Allow customers to set custom resource attributes as container Tags

.chloggen/dinesh.gurumurthy_expand-resource-attributes.yaml

@@ -0,0 +1,23 @@
+# 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: datadog/connetor
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Enable connetor to use any attribute from the resource as Container Tag
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32224]
+# (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: |
+  This change allows the connector to use any attribute from the resource as a container tag. This is useful when you want to use a custom attribute from the resource as a container tag. For example, you can use the `namespace` attribute from the resource as a container tag. This change is backward compatible and does not require any changes to the existing configuration.
+# 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/connector.go

@@ -40,7 +40,7 @@ type traceToMetricConnector struct {
 	// from the agent to OTLP Metrics.
 	translator *metrics.Translator
 
-	resourceAttrs     map[string]string
+	enrichedTags      map[string]string
 	containerTagCache *cache.Cache
 
 	// in specifies the channel through which the agent will output Stats Payloads
@@ -81,7 +81,17 @@ func newTraceToMetricConnector(set component.TelemetrySettings, cfg component.Co
 	for _, val := range cfg.(*Config).Traces.ResourceAttributesAsContainerTags {
 		ctags[val] = ""
 	}
-	ddtags := attributes.ContainerTagFromAttributes(ctags)
+	ddTags := attributes.ContainerTagFromAttributes(ctags)
+
+	for key := range ctags {
+		// TODO: we need to expose the container mappings from the attributes package
+		tags := attributes.ContainerTagFromAttributes(map[string]string{key: ""})
+		// It means there is no mapping for the resource attribute
+		if len(tags) == 0 {
+			ddTags[key] = ""
+		}
+
+	}
 
 	ctx := context.Background()
 	return &traceToMetricConnector{
@@ -90,7 +100,7 @@ func newTraceToMetricConnector(set component.TelemetrySettings, cfg component.Co
 		translator:        trans,
 		in:                in,
 		metricsConsumer:   metricsConsumer,
-		resourceAttrs:     ddtags,
+		enrichedTags:      ddTags,
 		containerTagCache: cache.New(cacheExpiration, cacheCleanupInterval),
 		exit:              make(chan struct{}),
 	}, nil
@@ -146,31 +156,34 @@ func (c *traceToMetricConnector) Capabilities() consumer.Capabilities {
 	return consumer.Capabilities{MutatesData: false}
 }
 
+func (c *traceToMetricConnector) addToCache(containerID string, key string) {
+	if tags, ok := c.containerTagCache.Get(containerID); ok {
+		tagList := tags.(*sync.Map)
+		tagList.Store(key, struct{}{})
+	} else {
+		tagList := &sync.Map{}
+		tagList.Store(key, struct{}{})
+		c.containerTagCache.Set(containerID, tagList, cache.DefaultExpiration)
+	}
+}
+
 func (c *traceToMetricConnector) populateContainerTagsCache(traces ptrace.Traces) {
 	for i := 0; i < traces.ResourceSpans().Len(); i++ {
 		rs := traces.ResourceSpans().At(i)
 		attrs := rs.Resource().Attributes()
+
 		containerID, ok := attrs.Get(semconv.AttributeContainerID)
 		if !ok {
 			continue
 		}
 		ddContainerTags := attributes.ContainerTagsFromResourceAttributes(attrs)
-		for attr := range c.resourceAttrs {
+		for attr := range c.enrichedTags {
 			if val, ok := ddContainerTags[attr]; ok {
 				key := fmt.Sprintf("%s:%s", attr, val)
-				if v, ok := c.containerTagCache.Get(containerID.AsString()); ok {
-					cacheVal := v.(*sync.Map)
-					// check if the key already exists in the cache
-					if _, ok := cacheVal.Load(key); ok {
-						continue
-					}
-					cacheVal.Store(key, struct{}{})
-				} else {
-					cacheVal := &sync.Map{}
-					cacheVal.Store(key, struct{}{})
-					c.containerTagCache.Set(containerID.AsString(), cacheVal, cache.DefaultExpiration)
-				}
-
+				c.addToCache(containerID.AsString(), key)
+			} else if incomingVal, ok := attrs.Get(attr); ok {
+				key := fmt.Sprintf("%s:%s", attr, incomingVal.Str())
+				c.addToCache(containerID.AsString(), key)
 			}
 		}
 	}
@@ -214,7 +227,7 @@ func (c *traceToMetricConnector) run() {
 			var mx pmetric.Metrics
 			var err error
 			// Enrich the stats with container tags
-			if len(c.resourceAttrs) > 0 {
+			if len(c.enrichedTags) > 0 {
 				c.enrichStatsPayload(stats)
 			}
 

connector/datadogconnector/connector_test.go

@@ -68,6 +68,8 @@ func generateTrace() ptrace.Traces {
 	res.Attributes().PutStr("container.id", "my-container-id")
 	res.Attributes().PutStr("cloud.availability_zone", "my-zone")
 	res.Attributes().PutStr("cloud.region", "my-region")
+	// add a custom Resource attribute
+	res.Attributes().PutStr("az", "my-az")
 
 	ss := td.ResourceSpans().At(0).ScopeSpans().AppendEmpty().Spans()
 	ss.EnsureCapacity(1)
@@ -103,7 +105,7 @@ func creteConnector(t *testing.T) (*traceToMetricConnector, *consumertest.Metric
 
 	creationParams := connectortest.NewNopCreateSettings()
 	cfg := factory.CreateDefaultConfig().(*Config)
-	cfg.Traces.ResourceAttributesAsContainerTags = []string{semconv.AttributeCloudAvailabilityZone, semconv.AttributeCloudRegion}
+	cfg.Traces.ResourceAttributesAsContainerTags = []string{semconv.AttributeCloudAvailabilityZone, semconv.AttributeCloudRegion, "az"}
 
 	metricsSink := &consumertest.MetricsSink{}
 
@@ -142,7 +144,7 @@ func TestContainerTags(t *testing.T) {
 		count++
 		return true
 	})
-	assert.Equal(t, 2, count)
+	assert.Equal(t, 3, count)
 
 	for {
 		if len(metricsSink.AllMetrics()) > 0 {
@@ -166,8 +168,8 @@ func TestContainerTags(t *testing.T) {
 	require.NoError(t, err)
 
 	tags := sp.Stats[0].Tags
-	assert.Equal(t, 2, len(tags))
-	assert.ElementsMatch(t, []string{"region:my-region", "zone:my-zone"}, tags)
+	assert.Equal(t, 3, len(tags))
+	assert.ElementsMatch(t, []string{"region:my-region", "zone:my-zone", "az:my-az"}, tags)
 }
 
 func newTranslatorWithStatsChannel(t *testing.T, logger *zap.Logger, ch chan []byte) *otlpmetrics.Translator {