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

Issue:
Metrics for replication lag
## Feature Request
I would like metrics for replication lag.

#### Use Case:
There are issues we've seen where source replication lag spikes when looking at the output of `nats stream report`. We'd like to be able to alert on these in our monitoring infrastructure, but that's really difficult without getting the lag data itself. We've summed the stream message counts, but that can be a really sensitive alert.

#### Proposed Change:
Add replication lag metrics.

#### Who Benefits From The Change(s)?
Anyone who needs to monitor jetstream replication.

#### Alternative Approaches
We can sum total messages for streams and compare it to the stream that is being replicated into, but this seems really clunky. It works for mirroring just fine, so that's not as much of a concern right now.


Pull request:
[ADDED] Replication lag to stream monitoring
 - [x] Link to issue, e.g. `Resolves #NNN`
 - [ ] Documentation added (if applicable)
 - [x] Tests added
 - [x] Branch rebased on top of current main (`git pull --rebase origin main`)
 - [x] Changes squashed to a single commit (described [here](http://gitready.com/advanced/2009/02/10/squashing-commits-with-rebase.html))
 - [x] Build is green in Travis CI
 - [x] You have certified that the contribution is your original work and that you license the work to the project under the [Apache 2 license](https://github.com/nats-io/nats-server/blob/main/LICENSE)

[Resolves #2778 ](https://github.com/nats-io/nats-server/issues/2778)

### Changes proposed in this pull request:
- Adds ReplicationLagInfo list to StreamDetail
- Adds option to populate ReplicationLagInfo

/cc @nats-io/core


Patch:
diff --git a/server/accounts.go b/server/accounts.go
index eae82105..d2102589 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -731,16 +731,9 @@ func (a *Account) hasMappings() bool {
 		return false
 	}
 	a.mu.RLock()
-	hm := a.hasMappingsLocked()
+	n := len(a.mappings)
 	a.mu.RUnlock()
-	return hm
-}
-
-// Indicates we have mapping entries.
-// The account has been verified to be non-nil.
-// Read or Write lock held on entry.
-func (a *Account) hasMappingsLocked() bool {
-	return len(a.mappings) > 0
+	return n > 0
 }
 
 // This performs the logic to map to a new dest subject based on mappings.
diff --git a/server/auth.go b/server/auth.go
index e7b4d56a..f2e76692 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -171,19 +171,16 @@ func (p *Permissions) clone() *Permissions {
 // Lock is assumed held.
 func (s *Server) checkAuthforWarnings() {
 	warn := false
-	if s.opts.Password != _EMPTY_ && !isBcrypt(s.opts.Password) {
+	if s.opts.Password != "" && !isBcrypt(s.opts.Password) {
 		warn = true
 	}
 	for _, u := range s.users {
 		// Skip warn if using TLS certs based auth
 		// unless a password has been left in the config.
-		if u.Password == _EMPTY_ && s.opts.TLSMap {
-			continue
-		}
-		// Check if this is our internal sys client created on the fly.
-		if s.sysAccOnlyNoAuthUser != _EMPTY_ && u.Username == s.sysAccOnlyNoAuthUser {
+		if u.Password == "" && s.opts.TLSMap {
 			continue
 		}
+
 		if !isBcrypt(u.Password) {
 			warn = true
 			break
@@ -709,7 +706,7 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 		acc.mu.RLock()
 		c.Debugf("Authenticated JWT: %s %q (claim-name: %q, claim-tags: %q) "+
 			"signed with %q by Account %q (claim-name: %q, claim-tags: %q) signed with %q has mappings %t accused %p",
-			c.kindString(), juc.Subject, juc.Name, juc.Tags, juc.Issuer, issuer, acc.nameTag, acc.tags, acc.Issuer, acc.hasMappingsLocked(), acc)
+			c.kindString(), juc.Subject, juc.Name, juc.Tags, juc.Issuer, issuer, acc.nameTag, acc.tags, acc.Issuer, acc.hasMappings(), acc)
 		acc.mu.RUnlock()
 		return true
 	}
@@ -1145,7 +1142,7 @@ func validateAllowedConnectionTypes(m map[string]struct{}) error {
 }
 
 func validateNoAuthUser(o *Options, noAuthUser string) error {
-	if noAuthUser == _EMPTY_ {
+	if noAuthUser == "" {
 		return nil
 	}
 	if len(o.TrustedOperators) > 0 {
diff --git a/server/const.go b/server/const.go
index e178413f..c6e1b6db 100644
--- a/server/const.go
+++ b/server/const.go
@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.7.1-beta02"
+	VERSION = "2.7.1-beta01"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/monitor.go b/server/monitor.go
index 1f8c484b..6f604ea9 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -2388,11 +2388,13 @@ type JSzOptions struct {
 }
 
 type StreamDetail struct {
-	Name     string          `json:"name"`
-	Cluster  *ClusterInfo    `json:"cluster,omitempty"`
-	Config   *StreamConfig   `json:"config,omitempty"`
-	State    StreamState     `json:"state,omitempty"`
-	Consumer []*ConsumerInfo `json:"consumer_detail,omitempty"`
+	Name     string              `json:"name"`
+	Cluster  *ClusterInfo        `json:"cluster,omitempty"`
+	Config   *StreamConfig       `json:"config,omitempty"`
+	State    StreamState         `json:"state,omitempty"`
+	Consumer []*ConsumerInfo     `json:"consumer_detail,omitempty"`
+	Mirror   *StreamSourceInfo   `json:"mirror,omitempty"`
+	Sources  []*StreamSourceInfo `json:"sources,omitempty"`
 }
 
 type AccountDetail struct {
@@ -2469,6 +2471,8 @@ func (s *Server) accountDetail(jsa *jsAccount, optStreams, optConsumers, optCfg
 				State:   stream.state(),
 				Cluster: ci,
 				Config:  cfg,
+				Mirror:  stream.mirrorInfo(),
+				Sources: stream.sourcesInfo(),
 			}
 			if optConsumers {
 				for _, consumer := range stream.getPublicConsumers() {
diff --git a/server/opts.go b/server/opts.go
index fba65710..dc170786 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -663,7 +663,7 @@ func configureSystemAccount(o *Options, m map[string]interface{}) (retErr error)
 // or was present but set to false.
 func (o *Options) ProcessConfigFile(configFile string) error {
 	o.ConfigFile = configFile
-	if configFile == _EMPTY_ {
+	if configFile == "" {
 		return nil
 	}
 	m, err := conf.ParseFileWithChecks(configFile)
@@ -1018,7 +1018,7 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 				}
 			}
 			// In case "system_account" is defined as well, it takes precedence
-			if o.SystemAccount == _EMPTY_ {
+			if o.SystemAccount == "" {
 				o.SystemAccount = o.TrustedOperators[0].SystemAccount
 			}
 		}
diff --git a/server/parser.go b/server/parser.go
index d5c9b495..a9c3533a 100644
--- a/server/parser.go
+++ b/server/parser.go
@@ -147,7 +147,8 @@ func (c *client) parse(buf []byte) error {
 	// proper CONNECT if needed.
 	authSet := c.awaitingAuth()
 	// Snapshot max control line as well.
-	s, mcl, trace := c.srv, c.mcl, c.trace
+	mcl := c.mcl
+	trace := c.trace
 	c.mu.Unlock()
 
 	// Move to loop instead of range syntax to allow jumping of i
@@ -159,28 +160,7 @@ func (c *client) parse(buf []byte) error {
 			c.op = b
 			if b != 'C' && b != 'c' {
 				if authSet {
-					if s == nil {
-						goto authErr
-					}
-					var ok bool
-					// Check here for NoAuthUser. If this is set allow non CONNECT protos as our first.
-					// E.g. telnet proto demos.
-					if noAuthUser := s.getOpts().NoAuthUser; noAuthUser != _EMPTY_ {
-						s.mu.Lock()
-						user, exists := s.users[noAuthUser]
-						s.mu.Unlock()
-						if exists {
-							c.RegisterUser(user)
-							c.mu.Lock()
-							c.clearAuthTimer()
-							c.flags.set(connectReceived)
-							c.mu.Unlock()
-							authSet, ok = false, true
-						}
-					}
-					if !ok {
-						goto authErr
-					}
+					goto authErr
 				}
 				// If the connection is a gateway connection, make sure that
 				// if this is an inbound, it starts with a CONNECT.
diff --git a/server/server.go b/server/server.go
index 785460e7..1a44935f 100644
--- a/server/server.go
+++ b/server/server.go
@@ -2362,6 +2362,7 @@ func (s *Server) startMonitoring(secure bool) error {
 			}
 		}
 		srv.Close()
+		srv.Handler = nil
 		s.mu.Lock()
 		s.httpHandler = nil
 		s.mu.Unlock()


Test patch:
diff --git a/server/auth_test.go b/server/auth_test.go
index dcd87649..65cc6fcb 100644
--- a/server/auth_test.go
+++ b/server/auth_test.go
@@ -15,13 +15,11 @@ package server
 
 import (
 	"fmt"
-	"net"
 	"net/url"
 	"os"
 	"reflect"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
@@ -275,46 +273,3 @@ func TestNoAuthUser(t *testing.T) {
 		})
 	}
 }
-
-func TestNoAuthUserNoConnectProto(t *testing.T) {
-	conf := createConfFile(t, []byte(`
-		listen: "127.0.0.1:-1"
-		accounts {
-			A { users [{user: "foo", password: "pwd"}] }
-		}
-		authorization { timeout: 1 }
-		no_auth_user: "foo"
-	`))
-	defer os.Remove(conf)
-	s, o := RunServerWithConfig(conf)
-	defer s.Shutdown()
-
-	checkClients := func(n int) {
-		t.Helper()
-		time.Sleep(100 * time.Millisecond)
-		if nc := s.NumClients(); nc != n {
-			t.Fatalf("Expected %d clients, got %d", n, nc)
-		}
-	}
-
-	conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", o.Host, o.Port))
-	require_NoError(t, err)
-	defer conn.Close()
-	checkClientsCount(t, s, 1)
-
-	// With no auth user we should not require a CONNECT.
-	// Make sure we are good on not sending CONN first.
-	_, err = conn.Write([]byte("PUB foo 2\r\nok\r\n"))
-	require_NoError(t, err)
-	checkClients(1)
-	conn.Close()
-
-	// Now make sure we still do get timed out though.
-	conn, err = net.Dial("tcp", fmt.Sprintf("%s:%d", o.Host, o.Port))
-	require_NoError(t, err)
-	defer conn.Close()
-	checkClientsCount(t, s, 1)
-
-	time.Sleep(1200 * time.Millisecond)
-	checkClientsCount(t, s, 0)
-}
diff --git a/server/monitor_test.go b/server/monitor_test.go
index c4b75903..446b0a3c 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -4056,6 +4056,14 @@ func TestMonitorJsz(t *testing.T) {
 		Replicas: 1,
 	})
 	require_NoError(t, err)
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "my-stream-mirror",
+		Replicas: 2,
+		Mirror: &nats.StreamSource{
+			Name: "my-stream-replicated",
+		},
+	})
+	require_NoError(t, err)
 	_, err = js.AddConsumer("my-stream-replicated", &nats.ConsumerConfig{
 		Durable:   "my-consumer-replicated",
 		AckPolicy: nats.AckExplicitPolicy,
@@ -4066,9 +4074,16 @@ func TestMonitorJsz(t *testing.T) {
 		AckPolicy: nats.AckExplicitPolicy,
 	})
 	require_NoError(t, err)
+	_, err = js.AddConsumer("my-stream-mirror", &nats.ConsumerConfig{
+		Durable:   "my-consumer-mirror",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
 	nc.Flush()
 	_, err = js.Publish("foo", nil)
 	require_NoError(t, err)
+	// Wait for mirror replication
+	time.Sleep(100 * time.Millisecond)
 
 	monUrl1 := fmt.Sprintf("http://127.0.0.1:%d/jsz", 7501)
 	monUrl2 := fmt.Sprintf("http://127.0.0.1:%d/jsz", 5501)
@@ -4080,13 +4095,13 @@ func TestMonitorJsz(t *testing.T) {
 				t.Fatalf("expected no account to be returned by %s but got %v", url, info)
 			}
 			if info.Streams == 0 {
-				t.Fatalf("expected stream count to be 2 but got %d", info.Streams)
+				t.Fatalf("expected stream count to be 3 but got %d", info.Streams)
 			}
 			if info.Consumers == 0 {
-				t.Fatalf("expected consumer count to be 2 but got %d", info.Consumers)
+				t.Fatalf("expected consumer count to be 3 but got %d", info.Consumers)
 			}
-			if info.Messages != 1 {
-				t.Fatalf("expected one message but got %d", info.Messages)
+			if info.Messages != 2 {
+				t.Fatalf("expected two message but got %d", info.Messages)
 			}
 		}
 	})
@@ -4192,6 +4207,31 @@ func TestMonitorJsz(t *testing.T) {
 			}
 		}
 	})
+	t.Run("replication", func(t *testing.T) {
+		// The replication lag may only be present on the leader
+		replicationFound := false
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url + "?acc=ACC&streams=true")
+			if len(info.AccountDetails) != 1 {
+				t.Fatalf("expected account ACC to be returned by %s but got %v", url, info)
+			}
+			streamFound := false
+			for _, stream := range info.AccountDetails[0].Streams {
+				if stream.Name == "my-stream-mirror" {
+					streamFound = true
+					if stream.Mirror != nil {
+						replicationFound = true
+					}
+				}
+			}
+			if !streamFound {
+				t.Fatalf("Did not locate my-stream-mirror stream in results")
+			}
+		}
+		if !replicationFound {
+			t.Fatal("ReplicationLag expected to be present for my-stream-mirror stream")
+		}
+	})
 	t.Run("account-non-existing", func(t *testing.T) {
 		for _, url := range []string{monUrl1, monUrl2} {
 			info := readJsInfo(url + "?acc=DOES_NOT_EXIT")