Repository: nats-io/nats-server
Pull request #2164 (2021-04-29 01:44:22)

[FIXED] MQTT: asset placement in origin cluster
In a setup with shared system account and a cluster of leaf nodes,
the JS requests did not contain the origin cluster, which caused
assets to possibly be created in the HUB. With this change, the
assets will be created in the origin cluster.

Also, removed use of acc.JetStreamEnabled() but instead fail
start of the server if mqtt is enabled in standalone mode and JS
is not enabled. If JS is enabled, we will get proper error if
account has no JS enabled.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/server/jwt.go b/server/jwt.go
index a41f79d7..dda182a9 100644
--- a/server/jwt.go
+++ b/server/jwt.go
@@ -96,11 +96,11 @@ func validateTrustedOperators(o *Options) error {
 	if len(o.TrustedOperators) > 0 && len(o.TrustedKeys) > 0 {
 		return fmt.Errorf("conflicting options for 'TrustedKeys' and 'TrustedOperators'")
 	}
-	if o.SystemAccount != _EMPTY_ {
+	if o.SystemAccount != "" {
 		foundSys := false
 		foundNonEmpty := false
 		for _, op := range o.TrustedOperators {
-			if op.SystemAccount != _EMPTY_ {
+			if op.SystemAccount != "" {
 				foundNonEmpty = true
 			}
 			if op.SystemAccount == o.SystemAccount {
@@ -111,13 +111,6 @@ func validateTrustedOperators(o *Options) error {
 		if foundNonEmpty && !foundSys {
 			return fmt.Errorf("system_account in config and operator JWT must be identical")
 		}
-	} else if o.TrustedOperators[0].SystemAccount == _EMPTY_ {
-		// In case the system account is neither defined in config nor in the first operator.
-		// If it would be needed due to the nats account resolver, raise an error.
-		switch o.AccountResolver.(type) {
-		case *DirAccResolver, *CacheDirAccResolver:
-			return fmt.Errorf("using nats based account resolver - the system account needs to be specified in configuration or the operator jwt")
-		}
 	}
 	ver := strings.Split(strings.Split(VERSION, "-")[0], ".RC")[0]
 	srvMajor, srvMinor, srvUpdate, _ := jwt.ParseServerVersion(ver)
diff --git a/server/monitor.go b/server/monitor.go
index 936e2a66..a077dfee 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1101,18 +1101,11 @@ type LeafNodeOptsVarz struct {
 	Remotes     []RemoteLeafOptsVarz `json:"remotes,omitempty"`
 }
 
-// Contains lists of subjects not allowed to be imported/exported
-type DenyRules struct {
-	Exports []string `json:"exports,omitempty"`
-	Imports []string `json:"imports,omitempty"`
-}
-
 // RemoteLeafOptsVarz contains monitoring remote leaf node information
 type RemoteLeafOptsVarz struct {
-	LocalAccount string     `json:"local_account,omitempty"`
-	TLSTimeout   float64    `json:"tls_timeout,omitempty"`
-	URLs         []string   `json:"urls,omitempty"`
-	Deny         *DenyRules `json:"deny,omitempty"`
+	LocalAccount string   `json:"local_account,omitempty"`
+	TLSTimeout   float64  `json:"tls_timeout,omitempty"`
+	URLs         []string `json:"urls,omitempty"`
 }
 
 // VarzOptions are the options passed to Varz().
@@ -1285,18 +1278,10 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 	if l := len(ln.Remotes); l > 0 {
 		rlna := make([]RemoteLeafOptsVarz, l)
 		for i, r := range ln.Remotes {
-			var deny *DenyRules
-			if len(r.DenyImports) > 0 || len(r.DenyExports) > 0 {
-				deny = &DenyRules{
-					Imports: r.DenyImports,
-					Exports: r.DenyExports,
-				}
-			}
 			rlna[i] = RemoteLeafOptsVarz{
 				LocalAccount: r.LocalAccount,
 				URLs:         urlsToStrings(r.URLs),
 				TLSTimeout:   r.TLSTimeout,
-				Deny:         deny,
 			}
 		}
 		varz.LeafNode.Remotes = rlna
diff --git a/server/mqtt.go b/server/mqtt.go
index c9a1ee26..ce39010d 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -526,6 +526,10 @@ func validateMQTTOptions(o *Options) error {
 	if mo.AckWait < 0 {
 		return fmt.Errorf("ack wait must be a positive value")
 	}
+	// If standalone and there is no JS enabled, then it won't work...
+	if o.Cluster.Port == 0 && o.Gateway.Port == 0 && !o.JetStream {
+		return fmt.Errorf("mqtt requires JetStream to be enabled if running in standalone mode")
+	}
 	return nil
 }
 
@@ -825,14 +829,6 @@ func (s *Server) getOrCreateMQTTAccountSessionManager(clientID string, c *client
 	if ok {
 		return asm, nil
 	}
-	// First check that we have JS enabled for this account.
-	// TODO: Since we check only when creating a session manager for this
-	// account, would probably need to do some cleanup if JS can be disabled
-	// on config reload.
-	if !acc.JetStreamEnabled() {
-		return nil, fmt.Errorf("JetStream must be enabled for account %q used by MQTT client ID %q",
-			accName, clientID)
-	}
 	// Need to create one here.
 	asm, err := s.mqttCreateAccountSessionManager(acc, quitCh)
 	if err != nil {
@@ -908,7 +904,7 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 	// Start the go routine that will send JS API requests.
 	s.startGoRoutine(func() {
 		defer s.grWG.Done()
-		as.sendJSAPIrequests(sc, accName, closeCh)
+		as.sendJSAPIrequests(s, sc, accName, closeCh)
 	})
 
 	// Create the stream for the messages.
@@ -1412,11 +1408,12 @@ func (as *mqttAccountSessionManager) createSubscription(subject string, cb msgHa
 // only used when the server shutdown.
 //
 // No lock held on entry.
-func (as *mqttAccountSessionManager) sendJSAPIrequests(c *client, accName string, closeCh chan struct{}) {
+func (as *mqttAccountSessionManager) sendJSAPIrequests(s *Server, c *client, accName string, closeCh chan struct{}) {
+	cluster := s.cachedClusterName()
 	as.mu.RLock()
 	sendq := as.jsa.sendq
 	quitCh := as.jsa.quitCh
-	ci := ClientInfo{Account: accName}
+	ci := ClientInfo{Account: accName, Cluster: cluster}
 	as.mu.RUnlock()
 
 	// The account session manager does not have a suhtdown API per-se, instead,
diff --git a/server/server.go b/server/server.go
index a9612e76..d6a4ec44 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1664,9 +1664,6 @@ func (s *Server) Start() {
 // Shutdown will shutdown the server instance by kicking out the AcceptLoop
 // and closing all associated clients.
 func (s *Server) Shutdown() {
-	if s == nil {
-		return
-	}
 	// Transfer off any raft nodes that we are a leader by shutting them all down.
 	s.shutdownRaftNodes()
 


Test patch:
diff --git a/server/dirstore_test.go b/server/dirstore_test.go
index af0b9948..42ad1651 100644
--- a/server/dirstore_test.go
+++ b/server/dirstore_test.go
@@ -44,7 +44,7 @@ func require_False(t *testing.T, b bool) {
 	}
 }
 
-func require_NoError(t *testing.T, err error) {
+func require_NoError(t testing.TB, err error) {
 	t.Helper()
 	if err != nil {
 		t.Fatalf("require no error, but got: %v", err)
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 3275cb7c..825cb59e 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -6333,6 +6333,8 @@ var jsClusterTemplWithLeafNode = `
 		routes = [%s]
 	}
 
+	{{mqtt}}
+
 	# For access to system account.
 	accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
 `
@@ -6340,7 +6342,7 @@ var jsClusterTemplWithLeafNode = `
 var jsLeafFrag = `
 	leaf {
 		remotes [
-			{ urls: [ %s ], deny_exports: ["$JS.API.>"] }
+			{ urls: [ %s ], deny_exports: ["$JS.API.>"], deny_imports: ["$JS.API.>"] }
 			{ urls: [ %s ], account: "$SYS" }
 		]
 	}
@@ -6352,7 +6354,10 @@ func (c *cluster) createLeafNodes(clusterName string, numServers int) *cluster {
 }
 
 func (c *cluster) createLeafNodesWithStartPort(clusterName string, numServers int, portStart int) *cluster {
+	return c.createLeafNodesWithStartPortAndMQTT(clusterName, numServers, portStart, _EMPTY_)
+}
 
+func (c *cluster) createLeafNodesWithStartPortAndMQTT(clusterName string, numServers int, portStart int, mqtt string) *cluster {
 	// Create our leafnode cluster template first.
 	var lns, lnss []string
 	for _, s := range c.servers {
@@ -6364,6 +6369,7 @@ func (c *cluster) createLeafNodesWithStartPort(clusterName string, numServers in
 	lnsc := strings.Join(lnss, ", ")
 	lconf := fmt.Sprintf(jsLeafFrag, lnc, lnsc)
 	tmpl := strings.Replace(jsClusterTemplWithLeafNode, "{{leaf}}", lconf, 1)
+	tmpl = strings.Replace(tmpl, "{{mqtt}}", mqtt, 1)
 
 	pre := clusterName + "-"
 	lc := createJetStreamCluster(c.t, tmpl, clusterName, pre, numServers, portStart, false)
diff --git a/server/jwt_test.go b/server/jwt_test.go
index 27e7cb61..0cf7b9c3 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -3309,7 +3309,7 @@ func require_JWTEqual(t *testing.T, dir string, pub string, jwt string) {
 	require_Equal(t, string(content), jwt)
 }
 
-func createDir(t *testing.T, prefix string) string {
+func createDir(t testing.TB, prefix string) string {
 	t.Helper()
 	err := os.MkdirAll(tempRoot, 0700)
 	require_NoError(t, err)
@@ -5739,25 +5739,3 @@ func TestJWTMappings(t *testing.T) {
 	require_Len(t, 1, updateJwt(t, srv.ClientURL(), sysCreds, aJwtNoM, 1))
 	test("foo2", "bar2", true)
 }
-
-func TestJWTNoSystemAccountButNatsResolver(t *testing.T) {
-	dirSrv := createDir(t, "srv")
-	defer removeDir(t, dirSrv)
-	for _, resType := range []string{"full", "cache"} {
-		t.Run(resType, func(t *testing.T) {
-			conf := createConfFile(t, []byte(fmt.Sprintf(`
-			listen: -1
-			operator: %s
-			resolver: {
-				type: %s
-				dir: %s
-			}`, ojwt, resType, dirSrv)))
-			defer removeFile(t, conf)
-			opts := LoadConfig(conf)
-			s, err := NewServer(opts)
-			s.Shutdown()
-			require_Error(t, err)
-			require_Contains(t, err.Error(), "the system account needs to be specified in configuration or the operator jwt")
-		})
-	}
-}
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index d4559226..df989668 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -28,8 +28,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/nats-io/nkeys"
-
 	jwt "github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
 )
@@ -2457,10 +2455,6 @@ func TestLeafNodeRouteParseLSUnsub(t *testing.T) {
 }
 
 func TestLeafNodeOperatorBadCfg(t *testing.T) {
-	sysAcc, err := nkeys.CreateAccount()
-	require_NoError(t, err)
-	sysAccPk, err := sysAcc.PublicKey()
-	require_NoError(t, err)
 	tmpDir := createDir(t, "_nats-server")
 	defer removeDir(t, tmpDir)
 	for errorText, cfg := range map[string]string{
@@ -2480,7 +2474,6 @@ func TestLeafNodeOperatorBadCfg(t *testing.T) {
 			conf := createConfFile(t, []byte(fmt.Sprintf(`
 		port: -1
 		operator: %s
-		system_account: %s
 		resolver: {
 			type: cache
 			dir: %s
@@ -2488,7 +2481,7 @@ func TestLeafNodeOperatorBadCfg(t *testing.T) {
 		leafnodes: {
 			%s
 		}
-	`, ojwt, sysAccPk, tmpDir, cfg)))
+	`, ojwt, tmpDir, cfg)))
 			defer removeFile(t, conf)
 			opts := LoadConfig(conf)
 			s, err := NewServer(opts)
diff --git a/server/monitor_test.go b/server/monitor_test.go
index 6a321db5..756c3242 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -2821,7 +2821,7 @@ func TestMonitorLeafNode(t *testing.T) {
 		opts.LeafNode.TLSConfig != nil,
 		[]RemoteLeafOptsVarz{
 			{
-				"acc", 1, []string{"localhost:1234"}, nil,
+				"acc", 1, []string{"localhost:1234"},
 			},
 		},
 	}
@@ -2840,7 +2840,7 @@ func TestMonitorLeafNode(t *testing.T) {
 
 		// Having this here to make sure that if fields are added in ClusterOptsVarz,
 		// we make sure to update this test (compiler will report an error if we don't)
-		_ = LeafNodeOptsVarz{"", 0, 0, 0, false, false, []RemoteLeafOptsVarz{{"", 0, nil, nil}}}
+		_ = LeafNodeOptsVarz{"", 0, 0, 0, false, false, []RemoteLeafOptsVarz{{"", 0, nil}}}
 
 		// Alter the fields to make sure that we have a proper deep copy
 		// of what may be stored in the server. Anything we change here
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index 609c87d2..5d2aa909 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -36,6 +36,8 @@ import (
 	"github.com/nats-io/nuid"
 )
 
+var testMQTTTimeout = 4 * time.Second
+
 var jsClusterTemplWithMQTT = `
 	listen: 127.0.0.1:-1
 	server_name: %s
@@ -211,6 +213,9 @@ func testMQTTDefaultOptions() *Options {
 func testMQTTRunServer(t testing.TB, o *Options) *Server {
 	t.Helper()
 	o.NoLog = false
+	if o.StoreDir == _EMPTY_ {
+		o.StoreDir = createDir(t, "mqtt_js")
+	}
 	s, err := NewServer(o)
 	if err != nil {
 		t.Fatalf("Error creating server: %v", err)
@@ -256,8 +261,29 @@ func testMQTTDefaultTLSOptions(t *testing.T, verify bool) *Options {
 	return o
 }
 
+func TestMQTTStandaloneRequiresJetStream(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		mqtt {
+			port: -1
+			tls {
+				cert_file: "./configs/certs/server.pem"
+				key_file: "./configs/certs/key.pem"
+			}
+		}
+	`))
+	defer removeFile(t, conf)
+	o, err := ProcessConfigFile(conf)
+	if err != nil {
+		t.Fatalf("Error processing config file: %v", err)
+	}
+	if _, err := NewServer(o); err == nil || !strings.Contains(err.Error(), "standalone") {
+		t.Fatalf("Expected error about requiring JetStream in standalone mode, got %v", err)
+	}
+}
+
 func TestMQTTConfig(t *testing.T) {
 	conf := createConfFile(t, []byte(`
+		jetstream: enabled
 		mqtt {
 			port: -1
 			tls {
@@ -618,7 +644,7 @@ func testMQTTGetClient(t testing.TB, s *Server, clientID string) *client {
 func testMQTTRead(c net.Conn) ([]byte, error) {
 	var buf [512]byte
 	// Make sure that test does not block
-	c.SetReadDeadline(time.Now().Add(2 * time.Second))
+	c.SetReadDeadline(time.Now().Add(testMQTTTimeout))
 	n, err := c.Read(buf[:])
 	if err != nil {
 		return nil, err
@@ -628,7 +654,7 @@ func testMQTTRead(c net.Conn) ([]byte, error) {
 }
 
 func testMQTTWrite(c net.Conn, buf []byte) (int, error) {
-	c.SetWriteDeadline(time.Now().Add(2 * time.Second))
+	c.SetWriteDeadline(time.Now().Add(testMQTTTimeout))
 	n, err := c.Write(buf)
 	c.SetWriteDeadline(time.Time{})
 	return n, err
@@ -716,7 +742,7 @@ func mqttCreateConnectProto(ci *mqttConnInfo) []byte {
 
 func testMQTTCheckConnAck(t testing.TB, r *mqttReader, rc byte, sessionPresent bool) {
 	t.Helper()
-	r.reader.SetReadDeadline(time.Now().Add(2 * time.Second))
+	r.reader.SetReadDeadline(time.Now().Add(testMQTTTimeout))
 	if err := r.ensurePacketInBuffer(4); err != nil {
 		t.Fatalf("Error ensuring packet in buffer: %v", err)
 	}
@@ -1683,7 +1709,7 @@ func TestMQTTFilterConversion(t *testing.T) {
 
 func testMQTTReaderHasAtLeastOne(t testing.TB, r *mqttReader) {
 	t.Helper()
-	r.reader.SetReadDeadline(time.Now().Add(2 * time.Second))
+	r.reader.SetReadDeadline(time.Now().Add(testMQTTTimeout))
 	if err := r.ensurePacketInBuffer(1); err != nil {
 		t.Fatal(err)
 	}
@@ -2902,6 +2928,58 @@ func TestMQTTClusterReplicasCount(t *testing.T) {
 	}
 }
 
+func TestMQTTClusterPlacement(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "HUB", 3)
+	defer c.shutdown()
+
+	lnc := c.createLeafNodesWithStartPortAndMQTT("SPOKE", 3, 22111, `mqtt { listen: 127.0.0.1:-1 }`)
+	defer lnc.shutdown()
+
+	c.waitOnPeerCount(6)
+	c.waitOnLeader()
+
+	for i := 0; i < 10; i++ {
+		mc, rc := testMQTTConnect(t, &mqttConnInfo{cleanSess: true}, lnc.opts[i%3].MQTT.Host, lnc.opts[i%3].MQTT.Port)
+		defer mc.Close()
+		testMQTTCheckConnAck(t, rc, mqttConnAckRCConnectionAccepted, false)
+	}
+
+	// Now check that MQTT assets have been created in the LEAF node's side, not the Hub.
+	nc := natsConnect(t, lnc.servers[0].ClientURL())
+	defer nc.Close()
+	js, err := nc.JetStream()
+	if err != nil {
+		t.Fatalf("Unable to get JetStream: %v", err)
+	}
+	count := 0
+	for si := range js.StreamsInfo() {
+		if si.Cluster == nil || si.Cluster.Name != "SPOKE" {
+			t.Fatalf("Expected asset %q to be placed on spoke cluster, was placed on %+v", si.Config.Name, si.Cluster)
+		}
+		for _, repl := range si.Cluster.Replicas {
+			if !strings.HasPrefix(repl.Name, "SPOKE-") {
+				t.Fatalf("Replica on the wrong cluster: %+v", repl)
+			}
+		}
+		if si.State.Consumers > 0 {
+			for ci := range js.ConsumersInfo(si.Config.Name) {
+				if ci.Cluster == nil || ci.Cluster.Name != "SPOKE" {
+					t.Fatalf("Expected asset %q to be placed on spoke cluster, was placed on %+v", ci.Name, si.Cluster)
+				}
+				for _, repl := range ci.Cluster.Replicas {
+					if !strings.HasPrefix(repl.Name, "SPOKE-") {
+						t.Fatalf("Replica on the wrong cluster: %+v", repl)
+					}
+				}
+			}
+		}
+		count++
+	}
+	if count == 0 {
+		t.Fatal("No stream found!")
+	}
+}
+
 func TestMQTTParseUnsub(t *testing.T) {
 	eofr := testNewEOFReader()
 	for _, test := range []struct {