Repository: nats-io/nats-server
Pull request #1927 (2021-02-19 18:57:51)

Allow placement directives for meta controllers
Allow metacontroller leaders to be moved to specific clusters in a supercluster setup.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/gateway.go b/server/gateway.go
index d3ddc763..aecdf2cd 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -2622,15 +2622,6 @@ func (g *srvGateway) getClusterHash() []byte {
 	return clusterHash
 }
 
-// Returns the route with given hash or nil if not found.
-func (s *Server) getRouteByHash(srvHash []byte) *client {
-	var route *client
-	if v, ok := s.routesByHash.Load(string(srvHash)); ok {
-		route = v.(*client)
-	}
-	return route
-}
-
 // Store this route in map with the key being the remote server's name hash
 // and the remote server's ID hash used by gateway replies mapping routing.
 func (s *Server) storeRouteByHash(srvNameHash, srvIDHash string, c *client) {
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index e764f367..280d765b 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"math/rand"
 	"os"
 	"sort"
 	"strconv"
@@ -423,6 +424,11 @@ type JSApiConsumerLeaderStepDownResponse struct {
 
 const JSApiConsumerLeaderStepDownResponseType = "io.nats.jetstream.api.v1.consumer_leader_stepdown_response"
 
+// JSApiLeaderStepdownRequest allows placement control over the meta leader placement.
+type JSApiLeaderStepdownRequest struct {
+	Placement *Placement `json:"placement,omitempty"`
+}
+
 // JSApiLeaderStepDownResponse is the response to a meta leader stepdown request.
 type JSApiLeaderStepDownResponse struct {
 	ApiResponse
@@ -552,6 +558,8 @@ var (
 	jsClusterRequiredErr   = &ApiError{Code: 503, Description: "JetStream clustering support required"}
 	jsPeerNotMemberErr     = &ApiError{Code: 400, Description: "peer not a member"}
 	jsClusterIncompleteErr = &ApiError{Code: 503, Description: "incomplete results"}
+	jsClusterTagsErr       = &ApiError{Code: 400, Description: "tags placement not supported for operation"}
+	jsClusterNoPeersErr    = &ApiError{Code: 400, Description: "no suitable peers for placement"}
 )
 
 // For easier handling of exports and imports.
@@ -1699,10 +1707,47 @@ func (s *Server) jsLeaderStepDownRequest(sub *subscription, c *client, subject,
 		return
 	}
 
+	var preferredLeader string
+	var resp = JSApiLeaderStepDownResponse{ApiResponse: ApiResponse{Type: JSApiLeaderStepDownResponseType}}
+
+	if !isEmptyRequest(msg) {
+		var req JSApiLeaderStepdownRequest
+		if err := json.Unmarshal(msg, &req); err != nil {
+			resp.Error = jsInvalidJSONErr
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		if len(req.Placement.Tags) > 0 {
+			// Tags currently not supported.
+			resp.Error = jsClusterTagsErr
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		cn := req.Placement.Cluster
+		var peers []string
+		for _, p := range cc.meta.Peers() {
+			if si, ok := s.nodeToInfo.Load(p.ID); ok && si != nil {
+				if ni := si.(*nodeInfo); ni.offline || ni.cluster != cn {
+					continue
+				}
+				peers = append(peers, p.ID)
+			}
+		}
+		if len(peers) == 0 {
+			resp.Error = jsClusterNoPeersErr
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		// Randomize and select.
+		if len(peers) > 1 {
+			rand.Shuffle(len(peers), func(i, j int) { peers[i], peers[j] = peers[j], peers[i] })
+		}
+		preferredLeader = peers[0]
+	}
+
 	// Call actual stepdown.
-	err = cc.meta.StepDown()
+	err = cc.meta.StepDown(preferredLeader)
 
-	var resp = JSApiLeaderStepDownResponse{ApiResponse: ApiResponse{Type: JSApiLeaderStepDownResponseType}}
 	if err != nil {
 		resp.Error = jsError(err)
 	} else {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 4df00a97..1680d9cb 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -48,7 +48,7 @@ type jetStreamCluster struct {
 	stepdown *subscription
 }
 
-// Used to guide placement of streams in clustered JetStream.
+// Used to guide placement of streams and meta controllers in clustered JetStream.
 type Placement struct {
 	Cluster string   `json:"cluster"`
 	Tags    []string `json:"tags,omitempty"`
@@ -295,10 +295,11 @@ func (s *Server) JetStreamClusterPeers() []string {
 	peers := cc.meta.Peers()
 	var nodes []string
 	for _, p := range peers {
-		if si, ok := s.nodeToInfo.Load(p.ID); !ok || si.(*nodeInfo).offline {
+		si, ok := s.nodeToInfo.Load(p.ID)
+		if !ok || si.(*nodeInfo).offline {
 			continue
 		}
-		nodes = append(nodes, p.ID)
+		nodes = append(nodes, si.(*nodeInfo).name)
 	}
 	return nodes
 }
diff --git a/server/raft.go b/server/raft.go
index 11971fad..0ec51220 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -47,7 +47,7 @@ type RaftNode interface {
 	Quorum() bool
 	Current() bool
 	GroupLeader() string
-	StepDown() error
+	StepDown(preferred ...string) error
 	Campaign() error
 	ID() string
 	Group() string
@@ -219,6 +219,8 @@ var (
 	errBadSnapName     = errors.New("raft: snapshot name could not be parsed")
 	errNoSnapAvailable = errors.New("raft: no snapshot available")
 	errSnapshotCorrupt = errors.New("raft: snapshot corrupt")
+	errTooManyPrefs    = errors.New("raft: stepdown requires at most one preferred new leader")
+	errStepdownNoPeer  = errors.New("raft: stepdown failed, could not match new leader")
 )
 
 // This will bootstrap a raftNode by writing its config into the store directory.
@@ -911,9 +913,13 @@ func (n *raft) GroupLeader() string {
 }
 
 // StepDown will have a leader stepdown and optionally do a leader transfer.
-func (n *raft) StepDown() error {
+func (n *raft) StepDown(preferred ...string) error {
 	n.Lock()
 
+	if len(preferred) > 1 {
+		return errTooManyPrefs
+	}
+
 	if n.state != Leader {
 		n.Unlock()
 		return errNotLeader
@@ -924,19 +930,30 @@ func (n *raft) StepDown() error {
 	// See if we have up to date followers.
 	nowts := time.Now().UnixNano()
 	maybeLeader := noLeader
+	if len(preferred) > 0 {
+		maybeLeader = preferred[0]
+	}
 	for peer, ps := range n.peers {
 		// If not us and alive and caughtup.
-		if peer != n.id && (nowts-ps.ts) < int64(hbInterval*2) {
-			if n.s.getRouteByHash([]byte(peer)) != nil {
-				n.debug("Looking at %q which is %v behind", peer, time.Duration(nowts-ps.ts))
-				maybeLeader = peer
-				break
+		if peer != n.id && (nowts-ps.ts) < int64(hbInterval*3) {
+			if maybeLeader != noLeader && maybeLeader != peer {
+				continue
 			}
+			if si, ok := n.s.nodeToInfo.Load(peer); !ok || si.(*nodeInfo).offline {
+				continue
+			}
+			n.debug("Looking at %q which is %v behind", peer, time.Duration(nowts-ps.ts))
+			maybeLeader = peer
+			break
 		}
 	}
 	stepdown := n.stepdown
 	n.Unlock()
 
+	if len(preferred) > 0 && maybeLeader == noLeader {
+		return errStepdownNoPeer
+	}
+
 	if maybeLeader != noLeader {
 		n.debug("Stepping down, selected %q for new leader", maybeLeader)
 		n.sendAppendEntry([]*Entry{&Entry{EntryLeaderTransfer, []byte(maybeLeader)}})


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 7339e4e8..34d2de57 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -678,7 +678,7 @@ func TestJetStreamClusterMetaSnapshotsMultiChange(t *testing.T) {
 	// Add in a new server to the group. This way we know we can delete the original streams and consumers.
 	rs := c.addInNewServer()
 	c.waitOnServerCurrent(rs)
-	rsnn := rs.NodeName()
+	rsn := rs.Name()
 
 	// Shut it down.
 	rs.Shutdown()
@@ -686,7 +686,7 @@ func TestJetStreamClusterMetaSnapshotsMultiChange(t *testing.T) {
 	// Wait for the peer to be removed.
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		for _, p := range s.JetStreamClusterPeers() {
-			if p == rsnn {
+			if p == rsn {
 				return fmt.Errorf("Old server still in peer set")
 			}
 		}
@@ -3445,6 +3445,68 @@ func TestJetStreamClusterStreamGetMsg(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterMetaPlacement(t *testing.T) {
+	sc := createJetStreamSuperCluster(t, 3, 3)
+	defer sc.shutdown()
+
+	// We want to influence where the meta leader will place itself when we ask the
+	// current leader to stepdown.
+	ml := sc.leader()
+	cn := ml.ClusterName()
+	var pcn string
+	for _, c := range sc.clusters {
+		if c.name != cn {
+			pcn = c.name
+			break
+		}
+	}
+
+	// Client based API
+	s := sc.randomCluster().randomServer()
+	nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+	if err != nil {
+		t.Fatalf("Failed to create system client: %v", err)
+	}
+	defer nc.Close()
+
+	stepdown := func(cn string) *JSApiLeaderStepDownResponse {
+		req := &JSApiLeaderStepdownRequest{Placement: &Placement{Cluster: cn}}
+		jreq, err := json.Marshal(req)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		resp, err := nc.Request(JSApiLeaderStepDown, jreq, time.Second)
+		if err != nil {
+			t.Fatalf("Error on stepdown request: %v", err)
+		}
+		var sdr JSApiLeaderStepDownResponse
+		if err := json.Unmarshal(resp.Data, &sdr); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		return &sdr
+	}
+
+	// Make sure we get correct errors for tags and bad or unavailable cluster placement.
+	sdr := stepdown("C22")
+	if sdr.Error == nil || !strings.Contains(sdr.Error.Description, "no suitable peers") {
+		t.Fatalf("Got incorrect error result: %+v", sdr.Error)
+	}
+	// Should work.
+	sdr = stepdown(pcn)
+	if sdr.Error != nil {
+		t.Fatalf("Got an error on stepdown: %+v", sdr.Error)
+	}
+
+	sc.waitOnLeader()
+	ml = sc.leader()
+	cn = ml.ClusterName()
+
+	if cn != pcn {
+		t.Fatalf("Expected new metaleader to be in cluster %q, got %q", pcn, cn)
+	}
+}
+
 func TestJetStreamClusterSuperClusterBasics(t *testing.T) {
 	sc := createJetStreamSuperCluster(t, 3, 3)
 	defer sc.shutdown()