Repository: nats-io/nats-server
Pull request #1605 (2020-09-22 19:03:54)

Expose SUBSZ and CONNZ monitoring endpoints for accounts
Signed-off-by: Matthias Hanel <mh@synadia.com>

This is the low hanging fruit.
Take existing monitoring endpoints that allow to filter by account.
Expose these endpoints on an account specific subject. 
Extract account id from subject and set the account filter. (this will overwrite whatever other value may have been specified) 

I also moved CONNS into that setup loop.



Patch:
diff --git a/server/events.go b/server/events.go
index e8aace92..71c558cb 100644
--- a/server/events.go
+++ b/server/events.go
@@ -39,7 +39,7 @@ const (
 
 	connectEventSubj    = "$SYS.ACCOUNT.%s.CONNECT"
 	disconnectEventSubj = "$SYS.ACCOUNT.%s.DISCONNECT"
-	accConnsReqSubj     = "$SYS.REQ.ACCOUNT.%s.CONNS"
+	accReqSubj          = "$SYS.REQ.ACCOUNT.%s.%s"
 	// kept for backward compatibility when using http resolver
 	// this overlaps with the names for events but you'd have to have the operator private key in order to succeed.
 	accUpdateEventSubjOld    = "$SYS.ACCOUNT.%s.CLAIMS.UPDATE"
@@ -587,11 +587,6 @@ func (s *Server) initEventTracking() {
 	if _, err := s.sysSubscribe(subject, s.remoteConnsUpdate); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 	}
-	// Listen for broad requests to respond with account info.
-	subject = fmt.Sprintf(accConnsReqSubj, "*")
-	if _, err := s.sysSubscribe(subject, s.connsRequest); err != nil {
-		s.Errorf("Error setting up internal tracking: %v", err)
-	}
 	// Listen for broad requests to respond with number of subscriptions for a given subject.
 	if _, err := s.sysSubscribe(accNumSubsReqSubj, s.nsubsRequest); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
@@ -655,6 +650,44 @@ func (s *Server) initEventTracking() {
 			s.Errorf("Error setting up internal tracking: %v", err)
 		}
 	}
+	extractAccount := func(subject string) (string, error) {
+		if tk := strings.Split(subject, tsep); len(tk) != accReqTokens {
+			return "", fmt.Errorf("subject %q is malformed", subject)
+		} else {
+			return tk[accReqAccIndex], nil
+		}
+	}
+	monAccSrvc := map[string]msgHandler{
+		"SUBSZ": func(sub *subscription, _ *client, subject, reply string, msg []byte) {
+			optz := &SubszEventOptions{}
+			s.zReq(reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+				if acc, err := extractAccount(subject); err != nil {
+					return nil, err
+				} else {
+					optz.SubszOptions.Subscriptions = true
+					optz.SubszOptions.Account = acc
+					return s.Subsz(&optz.SubszOptions)
+				}
+			})
+		},
+		"CONNZ": func(sub *subscription, _ *client, subject, reply string, msg []byte) {
+			optz := &ConnzEventOptions{}
+			s.zReq(reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+				if acc, err := extractAccount(subject); err != nil {
+					return nil, err
+				} else {
+					optz.ConnzOptions.Account = acc
+					return s.Connz(&optz.ConnzOptions)
+				}
+			})
+		},
+		"CONNS": s.connsRequest,
+	}
+	for name, req := range monAccSrvc {
+		if _, err := s.sysSubscribe(fmt.Sprintf(accReqSubj, "*", name), req); err != nil {
+			s.Errorf("Error setting up internal tracking: %v", err)
+		}
+	}
 
 	// Listen for updates when leaf nodes connect for a given account. This will
 	// force any gateway connections to move to `modeInterestOnly`
@@ -822,7 +855,7 @@ func (s *Server) connsRequest(sub *subscription, _ *client, subject, reply strin
 	if !s.eventsRunning() {
 		return
 	}
-	tk := strings.Split(subject, ".")
+	tk := strings.Split(subject, tsep)
 	if len(tk) != accReqTokens {
 		s.sys.client.Errorf("Bad subject account connections request message")
 		return
@@ -1054,7 +1087,7 @@ func (s *Server) enableAccountTracking(a *Account) {
 	// May need to ensure we do so only if there is a known interest.
 	// This can get complicated with gateways.
 
-	subj := fmt.Sprintf(accConnsReqSubj, a.Name)
+	subj := fmt.Sprintf(accReqSubj, a.Name, "CONNS")
 	reply := fmt.Sprintf(connsRespSubj, s.info.ID)
 	m := accNumConnsReq{Account: a.Name}
 	s.sendInternalMsg(subj, reply, &m.Server, &m)


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index 0606261e..3c204b0b 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1203,6 +1203,61 @@ func TestAccountClaimsUpdates(t *testing.T) {
 	})
 }
 
+func TestAccountReqMonitoring(t *testing.T) {
+	s, opts := runTrustedServer(t)
+	defer s.Shutdown()
+	sacc, sakp := createAccount(s)
+	s.setSystemAccount(sacc)
+	acc, akp := createAccount(s)
+	if acc == nil {
+		t.Fatalf("did not create account")
+	}
+	subsz := fmt.Sprintf(accReqSubj, acc.Name, "SUBSZ")
+	connz := fmt.Sprintf(accReqSubj, acc.Name, "CONNZ")
+	// Create system account connection to query
+	url := fmt.Sprintf("nats://%s:%d", opts.Host, opts.Port)
+	ncSys, err := nats.Connect(url, createUserCreds(t, s, sakp))
+	if err != nil {
+		t.Fatalf("Error on connect: %v", err)
+	}
+	defer ncSys.Close()
+	// Create a connection that we can query
+	nc, err := nats.Connect(url, createUserCreds(t, s, akp))
+	if err != nil {
+		t.Fatalf("Error on connect: %v", err)
+	}
+	defer nc.Close()
+	// query SUBSZ for account
+	if resp, err := ncSys.Request(subsz, nil, time.Second); err != nil {
+		t.Fatalf("Error on request: %v", err)
+	} else if !strings.Contains(string(resp.Data), `"num_subscriptions": 0`) {
+		t.Fatalf("unexpected subs count (expected 0): %v", string(resp.Data))
+	}
+	// create a subscription
+	if sub, err := nc.Subscribe("foo", func(msg *nats.Msg) {}); err != nil {
+		t.Fatalf("error on subscribe %v", err)
+	} else {
+		defer sub.Unsubscribe()
+	}
+	nc.Flush()
+	// query SUBSZ for account
+	if resp, err := ncSys.Request(subsz, nil, time.Second); err != nil {
+		t.Fatalf("Error on request: %v", err)
+	} else if !strings.Contains(string(resp.Data), `"num_subscriptions": 1`) {
+		t.Fatalf("unexpected subs count (expected 1): %v", string(resp.Data))
+	} else if !strings.Contains(string(resp.Data), `"subject": "foo"`) {
+		t.Fatalf("expected subscription foo: %v", string(resp.Data))
+	}
+	// query connections for account
+	if resp, err := ncSys.Request(connz, nil, time.Second); err != nil {
+		t.Fatalf("Error on request: %v", err)
+	} else if !strings.Contains(string(resp.Data), `"num_connections": 1`) {
+		t.Fatalf("unexpected subs count (expected 1): %v", string(resp.Data))
+	} else if !strings.Contains(string(resp.Data), `"total": 2`) { // includes system acc connection
+		t.Fatalf("unexpected subs count (expected 1): %v", string(resp.Data))
+	}
+}
+
 func TestAccountClaimsUpdatesWithServiceImports(t *testing.T) {
 	s, opts := runTrustedServer(t)
 	defer s.Shutdown()
@@ -1398,7 +1453,7 @@ func TestSystemAccountWithGateways(t *testing.T) {
 
 	// If this tests fails with wrong number after 10 seconds we may have
 	// added a new inititial subscription for the eventing system.
-	checkExpectedSubs(t, 28, sa)
+	checkExpectedSubs(t, 30, sa)
 
 	// Create a client on B and see if we receive the event
 	urlb := fmt.Sprintf("nats://%s:%d", ob.Host, ob.Port)