Add support of changing raft peers via HTTP api (#232)

Author: git-hulk

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 (

controller/controller.go

@@ -104,6 +104,7 @@ func (c *Controller) resume(ctx context.Context) error {
 		}
 		for _, cluster := range clusters {
 			c.addCluster(ns, cluster)
+			logger.Get().Debug("Resume the cluster", zap.String("namespace", ns), zap.String("cluster", cluster))
 		}
 	}
 	return nil

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,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+package api
+
+import (
+	"errors"
+	"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"
+
+	"github.com/gin-gonic/gin"
+	"go.uber.org/zap"
+)
+
+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)
+	helper.ResponseOK(c, gin.H{
+		"leader": raftNode.GetRaftLead(),
+		"peers":  raftNode.ListPeers(),
+	})
+}
+
+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)
+	peers := raftNode.ListPeers()
+
+	var err error
+	if req.Operation == OperationAdd {
+		for _, peer := range peers {
+			if peer == req.Peer {
+				helper.ResponseError(c, fmt.Errorf("peer '%s' already exists", req.Peer))
+				return
+			}
+		}
+		err = raftNode.AddPeer(c, req.ID, req.Peer)
+	} else {
+		if _, ok := peers[req.ID]; !ok {
+			helper.ResponseBadRequest(c, errors.New("peer not exists"))
+			return
+		}
+		if len(peers) == 1 {
+			helper.ResponseBadRequest(c, errors.New("can't remove the last peer"))
+			return
+		}
+		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 node 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/config.go

@@ -20,15 +20,23 @@
 
 package raft
 
-import "errors"
+import (
+	"errors"
+	"strings"
+)
+
+const (
+	ClusterStateNew      = "new"
+	ClusterStateExisting = "existing"
+)
 
 type Config struct {
 	// ID is the identity of the local raft. ID cannot be 0.
 	ID uint64 `yaml:"id"`
 	// DataDir is the directory to store the raft data which includes snapshot and WALs.
 	DataDir string `yaml:"data_dir"`
-	// Join should be set to true if the node is joining an existing cluster.
-	Join bool `yaml:"join"`
+	// ClusterState is the state of the cluster, can be one of "new" and "existing".
+	ClusterState string `yaml:"cluster_state"`
 	// Peers is the list of raft peers.
 	Peers []string `yaml:"peers"`
 	// HeartbeatSeconds is the interval to send heartbeat message. Default is 2 seconds.
@@ -47,10 +55,15 @@ func (c *Config) validate() error {
 	if c.ID > uint64(len(c.Peers)) {
 		return errors.New("ID cannot be greater than the number of peers")
 	}
+	clusterState := strings.ToLower(c.ClusterState)
+	if clusterState != ClusterStateNew && clusterState != ClusterStateExisting {
+		return errors.New("cluster state must be one of [new, existing]")
+	}
 	return nil
 }
 
 func (c *Config) init() {
+	c.ClusterState = ClusterStateNew
 	if c.DataDir == "" {
 		c.DataDir = "."
 	}

store/engine/raft/config_test.go

@@ -28,6 +28,7 @@ import (
 
 func TestConfig_Validate(t *testing.T) {
 	c := &Config{}
+	c.init()
 
 	// missing ID
 	require.ErrorContains(t, c.validate(), "ID cannot be 0")
@@ -40,6 +41,12 @@ func TestConfig_Validate(t *testing.T) {
 	// ID greater than the number of peers
 	c.ID = 2
 	require.ErrorContains(t, c.validate(), "ID cannot be greater than the number of peers")
+
+	c.ID = 1
+	c.ClusterState = "invalid"
+	require.ErrorContains(t, c.validate(), "cluster state must be one of [new, existing]")
+	c.ClusterState = ClusterStateNew
+	require.NoError(t, c.validate())
 }
 
 func TestConfig_Init(t *testing.T) {

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)
 	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
@@ -165,7 +166,7 @@ func (n *Node) run() error {
 	n.snapshotIndex = snapshot.Metadata.Index
 	n.confState = snapshot.Metadata.ConfState
 
-	if n.config.Join || walExists {
+	if n.config.ClusterState == ClusterStateExisting || walExists {
 		n.raftNode = raft.RestartNode(raftConfig)
 	} else {
 		n.raftNode = raft.StartNode(raftConfig, peers)
@@ -174,6 +175,7 @@ func (n *Node) run() error {
 	if err := n.runTransport(); err != nil {
 		return err
 	}
+	n.watchLeaderChange()
 	return n.runRaftMessages()
 }
 
@@ -224,10 +226,33 @@ func (n *Node) runTransport() error {
 	return nil
 }
 
+func (n *Node) watchLeaderChange() {
+	n.wg.Add(1)
+	go func() {
+		defer n.wg.Done()
+
+		ticker := time.NewTicker(time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-n.shutdown:
+				return
+			case <-ticker.C:
+				lead := n.GetRaftLead()
+				if lead != n.leader {
+					n.leader = lead
+					n.leaderChanged <- true
+					n.logger.Info("Found leader changed", zap.Uint64("leader", lead))
+				}
+			}
+		}
+	}()
+}
+
 func (n *Node) runRaftMessages() error {
 	n.wg.Add(1)
 	go func() {
-		ticker := time.NewTicker(100 * time.Millisecond)
+		ticker := time.NewTicker(time.Second)
 		defer func() {
 			ticker.Stop()
 			n.wg.Done()
@@ -252,9 +277,7 @@ func (n *Node) runRaftMessages() error {
 				if err := n.applySnapshot(rd.Snapshot); err != nil {
 					n.logger.Error("Failed to apply snapshot", zap.Error(err))
 				}
-				if len(rd.Entries) > 0 {
-					_ = n.dataStore.raftStorage.Append(rd.Entries)
-				}
+				_ = n.dataStore.raftStorage.Append(rd.Entries)
 
 				for _, msg := range rd.Messages {
 					if msg.Type == raftpb.MsgApp {
@@ -465,13 +488,14 @@ func (n *Node) applyEntry(entry raftpb.Entry) error {
 				n.peers.Store(cc.NodeID, string(cc.Context))
 			}
 		case raftpb.ConfChangeRemoveNode:
-			n.peers.Delete(cc.NodeID)
-			n.transport.RemovePeer(types.ID(cc.NodeID))
 			if cc.NodeID == n.config.ID {
+				n.logger.Info("I have been removed from the cluster, will shutdown")
 				n.Close()
-				n.logger.Info("Node removed from the cluster")
 				return nil
 			}
+			n.transport.RemovePeer(types.ID(cc.NodeID))
+			n.peers.Delete(cc.NodeID)
+			n.logger.Info("Remove the peer", zap.Uint64("node_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 {