feat(autodiscovery): tag configuration-discovery instances to mitigate duplicate metrics risk (#54660)

### What does this PR do?

Adds a `dd_config_discovery:true` tag to every check instance scheduled via the Autodiscovery configuration-discovery mechanism (i.e. any `auto_conf.yaml` template with `discovery: {}`, resolved through `comp/core/autodiscovery/impl/configmgr_discovery.go`'s `applyDiscoveredConfigsLocked`).

The tag used is `dd_config_discovery:true`, a plain `dd_`-prefixed key, following the precedent of other agent-added, customer-visible marker/provenance tags already in the codebase:
- `dd_remote_config_id` / `dd_remote_config_rev` (`comp/core/tagger/tags/tags.go`)
- `dd_enable_check_intake` (`pkg/collector/worker/worker.go`)

### Motivation

[DSCVR-651](https://datadoghq.atlassian.net/browse/DSCVR-651): there is a risk that an agent on host A is monitoring a service on host B with a manually-configured check (e.g. a generic `openmetrics` check), while the agent running locally on host B also autodiscovers and schedules a dedicated integration for the same service via configuration discovery. Neither agent's local anti-duplication logic can see the other's config, so both submit metrics for the same underlying data.

This tag doesn't prevent the duplication, but lets users identify and, if needed, exclude the autodiscovered side of it (e.g. `metric{!dd_config_discovery:true}`), both for the cross-host case above and for any single-host case the automatic suppression doesn't catch.

### Describe how you validated your changes

Unit and E2E tests.

---
🤖 This PR description and implementation were generated with assistance from [Claude Code](https://claude.com/claude-code).


[DSCVR-651]: https://datadoghq.atlassian.net/browse/DSCVR-651?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

Co-authored-by: vincent.whitchurch <vincent.whitchurch@datadoghq.com>

Author: vitkyrka

comp/core/autodiscovery/impl/BUILD.bazel

@@ -120,6 +120,7 @@ dd_agent_go_test(
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@com_github_stretchr_testify//suite",
+        "@in_yaml_go_yaml_v2//:yaml",
         "@org_uber_go_atomic//:atomic",
         "@org_uber_go_fx//:fx",
     ],

comp/core/autodiscovery/impl/common_test.go

@@ -21,6 +21,7 @@ type dummyService struct {
 	Ports           []workloadmeta.ContainerPort
 	Pid             int
 	Hostname        string
+	Tags            []string
 	filterTemplates func(map[string]integration.Config)
 }
 
@@ -51,7 +52,7 @@ func (s *dummyService) GetPorts() ([]workloadmeta.ContainerPort, error) {
 
 // GetTags returns the tags for this service
 func (s *dummyService) GetTags() ([]string, error) {
-	return nil, nil
+	return s.Tags, nil
 }
 
 // GetTagsWithCardinality returns the tags for this service

comp/core/autodiscovery/impl/configmgr_discovery.go

@@ -35,6 +35,23 @@ type discoveryState struct {
 // of completions without blocking the worker goroutine on a busy scheduler.
 const discoveredChangesBuffer = 128
 
+// configDiscoveryTag is added to every instance of a check scheduled via
+// configuration discovery. Configuration discovery has no way of knowing
+// about a manually-configured, differently-named check (e.g. a generic
+// `openmetrics` check) that a user has pointed at the same service from
+// elsewhere, so the two can end up scraping the same target and duplicating
+// (or, for additive metric types, doubling) submitted metrics. This tag lets
+// users spot and exclude the autodiscovered side of such a duplication.
+//
+// Uses a plain `dd_`-prefixed key (not `dd.internal.*`, which is reserved for
+// tags consumed and stripped internally before reaching the backend, e.g.
+// dd.internal.resource in pkg/metrics/series.go) so it survives to the
+// backend and stays queryable, following the precedent of other
+// agent-added, customer-visible marker tags such as dd_remote_config_id /
+// dd_remote_config_rev (comp/core/tagger/tags/tags.go) and
+// dd_enable_check_intake (pkg/collector/worker/worker.go).
+const configDiscoveryTag = "dd_config_discovery:true"
+
 // initDiscoveryWorker wires the workqueue-backed discovery worker into cm.
 func initDiscoveryWorker(cm *reconcilingConfigManager, disco discoverer.ConfigDiscoverer) {
 	cm.discoveredCh = make(chan integration.ConfigChanges, discoveredChangesBuffer)
@@ -152,6 +169,11 @@ func (cm *reconcilingConfigManager) applyDiscoveredConfigsLocked(svcID, tplDiges
 		return changes
 	}
 	resolved.Source = rewriteSource(resolved.Source, svcAndADIDs.svc)
+	for i := range resolved.Instances {
+		if err := resolved.Instances[i].MergeAdditionalTags([]string{configDiscoveryTag}); err != nil {
+			log.Errorf("error adding configuration-discovery tag to config %s for service %s: %v", resolved.Name, svcID, err)
+		}
+	}
 	decrypted, err := decryptConfig(resolved, cm.secretResolver, tplDigest)
 	if err != nil {
 		log.Errorf("error decrypting discovered config %s for service %s: %v", resolved.Name, svcID, err)

comp/core/autodiscovery/impl/configmgr_discovery_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
+	yaml "go.yaml.in/yaml/v2"
 
 	"github.com/DataDog/datadog-agent/comp/core/autodiscovery/discoverer"
 	"github.com/DataDog/datadog-agent/comp/core/autodiscovery/integration"
@@ -154,6 +155,8 @@ func TestConfigMgr_DiscoveryTemplate_RoutesThroughDiscoverer(t *testing.T) {
 					"discovered instance config should have %%host%% resolved via the configresolver path")
 				assert.Equal(t, tc.wantSource, discovered.Schedule[0].Source,
 					"discovered config's Source should be tagged with the discovery provider")
+				assert.Contains(t, string(discovered.Schedule[0].Instances[0]), configDiscoveryTag,
+					"discovered instance config should carry the configuration-discovery marker tag")
 			case <-time.After(2 * time.Second):
 				t.Fatalf("timed out waiting for discovered changes")
 			}
@@ -163,6 +166,145 @@ func TestConfigMgr_DiscoveryTemplate_RoutesThroughDiscoverer(t *testing.T) {
 	}
 }
 
+// countTag returns how many times tag appears in tags.
+func countTag(tags []string, tag string) int {
+	n := 0
+	for _, t := range tags {
+		if t == tag {
+			n++
+		}
+	}
+	return n
+}
+
+// instanceTags unmarshals the `tags` field of a resolved instance for
+// precise assertions (exact membership/count), rather than substring
+// matching on the raw YAML.
+func instanceTags(t *testing.T, instance integration.Data) []string {
+	t.Helper()
+	var parsed struct {
+		Tags []string `yaml:"tags"`
+	}
+	require.NoError(t, yaml.Unmarshal(instance, &parsed))
+	return parsed.Tags
+}
+
+// TestConfigMgr_DiscoveryTemplate_TagsAllInstances verifies that the
+// configuration-discovery marker tag is added exactly once to every instance of
+// a discovered config, is merged alongside any tags the discovered instance
+// already carries rather than replacing them, and is still added even when the
+// discovered config opts out of ordinary autodiscovery service tags via
+// ignore_autodiscovery_tags (which must still suppress the service's own tags,
+// proving the marker tag is applied independently of that mechanism).
+func TestConfigMgr_DiscoveryTemplate_TagsAllInstances(t *testing.T) {
+	mockResolver := MockSecretResolver{}
+	disco := newStubDiscoverer(func(_, _ string) (string, error) {
+		return `[{
+			"instances": [
+				{"openmetrics_endpoint": "http://%%host%%:8080/metrics", "tags": ["custom:tag"]},
+				{"openmetrics_endpoint": "http://%%host%%:8081/metrics"}
+			],
+			"ignore_autodiscovery_tags": true
+		}]`, nil
+	})
+	cm := newReconcilingConfigManager(&mockResolver, nil, nil, disco, nil).(*reconcilingConfigManager)
+	cm.start()
+	defer cm.stop()
+
+	tpl := integration.Config{
+		Name:          "krakend",
+		ADIdentifiers: []string{"krakend"},
+		Discovery:     &integration.DiscoveryConfig{},
+		Source:        "file:/etc/datadog-agent/conf.d/krakend.d/auto_conf.yaml",
+		Provider:      names.File,
+	}
+	svc := &dummyService{
+		ID:            "docker://k1",
+		ADIdentifiers: []string{"krakend"},
+		Hosts:         map[string]string{"main": "10.0.0.1"},
+		Tags:          []string{"service:tag"},
+	}
+
+	_, _ = cm.processNewConfig(tpl)
+	changes := cm.processNewService(svc)
+	assertConfigsMatch(t, changes.Schedule)
+	assertConfigsMatch(t, changes.Unschedule)
+
+	ch := cm.discoveredChanges()
+	require.NotNil(t, ch)
+	select {
+	case discovered := <-ch:
+		require.Len(t, discovered.Schedule, 1)
+		require.Len(t, discovered.Schedule[0].Instances, 2)
+
+		firstTags := instanceTags(t, discovered.Schedule[0].Instances[0])
+		assert.Equal(t, 1, countTag(firstTags, configDiscoveryTag),
+			"marker tag should appear exactly once on the first instance, got %v", firstTags)
+		assert.Contains(t, firstTags, "custom:tag",
+			"first instance's own tag should be preserved alongside the marker tag")
+		assert.NotContains(t, firstTags, "service:tag",
+			"ignore_autodiscovery_tags should still suppress ordinary service tags")
+
+		secondTags := instanceTags(t, discovered.Schedule[0].Instances[1])
+		assert.Equal(t, 1, countTag(secondTags, configDiscoveryTag),
+			"marker tag should appear exactly once on the second instance, got %v", secondTags)
+	case <-time.After(2 * time.Second):
+		t.Fatalf("timed out waiting for discovered changes")
+	}
+}
+
+// TestConfigMgr_DiscoveryTemplate_TagPersistsAcrossRediscovery verifies that
+// a second discovery result for the same service+template (e.g. the
+// integration's discover_config returning updated instance data on a later
+// probe) still carries exactly one configDiscoveryTag on the replacement
+// config, rather than accumulating duplicates or losing the tag across
+// re-resolution.
+func TestConfigMgr_DiscoveryTemplate_TagPersistsAcrossRediscovery(t *testing.T) {
+	mockResolver := MockSecretResolver{}
+	tpl := integration.Config{
+		Name:          "krakend",
+		ADIdentifiers: []string{"krakend"},
+		Discovery:     &integration.DiscoveryConfig{},
+		Source:        "file:/etc/datadog-agent/conf.d/krakend.d/auto_conf.yaml",
+		Provider:      names.File,
+	}
+	svc := &dummyService{
+		ID:            "docker://k1",
+		ADIdentifiers: []string{"krakend"},
+		Hosts:         map[string]string{"main": "10.0.0.1"},
+	}
+
+	// The discovery worker is never started: this test drives
+	// applyDiscoveredConfigsLocked directly (as the worker's callback would),
+	// so no discoverer or running goroutine is needed.
+	cm := newReconcilingConfigManager(&mockResolver, nil, nil, nil, nil).(*reconcilingConfigManager)
+	_, _ = cm.processNewConfig(tpl)
+	_ = cm.processNewService(svc)
+
+	tplDigest := tpl.Digest()
+	firstConfigs := []integration.Config{{
+		Instances: []integration.Data{integration.Data(`openmetrics_endpoint: http://%%host%%:8080/metrics`)},
+	}}
+	cm.m.Lock()
+	changes := cm.applyDiscoveredConfigsLocked(svc.ID, tplDigest, firstConfigs)
+	cm.m.Unlock()
+	require.Len(t, changes.Schedule, 1)
+	firstTags := instanceTags(t, changes.Schedule[0].Instances[0])
+	require.Equal(t, 1, countTag(firstTags, configDiscoveryTag))
+
+	secondConfigs := []integration.Config{{
+		Instances: []integration.Data{integration.Data(`openmetrics_endpoint: http://%%host%%:9090/metrics`)},
+	}}
+	cm.m.Lock()
+	changes = cm.applyDiscoveredConfigsLocked(svc.ID, tplDigest, secondConfigs)
+	cm.m.Unlock()
+	require.Len(t, changes.Schedule, 1, "rediscovery should schedule the replacement config")
+	require.Len(t, changes.Unschedule, 1, "rediscovery should unschedule the previous config")
+	secondTags := instanceTags(t, changes.Schedule[0].Instances[0])
+	assert.Equal(t, 1, countTag(secondTags, configDiscoveryTag),
+		"marker tag should appear exactly once on the re-discovered config, got %v", secondTags)
+}
+
 // TestConfigMgr_DiscoveryTemplate_ServiceDeletionCancels confirms that
 // deleting the service forgets in-flight probes so the worker stops retrying.
 func TestConfigMgr_DiscoveryTemplate_ServiceDeletionCancels(t *testing.T) {

releasenotes/notes/config-discovery-tag-a2e43a3bd95e2321.yaml

@@ -0,0 +1,14 @@
+# Each section from every release note are combined when the
+# CHANGELOG.rst is rendered. So the text needs to be worded so that
+# it does not depend on any information only available in another
+# section. This may mean repeating some details, but each section
+# must be readable independently of the other.
+#
+# Each section note must be formatted as reStructuredText.
+---
+enhancements:
+  - |
+    Check instances scheduled via configuration discovery now carry the
+    ``dd_config_discovery:true`` tag. This can be used to identify,
+    and if needed exclude, metrics submitted by an autodiscovered check that
+    duplicates a check configured manually elsewhere for the same service.

test/new-e2e/tests/discovery/BUILD.bazel

@@ -83,5 +83,13 @@ dd_agent_go_test(
         "@com_github_stretchr_testify//require",
         "@io_k8s_api//core/v1:core",
         "@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
-    ],
+    ] + select({
+        "@rules_go//go/platform:android": [
+            "//test/fakeintake/client",
+        ],
+        "@rules_go//go/platform:linux": [
+            "//test/fakeintake/client",
+        ],
+        "//conditions:default": [],
+    }),
 )

test/new-e2e/tests/discovery/config_discovery_linux_test.go

@@ -12,14 +12,17 @@ import (
 	"testing"
 	"time"
 
-	"github.com/DataDog/datadog-agent/test/e2e-framework/components/datadog/dockeragentparams"
-	scendocker "github.com/DataDog/datadog-agent/test/e2e-framework/scenarios/aws/ec2docker"
 	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/DataDog/datadog-agent/test/e2e-framework/components/datadog/dockeragentparams"
+	scendocker "github.com/DataDog/datadog-agent/test/e2e-framework/scenarios/aws/ec2docker"
+
 	"github.com/DataDog/datadog-agent/test/e2e-framework/testing/e2e"
 	"github.com/DataDog/datadog-agent/test/e2e-framework/testing/environments"
 	awsdocker "github.com/DataDog/datadog-agent/test/e2e-framework/testing/provisioners/aws/docker"
+	"github.com/DataDog/datadog-agent/test/fakeintake/aggregator"
+	fakeintakeclient "github.com/DataDog/datadog-agent/test/fakeintake/client"
 )
 
 //go:embed testdata/compose/docker-compose.fake-krakend.yaml
@@ -77,6 +80,10 @@ func (s *configDiscoverySuite) verifyKrakendConfigDiscovery(c *assert.CollectT)
 		t.Logf("configcheck output: %s", configCheckOutput)
 		return
 	}
+	if !assert.True(c, strings.Contains(configCheckOutput, configDiscoveryTag), "krakend config resolved via configuration discovery should carry the %s marker tag", configDiscoveryTag) {
+		t.Logf("configcheck output: %s", configCheckOutput)
+		return
+	}
 
 	statusOutput := s.Env().Docker.Client.ExecuteCommand(s.Env().Agent.ContainerName, "agent", "status", "collector", "--json")
 	var status collectorStatus
@@ -105,6 +112,26 @@ func (s *configDiscoverySuite) verifyKrakendConfigDiscovery(c *assert.CollectT)
 	// config came from the configuration-discovery path (via the Docker
 	// listener), not a plain file provider.
 	s.verifyKrakendCheckProvider(c)
+
+	// Verify the metric actually submitted by the discovered krakend check
+	// carries the configuration-discovery marker tag, not just the resolved
+	// config (checked above via configcheck).
+	s.verifyKrakendMetricHasConfigDiscoveryTag(c)
+}
+
+// verifyKrakendMetricHasConfigDiscoveryTag checks, via fakeintake, that a
+// metric submitted by the discovered krakend check (krakend.api.go.goroutines,
+// from the fake container's go_goroutines gauge) carries the
+// configDiscoveryTag marker tag end to end, not just in the resolved config.
+func (s *configDiscoverySuite) verifyKrakendMetricHasConfigDiscoveryTag(c *assert.CollectT) {
+	const metricName = "krakend.api.go.goroutines"
+
+	metrics, err := s.Env().FakeIntake.Client().FilterMetrics(metricName,
+		fakeintakeclient.WithTags[*aggregator.MetricSeries]([]string{configDiscoveryTag}))
+	if !assert.NoError(c, err, "failed to query fakeintake for %s", metricName) {
+		return
+	}
+	assert.NotEmpty(c, metrics, "expected at least one %s series tagged with %s", metricName, configDiscoveryTag)
 }
 
 // adContainerDiscoveryProvider mirrors names.ADContainerDiscovery in
@@ -115,6 +142,15 @@ func (s *configDiscoverySuite) verifyKrakendConfigDiscovery(c *assert.CollectT)
 // (e.g. containers, discovered via the Docker listener here).
 const adContainerDiscoveryProvider = "ad-container-discovery+file"
 
+// configDiscoveryTag mirrors configDiscoveryTag in
+// comp/core/autodiscovery/impl/configmgr_discovery.go (not importable here:
+// it lives in the root module, which test/new-e2e does not depend on). It is
+// the marker tag configuration discovery adds to every instance it
+// schedules, so users can identify (and, if needed, exclude) metrics
+// submitted by an autodiscovered check that duplicates a manually-configured
+// one pointed at the same service from elsewhere.
+const configDiscoveryTag = "dd_config_discovery:true"
+
 // verifyKrakendCheckProvider checks that the krakend check has
 // config.provider = adContainerDiscoveryProvider in the inventory-checks
 // metadata, confirming it was resolved via configuration discovery against