feat(api): expose cluster membership service

cmd/kv/main.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/Aetherance/kv/engine/config"
 	"github.com/Aetherance/kv/engine/storage/raft_storage"
+	"github.com/Aetherance/kv/proto/pkg/clusterpb"
 	"github.com/Aetherance/kv/proto/pkg/kvpb"
 	rspb "github.com/Aetherance/kv/proto/pkg/raft_serverpb"
 	"github.com/Aetherance/kv/server"
@@ -46,6 +47,7 @@ func main() {
 	grpcServer := grpc.NewServer()
 	kvpb.RegisterKvServer(grpcServer, server.NewServer(rs))
 	rspb.RegisterRaftServiceServer(grpcServer, rs)
+	clusterpb.RegisterClusterServer(grpcServer, rs)
 
 	lis, err := net.Listen("tcp", addr)
 	if err != nil {

engine/storage/raft_storage/cluster_api.go

@@ -0,0 +1,154 @@
+package raft_storage
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/Aetherance/kv/proto/pkg/clusterpb"
+	"github.com/Aetherance/kv/proto/pkg/raftpb"
+	"github.com/Aetherance/kv/raft"
+)
+
+func (rs *RaftStorage) MemberList(ctx context.Context, _ *clusterpb.MemberListRequest) (*clusterpb.MemberListResponse, error) {
+	clusterStatus, err := rs.status(ctx)
+	if err != nil {
+		return nil, membershipRPCError(err)
+	}
+	return memberListResponse(clusterStatus), nil
+}
+
+func (rs *RaftStorage) MemberAdd(ctx context.Context, request *clusterpb.MemberAddRequest) (*clusterpb.MemberAddResponse, error) {
+	if request == nil || !request.Learner {
+		return nil, status.Error(codes.InvalidArgument, "members must first be added as learners")
+	}
+	member := &clusterpb.Member{Id: request.Id, RaftAddress: strings.TrimSpace(request.RaftAddress)}
+	if err := rs.proposeMembership(ctx, raftpb.ConfChangeType_AddLearnerNode, member); err != nil {
+		return nil, membershipRPCError(err)
+	}
+	list, err := rs.MemberList(ctx, &clusterpb.MemberListRequest{})
+	return &clusterpb.MemberAddResponse{Cluster: list}, err
+}
+
+func (rs *RaftStorage) MemberPromote(ctx context.Context, request *clusterpb.MemberPromoteRequest) (*clusterpb.MemberPromoteResponse, error) {
+	if request == nil || request.Id == 0 {
+		return nil, status.Error(codes.InvalidArgument, "member ID must be non-zero")
+	}
+	clusterStatus, err := rs.status(ctx)
+	if err != nil {
+		return nil, membershipRPCError(err)
+	}
+	member, _ := findMember(clusterStatus.metadata, request.Id)
+	if member == nil {
+		return nil, membershipRPCError(errMemberNotFound)
+	}
+	if err := rs.proposeMembership(ctx, raftpb.ConfChangeType_AddNode, member); err != nil {
+		return nil, membershipRPCError(err)
+	}
+	list, err := rs.MemberList(ctx, &clusterpb.MemberListRequest{})
+	return &clusterpb.MemberPromoteResponse{Cluster: list}, err
+}
+
+func (rs *RaftStorage) MemberRemove(ctx context.Context, request *clusterpb.MemberRemoveRequest) (*clusterpb.MemberRemoveResponse, error) {
+	if request == nil || request.Id == 0 {
+		return nil, status.Error(codes.InvalidArgument, "member ID must be non-zero")
+	}
+	clusterStatus, err := rs.status(ctx)
+	if err != nil {
+		return nil, membershipRPCError(err)
+	}
+	member, _ := findMember(clusterStatus.metadata, request.Id)
+	if member == nil {
+		member = &clusterpb.Member{Id: request.Id}
+	}
+	if err := rs.proposeMembership(ctx, raftpb.ConfChangeType_RemoveNode, member); err != nil {
+		return nil, membershipRPCError(err)
+	}
+	list, err := rs.MemberList(ctx, &clusterpb.MemberListRequest{})
+	return &clusterpb.MemberRemoveResponse{Cluster: list}, err
+}
+
+func (rs *RaftStorage) MemberUpdate(ctx context.Context, request *clusterpb.MemberUpdateRequest) (*clusterpb.MemberUpdateResponse, error) {
+	if request == nil {
+		return nil, status.Error(codes.InvalidArgument, "request is required")
+	}
+	member := &clusterpb.Member{Id: request.Id, RaftAddress: strings.TrimSpace(request.RaftAddress)}
+	if err := rs.proposeMembership(ctx, raftpb.ConfChangeType_UpdateNode, member); err != nil {
+		return nil, membershipRPCError(err)
+	}
+	list, err := rs.MemberList(ctx, &clusterpb.MemberListRequest{})
+	return &clusterpb.MemberUpdateResponse{Cluster: list}, err
+}
+
+func (rs *RaftStorage) MemberStatus(ctx context.Context, request *clusterpb.MemberStatusRequest) (*clusterpb.MemberStatusResponse, error) {
+	if request == nil || request.Id == 0 {
+		return nil, status.Error(codes.InvalidArgument, "member ID must be non-zero")
+	}
+	clusterStatus, err := rs.status(ctx)
+	if err != nil {
+		return nil, membershipRPCError(err)
+	}
+	member, _ := findMember(clusterStatus.metadata, request.Id)
+	if member == nil {
+		return nil, membershipRPCError(errMemberNotFound)
+	}
+	return &clusterpb.MemberStatusResponse{
+		LeaderId:    clusterStatus.leaderID,
+		CommitIndex: clusterStatus.commitIndex,
+		Member:      memberInfo(clusterStatus, member),
+	}, nil
+}
+
+func memberListResponse(clusterStatus *clusterStatus) *clusterpb.MemberListResponse {
+	response := &clusterpb.MemberListResponse{
+		ClusterId:    clusterStatus.metadata.ClusterId,
+		LeaderId:     clusterStatus.leaderID,
+		ConfRevision: clusterStatus.metadata.ConfRevision,
+		Members:      make([]*clusterpb.MemberInfo, 0, len(clusterStatus.metadata.Members)),
+	}
+	for _, member := range clusterStatus.metadata.Members {
+		response.Members = append(response.Members, memberInfo(clusterStatus, member))
+	}
+	return response
+}
+
+func memberInfo(clusterStatus *clusterStatus, member *clusterpb.Member) *clusterpb.MemberInfo {
+	progress, exists := clusterStatus.progress[member.Id]
+	active := exists && progress.RecentActive
+	if member.Id == clusterStatus.leaderID {
+		active = true
+	}
+	return &clusterpb.MemberInfo{
+		Member:     proto.Clone(member).(*clusterpb.Member),
+		Role:       memberRole(clusterStatus.confState, member.Id),
+		Active:     active,
+		MatchIndex: progress.Match,
+	}
+}
+
+func membershipRPCError(err error) error {
+	if err == nil {
+		return nil
+	}
+	var notLeader *NotLeaderError
+	switch {
+	case errors.As(err, &notLeader):
+		return status.Error(codes.FailedPrecondition, err.Error())
+	case errors.Is(err, errMemberNotFound):
+		return status.Error(codes.NotFound, err.Error())
+	case errors.Is(err, errMemberRemoved), errors.Is(err, errMemberAlreadyExists), errors.Is(err, errAddressAlreadyExists):
+		return status.Error(codes.AlreadyExists, err.Error())
+	case errors.Is(err, errLearnerNotReady), errors.Is(err, errLastVoter), errors.Is(err, errTooManyLearners), errors.Is(err, errUnsafeReconfiguration), errors.Is(err, raft.ErrConfChangePending):
+		return status.Error(codes.FailedPrecondition, err.Error())
+	case errors.Is(err, context.Canceled):
+		return status.Error(codes.Canceled, err.Error())
+	case errors.Is(err, context.DeadlineExceeded):
+		return status.Error(codes.DeadlineExceeded, err.Error())
+	default:
+		return status.Error(codes.InvalidArgument, err.Error())
+	}
+}

engine/storage/raft_storage/cluster_api_test.go

@@ -0,0 +1,112 @@
+package raft_storage
+
+import (
+	"context"
+	"testing"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/Aetherance/kv/proto/pkg/clusterpb"
+)
+
+func TestClusterMembershipAPI(t *testing.T) {
+	store := startMembershipStore(t)
+	ctx := context.Background()
+
+	list, err := store.MemberList(ctx, &clusterpb.MemberListRequest{})
+	if err != nil {
+		t.Fatalf("list initial members: %v", err)
+	}
+	assertMemberList(t, list, 42, 1, 0, []memberExpectation{{
+		id: 1, address: "127.0.0.1:1", role: clusterpb.MemberRole_MemberRoleVoter, active: true,
+	}})
+
+	if _, err := store.MemberAdd(ctx, &clusterpb.MemberAddRequest{
+		Id: 2, RaftAddress: "127.0.0.1:2",
+	}); status.Code(err) != codes.InvalidArgument {
+		t.Fatalf("direct voter add code = %s, want InvalidArgument", status.Code(err))
+	}
+	added, err := store.MemberAdd(ctx, &clusterpb.MemberAddRequest{
+		Id: 2, RaftAddress: " 127.0.0.1:2 ", Learner: true,
+	})
+	if err != nil {
+		t.Fatalf("add learner: %v", err)
+	}
+	assertMemberList(t, added.Cluster, 42, 1, 1, []memberExpectation{
+		{id: 1, address: "127.0.0.1:1", role: clusterpb.MemberRole_MemberRoleVoter, active: true},
+		{id: 2, address: "127.0.0.1:2", role: clusterpb.MemberRole_MemberRoleLearner},
+	})
+
+	memberStatus, err := store.MemberStatus(ctx, &clusterpb.MemberStatusRequest{Id: 2})
+	if err != nil {
+		t.Fatalf("learner status: %v", err)
+	}
+	if memberStatus.LeaderId != 1 || memberStatus.Member.Member.Id != 2 || memberStatus.Member.Role != clusterpb.MemberRole_MemberRoleLearner {
+		t.Fatalf("unexpected learner status: %v", memberStatus)
+	}
+
+	updated, err := store.MemberUpdate(ctx, &clusterpb.MemberUpdateRequest{Id: 2, RaftAddress: "127.0.0.1:22"})
+	if err != nil {
+		t.Fatalf("update learner: %v", err)
+	}
+	assertMemberList(t, updated.Cluster, 42, 1, 2, []memberExpectation{
+		{id: 1, address: "127.0.0.1:1", role: clusterpb.MemberRole_MemberRoleVoter, active: true},
+		{id: 2, address: "127.0.0.1:22", role: clusterpb.MemberRole_MemberRoleLearner},
+	})
+
+	removed, err := store.MemberRemove(ctx, &clusterpb.MemberRemoveRequest{Id: 2})
+	if err != nil {
+		t.Fatalf("remove learner: %v", err)
+	}
+	assertMemberList(t, removed.Cluster, 42, 1, 3, []memberExpectation{{
+		id: 1, address: "127.0.0.1:1", role: clusterpb.MemberRole_MemberRoleVoter, active: true,
+	}})
+	removed, err = store.MemberRemove(ctx, &clusterpb.MemberRemoveRequest{Id: 2})
+	if err != nil || removed.Cluster.ConfRevision != 3 {
+		t.Fatalf("idempotent remove = %v, %v", removed, err)
+	}
+	if _, err := store.MemberAdd(ctx, &clusterpb.MemberAddRequest{
+		Id: 2, RaftAddress: "127.0.0.1:22", Learner: true,
+	}); status.Code(err) != codes.AlreadyExists {
+		t.Fatalf("re-add removed member code = %s, want AlreadyExists", status.Code(err))
+	}
+}
+
+func TestClusterMembershipAPIErrors(t *testing.T) {
+	store := startMembershipStore(t)
+	ctx := context.Background()
+
+	if _, err := store.MemberStatus(ctx, &clusterpb.MemberStatusRequest{Id: 99}); status.Code(err) != codes.NotFound {
+		t.Fatalf("unknown member status code = %s, want NotFound", status.Code(err))
+	}
+	if _, err := store.MemberRemove(ctx, &clusterpb.MemberRemoveRequest{Id: 99}); status.Code(err) != codes.NotFound {
+		t.Fatalf("unknown member remove code = %s, want NotFound", status.Code(err))
+	}
+	if _, err := store.MemberRemove(ctx, &clusterpb.MemberRemoveRequest{Id: 1}); status.Code(err) != codes.FailedPrecondition {
+		t.Fatalf("last voter remove code = %s, want FailedPrecondition", status.Code(err))
+	}
+}
+
+type memberExpectation struct {
+	id      uint64
+	address string
+	role    clusterpb.MemberRole
+	active  bool
+}
+
+func assertMemberList(t *testing.T, response *clusterpb.MemberListResponse, clusterID, leaderID, revision uint64, members []memberExpectation) {
+	t.Helper()
+	if response == nil || response.ClusterId != clusterID || response.LeaderId != leaderID || response.ConfRevision != revision {
+		t.Fatalf("unexpected cluster response: %v", response)
+	}
+	if len(response.Members) != len(members) {
+		t.Fatalf("members = %v, want %v", response.Members, members)
+	}
+	for index, want := range members {
+		got := response.Members[index]
+		if got.Member.Id != want.id || got.Member.RaftAddress != want.address || got.Role != want.role || got.Active != want.active {
+			t.Fatalf("member %d = %v, want %+v", index, got, want)
+		}
+	}
+}

engine/storage/raft_storage/membership.go

@@ -138,6 +138,17 @@ func roleOf(state *raftpb.ConfState, id uint64) raftMemberRole {
 	return memberRoleUnknown
 }
 
+func memberRole(state *raftpb.ConfState, id uint64) clusterpb.MemberRole {
+	switch roleOf(state, id) {
+	case memberRoleVoter:
+		return clusterpb.MemberRole_MemberRoleVoter
+	case memberRoleLearner:
+		return clusterpb.MemberRole_MemberRoleLearner
+	default:
+		return clusterpb.MemberRole_MemberRoleUnknown
+	}
+}
+
 func (rs *RaftStorage) validateConfChange(change *raftpb.ConfChange) error {
 	var context clusterpb.ConfChangeContext
 	if change == nil || proto.Unmarshal(change.GetContext(), &context) != nil || context.Member == nil || context.Member.Id != change.NodeId {

engine/storage/raft_storage/raft.go

@@ -33,19 +33,31 @@ const (
 	opStep raftOperation = iota
 	opPropose
 	opProposeConfChange
+	opStatus
 )
 
 type raftEvent struct {
 	op      raftOperation
 	message *raftpb.Message
 	data    []byte
 	change  *raftpb.ConfChange
+	status  *clusterStatus
 	done    chan error
 }
 
+type clusterStatus struct {
+	metadata    *clusterpb.ClusterMetadata
+	confState   *raftpb.ConfState
+	leaderID    uint64
+	commitIndex uint64
+	progress    map[uint64]raft.Progress
+}
+
 func (rs *RaftStorage) run(ctx context.Context) {
 	err := rs.runLoop(ctx)
+	rs.lifecycleMu.Lock()
 	rs.runErr = err
+	rs.lifecycleMu.Unlock()
 
 	pendingErr := err
 	if pendingErr == nil {
@@ -107,6 +119,16 @@ func (rs *RaftStorage) handleEvent(event raftEvent) (error, bool) {
 		if err := rs.node.ProposeConfChange(event.change); err != nil {
 			return err, false
 		}
+	case opStatus:
+		if event.status == nil {
+			return errors.New("raft storage: nil status target"), false
+		}
+		event.status.metadata = cloneClusterMetadata(rs.state.cluster)
+		event.status.confState = rs.node.ConfState()
+		event.status.leaderID = rs.node.LeaderID()
+		event.status.commitIndex = rs.node.CommitIndex()
+		event.status.progress = rs.node.GetProgress()
+		return nil, false
 	default:
 		return errors.New("raft storage: unknown operation"), false
 	}
@@ -194,6 +216,14 @@ func (rs *RaftStorage) proposeConfChangeData(ctx context.Context, change *raftpb
 	return rs.submit(ctx, raftEvent{op: opProposeConfChange, change: change})
 }
 
+func (rs *RaftStorage) status(ctx context.Context) (*clusterStatus, error) {
+	status := new(clusterStatus)
+	if err := rs.submit(ctx, raftEvent{op: opStatus, status: status}); err != nil {
+		return nil, err
+	}
+	return status, nil
+}
+
 func (rs *RaftStorage) submit(ctx context.Context, event raftEvent) error {
 	if rs.inbox == nil || rs.done == nil {
 		return errStopped
@@ -218,6 +248,8 @@ func (rs *RaftStorage) submit(ctx context.Context, event raftEvent) error {
 }
 
 func (rs *RaftStorage) stoppedError() error {
+	rs.lifecycleMu.Lock()
+	defer rs.lifecycleMu.Unlock()
 	if rs.runErr != nil {
 		return rs.runErr
 	}