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

Issue:
Add a tag map to the server configuration
## Feature Request

Add support for an arbitrary key/value map for configuration properties per server.

#### Use Case:

From an ops perspective it is useful to be able to associate meta-tags to an instance that would be queryable through the various mechanisms supported by the server.

The server name (which defaults to the server id) is a singular instance of why you'd want to tag an instance with more human-friendly metadata. But there are many other examples.

- Tag the server with the latitude and longitude of its data center.
- Tag the server with its cloud, region, and availability zone.
- Tag the server with its environment (e.g., test or production).

Note that we wouldn't expect this tagging to be unmanageably large, nor to change often. Perhaps limits as to number of tags and key/value sizing would be reasonable.

#### Proposed Change:

Add a string to string key/value dictionary to the nats server configuration. Read it on startup, and perhaps re-read it during reload signal processing. Emit its values in, say, `varz`, and/or make it queryable through server messages.

#### Who Benefits From The Change(s)?

Anyone who needs more domain-specific tagging in a large cloud of nats servers.

#### Alternative Approaches

The configured server name could be used as a key or key fragment to some external key/value store. But that adds extra moving parts where none might be needed.

Pull request description:
[added] support for tags and filter ping monitoring requests by tags
fixes #1588

Signed-off-by: Matthias Hanel <mh@synadia.com>

original issue asks for map, but I'd rather stick to the same label format we have in other places.
For now I also refrained from sending them between server.

Patch:
diff --git a/server/events.go b/server/events.go
index 2a76192a..a660bd92 100644
--- a/server/events.go
+++ b/server/events.go
@@ -957,9 +957,10 @@ func (s *Server) leafNodeConnected(sub *subscription, _ *client, subject, reply
 
 // Common filter options for system requests STATSZ VARZ SUBSZ CONNZ ROUTEZ GATEWAYZ LEAFZ
 type EventFilterOptions struct {
-	Name    string `json:"server_name,omitempty"` // filter by server name
-	Cluster string `json:"cluster,omitempty"`     // filter by cluster name
-	Host    string `json:"host,omitempty"`        // filter by host name
+	Name    string   `json:"server_name,omitempty"` // filter by server name
+	Cluster string   `json:"cluster,omitempty"`     // filter by cluster name
+	Host    string   `json:"host,omitempty"`        // filter by host name
+	Tags    []string `json:"tags,omitempty"`        // filter by tags (must match all tags)
 }
 
 // StatszEventOptions are options passed to Statsz
@@ -1033,6 +1034,14 @@ func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
 			return true
 		}
 	}
+	if len(fOpts.Tags) > 0 {
+		opts := s.getOpts()
+		for _, t := range fOpts.Tags {
+			if !opts.Tags.Contains(t) {
+				return true
+			}
+		}
+	}
 	return false
 }
 
diff --git a/server/monitor.go b/server/monitor.go
index 22b12fd3..5266e608 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1036,6 +1036,7 @@ type Varz struct {
 	Subscriptions     uint32            `json:"subscriptions"`
 	HTTPReqStats      map[string]uint64 `json:"http_req_stats"`
 	ConfigLoadTime    time.Time         `json:"config_load_time"`
+	Tags              jwt.TagList       `json:"tags,omitempty"`
 }
 
 // JetStreamVarz contains basic runtime information about jetstream
@@ -1228,6 +1229,7 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 		MaxSubs:  opts.MaxSubs,
 		Cores:    numCores,
 		MaxProcs: maxProcs,
+		Tags:     opts.Tags,
 	}
 	if len(opts.Routes) > 0 {
 		varz.Cluster.URLs = urlsToStrings(opts.Routes)
diff --git a/server/opts.go b/server/opts.go
index 915e01d3..9330a21a 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -247,6 +247,10 @@ type Options struct {
 	// that this applies to reconnect events.
 	ReconnectErrorReports int
 
+	// Tags describing the server. They will be included in varz
+	// and used as a filter criteria for some system requests
+	Tags jwt.TagList `json:"-"`
+
 	// private fields, used to know if bool options are explicitly
 	// defined in config and/or command line params.
 	inConfig  map[string]bool
@@ -1060,6 +1064,34 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 			*errors = append(*errors, err)
 			return
 		}
+	case "server_tags":
+		var err error
+		switch v := v.(type) {
+		case string:
+			o.Tags.Add(v)
+		case []string:
+			o.Tags.Add(v...)
+		case []interface{}:
+			for _, t := range v {
+				if t, ok := t.(token); ok {
+					if t, ok := t.Value().(string); ok {
+						o.Tags.Add(t)
+						continue
+					} else {
+						err = &configErr{tk, fmt.Sprintf("error parsing tags: unsupported type %T where string is expected", t)}
+					}
+				} else {
+					err = &configErr{tk, fmt.Sprintf("error parsing tags: unsupported type %T", t)}
+				}
+				break
+			}
+		default:
+			err = &configErr{tk, fmt.Sprintf("error parsing tags: unsupported type %T", v)}
+		}
+		if err != nil {
+			*errors = append(*errors, err)
+			return
+		}
 	default:
 		if au := atomic.LoadInt32(&allowUnknownTopLevelField); au == 0 && !tk.IsUsedVariable() {
 			err := &unknownConfigFieldErr{
diff --git a/server/reload.go b/server/reload.go
index a5339920..741452a5 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -788,8 +788,8 @@ func imposeOrder(value interface{}) error {
 		})
 	case WebsocketOpts:
 		sort.Strings(value.AllowedOrigins)
-	case string, bool, int, int32, int64, time.Duration, float64, nil,
-		LeafNodeOpts, ClusterOpts, *tls.Config, *URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts:
+	case string, bool, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config,
+		*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList:
 		// explicitly skipped types
 	default:
 		// this will fail during unit tests


Test patch:
diff --git a/server/dirstore_test.go b/server/dirstore_test.go
index e47dfef8..8d0d43c0 100644
--- a/server/dirstore_test.go
+++ b/server/dirstore_test.go
@@ -54,6 +54,15 @@ func require_NoError(t *testing.T, err error) {
 	}
 }
 
+func require_Contains(t *testing.T, s string, subStrs ...string) {
+	t.Helper()
+	for _, subStr := range subStrs {
+		if !strings.Contains(s, subStr) {
+			t.Fatalf("require %q to be contained in %q", subStr, s)
+		}
+	}
+}
+
 func require_Error(t *testing.T, err error) {
 	t.Helper()
 	if err == nil {
diff --git a/server/events_test.go b/server/events_test.go
index 7d5bcca2..0b9874ff 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -2178,3 +2178,90 @@ func TestServerEventsReceivedByQSubs(t *testing.T) {
 		t.Fatalf("Expected auth error, got %q", dem.Reason)
 	}
 }
+
+func TestServerEventsFilteredByTag(t *testing.T) {
+	confA := createConfFile(t, []byte(`
+		listen: -1
+		server_name: srv-A
+		server_tags: ["foo", "bar"]
+		cluster {
+			name: clust
+			listen: -1
+			no_advertise: true
+		}
+		system_account: SYS
+		accounts: {
+			SYS: {
+				users: [
+					{user: b, password: b}
+				]
+			}
+		}
+		no_auth_user: b
+    `))
+	defer os.Remove(confA)
+	sA, _ := RunServerWithConfig(confA)
+	defer sA.Shutdown()
+	confB := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: -1
+		server_name: srv-B
+		server_tags: ["bar", "baz"]
+		cluster {
+			name: clust
+			listen: -1
+			no_advertise: true
+			routes [
+				nats-route://localhost:%d
+			]
+		}
+		system_account: SYS
+		accounts: {
+			SYS: {
+				users: [
+					{user: b, password: b}
+				]
+			}
+		}
+		no_auth_user: b
+    `, sA.opts.Cluster.Port)))
+	defer os.Remove(confB)
+	sB, _ := RunServerWithConfig(confB)
+	defer sB.Shutdown()
+	checkClusterFormed(t, sA, sB)
+	nc := natsConnect(t, sA.ClientURL())
+	defer nc.Close()
+
+	ib := nats.NewInbox()
+	req := func(tags ...string) {
+		t.Helper()
+		r, err := json.Marshal(VarzEventOptions{EventFilterOptions: EventFilterOptions{Tags: tags}})
+		require_NoError(t, err)
+		err = nc.PublishRequest(fmt.Sprintf(serverPingReqSubj, "VARZ"), ib, r)
+		require_NoError(t, err)
+	}
+
+	msgs := make(chan *nats.Msg, 10)
+	defer close(msgs)
+	_, err := nc.ChanSubscribe(ib, msgs)
+	require_NoError(t, err)
+	req("none")
+	select {
+	case <-msgs:
+		t.Fatalf("no message expected")
+	case <-time.After(200 * time.Millisecond):
+	}
+	req("foo")
+	m := <-msgs
+	require_Contains(t, string(m.Data), "srv-A", "foo", "bar")
+	req("foo", "bar")
+	m = <-msgs
+	require_Contains(t, string(m.Data), "srv-A", "foo", "bar")
+	req("baz")
+	m = <-msgs
+	require_Contains(t, string(m.Data), "srv-B", "bar", "baz")
+	req("bar")
+	m1 := <-msgs
+	m2 := <-msgs
+	require_Contains(t, string(m1.Data)+string(m2.Data), "srv-A", "srv-B", "foo", "bar", "baz")
+	require_Len(t, len(msgs), 0)
+}
diff --git a/server/monitor_test.go b/server/monitor_test.go
index 51dc8f58..3955b7b9 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -51,6 +51,7 @@ func DefaultMonitorOptions() *Options {
 		ServerName:   "monitor_server",
 		NoLog:        true,
 		NoSigs:       true,
+		Tags:         []string{"tag"},
 	}
 }
 
@@ -244,6 +245,9 @@ func TestHandleVarz(t *testing.T) {
 		if v.Name != "monitor_server" {
 			t.Fatal("Expected ServerName to be 'monitor_server'")
 		}
+		if !v.Tags.Contains("tag") {
+			t.Fatal("Expected tags to be 'tag'")
+		}
 	}
 
 	// Test JSONP