[connector/datadog] Improve performance in cases with many peer tags

.chloggen/dd-conn-peer-tags.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: datadogconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Optimize Datadog connector when there are many peer tags and `connector.datadogconnector.NativeIngest` is enabled"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: []
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: "`connector.datadogconnector.NativeIngest` is currently enabled by default"
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

connector/datadogconnector/benchmark_test.go

@@ -0,0 +1,95 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package datadogconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/datadogconnector"
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/connector/connectortest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/featuregate"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+func genTrace() ptrace.Traces {
+	start := time.Now().Add(-1 * time.Second)
+	end := time.Now()
+	traces := ptrace.NewTraces()
+	rspan := traces.ResourceSpans().AppendEmpty()
+	rattrs := rspan.Resource().Attributes()
+	rattrs.PutStr("deployment.environment", "test_env")
+	rattrs.PutStr("service.name", "test_svc")
+	sspan := rspan.ScopeSpans().AppendEmpty()
+	span := sspan.Spans().AppendEmpty()
+	span.SetTraceID(testTraceID)
+	span.SetSpanID(testSpanID1)
+	span.SetStartTimestamp(pcommon.NewTimestampFromTime(start))
+	span.SetEndTimestamp(pcommon.NewTimestampFromTime(end))
+	span.SetName("span_name")
+	span.SetKind(ptrace.SpanKindClient)
+	span.Attributes().PutStr("peer.service", "my_peer_svc")
+	span.Attributes().PutStr("rpc.service", "my_rpc_svc")
+	span.Attributes().PutStr("net.peer.name", "my_net_peer")
+	return traces
+}
+
+func BenchmarkPeerTags_Native(b *testing.B) {
+	benchmarkPeerTags(b)
+}
+
+func BenchmarkPeerTags_Legacy(b *testing.B) {
+	err := featuregate.GlobalRegistry().Set(NativeIngestFeatureGate.ID(), false)
+	assert.NoError(b, err)
+	defer func() {
+		_ = featuregate.GlobalRegistry().Set(NativeIngestFeatureGate.ID(), true)
+	}()
+
+	benchmarkPeerTags(b)
+}
+
+func benchmarkPeerTags(b *testing.B) {
+	cfg := NewFactory().CreateDefaultConfig().(*Config)
+	cfg.Traces.ComputeStatsBySpanKind = true
+	cfg.Traces.PeerTagsAggregation = true
+	cfg.Traces.BucketInterval = 10 * time.Millisecond
+	cfg.Traces.TraceBuffer = 0
+
+	factory := NewFactory()
+	creationParams := connectortest.NewNopSettings()
+	metricsSink := &consumertest.MetricsSink{}
+
+	tconn, err := factory.CreateTracesToMetrics(context.Background(), creationParams, cfg, metricsSink)
+	assert.NoError(b, err)
+
+	err = tconn.Start(context.Background(), componenttest.NewNopHost())
+	if err != nil {
+		b.Errorf("Error starting connector: %v", err)
+		return
+	}
+	defer func() {
+		require.NoError(b, tconn.Shutdown(context.Background()))
+	}()
+
+	b.ResetTimer()
+
+	for n := 0; n < b.N; n++ {
+		err = tconn.ConsumeTraces(context.Background(), genTrace())
+		assert.NoError(b, err)
+		for {
+			metrics := metricsSink.AllMetrics()
+			if len(metrics) > 0 {
+				assert.Equal(b, 1, len(metrics))
+				break
+			}
+			time.Sleep(5 * time.Millisecond)
+		}
+		metricsSink.Reset()
+	}
+}

connector/datadogconnector/connector_native.go

@@ -37,6 +37,9 @@ type traceToMetricConnectorNative struct {
 	// ctagKeys are container tag keys
 	ctagKeys []string
 
+	// peerTagKeys are peer tag keys to group APM stats
+	peerTagKeys []string
+
 	// translator specifies the translator used to transform APM Stats Payloads
 	// from the agent to OTLP Metrics.
 	translator *metrics.Translator
@@ -75,6 +78,7 @@ func newTraceToMetricConnectorNative(set component.TelemetrySettings, cfg compon
 		translator:      trans,
 		tcfg:            tcfg,
 		ctagKeys:        cfg.(*Config).Traces.ResourceAttributesAsContainerTags,
+		peerTagKeys:     tcfg.ConfiguredPeerTags(),
 		concentrator:    stats.NewConcentrator(tcfg, statsWriter, time.Now(), metricsClient),
 		statsout:        statsout,
 		metricsConsumer: metricsConsumer,
@@ -114,7 +118,7 @@ func (c *traceToMetricConnectorNative) Capabilities() consumer.Capabilities {
 }
 
 func (c *traceToMetricConnectorNative) ConsumeTraces(_ context.Context, traces ptrace.Traces) error {
-	inputs := stats.OTLPTracesToConcentratorInputs(traces, c.tcfg, c.ctagKeys)
+	inputs := stats.OTLPTracesToConcentratorInputs(traces, c.tcfg, c.ctagKeys, c.peerTagKeys)
 	for _, input := range inputs {
 		c.concentrator.Add(input)
 	}

connector/datadogconnector/go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/DataDog/datadog-agent/comp/otelcol/otlp/components/metricsclient v0.56.0
 	github.com/DataDog/datadog-agent/comp/otelcol/otlp/components/statsprocessor v0.56.0
 	github.com/DataDog/datadog-agent/pkg/proto v0.56.0
-	github.com/DataDog/datadog-agent/pkg/trace v0.57.0-devel.0.20240722160158-ad956a31a730
+	github.com/DataDog/datadog-agent/pkg/trace v0.58.0-devel.0.20240830155027-f44a9da12951
 	github.com/DataDog/datadog-go/v5 v5.5.0
 	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/attributes v0.19.0
 	github.com/DataDog/opentelemetry-mapping-go/pkg/otlp/metrics v0.19.0
@@ -70,7 +70,7 @@ require (
 	github.com/DataDog/datadog-agent/pkg/logs/sources v0.56.0 // indirect
 	github.com/DataDog/datadog-agent/pkg/logs/status/statusinterface v0.56.0 // indirect
 	github.com/DataDog/datadog-agent/pkg/logs/status/utils v0.56.0 // indirect
-	github.com/DataDog/datadog-agent/pkg/obfuscate v0.56.0 // indirect
+	github.com/DataDog/datadog-agent/pkg/obfuscate v0.58.0-devel.0.20240830155027-f44a9da12951 // indirect
 	github.com/DataDog/datadog-agent/pkg/remoteconfig/state v0.56.0 // indirect
 	github.com/DataDog/datadog-agent/pkg/status/health v0.56.0 // indirect
 	github.com/DataDog/datadog-agent/pkg/telemetry v0.56.0 // indirect
@@ -93,7 +93,7 @@ require (
 	github.com/DataDog/datadog-agent/pkg/version v0.56.0 // indirect
 	github.com/DataDog/datadog-api-client-go/v2 v2.29.0 // indirect
 	github.com/DataDog/dd-sensitive-data-scanner/sds-go/go v0.0.0-20240419161837-f1b2f553edfe // indirect
-	github.com/DataDog/go-sqllexer v0.0.12 // indirect
+	github.com/DataDog/go-sqllexer v0.0.13 // indirect
 	github.com/DataDog/go-tuf v1.1.0-0.5.2 // indirect
 	github.com/DataDog/gohai v0.0.0-20230524154621-4316413895ee // indirect
 	github.com/DataDog/opentelemetry-mapping-go/pkg/inframetadata v0.19.0 // indirect
@@ -273,7 +273,7 @@ require (
 	golang.org/x/sys v0.24.0 // indirect
 	golang.org/x/term v0.23.0 // indirect
 	golang.org/x/text v0.17.0 // indirect
-	golang.org/x/time v0.5.0 // indirect
+	golang.org/x/time v0.6.0 // indirect
 	gonum.org/v1/gonum v0.15.1 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240822170219-fc7c04adadcd // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240822170219-fc7c04adadcd // indirect