[apm] peer.service aggregation for trace stats, option to compute stats based on span.kind

cmd/trace-agent/config/config.go

@@ -193,6 +193,12 @@ func applyDatadogConfig(c *config.AgentConfig) error {
 	if coreconfig.Datadog.IsSet("apm_config.connection_limit") {
 		c.ConnectionLimit = coreconfig.Datadog.GetInt("apm_config.connection_limit")
 	}
+	if coreconfig.Datadog.IsSet("apm_config.peer_service_stats_aggregation") {
+		c.PeerServiceStatsAggregation = coreconfig.Datadog.GetBool("apm_config.peer_service_stats_aggregation")
+	}
+	if coreconfig.Datadog.IsSet("apm_config.compute_stats_by_span_kind") {
+		c.ComputeStatsBySpanKind = coreconfig.Datadog.GetBool("apm_config.compute_stats_by_span_kind")
+	}
 	if coreconfig.Datadog.IsSet("apm_config.extra_sample_rate") {
 		c.ExtraSampleRate = coreconfig.Datadog.GetFloat64("apm_config.extra_sample_rate")
 	}

cmd/trace-agent/config/config_test.go

@@ -1094,3 +1094,47 @@ func TestSetMaxMemCPU(t *testing.T) {
 		assert.Equal(t, 300.0, c.MaxMemory)
 	})
 }
+
+func TestPeerServiceAggregation(t *testing.T) {
+	t.Run("disabled", func(t *testing.T) {
+		defer cleanConfig()
+		cfg := config.New()
+		err := applyDatadogConfig(cfg)
+
+		assert := assert.New(t)
+		assert.NoError(err)
+		assert.False(cfg.PeerServiceStatsAggregation)
+	})
+	t.Run("enabled", func(t *testing.T) {
+		defer cleanConfig()
+		coreconfig.Datadog.Set("apm_config.peer_service_stats_aggregation", true)
+		cfg := config.New()
+		err := applyDatadogConfig(cfg)
+
+		assert := assert.New(t)
+		assert.NoError(err)
+		assert.True(cfg.PeerServiceStatsAggregation)
+	})
+}
+
+func TestComputeStatsBySpanKind(t *testing.T) {
+	t.Run("disabled", func(t *testing.T) {
+		defer cleanConfig()
+		cfg := config.New()
+		err := applyDatadogConfig(cfg)
+
+		assert := assert.New(t)
+		assert.NoError(err)
+		assert.False(cfg.ComputeStatsBySpanKind)
+	})
+	t.Run("enabled", func(t *testing.T) {
+		defer cleanConfig()
+		coreconfig.Datadog.Set("apm_config.compute_stats_by_span_kind", true)
+		cfg := config.New()
+		err := applyDatadogConfig(cfg)
+
+		assert := assert.New(t)
+		assert.NoError(err)
+		assert.True(cfg.ComputeStatsBySpanKind)
+	})
+}

pkg/config/apm.go

@@ -68,6 +68,8 @@ func setupAPM(config Config) {
 	config.BindEnvAndSetDefault("apm_config.windows_pipe_buffer_size", 1_000_000, "DD_APM_WINDOWS_PIPE_BUFFER_SIZE")                          //nolint:errcheck
 	config.BindEnvAndSetDefault("apm_config.windows_pipe_security_descriptor", "D:AI(A;;GA;;;WD)", "DD_APM_WINDOWS_PIPE_SECURITY_DESCRIPTOR") //nolint:errcheck
 	config.BindEnvAndSetDefault("apm_config.remote_tagger", true, "DD_APM_REMOTE_TAGGER")                                                     //nolint:errcheck
+	config.BindEnvAndSetDefault("apm_config.peer_service_stats_aggregation", false, "DD_APM_PEER_SERVICE_STATS_AGGREGATION")                  //nolint:errcheck
+	config.BindEnvAndSetDefault("apm_config.compute_stats_by_span_kind", false, "DD_APM_COMPUTE_STATS_BY_SPAN_KIND")                          //nolint:errcheck
 
 	config.BindEnv("apm_config.max_catalog_services", "DD_APM_MAX_CATALOG_SERVICES")
 	config.BindEnv("apm_config.receiver_timeout", "DD_APM_RECEIVER_TIMEOUT")

pkg/config/config_template.yaml

@@ -1222,10 +1222,21 @@ api_key:
   #
   # connection_limit: 2000
 
+  ## @param compute_stats_by_span_kind - bool - default: false
+  ## @env DD_APM_COMPUTE_STATS_BY_SPAN_KIND - bool - default: false
+  ## Enables an additional stats computation check on spans to see they have an eligible `span.kind` (server, consumer, client, producer).
+  ## If enabled, a span with an eligible `span.kind` will have stats computed. If disabled, only top-level and measured spans will have stats computed.
+  # compute_stats_by_span_kind: false
+
+  ## @param peer_service_stats_aggregation - bool - default: false
+  ## @env DD_APM_PEER_SERVICE_STATS_AGGREGATION - bool - default: false
+  ## Enables peer.service stats aggregation in the agent. If disabled, aggregated stats will not include peer.service as a dimension.
+  # peer_service_stats_aggregation: false
+
   ## @param features - list of strings - optional
   ## @env DD_APM_FEATURES - comma separated list of strings - optional
   ## Configure additional beta APM features.
-  ## The list of items available under apm_config.features is not guaranteed to persist across versions; 
+  ## The list of items available under apm_config.features is not guaranteed to persist across versions;
   ## a feature may eventually be promoted to its own configuration option on the agent, or dropped entirely.
   #
   # features: ["error_rare_sample_tracer_drop","table_names","component2name","sql_cache"]

pkg/config/config_test.go

@@ -1287,3 +1287,62 @@ fips:
 	err := setupFipsEndpoints(testConfig)
 	require.Error(t, err)
 }
+
+func TestEnablePeerServiceStatsAggregationYAML(t *testing.T) {
+	datadogYaml := `
+apm_config:
+  peer_service_stats_aggregation: true
+`
+	testConfig := setupConfFromYAML(datadogYaml)
+	err := setupFipsEndpoints(testConfig)
+	require.NoError(t, err)
+	require.True(t, testConfig.GetBool("apm_config.peer_service_stats_aggregation"))
+
+	datadogYaml = `
+apm_config:
+  peer_service_stats_aggregation: false
+`
+	testConfig = setupConfFromYAML(datadogYaml)
+	err = setupFipsEndpoints(testConfig)
+	require.NoError(t, err)
+	require.False(t, testConfig.GetBool("apm_config.peer_service_stats_aggregation"))
+}
+
+func TestEnablePeerServiceStatsAggregationEnv(t *testing.T) {
+	t.Setenv("DD_APM_PEER_SERVICE_STATS_AGGREGATION", "true")
+	testConfig := setupConfFromYAML("")
+	require.True(t, testConfig.GetBool("apm_config.peer_service_stats_aggregation"))
+	t.Setenv("DD_APM_PEER_SERVICE_STATS_AGGREGATION", "false")
+	testConfig = setupConfFromYAML("")
+	require.False(t, testConfig.GetBool("apm_config.peer_service_stats_aggregation"))
+}
+
+func TestEnableStatsComputationBySpanKindYAML(t *testing.T) {
+	datadogYaml := `
+apm_config:
+  compute_stats_by_span_kind: false
+`
+	testConfig := setupConfFromYAML(datadogYaml)
+	err := setupFipsEndpoints(testConfig)
+	require.NoError(t, err)
+	require.False(t, testConfig.GetBool("apm_config.compute_stats_by_span_kind"))
+
+	datadogYaml = `
+apm_config:
+  compute_stats_by_span_kind: true
+`
+	testConfig = setupConfFromYAML(datadogYaml)
+	err = setupFipsEndpoints(testConfig)
+	require.NoError(t, err)
+	require.True(t, testConfig.GetBool("apm_config.compute_stats_by_span_kind"))
+
+}
+
+func TestComputeStatsBySpanKindEnv(t *testing.T) {
+	t.Setenv("DD_APM_COMPUTE_STATS_BY_SPAN_KIND", "false")
+	testConfig := setupConfFromYAML("")
+	require.False(t, testConfig.GetBool("apm_config.compute_stats_by_span_kind"))
+	t.Setenv("DD_APM_COMPUTE_STATS_BY_SPAN_KIND", "true")
+	testConfig = setupConfFromYAML("")
+	require.True(t, testConfig.GetBool("apm_config.compute_stats_by_span_kind"))
+}

pkg/trace/agent/normalizer.go

@@ -28,6 +28,8 @@ const (
 	// tagSamplingPriority specifies the sampling priority of the trace.
 	// DEPRECATED: Priority is now specified as a TraceChunk field.
 	tagSamplingPriority = "_sampling_priority_v1"
+	// peerServiceKey is the key for the peer.service meta field.
+	peerServiceKey = "peer.service"
 )
 
 var (
@@ -60,6 +62,20 @@ func (a *Agent) normalize(ts *info.TagStats, s *pb.Span) error {
 	}
 	s.Service = svc
 
+	pSvc, ok := s.Meta[peerServiceKey]
+	if ok {
+		ps, err := traceutil.NormalizePeerService(pSvc)
+		switch err {
+		case traceutil.ErrTooLong:
+			ts.SpansMalformed.PeerServiceTruncate.Inc()
+			log.Debugf("Fixing malformed trace. peer.service is too long (reason:peer_service_truncate), truncating peer.service to length=%d: %s", traceutil.MaxServiceLen, ps)
+		case traceutil.ErrInvalid:
+			ts.SpansMalformed.PeerServiceInvalid.Inc()
+			log.Debugf("Fixing malformed trace. peer.service is invalid (reason:peer_service_invalid), replacing invalid peer.service=%s with empty string", pSvc)
+		}
+		s.Meta[peerServiceKey] = ps
+	}
+
 	if a.conf.HasFeature("component2name") {
 		// This feature flag determines the component tag to become the span name.
 		//

pkg/trace/api/info_test.go

@@ -243,17 +243,17 @@ func TestInfoHandler(t *testing.T) {
 			Host:    "https://target-intake.datadoghq.com",
 			NoProxy: true,
 		}},
-		BucketInterval:   time.Second,
-		ExtraAggregators: []string{"agg:val"},
-		ExtraSampleRate:  2.4,
-		TargetTPS:        11,
-		MaxEPS:           12,
-		ReceiverHost:     "localhost",
-		ReceiverPort:     8111,
-		ReceiverSocket:   "/sock/path",
-		ConnectionLimit:  12,
-		ReceiverTimeout:  100,
-		MaxRequestBytes:  123,
+		BucketInterval:              time.Second,
+		PeerServiceStatsAggregation: true,
+		ExtraSampleRate:             2.4,
+		TargetTPS:                   11,
+		MaxEPS:                      12,
+		ReceiverHost:                "localhost",
+		ReceiverPort:                8111,
+		ReceiverSocket:              "/sock/path",
+		ConnectionLimit:             12,
+		ReceiverTimeout:             100,
+		MaxRequestBytes:             123,
 		StatsWriter: &config.WriterConfig{
 			ConnectionLimit:    20,
 			QueueSize:          12,

pkg/trace/config/config.go

@@ -311,8 +311,9 @@ type AgentConfig struct {
 	Endpoints []*Endpoint
 
 	// Concentrator
-	BucketInterval   time.Duration // the size of our pre-aggregation per bucket
-	ExtraAggregators []string
+	BucketInterval              time.Duration // the size of our pre-aggregation per bucket
+	PeerServiceStatsAggregation bool          // enables/disables stats aggregation for peer.service
+	ComputeStatsBySpanKind      bool          // enables/disables the computing of stats based on a span's `span.kind` field
 
 	// Sampler configuration
 	ExtraSampleRate float64

pkg/trace/info/info_test.go

@@ -442,6 +442,8 @@ func TestPublishReceiverStats(t *testing.T) {
 				atom(10),
 				atom(11),
 				atom(12),
+				atom(13),
+				atom(14),
 			},
 			TracesFiltered:     atom(4),
 			TracesPriorityNone: atom(5),
@@ -487,14 +489,16 @@ func TestPublishReceiverStats(t *testing.T) {
 				"ServiceEmpty":          2.0,
 				"ServiceTruncate":       3.0,
 				"ServiceInvalid":        4.0,
-				"SpanNameEmpty":         5.0,
-				"SpanNameTruncate":      6.0,
-				"SpanNameInvalid":       7.0,
-				"ResourceEmpty":         8.0,
-				"TypeTruncate":          9.0,
-				"InvalidStartDate":      10.0,
-				"InvalidDuration":       11.0,
-				"InvalidHTTPStatusCode": 12.0,
+				"PeerServiceTruncate":   5.0,
+				"PeerServiceInvalid":    6.0,
+				"SpanNameEmpty":         7.0,
+				"SpanNameTruncate":      8.0,
+				"SpanNameInvalid":       9.0,
+				"ResourceEmpty":         10.0,
+				"TypeTruncate":          11.0,
+				"InvalidStartDate":      12.0,
+				"InvalidDuration":       13.0,
+				"InvalidHTTPStatusCode": 14.0,
 			},
 			"SpansReceived": 10.0,
 			"TracerVersion": "",

pkg/trace/info/stats.go

@@ -256,6 +256,10 @@ type SpansMalformed struct {
 	ServiceTruncate atomic.Int64
 	// ServiceInvalid is when a span's Service doesn't conform to Datadog tag naming standards
 	ServiceInvalid atomic.Int64
+	// PeerServiceTruncate is when a span's peer.service is truncated for exceeding the max length
+	PeerServiceTruncate atomic.Int64
+	// PeerServiceInvalid is when a span's peer.service doesn't conform to Datadog tag naming standards
+	PeerServiceInvalid atomic.Int64
 	// SpanNameEmpty is when a span's Name is empty
 	SpanNameEmpty atomic.Int64
 	// SpanNameTruncate is when a span's Name is truncated for exceeding the max length
@@ -280,6 +284,8 @@ func (s *SpansMalformed) tagCounters() map[string]*atomic.Int64 {
 		"service_empty":            &s.ServiceEmpty,
 		"service_truncate":         &s.ServiceTruncate,
 		"service_invalid":          &s.ServiceInvalid,
+		"peer_service_truncate":    &s.PeerServiceTruncate,
+		"peer_service_invalid":     &s.PeerServiceInvalid,
 		"span_name_empty":          &s.SpanNameEmpty,
 		"span_name_truncate":       &s.SpanNameTruncate,
 		"span_name_invalid":        &s.SpanNameInvalid,
@@ -420,6 +426,8 @@ func (s *Stats) update(recent *Stats) {
 	s.SpansMalformed.ServiceEmpty.Add(recent.SpansMalformed.ServiceEmpty.Load())
 	s.SpansMalformed.ServiceTruncate.Add(recent.SpansMalformed.ServiceTruncate.Load())
 	s.SpansMalformed.ServiceInvalid.Add(recent.SpansMalformed.ServiceInvalid.Load())
+	s.SpansMalformed.PeerServiceTruncate.Add(recent.SpansMalformed.PeerServiceTruncate.Load())
+	s.SpansMalformed.PeerServiceInvalid.Add(recent.SpansMalformed.PeerServiceInvalid.Load())
 	s.SpansMalformed.SpanNameEmpty.Add(recent.SpansMalformed.SpanNameEmpty.Load())
 	s.SpansMalformed.SpanNameTruncate.Add(recent.SpansMalformed.SpanNameTruncate.Load())
 	s.SpansMalformed.SpanNameInvalid.Add(recent.SpansMalformed.SpanNameInvalid.Load())

pkg/trace/info/stats_test.go

@@ -8,12 +8,13 @@ package info
 import (
 	"bytes"
 	"fmt"
-	"github.com/DataDog/datadog-agent/pkg/trace/log"
 	"reflect"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/DataDog/datadog-agent/pkg/trace/log"
+
 	"go.uber.org/atomic"
 
 	"github.com/DataDog/datadog-agent/pkg/trace/metrics"
@@ -60,6 +61,8 @@ func TestSpansMalformed(t *testing.T) {
 			"span_name_invalid":        0,
 			"span_name_empty":          0,
 			"service_truncate":         0,
+			"peer_service_truncate":    0,
+			"peer_service_invalid":     0,
 			"invalid_start_date":       0,
 			"invalid_http_status_code": 0,
 			"invalid_duration":         0,
@@ -241,12 +244,14 @@ func TestReceiverStats(t *testing.T) {
 		stats.SpansMalformed.ServiceInvalid.Store(4)
 		stats.SpansMalformed.SpanNameEmpty.Store(5)
 		stats.SpansMalformed.SpanNameTruncate.Store(6)
-		stats.SpansMalformed.SpanNameInvalid.Store(7)
-		stats.SpansMalformed.ResourceEmpty.Store(8)
-		stats.SpansMalformed.TypeTruncate.Store(9)
-		stats.SpansMalformed.InvalidStartDate.Store(10)
-		stats.SpansMalformed.InvalidDuration.Store(11)
-		stats.SpansMalformed.InvalidHTTPStatusCode.Store(12)
+		stats.SpansMalformed.PeerServiceTruncate.Store(7)
+		stats.SpansMalformed.PeerServiceInvalid.Store(8)
+		stats.SpansMalformed.SpanNameInvalid.Store(9)
+		stats.SpansMalformed.ResourceEmpty.Store(10)
+		stats.SpansMalformed.TypeTruncate.Store(11)
+		stats.SpansMalformed.InvalidStartDate.Store(12)
+		stats.SpansMalformed.InvalidDuration.Store(13)
+		stats.SpansMalformed.InvalidHTTPStatusCode.Store(14)
 		return &ReceiverStats{
 			Stats: map[Tags]*TagStats{
 				tags: {