feat(shardnode): complete shardserver transport and support raft stat api

with #22357426

Signed-off-by: xiejian <xiejian3@oppo.com>

blobstore/api/shardnode/shard.go

@@ -17,9 +17,24 @@ package shardnode
 import (
 	"context"
 
+	"github.com/cubefs/cubefs/blobstore/common/raft"
+	"github.com/cubefs/cubefs/blobstore/common/sharding"
+
 	"github.com/cubefs/cubefs/blobstore/api/clustermgr"
+	"github.com/cubefs/cubefs/blobstore/common/proto"
 )
 
+type ShardStats struct {
+	Suid         proto.Suid
+	AppliedIndex uint64
+	LeaderIdx    uint32
+	RouteVersion proto.RouteVersion
+	Range        sharding.Range
+	Units        []clustermgr.ShardUnit
+	Learner      bool
+	RaftStat     raft.Stat
+}
+
 func (c *Client) AddShard(ctx context.Context, host string, args AddShardArgs) error {
 	return nil
 }
@@ -31,3 +46,7 @@ func (c *Client) UpdateShard(ctx context.Context, host string, args UpdateShardA
 func (c *Client) GetShardUintInfo(ctx context.Context, host string, args GetShardArgs) (ret clustermgr.ShardUnitInfo, err error) {
 	return clustermgr.ShardUnitInfo{}, nil
 }
+
+func (c *Client) GetShardStats(ctx context.Context, host string, diskID proto.DiskID, suid proto.Suid) (ret ShardStats, err error) {
+	return ShardStats{}, nil
+}

blobstore/shardnode/base/mock_transport.go

@@ -156,6 +156,22 @@ func (mr *MockTransportMockRecorder) GetNode(ctx, nodeID interface{}) *gomock.Ca
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNode", reflect.TypeOf((*MockTransport)(nil).GetNode), ctx, nodeID)
 }
 
+// GetRouteUpdate mocks base method.
+func (m *MockTransport) GetRouteUpdate(ctx context.Context, routeVersion proto.RouteVersion) (proto.RouteVersion, []clustermgr.CatalogChangeItem, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetRouteUpdate", ctx, routeVersion)
+	ret0, _ := ret[0].(proto.RouteVersion)
+	ret1, _ := ret[1].([]clustermgr.CatalogChangeItem)
+	ret2, _ := ret[2].(error)
+	return ret0, ret1, ret2
+}
+
+// GetRouteUpdate indicates an expected call of GetRouteUpdate.
+func (mr *MockTransportMockRecorder) GetRouteUpdate(ctx, routeVersion interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRouteUpdate", reflect.TypeOf((*MockTransport)(nil).GetRouteUpdate), ctx, routeVersion)
+}
+
 // GetSpace mocks base method.
 func (m *MockTransport) GetSpace(ctx context.Context, sid proto.SpaceID) (*clustermgr.Space, error) {
 	m.ctrl.T.Helper()

blobstore/shardnode/base/transport.go

@@ -30,6 +30,7 @@ type (
 	Transport interface {
 		GetConfig(ctx context.Context, key string) (string, error)
 		ShardReport(ctx context.Context, reports []clustermgr.ShardUnitInfo) ([]clustermgr.ShardTask, error)
+		GetRouteUpdate(ctx context.Context, routeVersion proto.RouteVersion) (proto.RouteVersion, []clustermgr.CatalogChangeItem, error)
 		NodeTransport
 		SpaceTransport
 		AllocVolTransport
@@ -124,8 +125,7 @@ func (t *transport) AllocDiskID(ctx context.Context) (proto.DiskID, error) {
 }
 
 func (t *transport) RegisterDisk(ctx context.Context, disk *clustermgr.ShardNodeDiskInfo) error {
-	return nil
-	// return t.cmClient.AddDisk(ctx, Disk)
+	return t.cmClient.AddShardNodeDisk(ctx, disk)
 }
 
 func (t *transport) SetDiskBroken(ctx context.Context, diskID proto.DiskID) error {
@@ -148,62 +148,87 @@ func (t *transport) GetMyself() *clustermgr.ShardNodeInfo {
 }
 
 func (t *transport) GetSpace(ctx context.Context, sid proto.SpaceID) (*clustermgr.Space, error) {
-	// todo: add singleflight group to avoid too much get space request go through to master
-	/*resp, err := t.cmClient.GetSpace(ctx, &proto.GetSpaceRequest{
-		SpaceID: sid,
+	v, err, _ := t.singleRun.Do(strconv.Itoa(int(sid)), func() (interface{}, error) {
+		space, err := t.cmClient.GetSpaceByID(ctx, &clustermgr.GetSpaceByIDArgs{SpaceID: sid})
+		if err != nil {
+			return nil, err
+		}
+		return space, nil
 	})
 	if err != nil {
-		return proto.SpaceMeta{}, err
+		return nil, err
 	}
-
-	return resp.Info, nil*/
-	return nil, nil
+	return v.(*clustermgr.Space), nil
 }
 
 func (t *transport) GetAllSpaces(ctx context.Context) ([]clustermgr.Space, error) {
-	return nil, nil
+	args := &clustermgr.ListSpaceArgs{Count: uint32(10000)}
+	args.Count = uint32(10000)
+
+	spaces := make([]clustermgr.Space, 0)
+	for {
+		ret, err := t.cmClient.ListSpace(ctx, args)
+		if err != nil {
+			return nil, err
+		}
+		if len(ret.Spaces) < 1 {
+			break
+		}
+		for _, s := range ret.Spaces {
+			spaces = append(spaces, *s)
+		}
+		args.Marker = ret.Marker
+	}
+	return spaces, nil
 }
 
-/*func (t *Transport) GetRouteUpdate(ctx context.Context, routeVersion uint64) (uint64, []proto.CatalogChangeItem, error) {
-	resp, err := t.cmClient.GetCatalogChanges(ctx, &proto.GetCatalogChangesRequest{RouteVersion: routeVersion, NodeID: t.myself.ID})
+func (t *transport) GetRouteUpdate(ctx context.Context, routeVersion proto.RouteVersion) (proto.RouteVersion, []clustermgr.CatalogChangeItem, error) {
+	resp, err := t.cmClient.GetCatalogChanges(ctx, &clustermgr.GetCatalogChangesArgs{RouteVersion: routeVersion, NodeID: t.myself.NodeID})
 	if err != nil {
 		return 0, nil, err
 	}
-
 	return resp.RouteVersion, resp.Items, nil
-}*/
+}
 
 func (t *transport) ShardReport(ctx context.Context, reports []clustermgr.ShardUnitInfo) ([]clustermgr.ShardTask, error) {
-	/*resp, err := t.cmClient.Report(ctx, &proto.ReportRequest{
-		NodeID: t.myself.ID,
-		Infos:  infos,
+	resp, err := t.cmClient.ReportShard(ctx, &clustermgr.ShardReportArgs{
+		ShardReport: clustermgr.ShardReport{
+			Shards: reports,
+		},
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	return resp.Tasks, err*/
-	return nil, nil
+	return resp, err
 }
 
 func (t *transport) ListDisks(ctx context.Context) ([]clustermgr.ShardNodeDiskInfo, error) {
-	// todo: change api to shard node api
-	/*resp, err := t.cmClient.ListDisk(ctx, &clustermgr.ListOptionArgs{
+	args := &clustermgr.ListOptionArgs{
+		Idc:   t.myself.Idc,
+		Rack:  t.myself.Rack,
 		Host:  t.myself.Host,
 		Count: 10000,
-	})
-	if err != nil {
-		return nil, err
 	}
-
-	return resp.Disks, nil*/
-	return nil, nil
+	disks := make([]clustermgr.ShardNodeDiskInfo, 0)
+	for {
+		ret, err := t.cmClient.ListShardNodeDisk(ctx, args)
+		if err != nil {
+			return nil, err
+		}
+		if len(ret.Disks) < 1 {
+			break
+		}
+		for _, d := range ret.Disks {
+			disks = append(disks, *d)
+		}
+		args.Marker = ret.Marker
+	}
+	return disks, nil
 }
 
 func (t *transport) HeartbeatDisks(ctx context.Context, disks []clustermgr.ShardNodeDiskHeartbeatInfo) error {
-	//_, err := t.cmClient.HeartbeatDisk(ctx, disks)
-	//return err
-	return nil
+	return t.cmClient.HeartbeatShardNodeDisk(ctx, disks)
 }
 
 func (t *transport) NodeID() proto.NodeID {

blobstore/shardnode/shard.go

@@ -47,13 +47,16 @@ func (s *service) UpdateShard(ctx context.Context, req *shardnode.UpdateShardArg
 	return disk.UpdateShard(ctx, req.Suid, req.ShardUpdateType, req.Unit)
 }
 
-func (s *service) GetShardInfo(ctx context.Context, diskID proto.DiskID, suid proto.Suid) (ret clustermgr.ShardUnitInfo, err error) {
+func (s *service) GetShardUintInfo(ctx context.Context, diskID proto.DiskID, suid proto.Suid) (ret clustermgr.ShardUnitInfo, err error) {
 	shard, err := s.GetShard(diskID, suid)
 	if err != nil {
 		return
 	}
 
-	shardStat := shard.Stats()
+	shardStat, err := shard.Stats()
+	if err != nil {
+		return
+	}
 
 	return clustermgr.ShardUnitInfo{
 		Suid:         suid,
@@ -65,6 +68,20 @@ func (s *service) GetShardInfo(ctx context.Context, diskID proto.DiskID, suid pr
 	}, nil
 }
 
+func (s *service) GetShardStats(ctx context.Context, diskID proto.DiskID, suid proto.Suid) (ret shardnode.ShardStats, err error) {
+	shard, err := s.GetShard(diskID, suid)
+	if err != nil {
+		return
+	}
+
+	shardStat, err := shard.Stats()
+	if err != nil {
+		return
+	}
+
+	return shardStat, nil
+}
+
 func (s *service) GetShard(diskID proto.DiskID, suid proto.Suid) (storage.ShardHandler, error) {
 	disk, err := s.getDisk(diskID)
 	if err != nil {
@@ -117,7 +134,11 @@ func (s *service) loop(ctx context.Context) {
 			disks := s.getAllDisks()
 			for _, disk := range disks {
 				disk.RangeShard(func(shard storage.ShardHandler) bool {
-					stats := shard.Stats()
+					stats, err := shard.Stats()
+					if err != nil {
+						span.Errorf("get shard stat err: %s", err)
+						return false
+					}
 					shardReports = append(shardReports, clustermgr.ShardUnitInfo{
 						Suid:         stats.Suid,
 						DiskID:       disk.DiskID(),

blobstore/shardnode/storage/disk_mock.go

@@ -17,6 +17,7 @@ package storage
 import (
 	"context"
 	"fmt"
+	"math"
 	"math/rand"
 	"os"
 	"path"
@@ -40,9 +41,9 @@ var (
 	_, ctx = trace.StartSpanFromContext(context.Background(), "Testing")
 )
 
-func tempPath() (string, func()) {
+func tempPath(tb testing.TB) (string, func()) {
 	rand.Seed(time.Now().Unix())
-	tmp := path.Join(os.TempDir(), fmt.Sprintf("shardserver_disk_%d", rand.Int31n(10000)+10000))
+	tmp := path.Join(os.TempDir(), fmt.Sprintf("shardserver_disk_%s_%d", tb.Name(), rand.Int63n(math.MaxInt)))
 	return tmp, func() { os.RemoveAll(tmp) }
 }
 
@@ -52,7 +53,7 @@ type MockDisk struct {
 }
 
 func NewMockDisk(tb testing.TB) (*MockDisk, func()) {
-	diskPath, pathClean := tempPath()
+	diskPath, pathClean := tempPath(tb)
 	var cfg DiskConfig
 	cfg.DiskPath = diskPath
 	cfg.StoreConfig.KVOption.CreateIfMissing = true

blobstore/shardnode/storage/mock_shard.go

@@ -395,11 +395,12 @@ func (mr *MockSpaceShardHandlerMockRecorder) ListItem(ctx, h, prefix, id, count
 }
 
 // Stats mocks base method.
-func (m *MockSpaceShardHandler) Stats() ShardStats {
+func (m *MockSpaceShardHandler) Stats() (shardnode.ShardStats, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Stats")
-	ret0, _ := ret[0].(ShardStats)
-	return ret0
+	ret0, _ := ret[0].(shardnode.ShardStats)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
 }
 
 // Stats indicates an expected call of Stats.

blobstore/shardnode/storage/raft_impl.go

@@ -98,7 +98,14 @@ type raftStorage struct {
 }
 
 func (w *raftStorage) Get(key []byte) (raft.ValGetter, error) {
-	return w.kvStore.Get(context.TODO(), raftWalCF, key, nil)
+	vg, err := w.kvStore.Get(context.TODO(), raftWalCF, key, nil)
+	if err != nil {
+		if err == kvstore.ErrNotFound {
+			err = raft.ErrNotFound
+		}
+		return nil, err
+	}
+	return vg, err
 }
 
 func (w *raftStorage) Iter(prefix []byte) raft.Iterator {

blobstore/shardnode/storage/shard.go

@@ -64,7 +64,7 @@ type (
 		ShardItemHandler
 		GetRouteVersion() proto.RouteVersion
 		Checkpoint(ctx context.Context) error
-		Stats() ShardStats
+		Stats() (shardnode.ShardStats, error)
 	}
 
 	OpHeader struct {
@@ -77,16 +77,6 @@ type (
 		TruncateWalLogInterval uint64                     `json:"truncate_wal_log_interval"`
 	}
 
-	ShardStats struct {
-		Suid         proto.Suid
-		AppliedIndex uint64
-		LeaderIdx    uint32
-		RouteVersion proto.RouteVersion
-		Range        sharding.Range
-		Units        []shardUnitInfo
-		Learner      bool
-	}
-
 	shardConfig struct {
 		*ShardBaseConfig
 		suid         proto.Suid
@@ -426,9 +416,9 @@ func (s *shard) UpdateShardRouteVersion(version proto.RouteVersion) {
 	s.shardInfoMu.Unlock()
 }
 
-func (s *shard) Stats() ShardStats {
+func (s *shard) Stats() (shardnode.ShardStats, error) {
 	if err := s.shardState.prepRWCheck(); err != nil {
-		return ShardStats{}
+		return shardnode.ShardStats{}, err
 	}
 	defer s.shardState.prepRWCheckDone()
 
@@ -446,7 +436,12 @@ func (s *shard) Stats() ShardStats {
 	learner := !(s.shardInfoMu.leader == s.diskID)
 	s.shardInfoMu.RUnlock()
 
-	return ShardStats{
+	raftStat, err := s.raftGroup.Stat()
+	if err != nil {
+		return shardnode.ShardStats{}, err
+	}
+
+	return shardnode.ShardStats{
 		Suid:         s.suid,
 		AppliedIndex: appliedIndex,
 		LeaderIdx:    leaderIdx,
@@ -454,7 +449,8 @@ func (s *shard) Stats() ShardStats {
 		Range:        rg,
 		Units:        units,
 		Learner:      learner,
-	}
+		RaftStat:     *raftStat,
+	}, nil
 }
 
 // Checkpoint do checkpoint job with raft group
@@ -553,9 +549,9 @@ func (s *shard) DeleteShard(ctx context.Context, nodeHost string) error {
 	if err != nil {
 		return errors.Info(err, "raft stat failed")
 	}
-	if len(stat.Nodes) > 1 {
-		for _, diskID := range stat.Nodes {
-			if uint64(s.diskID) == diskID {
+	if len(stat.Peers) > 1 {
+		for _, pr := range stat.Peers {
+			if uint64(s.diskID) == pr.NodeID {
 				if err := s.raftGroup.MemberChange(ctx, &raft.Member{
 					NodeID: uint64(s.diskID),
 					Host:   nodeHost,

blobstore/shardnode/storage/shard_test.go

@@ -176,3 +176,13 @@ func TestServerShard_Item(t *testing.T) {
 	require.Equal(t, apierr.ErrShardNodeNotLeader, mockShard.shard.Delete(ctx, newShardOpHeader, oldProtoItem.ID))
 	mockShard.shard.diskID = 1
 }
+
+func TestServerShard_Stats(t *testing.T) {
+	mockShard, shardClean := newMockShard(t)
+	defer shardClean()
+
+	mockShard.mockRaftGroup.EXPECT().Stat().Return(&raft.Stat{}, nil)
+
+	_, err := mockShard.shard.Stats()
+	require.Nil(t, err)
+}