[SVLS-9121] Identify Azure Container Apps OTel Payloads and emit Datadog exporter running metric

comp/otelcol/otlp/components/exporter/datadogexporter/traces_exporter.go

@@ -88,9 +88,9 @@ func (exp *traceExporter) consumeTraces(
 		}
 		switch src.Kind {
 		case source.HostnameKind:
-			hosts[src.Identifier] = struct{}{}
+			hosts[src.Identifier] = struct{}{} //nolint:staticcheck // SA1019: intentional during Step 1 of the Source.Identifier migration (datadog-agent#51116); this call site migrates to SourceIdentifier.Primary in Step 2
 		case source.AWSECSFargateKind:
-			ecsFargateArns[src.Identifier] = struct{}{}
+			ecsFargateArns[src.Identifier] = struct{}{} //nolint:staticcheck // SA1019: intentional during Step 1 of the Source.Identifier migration (datadog-agent#51116); this call site migrates to SourceIdentifier.Primary in Step 2
 		case source.InvalidKind:
 		}
 	}

comp/otelcol/otlp/components/exporter/serializerexporter/consumer.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"slices"
 	"strings"
 	"time"
 
@@ -23,6 +24,7 @@ import (
 
 	"github.com/DataDog/datadog-agent/comp/core/telemetry/def"
 	"github.com/DataDog/datadog-agent/pkg/metrics"
+	"github.com/DataDog/datadog-agent/pkg/opentelemetry-mapping-go/otlp/attributes/source"
 	otlpmetrics "github.com/DataDog/datadog-agent/pkg/opentelemetry-mapping-go/otlp/metrics"
 	pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/trace"
 	"github.com/DataDog/datadog-agent/pkg/serializer"
@@ -98,7 +100,7 @@ type serializerConsumer struct {
 	apmReceiverAddr string
 	ipath           ingestionPath
 	hosts           map[string]struct{}
-	ecsFargateTags  map[string]struct{}
+	fargateTagSets  map[tagSetKey][]string
 }
 
 // ingestionPath specifies which ingestion path is using the serializer exporter
@@ -209,8 +211,13 @@ func (c *serializerConsumer) addTelemetryMetric(agentHostname string, params exp
 		for host := range c.hosts {
 			coatUsageMetric.Set(1.0, buildInfo.Version, buildInfo.Command, host, "")
 		}
-		for ecsFargateTag := range c.ecsFargateTags {
-			taskArn := strings.Split(ecsFargateTag, ":")[1]
+		for _, tags := range c.fargateTagSets {
+			prefix := string(source.AWSECSFargateKind) + ":"
+			idx := slices.IndexFunc(tags, func(t string) bool { return strings.HasPrefix(t, prefix) })
+			if idx == -1 {
+				continue
+			}
+			taskArn := strings.TrimPrefix(tags[idx], prefix)
 			coatUsageMetric.Set(1.0, buildInfo.Version, buildInfo.Command, "", taskArn)
 		}
 	case agentOTLPIngest:
@@ -324,7 +331,13 @@ func (c *serializerConsumer) ConsumeHost(host string) {
 	c.hosts[host] = struct{}{}
 }
 
-// ConsumeTag implements the metrics.TagsConsumer interface.
-func (c *serializerConsumer) ConsumeTag(tag string) {
-	c.ecsFargateTags[tag] = struct{}{}
+// ConsumeTagSet implements the metrics.TagSetConsumer interface.
+func (c *serializerConsumer) ConsumeTagSet(metricSuffix string, tags []string) {
+	if metricSuffix != "fargate" {
+		return
+	}
+	sorted := slices.Clone(tags)
+	slices.Sort(sorted)
+	dedupKey := tagSetKey{metricSuffix: metricSuffix, sortedTags: strings.Join(sorted, ",")}
+	c.fargateTagSets[dedupKey] = sorted
 }

comp/otelcol/otlp/components/exporter/serializerexporter/consumer_collector.go

@@ -7,23 +7,31 @@ package serializerexporter
 
 import (
 	"fmt"
+	"slices"
 	"strings"
 
-	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/exporter"
-
 	telemetry "github.com/DataDog/datadog-agent/comp/core/telemetry/def"
 	"github.com/DataDog/datadog-agent/pkg/metrics"
-	"github.com/DataDog/datadog-agent/pkg/opentelemetry-mapping-go/otlp/attributes/source"
 	"github.com/DataDog/datadog-agent/pkg/tagset"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter"
 )
 
+// tagSetKey namespaces a ConsumeTagSet dedup key by metricSuffix, so that two
+// different workload types can never collide in seenTagSets even if their
+// tag content happens to coincide. sortedTags is derived from tags themselves
+// (sorted and joined) so that two calls with identical tags always dedup
+type tagSetKey struct {
+	metricSuffix string
+	sortedTags   string
+}
+
 // collectorConsumer is a consumer OSS collector uses to send metrics to the DataDog.
 type collectorConsumer struct {
 	*serializerConsumer
-	seenHosts map[string]struct{}
-	seenTags  map[string]struct{}
-	buildInfo component.BuildInfo
+	seenHosts   map[string]struct{}
+	seenTagSets map[tagSetKey][]string // value: full "key:value" tag slice for the metric
+	buildInfo   component.BuildInfo
 	// getPushTime returns a Unix time in nanoseconds, representing the time pushing metrics.
 	// It will be overwritten in tests.
 	getPushTime func() uint64
@@ -41,17 +49,13 @@ func (c *collectorConsumer) addRuntimeTelemetryMetric(_ string, languageTags []s
 		series = append(series, runningMetric)
 	}
 
-	var nonFargateTags []string
-	for tag := range c.seenTags {
-		if strings.HasPrefix(tag, string(source.AWSECSFargateKind)+":") {
-			series = append(series, exporterFargateMetrics(timestamp, append(buildTags, tag)))
-		} else {
-			nonFargateTags = append(nonFargateTags, tag)
-		}
+	for k, tags := range c.seenTagSets {
+		allTags := append(slices.Clone(buildTags), tags...)
+		series = append(series, exporterWorkloadMetrics(k.metricSuffix, timestamp, allTags))
 	}
-	if (len(c.seenHosts) > 0 && len(c.seenTags) == 0) || len(nonFargateTags) > 0 {
-		tags := append(buildTags, nonFargateTags...)
-		series = append(series, exporterDefaultMetrics("metrics", "", timestamp, tags))
+
+	if len(c.seenHosts) > 0 && len(c.seenTagSets) == 0 {
+		series = append(series, exporterDefaultMetrics("metrics", "", timestamp, buildTags))
 	}
 
 	for _, lang := range languageTags {
@@ -70,9 +74,12 @@ func (c *collectorConsumer) ConsumeHost(host string) {
 	c.seenHosts[host] = struct{}{}
 }
 
-// ConsumeTag implements the metrics.TagsConsumer interface.
-func (c *collectorConsumer) ConsumeTag(tag string) {
-	c.seenTags[tag] = struct{}{}
+// ConsumeTagSet implements the metrics.TagSetConsumer interface.
+func (c *collectorConsumer) ConsumeTagSet(metricSuffix string, tags []string) {
+	sorted := slices.Clone(tags)
+	slices.Sort(sorted)
+	dedupKey := tagSetKey{metricSuffix: metricSuffix, sortedTags: strings.Join(sorted, ",")}
+	c.seenTagSets[dedupKey] = sorted
 }
 
 // exporterDefaultMetrics creates built-in metrics to report that an exporter is running
@@ -93,10 +100,13 @@ func exporterDefaultMetrics(exporterType string, hostname string, timestamp uint
 	return metrics
 }
 
-// exporterFargateMetrics creates a built-in metric to report that a Fargate exporter is running.
-func exporterFargateMetrics(timestamp uint64, tags []string) *metrics.Serie {
+// exporterWorkloadMetrics creates a built-in metric to report that a
+// workload-specific exporter (e.g. Fargate, Azure Container Apps) is
+// running. The resulting metric name is
+// "otel.datadog_exporter.metrics.running.<metricSuffix>".
+func exporterWorkloadMetrics(metricSuffix string, timestamp uint64, tags []string) *metrics.Serie {
 	return &metrics.Serie{
-		Name: "otel.datadog_exporter.metrics.running.fargate",
+		Name: "otel.datadog_exporter.metrics.running." + metricSuffix,
 		Points: []metrics.Point{
 			{
 				Ts:    float64(timestamp),

comp/otelcol/otlp/components/exporter/serializerexporter/consumer_collector_test.go

@@ -9,6 +9,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
 )
 
@@ -17,21 +18,44 @@ func newTestCollectorConsumer(buildInfo component.BuildInfo) *collectorConsumer
 	return &collectorConsumer{
 		serializerConsumer: s,
 		seenHosts:          make(map[string]struct{}),
-		seenTags:           make(map[string]struct{}),
+		seenTagSets:        make(map[tagSetKey][]string),
 		buildInfo:          buildInfo,
 		getPushTime:        func() uint64 { return uint64(2e9) },
 	}
 }
 
-func TestExporterFargateMetrics(t *testing.T) {
-	tags := []string{"version:1.0", "command:otelcontribcol"}
-	serie := exporterFargateMetrics(uint64(2e9), tags)
+func TestExporterWorkloadMetrics(t *testing.T) {
+	tests := []struct {
+		name         string
+		metricSuffix string
+		tags         []string
+		wantName     string
+	}{
+		{
+			name:         "fargate",
+			metricSuffix: "fargate",
+			tags:         []string{"version:1.0", "command:otelcontribcol", "task_arn:arn:aws:ecs:us-east-1:123:task/cluster/abc"},
+			wantName:     "otel.datadog_exporter.metrics.running.fargate",
+		},
+		{
+			name:         "azurecontainerapps",
+			metricSuffix: "azurecontainerapps",
+			tags:         []string{"version:1.0", "command:otelcontribcol", "replica_name:replica-1"},
+			wantName:     "otel.datadog_exporter.metrics.running.azurecontainerapps",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			serie := exporterWorkloadMetrics(tt.metricSuffix, uint64(2e9), tt.tags)
 
-	assert.Equal(t, "otel.datadog_exporter.metrics.running.fargate", serie.Name)
-	assert.Equal(t, 1, len(serie.Points))
-	assert.Equal(t, float64(2e9), serie.Points[0].Ts)
-	assert.Equal(t, 1.0, serie.Points[0].Value)
-	assert.Equal(t, "", serie.Host)
+			assert.Equal(t, tt.wantName, serie.Name)
+			assert.Equal(t, 1, len(serie.Points))
+			assert.Equal(t, float64(2e9), serie.Points[0].Ts)
+			assert.Equal(t, 1.0, serie.Points[0].Value)
+			assert.Equal(t, "", serie.Host)
+		})
+	}
 }
 
 func TestAddRuntimeTelemetryMetric_NoTags(t *testing.T) {
@@ -64,7 +88,8 @@ func TestAddRuntimeTelemetryMetric_HostSource(t *testing.T) {
 func TestAddRuntimeTelemetryMetric_FargateTags(t *testing.T) {
 	buildInfo := component.BuildInfo{Version: "1.0", Command: "otelcontribcol"}
 	c := newTestCollectorConsumer(buildInfo)
-	c.ConsumeTag("task_arn:arn:aws:ecs:us-east-1:123:task/cluster/abc")
+	tag := "task_arn:arn:aws:ecs:us-east-1:123:task/cluster/abc"
+	c.ConsumeTagSet("fargate", []string{tag})
 
 	c.addRuntimeTelemetryMetric("", nil)
 
@@ -79,7 +104,8 @@ func TestAddRuntimeTelemetryMetric_HostAndFargate(t *testing.T) {
 	buildInfo := component.BuildInfo{Version: "1.0", Command: "otelcontribcol"}
 	c := newTestCollectorConsumer(buildInfo)
 	c.ConsumeHost("my-hostname")
-	c.ConsumeTag("task_arn:arn:aws:ecs:us-east-1:123:task/cluster/abc")
+	tag := "task_arn:arn:aws:ecs:us-east-1:123:task/cluster/abc"
+	c.ConsumeTagSet("fargate", []string{tag})
 
 	c.addRuntimeTelemetryMetric("", nil)
 
@@ -91,3 +117,33 @@ func TestAddRuntimeTelemetryMetric_HostAndFargate(t *testing.T) {
 	assert.ElementsMatch(t, metricsByName["otel.datadog_exporter.metrics.running.fargate"], []string{""})
 	assert.Len(t, c.series, 2)
 }
+
+func TestAzureContainerAppsMetric(t *testing.T) {
+	buildInfo := component.BuildInfo{}
+	c := newTestCollectorConsumer(buildInfo)
+	tags := []string{
+		"replica_name:replica-1",
+		"name:my-app",
+		"subscription_id:sub-123",
+		"resource_group:my-rg",
+	}
+	reordered := []string{
+		"resource_group:my-rg",
+		"subscription_id:sub-123",
+		"name:my-app",
+		"replica_name:replica-1",
+	}
+	c.ConsumeTagSet("azurecontainerapps", tags)
+	// Same tags, different order should dedup
+	c.ConsumeTagSet("azurecontainerapps", reordered)
+	c.addRuntimeTelemetryMetric("", nil)
+
+	require.Len(t, c.series, 1, "expected exactly one series (ACA only)")
+	found := c.series[0]
+	assert.Equal(t, "otel.datadog_exporter.metrics.running.azurecontainerapps", found.Name)
+	tagStrs := found.Tags.UnsafeToReadOnlySliceString()
+	assert.Contains(t, tagStrs, "replica_name:replica-1")
+	assert.Contains(t, tagStrs, "name:my-app")
+	assert.Contains(t, tagStrs, "subscription_id:sub-123")
+	assert.Contains(t, tagStrs, "resource_group:my-rg")
+}

comp/otelcol/otlp/components/exporter/serializerexporter/exporter.go

@@ -151,7 +151,11 @@ func (f SourceProviderFunc) Source(ctx context.Context) (source.Source, error) {
 		return source.Source{}, err
 	}
 
-	return source.Source{Kind: source.HostnameKind, Identifier: hostnameIdentifier}, nil
+	return source.Source{
+		Kind:             source.HostnameKind,
+		Identifier:       hostnameIdentifier,
+		SourceIdentifier: source.SourceIdentifier{Primary: hostnameIdentifier},
+	}, nil
 }
 
 // Exporter translate OTLP metrics into the Datadog format and sends

comp/otelcol/otlp/components/exporter/serializerexporter/exporter_test.go

@@ -501,7 +501,7 @@ func TestRunningMetricForPayloadContents(t *testing.T) {
 				return &collectorConsumer{
 					serializerConsumer: &serializerConsumer{},
 					seenHosts:          make(map[string]struct{}),
-					seenTags:           make(map[string]struct{}),
+					seenTagSets:        make(map[tagSetKey][]string),
 					getPushTime:        func() uint64 { return 0 },
 				}
 			}
@@ -1185,7 +1185,7 @@ func initSyncSerializerForTest(t testing.TB, logger *zap.Logger, cfg *ExporterCo
 			if err != nil {
 				return ""
 			}
-			return s.Identifier
+			return s.Identifier //nolint:staticcheck // SA1019: intentional during Step 1 of the Source.Identifier migration (datadog-agent#51116); this call site migrates to SourceIdentifier.Primary in Step 2
 		}),
 		fx.Provide(newOrchestratorinterfaceimpl),
 		fx.Provide(serializer.NewSerializer),

comp/otelcol/otlp/components/exporter/serializerexporter/factory.go

@@ -121,7 +121,7 @@ func newFactoryForAgentWithType(
 				apmReceiverAddr: apmReceiverAddr,
 				ipath:           ipath,
 				hosts:           make(map[string]struct{}),
-				ecsFargateTags:  make(map[string]struct{}),
+				fargateTagSets:  make(map[tagSetKey][]string),
 			}
 		},
 		options:      options,
@@ -172,7 +172,7 @@ func NewFactoryForOSSExporter(typ component.Type, statsIn chan []byte) exp.Facto
 			return &collectorConsumer{
 				serializerConsumer: s,
 				seenHosts:          make(map[string]struct{}),
-				seenTags:           make(map[string]struct{}),
+				seenTagSets:        make(map[tagSetKey][]string),
 				buildInfo:          buildInfo,
 				getPushTime:        func() uint64 { return uint64(time.Now().Unix()) },
 			}

comp/otelcol/otlp/components/exporter/serializerexporter/serializer.go

@@ -188,7 +188,7 @@ func initSerializerInternal(logger *zap.Logger, cfg *ExporterConfig, sourceProvi
 			if err != nil {
 				return ""
 			}
-			return s.Identifier
+			return s.Identifier //nolint:staticcheck // SA1019: intentional during Step 1 of the Source.Identifier migration (datadog-agent#51116); this call site migrates to SourceIdentifier.Primary in Step 2
 		}),
 		fx.Provide(newOrchestratorinterfaceimpl),
 		fx.Provide(serializer.NewSerializer),

pkg/opentelemetry-mapping-go/inframetadata/reporter.go

@@ -121,7 +121,7 @@ func (r *Reporter) hostname(res pcommon.Resource) (string, bool) {
 		// The resource does not identify a host (e.g. serverless resource)
 		return "", false
 	}
-	return src.Identifier, true
+	return src.Identifier, true //nolint:staticcheck // SA1019: intentional during Step 1 of the Source.Identifier migration (datadog-agent#51116); this call site migrates to SourceIdentifier.Primary in Step 2
 }
 
 // ConsumeResource for host metadata reporting purposes.

pkg/opentelemetry-mapping-go/otlp/attributes/BUILD.bazel

@@ -28,6 +28,7 @@ go_library(
         "@io_opentelemetry_go_otel//semconv/v1.12.0:v1_12_0",
         "@io_opentelemetry_go_otel//semconv/v1.17.0:v1_17_0",
         "@io_opentelemetry_go_otel//semconv/v1.27.0:v1_27_0",
+        "@io_opentelemetry_go_otel//semconv/v1.40.0:v1_40_0",
         "@io_opentelemetry_go_otel//semconv/v1.6.1:v1_6_1",
         "@io_opentelemetry_go_otel_metric//:metric",
     ],
@@ -60,6 +61,7 @@ go_test(
         "@io_opentelemetry_go_otel//semconv/v1.12.0:v1_12_0",
         "@io_opentelemetry_go_otel//semconv/v1.17.0:v1_17_0",
         "@io_opentelemetry_go_otel//semconv/v1.27.0:v1_27_0",
+        "@io_opentelemetry_go_otel//semconv/v1.40.0:v1_40_0",
         "@io_opentelemetry_go_otel//semconv/v1.6.1:v1_6_1",
         "@io_opentelemetry_go_otel_sdk_metric//:metric",
         "@io_opentelemetry_go_otel_sdk_metric//metricdata",