Repository: nats-io/nats.go
Pull request #1231 (2023-03-12 11:51:40)

support metadata for micro

ripienaar: Implements https://github.com/nats-io/nats-architecture-and-design/pull/205
coveralls: 
[![Coverage Status](https://coveralls.io/builds/57765296/badge)](https://coveralls.io/builds/57765296)

Coverage: 85.229% (+0.05%) from 85.176% when pulling **9c49d99c31a04dc7872086eb7424d03c1dc9b95d on ripienaar:micro_metadata** into **3c6d9b83464299fddbf4b44131304160587e9041 on nats-io:main**.

ripienaar: @piotrpio tests green now, pls merge when happy.

Patch:
diff --git a/micro/service.go b/micro/service.go
index 9d413de..79f0cc9 100644
--- a/micro/service.go
+++ b/micro/service.go
@@ -74,8 +74,9 @@ type (
 	EndpointOpt func(*endpointOpts) error
 
 	endpointOpts struct {
-		subject string
-		schema  *Schema
+		subject  string
+		schema   *Schema
+		metadata map[string]string
 	}
 
 	// ErrHandler is a function used to configure a custom error handler for a service,
@@ -90,9 +91,10 @@ type (
 
 	// ServiceIdentity contains fields helping to identity a service instance.
 	ServiceIdentity struct {
-		Name    string `json:"name"`
-		ID      string `json:"id"`
-		Version string `json:"version"`
+		Name     string            `json:"name"`
+		ID       string            `json:"id"`
+		Version  string            `json:"version"`
+		Metadata map[string]string `json:"metadata"`
 	}
 
 	// Stats is the type returned by STATS monitoring endpoint.
@@ -139,9 +141,10 @@ type (
 	}
 
 	EndpointSchema struct {
-		Name    string `json:"name"`
-		Subject string `json:"subject"`
-		Schema  Schema `json:"schema,omitempty"`
+		Name     string            `json:"name"`
+		Subject  string            `json:"subject"`
+		Metadata map[string]string `json:"metadata"`
+		Schema   Schema            `json:"schema,omitempty"`
 	}
 
 	// Schema can be used to configure a schema for a service.
@@ -187,6 +190,9 @@ type (
 		// APIURL is an optional url pointing to API specification.
 		APIURL string `json:"api_url"`
 
+		// Metadata annotates the service
+		Metadata map[string]string `json:"metadata,omitempty"`
+
 		// StatsHandler is a user-defined custom function.
 		// used to calculate additional service stats.
 		StatsHandler StatsHandler
@@ -207,6 +213,9 @@ type (
 
 		// Schema is an optional request/response endpoint schema.
 		Schema *Schema
+
+		// Metadata annotates the service
+		Metadata map[string]string `json:"metadata,omitempty"`
 	}
 
 	// NATSError represents an error returned by a NATS Subscription.
@@ -331,11 +340,6 @@ func AddService(nc *nats.Conn, config Config) (Service, error) {
 		verbSubs:  make(map[string]*nats.Subscription),
 		endpoints: make([]*Endpoint, 0),
 	}
-	svcIdentity := ServiceIdentity{
-		Name:    config.Name,
-		ID:      id,
-		Version: config.Version,
-	}
 
 	svc.setupAsyncCallbacks()
 
@@ -346,6 +350,9 @@ func AddService(nc *nats.Conn, config Config) (Service, error) {
 		if config.Endpoint.Schema != nil {
 			opts = append(opts, WithEndpointSchema(config.Endpoint.Schema))
 		}
+		if config.Endpoint.Metadata != nil {
+			opts = append(opts, WithEndpointMetadata(config.Endpoint.Metadata))
+		}
 		if err := svc.AddEndpoint("default", config.Endpoint.Handler, opts...); err != nil {
 			svc.asyncDispatcher.close()
 			return nil, err
@@ -363,7 +370,7 @@ func AddService(nc *nats.Conn, config Config) (Service, error) {
 	}
 
 	ping := Ping{
-		ServiceIdentity: svcIdentity,
+		ServiceIdentity: svc.serviceIdentity(),
 		Type:            PingResponseType,
 	}
 	pingHandler := func(req Request) {
@@ -426,10 +433,10 @@ func (s *service) AddEndpoint(name string, handler Handler, opts ...EndpointOpt)
 	if options.subject != "" {
 		subject = options.subject
 	}
-	return addEndpoint(s, name, subject, handler, options.schema)
+	return addEndpoint(s, name, subject, handler, options.schema, options.metadata)
 }
 
-func addEndpoint(s *service, name, subject string, handler Handler, schema *Schema) error {
+func addEndpoint(s *service, name, subject string, handler Handler, schema *Schema, metadata map[string]string) error {
 	if !nameRegexp.MatchString(name) {
 		return fmt.Errorf("%w: invalid endpoint name", ErrConfigValidation)
 	}
@@ -439,9 +446,10 @@ func addEndpoint(s *service, name, subject string, handler Handler, schema *Sche
 	endpoint := &Endpoint{
 		service: s,
 		EndpointConfig: EndpointConfig{
-			Subject: subject,
-			Handler: handler,
-			Schema:  schema,
+			Subject:  subject,
+			Handler:  handler,
+			Schema:   schema,
+			Metadata: metadata,
 		},
 	}
 	sub, err := s.nc.QueueSubscribe(
@@ -684,21 +692,27 @@ func restoreAsyncHandlers(nc *nats.Conn, handlers handlers) {
 	nc.SetErrorHandler(handlers.asyncErr)
 }
 
+func (s *service) serviceIdentity() ServiceIdentity {
+	return ServiceIdentity{
+		Name:     s.Config.Name,
+		ID:       s.id,
+		Version:  s.Config.Version,
+		Metadata: s.Config.Metadata,
+	}
+}
+
 // Info returns information about the service
 func (s *service) Info() Info {
 	endpoints := make([]string, 0, len(s.endpoints))
 	for _, e := range s.endpoints {
 		endpoints = append(endpoints, e.Subject)
 	}
+
 	return Info{
-		ServiceIdentity: ServiceIdentity{
-			Name:    s.Config.Name,
-			ID:      s.id,
-			Version: s.Config.Version,
-		},
-		Type:        InfoResponseType,
-		Description: s.Config.Description,
-		Subjects:    endpoints,
+		ServiceIdentity: s.serviceIdentity(),
+		Type:            InfoResponseType,
+		Description:     s.Config.Description,
+		Subjects:        endpoints,
 	}
 }
 
@@ -706,16 +720,12 @@ func (s *service) Info() Info {
 func (s *service) Stats() Stats {
 	s.m.Lock()
 	defer s.m.Unlock()
-	info := s.Info()
+
 	stats := Stats{
-		ServiceIdentity: ServiceIdentity{
-			Name:    info.Name,
-			ID:      info.ID,
-			Version: info.Version,
-		},
-		Endpoints: make([]*EndpointStats, 0),
-		Type:      StatsResponseType,
-		Started:   s.started,
+		ServiceIdentity: s.serviceIdentity(),
+		Endpoints:       make([]*EndpointStats, 0),
+		Type:            StatsResponseType,
+		Started:         s.started,
 	}
 	for _, endpoint := range s.endpoints {
 		endpointStats := &EndpointStats{
@@ -763,14 +773,15 @@ func (s *service) schema() SchemaResp {
 		}
 
 		endpoints = append(endpoints, EndpointSchema{
-			Name:    e.stats.Name,
-			Subject: e.stats.Subject,
-			Schema:  schema,
+			Name:     e.stats.Name,
+			Subject:  e.stats.Subject,
+			Metadata: e.EndpointConfig.Metadata,
+			Schema:   schema,
 		})
 	}
 
 	return SchemaResp{
-		ServiceIdentity: s.Info().ServiceIdentity,
+		ServiceIdentity: s.serviceIdentity(),
 		Type:            SchemaResponseType,
 		APIURL:          s.APIURL,
 		Endpoints:       endpoints,
@@ -796,7 +807,7 @@ func (g *group) AddEndpoint(name string, handler Handler, opts ...EndpointOpt) e
 	if g.prefix == "" {
 		endpointSubject = subject
 	}
-	return addEndpoint(g.service, name, endpointSubject, handler, options.schema)
+	return addEndpoint(g.service, name, endpointSubject, handler, options.schema, options.metadata)
 }
 
 func (g *group) AddGroup(name string) Group {
@@ -875,3 +886,10 @@ func WithEndpointSchema(schema *Schema) EndpointOpt {
 		return nil
 	}
 }
+
+func WithEndpointMetadata(metadata map[string]string) EndpointOpt {
+	return func(e *endpointOpts) error {
+		e.metadata = metadata
+		return nil
+	}
+}


Test patch:
diff --git a/micro/test/service_test.go b/micro/test/service_test.go
index ac5af67..d168fbe 100644
--- a/micro/test/service_test.go
+++ b/micro/test/service_test.go
@@ -67,6 +67,7 @@ func TestServiceBasics(t *testing.T) {
 		Name:        "CoolAddService",
 		Version:     "0.1.0",
 		Description: "Add things together",
+		Metadata:    map[string]string{"basic": "metadata"},
 		Endpoint: &micro.EndpointConfig{
 			Subject: "svc.add",
 			Handler: micro.HandlerFunc(doAdd),
@@ -98,6 +99,9 @@ func TestServiceBasics(t *testing.T) {
 		if len(info.Description) == 0 || len(info.Version) == 0 {
 			t.Fatalf("Expected non empty description and version")
 		}
+		if !reflect.DeepEqual(info.Metadata, map[string]string{"basic": "metadata"}) {
+			t.Fatalf("invalid metadata: %v", info.Metadata)
+		}
 	}
 
 	// Make sure we can request info, 1 response.
@@ -206,14 +210,16 @@ func TestAddService(t *testing.T) {
 		{
 			name: "minimal config",
 			givenConfig: micro.Config{
-				Name:    "test_service",
-				Version: "0.1.0",
+				Name:     "test_service",
+				Version:  "0.1.0",
+				Metadata: map[string]string{"basic": "metadata"},
 			},
 			expectedPing: micro.Ping{
 				Type: micro.PingResponseType,
 				ServiceIdentity: micro.ServiceIdentity{
-					Name:    "test_service",
-					Version: "0.1.0",
+					Name:     "test_service",
+					Version:  "0.1.0",
+					Metadata: map[string]string{"basic": "metadata"},
 				},
 			},
 		},
@@ -223,8 +229,9 @@ func TestAddService(t *testing.T) {
 				Name:    "test_service",
 				Version: "0.1.0",
 				Endpoint: &micro.EndpointConfig{
-					Subject: "test",
-					Handler: micro.HandlerFunc(testHandler),
+					Subject:  "test",
+					Handler:  micro.HandlerFunc(testHandler),
+					Metadata: map[string]string{"basic": "endpoint_metadata"},
 				},
 			},
 			expectedPing: micro.Ping{
@@ -476,7 +483,7 @@ func TestAddService(t *testing.T) {
 			}
 
 			test.expectedPing.ID = info.ID
-			if test.expectedPing != ping {
+			if !reflect.DeepEqual(test.expectedPing, ping) {
 				t.Fatalf("Invalid ping response; want: %+v; got: %+v", test.expectedPing, ping)
 			}
 
@@ -752,8 +759,9 @@ func TestMonitoringHandlers(t *testing.T) {
 		APIURL:       "http://someapi.com/v1",
 		ErrorHandler: errHandler,
 		Endpoint: &micro.EndpointConfig{
-			Subject: "test.func",
-			Handler: micro.HandlerFunc(func(r micro.Request) {}),
+			Subject:  "test.func",
+			Handler:  micro.HandlerFunc(func(r micro.Request) {}),
+			Metadata: map[string]string{"basic": "schema"},
 			Schema: &micro.Schema{
 				Request:  "request_schema",
 				Response: "response_schema",
@@ -867,8 +875,9 @@ func TestMonitoringHandlers(t *testing.T) {
 				APIURL: "http://someapi.com/v1",
 				Endpoints: []micro.EndpointSchema{
 					{
-						Name:    "default",
-						Subject: "test.func",
+						Name:     "default",
+						Subject:  "test.func",
+						Metadata: map[string]string{"basic": "schema"},
 						Schema: micro.Schema{
 							Request:  "request_schema",
 							Response: "response_schema",
@@ -890,8 +899,9 @@ func TestMonitoringHandlers(t *testing.T) {
 				APIURL: "http://someapi.com/v1",
 				Endpoints: []micro.EndpointSchema{
 					{
-						Name:    "default",
-						Subject: "test.func",
+						Name:     "default",
+						Subject:  "test.func",
+						Metadata: map[string]string{"basic": "schema"},
 						Schema: micro.Schema{
 							Request:  "request_schema",
 							Response: "response_schema",
@@ -913,8 +923,9 @@ func TestMonitoringHandlers(t *testing.T) {
 				APIURL: "http://someapi.com/v1",
 				Endpoints: []micro.EndpointSchema{
 					{
-						Name:    "default",
-						Subject: "test.func",
+						Name:     "default",
+						Subject:  "test.func",
+						Metadata: map[string]string{"basic": "schema"},
 						Schema: micro.Schema{
 							Request:  "request_schema",
 							Response: "response_schema",