streaming: store services with a unique ID that includes namespace

agent/cache-types/streaming_health_services.go

@@ -154,17 +154,18 @@ func (s *healthView) Update(events []*pbsubscribe.Event) error {
 			return fmt.Errorf("unexpected event type for service health view: %T",
 				event.GetPayload())
 		}
-		node := serviceHealth.CheckServiceNode
-		id := fmt.Sprintf("%s/%s", node.Node.Node, node.Service.ID)
 
+		id := serviceHealth.CheckServiceNode.UniqueID()
 		switch serviceHealth.Op {
 		case pbsubscribe.CatalogOp_Register:
-			checkServiceNode := pbservice.CheckServiceNodeToStructs(serviceHealth.CheckServiceNode)
-			s.state[id] = *checkServiceNode
+			csn := pbservice.CheckServiceNodeToStructs(serviceHealth.CheckServiceNode)
+			s.state[id] = *csn
 		case pbsubscribe.CatalogOp_Deregister:
 			delete(s.state, id)
 		}
 	}
+	// TODO(streaming): should this filter be applied to only the new CheckServiceNode
+	// instead of the full map, which should already be filtered.
 	if s.filter != nil {
 		filtered, err := s.filter.Execute(s.state)
 		if err != nil {

proto/pbservice/ids.go

@@ -0,0 +1,29 @@
+package pbservice
+
+import (
+	"strings"
+)
+
+// UniqueID returns a unique identifier for this CheckServiceNode, which includes
+// the node name, service namespace, and service ID.
+//
+// The returned ID uses slashes to separate the identifiers, however the node name
+// may also contain a slash, so it is not possible to parse this identifier to
+// retrieve its constituent parts.
+//
+// This function is similar to structs.UniqueID, however at this time no guarantees
+// are made that it will remain the same.
+func (m *CheckServiceNode) UniqueID() string {
+	if m == nil {
+		return ""
+	}
+	builder := new(strings.Builder)
+	if m.Node != nil {
+		builder.WriteString(m.Node.Node + "/")
+	}
+	if m.Service != nil {
+		builder.WriteString(m.Service.EnterpriseMeta.Namespace + "/")
+		builder.WriteString(m.Service.ID)
+	}
+	return builder.String()
+}

proto/pbservice/ids_test.go

@@ -0,0 +1,67 @@
+package pbservice
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/hashicorp/consul/proto/pbcommon"
+)
+
+func TestCheckServiceNode_UniqueID(t *testing.T) {
+	type testCase struct {
+		name     string
+		csn      CheckServiceNode
+		expected string
+	}
+	fn := func(t *testing.T, tc testCase) {
+		require.Equal(t, tc.expected, tc.csn.UniqueID())
+	}
+
+	var testCases = []testCase{
+		{
+			name: "full",
+			csn: CheckServiceNode{
+				Node: &Node{Node: "the-node-name"},
+				Service: &NodeService{
+					ID:             "the-service-id",
+					EnterpriseMeta: pbcommon.EnterpriseMeta{Namespace: "the-namespace"},
+				},
+			},
+			expected: "the-node-name/the-namespace/the-service-id",
+		},
+		{
+			name: "without node",
+			csn: CheckServiceNode{
+				Service: &NodeService{
+					ID:             "the-service-id",
+					EnterpriseMeta: pbcommon.EnterpriseMeta{Namespace: "the-namespace"},
+				},
+			},
+			expected: "the-namespace/the-service-id",
+		},
+		{
+			name: "without service",
+			csn: CheckServiceNode{
+				Node: &Node{Node: "the-node-name"},
+			},
+			expected: "the-node-name/",
+		},
+		{
+			name: "without namespace",
+			csn: CheckServiceNode{
+				Node: &Node{Node: "the-node-name"},
+				Service: &NodeService{
+					ID: "the-service-id",
+				},
+			},
+			expected: "the-node-name//the-service-id",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fn(t, tc)
+		})
+	}
+
+}