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

refactor(shardnode): replace msg.go from base to blobdeleter

Browse Source 
with: #1000042472

Signed-off-by: xiejian <xiejian3@oppo.com>
This commit is contained in:
xiejian 2025-08-13 16:14:52 +08:00 committed by slasher
parent a851e43681
commit 59ec91cf3c
6 changed files with 32 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
blobstore/shardnode/blobdeleter/blob_deleter.go
View File

@ -543,7 +543,7 @@ func (m *BlobDeleteMgr) punish(ctx context.Context, msgExt *delMsgExt) error {
return nil
}
shard := v.(*shardListReader)
_, _, _, shardKeys, err := base.DecodeDelMsgKey(msgExt.msgKey, shard.ShardingSubRangeCount())
_, _, _, shardKeys, err := DecodeDelMsgKey(msgExt.msgKey, shard.ShardingSubRangeCount())
if err != nil {
span.Errorf("shard[%d] decode shardKeys failed, key: %+v, err: %s", msgExt.suid, msgExt.msgKey, err.Error())
return err
@ -551,7 +551,7 @@ func (m *BlobDeleteMgr) punish(ctx context.Context, msgExt *delMsgExt) error {
punishDr := time.Duration(m.cfg.PunishTimeoutH) * time.Hour
ts := m.tsGen.CurrentTs().Add(punishDr)
punishMsgKey := base.EncodeDelMsgKey(ts, msg.Slice.Vid, msg.Slice.MinSliceID, shardKeys)
punishMsgKey := EncodeDelMsgKey(ts, msg.Slice.Vid, msg.Slice.MinSliceID, shardKeys)
msg.Retry = 0
msg.Time = ts.TimeUnix()
@ -594,7 +594,7 @@ func (m *BlobDeleteMgr) clearShardMessages(ctx context.Context, suid proto.Suid,
return
}
shard := v.(*shardListReader)
_, _, _, shardKeys, err := base.DecodeDelMsgKey(items[0].ID, shard.ShardingSubRangeCount())
_, _, _, shardKeys, err := DecodeDelMsgKey(items[0].ID, shard.ShardingSubRangeCount())
if err != nil {
span.Errorf("shard[%d] decode shardKeys failed, key: %+v, err: %s", suid, items[0].ID, err.Error())
return
@ -654,7 +654,7 @@ func (m *BlobDeleteMgr) slicesToDeleteMsgItems(ctx context.Context, slices []pro
items := make([]snapi.Item, len(slices))
for i := range slices {
ts := m.tsGen.GenerateTs()
key := base.EncodeDelMsgKey(ts, slices[i].Vid, slices[i].MinSliceID, shardKeys)
key := EncodeDelMsgKey(ts, slices[i].Vid, slices[i].MinSliceID, shardKeys)
msg := snproto.DeleteMsg{
Slice: slices[i],
Time: ts.TimeUnix(),
@ -674,7 +674,7 @@ func (m *BlobDeleteMgr) slicesToDeleteMsgItems(ctx context.Context, slices []pro
func (m *BlobDeleteMgr) sliceToDeleteMsgItemRaw(ctx context.Context, slices proto.Slice, tagNum int) (snapi.Item, [][]byte, error) {
span := trace.SpanFromContextSafe(ctx)
ts := m.tsGen.GenerateTs()
key, shardKeys := base.EncodeRawDelMsgKey(ts, slices.Vid, slices.MinSliceID, tagNum)
key, shardKeys := EncodeRawDelMsgKey(ts, slices.Vid, slices.MinSliceID, tagNum)
msg := snproto.DeleteMsg{
Slice: slices,

10
blobstore/shardnode/blobdeleter/blob_deleter_test.go
View File

@ -221,7 +221,7 @@ func TestBlobDeleteMgr_RunConsumeTask(t *testing.T) {
slice.Count = cnt
j += cnt
ts := mgr.tsGen.GenerateTs()
id, _ := base.EncodeRawDelMsgKey(ts, slice.Vid, slice.MinSliceID, 2)
id, _ := EncodeRawDelMsgKey(ts, slice.Vid, slice.MinSliceID, 2)
mgr.msgChannels[0] <- &delMsgExt{
suid: suid,
msg: snproto.DeleteMsg{
@ -404,7 +404,7 @@ func TestBlobDeleteMgr_DeleteFailed_Punish(t *testing.T) {
tagNum := 2
bidCnt := 5
ts := mgr.tsGen.GenerateTs()
id, _ := base.EncodeRawDelMsgKey(ts, vid, minBid, tagNum)
id, _ := EncodeRawDelMsgKey(ts, vid, minBid, tagNum)
msg := &delMsgExt{
msgKey: id,
suid: suid,
@ -550,7 +550,7 @@ func TestBlobDeleteMgr_Punish(t *testing.T) {
vid := proto.Vid(1)
bid := proto.BlobID(100)
ts := mgr.tsGen.GenerateTs()
id, _ := base.EncodeRawDelMsgKey(ts, vid, bid, tagNum)
id, _ := EncodeRawDelMsgKey(ts, vid, bid, tagNum)
msg := &delMsgExt{
suid: proto.Suid(123),
@ -591,7 +591,7 @@ func TestBlobDeleteMgr_ClearShardMessages(t *testing.T) {
delItems := make([]storage.BatchItemElem, 3)
for i := 0; i < 3; i++ {
ts := mgr.tsGen.GenerateTs()
id, _ := base.EncodeRawDelMsgKey(ts, proto.Vid(i), proto.BlobID(i), tagNum)
id, _ := EncodeRawDelMsgKey(ts, proto.Vid(i), proto.BlobID(i), tagNum)
delItems[i] = storage.NewBatchItemElemDelete(id)
}
mgr.clearShardMessages(context.Background(), proto.Suid(123), delItems)
@ -802,7 +802,7 @@ func TestBlobDeleteMgr_SlicesToDeleteMsgItems(t *testing.T) {
items, err := mgr.SlicesToDeleteMsgItems(context.Background(), []proto.Slice{slice1, slice2}, shardKeys)
require.Nil(t, err)
for i := range items {
_, _, _, _shardKeys, err := base.DecodeDelMsgKey(items[i].ID, len(shardKeys))
_, _, _, _shardKeys, err := DecodeDelMsgKey(items[i].ID, len(shardKeys))
require.Nil(t, err)
require.Equal(t, shardKeys, _shardKeys)
require.True(t, bytes.Contains(items[i].ID, snproto.DeleteMsgPrefix))

13
blobstore/shardnode/base/msg.go → blobstore/shardnode/blobdeleter/msg.go
View File

@ -12,7 +12,7 @@
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package base
package blobdeleter
import (
"encoding/binary"
@ -21,6 +21,7 @@ import (
snapi "github.com/cubefs/cubefs/blobstore/api/shardnode"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/shardnode/base"
snproto "github.com/cubefs/cubefs/blobstore/shardnode/proto"
)
@ -28,7 +29,7 @@ const minMsgKeyLen = 21 // 1(prefix) + 8(ts) + 4(vid) + 8(bid)
var errInvalidMsgKey = errors.New("invalid msg key")
func EncodeDelMsgKey(ts Ts, vid proto.Vid, bid proto.BlobID, shardKeys [][]byte) []byte {
func EncodeDelMsgKey(ts base.Ts, vid proto.Vid, bid proto.BlobID, shardKeys [][]byte) []byte {
shardKeyLen := 0
for _, sk := range shardKeys {
shardKeyLen += len(sk)
@ -55,7 +56,7 @@ func EncodeDelMsgKey(ts Ts, vid proto.Vid, bid proto.BlobID, shardKeys [][]byte)
return buf
}
func EncodeRawDelMsgKey(ts Ts, vid proto.Vid, bid proto.BlobID, tagNum int) (key []byte, shardKeys [][]byte) {
func EncodeRawDelMsgKey(ts base.Ts, vid proto.Vid, bid proto.BlobID, tagNum int) (key []byte, shardKeys [][]byte) {
key1 := []byte(fmt.Sprintf("%d", uint32(vid)))
key2 := []byte(fmt.Sprintf("%d", uint64(bid)))
@ -71,11 +72,11 @@ func EncodeRawDelMsgKey(ts Ts, vid proto.Vid, bid proto.BlobID, tagNum int) (key
return EncodeDelMsgKey(ts, vid, bid, shardKeys), shardKeys
}
func DecodeDelMsgKey(key []byte, tagNum int) (Ts, proto.Vid, proto.BlobID, [][]byte, error) {
func DecodeDelMsgKey(key []byte, tagNum int) (base.Ts, proto.Vid, proto.BlobID, [][]byte, error) {
if len(key) < minMsgKeyLen+2*tagNum {
return Ts(0), proto.InvalidVid, proto.InValidBlobID, nil, errInvalidMsgKey
return base.Ts(0), proto.InvalidVid, proto.InValidBlobID, nil, errInvalidMsgKey
}
ts := Ts(binary.BigEndian.Uint64(key[1:9]))
ts := base.Ts(binary.BigEndian.Uint64(key[1:9]))
vid := proto.Vid(binary.BigEndian.Uint32(key[9:13]))
bid := proto.BlobID(binary.BigEndian.Uint64(key[13:21]))
return ts, vid, bid, snapi.ParseShardKeys(key[minMsgKeyLen:], tagNum), nil

11
blobstore/shardnode/base/msg_test.go → blobstore/shardnode/blobdeleter/msg_test.go
View File

@ -12,7 +12,7 @@
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package base
package blobdeleter
import (
"bytes"
@ -28,11 +28,12 @@ import (
kvstore "github.com/cubefs/cubefs/blobstore/common/kvstorev2"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/shardnode/base"
)
func TestEncodeDecodeDelMsgKey(t *testing.T) {
t.Run("EncodeDelMsgKey", func(t *testing.T) {
g := NewTsGenerator(0)
g := base.NewTsGenerator(0)
ts := g.GenerateTs()
vid := proto.Vid(123)
bid := proto.BlobID(456)
@ -54,7 +55,7 @@ func TestEncodeDecodeDelMsgKey(t *testing.T) {
})
t.Run("EncodeRawDelMsgKey", func(t *testing.T) {
g := NewTsGenerator(0)
g := base.NewTsGenerator(0)
ts := g.GenerateTs()
vid := proto.Vid(365)
bid := proto.BlobID(45542013)
@ -77,7 +78,7 @@ func TestEncodeDecodeDelMsgKey(t *testing.T) {
})
t.Run("CompositeOrdering", func(t *testing.T) {
g := NewTsGenerator(0)
g := base.NewTsGenerator(0)
ts1 := g.GenerateTs()
key1 := EncodeDelMsgKey(ts1, 200, 50, [][]byte{[]byte("part1"), []byte("z")})
time.Sleep(time.Millisecond)
@ -100,7 +101,7 @@ func Benchmark(b *testing.B) {
require.Nil(b, err)
defer store.Close()
g := NewTsGenerator(0)
g := base.NewTsGenerator(0)
b.ResetTimer()
b.Run("Generate message key and store", func(b *testing.B) {
ts := g.GenerateTs()

3
blobstore/shardnode/blobdeleter/shard_reader.go
View File

@ -21,7 +21,6 @@ import (
snapi "github.com/cubefs/cubefs/blobstore/api/shardnode"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/blobstore/shardnode/base"
snproto "github.com/cubefs/cubefs/blobstore/shardnode/proto"
"github.com/cubefs/cubefs/blobstore/shardnode/storage"
"github.com/cubefs/cubefs/blobstore/util/log"
@ -121,7 +120,7 @@ func (r *shardListReader) listFromStorage(ctx context.Context, protectDuration t
}
// if nextMarker's timeUnix if protected, set protected
ts, _, _, _, err := base.DecodeDelMsgKey(r.nextMarker, r.ShardingSubRangeCount())
ts, _, _, _, err := DecodeDelMsgKey(r.nextMarker, r.ShardingSubRangeCount())
if err != nil {
span.Errorf("decode nextMarker[%+v] failed, err: %v", r.nextMarker, err)
return nil, err

16
blobstore/shardnode/blobdeleter/shard_reader_test.go
View File

@ -142,7 +142,7 @@ func TestShardListReader_ListFromStorage_Success(t *testing.T) {
g := base.NewTsGenerator(0)
tagNum := 2
ts := g.GenerateTs()
id, _ := base.EncodeRawDelMsgKey(ts, proto.Vid(1), proto.BlobID(100), tagNum)
id, _ := EncodeRawDelMsgKey(ts, proto.Vid(1), proto.BlobID(100), tagNum)
// mock item
oldTime := time.Now().Add(-2 * time.Hour).Unix()
item := snapi.Item{
@ -178,9 +178,9 @@ func TestShardListReader_ListFromStorage_WithProtectedMessages(t *testing.T) {
g := base.NewTsGenerator(0)
tagNum := 2
ts1 := g.GenerateTs()
id1, _ := base.EncodeRawDelMsgKey(ts1, proto.Vid(1), proto.BlobID(100), tagNum)
id1, _ := EncodeRawDelMsgKey(ts1, proto.Vid(1), proto.BlobID(100), tagNum)
ts2 := g.GenerateTs()
id2, _ := base.EncodeRawDelMsgKey(ts2, proto.Vid(1), proto.BlobID(101), tagNum)
id2, _ := EncodeRawDelMsgKey(ts2, proto.Vid(1), proto.BlobID(101), tagNum)
oldTime := time.Now().Add(-2 * time.Hour).Unix()
now := time.Now().Unix()
@ -210,7 +210,7 @@ func TestShardListReader_ListFromStorage_WithProtectedMessages(t *testing.T) {
}
ts := base.NewTs(time.Now().Unix())
protectedNextMarker, _ := base.EncodeRawDelMsgKey(ts, proto.Vid(1), proto.BlobID(102), tagNum)
protectedNextMarker, _ := EncodeRawDelMsgKey(ts, proto.Vid(1), proto.BlobID(102), tagNum)
handler := mock.NewMockSpaceShardHandler(ctr(t))
handler.EXPECT().GetRouteVersion().Return(proto.RouteVersion(1))
@ -262,8 +262,8 @@ func TestShardListReader_IsProtected(t *testing.T) {
// nextMarker > 0, and is unprotected
ts := base.NewTs(time.Now().Add(-2 * time.Hour).Unix())
key, _ := base.EncodeRawDelMsgKey(ts, vid, bid, tagNum)
_ts, _, _, _, err := base.DecodeDelMsgKey(key, tagNum)
key, _ := EncodeRawDelMsgKey(ts, vid, bid, tagNum)
_ts, _, _, _, err := DecodeDelMsgKey(key, tagNum)
require.NoError(t, err)
reader.setProtected(_ts.TimeUnix())
@ -271,8 +271,8 @@ func TestShardListReader_IsProtected(t *testing.T) {
// nextMarker > 0, and is protected
ts = base.NewTs(time.Now().Unix())
key, _ = base.EncodeRawDelMsgKey(ts, vid, bid, tagNum)
_ts, _, _, _, err = base.DecodeDelMsgKey(key, tagNum)
key, _ = EncodeRawDelMsgKey(ts, vid, bid, tagNum)
_ts, _, _, _, err = DecodeDelMsgKey(key, tagNum)
require.NoError(t, err)
reader.setProtected(_ts.TimeUnix())