gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(client): add debug log for parameter error

Browse Source 
close:#1000080725

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-04-24 16:23:17 +08:00 committed by zhumingze1108
parent 8056986b3e
commit fb7b1ea7dc
4 changed files with 31 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
flashnode/cachengine/block.go
View File

@ -142,7 +142,7 @@ func IsDiskErr(errMsg string) bool {
}
// WriteAt writes data to an cacheBlock, only append write supported
func (cb *CacheBlock) WriteAt(data []byte, offset, size int64) (err error) {
func (cb *CacheBlock) WriteAt(data []byte, offset, size int64, reqID int64) (err error) {
var file *os.File
bgTime := stat.BeginStat()
startTime := time.Now()
@ -160,7 +160,7 @@ func (cb *CacheBlock) WriteAt(data []byte, offset, size int64) (err error) {
}
}()
if alloced := cb.getAllocSize(); offset >= alloced || size == 0 || offset+size > alloced {
return fmt.Errorf("parameter offset=%d size=%d allocSize:%d", offset, size, alloced)
return fmt.Errorf("reqID %v parameter offset=%d size=%d allocSize:%d", reqID, offset, size, alloced)
}
if file, err = cb.GetOrOpenFileHandler(); err != nil {
@ -300,7 +300,7 @@ func (cb *CacheBlock) checkCacheBlockFileHeader(file *os.File) (allocSize, usedS
return
}
func (cb *CacheBlock) initFilePath(isLoad bool) (err error) {
func (cb *CacheBlock) initFilePath(isLoad bool, reqID int64) (err error) {
defer func() {
if err != nil {
if IsDiskErr(err.Error()) {
@ -362,14 +362,14 @@ func (cb *CacheBlock) initFilePath(isLoad bool) (err error) {
}
_, err = os.Stat(cb.filePath)
if !isLoad {
msg := fmt.Sprintf("init cache block(%s) to local : err %v", cb.info(), err)
msg := fmt.Sprintf("init cache block(%s) to local reqID %v: err %v", cb.info(), reqID, err)
log.LogDebugf("%v", msg)
auditlog.LogFlashNodeOp("BlockInit", msg, err)
}
return
}
func (cb *CacheBlock) Init(sources []*proto.DataSource, readDataNodeTimeout int) {
func (cb *CacheBlock) Init(sources []*proto.DataSource, readDataNodeTimeout int, reqID int64) {
var err error
var file *os.File
bgTime := stat.BeginStat()
@ -395,7 +395,7 @@ func (cb *CacheBlock) Init(sources []*proto.DataSource, readDataNodeTimeout int)
for i := 0; i < util.Min(20, len(sources)); i++ {
wg.Add(1)
go func() {
if err := cb.prepareSource(ctx, sourceTaskCh, readDataNodeTimeout); err != nil {
if err := cb.prepareSource(ctx, sourceTaskCh, readDataNodeTimeout, reqID); err != nil {
cancel()
}
wg.Done()
@ -439,7 +439,7 @@ func (cb *CacheBlock) Init(sources []*proto.DataSource, readDataNodeTimeout int)
cb.notifyReady()
}
func (cb *CacheBlock) prepareSource(ctx context.Context, sourceCh <-chan *proto.DataSource, readDataNodeTimeout int) (err error) {
func (cb *CacheBlock) prepareSource(ctx context.Context, sourceCh <-chan *proto.DataSource, readDataNodeTimeout int, reqID int64) (err error) {
bgTime := stat.BeginStat()
defer func() {
stat.EndStat("CacheBlock:prepareSource", err, bgTime, 1)
@ -454,7 +454,7 @@ func (cb *CacheBlock) prepareSource(ctx context.Context, sourceCh <-chan *proto.
}
offset := int64(source.FileOffset) & (proto.CACHE_BLOCK_SIZE - 1)
writeCacheAfterRead := func(data []byte, size int64) error {
if e := cb.WriteAt(data, offset, size); e != nil {
if e := cb.WriteAt(data, offset, size, reqID); e != nil {
return e
}
offset += size
@ -522,8 +522,8 @@ func computeAllocSize(sources []*proto.DataSource) (alloc uint64) {
return
}
func (cb *CacheBlock) InitOnce(engine *CacheEngine, sources []*proto.DataSource) {
cb.initOnce.Do(func() { cb.Init(sources, engine.readDataNodeTimeout) })
func (cb *CacheBlock) InitOnce(engine *CacheEngine, sources []*proto.DataSource, reqID int64) {
cb.initOnce.Do(func() { cb.Init(sources, engine.readDataNodeTimeout, reqID) })
select {
case <-cb.closeCh:
engine.deleteCacheBlock(cb.blockKey)
@ -563,9 +563,9 @@ func (cb *CacheBlock) GetRootPath() string {
return cb.rootPath
}
func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto.DataSource, done chan struct{}) {
func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto.DataSource, done chan struct{}, reqID int64) {
cb.initOnce.Do(func() {
cb.InitForCacheRead(sources, engine.readDataNodeTimeout)
cb.InitForCacheRead(sources, engine.readDataNodeTimeout, reqID)
select {
case <-cb.closeCh:
engine.deleteCacheBlock(cb.blockKey)
@ -576,7 +576,7 @@ func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto
close(done)
}
func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNodeTimeout int) {
func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNodeTimeout int, reqID int64) {
var err error
var file *os.File
bgTime := stat.BeginStat()
@ -598,7 +598,7 @@ func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNode
for _, s := range sources {
offset := int64(s.FileOffset) & (proto.CACHE_BLOCK_SIZE - 1)
writeCacheAfterRead := func(data []byte, size int64) error {
if e := cb.WriteAt(data, offset, size); e != nil {
if e := cb.WriteAt(data, offset, size, reqID); e != nil {
return e
}
offset += size
@ -645,5 +645,5 @@ func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNode
}
func (cb *CacheBlock) info() string {
return fmt.Sprintf("path(%v)_from(%v)", cb.filePath, cb.clientIP)
return fmt.Sprintf("path(%v)_from(%v)_size(%v)", cb.filePath, cb.clientIP, cb.allocSize)
}

14
flashnode/cachengine/engine.go
View File

@ -604,7 +604,7 @@ func (c *CacheEngine) createCacheBlockFromExist(dataPath string, volume string,
}
}()
if err = block.initFilePath(true); err != nil {
if err = block.initFilePath(true, 0); err != nil {
return
}
@ -616,7 +616,7 @@ func (c *CacheEngine) createCacheBlockFromExist(dataPath string, volume string,
return
}
func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64, version uint32, ttl int64, allocSize uint64, clientIP string, isPrepare bool) (block *CacheBlock, err error) {
func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64, version uint32, ttl int64, allocSize uint64, clientIP string, isPrepare bool, reqID int64) (block *CacheBlock, err error) {
if allocSize == 0 {
return nil, fmt.Errorf("alloc size is zero")
}
@ -684,7 +684,7 @@ func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64,
return
}
if err = block.initFilePath(false); err != nil {
if err = block.initFilePath(false, reqID); err != nil {
return
}
if _, err = cacheItem.lruCache.Set(key, block, time.Duration(ttl)*time.Second); err != nil {
@ -732,7 +732,7 @@ func (c *CacheEngine) startCachePrepareWorkers() {
return
case task := <-c.cachePrepareTaskCh:
r := task.request
if _, err := c.CreateBlock(r, task.clientIP, true); err != nil {
if _, err := c.CreateBlock(r, task.clientIP, true, task.reqID); err != nil {
log.LogWarnf("action[startCachePrepareWorkers] ReqID(%d) create block failed, err:%v", task.reqID, err)
continue
}
@ -740,7 +740,7 @@ func (c *CacheEngine) startCachePrepareWorkers() {
if err != nil {
log.LogWarnf("action[startCachePrepareWorkers] ReqID(%d) cache block not found, err:%v", task.reqID, err)
} else {
block.InitOnce(c, r.Sources)
block.InitOnce(c, r.Sources, task.reqID)
}
}
}
@ -757,11 +757,11 @@ func (c *CacheEngine) PrepareCache(reqID int64, req *proto.CacheRequest, clientI
return
}
func (c *CacheEngine) CreateBlock(req *proto.CacheRequest, clientIP string, isPrepare bool) (block *CacheBlock, err error) {
func (c *CacheEngine) CreateBlock(req *proto.CacheRequest, clientIP string, isPrepare bool, reqID int64) (block *CacheBlock, err error) {
if len(req.Sources) == 0 {
return nil, fmt.Errorf("no source data")
}
if block, err = c.createCacheBlock(req.Volume, req.Inode, req.FixedFileOffset, req.Version, req.TTL, computeAllocSize(req.Sources), clientIP, isPrepare); err != nil {
if block, err = c.createCacheBlock(req.Volume, req.Inode, req.FixedFileOffset, req.Version, req.TTL, computeAllocSize(req.Sources), clientIP, isPrepare, reqID); err != nil {
log.LogWarnf("action[CreateBlock] createCacheBlock(%v) failed err %v ",
GenCacheBlockKey(req.Volume, req.Inode, req.FixedFileOffset, req.Version), err)
c.deleteCacheBlock(GenCacheBlockKey(req.Volume, req.Inode, req.FixedFileOffset, req.Version))

4
flashnode/flashnode_op.go
View File

@ -194,12 +194,12 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
reqSize += int(source.Size_)
}
if err = f.limitWrite.TryRunAsync(ctx, reqSize, f.waitForCacheBlock, func() {
if block2, err2 := f.cacheEngine.CreateBlock(cr, conn.RemoteAddr().String(), false); err2 != nil {
if block2, err2 := f.cacheEngine.CreateBlock(cr, conn.RemoteAddr().String(), false, p.ReqID); err2 != nil {
log.LogWarnf("opCacheRead: CreateBlock failed, req(%v) err(%v)", req, err2)
close(missTaskDone)
return
} else {
block2.InitOnceForCacheRead(f.cacheEngine, cr.Sources, missTaskDone)
block2.InitOnceForCacheRead(f.cacheEngine, cr.Sources, missTaskDone, p.ReqID)
}
}); err != nil {
stat.EndStat("MissCacheReadLimit", err, bgTime2, 1)

9
sdk/data/stream/stream_remote_cache.go
View File

@ -126,6 +126,8 @@ func (s *Streamer) readFromRemoteCache(ctx context.Context, offset, size uint64,
}
return
} else {
log.LogDebugf("readFromRemoteCache: inode(%d) cacheReadRequest version %v, source %v",
s.inode, req.CacheRequest.Version, req.CacheRequest.Sources)
total += read
}
}
@ -171,8 +173,10 @@ func (s *Streamer) getDataSource(start, size, fixedFileOffset uint64, isRead boo
Hosts: sortedHosts,
}
sources = append(sources, source)
log.LogDebugf("getDataSource: append source inode %v PartitionID %v ExtentID %v FileOffset %v ExtentOffset %v",
s.inode, source.PartitionID, source.ExtentID, source.FileOffset, source.ExtentOffset)
log.LogDebugf("getDataSource: append source inode %v PartitionID %v ExtentID %v FileOffset %v "+
"ExtentOffset %v fixedFileOffset %v size %v",
s.inode, source.PartitionID, source.ExtentID, source.FileOffset, source.ExtentOffset, fixedFileOffset,
source.Size_)
}
return sources, nil
}
@ -217,6 +221,7 @@ func (s *Streamer) prepareCacheRequests(offset, size uint64, data []byte, gen ui
cReadRequests = append(cReadRequests, cReadRequest)
}
}
log.LogDebugf("prepareCacheRequests: inode %v extent[offset=%v,size=%v] cReadRequests %v ", s.inode, offset, size, cReadRequests)
return cReadRequests, nil
}