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(flashnode): add volume, clientIp, inode for DataSource.#23013250

Browse Source 
Signed-off-by: baihailong <baihailong@oppo.com>
This commit is contained in:
baihailong 2025-02-06 17:23:19 +08:00 committed by zhumingze1108
parent eb88697dac
commit 7bcd569001
7 changed files with 41 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
client/fs/dir.go
View File

@ -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
}

6
flashnode/cachengine/block.go
View File

@ -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
}

14
flashnode/cachengine/block_test.go
View File

@ -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)

16
flashnode/cachengine/engine.go
View File

@ -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
}

8
flashnode/cachengine/engine_test.go
View File

@ -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)

23
flashnode/extent_reader.go
View File

@ -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
}

4
flashnode/flashnode_op.go
View File

@ -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
}