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(lcnode): multipart migration to blobstore

Browse Source 
Signed-off-by: zhaochenyang <zhaochenyang@oppo.com>
This commit is contained in:
zhaochenyang 2024-02-29 14:39:36 +08:00 committed by AmazingChi
parent 582b6f19aa
commit 08665970fd
4 changed files with 81 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
lcnode/lc_transition.go
View File

@ -19,6 +19,7 @@ import (
"crypto/md5"
"encoding/hex"
"io"
"strings"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
@ -36,7 +37,7 @@ type ExtentApi interface {
}
type EbsApi interface {
Put(ctx context.Context, volName string, f io.Reader, size uint64) (oek proto.ObjExtentKey, md5 []byte, err error)
Put(ctx context.Context, volName string, f io.Reader, size uint64) (oek []proto.ObjExtentKey, md5 [][]byte, err error)
Get(ctx context.Context, volName string, offset uint64, size uint64, oek proto.ObjExtentKey) (body io.ReadCloser, err error)
}
@ -122,7 +123,7 @@ func (t *TransitionMgr) migrate(e *proto.ScanDentry) (err error) {
//check read from src extent
srcMd5Hash := md5.New()
err = t.readFromExtentClient(e, srcMd5Hash, false)
err = t.readFromExtentClient(e, srcMd5Hash, false, 0, 0)
if err != nil {
log.LogErrorf("check: read from src extent err: %v, inode(%v)", err, e.Inode)
return
@ -137,7 +138,7 @@ func (t *TransitionMgr) migrate(e *proto.ScanDentry) (err error) {
//check read from dst migration extent
dstMd5Hash := md5.New()
err = t.readFromExtentClient(e, dstMd5Hash, true)
err = t.readFromExtentClient(e, dstMd5Hash, true, 0, 0)
if err != nil {
log.LogErrorf("check: read from dst extent err: %v, inode(%v)", err, e.Inode)
return
@ -154,7 +155,7 @@ func (t *TransitionMgr) migrate(e *proto.ScanDentry) (err error) {
return
}
func (t *TransitionMgr) readFromExtentClient(e *proto.ScanDentry, writer io.Writer, isMigrationExtent bool) (err error) {
func (t *TransitionMgr) readFromExtentClient(e *proto.ScanDentry, writer io.Writer, isMigrationExtent bool, from, size int) (err error) {
var (
readN int
readOffset int
@ -163,8 +164,14 @@ func (t *TransitionMgr) readFromExtentClient(e *proto.ScanDentry, writer io.Writ
buf = make([]byte, 2*util.BlockSize)
)
if size > 0 {
readOffset = from
} else {
size = int(e.Size)
}
for {
if rest = int(e.Size) - readOffset; rest <= 0 {
if rest = (from + size) - readOffset; rest <= 0 {
break
}
readSize = len(buf)
@ -196,7 +203,7 @@ func (t *TransitionMgr) readFromExtentClient(e *proto.ScanDentry, writer io.Writ
return
}
func (t *TransitionMgr) migrateToEbs(e *proto.ScanDentry) (oeks []proto.ObjExtentKey, err error) {
func (t *TransitionMgr) migrateToEbs(e *proto.ScanDentry) (oek []proto.ObjExtentKey, err error) {
if err = t.ec.OpenStream(e.Inode, false, false); err != nil {
log.LogErrorf("migrateToEbs: OpenStream fail, inode(%v) err: %v", e.Inode, err)
return
@ -209,7 +216,7 @@ func (t *TransitionMgr) migrateToEbs(e *proto.ScanDentry) (oeks []proto.ObjExten
r, w := io.Pipe()
go func() {
err = t.readFromExtentClient(e, w, false)
err = t.readFromExtentClient(e, w, false, 0, 0)
if err != nil {
log.LogErrorf("migrateToEbs: read from extent err: %v, inode(%v)", err, e.Inode)
}
@ -217,31 +224,48 @@ func (t *TransitionMgr) migrateToEbs(e *proto.ScanDentry) (oeks []proto.ObjExten
}()
ctx := context.Background()
oek, m, err := t.ebsClient.Put(ctx, t.volume, r, e.Size)
oek, _md5, err := t.ebsClient.Put(ctx, t.volume, r, e.Size)
if err != nil {
log.LogErrorf("migrateToEbs: ebs put err: %v, inode(%v)", err, e.Inode)
r.Close()
return
}
r.Close()
md5Value := hex.EncodeToString(m)
var md5Value []string
for _, m := range _md5 {
md5Value = append(md5Value, hex.EncodeToString(m))
}
log.LogDebugf("migrateToEbs finished, inode(%v), oek: %v, md5Value: %v", e.Inode, oek, md5Value)
//check read from extent
srcMd5Hash := md5.New()
err = t.readFromExtentClient(e, srcMd5Hash, false)
if err != nil {
log.LogErrorf("migrateToEbs: check err: %v, inode(%v)", err, e.Inode)
return
var srcMd5 []string
var from int
var part uint64 = util.ExtentSize
var rest = e.Size
for rest > 0 {
var getSize uint64
if rest > part {
getSize = part
} else {
getSize = rest
}
rest -= getSize
srcMd5Hash := md5.New()
err = t.readFromExtentClient(e, srcMd5Hash, false, from, int(getSize))
if err != nil {
log.LogErrorf("migrateToEbs: check err: %v, inode(%v)", err, e.Inode)
return
}
srcMd5 = append(srcMd5, hex.EncodeToString(srcMd5Hash.Sum(nil)))
from += int(getSize)
}
srcMd5 := hex.EncodeToString(srcMd5Hash.Sum(nil))
log.LogDebugf("migrateToEbs check finished, inode(%v), srcmd5: %v", e.Inode, srcMd5)
if srcMd5 != md5Value {
if strings.Join(srcMd5, ",") != strings.Join(md5Value, ",") {
err = errors.NewErrorf("migrateToEbs check md5 inconsistent, srcMd5: %v, md5Value: %v", srcMd5, md5Value)
return
}
oeks = []proto.ObjExtentKey{oek}
log.LogInfof("migrateToEbs and check finished, inode(%v)", e.Inode)
return
}

4
lcnode/lc_transition_test.go
View File

@ -73,11 +73,11 @@ func NewMockEbsClient() *MockEbsClient {
return &MockEbsClient{}
}
func (m *MockEbsClient) Put(ctx context.Context, volName string, f io.Reader, size uint64) (oek proto.ObjExtentKey, _md5 []byte, err error) {
func (m *MockEbsClient) Put(ctx context.Context, volName string, f io.Reader, size uint64) (oek []proto.ObjExtentKey, _md5 [][]byte, err error) {
m.data, _ = ioutil.ReadAll(f)
md5Hash := md5.New()
md5Hash.Write(m.data)
_md5 = md5Hash.Sum(nil)
_md5 = append(_md5, md5Hash.Sum(nil))
return
}

2
lcnode/server.go
View File

@ -504,7 +504,7 @@ func (l *LcNode) debugServiceGetFile(w http.ResponseWriter, r *http.Request) {
Inode: ino,
StorageClass: uint32(sc),
}
if err = t.readFromExtentClient(e, w, true); err != nil {
if err = t.readFromExtentClient(e, w, true, 0, 0); err != nil {
http.Error(w, fmt.Sprintf("readFromExtentClient err: %v", err.Error()), http.StatusBadRequest)
return
}

53
sdk/data/blobstore/blobstore_client.go
View File

@ -235,7 +235,7 @@ func createOPMetricBySize(size uint64, tag string) string {
return tag + "1G_"
}
func (ebs *BlobStoreClient) Put(ctx context.Context, volName string, f io.Reader, size uint64) (oek proto.ObjExtentKey, md5 []byte, err error) {
func (ebs *BlobStoreClient) Put(ctx context.Context, volName string, f io.Reader, size uint64) (oek []proto.ObjExtentKey, md5 [][]byte, err error) {
bgTime := stat.BeginStat()
defer func() {
stat.EndStat("ebs-write", err, bgTime, 1)
@ -250,22 +250,45 @@ func (ebs *BlobStoreClient) Put(ctx context.Context, volName string, f io.Reader
metric.SetWithLabels(err, map[string]string{exporter.Vol: volName})
}()
location, hash, err := ebs.client.Put(ctx, &access.PutArgs{
Size: int64(size),
Hashes: access.HashAlgMD5,
Body: f,
})
if err != nil {
log.LogErrorf("TRACE Ebs Put, err(%v),requestId(%v)", err.Error(), requestId)
return
var part uint64 = util.ExtentSize
var rest = size
for rest > 0 {
var putSize uint64
if rest > part {
putSize = part
} else {
putSize = rest
}
rest -= putSize
var location access.Location
var hash access.HashSumMap
location, hash, err = ebs.client.Put(ctx, &access.PutArgs{
Size: int64(putSize),
Hashes: access.HashAlgMD5,
Body: f,
})
if err != nil {
log.LogErrorf("TRACE Ebs Put, err(%v),requestId(%v)", err.Error(), requestId)
return
}
var _md5 []byte
_md5, err = hex.DecodeString(hash.GetSumVal(access.HashAlgMD5).(string))
if err != nil {
log.LogErrorf("decode md5 %v, err %v", hash.GetSumVal(access.HashAlgMD5).(string), err)
return
}
oek = append(oek, locationToObjExtentKey(location))
md5 = append(md5, _md5)
log.LogDebugf("TRACE Ebs Put, requestId(%v) loc(%v) putSize(%v)", requestId, location, putSize)
}
md5, err = hex.DecodeString(hash.GetSumVal(access.HashAlgMD5).(string))
if err != nil {
log.LogErrorf("decode md5 %v, err %v", hash.GetSumVal(access.HashAlgMD5).(string), err)
return
}
elapsed := time.Since(start)
log.LogDebugf("TRACE Ebs Put Exit, requestId(%v) oek(%v) md5(%v) size(%v) consume(%v)ns", requestId, oek, md5, size, elapsed.Nanoseconds())
return
}
func locationToObjExtentKey(location access.Location) (oek proto.ObjExtentKey) {
blobs := make([]proto.Blob, 0)
for _, info := range location.Blobs {
blob := proto.Blob{
@ -285,8 +308,6 @@ func (ebs *BlobStoreClient) Put(ctx context.Context, volName string, f io.Reader
FileOffset: 0,
Crc: location.Crc,
}
elapsed := time.Since(start)
log.LogDebugf("TRACE Ebs Put Exit, requestId(%v) loc(%v) oek(%v) md5(%v) size(%v) consume(%v)ns", requestId, location, oek, string(md5), size, elapsed.Nanoseconds())
return
}