From 7bcd569001ff3ff8660b8da87ece4b54bcee994a Mon Sep 17 00:00:00 2001 From: baihailong Date: Thu, 6 Feb 2025 17:23:19 +0800 Subject: [PATCH] fix(flashnode): add volume, clientIp, inode for DataSource.#23013250 Signed-off-by: baihailong --- client/fs/dir.go | 2 +- flashnode/cachengine/block.go | 6 ++++-- flashnode/cachengine/block_test.go | 14 +++++++------- flashnode/cachengine/engine.go | 16 ++++++++-------- flashnode/cachengine/engine_test.go | 8 ++++---- flashnode/extent_reader.go | 23 +++++++++++++++-------- flashnode/flashnode_op.go | 4 ++-- 7 files changed, 41 insertions(+), 32 deletions(-) diff --git a/client/fs/dir.go b/client/fs/dir.go index 408f84f35..d93287e22 100644 --- a/client/fs/dir.go +++ b/client/fs/dir.go @@ -1066,7 +1066,7 @@ func (d *Dir) addParentInode(inos []uint64) { } func (d *Dir) getCwd() string { - log.LogDebugf("getCwd,fullPath()", d.fullPath) + log.LogDebugf("getCwd,fullPath(%v)", d.fullPath) return d.fullPath } diff --git a/flashnode/cachengine/block.go b/flashnode/cachengine/block.go index bcaf9a6ef..244bce2f2 100644 --- a/flashnode/cachengine/block.go +++ b/flashnode/cachengine/block.go @@ -59,10 +59,11 @@ type CacheBlock struct { readyCh chan struct{} closeOnce sync.Once closeCh chan struct{} + clientIP string } // NewCacheBlock create and returns a new extent instance. -func NewCacheBlock(rootPath string, volume string, inode, fixedOffset uint64, version uint32, allocSize uint64, reader ReadExtentData) (cb *CacheBlock) { +func NewCacheBlock(rootPath string, volume string, inode, fixedOffset uint64, version uint32, allocSize uint64, reader ReadExtentData, clientIP string) (cb *CacheBlock) { cb = new(CacheBlock) cb.volume = volume cb.inode = inode @@ -75,6 +76,7 @@ func NewCacheBlock(rootPath string, volume string, inode, fixedOffset uint64, ve cb.sourceReader = reader cb.readyCh = make(chan struct{}) cb.closeCh = make(chan struct{}) + cb.clientIP = clientIP return } @@ -350,7 +352,7 @@ func (cb *CacheBlock) prepareSource(ctx context.Context, sourceCh <-chan *proto. if log.EnableDebug() { log.LogDebugf("%s start", logPrefix()) } - if _, err = cb.sourceReader(source, writeCacheAfterRead, readDataNodeTimeout); err != nil { + if _, err = cb.sourceReader(source, writeCacheAfterRead, readDataNodeTimeout, cb.volume, cb.inode, cb.clientIP); err != nil { log.LogErrorf("%s err:%v", logPrefix(), err) return } diff --git a/flashnode/cachengine/block_test.go b/flashnode/cachengine/block_test.go index 165a7fea3..285e5d64c 100644 --- a/flashnode/cachengine/block_test.go +++ b/flashnode/cachengine/block_test.go @@ -66,7 +66,7 @@ func TestBlockWriteCache(t *testing.T) { } func testWriteSingleFile(t *testing.T) { - cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE, nil) + cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE, nil, "") require.NoError(t, cacheBlock.initFilePath(false)) defer func() { require.NoError(t, cacheBlock.Delete()) }() bytes := randTestData(1024) @@ -75,7 +75,7 @@ func testWriteSingleFile(t *testing.T) { } func testWriteSingleFileError(t *testing.T) { - cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE, nil) + cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE, nil, "") require.NoError(t, cacheBlock.initFilePath(false)) defer func() { require.NoError(t, cacheBlock.Delete()) }() bytes := randTestData(1024) @@ -86,7 +86,7 @@ func testWriteSingleFileError(t *testing.T) { func testWriteCacheBlockFull(t *testing.T) { var err error - cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE, nil) + cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE, nil, "") require.NoError(t, cacheBlock.initFilePath(false)) defer func() { require.NoError(t, cacheBlock.Delete()) }() bytes := randTestData(1024) @@ -104,14 +104,14 @@ func testWriteCacheBlockFull(t *testing.T) { } func newCacheBlockWithDiffInode(volume string, index int, allocSize uint64) (cacheBlock *CacheBlock, err error) { - cacheBlock = NewCacheBlock(testTmpFS, volume, uint64(index), 1024, 112456871, allocSize, nil) + cacheBlock = NewCacheBlock(testTmpFS, volume, uint64(index), 1024, 112456871, allocSize, nil, "") err = cacheBlock.initFilePath(false) return } func newCacheBlockWithDiffVolume(volume string, index int, allocSize uint64) (cacheBlock *CacheBlock, err error) { newVolume := fmt.Sprintf("%s_%d", volume, index) - cacheBlock = NewCacheBlock(testTmpFS, newVolume, 1, 1024, 112456871, allocSize, nil) + cacheBlock = NewCacheBlock(testTmpFS, newVolume, 1, 1024, 112456871, allocSize, nil, "") err = cacheBlock.initFilePath(false) return } @@ -151,7 +151,7 @@ func TestBlockReadCache(t *testing.T) { require.NoError(t, err) defer func() { require.NoError(t, umount()) }() - cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 2568748711, proto.CACHE_BLOCK_SIZE, nil) + cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 2568748711, proto.CACHE_BLOCK_SIZE, nil, "") cacheBlock.lruFhCache = NewCache(LRUFileHandleCacheType, 1, -1, time.Hour, func(v interface{}) error { file := v.(*os.File) @@ -188,7 +188,7 @@ func testParallelOperation(t *testing.T) { require.NoError(t, err) defer func() { require.NoError(t, umount()) }() - cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE, nil) + cacheBlock := NewCacheBlock(testTmpFS, t.Name(), 1, 1024, 112456871, proto.CACHE_BLOCK_SIZE, nil, "") cacheBlock.lruFhCache = NewCache(LRUFileHandleCacheType, 1, -1, time.Hour, func(v interface{}) error { file := v.(*os.File) diff --git a/flashnode/cachengine/engine.go b/flashnode/cachengine/engine.go index 088dbb7c9..7e089a2d5 100644 --- a/flashnode/cachengine/engine.go +++ b/flashnode/cachengine/engine.go @@ -94,7 +94,7 @@ type CacheEngine struct { type ( ReadExtentAfter func([]byte, int64) error - ReadExtentData func(source *proto.DataSource, afterReadFunc ReadExtentAfter, timeout int) (n int, err error) + ReadExtentData func(source *proto.DataSource, afterReadFunc ReadExtentAfter, timeout int, volume string, ino uint64, clientIP string) (n int, err error) ) func NewCacheEngine(dataDir string, totalSize int64, maxUseRatio float64, @@ -223,7 +223,7 @@ func (c *CacheEngine) LoadCacheBlock() (err error) { log.LogDebugf("acton[LoadCacheBlock] dataPath(%v) cacheBlockName(%v) volume(%v) inode(%v) offset(%v) version(%v).", fullPath, fileInfo.Name(), volume, inode, offset, version) - if _, err = c.createCacheBlock(volume, inode, offset, version, c.config.LoadCbTTL, 0, true); err != nil { + if _, err = c.createCacheBlock(volume, inode, offset, version, c.config.LoadCbTTL, 0, true, ""); err != nil { c.deleteCacheBlock(GenCacheBlockKey(volume, inode, offset, version)) log.LogInfof("action[LoadCacheBlock] createCacheBlock(%v) from dataPath(%v) volume(%v) err(%v) ", filename, fullPath, volume, err.Error()) @@ -363,7 +363,7 @@ func (c *CacheEngine) PeekCacheBlock(key string) (block *CacheBlock, err error) return nil, errors.NewErrorf("cache block peek failed:%v", err) } -func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64, version uint32, ttl int64, allocSize uint64, isLoad bool) (block *CacheBlock, err error) { +func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64, version uint32, ttl int64, allocSize uint64, isLoad bool, clientIP string) (block *CacheBlock, err error) { if !isLoad && allocSize == 0 { return nil, fmt.Errorf("alloc size is zero") } @@ -393,7 +393,7 @@ func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64, } } - block = NewCacheBlock(c.dataPath, volume, inode, fixedOffset, version, allocSize, c.readSourceFunc) + block = NewCacheBlock(c.dataPath, volume, inode, fixedOffset, version, allocSize, c.readSourceFunc, clientIP) if ttl <= 0 { ttl = proto.DefaultCacheTTLSec } @@ -441,8 +441,8 @@ func (c *CacheEngine) startCachePrepareWorkers() { } } -func (c *CacheEngine) PrepareCache(reqID int64, req *proto.CacheRequest) (err error) { - if _, err = c.CreateBlock(req); err != nil { +func (c *CacheEngine) PrepareCache(reqID int64, req *proto.CacheRequest, clientIP string) (err error) { + if _, err = c.CreateBlock(req, clientIP); err != nil { return } select { @@ -453,11 +453,11 @@ func (c *CacheEngine) PrepareCache(reqID int64, req *proto.CacheRequest) (err er return } -func (c *CacheEngine) CreateBlock(req *proto.CacheRequest) (block *CacheBlock, err error) { +func (c *CacheEngine) CreateBlock(req *proto.CacheRequest, clientIP string) (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), false); err != nil { + if block, err = c.createCacheBlock(req.Volume, req.Inode, req.FixedFileOffset, req.Version, req.TTL, computeAllocSize(req.Sources), false, clientIP); err != nil { c.deleteCacheBlock(GenCacheBlockKey(req.Volume, req.Inode, req.FixedFileOffset, req.Version)) return nil, err } diff --git a/flashnode/cachengine/engine_test.go b/flashnode/cachengine/engine_test.go index 79ae04858..19e7bce94 100644 --- a/flashnode/cachengine/engine_test.go +++ b/flashnode/cachengine/engine_test.go @@ -53,7 +53,7 @@ func TestEngineNew(t *testing.T) { defer func() { require.NoError(t, ce.Stop()) }() var cb *CacheBlock inode, fixedOffset, version := uint64(1), uint64(1024), uint32(112358796) - cb, err = ce.createCacheBlock(t.Name(), inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, false) + cb, err = ce.createCacheBlock(t.Name(), inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, false, "") require.NoError(t, err) require.NoError(t, cb.WriteAt(bytesCommon, 0, 1024)) } @@ -83,7 +83,7 @@ func TestEngineOverFlow(t *testing.T) { inode, fixedOffset, version := uint64(1), uint64(1024), uint32(112358796) cb, err1 := ce.createCacheBlock(fmt.Sprintf("%s_%d_%d", t.Name(), round, thread), - inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, false) + inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, false, "") if err1 != nil { isErr.Store(true) return @@ -140,7 +140,7 @@ func TestEngineTTL(t *testing.T) { wg.Add(1) go func(index int) { defer wg.Done() - cb, err := ce.createCacheBlock(fmt.Sprintf("%s_%d", t.Name(), index), inode, fixedOffset, version, ttl, proto.CACHE_BLOCK_SIZE, false) + cb, err := ce.createCacheBlock(fmt.Sprintf("%s_%d", t.Name(), index), inode, fixedOffset, version, ttl, proto.CACHE_BLOCK_SIZE, false, "") require.NoError(t, err) var offset int64 for { @@ -183,7 +183,7 @@ func TestEngineLru(t *testing.T) { var cb *CacheBlock var offset int64 inode, fixedOffset, version := uint64(1), uint64(1024), uint32(112358796) - cb, err = ce.createCacheBlock(fmt.Sprintf("%s_%d", t.Name(), j), inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, false) + cb, err = ce.createCacheBlock(fmt.Sprintf("%s_%d", t.Name(), j), inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, false, "") require.NoError(t, err) for { err = cb.WriteAt(bytesCommon, offset, 1024) diff --git a/flashnode/extent_reader.go b/flashnode/extent_reader.go index 2014fcec0..e46dcd0f3 100644 --- a/flashnode/extent_reader.go +++ b/flashnode/extent_reader.go @@ -42,49 +42,56 @@ func initExtentConnPool() { extentReaderConnPool = util.NewConnectPoolWithTimeoutAndCap(5, 100, _connPoolIdleTimeout, 1) } -func ReadExtentData(source *proto.DataSource, afterReadFunc cachengine.ReadExtentAfter, timeout int) (readBytes int, err error) { +func ReadExtentData(source *proto.DataSource, afterReadFunc cachengine.ReadExtentAfter, timeout int, volume string, ino uint64, clientIP string) (readBytes int, err error) { reqPacket := newReadPacket(&proto.ExtentKey{ PartitionId: source.PartitionID, ExtentId: source.ExtentID, }, int(source.ExtentOffset), int(source.Size_), source.FileOffset, true) - readBytes, err = extentReadWithRetry(reqPacket, source.Hosts, afterReadFunc, timeout) + readBytes, err = extentReadWithRetry(reqPacket, source, afterReadFunc, timeout, volume, ino, clientIP) if err != nil { log.LogErrorf("read extent err(%v)", err) } return } -func extentReadWithRetry(reqPacket *proto.Packet, hosts []string, afterReadFunc cachengine.ReadExtentAfter, timeout int) (readBytes int, err error) { +func extentReadWithRetry(reqPacket *proto.Packet, source *proto.DataSource, afterReadFunc cachengine.ReadExtentAfter, timeout int, volume string, ino uint64, clientIP string) (readBytes int, err error) { errMap := make(map[string]error) startTime := time.Now() + hosts := source.Hosts + for try := range [_extentReadMaxRetry]struct{}{} { try++ for _, addr := range hosts { if addr == "" { continue } - log.LogDebugf("extentReadWithRetry: try(%d) addr(%s) reqPacket(%v)", try, addr, reqPacket) + log.LogDebugf("extentReadWithRetry: try(%d) addr(%s) hosts(%v) reqPacket(%v) volume(%v) ino(%v) client(%v)", + try, addr, hosts, reqPacket, volume, ino, clientIP) if readBytes, err = readFromDataPartition(addr, reqPacket, afterReadFunc, timeout); err == nil { return } errMap[addr] = err - log.LogWarnf("extentReadWithRetry: try(%d) addr(%s) reqPacket(%v) err(%v)", try, addr, reqPacket, err) + log.LogWarnf("extentReadWithRetry: try(%d) addr(%s) hosts(%v) reqPacket(%v) volume(%v) ino(%v) client(%v) err(%v)", + try, addr, hosts, reqPacket, volume, ino, clientIP, err) if strings.Contains(err.Error(), proto.ErrTmpfsNoSpace.Error()) { break } } if time.Since(startTime) > time.Duration(timeout)*time.Second { - log.LogWarnf("extentReadWithRetry: retry timeout req(%v) time(%v)", reqPacket, time.Since(startTime)) + log.LogWarnf("extentReadWithRetry: retry timeout req(%v) time(%v) hosts(%v) volume(%v) ino(%v) client(%v)", + reqPacket, time.Since(startTime), hosts, volume, ino, clientIP) break } - log.LogWarnf("extentReadWithRetry: errMap(%v) reqPacket(%v) try the next round", errMap, reqPacket) + log.LogWarnf("extentReadWithRetry: errMap(%v) reqPacket(%v) hosts(%v) volume(%v) ino(%v) client(%v) try the next round", + errMap, reqPacket, hosts, volume, ino, clientIP) rand.Seed(time.Now().UnixNano()) sleepDuration := rand.Intn(401) + 100 duration := time.Duration(sleepDuration) * time.Millisecond time.Sleep(duration) } - err = errors.NewErrorf("FollowerRead: tried %d times reqPacket(%v) errMap(%v)", _extentReadMaxRetry, reqPacket, errMap) + err = errors.NewErrorf("FollowerRead: tried %d times hosts(%v) reqPacket(%v) volume(%v) ino(%v) client(%v) errMap(%v)", + _extentReadMaxRetry, hosts, reqPacket, volume, ino, clientIP, errMap) return } diff --git a/flashnode/flashnode_op.go b/flashnode/flashnode_op.go index 01b6fda9b..6d23d3c20 100644 --- a/flashnode/flashnode_op.go +++ b/flashnode/flashnode_op.go @@ -151,7 +151,7 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) { errMetric := exporter.NewCounter("lowerHitRate") errMetric.AddWithLabels(1, map[string]string{exporter.FlashNode: f.localAddr, exporter.Err: "LowerHitRate"}) } - if block, err = f.cacheEngine.CreateBlock(cr); err != nil { + if block, err = f.cacheEngine.CreateBlock(cr, conn.RemoteAddr().String()); err != nil { log.LogErrorf("opCacheRead: CreateBlock failed, req(%v) err(%v)", req, err) return err } @@ -255,7 +255,7 @@ func (f *FlashNode) opCachePrepare(conn net.Conn, p *proto.Packet) (err error) { } volume = req.CacheRequest.Volume - if err = f.cacheEngine.PrepareCache(p.ReqID, req.CacheRequest); err != nil { + if err = f.cacheEngine.PrepareCache(p.ReqID, req.CacheRequest, conn.RemoteAddr().String()); err != nil { log.LogErrorf("%s prepare %v", action, err) return }