Repository: nats-io/nats-server
Pull request #3011 (2022-04-06 22:46:34)

Adding unique_placement_prefix to ensure matching tags are not used twice

Allows to not place a stream in the same availability zone twice

Signed-off-by: Matthias Hanel <mh@synadia.com>


kozlovic: @matthiashanel Could you describe a bit in the PR description how the feature would be used and give an example. I am looking at the test case added (which is good) but I don't have a sense of how this works. Having more details in the PR can then be used as base for documentation, so it is not wasted. Thanks!
matthiashanel: @kozlovic the new filed specifies a mandatory prefix of a server_tag used for placement.
When the server makes a placement decision in addition to other factors (such as placment tags provided, or disk utilization) it picks only server that have `server_tags` with the specified value as prefix and where the matching tag is unique across all selected server.  

Say a three server have the server tags below and the `unique_placement_prefix` is `az`:
```
server 1: server_tags: [az:1]
server 2: server_tags: [az:2]
server 3: server_tags: [az:3]
server 4: server_tags: [az:1]
```

An R3 stream would only be placed on servers 1-3 or 2-4. Never would server 1 and 4 be used for to host the same R3 stream. 
kozlovic: > An R3 stream would only be placed on servers 1-3 or 2-4. Never would server 1 and 4 be used for to host the same R3 stream.

Thanks, that helps a lot. Will review...

Patch:
diff --git a/server/client.go b/server/client.go
index bea7fea7..f6e04a72 100644
--- a/server/client.go
+++ b/server/client.go
@@ -202,6 +202,7 @@ const (
 	DuplicateRemoteLeafnodeConnection
 	DuplicateClientID
 	DuplicateServerName
+	MinimumVersionRequired
 )
 
 // Some flags passed to processMsgResults
@@ -1226,6 +1227,15 @@ func (c *client) readLoop(pre []byte) {
 		// to process messages, etc.
 		for i := 0; i < len(bufs); i++ {
 			if err := c.parse(bufs[i]); err != nil {
+				if err == ErrMinimumVersionRequired {
+					// Special case here, currently only for leaf node connections.
+					// When process the CONNECT protocol, if the minimum version
+					// required was not met, an error was printed and sent back to
+					// the remote, and connection was closed after a certain delay
+					// (to avoid "rapid" reconnection from the remote).
+					// We don't need to do any of the things below, simply return.
+					return
+				}
 				if dur := time.Since(start); dur >= readLoopReportThreshold {
 					c.Warnf("Readloop processing time: %v", dur)
 				}
diff --git a/server/errors.go b/server/errors.go
index a5ba0ac9..56b5af3f 100644
--- a/server/errors.go
+++ b/server/errors.go
@@ -188,6 +188,9 @@ var (
 	// ErrDuplicateServerName is returned when processing a server remote connection and
 	// the server reports that this server name is already used in the cluster.
 	ErrDuplicateServerName = errors.New("duplicate server name")
+
+	// ErrMinimumVersionRequired is returned when a connection is not at the minimum version required.
+	ErrMinimumVersionRequired = errors.New("minimum version required")
 )
 
 // configErr is a configuration error.
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 3ac85eb3..1931dcc3 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -4049,6 +4049,7 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 	}
 
 	var nodes []wn
+	// peers is a randomized list
 	s, peers := cc.s, cc.meta.Peers()
 
 	// Map existing.
@@ -4063,6 +4064,18 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 		}
 	}
 
+	uniqueTagPrefix := s.getOpts().JetStreamUniqueTag
+	if uniqueTagPrefix != _EMPTY_ {
+		for _, tag := range tags {
+			if strings.HasPrefix(tag, uniqueTagPrefix) {
+				// disable uniqueness check of explicitly listed in tags
+				uniqueTagPrefix = _EMPTY_
+				break
+			}
+		}
+	}
+	var uniqueTags = make(map[string]struct{})
+
 	for _, p := range peers {
 		si, ok := s.nodeToInfo.Load(p.ID)
 		if !ok || si == nil {
@@ -4119,6 +4132,23 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 		if maxBytes > 0 && maxBytes > available {
 			continue
 		}
+
+		if uniqueTagPrefix != _EMPTY_ {
+			// default requires the unique prefix to be present
+			isUnique := false
+			for _, t := range ni.tags {
+				if strings.HasPrefix(t, uniqueTagPrefix) {
+					if _, ok := uniqueTags[t]; !ok {
+						uniqueTags[t] = struct{}{}
+						isUnique = true
+					}
+					break
+				}
+			}
+			if !isUnique {
+				continue
+			}
+		}
 		// Add to our list of potential nodes.
 		nodes = append(nodes, wn{p.ID, available})
 	}
diff --git a/server/leafnode.go b/server/leafnode.go
index 40f068e1..78adef19 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -56,6 +56,10 @@ const leafNodeLoopDetectionSubjectPrefix = "$LDS."
 // LEAF connection as opposed to a CLIENT.
 const leafNodeWSPath = "/leafnode"
 
+// This is the time the server will wait, when receiving a CONNECT,
+// before closing the connection if the required minimum version is not met.
+const leafNodeWaitBeforeClose = 5 * time.Second
+
 type leaf struct {
 	// We have any auth stuff here for solicited connections.
 	remote *leafNodeCfg
@@ -252,6 +256,18 @@ func validateLeafNode(o *Options) error {
 	if o.LeafNode.Port == 0 {
 		return nil
 	}
+
+	// If MinVersion is defined, check that it is valid.
+	if mv := o.LeafNode.MinVersion; mv != _EMPTY_ {
+		if err := checkLeafMinVersionConfig(mv); err != nil {
+			return err
+		}
+	}
+
+	// The checks below will be done only when detecting that we are configured
+	// with gateways. So if an option validation needs to be done regardless,
+	// it MUST be done before this point!
+
 	if o.Gateway.Name == "" && o.Gateway.Port == 0 {
 		return nil
 	}
@@ -266,6 +282,17 @@ func validateLeafNode(o *Options) error {
 	return nil
 }
 
+func checkLeafMinVersionConfig(mv string) error {
+	if ok, err := versionAtLeastCheckError(mv, 2, 8, 0); !ok || err != nil {
+		if err != nil {
+			return fmt.Errorf("invalid leafnode's minimum version: %v", err)
+		} else {
+			return fmt.Errorf("the minimum version should be at least 2.8.0")
+		}
+	}
+	return nil
+}
+
 // Used to validate user names in LeafNode configuration.
 // - rejects mix of single and multiple users.
 // - rejects duplicate user names.
@@ -613,6 +640,7 @@ var credsRe = regexp.MustCompile(`\s*(?:(?:[-]{3,}[^\n]*[-]{3,}\n)(.+)(?:\n\s*[-
 func (c *client) sendLeafConnect(clusterName string, tlsRequired, headers bool) error {
 	// We support basic user/pass and operator based user JWT with signatures.
 	cinfo := leafConnectInfo{
+		Version:   VERSION,
 		TLS:       tlsRequired,
 		ID:        c.srv.info.ID,
 		Domain:    c.srv.info.Domain,
@@ -1316,6 +1344,7 @@ func (s *Server) removeLeafNodeConnection(c *client) {
 
 // Connect information for solicited leafnodes.
 type leafConnectInfo struct {
+	Version   string   `json:"version,omitempty"`
 	JWT       string   `json:"jwt,omitempty"`
 	Sig       string   `json:"sig,omitempty"`
 	User      string   `json:"user,omitempty"`
@@ -1363,6 +1392,25 @@ func (c *client) processLeafNodeConnect(s *Server, arg []byte, lang string) erro
 		return ErrWrongGateway
 	}
 
+	if mv := s.getOpts().LeafNode.MinVersion; mv != _EMPTY_ {
+		major, minor, update, _ := versionComponents(mv)
+		if !versionAtLeast(proto.Version, major, minor, update) {
+			// We are going to send back an INFO because otherwise recent
+			// versions of the remote server would simply break the connection
+			// after 2 seconds if not receiving it. Instead, we want the
+			// other side to just "stall" until we finish waiting for the holding
+			// period and close the connection below.
+			s.sendPermsAndAccountInfo(c)
+			c.sendErrAndErr(fmt.Sprintf("connection rejected since minimum version required is %q", mv))
+			select {
+			case <-c.srv.quitCh:
+			case <-time.After(leafNodeWaitBeforeClose):
+			}
+			c.closeConnection(MinimumVersionRequired)
+			return ErrMinimumVersionRequired
+		}
+	}
+
 	// Check if this server supports headers.
 	supportHeaders := c.srv.supportsHeaders()
 
diff --git a/server/monitor.go b/server/monitor.go
index 01918ab9..44c02f21 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -2230,6 +2230,8 @@ func (reason ClosedState) String() string {
 		return "Duplicate Client ID"
 	case DuplicateServerName:
 		return "Duplicate Server Name"
+	case MinimumVersionRequired:
+		return "Minimum Version Required"
 	}
 
 	return "Unknown State"
diff --git a/server/opts.go b/server/opts.go
index cd790108..95af6e20 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -143,6 +143,13 @@ type LeafNodeOpts struct {
 	// For solicited connections to other clusters/superclusters.
 	Remotes []*RemoteLeafOpts `json:"remotes,omitempty"`
 
+	// This is the minimum version that is accepted for remote connections.
+	// Note that since the server version in the CONNECT protocol was added
+	// only starting at v2.8.0, any version below that will be rejected
+	// (since empty version string in CONNECT would fail the "version at
+	// least" test).
+	MinVersion string
+
 	// Not exported, for tests.
 	resolver    netResolver
 	dialTimeout time.Duration
@@ -180,6 +187,7 @@ type RemoteLeafOpts struct {
 type JSLimitOpts struct {
 	MaxAckPending int
 	Duplicates    time.Duration
+	UniqueTag     string
 }
 
 // Options block for nats-server.
@@ -231,6 +239,7 @@ type Options struct {
 	JetStreamDomain       string        `json:"-"`
 	JetStreamExtHint      string        `json:"-"`
 	JetStreamKey          string        `json:"-"`
+	JetStreamUniqueTag    string
 	JetStreamLimits       JSLimitOpts
 	StoreDir              string            `json:"-"`
 	JsAccDefaultDomain    map[string]string `json:"-"` // account to domain name mapping
@@ -1886,6 +1895,8 @@ func parseJetStream(v interface{}, opts *Options, errors *[]error, warnings *[]e
 				if err := parseJetStreamLimits(tk, opts, errors, warnings); err != nil {
 					return err
 				}
+			case "unique_tag":
+				opts.JetStreamUniqueTag = strings.ToLower(strings.TrimSpace(mv.(string)))
 			default:
 				if !tk.IsUsedVariable() {
 					err := &unknownConfigFieldErr{
@@ -1981,6 +1992,14 @@ func parseLeafNodes(v interface{}, opts *Options, errors *[]error, warnings *[]e
 		case "no_advertise":
 			opts.LeafNode.NoAdvertise = mv.(bool)
 			trackExplicitVal(opts, &opts.inConfig, "LeafNode.NoAdvertise", opts.LeafNode.NoAdvertise)
+		case "min_version", "minimum_version":
+			version := mv.(string)
+			if err := checkLeafMinVersionConfig(version); err != nil {
+				err = &configErr{tk, err.Error()}
+				*errors = append(*errors, err)
+				continue
+			}
+			opts.LeafNode.MinVersion = version
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
diff --git a/server/util.go b/server/util.go
index 24f4de9f..24cef1ac 100644
--- a/server/util.go
+++ b/server/util.go
@@ -59,17 +59,22 @@ func versionComponents(version string) (major, minor, patch int, err error) {
 	return major, minor, patch, err
 }
 
-func versionAtLeast(version string, emajor, eminor, epatch int) bool {
+func versionAtLeastCheckError(version string, emajor, eminor, epatch int) (bool, error) {
 	major, minor, patch, err := versionComponents(version)
 	if err != nil {
-		return false
+		return false, err
 	}
 	if major > emajor ||
 		(major == emajor && minor > eminor) ||
 		(major == emajor && minor == eminor && patch >= epatch) {
-		return true
+		return true, nil
 	}
-	return false
+	return false, err
+}
+
+func versionAtLeast(version string, emajor, eminor, epatch int) bool {
+	res, _ := versionAtLeastCheckError(version, emajor, eminor, epatch)
+	return res
 }
 
 // parseSize expects decimal positive numbers. We


Test patch:
diff --git a/server/config_check_test.go b/server/config_check_test.go
index edb808d4..dd476c63 100644
--- a/server/config_check_test.go
+++ b/server/config_check_test.go
@@ -1238,6 +1238,39 @@ func TestConfigCheck(t *testing.T) {
 			errorLine: 4,
 			errorPos:  18,
 		},
+		{
+			name: "when leafnode min_version is wrong type",
+			config: `
+				leafnodes {
+					port: -1
+					min_version = 123
+				}`,
+			err:       errors.New(`interface conversion: interface {} is int64, not string`),
+			errorLine: 4,
+			errorPos:  6,
+		},
+		{
+			name: "when leafnode min_version has parsing error",
+			config: `
+				leafnodes {
+					port: -1
+					min_version = bad.version
+				}`,
+			err:       errors.New(`invalid leafnode's minimum version: invalid semver`),
+			errorLine: 4,
+			errorPos:  6,
+		},
+		{
+			name: "when leafnode min_version is too low",
+			config: `
+				leafnodes {
+					port: -1
+					min_version = 2.7.9
+				}`,
+			err:       errors.New(`the minimum version should be at least 2.8.0`),
+			errorLine: 4,
+			errorPos:  6,
+		},
 		{
 			name: "when setting latency tracking with a system account",
 			config: `
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index dd979894..11111d34 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -4563,6 +4563,85 @@ func TestJetStreamClusterSuperClusterMetaPlacement(t *testing.T) {
 	}
 }
 
+func TestJetStreamUniquePlacementTag(t *testing.T) {
+	tmlp := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s', unique_tag: az}
+		leaf {listen: 127.0.0.1:-1}
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+		# For access to system account.
+		accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+	`
+	s := createJetStreamSuperClusterWithTemplateAndModHook(t, tmlp, 5, 2,
+		func(serverName, clustername, storeDir, conf string) string {
+			azTag := map[string]string{
+				"C1-S1": "az:same",
+				"C1-S2": "az:same",
+				"C1-S3": "az:same",
+				"C1-S4": "az:same",
+				"C1-S5": "az:same",
+				"C2-S1": "az:1",
+				"C2-S2": "az:2",
+				"C2-S3": "az:1",
+				"C2-S4": "az:2",
+				"C2-S5": "az:1",
+			}
+			return conf + fmt.Sprintf("\nserver_tags: [cloud:%s-tag, %s]\n", clustername, azTag[serverName])
+		})
+	defer s.shutdown()
+
+	nc := natsConnect(t, s.randomServer().ClientURL())
+	defer nc.Close()
+
+	js, err := nc.JetStream()
+	require_NoError(t, err)
+
+	for i, test := range []struct {
+		placement *nats.Placement
+		replicas  int
+		fail      bool
+		cluster   string
+	}{
+		// these pass because replica count is 1
+		{&nats.Placement{Tags: []string{"az:same"}}, 1, false, "C1"},
+		{&nats.Placement{Tags: []string{"cloud:C1-tag", "az:same"}}, 1, false, "C1"},
+		{&nats.Placement{Tags: []string{"cloud:C1-tag"}}, 1, false, "C1"},
+		// pass because az is set, which disables the filter
+		{&nats.Placement{Tags: []string{"az:same"}}, 2, false, "C1"},
+		{&nats.Placement{Tags: []string{"cloud:C1-tag", "az:same"}}, 2, false, "C1"},
+		// fails because this cluster only has the same az
+		{&nats.Placement{Tags: []string{"cloud:C1-tag"}}, 2, true, ""},
+		// fails because no 3 unique tags exist
+		{&nats.Placement{Tags: []string{"cloud:C2-tag"}}, 3, true, ""},
+		{nil, 3, true, ""},
+		// pass because replica count is low enough
+		{nil, 2, false, "C2"},
+		{&nats.Placement{Tags: []string{"cloud:C2-tag"}}, 2, false, "C2"},
+		// pass because az is provided
+		{&nats.Placement{Tags: []string{"az:1"}}, 3, false, "C2"},
+		{&nats.Placement{Tags: []string{"az:2"}}, 2, false, "C2"},
+	} {
+		name := fmt.Sprintf("test-%d", i)
+		t.Run(name, func(t *testing.T) {
+			ci, err := js.AddStream(&nats.StreamConfig{Name: name, Replicas: test.replicas, Placement: test.placement})
+			if test.fail {
+				require_Error(t, err)
+				require_Equal(t, err.Error(), "insufficient resources")
+				return
+			}
+			require_NoError(t, err)
+			if test.cluster != _EMPTY_ {
+				require_Equal(t, ci.Cluster.Name, test.cluster)
+			}
+		})
+	}
+}
+
 func TestJetStreamClusterSuperClusterBasics(t *testing.T) {
 	sc := createJetStreamSuperCluster(t, 3, 3)
 	defer sc.shutdown()
diff --git a/server/jwt_test.go b/server/jwt_test.go
index 09fd9b70..2b36e34a 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -5769,7 +5769,7 @@ func TestJWTClusteredJetStreamTiers(t *testing.T) {
 	_, err = js.Publish("testR1-2", msg[:])
 	require_NoError(t, err)
 
-	time.Sleep(1700 * time.Millisecond) // wait for update timer to synchronize totals
+	time.Sleep(2000 * time.Millisecond) // wait for update timer to synchronize totals
 
 	// test exceeding tiered storage limit
 	_, err = js.Publish("testR1-1", []byte("1"))
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index 1e47d93c..257f7294 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -5280,3 +5280,142 @@ leafnodes:{
 	require_NoError(t, err)
 	test(jsAA, jsLL)
 }
+
+type checkLeafMinVersionLogger struct {
+	DummyLogger
+	errCh  chan string
+	connCh chan string
+}
+
+func (l *checkLeafMinVersionLogger) Errorf(format string, args ...interface{}) {
+	msg := fmt.Sprintf(format, args...)
+	if strings.Contains(msg, "minimum version") {
+		select {
+		case l.errCh <- msg:
+		default:
+		}
+	}
+}
+
+func (l *checkLeafMinVersionLogger) Noticef(format string, args ...interface{}) {
+	msg := fmt.Sprintf(format, args...)
+	if strings.Contains(msg, "Leafnode connection created") {
+		select {
+		case l.connCh <- msg:
+		default:
+		}
+	}
+}
+
+func TestLeafNodeMinVersion(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		port: -1
+		leafnodes {
+			port: -1
+			min_version: 2.8.0
+		}
+	`))
+	defer removeFile(t, conf)
+	s, o := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	rconf := createConfFile(t, []byte(fmt.Sprintf(`
+		port: -1
+		leafnodes {
+			remotes [
+				{url: "nats://127.0.0.1:%d" }
+			]
+		}
+	`, o.LeafNode.Port)))
+	defer removeFile(t, rconf)
+	ln, _ := RunServerWithConfig(rconf)
+	defer ln.Shutdown()
+
+	checkLeafNodeConnected(t, s)
+	checkLeafNodeConnected(t, ln)
+
+	ln.Shutdown()
+	s.Shutdown()
+
+	// Now makes sure we validate options, not just config file.
+	for _, test := range []struct {
+		name    string
+		version string
+		err     string
+	}{
+		{"invalid version", "abc", "semver"},
+		{"version too low", "2.7.9", "the minimum version should be at least 2.8.0"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			o.Port = -1
+			o.LeafNode.Port = -1
+			o.LeafNode.MinVersion = test.version
+			if s, err := NewServer(o); err == nil || !strings.Contains(err.Error(), test.err) {
+				if s != nil {
+					s.Shutdown()
+				}
+				t.Fatalf("Expected error to contain %q, got %v", test.err, err)
+			}
+		})
+	}
+
+	// Ok, so now to verify that a server rejects a leafnode connection
+	// we will set the min_version above our current VERSION. So first
+	// decompose the version:
+	major, minor, _, err := versionComponents(VERSION)
+	if err != nil {
+		t.Fatalf("The current server version %q is not valid: %v", VERSION, err)
+	}
+	// Let's make our minimum server an minor version above
+	mv := fmt.Sprintf("%d.%d.0", major, minor+1)
+	conf = createConfFile(t, []byte(fmt.Sprintf(`
+		port: -1
+		leafnodes {
+			port: -1
+			min_version: "%s"
+		}
+	`, mv)))
+	defer removeFile(t, conf)
+	s, o = RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	l := &checkLeafMinVersionLogger{errCh: make(chan string, 1), connCh: make(chan string, 1)}
+	s.SetLogger(l, false, false)
+
+	rconf = createConfFile(t, []byte(fmt.Sprintf(`
+		port: -1
+		leafnodes {
+			remotes [
+				{url: "nats://127.0.0.1:%d" }
+			]
+		}
+	`, o.LeafNode.Port)))
+	defer removeFile(t, rconf)
+	lo := LoadConfig(rconf)
+	lo.LeafNode.ReconnectInterval = 50 * time.Millisecond
+	ln = RunServer(lo)
+	defer ln.Shutdown()
+
+	select {
+	case <-l.connCh:
+	case <-time.After(time.Second):
+		t.Fatal("Remote did not try to connect")
+	}
+
+	select {
+	case <-l.errCh:
+	case <-time.After(time.Second):
+		t.Fatal("Did not get the minimum version required error")
+	}
+
+	// Since we have a very small reconnect interval, if the connection was
+	// closed "right away", then we should have had a reconnect attempt with
+	// another failure. This should not be the case because the server will
+	// wait 5s before closing the connection.
+	select {
+	case <-l.connCh:
+		t.Fatal("Should not have tried to reconnect")
+	case <-time.After(250 * time.Millisecond):
+		// OK
+	}
+}
diff --git a/test/leafnode_test.go b/test/leafnode_test.go
index 89ff98ba..cb29db33 100644
--- a/test/leafnode_test.go
+++ b/test/leafnode_test.go
@@ -1684,6 +1684,7 @@ func TestLeafNodeOperatorAndPermissions(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Error on subscribe: %v", err)
 	}
+	leafnc.Flush()
 
 	// Make sure the interest on "bar" from "sl" server makes it to the "s" server.
 	checkSubInterest(t, s, acc.GetName(), "bar", time.Second)