Repository: nats-io/nats-streaming-server
Pull request #1090 (2020-09-17 23:25:04)

[ADDED] Clustering: ability to add/remove nodes at runtime
If option `allow_add_remove_node` in `streaming{cluster{}}` block
is enabled (or command line `--cluster_allow_add_remove_node`)
then the leader will listen to `_STAN.raft.<cluster id>.node.remove`
(and `...node.add`) to which an user can send a simple NATS request
with the value being the name of the node to add/remove.

If successful, the leader will return `+OK`, if not `-ERR` with
the error string explaining why it failed.

Note that removing a non-existent node will still return `+OK`.

If the leader receives a request to remove itself, it will work
but the process will then exit. It can be added back by sending
the request to the `..node.add` subject.

Note that this updates the RAFT configuration in the RAFT log but
does NOT change the configuration file. The admin will be responsible
to change the configuration on file so that when nodes are restarted
they reflect the new configuration (list of peers).

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>


Patch:
diff --git a/nats-streaming-server.go b/nats-streaming-server.go
index 1691ac6..8dde3b5 100644
--- a/nats-streaming-server.go
+++ b/nats-streaming-server.go
@@ -52,18 +52,19 @@ Streaming Server Options:
           --encrypt <bool>               Specify if server should use encryption at rest
           --encryption_cipher <string>   Cipher to use for encryption. Currently support AES and CHAHA (ChaChaPoly). Defaults to AES
           --encryption_key <string>      Encryption Key. It is recommended to specify it through the NATS_STREAMING_ENCRYPTION_KEY environment variable instead
-    
+
 Streaming Server Clustering Options:
-    --clustered <bool>                   Run the server in a clustered configuration (default: false)
-    --cluster_node_id <string>           ID of the node within the cluster if there is no stored ID (default: random UUID)
-    --cluster_bootstrap <bool>           Bootstrap the cluster if there is no existing state by electing self as leader (default: false)
-    --cluster_peers <string, ...>        Comma separated list of cluster peer node IDs to bootstrap cluster state
-    --cluster_log_path <string>          Directory to store log replication data
-    --cluster_log_cache_size <int>       Number of log entries to cache in memory to reduce disk IO (default: 512)
-    --cluster_log_snapshots <int>        Number of log snapshots to retain (default: 2)
-    --cluster_trailing_logs <int>        Number of log entries to leave after a snapshot and compaction
-    --cluster_sync <bool>                Do a file sync after every write to the replication log and message store
-    --cluster_raft_logging <bool>        Enable logging from the Raft library (disabled by default)
+    --clustered <bool>                     Run the server in a clustered configuration (default: false)
+    --cluster_node_id <string>             ID of the node within the cluster if there is no stored ID (default: random UUID)
+    --cluster_bootstrap <bool>             Bootstrap the cluster if there is no existing state by electing self as leader (default: false)
+    --cluster_peers <string, ...>          Comma separated list of cluster peer node IDs to bootstrap cluster state
+    --cluster_log_path <string>            Directory to store log replication data
+    --cluster_log_cache_size <int>         Number of log entries to cache in memory to reduce disk IO (default: 512)
+    --cluster_log_snapshots <int>          Number of log snapshots to retain (default: 2)
+    --cluster_trailing_logs <int>          Number of log entries to leave after a snapshot and compaction
+    --cluster_sync <bool>                  Do a file sync after every write to the replication log and message store
+    --cluster_raft_logging <bool>          Enable logging from the Raft library (disabled by default)
+    --cluster_allow_add_remove_node <bool> Enable the ability to send NATS requests to the leader to add/remove cluster nodes
 
 Streaming Server File Store Options:
     --file_compact_enabled <bool>        Enable file compaction
diff --git a/server/clustering.go b/server/clustering.go
index 48b27da..3b75f7e 100644
--- a/server/clustering.go
+++ b/server/clustering.go
@@ -15,6 +15,7 @@ package server
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -28,6 +29,11 @@ import (
 	"github.com/nats-io/nats.go"
 )
 
+const (
+	addClusterNodeSubj    = defaultRaftPrefix + ".%s.node.add"
+	removeClusterNodeSubj = defaultRaftPrefix + ".%s.node.remove"
+)
+
 const (
 	defaultJoinRaftGroupTimeout = time.Second
 	defaultRaftHBTimeout        = 2 * time.Second
@@ -64,6 +70,12 @@ type ClusteringOptions struct {
 	Sync         bool     // Do a file sync after every write to the Raft log and message store.
 	RaftLogging  bool     // Enable logging of Raft library (disabled by default since really verbose).
 
+	// If this is enabled, the leader of the cluster will listen to add/remove
+	// requests on NATS subject "_STAN.raft.<cluster ID>.node.[add|remove]".
+	// Admin can/should limit permissions to send to this subject to prevent
+	// a user to inadvertently change the cluster configuration.
+	AllowAddRemoveNode bool
+
 	// When a node processes a snapshot (either on startup or if falling behind) and its is
 	// not in phase with the message store's state, it is required to reconcile its state
 	// with the current leader. If it is unable, the node will fail to start or exit.
@@ -672,3 +684,47 @@ func (r *raftFSM) lookupOrCreateChannel(name string, id uint64) (*channel, error
 	// Channel does exist or has been deleted. Create now with given ID.
 	return cs.createChannelLocked(s, name, id)
 }
+
+func (s *StanServer) processAddNode(m *nats.Msg) {
+	var err error
+	nodeID := string(m.Data)
+	if nodeID != "" {
+		addr := s.getClusteringPeerAddr(s.opts.ID, nodeID)
+		err = s.raft.AddVoter(raft.ServerID(nodeID), raft.ServerAddress(addr), 0, 0).Error()
+		if err == nil {
+			s.log.Noticef("Added node %q", nodeID)
+			m.Respond([]byte("+OK"))
+			return
+		}
+	} else {
+		err = errors.New("invalid node ID")
+	}
+	s.log.Errorf("Error adding node %q: %v", nodeID, err)
+	m.Respond([]byte(fmt.Sprintf("-ERR adding node %q: %v", nodeID, err)))
+}
+
+func (s *StanServer) processRemoveNode(m *nats.Msg) {
+	nodeID := string(m.Data)
+	err := s.raft.RemoveServer(raft.ServerID(nodeID), 0, 0).Error()
+	if err == nil {
+		s.log.Noticef("Removed node %q", nodeID)
+		m.Respond([]byte("+OK"))
+		if nodeID == s.opts.Clustering.NodeID {
+			s.nc.Flush()
+			// Wait that we step down...
+			timeout := time.Now().Add(5 * time.Second)
+			for time.Now().Before(timeout) {
+				if atomic.LoadInt64(&(s.raft.leader)) == 0 {
+					break
+				}
+			}
+			s.Shutdown()
+			if !runningInTests {
+				os.Exit(0)
+			}
+		}
+		return
+	}
+	s.log.Errorf("Error removing node %q: %v", nodeID, err)
+	m.Respond([]byte(fmt.Sprintf("-ERR removing node %q: %v", nodeID, err)))
+}
diff --git a/server/conf.go b/server/conf.go
index 0a9f2fa..512be0e 100644
--- a/server/conf.go
+++ b/server/conf.go
@@ -292,6 +292,11 @@ func parseCluster(itf interface{}, opts *Options) error {
 				return err
 			}
 			opts.Clustering.ProceedOnRestoreFailure = v.(bool)
+		case "allow_add_remove_node":
+			if err := checkType(k, reflect.Bool, v); err != nil {
+				return err
+			}
+			opts.Clustering.AllowAddRemoveNode = v.(bool)
 		case "raft_logging":
 			if err := checkType(k, reflect.Bool, v); err != nil {
 				return err
@@ -652,6 +657,7 @@ func ConfigureOptions(fs *flag.FlagSet, args []string, printVersion, printHelp,
 	fs.BoolVar(&sopts.Clustering.Sync, "cluster_sync", false, "stan.Clustering.Sync")
 	fs.BoolVar(&sopts.Clustering.RaftLogging, "cluster_raft_logging", false, "")
 	fs.BoolVar(&sopts.Clustering.ProceedOnRestoreFailure, "cluster_proceed_on_restore_failure", false, "")
+	fs.BoolVar(&sopts.Clustering.AllowAddRemoveNode, "cluster_allow_add_remove_node", false, "")
 	fs.StringVar(&sopts.SQLStoreOpts.Driver, "sql_driver", "", "SQL Driver")
 	fs.StringVar(&sopts.SQLStoreOpts.Source, "sql_source", "", "SQL Data Source")
 	defSQLOpts := stores.DefaultSQLStoreOptions()
diff --git a/server/server.go b/server/server.go
index bdcf140..6fa30b7 100644
--- a/server/server.go
+++ b/server/server.go
@@ -751,6 +751,8 @@ type StanServer struct {
 	subCloseSub *nats.Subscription
 	subUnsubSub *nats.Subscription
 	cliPingSub  *nats.Subscription
+	addNodeSub  *nats.Subscription
+	rmNodeSub   *nats.Subscription
 
 	// For sending responses to client PINGS. Used to be global but would
 	// cause races when running more than 1 server in a program or test.
@@ -2710,7 +2712,22 @@ func (s *StanServer) initInternalSubs(createPub bool) error {
 	}
 	// Receive PINGs from clients.
 	s.cliPingSub, err = s.createSub(s.info.Discovery+".pings", s.processClientPings, "client pings")
-	return err
+	if err != nil {
+		return err
+	}
+	if s.isClustered && s.opts.Clustering.AllowAddRemoveNode {
+		// Add cluster node requests
+		s.addNodeSub, err = s.createSub(fmt.Sprintf(addClusterNodeSubj, s.opts.ID), s.processAddNode, "add node")
+		if err != nil {
+			return err
+		}
+		// Remove cluster node requests
+		s.rmNodeSub, err = s.createSub(fmt.Sprintf(removeClusterNodeSubj, s.opts.ID), s.processRemoveNode, "remove node")
+		if err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 func (s *StanServer) unsubscribeInternalSubs() {
@@ -2746,6 +2763,14 @@ func (s *StanServer) unsubscribeInternalSubs() {
 		s.snapReqSub.Unsubscribe()
 		s.snapReqSub = nil
 	}
+	if s.addNodeSub != nil {
+		s.addNodeSub.Unsubscribe()
+		s.addNodeSub = nil
+	}
+	if s.rmNodeSub != nil {
+		s.rmNodeSub.Unsubscribe()
+		s.rmNodeSub = nil
+	}
 }
 
 func (s *StanServer) createSub(subj string, f nats.MsgHandler, errTxt string) (*nats.Subscription, error) {


Test patch:
diff --git a/server/clustering_test.go b/server/clustering_test.go
index 5729a4c..b573303 100644
--- a/server/clustering_test.go
+++ b/server/clustering_test.go
@@ -274,12 +274,14 @@ func shutdownServers(servers []*StanServer) {
 }
 
 func removeServer(servers []*StanServer, s *StanServer) []*StanServer {
-	for i, srv := range servers {
+	newservers := make([]*StanServer, 0, len(servers))
+	for _, srv := range servers {
 		if srv == s {
-			servers = append(servers[:i], servers[i+1:]...)
+			continue
 		}
+		newservers = append(newservers, srv)
 	}
-	return servers
+	return newservers
 }
 
 func assertMsg(t *testing.T, msg pb.MsgProto, expectedData []byte, expectedSeq uint64) {
@@ -7718,3 +7720,281 @@ func TestClusteringRedeliveryCount(t *testing.T) {
 		t.Fatalf("Timedout")
 	}
 }
+
+func testRemoveNode(t *testing.T, nc *nats.Conn, node string, timeoutExpected bool) {
+	t.Helper()
+	timeout := 2 * time.Second
+	if timeoutExpected {
+		timeout = 100 * time.Millisecond
+	}
+	resp, err := nc.Request(fmt.Sprintf(removeClusterNodeSubj, clusterName), []byte(node), timeout)
+	if timeoutExpected {
+		if err != nats.ErrTimeout {
+			t.Fatalf("Expected timeout, got %v", err)
+		}
+		return
+	}
+	if string(resp.Data) != "+OK" {
+		t.Fatalf("Removing node %q returned response error: %q", node, resp.Data)
+	}
+}
+
+func TestClusteringAddRemoveClusterNodes(t *testing.T) {
+	cleanupDatastore(t)
+	defer cleanupDatastore(t)
+	cleanupRaftLog(t)
+	defer cleanupRaftLog(t)
+
+	// For this test, use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// First create a cluster and make sure that if option is not
+	// enabled, the request will fail.
+	// Configure first server
+	s1sOpts := getTestDefaultOptsForClustering("a", true)
+	s1 := runServerWithOpts(t, s1sOpts, nil)
+	defer s1.Shutdown()
+
+	// Configure second server.
+	s2sOpts := getTestDefaultOptsForClustering("b", false)
+	s2 := runServerWithOpts(t, s2sOpts, nil)
+	defer s2.Shutdown()
+
+	getLeader(t, 10*time.Second, s1, s2)
+
+	sc := NewDefaultConnection(t)
+	defer sc.Close()
+	testRemoveNode(t, sc.NatsConn(), "b", true)
+	sc.Close()
+	s2.Shutdown()
+	s1.Shutdown()
+
+	cleanupDatastore(t)
+	cleanupRaftLog(t)
+
+	peers := []string{"a", "b", "c", "d", "e"}
+	servers := make([]*StanServer, 0, 5)
+
+	for _, nodeID := range peers {
+		opts := getTestDefaultOptsForClustering(nodeID, false)
+		opts.Clustering.AllowAddRemoveNode = true
+		opts.Clustering.Peers = peers
+		s := runServerWithOpts(t, opts, nil)
+		defer s.Shutdown()
+		servers = append(servers, s)
+	}
+
+	leader := getLeader(t, 10*time.Second, servers...)
+
+	sc = NewDefaultConnection(t)
+	defer sc.Close()
+
+	sc.Publish("foo", []byte("msg"))
+	checkChannelsInAllServers(t, []string{"foo"}, 2*time.Second, servers...)
+
+	// We will use the NATS connection from the STAN connection
+	// to send those add/remove requests
+	nc := sc.NatsConn()
+
+	// Now let's remove 2 followers
+	followers := removeServer(servers, leader)
+	for i := 0; i < 2; i++ {
+		f := followers[i]
+		testRemoveNode(t, nc, f.opts.Clustering.NodeID, false)
+		servers = removeServer(servers, f)
+		f.Shutdown()
+	}
+
+	// Now remove the leader itself
+	testRemoveNode(t, nc, leader.opts.Clustering.NodeID, false)
+	servers = removeServer(servers, leader)
+
+	// The current leader should stepdown. Wait for that to happen.
+	waitFor(t, 5*time.Second, 15*time.Millisecond, func() error {
+		if leader.raft.State() == raft.Leader {
+			return fmt.Errorf("still leader")
+		}
+		return nil
+	})
+
+	// Wait for a new leader
+	oldLeader := leader
+	// Since we sent the remove request to the leader itself,
+	// it should shutdown (and exit, but we don't do it when running in tests).
+	// Check that the status is shutdown.
+	waitFor(t, 2*time.Second, 15*time.Millisecond, func() error {
+		oldLeader.mu.Lock()
+		isShutdown := oldLeader.shutdown
+		oldLeader.mu.Unlock()
+		if !isShutdown {
+			return fmt.Errorf("old leader did not shutdown")
+		}
+		return nil
+	})
+
+	newLeader := getLeader(t, 10*time.Second, servers...)
+	if newLeader == oldLeader {
+		t.Fatalf("Leader %q was not replaced", oldLeader.opts.Clustering.NodeID)
+	}
+
+	// Ask the node to be added back
+	nodeID := oldLeader.opts.Clustering.NodeID
+	resp, err := nc.Request(fmt.Sprintf(addClusterNodeSubj, clusterName), []byte(nodeID), 2*time.Second)
+	if err != nil {
+		t.Fatalf("Request to add node failed: %v", err)
+	}
+	if string(resp.Data) != "+OK" {
+		t.Fatalf("Adding node %q returned response error: %q", nodeID, resp.Data)
+	}
+
+	s := runServerWithOpts(t, oldLeader.opts, nil)
+	defer s.Shutdown()
+	servers = append(servers, s)
+
+	getLeader(t, 10*time.Second, servers...)
+
+	sc.Publish("bar", []byte("msg"))
+	checkChannelsInAllServers(t, []string{"foo", "bar"}, 2*time.Second, servers...)
+}
+
+type captureFailedToContactNodeLogger struct {
+	dummyLogger
+	node        string
+	errMsgFound chan struct{}
+}
+
+func (l *captureFailedToContactNodeLogger) Errorf(format string, args ...interface{}) {
+	l.Lock()
+	msg := fmt.Sprintf(format, args...)
+	if l.node != "" {
+		errMsg := fmt.Sprintf("failed to heartbeat to: peer=%s.%s.%s", clusterName, l.node, clusterName)
+		if strings.Contains(msg, errMsg) {
+			select {
+			case l.errMsgFound <- struct{}{}:
+				l.node = ""
+			default:
+			}
+		}
+	}
+	l.Unlock()
+}
+
+func (l *captureFailedToContactNodeLogger) waitForFailedHearbeat(t *testing.T, node string) {
+	l.Lock()
+	l.node = node
+	l.errMsgFound = make(chan struct{}, 1)
+	l.Unlock()
+	select {
+	case <-l.errMsgFound:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Did not get heartbeat failures yet")
+	}
+}
+
+func (l *captureFailedToContactNodeLogger) makeSureNodeIsNoLongerContacted(t *testing.T, node string) {
+	timeout := time.Now().Add(2 * time.Second)
+	for time.Now().Before(timeout) {
+		l.Lock()
+		l.node = node
+		l.errMsgFound = make(chan struct{}, 1)
+		l.Unlock()
+		select {
+		case <-l.errMsgFound:
+		case <-time.After(1000 * time.Millisecond):
+			// OK!
+			return
+		}
+	}
+	t.Fatalf("Node %q is still being contacted", node)
+}
+
+func TestClusteringAddRemoveClusterNodesWithBootstrap(t *testing.T) {
+	cleanupDatastore(t)
+	defer cleanupDatastore(t)
+	cleanupRaftLog(t)
+	defer cleanupRaftLog(t)
+
+	// For this test, use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	l := &captureFailedToContactNodeLogger{}
+
+	// Configure first server
+	s1sOpts := getTestDefaultOptsForClustering("a", true)
+	s1sOpts.Clustering.AllowAddRemoveNode = true
+	s1sOpts.CustomLogger = l
+	s1 := runServerWithOpts(t, s1sOpts, nil)
+	defer s1.Shutdown()
+
+	// Wait for it to bootstrap
+	getLeader(t, 10*time.Second, s1)
+
+	// Configure second server.
+	s2sOpts := getTestDefaultOptsForClustering("b", false)
+	s2sOpts.Clustering.AllowAddRemoveNode = true
+	s2 := runServerWithOpts(t, s2sOpts, nil)
+	defer s2.Shutdown()
+
+	// Configure third server.
+	s3sOpts := getTestDefaultOptsForClustering("c", false)
+	s3sOpts.Clustering.AllowAddRemoveNode = true
+	s3 := runServerWithOpts(t, s3sOpts, nil)
+	defer s3.Shutdown()
+
+	// Configure fourth server.
+	s4sOpts := getTestDefaultOptsForClustering("d", false)
+	s4sOpts.Clustering.AllowAddRemoveNode = true
+	s4 := runServerWithOpts(t, s4sOpts, nil)
+	defer s4.Shutdown()
+
+	getLeader(t, 10*time.Second, s1, s2, s3, s4)
+	sc := NewDefaultConnection(t)
+	defer sc.Close()
+
+	s4.Shutdown()
+
+	l.waitForFailedHearbeat(t, "d")
+
+	testRemoveNode(t, sc.NatsConn(), "d", false)
+
+	l.makeSureNodeIsNoLongerContacted(t, "d")
+
+	// Check for error cases
+	resp, err := sc.NatsConn().Request(fmt.Sprintf(addClusterNodeSubj, clusterName), []byte(""), 2*time.Second)
+	if err != nil {
+		t.Fatalf("Request to add node failed: %v", err)
+	}
+	respStr := string(resp.Data)
+	if !strings.Contains(respStr, "-ERR adding node") {
+		t.Fatalf("Expected error response, got %q", respStr)
+	}
+
+	// Bring down the cluster to size 2, barely new quorum.
+	s3.Shutdown()
+
+	// Removing inexistent node does not fail, but it will if the leader loses leadership.
+	// So loop removing a node ID and then shutdown s2 (so s1 loses leadership)
+	subj := fmt.Sprintf(removeClusterNodeSubj, clusterName)
+	timeout := time.Now().Add(5 * time.Second)
+	time.AfterFunc(500*time.Millisecond, func() {
+		s2.Shutdown()
+	})
+	// We will try to get the error, but let's have a limit on how long we try.
+	for time.Now().Before(timeout) {
+		resp, err := sc.NatsConn().Request(subj, []byte("somenode"), time.Second)
+		// If the leader lost leadership before request is sent, it will timeout.
+		// So we are done (could not check the error)
+		if err != nil {
+			break
+		}
+		respStr := string(resp.Data)
+		if respStr == "+OK" {
+			continue
+		}
+		if strings.Contains(respStr, "-ERR removing node \"somenode\": leadership lost") {
+			break
+		}
+	}
+}
diff --git a/server/conf_test.go b/server/conf_test.go
index 14fc26a..9f2fe48 100644
--- a/server/conf_test.go
+++ b/server/conf_test.go
@@ -260,6 +260,9 @@ func TestParseConfig(t *testing.T) {
 	if opts.Clustering.RaftCommitTimeout != 50*time.Millisecond {
 		t.Fatalf("Expected RaftCommitTimeout to be 50ms, got %v", opts.Clustering.RaftCommitTimeout)
 	}
+	if !opts.Clustering.AllowAddRemoveNode {
+		t.Fatal("Expected AllowAddRemoveNode to be true")
+	}
 	if opts.SQLStoreOpts.Driver != "mysql" {
 		t.Fatalf("Expected SQL Driver to be %q, got %q", "mysql", opts.SQLStoreOpts.Driver)
 	}
@@ -485,6 +488,7 @@ func TestParseWrongTypes(t *testing.T) {
 	expectFailureFor(t, "cluster:{raft_lease_timeout:\"not_a_time\"}", wrongTimeErr)
 	expectFailureFor(t, "cluster:{raft_commit_timeout:123}", wrongTypeErr)
 	expectFailureFor(t, "cluster:{raft_commit_timeout:\"not_a_time\"}", wrongTimeErr)
+	expectFailureFor(t, "cluster:{allow_add_remove_node:1}", wrongTypeErr)
 	expectFailureFor(t, "sql:{driver:false}", wrongTypeErr)
 	expectFailureFor(t, "sql:{source:false}", wrongTypeErr)
 	expectFailureFor(t, "sql:{no_caching:123}", wrongTypeErr)
diff --git a/test/configs/test_parse.conf b/test/configs/test_parse.conf
index 66b86b5..000b054 100644
--- a/test/configs/test_parse.conf
+++ b/test/configs/test_parse.conf
@@ -87,6 +87,7 @@ streaming: {
       raft_election_timeout: "1s"
       raft_lease_timeout: "500ms"
       raft_commit_timeout: "50ms"
+      allow_add_remove_node: true
   }
 
   sql: {