Repository: nats-io/nats-server (go)

Issue:
Remote PULL consumer SPOF 
## Problem description

In a system where two NATS clusters are connected using leaf node connections, clients connected to the remote NATS cluster cannot consume messages from a stream when the NATS server hosting the consumer leader is unable to establish leaf node connections due to network failure. In this scenario a connection to the remote system is still available from other nodes in the cluster but messages cannot be routed to the consumer leader resulting in an unexpected point of failure for the consuming application.


## More Details

<img width="932" alt="image" src="https://user-images.githubusercontent.com/48684307/172841297-0eec4575-4634-4f35-bd82-fea7841edc60.png">


To consume message from the consumer C attached to the stream S, the Application publishes NATS messages on the subject `$JS.a.API.CONSUMER.MSG.NEXT.S.C`
Messages published from the application in Cluster-B cannot reach the consumer leader which resides on `Cluster-A.nats-server-1`.
Current NATS routing does not allow message forwarding within the Cluster-A even if one instance of the Cluster-A has a healthy leaf connection to the Cluster-B.


## Remediation

The current remediation for this is to manually move the consumer leader to a node in the cluster that still has leaf node connections, while this is possible to automate externally it would be a beneficial enhancement to JetStream to have this as a built in feature.

Pull request description:
Direct access to JetStream resources when a leafnode connection is down.
This allows a soliciting leafnode config to ask that any JetStream cluster assets that are a current leader have the leader stepdown. This will allow a new leader to take over and service direct access.

Signed-off-by: Derek Collison <derek@nats.io>

Resolves #3178 

/cc @nats-io/core


Patch:
diff --git a/server/errors.go b/server/errors.go
index 6e13f4df..0dd73b89 100644
--- a/server/errors.go
+++ b/server/errors.go
@@ -79,6 +79,9 @@ var (
 	// and it has the same cluster name as the hub cluster.
 	ErrLeafNodeHasSameClusterName = errors.New("remote leafnode has same cluster name")
 
+	// ErrLeafNodeDisabled is when we disable leafnodes.
+	ErrLeafNodeDisabled = errors.New("leafnodes disabled")
+
 	// ErrConnectedToWrongPort represents an error condition when a connection is attempted
 	// to the wrong listen port (for instance a LeafNode to a client port, etc...)
 	ErrConnectedToWrongPort = errors.New("attempted to connect to wrong port")
diff --git a/server/leafnode.go b/server/leafnode.go
index c3e2dfd3..9cb5518a 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -495,8 +495,13 @@ func (s *Server) connectToRemoteLeafNode(remote *leafNodeCfg, firstConnect bool)
 			if url != rURL.Host {
 				ipStr = fmt.Sprintf(" (%s)", url)
 			}
-			s.Debugf("Trying to connect as leafnode to remote server on %q%s", rURL.Host, ipStr)
-			conn, err = natsDialTimeout("tcp", url, dialTimeout)
+			if s.isLeafNodeEnabled() {
+				s.Debugf("Trying to connect as leafnode to remote server on %q%s", rURL.Host, ipStr)
+				conn, err = natsDialTimeout("tcp", url, dialTimeout)
+			} else {
+				s.Debugf("Will not attempt to connect to remote server on %q%s, leafnodes currently disabled", rURL.Host, ipStr)
+				err = ErrLeafNodeDisabled
+			}
 		}
 		if err != nil {
 			attempts++
@@ -509,6 +514,8 @@ func (s *Server) connectToRemoteLeafNode(remote *leafNodeCfg, firstConnect bool)
 			case <-s.quitCh:
 				return
 			case <-time.After(reconnectDelay):
+				// Check if we should migrate any JetStream assets while this remote is down.
+				s.checkJetStreamMigrate(remote)
 				continue
 			}
 		}
@@ -524,6 +531,50 @@ func (s *Server) connectToRemoteLeafNode(remote *leafNodeCfg, firstConnect bool)
 	}
 }
 
+// Check to see if we should migrate any assets from this account.
+func (s *Server) checkJetStreamMigrate(remote *leafNodeCfg) {
+	s.mu.RLock()
+	accName, shouldMigrate := remote.LocalAccount, remote.JetStreamClusterMigrate
+	s.mu.RUnlock()
+
+	if !shouldMigrate {
+		return
+	}
+
+	acc, err := s.LookupAccount(accName)
+	if err != nil {
+		s.Debugf("Error looking up account [%s] checking for JetStream migration on a leafnode", accName)
+		return
+	}
+
+	// Walk all streams looking for any clustered stream, skip otherwise.
+	// If we are the leader force stepdown.
+	for _, mset := range acc.streams() {
+		node := mset.raftNode()
+		if node == nil {
+			// Not R>1
+			continue
+		}
+		// Collect any consumers
+		for _, o := range mset.getConsumers() {
+			if n := o.raftNode(); n != nil && n.Leader() {
+				n.StepDown()
+			}
+		}
+		// Stepdown if this stream was leader.
+		if node.Leader() {
+			node.StepDown()
+		}
+	}
+}
+
+// Helper for checking.
+func (s *Server) isLeafNodeEnabled() bool {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.leafNodeEnabled
+}
+
 // Save off the tlsName for when we use TLS and mix hostnames and IPs. IPs usually
 // come from the server we connect to.
 //
diff --git a/server/opts.go b/server/opts.go
index cdb150c2..0753464f 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -182,6 +182,12 @@ type RemoteLeafOpts struct {
 	}
 
 	tlsConfigOpts *TLSConfigOpts
+
+	// If we are clustered and our local account has JetStream, if apps are accessing
+	// a stream or consumer leader through this LN and it gets dropped, the apps will
+	// not be able to work. This tells the system to migrate the leaders away from this server.
+	// This only changes leader for R>1 assets.
+	JetStreamClusterMigrate bool `json:"jetstream_cluster_migrate,omitempty"`
 }
 
 type JSLimitOpts struct {
@@ -2252,6 +2258,8 @@ func parseRemoteLeafNodes(v interface{}, errors *[]error, warnings *[]error) ([]
 				remote.Websocket.Compression = v.(bool)
 			case "ws_no_masking", "websocket_no_masking":
 				remote.Websocket.NoMasking = v.(bool)
+			case "jetstream_cluster_migrate", "js_cluster_migrate":
+				remote.JetStreamClusterMigrate = true
 			default:
 				if !tk.IsUsedVariable() {
 					err := &unknownConfigFieldErr{


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 7609a67f..76382dad 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -11269,3 +11269,60 @@ func TestJetStreamClusterSameClusterLeafNodes(t *testing.T) {
 		checkLeafNodeConnectedCount(t, s, 0)
 	}
 }
+
+// https://github.com/nats-io/nats-server/issues/3178
+func TestJetStreamClusterLeafNodeSPOFMigrateLeaders(t *testing.T) {
+	tmpl := strings.Replace(jsClusterTempl, "store_dir:", "domain: REMOTE, store_dir:", 1)
+	c := createJetStreamClusterWithTemplate(t, tmpl, "HUB", 2)
+	defer c.shutdown()
+
+	tmpl = strings.Replace(jsClusterTemplWithLeafNode, "store_dir:", "domain: CORE, store_dir:", 1)
+	lnc := c.createLeafNodesWithTemplateAndStartPort(tmpl, "LNC", 2, 22110)
+	defer lnc.shutdown()
+
+	lnc.waitOnClusterReady()
+
+	// Place JS assets in LN, and we will do a pull consumer from the HUB.
+	nc, js := jsClientConnect(t, lnc.randomServer())
+	defer nc.Close()
+
+	si, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 2,
+	})
+	require_NoError(t, err)
+	require_True(t, si.Cluster.Name == "LNC")
+
+	for i := 0; i < 100; i++ {
+		js.PublishAsync("foo", []byte("HELLO"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	// Create the consumer.
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "d", AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+
+	nc, _ = jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	dsubj := "$JS.CORE.API.CONSUMER.MSG.NEXT.TEST.d"
+	// Grab directly using domain based subject but from the HUB cluster.
+	_, err = nc.Request(dsubj, nil, time.Second)
+	require_NoError(t, err)
+
+	// Now we will force the consumer leader's server to drop and stall leafnode connections.
+	cl := lnc.consumerLeader("$G", "TEST", "d")
+	cl.setJetStreamMigrateOnRemoteLeaf()
+	cl.closeAndDisableLeafnodes()
+
+	// Now make sure we can eventually get a message again.
+	checkFor(t, 5*time.Second, 500*time.Millisecond, func() error {
+		_, err = nc.Request(dsubj, nil, 500*time.Millisecond)
+		return err
+	})
+}
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index 3124bfa3..ba8356c3 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -924,6 +924,31 @@ func (c *cluster) createLeafNodesWithTemplateAndStartPort(template, clusterName
 	return lc
 }
 
+// Helper function to close and disable leafnodes.
+func (s *Server) closeAndDisableLeafnodes() {
+	var leafs []*client
+	s.mu.Lock()
+	for _, ln := range s.leafs {
+		leafs = append(leafs, ln)
+	}
+	// Disable leafnodes for now.
+	s.leafNodeEnabled = false
+	s.mu.Unlock()
+
+	for _, ln := range leafs {
+		ln.closeConnection(Revocation)
+	}
+}
+
+// Helper to set the remote migrate feature.
+func (s *Server) setJetStreamMigrateOnRemoteLeaf() {
+	s.mu.Lock()
+	for _, cfg := range s.leafRemoteCfgs {
+		cfg.JetStreamClusterMigrate = true
+	}
+	s.mu.Unlock()
+}
+
 // Will add in the mapping for the account to each server.
 func (c *cluster) addSubjectMapping(account, src, dest string) {
 	c.t.Helper()