feat(raft): group Stat() api support peer detail

with #22357426

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

blobstore/common/raft/group.go

@@ -111,14 +111,14 @@ func (g *group) LeaderTransfer(ctx context.Context, nodeID uint64) error {
 		return err
 	}
 	nodeFound := false
-	for _, id := range stat.Nodes {
-		if id == nodeID {
+	for _, pr := range stat.Peers {
+		if pr.NodeID == nodeID {
 			nodeFound = true
 			break
 		}
 	}
 	if !nodeFound {
-		return fmt.Errorf("node[%d] not found in node list[%+v]", nodeID, stat.Nodes)
+		return fmt.Errorf("node[%d] not found in node list[%+v]", nodeID, stat.Peers)
 	}
 
 	(*internalGroupProcessor)(g).WithRaftRawNodeLocked(func(rn *raft.RawNode) error {
@@ -210,9 +210,26 @@ func (g *group) MemberChange(ctx context.Context, m *Member) error {
 
 func (g *group) Stat() (*Stat, error) {
 	raftStatus := g.raftStatusLocked()
-	peers := make([]uint64, 0, len(raftStatus.Progress))
-	for i := range raftStatus.Progress {
-		peers = append(peers, i)
+	peers := make([]Peer, 0, len(raftStatus.Progress))
+	for id, pr := range raftStatus.Progress {
+		var host string
+		if m, ok := g.storage.GetMember(id); ok {
+			host = m.Host
+		}
+		peer := Peer{
+			NodeID:          id,
+			Host:            host,
+			Match:           pr.Match,
+			Next:            pr.Next,
+			RaftState:       pr.State.String(),
+			Paused:          pr.IsPaused(),
+			PendingSnapshot: pr.PendingSnapshot,
+			RecentActive:    pr.RecentActive,
+			IsLearner:       pr.IsLearner,
+			InflightFull:    pr.Inflights.Full(),
+			InflightCount:   int64(pr.Inflights.Count()),
+		}
+		peers = append(peers, peer)
 	}
 	return &Stat{
 		ID:             g.id,
@@ -225,7 +242,7 @@ func (g *group) Stat() (*Stat, error) {
 		Applied:        g.storage.AppliedIndex(),
 		RaftApplied:    raftStatus.Applied,
 		LeadTransferee: raftStatus.LeadTransferee,
-		Nodes:          peers,
+		Peers:          peers,
 	}, nil
 }
 

blobstore/common/raft/manager_test.go

@@ -292,7 +292,7 @@ func TestManager_GroupInMultiServer(t *testing.T) {
 		require.NoError(t, err)
 
 		stat, _ := groups[leaderIndex].Stat()
-		require.Equal(t, len(allNodes)-1, len(stat.Nodes))
+		require.Equal(t, len(allNodes)-1, len(stat.Peers))
 
 		err = groups[leaderIndex].MemberChange(ctx, &allNodes[followerIndex])
 		require.NoError(t, err)

blobstore/common/raft/proto.go

@@ -16,12 +16,15 @@ package raft
 
 import (
 	"context"
+	"errors"
 
 	"go.etcd.io/etcd/raft/v3/raftpb"
 )
 
 const reqIDKey = "req-id"
 
+var ErrNotFound = errors.New("key not found")
+
 type (
 	StateMachine interface {
 		// Apply will notify the state machine to apply all proposal data
@@ -33,6 +36,7 @@ type (
 		ApplySnapshot(s Snapshot) error
 	}
 	Storage interface {
+		// Get should return ErrNotFound when key not exits
 		Get(key []byte) (ValGetter, error)
 		Iter(prefix []byte) Iterator
 		NewBatch() Batch
@@ -81,20 +85,6 @@ type (
 )
 
 type (
-	Stat struct {
-		ID             uint64   `json:"id"`
-		NodeID         uint64   `json:"node_id"`
-		Term           uint64   `json:"term"`
-		Vote           uint64   `json:"vote"`
-		Commit         uint64   `json:"commit"`
-		Leader         uint64   `json:"leader"`
-		RaftState      string   `json:"raftState"`
-		Applied        uint64   `json:"applied"`
-		RaftApplied    uint64   `json:"raftApplied"`
-		LeadTransferee uint64   `json:"transferee"`
-		Nodes          []uint64 `json:"nodes"`
-	}
-
 	ProposalResponse struct {
 		Data interface{}
 	}

blobstore/common/raft/proto_mock_test.go

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

blobstore/common/raft/raft.proto

@@ -111,4 +111,32 @@ message RaftSnapshotResponse {
     }
     Status status = 1;
     string message = 2;
+}
+
+message Peer {
+    uint64 node_id = 1 [(gogoproto.customname) = "NodeID"];
+    string host = 2;
+    uint64 match = 3;
+    uint64 next = 4;
+    string raft_state = 5;
+    bool paused = 6;
+    uint64 pending_snapshot = 7;
+    bool recent_active = 8;
+    bool is_learner = 9;
+    bool inflight_full = 10;
+    int64 inflight_count = 11;
+}
+
+message Stat {
+    uint64 id = 1 [(gogoproto.customname) = "ID"];
+    uint64 node_id = 2 [(gogoproto.customname) = "NodeID"];
+    uint64 term = 3;
+    uint64 vote = 4;
+    uint64 commit = 5;
+    uint64 leader = 6;
+    string raft_state = 7;
+    uint64 applied = 8;
+    uint64 raft_applied = 9;
+    uint64 Lead_transferee = 10;
+    repeated Peer peers = 11 [(gogoproto.nullable) = false];
 }

blobstore/common/raft/service.pb.go

@@ -1,17 +1,3 @@
-// Copyright 2022 The CubeFS Authors.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
-// implied. See the License for the specific language governing
-// permissions and limitations under the License.
-
 // Code generated by protoc-gen-gogo. DO NOT EDIT.
 // source: service.proto
 

blobstore/common/raft/storage.go

@@ -24,7 +24,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/cubefs/cubefs/blobstore/common/kvstorev2"
 	"github.com/google/uuid"
 	"go.etcd.io/etcd/raft/v3"
 	"go.etcd.io/etcd/raft/v3/raftpb"
@@ -59,7 +58,7 @@ func newStorage(cfg storageConfig) (*storage, error) {
 	}
 
 	value, err := cfg.raw.Get(encodeHardStateKey(cfg.id))
-	if err != nil && err != kvstore.ErrNotFound {
+	if err != nil && err != ErrNotFound {
 		return nil, err
 	}
 	if value != nil {
@@ -426,6 +425,13 @@ func (s *storage) MemberChange(member *Member) {
 	s.updateConfState()
 }
 
+func (s *storage) GetMember(id uint64) (Member, bool) {
+	s.membersMu.RLock()
+	defer s.membersMu.RUnlock()
+	m, hit := s.membersMu.members[id]
+	return m, hit
+}
+
 // Clear remove all raft log and hard state of the group
 func (s *storage) Clear() error {
 	batch := s.rawStg.NewBatch()

blobstore/common/raft/storage_test.go

@@ -284,6 +284,23 @@ func TestStorage_DecodeHardState(t *testing.T) {
 	t.Log("hs: ", hs, err)
 }
 
+func TestStorage_GetMember(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	s := initStorage(t, ctrl)
+
+	m, hit := s.GetMember(1)
+	require.True(t, hit)
+	require.Equal(t, uint64(1), m.NodeID)
+
+	m, hit = s.GetMember(2)
+	require.True(t, hit)
+	require.Equal(t, uint64(2), m.NodeID)
+
+	_, hit = s.GetMember(3)
+	require.False(t, hit)
+}
+
 func initStorage(t *testing.T, ctrl *gomock.Controller) *storage {
 	mockStorage := NewMockStorage(ctrl)
 	mockHardState := raftpb.HardState{

blobstore/common/raft/testraft/main.go

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

blobstore/common/raftserver/util.go

@@ -39,24 +39,24 @@ type Peer struct {
 	Next            uint64 `json:"next"`
 	State           string `json:"state"`
 	Paused          bool   `json:"paused"`
-	PendingSnapshot uint64 `json:"pendingSnapshot"`
+	PendingSnapshot uint64 `json:"pending_snapshot"`
 	RecentActive    bool   `json:"active"`
-	IsLearner       bool   `json:"isLearner"`
-	InflightFull    bool   `json:"isInflightFull"`
-	InflightCount   int    `json:"inflightCount"`
+	IsLearner       bool   `json:"is_learner"`
+	InflightFull    bool   `json:"is_inflight_full"`
+	InflightCount   int    `json:"inflight_count"`
 }
 
 type Status struct {
-	Id             uint64 `json:"nodeId"`
+	Id             uint64 `json:"node_id"`
 	Term           uint64 `json:"term"`
 	Vote           uint64 `json:"vote"`
 	Commit         uint64 `json:"commit"`
 	Leader         uint64 `json:"leader"`
-	RaftState      string `json:"raftState"`
+	RaftState      string `json:"raft_state"`
 	Applied        uint64 `json:"applied"`
-	RaftApplied    uint64 `json:"raftApplied"`
+	RaftApplied    uint64 `json:"raft_applied"`
 	LeadTransferee uint64 `json:"transferee"`
-	ApplyingLength int    `json:"applyingLength"`
+	ApplyingLength int    `json:"applying_length"`
 	Peers          []Peer `json:"peers"`
 }