Repository: nats-io/nats-server
Pull request #3473 (2022-09-15 00:25:34)

[ADDED] JetStream: ability to remove a server by peer ID instead of name
This can be helpful after a partial cluster restart since in that case the server name may not be known. However "server report jetstream" would report the peer ID that then can be used.

For instance here is the output after a cluster restart where server "C" is not restarted.

```
nats -s nats://sys:pwd@localhost:4222 server report jetstream
...
╭────────────────────────────────────────────────────────────────────────────────────────────────╮
│                                  RAFT Meta Group Information                                   │
├─────────────────────────────────────────────────────┬────────┬─────────┬────────┬────────┬─────┤
│ Name                                                │ Leader │ Current │ Online │ Active │ Lag │
├─────────────────────────────────────────────────────┼────────┼─────────┼────────┼────────┼─────┤
│ A                                                   │ yes    │ true    │ true   │ 0.00s  │ 0   │
│ B                                                   │        │ true    │ true   │ 0.53s  │ 0   │
│ Server name unknown at this time (peerID: jZ6RvVRH) │        │ false   │ false  │ 0.00s  │ 0   │
╰─────────────────────────────────────────────────────┴────────┴─────────┴────────┴────────┴─────╯
```

With a change to the NATS CLI we could have something like:
```
nats -s nats://sys:pwd@localhost:4222 server raft peer-remove jZ6RvVRH --by_id
```

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/.travis.yml b/.travis.yml
index f735f9e0..79c27f87 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -27,6 +27,8 @@ jobs:
       env: TEST_SUITE=js_cluster_tests_1
     - name: "Run JetStream cluster tests (2)"
       env: TEST_SUITE=js_cluster_tests_2
+    - name: "Run JetStream cluster tests (3)"
+      env: TEST_SUITE=js_cluster_tests_3
     - name: "Run JetStream super cluster tests"
       env: TEST_SUITE=js_super_cluster_tests
     - name: "Run MQTT tests"
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 0be83e5f..d4905a84 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -581,6 +581,9 @@ const JSApiLeaderStepDownResponseType = "io.nats.jetstream.api.v1.meta_leader_st
 type JSApiMetaServerRemoveRequest struct {
 	// Server name of the peer to be removed.
 	Server string `json:"peer"`
+	// Peer ID of the peer to be removed. If specified this is used
+	// instead of the server name.
+	Peer string `json:"peer_id,omitempty"`
 }
 
 // JSApiMetaServerRemoveResponse is the response to a peer removal request in the meta group.
@@ -2215,6 +2218,14 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, _ *Ac
 	var found string
 	js.mu.RLock()
 	for _, p := range cc.meta.Peers() {
+		// If Peer is specified, it takes precedence
+		if req.Peer != _EMPTY_ {
+			if p.ID == req.Peer {
+				found = req.Peer
+				break
+			}
+			continue
+		}
 		si, ok := s.nodeToInfo.Load(p.ID)
 		if ok && si.(nodeInfo).name == req.Server {
 			found = p.ID
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 3b9fd81d..0ef749d6 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -2000,10 +2000,10 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 				var newLeaderPeer, newLeader string
 				neededCurrent, current := replicas/2+1, 0
 				for _, r := range ci.Replicas {
-					if r.Current && newPeerSet[r.peer] {
+					if r.Current && newPeerSet[r.Peer] {
 						current++
 						if newLeader == _EMPTY_ {
-							newLeaderPeer, newLeader = r.peer, r.Name
+							newLeaderPeer, newLeader = r.Peer, r.Name
 						}
 					}
 				}
@@ -2427,7 +2427,7 @@ func (s *Server) replicas(node RaftNode) []*PeerInfo {
 	for _, rp := range node.Peers() {
 		if sir, ok := s.nodeToInfo.Load(rp.ID); ok && sir != nil {
 			si := sir.(nodeInfo)
-			pi := &PeerInfo{Name: si.name, Current: rp.Current, Active: now.Sub(rp.Last), Offline: si.offline, Lag: rp.Lag}
+			pi := &PeerInfo{Peer: rp.ID, Name: si.name, Current: rp.Current, Active: now.Sub(rp.Last), Offline: si.offline, Lag: rp.Lag}
 			replicas = append(replicas, pi)
 		}
 	}
@@ -3925,10 +3925,10 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 				var newLeaderPeer, newLeader, newCluster string
 				neededCurrent, current := replicas/2+1, 0
 				for _, r := range ci.Replicas {
-					if r.Current && newPeerSet[r.peer] {
+					if r.Current && newPeerSet[r.Peer] {
 						current++
 						if newCluster == _EMPTY_ {
-							newLeaderPeer, newLeader, newCluster = r.peer, r.Name, r.cluster
+							newLeaderPeer, newLeader, newCluster = r.Peer, r.Name, r.cluster
 						}
 					}
 				}
@@ -7087,7 +7087,7 @@ func (js *jetStream) offlineClusterInfo(rg *raftGroup) *ClusterInfo {
 	for _, peer := range rg.Peers {
 		if sir, ok := s.nodeToInfo.Load(peer); ok && sir != nil {
 			si := sir.(nodeInfo)
-			pi := &PeerInfo{Name: si.name, Current: false, Offline: true}
+			pi := &PeerInfo{Peer: peer, Name: si.name, Current: false, Offline: true}
 			ci.Replicas = append(ci.Replicas, pi)
 		}
 	}
@@ -7143,7 +7143,7 @@ func (js *jetStream) clusterInfo(rg *raftGroup) *ClusterInfo {
 				Offline: true,
 				Active:  lastSeen,
 				Lag:     rp.Lag,
-				peer:    rp.ID,
+				Peer:    rp.ID,
 			}
 			// If node is found, complete/update the settings.
 			if sir, ok := s.nodeToInfo.Load(rp.ID); ok && sir != nil {
@@ -7153,6 +7153,8 @@ func (js *jetStream) clusterInfo(rg *raftGroup) *ClusterInfo {
 				// If not, then add a name that indicates that the server name
 				// is unknown at this time, and clear the lag since it is misleading
 				// (the node may not have that much lag).
+				// Note: We return now the Peer ID in PeerInfo, so the "(peerID: %s)"
+				// would technically not be required, but keeping it for now.
 				pi.Name, pi.Lag = fmt.Sprintf("Server name unknown at this time (peerID: %s)", rp.ID), 0
 			}
 			ci.Replicas = append(ci.Replicas, pi)
diff --git a/server/stream.go b/server/stream.go
index ac4b25a1..45b2d4fc 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -143,9 +143,9 @@ type PeerInfo struct {
 	Offline bool          `json:"offline,omitempty"`
 	Active  time.Duration `json:"active"`
 	Lag     uint64        `json:"lag,omitempty"`
+	Peer    string        `json:"peer"`
 	// For migrations.
 	cluster string
-	peer    string
 }
 
 // StreamSourceInfo shows information about an upstream stream source.


Test patch:
diff --git a/scripts/runTestsOnTravis.sh b/scripts/runTestsOnTravis.sh
index 11ce2e3a..08d314ae 100755
--- a/scripts/runTestsOnTravis.sh
+++ b/scripts/runTestsOnTravis.sh
@@ -32,26 +32,34 @@ elif [ "$1" = "js_tests" ]; then
 
     # Run JetStream non-clustered tests. By convention, all JS tests start
     # with `TestJetStream`. We exclude the clustered and super-clustered
-    # tests by using the `skip_js_cluster_tests`, `skip_js_cluster_tests_2`
-    # and `skip_js_super_cluster_tests` build tags.
+    # tests by using the appropriate tags.
 
-    go test -race -v -run=TestJetStream ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_2,skip_js_super_cluster_tests -count=1 -vet=off -timeout=30m -failfast
+    go test -race -v -run=TestJetStream ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_2,skip_js_cluster_tests_3,skip_js_super_cluster_tests -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_cluster_tests_1" ]; then
 
     # Run JetStream clustered tests. By convention, all JS cluster tests
     # start with `TestJetStreamCluster`. Will run the first batch of tests,
-    # excluding the ones with the tag skip_js_cluster_tests_2.
+    # excluding others with use of proper tags.
 
-    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests_2 -count=1 -vet=off -timeout=30m -failfast
+    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests_2,skip_js_cluster_tests_3 -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_cluster_tests_2" ]; then
 
     # Run JetStream clustered tests. By convention, all JS cluster tests
     # start with `TestJetStreamCluster`. Will run the second batch of tests,
-    # excluding the ones with the tag skip_js_cluster_tests.
+    # excluding others with use of proper tags.
 
-    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests -count=1 -vet=off -timeout=30m -failfast
+    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_3 -count=1 -vet=off -timeout=30m -failfast
+
+elif [ "$1" = "js_cluster_tests_3" ]; then
+
+    # Run JetStream clustered tests. By convention, all JS cluster tests
+    # start with `TestJetStreamCluster`. Will run the third batch of tests,
+    # excluding others with use of proper tags.
+    #
+
+    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_2 -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_super_cluster_tests" ]; then
 
diff --git a/server/jetstream_cluster_2_test.go b/server/jetstream_cluster_2_test.go
index cf42cacb..4e6f69f7 100644
--- a/server/jetstream_cluster_2_test.go
+++ b/server/jetstream_cluster_2_test.go
@@ -7259,3 +7259,8 @@ func TestJetStreamClusterCompressedStreamMessages(t *testing.T) {
 		t.Fatalf("Did not receive completion signal")
 	}
 }
+
+//
+// DO NOT ADD NEW TESTS IN THIS FILE
+// Add at the end of jetstream_cluster_<n>_test.go, with <n> being the highest value.
+//
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
new file mode 100644
index 00000000..576616d8
--- /dev/null
+++ b/server/jetstream_cluster_3_test.go
@@ -0,0 +1,108 @@
+// Copyright 2022 The NATS 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.
+
+//go:build !skip_js_tests && !skip_js_cluster_tests_3
+// +build !skip_js_tests,!skip_js_cluster_tests_3
+
+package server
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+func TestJetStreamClusterRemovePeerByID(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	s := c.randomNonLeader()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo", "bar"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	// Wait for a leader
+	c.waitOnStreamLeader(globalAccountName, "TEST")
+
+	// Get the name of the one that is not restarted
+	srvName := c.opts[2].ServerName
+	// And its node ID
+	peerID := c.servers[2].Node()
+
+	nc.Close()
+	// Now stop the whole cluster
+	c.stopAll()
+	// Restart all but one
+	for i := 0; i < 2; i++ {
+		opts := c.opts[i]
+		s, o := RunServerWithConfig(opts.ConfigFile)
+		c.servers[i] = s
+		c.opts[i] = o
+	}
+
+	c.waitOnClusterReadyWithNumPeers(2)
+	c.waitOnStreamLeader(globalAccountName, "TEST")
+
+	// Now attempt to remove by name, this should fail because the cluster
+	// was restarted and names are not persisted.
+	ml := c.leader()
+	nc, err = nats.Connect(ml.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+	require_NoError(t, err)
+	defer nc.Close()
+
+	req := &JSApiMetaServerRemoveRequest{Server: srvName}
+	jsreq, err := json.Marshal(req)
+	require_NoError(t, err)
+	rmsg, err := nc.Request(JSApiRemoveServer, jsreq, 2*time.Second)
+	require_NoError(t, err)
+
+	var resp JSApiMetaServerRemoveResponse
+	err = json.Unmarshal(rmsg.Data, &resp)
+	require_NoError(t, err)
+	require_True(t, resp.Error != nil)
+	require_True(t, IsNatsErr(resp.Error, JSClusterServerNotMemberErr))
+
+	// Now try by ID, but first with an ID that does not match any peerID
+	req.Peer = "some_bad_id"
+	jsreq, err = json.Marshal(req)
+	require_NoError(t, err)
+	rmsg, err = nc.Request(JSApiRemoveServer, jsreq, 2*time.Second)
+	require_NoError(t, err)
+
+	resp = JSApiMetaServerRemoveResponse{}
+	err = json.Unmarshal(rmsg.Data, &resp)
+	require_NoError(t, err)
+	require_True(t, resp.Error != nil)
+	require_True(t, IsNatsErr(resp.Error, JSClusterServerNotMemberErr))
+
+	// Now with the proper peer ID
+	req.Peer = peerID
+	jsreq, err = json.Marshal(req)
+	require_NoError(t, err)
+	rmsg, err = nc.Request(JSApiRemoveServer, jsreq, 2*time.Second)
+	require_NoError(t, err)
+
+	resp = JSApiMetaServerRemoveResponse{}
+	err = json.Unmarshal(rmsg.Data, &resp)
+	require_NoError(t, err)
+	require_True(t, resp.Error == nil)
+	require_True(t, resp.Success)
+}
diff --git a/server/monitor_test.go b/server/monitor_test.go
index f8186937..b8eed2a3 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -4365,6 +4365,11 @@ func TestMonitorJsz(t *testing.T) {
 			info := readJsInfo(url + "")
 			if info.Meta.Replicas != nil {
 				found++
+				for _, r := range info.Meta.Replicas {
+					if r.Peer == _EMPTY_ {
+						t.Fatalf("Replicas' Peer is empty: %+v", r)
+					}
+				}
 				if info.Meta.Leader != srvs[i].Name() {
 					t.Fatalf("received cluster info from non leader: leader %s, server: %s", info.Meta.Leader, srvs[i].Name())
 				}