Support multiple tags for health and catalog http api endpoints (#4717)

* Support multiple tags for health and catalog api endpoints Fixes #1781. Adds a `ServiceTags` field to the ServiceSpecificRequest to support multiple tags, updates the filter logic in the catalog store, and propagates these change through to the health and catalog endpoints. Note: Leaves `ServiceTag` in the struct, since it is being used as part of the DNS lookup, which in turn uses the health check. * Update the api package to support multiple tags Includes additional tests. * Update new tests to use the `require` library * Update HealthConnect check after a bad merge
hashicorp · Oct 11, 2018 · 34e5516 · 34e5516
1 parent 51b33ef
commit 34e5516
Show file tree

Hide file tree

Showing 13 changed files with 399 additions and 127 deletions.
diff --git a/agent/catalog_endpoint.go b/agent/catalog_endpoint.go
@@ -188,7 +188,7 @@ func (s *HTTPServer) catalogServiceNodes(resp http.ResponseWriter, req *http.Req
 	// Check for a tag
 	params := req.URL.Query()
 	if _, ok := params["tag"]; ok {
-		args.ServiceTag = params.Get("tag")
+		args.ServiceTags = params["tag"]
 		args.TagFilter = true
 	}
 

diff --git a/agent/consul/catalog_endpoint.go b/agent/consul/catalog_endpoint.go
@@ -273,7 +273,7 @@ func (c *Catalog) ServiceNodes(args *structs.ServiceSpecificRequest, reply *stru
 			}
 
 			if args.TagFilter {
-				return s.ServiceTagNodes(ws, args.ServiceName, args.ServiceTag)
+				return s.ServiceTagNodes(ws, args.ServiceName, args.ServiceTags)
 			}
 
 			return s.ServiceNodes(ws, args.ServiceName)
@@ -334,6 +334,14 @@ func (c *Catalog) ServiceNodes(args *structs.ServiceSpecificRequest, reply *stru
 			metrics.IncrCounterWithLabels([]string{"catalog", key, "query-tag"}, 1,
 				[]metrics.Label{{Name: "service", Value: args.ServiceName}, {Name: "tag", Value: args.ServiceTag}})
 		}
+		if len(args.ServiceTags) > 0 {
+			// Build metric labels
+			labels := []metrics.Label{{Name: "service", Value: args.ServiceName}}
+			for _, tag := range args.ServiceTags {
+				labels = append(labels, metrics.Label{Name: "tag", Value: tag})
+			}
+			metrics.IncrCounterWithLabels([]string{"catalog", key, "query-tags"}, 1, labels)
+		}
 		if len(reply.ServiceNodes) == 0 {
 			metrics.IncrCounterWithLabels([]string{"catalog", key, "not-found"}, 1,
 				[]metrics.Label{{Name: "service", Value: args.ServiceName}})

diff --git a/agent/consul/catalog_endpoint_test.go b/agent/consul/catalog_endpoint_test.go
@@ -1589,7 +1589,7 @@ func TestCatalog_ListServiceNodes(t *testing.T) {
 	args := structs.ServiceSpecificRequest{
 		Datacenter:  "dc1",
 		ServiceName: "db",
-		ServiceTag:  "slave",
+		ServiceTags: []string{"slave"},
 		TagFilter:   false,
 	}
 	var out structs.IndexedServiceNodes
@@ -1647,16 +1647,16 @@ func TestCatalog_ListServiceNodes_NodeMetaFilter(t *testing.T) {
 	if err := s1.fsm.State().EnsureNode(2, node2); err != nil {
 		t.Fatalf("err: %v", err)
 	}
-	if err := s1.fsm.State().EnsureService(3, "foo", &structs.NodeService{ID: "db", Service: "db", Tags: []string{"primary"}, Address: "127.0.0.1", Port: 5000}); err != nil {
+	if err := s1.fsm.State().EnsureService(3, "foo", &structs.NodeService{ID: "db", Service: "db", Tags: []string{"primary", "v2"}, Address: "127.0.0.1", Port: 5000}); err != nil {
 		t.Fatalf("err: %v", err)
 	}
-	if err := s1.fsm.State().EnsureService(4, "bar", &structs.NodeService{ID: "db2", Service: "db", Tags: []string{"secondary"}, Address: "127.0.0.2", Port: 5000}); err != nil {
+	if err := s1.fsm.State().EnsureService(4, "bar", &structs.NodeService{ID: "db2", Service: "db", Tags: []string{"secondary", "v2"}, Address: "127.0.0.2", Port: 5000}); err != nil {
 		t.Fatalf("err: %v", err)
 	}
 
 	cases := []struct {
 		filters  map[string]string
-		tag      string
+		tags     []string
 		services structs.ServiceNodes
 	}{
 		// Basic meta filter
@@ -1667,7 +1667,7 @@ func TestCatalog_ListServiceNodes_NodeMetaFilter(t *testing.T) {
 		// Basic meta filter, tag
 		{
 			filters:  map[string]string{"somekey": "somevalue"},
-			tag:      "primary",
+			tags:     []string{"primary"},
 			services: structs.ServiceNodes{&structs.ServiceNode{Node: "foo", ServiceID: "db"}},
 		},
 		// Common meta filter
@@ -1681,7 +1681,7 @@ func TestCatalog_ListServiceNodes_NodeMetaFilter(t *testing.T) {
 		// Common meta filter, tag
 		{
 			filters: map[string]string{"common": "1"},
-			tag:     "secondary",
+			tags:    []string{"secondary"},
 			services: structs.ServiceNodes{
 				&structs.ServiceNode{Node: "bar", ServiceID: "db2"},
 			},
@@ -1699,7 +1699,22 @@ func TestCatalog_ListServiceNodes_NodeMetaFilter(t *testing.T) {
 		// Multiple filter values, tag
 		{
 			filters:  map[string]string{"somekey": "somevalue", "common": "1"},
-			tag:      "primary",
+			tags:     []string{"primary"},
+			services: structs.ServiceNodes{&structs.ServiceNode{Node: "foo", ServiceID: "db"}},
+		},
+		// Common meta filter, single tag
+		{
+			filters: map[string]string{"common": "1"},
+			tags:    []string{"v2"},
+			services: structs.ServiceNodes{
+				&structs.ServiceNode{Node: "bar", ServiceID: "db2"},
+				&structs.ServiceNode{Node: "foo", ServiceID: "db"},
+			},
+		},
+		// Common meta filter, multiple tags
+		{
+			filters:  map[string]string{"common": "1"},
+			tags:     []string{"v2", "primary"},
 			services: structs.ServiceNodes{&structs.ServiceNode{Node: "foo", ServiceID: "db"}},
 		},
 	}
@@ -1709,17 +1724,12 @@ func TestCatalog_ListServiceNodes_NodeMetaFilter(t *testing.T) {
 			Datacenter:      "dc1",
 			NodeMetaFilters: tc.filters,
 			ServiceName:     "db",
-			ServiceTag:      tc.tag,
-			TagFilter:       tc.tag != "",
+			ServiceTags:     tc.tags,
+			TagFilter:       len(tc.tags) > 0,
 		}
 		var out structs.IndexedServiceNodes
-		if err := msgpackrpc.CallWithCodec(codec, "Catalog.ServiceNodes", &args, &out); err != nil {
-			t.Fatalf("err: %v", err)
-		}
-
-		if len(out.ServiceNodes) != len(tc.services) {
-			t.Fatalf("bad: %v", out)
-		}
+		require.NoError(t, msgpackrpc.CallWithCodec(codec, "Catalog.ServiceNodes", &args, &out))
+		require.Len(t, out.ServiceNodes, len(tc.services))
 
 		for i, serviceNode := range out.ServiceNodes {
 			if serviceNode.Node != tc.services[i].Node || serviceNode.ServiceID != tc.services[i].ServiceID {

diff --git a/agent/consul/health_endpoint.go b/agent/consul/health_endpoint.go
@@ -170,6 +170,13 @@ func (h *Health) ServiceNodes(args *structs.ServiceSpecificRequest, reply *struc
 			metrics.IncrCounterWithLabels([]string{"health", key, "query-tag"}, 1,
 				[]metrics.Label{{Name: "service", Value: args.ServiceName}, {Name: "tag", Value: args.ServiceTag}})
 		}
+		if len(args.ServiceTags) > 0 {
+			labels := []metrics.Label{{Name: "service", Value: args.ServiceName}}
+			for _, tag := range args.ServiceTags {
+				labels = append(labels, metrics.Label{Name: "tag", Value: tag})
+			}
+			metrics.IncrCounterWithLabels([]string{"health", key, "query-tags"}, 1, labels)
+		}
 		if len(reply.Nodes) == 0 {
 			metrics.IncrCounterWithLabels([]string{"health", key, "not-found"}, 1,
 				[]metrics.Label{{Name: "service", Value: args.ServiceName}})
@@ -186,7 +193,12 @@ func (h *Health) serviceNodesConnect(ws memdb.WatchSet, s *state.Store, args *st
 }
 
 func (h *Health) serviceNodesTagFilter(ws memdb.WatchSet, s *state.Store, args *structs.ServiceSpecificRequest) (uint64, structs.CheckServiceNodes, error) {
-	return s.CheckServiceTagNodes(ws, args.ServiceName, args.ServiceTag)
+	// DNS service lookups populate the ServiceTag field. In this case,
+	// use ServiceTag instead of the ServiceTags field.
+	if args.ServiceTag != "" {
+		return s.CheckServiceTagNodes(ws, args.ServiceName, []string{args.ServiceTag})
+	}
+	return s.CheckServiceTagNodes(ws, args.ServiceName, args.ServiceTags)
 }
 
 func (h *Health) serviceNodesDefault(ws memdb.WatchSet, s *state.Store, args *structs.ServiceSpecificRequest) (uint64, structs.CheckServiceNodes, error) {

diff --git a/agent/consul/health_endpoint_test.go b/agent/consul/health_endpoint_test.go
@@ -11,6 +11,7 @@ import (
 	"github.com/hashicorp/consul/testrpc"
 	"github.com/hashicorp/net-rpc-msgpackrpc"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestHealth_ChecksInState(t *testing.T) {
@@ -602,6 +603,68 @@ func TestHealth_ServiceNodes(t *testing.T) {
 	}
 }
 
+func TestHealth_ServiceNodes_MultipleServiceTags(t *testing.T) {
+	t.Parallel()
+	dir1, s1 := testServer(t)
+	defer os.RemoveAll(dir1)
+	defer s1.Shutdown()
+	codec := rpcClient(t, s1)
+	defer codec.Close()
+
+	testrpc.WaitForLeader(t, s1.RPC, "dc1")
+
+	arg := structs.RegisterRequest{
+		Datacenter: "dc1",
+		Node:       "foo",
+		Address:    "127.0.0.1",
+		Service: &structs.NodeService{
+			ID:      "db",
+			Service: "db",
+			Tags:    []string{"master", "v2"},
+		},
+		Check: &structs.HealthCheck{
+			Name:      "db connect",
+			Status:    api.HealthPassing,
+			ServiceID: "db",
+		},
+	}
+	var out struct{}
+	require.NoError(t, msgpackrpc.CallWithCodec(codec, "Catalog.Register", &arg, &out))
+
+	arg = structs.RegisterRequest{
+		Datacenter: "dc1",
+		Node:       "bar",
+		Address:    "127.0.0.2",
+		Service: &structs.NodeService{
+			ID:      "db",
+			Service: "db",
+			Tags:    []string{"slave", "v2"},
+		},
+		Check: &structs.HealthCheck{
+			Name:      "db connect",
+			Status:    api.HealthWarning,
+			ServiceID: "db",
+		},
+	}
+	require.NoError(t, msgpackrpc.CallWithCodec(codec, "Catalog.Register", &arg, &out))
+
+	var out2 structs.IndexedCheckServiceNodes
+	req := structs.ServiceSpecificRequest{
+		Datacenter:  "dc1",
+		ServiceName: "db",
+		ServiceTags: []string{"master", "v2"},
+		TagFilter:   true,
+	}
+	require.NoError(t, msgpackrpc.CallWithCodec(codec, "Health.ServiceNodes", &req, &out2))
+
+	nodes := out2.Nodes
+	require.Len(t, nodes, 1)
+	require.Equal(t, nodes[0].Node.Node, "foo")
+	require.Contains(t, nodes[0].Service.Tags, "v2")
+	require.Contains(t, nodes[0].Service.Tags, "master")
+	require.Equal(t, nodes[0].Checks[0].Status, api.HealthPassing)
+}
+
 func TestHealth_ServiceNodes_NodeMetaFilter(t *testing.T) {
 	t.Parallel()
 	dir1, s1 := testServer(t)

diff --git a/agent/consul/state/catalog.go b/agent/consul/state/catalog.go
@@ -909,8 +909,8 @@ func (s *Store) serviceNodes(ws memdb.WatchSet, serviceName string, connect bool
 }
 
 // ServiceTagNodes returns the nodes associated with a given service, filtering
-// out services that don't contain the given tag.
-func (s *Store) ServiceTagNodes(ws memdb.WatchSet, service string, tag string) (uint64, structs.ServiceNodes, error) {
+// out services that don't contain the given tags.
+func (s *Store) ServiceTagNodes(ws memdb.WatchSet, service string, tags []string) (uint64, structs.ServiceNodes, error) {
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
@@ -928,7 +928,7 @@ func (s *Store) ServiceTagNodes(ws memdb.WatchSet, service string, tag string) (
 	var results structs.ServiceNodes
 	for service := services.Next(); service != nil; service = services.Next() {
 		svc := service.(*structs.ServiceNode)
-		if !serviceTagFilter(svc, tag) {
+		if !serviceTagsFilter(svc, tags) {
 			results = append(results, svc)
 		}
 	}
@@ -957,6 +957,20 @@ func serviceTagFilter(sn *structs.ServiceNode, tag string) bool {
 	return true
 }
 
+// serviceTagsFilter returns true (should filter) if the given service node
+// doesn't contain the given set of tags.
+func serviceTagsFilter(sn *structs.ServiceNode, tags []string) bool {
+	for _, tag := range tags {
+		if serviceTagFilter(sn, tag) {
+			// If any one of the expected tags was not found, filter the service
+			return true
+		}
+	}
+
+	// If all tags were found, don't filter the service
+	return false
+}
+
 // ServiceAddressNodes returns the nodes associated with a given service, filtering
 // out services that don't match the given serviceAddress
 func (s *Store) ServiceAddressNodes(ws memdb.WatchSet, address string) (uint64, structs.ServiceNodes, error) {
@@ -1645,7 +1659,7 @@ func (s *Store) checkServiceNodes(ws memdb.WatchSet, serviceName string, connect
 
 // CheckServiceTagNodes is used to query all nodes and checks for a given
 // service, filtering out services that don't contain the given tag.
-func (s *Store) CheckServiceTagNodes(ws memdb.WatchSet, serviceName, tag string) (uint64, structs.CheckServiceNodes, error) {
+func (s *Store) CheckServiceTagNodes(ws memdb.WatchSet, serviceName string, tags []string) (uint64, structs.CheckServiceNodes, error) {
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
@@ -1663,7 +1677,7 @@ func (s *Store) CheckServiceTagNodes(ws memdb.WatchSet, serviceName, tag string)
 	var results structs.ServiceNodes
 	for service := iter.Next(); service != nil; service = iter.Next() {
 		svc := service.(*structs.ServiceNode)
-		if !serviceTagFilter(svc, tag) {
+		if !serviceTagsFilter(svc, tags) {
 			results = append(results, svc)
 		}
 	}