From 5a41036559248a2b65403bd95fbeb7ff358ed166 Mon Sep 17 00:00:00 2001 From: Thiago Bertoldi Date: Wed, 14 Oct 2026 13:30:54 -0300 Subject: [PATCH 1/2] fix(kubernetes): emit each topology identity once per snapshot --- .../corechecks/cluster/kubeapi/README.md | 7 + .../cluster/topologycollectors/common.go | 57 ++++---- .../ingress_collector_test.go | 130 +----------------- .../topologycollectors/snapshot_dedup_test.go | 87 ++++++++++++ 4 files changed, 128 insertions(+), 153 deletions(-) create mode 100644 pkg/collector/corechecks/cluster/topologycollectors/snapshot_dedup_test.go diff --git a/pkg/collector/corechecks/cluster/kubeapi/README.md b/pkg/collector/corechecks/cluster/kubeapi/README.md index 475fc7d51471..0a8660e19713 100644 --- a/pkg/collector/corechecks/cluster/kubeapi/README.md +++ b/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** diff --git a/pkg/collector/corechecks/cluster/topologycollectors/common.go b/pkg/collector/corechecks/cluster/topologycollectors/common.go index c1b977008d8e..f298dfb3afb0 100644 --- a/pkg/collector/corechecks/cluster/topologycollectors/common.go +++ b/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) diff --git a/pkg/collector/corechecks/cluster/topologycollectors/ingress_collector_test.go b/pkg/collector/corechecks/cluster/topologycollectors/ingress_collector_test.go index f1342ad8bf6e..8856d4d99888 100644 --- a/pkg/collector/corechecks/cluster/topologycollectors/ingress_collector_test.go +++ b/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->" + diff --git a/pkg/collector/corechecks/cluster/topologycollectors/snapshot_dedup_test.go b/pkg/collector/corechecks/cluster/topologycollectors/snapshot_dedup_test.go new file mode 100644 index 000000000000..308517e3f3e3 --- /dev/null +++ b/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) +} From a5562597d88cafb62f6903298de50058c0a56371 Mon Sep 17 00:00:00 2001 From: Thiago Bertoldi Date: Wed, 14 Oct 2026 14:09:08 -0300 Subject: [PATCH 2/2] docs: note Kubernetes snapshot deduplication fix --- ...s-topology-snapshot-deduplication-c08e24704fa84ff7.yaml | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 releasenotes-dca/notes/kubernetes-topology-snapshot-deduplication-c08e24704fa84ff7.yaml diff --git a/releasenotes-dca/notes/kubernetes-topology-snapshot-deduplication-c08e24704fa84ff7.yaml b/releasenotes-dca/notes/kubernetes-topology-snapshot-deduplication-c08e24704fa84ff7.yaml new file mode 100644 index 000000000000..dde76fa7072f --- /dev/null +++ b/releasenotes-dca/notes/kubernetes-topology-snapshot-deduplication-c08e24704fa84ff7.yaml @@ -0,0 +1,7 @@ +fixes: + - | + Emit each Kubernetes topology component and relation identity only once + per snapshot. This prevents duplicate-item synchronization errors when + workloads share endpoints or host paths, or a container mounts multiple + files from the same ConfigMap. Existing identities and updates in later + snapshots are preserved.