[dsd] Support container ID field from dogstatsd 1.2

pkg/aggregator/aggregator.go

@@ -437,7 +437,7 @@ func (agg *BufferedAggregator) addServiceCheck(sc metrics.ServiceCheck) {
 		sc.Ts = time.Now().Unix()
 	}
 	tb := tagset.NewHashlessTagsAccumulatorFromSlice(sc.Tags)
-	tagger.EnrichTags(tb, sc.OriginID, sc.K8sOriginID, sc.Cardinality)
+	tagger.EnrichTags(tb, sc.OriginFromUDS, sc.OriginFromClient, sc.Cardinality)
 
 	tb.SortUniq()
 	sc.Tags = tb.Get()
@@ -451,7 +451,7 @@ func (agg *BufferedAggregator) addEvent(e metrics.Event) {
 		e.Ts = time.Now().Unix()
 	}
 	tb := tagset.NewHashlessTagsAccumulatorFromSlice(e.Tags)
-	tagger.EnrichTags(tb, e.OriginID, e.K8sOriginID, e.Cardinality)
+	tagger.EnrichTags(tb, e.OriginFromUDS, e.OriginFromClient, e.Cardinality)
 
 	tb.SortUniq()
 	e.Tags = tb.Get()

pkg/config/config.go

@@ -498,6 +498,7 @@ func InitConfig(config Config) {
 	// contexts will be deleted (see 'dogstatsd_expiry_seconds').
 	config.BindEnvAndSetDefault("dogstatsd_context_expiry_seconds", 300)
 	config.BindEnvAndSetDefault("dogstatsd_origin_detection", false) // Only supported for socket traffic
+	config.BindEnvAndSetDefault("dogstatsd_origin_detection_client", false)
 	config.BindEnvAndSetDefault("dogstatsd_so_rcvbuf", 0)
 	config.BindEnvAndSetDefault("dogstatsd_metrics_stats_enable", false)
 	config.BindEnvAndSetDefault("dogstatsd_tags", []string{})

pkg/config/config_template.yaml

@@ -1601,6 +1601,13 @@ api_key:
 #
 # dogstatsd_origin_detection: false
 
+## @param dogstatsd_origin_detection_client - boolean - optional - default: false
+## @env DD_DOGSTATSD_ORIGIN_DETECTION_CLIENT - boolean - optional - default: false
+## Whether the Agent should use a client-provided container ID to enrich the metrics, events and service checks with container tags.
+## Note: This requires using a client compatible with DogStatsD protocol version 1.2.
+#
+# dogstatsd_origin_detection_client: false
+
 ## @param dogstatsd_buffer_size - integer - optional - default: 8192
 ## @env DD_DOGSTATSD_BUFFER_SIZE - integer - optional - default: 8192
 ## The buffer size use to receive statsd packets, in bytes.

pkg/dogstatsd/enrich.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 
 	"github.com/DataDog/datadog-agent/pkg/metrics"
+	"github.com/DataDog/datadog-agent/pkg/util/containers"
 	"github.com/DataDog/datadog-agent/pkg/util/kubernetes/kubelet"
 )
 
@@ -20,16 +21,41 @@ var (
 	CardinalityTagPrefix = "dd.internal.card:"
 )
 
-func extractTagsMetadata(tags []string, defaultHostname string, originTags string, entityIDPrecedenceEnabled bool) ([]string, string, string, string, string) {
+// extractTagsMetadata returns tags (client tags + host tag) and information needed to query tagger (origins, cardinality).
+//
+// The following tables explain how the origins are chosen.
+// originFromUDS is the origin discovered via UDS origin detection (container ID).
+// originFromTag is the origin sent by the client via the dd.internal.entity_id tag (non-prefixed pod uid).
+// originFromMsg is the origin sent by the client via the container field (non-prefixed container ID).
+// entityIDPrecedenceEnabled refers to the dogstatsd_entity_id_precedence parameter.
+//
+//  ---------------------------------------------------------------------------------
+// | originFromUDS | originFromTag | entityIDPrecedenceEnabled || Result: udsOrigin  |
+// |---------------|---------------|---------------------------||--------------------|
+// | any           | any           | false                     || originFromUDS      |
+// | any           | any           | true                      || empty              |
+// | any           | empty         | any                       || originFromUDS      |
+//  ---------------------------------------------------------------------------------
+//
+//  ---------------------------------------------------------------------------------
+// | originFromTag          | originFromMsg   || Result: originFromClient            |
+// |------------------------|-----------------||-------------------------------------|
+// | not empty && not none  | any             || pod prefix + originFromTag          |
+// | empty                  | empty           || empty                               |
+// | none                   | empty           || empty                               |
+// | empty                  | not empty       || container prefix + originFromMsg    |
+// | none                   | not empty       || container prefix + originFromMsg    |
+//  ---------------------------------------------------------------------------------
+func extractTagsMetadata(tags []string, defaultHostname, originFromUDS string, originFromMsg []byte, entityIDPrecedenceEnabled bool) ([]string, string, string, string, string) {
 	host := defaultHostname
 
 	n := 0
-	entityIDValue, cardinality := "", ""
+	originFromTag, cardinality := "", ""
 	for _, tag := range tags {
 		if strings.HasPrefix(tag, hostTagPrefix) {
 			host = tag[len(hostTagPrefix):]
 		} else if strings.HasPrefix(tag, entityIDTagPrefix) {
-			entityIDValue = tag[len(entityIDTagPrefix):]
+			originFromTag = tag[len(entityIDTagPrefix):]
 		} else if strings.HasPrefix(tag, CardinalityTagPrefix) {
 			cardinality = tag[len(CardinalityTagPrefix):]
 		} else {
@@ -39,25 +65,31 @@ func extractTagsMetadata(tags []string, defaultHostname string, originTags strin
 	}
 	tags = tags[:n]
 
-	origin := ""
+	udsOrigin := ""
 	// We use the UDS socket origin if no origin ID was specify in the tags
 	// or 'dogstatsd_entity_id_precedence' is set to False (default false).
-	if entityIDValue == "" || !entityIDPrecedenceEnabled {
+	if originFromTag == "" || !entityIDPrecedenceEnabled {
 		// Add origin tags only if the entity id tags is not provided
-		origin = originTags
+		udsOrigin = originFromUDS
 	}
 
-	k8sOrigin := ""
-	// We set k8sOriginID if the metrics contain a 'dd.internal.entity_id' tag different from 'none'.
-	if entityIDValue != "" && entityIDValue != entityIDIgnoreValue {
+	// originFromClient can either be originFromTag or originFromMsg
+	originFromClient := ""
+
+	// We set originFromClient if the metrics contain a 'dd.internal.entity_id' tag different from 'none'.
+	if originFromTag != "" && originFromTag != entityIDIgnoreValue {
 		// Check if the value is not "none" in order to avoid calling
 		// the tagger for entity that doesn't exist.
 
 		// currently only supported for pods
-		k8sOrigin = kubelet.KubePodTaggerEntityPrefix + entityIDValue
+		originFromClient = kubelet.KubePodTaggerEntityPrefix + originFromTag
+	} else if originFromTag == "" && len(originFromMsg) > 0 {
+		// originFromMsg is the container id sent by the newer clients.
+		// Opt-in is handled when parsing.
+		originFromClient = containers.BuildTaggerEntityName(string(originFromMsg))
 	}
 
-	return tags, host, origin, k8sOrigin, cardinality
+	return tags, host, udsOrigin, originFromClient, cardinality
 }
 
 func enrichMetricType(dogstatsdMetricType metricType) metrics.MetricType {
@@ -101,7 +133,7 @@ func isMetricBlocklisted(metricName string, metricBlocklist []string) bool {
 func enrichMetricSample(metricSamples []metrics.MetricSample, ddSample dogstatsdMetricSample, namespace string, excludedNamespaces []string,
 	metricBlocklist []string, defaultHostname string, origin string, entityIDPrecedenceEnabled bool, serverlessMode bool) []metrics.MetricSample {
 	metricName := ddSample.name
-	tags, hostnameFromTags, originID, k8sOriginID, cardinality := extractTagsMetadata(ddSample.tags, defaultHostname, origin, entityIDPrecedenceEnabled)
+	tags, hostnameFromTags, udsOrigin, clientOrigin, cardinality := extractTagsMetadata(ddSample.tags, defaultHostname, origin, ddSample.containerID, entityIDPrecedenceEnabled)
 
 	if !isExcluded(metricName, namespace, excludedNamespaces) {
 		metricName = namespace + metricName
@@ -124,33 +156,33 @@ func enrichMetricSample(metricSamples []metrics.MetricSample, ddSample dogstatsd
 		for idx := range ddSample.values {
 			metricSamples = append(metricSamples,
 				metrics.MetricSample{
-					Host:        hostnameFromTags,
-					Name:        metricName,
-					Tags:        tags,
-					Mtype:       mtype,
-					Value:       ddSample.values[idx],
-					SampleRate:  ddSample.sampleRate,
-					RawValue:    ddSample.setValue,
-					OriginID:    originID,
-					K8sOriginID: k8sOriginID,
-					Cardinality: cardinality,
+					Host:             hostnameFromTags,
+					Name:             metricName,
+					Tags:             tags,
+					Mtype:            mtype,
+					Value:            ddSample.values[idx],
+					SampleRate:       ddSample.sampleRate,
+					RawValue:         ddSample.setValue,
+					OriginFromUDS:    udsOrigin,
+					OriginFromClient: clientOrigin,
+					Cardinality:      cardinality,
 				})
 		}
 		return metricSamples
 	}
 
 	// only one value contained, simple append it
 	return append(metricSamples, metrics.MetricSample{
-		Host:        hostnameFromTags,
-		Name:        metricName,
-		Tags:        tags,
-		Mtype:       mtype,
-		Value:       ddSample.value,
-		SampleRate:  ddSample.sampleRate,
-		RawValue:    ddSample.setValue,
-		OriginID:    originID,
-		K8sOriginID: k8sOriginID,
-		Cardinality: cardinality,
+		Host:             hostnameFromTags,
+		Name:             metricName,
+		Tags:             tags,
+		Mtype:            mtype,
+		Value:            ddSample.value,
+		SampleRate:       ddSample.sampleRate,
+		RawValue:         ddSample.setValue,
+		OriginFromUDS:    udsOrigin,
+		OriginFromClient: clientOrigin,
+		Cardinality:      cardinality,
 	})
 }
 
@@ -179,20 +211,20 @@ func enrichEventAlertType(dogstatsdAlertType alertType) metrics.EventAlertType {
 }
 
 func enrichEvent(event dogstatsdEvent, defaultHostname string, origin string, entityIDPrecedenceEnabled bool) *metrics.Event {
-	tags, hostnameFromTags, originID, k8sOriginID, cardinality := extractTagsMetadata(event.tags, defaultHostname, origin, entityIDPrecedenceEnabled)
+	tags, hostnameFromTags, udsOrigin, clientOrigin, cardinality := extractTagsMetadata(event.tags, defaultHostname, origin, event.containerID, entityIDPrecedenceEnabled)
 
 	enrichedEvent := &metrics.Event{
-		Title:          event.title,
-		Text:           event.text,
-		Ts:             event.timestamp,
-		Priority:       enrichEventPriority(event.priority),
-		Tags:           tags,
-		AlertType:      enrichEventAlertType(event.alertType),
-		AggregationKey: event.aggregationKey,
-		SourceTypeName: event.sourceType,
-		OriginID:       originID,
-		K8sOriginID:    k8sOriginID,
-		Cardinality:    cardinality,
+		Title:            event.title,
+		Text:             event.text,
+		Ts:               event.timestamp,
+		Priority:         enrichEventPriority(event.priority),
+		Tags:             tags,
+		AlertType:        enrichEventAlertType(event.alertType),
+		AggregationKey:   event.aggregationKey,
+		SourceTypeName:   event.sourceType,
+		OriginFromUDS:    udsOrigin,
+		OriginFromClient: clientOrigin,
+		Cardinality:      cardinality,
 	}
 
 	if len(event.hostname) != 0 {
@@ -218,17 +250,17 @@ func enrichServiceCheckStatus(status serviceCheckStatus) metrics.ServiceCheckSta
 }
 
 func enrichServiceCheck(serviceCheck dogstatsdServiceCheck, defaultHostname string, origin string, entityIDPrecedenceEnabled bool) *metrics.ServiceCheck {
-	tags, hostnameFromTags, originID, k8sOriginID, cardinality := extractTagsMetadata(serviceCheck.tags, defaultHostname, origin, entityIDPrecedenceEnabled)
+	tags, hostnameFromTags, udsOrigin, clientOrigin, cardinality := extractTagsMetadata(serviceCheck.tags, defaultHostname, origin, serviceCheck.containerID, entityIDPrecedenceEnabled)
 
 	enrichedServiceCheck := &metrics.ServiceCheck{
-		CheckName:   serviceCheck.name,
-		Ts:          serviceCheck.timestamp,
-		Status:      enrichServiceCheckStatus(serviceCheck.status),
-		Message:     serviceCheck.message,
-		Tags:        tags,
-		OriginID:    originID,
-		K8sOriginID: k8sOriginID,
-		Cardinality: cardinality,
+		CheckName:        serviceCheck.name,
+		Ts:               serviceCheck.timestamp,
+		Status:           enrichServiceCheckStatus(serviceCheck.status),
+		Message:          serviceCheck.message,
+		Tags:             tags,
+		OriginFromUDS:    udsOrigin,
+		OriginFromClient: clientOrigin,
+		Cardinality:      cardinality,
 	}
 
 	if len(serviceCheck.hostname) != 0 {

pkg/dogstatsd/enrich_bench_test.go

@@ -29,7 +29,7 @@ func BenchmarkExtractTagsMetadata(b *testing.B) {
 			sb.ResetTimer()
 
 			for n := 0; n < sb.N; n++ {
-				tags, _, _, _, _ = extractTagsMetadata(baseTags, "hostname", "", false)
+				tags, _, _, _, _ = extractTagsMetadata(baseTags, "hostname", "", []byte{}, false)
 			}
 		})
 	}