fix(kubernetes): emit each topology identity once per snapshot

pkg/collector/corechecks/cluster/kubeapi/README.md

@@ -23,6 +23,13 @@ Default: **enabled**
 ## Kubernetes Api Topology Check (_WIP_)
 This fetches all kubernetes topology components.
 
+Each snapshot emits a component or relation identity only once, even when several
+collectors discover it concurrently. Examples include shared host paths and a
+container mounting multiple files from one ConfigMap. The first observation is
+retained, matching the receiver's existing handling of duplicates; identifiers
+are unchanged. Relations still require both endpoints. Identity caches belong
+to one snapshot, so subsequent snapshots continue to refresh existing topology.
+
 The check is enabled using the: `collect_kubernetes_topology` config parameter or the `STS_COLLECT_KUBERNETES_TOPOLOGY` environment variable.
 
 Default: **enabled**

pkg/collector/corechecks/cluster/topologycollectors/common.go

@@ -69,17 +69,18 @@ type ClusterTopologyCommon interface {
 }
 
 type clusterTopologyCommon struct {
-	Instance                      topology.Instance
-	ClusterType                   ClusterType
-	APICollectorClient            apiserver.APICollectorClient
-	urn                           urn.Builder
-	componentChan                 chan<- *topology.Component
-	componentIDCache              sync.Map
-	relationChan                  chan<- *topology.Relation
-	possibleRelations             []*topology.Relation
-	k8sVersion                    *version.Info
-	useRelationCache              bool
-	relationCacheMu               sync.Mutex
+	Instance           topology.Instance
+	ClusterType        ClusterType
+	APICollectorClient apiserver.APICollectorClient
+	urn                urn.Builder
+	componentChan      chan<- *topology.Component
+	componentIDCache   sync.Map
+	relationChan       chan<- *topology.Relation
+	relationIDCache    sync.Map
+	possibleRelations  []*topology.Relation
+	k8sVersion         *version.Info
+	useRelationCache   bool
+	relationCacheMu    sync.Mutex
 }
 
 // NewClusterTopologyCommon creates a clusterTopologyCommon
@@ -92,28 +93,33 @@ func NewClusterTopologyCommon(
 	k8sVersion *version.Info,
 ) ClusterTopologyCommon {
 	return &clusterTopologyCommon{
-		Instance:                      instance,
-		ClusterType:                   clusterType,
-		APICollectorClient:            ac,
-		urn:                           urn.NewURNBuilder(urn.ClusterTypeFromString(instance.Type), instance.URL),
-		componentChan:                 componentChan,
-		componentIDCache:              sync.Map{},
-		relationChan:                  relationChan,
-		k8sVersion:                    k8sVersion,
-		useRelationCache:              true,
+		Instance:           instance,
+		ClusterType:        clusterType,
+		APICollectorClient: ac,
+		urn:                urn.NewURNBuilder(urn.ClusterTypeFromString(instance.Type), instance.URL),
+		componentChan:      componentChan,
+		componentIDCache:   sync.Map{},
+		relationChan:       relationChan,
+		k8sVersion:         k8sVersion,
+		useRelationCache:   true,
 	}
 }
 
-// SubmitComponent sends a component to the Component channel and adds its External ID to the cache if the relation cache is being used
+// SubmitComponent sends the first observation of each component identity in this snapshot.
+// Collectors can concurrently discover shared resources, such as host paths and endpoints.
 func (c *clusterTopologyCommon) SubmitComponent(component *topology.Component) {
-	c.componentChan <- component
-	if c.useRelationCache {
-		c.componentIDCache.Store(component.ExternalID, true)
+	if _, exists := c.componentIDCache.LoadOrStore(component.ExternalID, true); exists {
+		return
 	}
+	c.componentChan <- component
 }
 
-// SubmitRelation sends a relation to the Relation channel or adds it to the possibleRelations cache if it's being used
+// SubmitRelation accepts the first observation of each relation identity in this snapshot,
+// sending it immediately or waiting until its endpoints have been collected.
 func (c *clusterTopologyCommon) SubmitRelation(relation *topology.Relation) {
+	if _, exists := c.relationIDCache.LoadOrStore(relation.ExternalID, true); exists {
+		return
+	}
 	if c.useRelationCache {
 		_, sourceExists := c.componentIDCache.Load(relation.SourceID)
 		_, targetExists := c.componentIDCache.Load(relation.TargetID)
@@ -132,6 +138,7 @@ func (c *clusterTopologyCommon) SubmitRelation(relation *topology.Relation) {
 func (c *clusterTopologyCommon) CorrelateRelations() {
 	c.relationCacheMu.Lock()
 	possibleRelations := c.possibleRelations
+	c.possibleRelations = nil
 	c.relationCacheMu.Unlock()
 	for _, relation := range possibleRelations {
 		_, sourceExists := c.componentIDCache.Load(relation.SourceID)

pkg/collector/corechecks/cluster/topologycollectors/ingress_collector_test.go

@@ -60,13 +60,12 @@ func TestIngressCollector_1_18(t *testing.T) {
 			},
 		},
 		{
+			// Shared endpoint components were emitted with the first ingress.
 			testCase: "Test Service 2 - Default Backend",
 			assertions: []func(*testing.T, chan *topology.Component, chan *topology.Relation){
 				expectIngress212(creationTimeFormatted),
 				expectRelationIngress212Service(),
-				expectEndpointIP212(),
 				expectRelationEndpointIP21Ingress212(),
-				expectEndpointAmazon212(),
 				expectRelationEndpointAmazonIngress212(),
 			},
 		},
@@ -77,9 +76,7 @@ func TestIngressCollector_1_18(t *testing.T) {
 				expectRelationIngress213Service1(),
 				expectRelationIngress213Service2(),
 				expectRelationIngress213Service3(),
-				expectEndpointIP213(),
 				expectRelationEndpointIPIngress213(),
-				expectEndpointAmazon213(),
 				expectRelationEndpointAmazonIngress213(),
 			},
 		},
@@ -128,13 +125,12 @@ func TestIngressCollector_1_22(t *testing.T) {
 			},
 		},
 		{
+			// Shared endpoint components were emitted with the first ingress.
 			testCase: "Test Service 2 - Default Backend",
 			assertions: []func(*testing.T, chan *topology.Component, chan *topology.Relation){
 				expectIngress222(creationTimeFormatted),
 				expectRelationIngressService222(),
-				expectEndpointIP222(),
 				expectRelationEndpointIPIngress222(),
-				expectEndpointAmazon22(),
 				expectRelationEndpointAmazonIngress222(),
 			},
 		},
@@ -145,9 +141,7 @@ func TestIngressCollector_1_22(t *testing.T) {
 				expectRelationIngress223Service1(),
 				expectRelationIngress223Service2(),
 				expectRelationIngress223Service3(),
-				expectEndpointIP223(),
 				expectRelationEndpointIP22Ingress223(),
-				expectEndpointAmazon22(),
 				expectRelationEndpointAmazonIngress223(),
 			},
 		},
@@ -170,26 +164,6 @@ func expectRelationEndpointAmazonIngress213() func(*testing.T, chan *topology.Co
 	})
 }
 
-func expectEndpointAmazon213() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
-	return expectComponent(&topology.Component{
-		ExternalID: "urn:endpoint:/test-cluster-name:64047e8f24bb48e9a406ac8286ee8b7d.eu-west-1.elb.amazonaws.com",
-		Type:       topology.Type{Name: "endpoint"},
-		Data: topology.Data{
-			"name":              "64047e8f24bb48e9a406ac8286ee8b7d.eu-west-1.elb.amazonaws.com",
-			"kind":              "Endpoint",
-			"creationTimestamp": creationTime,
-			"tags": map[string]string{
-				"test":           "label",
-				"cluster-name":   "test-cluster-name",
-				"cluster-type":   "kubernetes",
-				"component-type": "kubernetes-endpoint",
-				"namespace":      "test-namespace",
-			},
-			"identifiers": []string{},
-		},
-	})
-}
-
 func expectRelationEndpointIPIngress213() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
 	return expectRelation(&topology.Relation{
 		ExternalID: "urn:endpoint:/test-cluster-name:34.100.200.15->urn:kubernetes:/test-cluster-name:test-namespace:ingress/test-ingress-3",
@@ -200,26 +174,6 @@ func expectRelationEndpointIPIngress213() func(*testing.T, chan *topology.Compon
 	})
 }
 
-func expectEndpointIP213() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
-	return expectComponent(&topology.Component{
-		ExternalID: "urn:endpoint:/test-cluster-name:34.100.200.15",
-		Type:       topology.Type{Name: "endpoint"},
-		Data: topology.Data{
-			"name":              "34.100.200.15",
-			"kind":              "Endpoint",
-			"creationTimestamp": creationTime,
-			"tags": map[string]string{
-				"test":           "label",
-				"cluster-name":   "test-cluster-name",
-				"cluster-type":   "kubernetes",
-				"component-type": "kubernetes-endpoint",
-				"namespace":      "test-namespace",
-			},
-			"identifiers": []string{},
-		},
-	})
-}
-
 func expectRelationIngress213Service3() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
 	return expectRelation(&topology.Relation{
 		ExternalID: "urn:kubernetes:/test-cluster-name:test-namespace:ingress/test-ingress-3->" +
@@ -334,26 +288,6 @@ func expectRelationEndpointAmazonIngress212() func(*testing.T, chan *topology.Co
 	})
 }
 
-func expectEndpointAmazon212() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
-	return expectComponent(&topology.Component{
-		ExternalID: "urn:endpoint:/test-cluster-name:64047e8f24bb48e9a406ac8286ee8b7d.eu-west-1.elb.amazonaws.com",
-		Type:       topology.Type{Name: "endpoint"},
-		Data: topology.Data{
-			"name":              "64047e8f24bb48e9a406ac8286ee8b7d.eu-west-1.elb.amazonaws.com",
-			"kind":              "Endpoint",
-			"creationTimestamp": creationTime,
-			"tags": map[string]string{
-				"test":           "label",
-				"cluster-name":   "test-cluster-name",
-				"cluster-type":   "kubernetes",
-				"component-type": "kubernetes-endpoint",
-				"namespace":      "test-namespace",
-			},
-			"identifiers": []string{},
-		},
-	})
-}
-
 func expectRelationEndpointIP21Ingress212() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
 	return expectRelation(&topology.Relation{
 		ExternalID: "urn:endpoint:/test-cluster-name:34.100.200.15->urn:kubernetes:/test-cluster-name:test-namespace:ingress/test-ingress-2",
@@ -364,26 +298,6 @@ func expectRelationEndpointIP21Ingress212() func(*testing.T, chan *topology.Comp
 	})
 }
 
-func expectEndpointIP212() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
-	return expectComponent(&topology.Component{
-		ExternalID: "urn:endpoint:/test-cluster-name:34.100.200.15",
-		Type:       topology.Type{Name: "endpoint"},
-		Data: topology.Data{
-			"name":              "34.100.200.15",
-			"kind":              "Endpoint",
-			"creationTimestamp": creationTime,
-			"tags": map[string]string{
-				"test":           "label",
-				"cluster-name":   "test-cluster-name",
-				"cluster-type":   "kubernetes",
-				"component-type": "kubernetes-endpoint",
-				"namespace":      "test-namespace",
-			},
-			"identifiers": []string{},
-		},
-	})
-}
-
 func expectRelationIngress212Service() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
 	return expectRelation(&topology.Relation{
 		ExternalID: "urn:kubernetes:/test-cluster-name:test-namespace:ingress/test-ingress-2->" +
@@ -592,26 +506,6 @@ func expectRelationEndpointIP22Ingress223() func(*testing.T, chan *topology.Comp
 	})
 }
 
-func expectEndpointIP223() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
-	return expectComponent(&topology.Component{
-		ExternalID: "urn:endpoint:/test-cluster-name:34.100.200.22",
-		Type:       topology.Type{Name: "endpoint"},
-		Data: topology.Data{
-			"name":              "34.100.200.22",
-			"kind":              "Endpoint",
-			"creationTimestamp": creationTime,
-			"tags": map[string]string{
-				"test":           "label22",
-				"cluster-name":   "test-cluster-name",
-				"cluster-type":   "kubernetes",
-				"component-type": "kubernetes-endpoint",
-				"namespace":      "test-namespace",
-			},
-			"identifiers": []string{},
-		},
-	})
-}
-
 func expectRelationIngress223Service3() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
 	return expectRelation(&topology.Relation{
 		ExternalID: "urn:kubernetes:/test-cluster-name:test-namespace:ingress/test-ingress22-3->" +
@@ -748,26 +642,6 @@ func expectRelationEndpointIPIngress222() func(*testing.T, chan *topology.Compon
 	})
 }
 
-func expectEndpointIP222() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
-	return expectComponent(&topology.Component{
-		ExternalID: "urn:endpoint:/test-cluster-name:34.100.200.22",
-		Type:       topology.Type{Name: "endpoint"},
-		Data: topology.Data{
-			"name":              "34.100.200.22",
-			"kind":              "Endpoint",
-			"creationTimestamp": creationTime,
-			"tags": map[string]string{
-				"test":           "label22",
-				"cluster-name":   "test-cluster-name",
-				"cluster-type":   "kubernetes",
-				"component-type": "kubernetes-endpoint",
-				"namespace":      "test-namespace",
-			},
-			"identifiers": []string{},
-		},
-	})
-}
-
 func expectRelationIngressService222() func(*testing.T, chan *topology.Component, chan *topology.Relation) {
 	return expectRelation(&topology.Relation{
 		ExternalID: "urn:kubernetes:/test-cluster-name:test-namespace:ingress/test-ingress22-2->" +

pkg/collector/corechecks/cluster/topologycollectors/snapshot_dedup_test.go

@@ -0,0 +1,87 @@
+//go:build kubeapiserver
+
+package topologycollectors
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/StackVista/stackstate-receiver-go-client/pkg/model/topology"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSnapshotKeepsFirstComponentAndRelation(t *testing.T) {
+	for _, cacheRelations := range []bool{true, false} {
+		t.Run(fmt.Sprintf("cacheRelations=%t", cacheRelations), func(t *testing.T) {
+			components := make(chan *topology.Component, 10)
+			relations := make(chan *topology.Relation, 10)
+			common := NewClusterTopologyCommon(topology.Instance{Type: "kubernetes", URL: "test"}, Kubernetes, nil,
+				components, relations, nil)
+			common.SetUseRelationCache(cacheRelations)
+
+			// A container can mount two files from the same ConfigMap. Both
+			// observations describe one relation, even when endpoints arrive later.
+			firstRelation := common.CreateRelationData("container", "configmap", "uses", map[string]interface{}{"mount_path": "/first"})
+			common.SubmitRelation(firstRelation)
+			firstComponent := &topology.Component{ExternalID: "container", Data: map[string]interface{}{"name": "first"}}
+			common.SubmitComponent(firstComponent)
+			common.SubmitComponent(&topology.Component{ExternalID: "configmap"})
+			common.SubmitComponent(&topology.Component{ExternalID: "container", Data: map[string]interface{}{"name": "second"}})
+			common.SubmitRelation(common.CreateRelationData("container", "configmap", "uses", map[string]interface{}{"mount_path": "/second"}))
+			common.CorrelateRelations()
+			common.CorrelateRelations()
+
+			require.Len(t, components, 2)
+			assert.Same(t, firstComponent, <-components)
+			require.Len(t, relations, 1)
+			assert.Same(t, firstRelation, <-relations)
+		})
+	}
+}
+
+func TestSnapshotDeduplicationIsConcurrentAndScopedToOneSnapshot(t *testing.T) {
+	const submitters, identities = 8, 50
+	components := make(chan *topology.Component, submitters*identities*2)
+	relations := make(chan *topology.Relation, submitters*identities*2)
+	for snapshot := 0; snapshot < 2; snapshot++ {
+		common := NewClusterTopologyCommon(topology.Instance{Type: "kubernetes", URL: "test"}, Kubernetes, nil,
+			components, relations, nil)
+		var submitted sync.WaitGroup
+		for s := 0; s < submitters; s++ {
+			submitted.Add(1)
+			go func() {
+				defer submitted.Done()
+				for i := 0; i < identities; i++ {
+					id := fmt.Sprintf("component-%d", i)
+					common.SubmitRelation(common.CreateRelation(id, "component-0", "uses"))
+					common.SubmitComponent(&topology.Component{ExternalID: id})
+				}
+			}()
+		}
+		submitted.Wait()
+		common.CorrelateRelations()
+		require.Len(t, components, identities)
+		require.Len(t, relations, identities)
+		seenComponents, seenRelations := map[string]bool{}, map[string]bool{}
+		for i := 0; i < identities; i++ {
+			component, relation := <-components, <-relations
+			assert.False(t, seenComponents[component.ExternalID])
+			assert.False(t, seenRelations[relation.ExternalID])
+			seenComponents[component.ExternalID], seenRelations[relation.ExternalID] = true, true
+		}
+	}
+}
+
+func TestSnapshotStillDropsRelationsWithMissingEndpoints(t *testing.T) {
+	components := make(chan *topology.Component, 2)
+	relations := make(chan *topology.Relation, 2)
+	common := NewClusterTopologyCommon(topology.Instance{Type: "kubernetes", URL: "test"}, Kubernetes, nil,
+		components, relations, nil)
+	common.SubmitComponent(&topology.Component{ExternalID: "source"})
+	common.SubmitRelation(common.CreateRelation("source", "missing", "uses"))
+	common.SubmitRelation(common.CreateRelation("missing", "source", "uses"))
+	common.CorrelateRelations()
+	assert.Empty(t, relations)
+}