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(shardnode): encode kv storage data key with shardID

Browse Source 
with #22357426

Signed-off-by: xiejian <xiejian3@oppo.com>
This commit is contained in:
xiejian 2024-10-25 17:40:40 +08:00 committed by slasher
parent 173a1cd157
commit 24edbfdbb1
6 changed files with 27 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
blobstore/common/errors/errors.go
View File

@ -162,6 +162,7 @@ var errCodeMap = map[int]string{
CodeShardNoLeader: "shardnode:shard has no leader",
CodeIllegalSlices: "shardnode:alloc with illegal slices",
CodeBlobAlreadyExists: "shardnode:blob already exists",
CodeShardConflicts: "shardnode:shard conflicts",
}
// HTTPError make rpc.HTTPError

2
blobstore/common/errors/shardnode.go
View File

@ -25,6 +25,7 @@ const (
CodeIllegalSlices = 1008
CodeBlobAlreadyExists = 1009
CodeUnsupport = 1010
CodeShardConflicts = 1011
)
// 10xx
@ -39,4 +40,5 @@ var (
ErrIllegalSlices = newError(CodeIllegalSlices, "illegal slices")
ErrBlobAlreadyExists = newError(CodeBlobAlreadyExists, "blob already exists")
ErrShardNodeUnsupport = newError(CodeUnsupport, "unsupport shard node")
ErrShardConflicts = newError(CodeShardConflicts, "shard conflicts")
)

13
blobstore/shardnode/storage/disk.go
View File

@ -132,6 +132,7 @@ func OpenDisk(ctx context.Context, cfg DiskConfig) (*Disk, error) {
disk.diskInfo = diskInfo
disk.store = store
disk.shardsMu.shards = make(map[proto.Suid]*shard)
disk.shardsMu.shardCheck = make(map[proto.ShardID]struct{})
// disk will be gc by finalizer
runtime.SetFinalizer(disk, func(disk *Disk) {
@ -181,7 +182,8 @@ type Disk struct {
shardsMu struct {
sync.RWMutex
// shard id as the map key
shards map[proto.Suid]*shard
shards map[proto.Suid]*shard
shardCheck map[proto.ShardID]struct{}
}
raftManager raft.Manager
store *store.Store
@ -243,6 +245,7 @@ func (d *Disk) Load(ctx context.Context) error {
d.shardsMu.Lock()
d.shardsMu.shards[suid] = shard
d.shardsMu.shardCheck[suid.ShardID()] = struct{}{}
d.shardsMu.Unlock()
shard.Start()
@ -269,6 +272,11 @@ func (d *Disk) AddShard(ctx context.Context, suid proto.Suid,
return nil
}
if _, ok := d.shardsMu.shardCheck[suid.ShardID()]; ok {
span.Errorf("shard[%d] already exist", suid.ShardID())
return apierr.ErrShardConflicts
}
shardUnits := make([]clustermgr.ShardUnit, len(nodes))
for i := range nodes {
shardUnits[i] = clustermgr.ShardUnit{
@ -305,6 +313,7 @@ func (d *Disk) AddShard(ctx context.Context, suid proto.Suid,
}
d.shardsMu.shards[suid] = shard
d.shardsMu.shardCheck[suid.ShardID()] = struct{}{}
shard.Start()
return nil
}
@ -375,6 +384,7 @@ func (d *Disk) DeleteShard(ctx context.Context, suid proto.Suid, version proto.R
}
delete(d.shardsMu.shards, suid)
delete(d.shardsMu.shardCheck, suid.ShardID())
return nil
}
@ -484,6 +494,7 @@ func (d *Disk) SetBroken() bool {
func (d *Disk) ResetShards() {
d.lock.Lock()
d.shardsMu.shards = make(map[proto.Suid]*shard)
d.shardsMu.shardCheck = make(map[proto.ShardID]struct{})
d.lock.Unlock()
}

14
blobstore/shardnode/storage/proto.go
View File

@ -52,7 +52,7 @@ type (
// todo: merge these encode and decode function into shard?
func shardDataPrefixSize() int {
return len(shardDataPrefix) + 8
return len(shardDataPrefix) + 4
}
func shardInfoPrefixSize() int {
@ -91,19 +91,19 @@ func decodeShardInfoPrefix(raw []byte) proto.Suid {
return proto.Suid(binary.BigEndian.Uint64(raw[prefixSize:]))
}
func encodeShardDataPrefix(suid proto.Suid, raw []byte) {
func encodeShardDataPrefix(shardID proto.ShardID, raw []byte) {
copy(raw, shardDataPrefix)
binary.BigEndian.PutUint64(raw[len(shardDataPrefix):], uint64(suid))
binary.BigEndian.PutUint32(raw[len(shardDataPrefix):], uint32(shardID))
}
func encodeShardItemPrefix(suid proto.Suid, raw []byte) {
func encodeShardItemPrefix(shardID proto.ShardID, raw []byte) {
shardPrefixSize := shardDataPrefixSize()
encodeShardDataPrefix(suid, raw)
encodeShardDataPrefix(shardID, raw)
copy(raw[shardPrefixSize:], itemSuffix)
}
func encodeShardDataMaxPrefix(suid proto.Suid, raw []byte) {
func encodeShardDataMaxPrefix(shardID proto.ShardID, raw []byte) {
shardPrefixSize := shardDataPrefixSize()
encodeShardDataPrefix(suid, raw)
encodeShardDataPrefix(shardID, raw)
copy(raw[shardPrefixSize:], maxSuffix)
}

6
blobstore/shardnode/storage/shard.go
View File

@ -880,7 +880,7 @@ type shardKeysGenerator struct {
func (s *shardKeysGenerator) encodeItemKey(key []byte) []byte {
shardItemPrefixSize := shardItemPrefixSize()
newKey := make([]byte, shardItemPrefixSize+len(key))
encodeShardItemPrefix(s.suid, newKey)
encodeShardItemPrefix(s.suid.ShardID(), newKey)
copy(newKey[shardItemPrefixSize:], key)
return newKey
}
@ -904,7 +904,7 @@ func (s *shardKeysGenerator) encodeShardInfoKey() []byte {
// it can be used for listing all shard's data or delete shard's data
func (s *shardKeysGenerator) encodeShardDataPrefix() []byte {
key := make([]byte, shardDataPrefixSize())
encodeShardDataPrefix(s.suid, key)
encodeShardDataPrefix(s.suid.ShardID(), key)
return key
}
@ -912,7 +912,7 @@ func (s *shardKeysGenerator) encodeShardDataPrefix() []byte {
// it can be used for delete shard's data
func (s *shardKeysGenerator) encodeShardDataMaxPrefix() []byte {
key := make([]byte, shardMaxPrefixSize())
encodeShardDataMaxPrefix(s.suid, key)
encodeShardDataMaxPrefix(s.suid.ShardID(), key)
return key
}

2
blobstore/shardnode/storage/shard_sm.go
View File

@ -196,6 +196,8 @@ func (s *shardSM) ApplySnapshot(header raft.RaftSnapshotHeader, snap raft.Snapsh
// clear all data with shard prefix
batch := kvStore.NewWriteBatch()
batch.DeleteRange(dataCF, s.shardKeys.encodeShardDataPrefix(), s.shardKeys.encodeShardDataMaxPrefix())
// flush
if err := kvStore.Write(ctx, batch, nil); err != nil {
return err
}