Add support of changing raft peers via HTTP api

consts/context_key.go

@@ -23,7 +23,7 @@ const (
 	ContextKeyStore        = "_context_key_storage"
 	ContextKeyCluster      = "_context_key_cluster"
 	ContextKeyClusterShard = "_context_key_cluster_shard"
-	ContextKeyHost         = "_context_key_host"
+	ContextKeyRaftNode     = "_context_key_raft_node"
 )
 
 const (

server/api/handler.go

@@ -20,13 +20,16 @@
 
 package api
 
-import "github.com/apache/kvrocks-controller/store"
+import (
+	"github.com/apache/kvrocks-controller/store"
+)
 
 type Handler struct {
 	Namespace *NamespaceHandler
 	Cluster   *ClusterHandler
 	Shard     *ShardHandler
 	Node      *NodeHandler
+	Raft      *RaftHandler
 }
 
 func NewHandler(s *store.ClusterStore) *Handler {
@@ -35,5 +38,6 @@ func NewHandler(s *store.ClusterStore) *Handler {
 		Cluster:   &ClusterHandler{s: s},
 		Shard:     &ShardHandler{s: s},
 		Node:      &NodeHandler{s: s},
+		Raft:      &RaftHandler{},
 	}
 }

server/api/raft.go

@@ -0,0 +1,71 @@
+package api
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/apache/kvrocks-controller/consts"
+	"github.com/apache/kvrocks-controller/logger"
+	"github.com/apache/kvrocks-controller/server/helper"
+	"github.com/apache/kvrocks-controller/store/engine/raft"
+	"go.uber.org/zap"
+
+	"github.com/gin-gonic/gin"
+)
+
+const (
+	OperationAdd    = "add"
+	OperationRemove = "remove"
+)
+
+type RaftHandler struct{}
+
+type MemberRequest struct {
+	ID        uint64 `json:"id" validate:"required,gt=0"`
+	Operation string `json:"operation" validate:"required"`
+	Peer      string `json:"peer"`
+}
+
+func (r *MemberRequest) validate() error {
+	r.Operation = strings.ToLower(r.Operation)
+	if r.Operation != OperationAdd && r.Operation != OperationRemove {
+		return fmt.Errorf("operation must be one of [%s]",
+			strings.Join([]string{OperationAdd, OperationRemove}, ","))
+	}
+	if r.Operation == OperationAdd && r.Peer == "" {
+		return fmt.Errorf("peer should NOT be empty")
+	}
+	return nil
+}
+
+func (handler *RaftHandler) ListPeers(c *gin.Context) {
+	raftNode, _ := c.MustGet(consts.ContextKeyRaftNode).(*raft.Node)
+	peers := raftNode.ListPeers()
+	helper.ResponseOK(c, gin.H{"peers": peers})
+}
+
+func (handler *RaftHandler) UpdatePeer(c *gin.Context) {
+	var req MemberRequest
+	if err := c.BindJSON(&req); err != nil {
+		helper.ResponseBadRequest(c, err)
+		return
+	}
+	if err := req.validate(); err != nil {
+		helper.ResponseBadRequest(c, err)
+		return
+	}
+
+	raftNode, _ := c.MustGet(consts.ContextKeyRaftNode).(*raft.Node)
+	var err error
+	if req.Operation == OperationAdd {
+		err = raftNode.AddPeer(c, req.ID, req.Peer)
+	} else {
+		err = raftNode.RemovePeer(c, req.ID)
+	}
+	if err != nil {
+		helper.ResponseError(c, err)
+	} else {
+		logger.Get().With(zap.Any("request", req)).Info("Update peer success")
+		helper.ResponseOK(c, nil)
+	}
+}

server/middleware/middleware.go

@@ -26,6 +26,7 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/apache/kvrocks-controller/store/engine/raft"
 	"github.com/gin-gonic/gin"
 	"github.com/prometheus/client_golang/prometheus"
 
@@ -66,7 +67,11 @@ func RedirectIfNotLeader(c *gin.Context) {
 		c.Abort()
 		return
 	}
-	if !storage.IsLeader() {
+
+	_, isRaftMode := storage.GetEngine().(*raft.Node)
+	// Raft engine will forward the request to the leader nod under the hood,
+	// so we don't need to do the redirect.
+	if !storage.IsLeader() && !isRaftMode {
 		if !c.GetBool(consts.HeaderIsRedirect) {
 			c.Set(consts.HeaderIsRedirect, true)
 			peerAddr := helper.ExtractAddrFromSessionID(storage.Leader())
@@ -131,3 +136,15 @@ func RequiredClusterShard(c *gin.Context) {
 	c.Set(consts.ContextKeyClusterShard, shard)
 	c.Next()
 }
+
+func RequiredRaftEngine(c *gin.Context) {
+	storage, _ := c.MustGet(consts.ContextKeyStore).(*store.ClusterStore)
+	raftNode, ok := storage.GetEngine().(*raft.Node)
+	if !ok {
+		helper.ResponseBadRequest(c, errors.New("raft engine is not enabled"))
+		c.Abort()
+		return
+	}
+	c.Set(consts.ContextKeyRaftNode, raftNode)
+	c.Next()
+}

server/route.go

@@ -47,6 +47,13 @@ func (srv *Server) initHandlers() {
 
 	apiV1 := engine.Group("/api/v1/")
 	{
+		raftAPI := apiV1.Group("raft")
+		{
+			raftAPI.Use(middleware.RequiredRaftEngine)
+			raftAPI.POST("/peers", handler.Raft.UpdatePeer)
+			raftAPI.GET("/peers", handler.Raft.ListPeers)
+		}
+
 		namespaces := apiV1.Group("namespaces")
 		{
 			namespaces.GET("", handler.Namespace.List)

store/engine/engine.go

@@ -17,6 +17,7 @@
  * under the License.
  *
  */
+
 package engine
 
 import (

store/engine/raft/node.go

@@ -115,11 +115,12 @@ func (n *Node) Addr() string {
 	return n.addr
 }
 
-func (n *Node) Peers() []string {
-	peers := make([]string, 0)
+func (n *Node) ListPeers() map[uint64]string {
+	peers := make(map[uint64]string, 0)
 	n.peers.Range(func(key, value interface{}) bool {
+		id, _ := key.(uint64)
 		peer, _ := value.(string)
-		peers = append(peers, peer)
+		peers[id] = peer
 		return true
 	})
 	return peers
@@ -466,12 +467,13 @@ func (n *Node) applyEntry(entry raftpb.Entry) error {
 			}
 		case raftpb.ConfChangeRemoveNode:
 			n.peers.Delete(cc.NodeID)
-			n.transport.RemovePeer(types.ID(cc.NodeID))
 			if cc.NodeID == n.config.ID {
 				n.Close()
 				n.logger.Info("Node removed from the cluster")
 				return nil
 			}
+			n.logger.Info("Remove the peer", zap.Uint64("node_id", cc.NodeID))
+			n.transport.RemovePeer(types.ID(cc.NodeID))
 		case raftpb.ConfChangeUpdateNode:
 			n.transport.UpdatePeer(types.ID(cc.NodeID), []string{string(cc.Context)})
 			if _, ok := n.peers.Load(cc.NodeID); ok {

store/engine/raft/node_test.go

@@ -68,7 +68,11 @@ func (c *TestCluster) createNode(peers []string) (*Node, error) {
 		HeartbeatSeconds: 1,
 		ElectionSeconds:  2,
 	})
-	return node, err
+	if err != nil {
+		return nil, err
+	}
+	c.nodes = append(c.nodes, node)
+	return node, nil
 }
 
 func (c *TestCluster) AddNode(ctx context.Context, nodeID uint64, peer string) error {
@@ -79,16 +83,18 @@ func (c *TestCluster) AddNode(ctx context.Context, nodeID uint64, peer string) e
 }
 
 func (c *TestCluster) RemoveNode(ctx context.Context, nodeID uint64) error {
+	var node *Node
 	for i, n := range c.nodes {
 		if n.config.ID == nodeID {
+			node = n
 			c.nodes = append(c.nodes[:i], c.nodes[i+1:]...)
 			break
 		}
 	}
-	if len(c.nodes) == 0 {
+	if len(c.nodes) == 0 || node == nil {
 		return nil
 	}
-	return c.nodes[0].RemovePeer(ctx, nodeID)
+	return node.RemovePeer(ctx, nodeID)
 }
 
 func (c *TestCluster) SetSnapshotThreshold(threshold uint64) {
@@ -234,13 +240,13 @@ func TestCluster_AddRemovePeer(t *testing.T) {
 			got, _ := n1.Get(ctx, "foo")
 			return string(got) == "bar-1"
 		}, 1*time.Second, 100*time.Millisecond)
-		require.Len(t, n1.Peers(), 4)
+		require.Len(t, n1.ListPeers(), 4)
 	})
 
 	t.Run("remove a peer node", func(t *testing.T) {
 		cluster.RemoveNode(ctx, 4)
 		require.Eventually(t, func() bool {
-			return len(n1.Peers()) == 3
+			return len(n1.ListPeers()) == 3
 		}, 10*time.Second, 100*time.Millisecond)
 	})
 }

store/store.go

@@ -310,6 +310,10 @@ func (s *ClusterStore) EmitEvent(event EventPayload) {
 	s.eventNotifyCh <- event
 }
 
+func (s *ClusterStore) GetEngine() engine.Engine {
+	return s.e
+}
+
 func (s *ClusterStore) LeaderChange() <-chan bool {
 	return s.e.LeaderChange()
 }