[chore]: datadog receiver: Tags translation (#33922)

**Description**:
This PR is a follow up to the former
#33631
extending the existing tags translation structure. This will be required
for the follow up PRs adding support for v1 and v2 series endpoints,
service checks, as well as sketches.

The full version of the code can be found in the
cedwards/datadog-metrics-receiver-full branch, or in Grafana Alloy:
https://github.com/grafana/alloy/tree/main/internal/etc/datadogreceiver

**Link to tracking Issue:**

#18278

**Testing**:
Unit tests have been added. More thorough tests will be included in
follow-up PRs as the remaining functionality is added.

**Notes**:
- Adding `[chore]` to the title of the PR because
https://github.com/grafana/opentelemetry-collector-contrib/blob/ab4d726aaaa07aad702ff3b312a8e261f2b38021/.chloggen/datadogreceiver_metrics.yaml#L1-L27
already exists.

---------

Signed-off-by: Jesus Vazquez <jesus.vazquez@grafana.com>

receiver/datadogreceiver/tags.go

@@ -0,0 +1,142 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package datadogreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/datadogreceiver"
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	semconv "go.opentelemetry.io/collector/semconv/v1.16.0"
+)
+
+// See:
+// https://docs.datadoghq.com/opentelemetry/schema_semantics/semantic_mapping/
+// https://github.com/DataDog/opentelemetry-mapping-go/blob/main/pkg/otlp/attributes/attributes.go
+var datadogKnownResourceAttributes = map[string]string{
+	"env":     semconv.AttributeDeploymentEnvironment,
+	"service": semconv.AttributeServiceName,
+	"version": semconv.AttributeServiceVersion,
+
+	// Container-related attributes
+	"container_id":   semconv.AttributeContainerID,
+	"container_name": semconv.AttributeContainerName,
+	"image_name":     semconv.AttributeContainerImageName,
+	"image_tag":      semconv.AttributeContainerImageTag,
+	"runtime":        semconv.AttributeContainerRuntime,
+
+	// Cloud-related attributes
+	"cloud_provider": semconv.AttributeCloudProvider,
+	"region":         semconv.AttributeCloudRegion,
+	"zone":           semconv.AttributeCloudAvailabilityZone,
+
+	// ECS-related attributes
+	"task_family":        semconv.AttributeAWSECSTaskFamily,
+	"task_arn":           semconv.AttributeAWSECSTaskARN,
+	"ecs_cluster_name":   semconv.AttributeAWSECSClusterARN,
+	"task_version":       semconv.AttributeAWSECSTaskRevision,
+	"ecs_container_name": semconv.AttributeAWSECSContainerARN,
+
+	// K8-related attributes
+	"kube_container_name": semconv.AttributeK8SContainerName,
+	"kube_cluster_name":   semconv.AttributeK8SClusterName,
+	"kube_deployment":     semconv.AttributeK8SDeploymentName,
+	"kube_replica_set":    semconv.AttributeK8SReplicaSetName,
+	"kube_stateful_set":   semconv.AttributeK8SStatefulSetName,
+	"kube_daemon_set":     semconv.AttributeK8SDaemonSetName,
+	"kube_job":            semconv.AttributeK8SJobName,
+	"kube_cronjob":        semconv.AttributeK8SCronJobName,
+	"kube_namespace":      semconv.AttributeK8SNamespaceName,
+	"pod_name":            semconv.AttributeK8SPodName,
+
+	// Other
+	"process_id":       semconv.AttributeProcessPID,
+	"error.stacktrace": semconv.AttributeExceptionStacktrace,
+	"error.msg":        semconv.AttributeExceptionMessage,
+}
+
+// translateDatadogTagToKeyValuePair translates a Datadog tag to a key value pair
+func translateDatadogTagToKeyValuePair(tag string) (key string, value string) {
+	if tag == "" {
+		return "", ""
+	}
+
+	key, val, ok := strings.Cut(tag, ":")
+	if !ok {
+		// Datadog allows for two tag formats, one of which includes a key such as 'env',
+		// followed by a value. Datadog also supports inputTags without the key, but OTel seems
+		// to only support key:value pairs.
+		// The following is a workaround to map unnamed inputTags to key:value pairs and its subject to future
+		// changes if OTel supports unnamed inputTags in the future or if there is a better way to do this.
+		key = fmt.Sprintf("unnamed_%s", tag)
+		val = tag
+	}
+	return key, val
+}
+
+// translateDatadogKeyToOTel translates a Datadog key to an OTel key
+func translateDatadogKeyToOTel(k string) string {
+	if otelKey, ok := datadogKnownResourceAttributes[strings.ToLower(k)]; ok {
+		return otelKey
+	}
+	return k
+}
+
+type StringPool struct {
+	sync.RWMutex
+	pool map[string]string
+}
+
+func newStringPool() *StringPool {
+	return &StringPool{
+		pool: make(map[string]string),
+	}
+}
+
+func (s *StringPool) Intern(str string) string {
+	s.RLock()
+	interned, ok := s.pool[str]
+	s.RUnlock()
+
+	if ok {
+		return interned
+	}
+
+	s.Lock()
+	// Double check if another goroutine has added the string after releasing the read lock
+	interned, ok = s.pool[str]
+	if !ok {
+		interned = str
+		s.pool[str] = str
+	}
+	s.Unlock()
+
+	return interned
+}
+
+func tagsToAttributes(tags []string, host string, stringPool *StringPool) (pcommon.Map, pcommon.Map, pcommon.Map) {
+	resourceAttrs := pcommon.NewMap()
+	scopeAttrs := pcommon.NewMap()
+	dpAttrs := pcommon.NewMap()
+
+	if host != "" {
+		resourceAttrs.PutStr(semconv.AttributeHostName, host)
+	}
+
+	var key, val string
+	for _, tag := range tags {
+		key, val = translateDatadogTagToKeyValuePair(tag)
+		if attr, ok := datadogKnownResourceAttributes[key]; ok {
+			val = stringPool.Intern(val) // No need to intern the key if we already have it
+			resourceAttrs.PutStr(attr, val)
+		} else {
+			key = stringPool.Intern(translateDatadogKeyToOTel(key))
+			val = stringPool.Intern(val)
+			dpAttrs.PutStr(key, val)
+		}
+	}
+
+	return resourceAttrs, scopeAttrs, dpAttrs
+}

receiver/datadogreceiver/tags_test.go

@@ -0,0 +1,156 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package datadogreceiver
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+)
+
+func TestGetMetricAttributes(t *testing.T) {
+	cases := []struct {
+		name                  string
+		tags                  []string
+		host                  string
+		expectedResourceAttrs pcommon.Map
+		expectedScopeAttrs    pcommon.Map
+		expectedDpAttrs       pcommon.Map
+	}{
+		{
+			name:                  "empty",
+			tags:                  []string{},
+			host:                  "",
+			expectedResourceAttrs: pcommon.NewMap(),
+			expectedScopeAttrs:    pcommon.NewMap(),
+			expectedDpAttrs:       pcommon.NewMap(),
+		},
+		{
+			name: "host",
+			tags: []string{},
+			host: "host",
+			expectedResourceAttrs: newMapFromKV(t, map[string]any{
+				"host.name": "host",
+			}),
+			expectedScopeAttrs: pcommon.NewMap(),
+			expectedDpAttrs:    pcommon.NewMap(),
+		},
+		{
+			name: "provides both host and tags where some tag keys have to replaced by otel conventions",
+			tags: []string{"env:prod", "service:my-service", "version:1.0"},
+			host: "host",
+			expectedResourceAttrs: newMapFromKV(t, map[string]any{
+				"host.name":              "host",
+				"deployment.environment": "prod",
+				"service.name":           "my-service",
+				"service.version":        "1.0",
+			}),
+			expectedScopeAttrs: pcommon.NewMap(),
+			expectedDpAttrs:    pcommon.NewMap(),
+		},
+		{
+			name: "provides host, tags and unnamed tags",
+			tags: []string{"env:prod", "foo"},
+			host: "host",
+			expectedResourceAttrs: newMapFromKV(t, map[string]any{
+				"host.name":              "host",
+				"deployment.environment": "prod",
+			}),
+			expectedScopeAttrs: pcommon.NewMap(),
+			expectedDpAttrs: newMapFromKV(t, map[string]any{
+				"unnamed_foo": "foo",
+			}),
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			pool := newStringPool()
+			resourceAttrs, scopeAttrs, dpAttrs := tagsToAttributes(c.tags, c.host, pool)
+
+			assert.Equal(t, c.expectedResourceAttrs.Len(), resourceAttrs.Len())
+			c.expectedResourceAttrs.Range(func(k string, _ pcommon.Value) bool {
+				ev, _ := c.expectedResourceAttrs.Get(k)
+				av, ok := resourceAttrs.Get(k)
+				assert.True(t, ok)
+				assert.Equal(t, ev, av)
+				return true
+			})
+
+			assert.Equal(t, c.expectedScopeAttrs.Len(), scopeAttrs.Len())
+			c.expectedScopeAttrs.Range(func(k string, _ pcommon.Value) bool {
+				ev, _ := c.expectedScopeAttrs.Get(k)
+				av, ok := scopeAttrs.Get(k)
+				assert.True(t, ok)
+				assert.Equal(t, ev, av)
+				return true
+			})
+
+			assert.Equal(t, c.expectedDpAttrs.Len(), dpAttrs.Len())
+			c.expectedDpAttrs.Range(func(k string, _ pcommon.Value) bool {
+				ev, _ := c.expectedDpAttrs.Get(k)
+				av, ok := dpAttrs.Get(k)
+				assert.True(t, ok)
+				assert.Equal(t, ev, av)
+				return true
+			})
+		})
+
+	}
+
+}
+
+func newMapFromKV(t *testing.T, kv map[string]any) pcommon.Map {
+	m := pcommon.NewMap()
+	err := m.FromRaw(kv)
+	assert.NoError(t, err)
+	return m
+}
+
+func TestDatadogTagToKeyValuePair(t *testing.T) {
+	cases := []struct {
+		name          string
+		input         string
+		expectedKey   string
+		expectedValue string
+	}{
+		{
+			name:          "empty",
+			input:         "",
+			expectedKey:   "",
+			expectedValue: "",
+		},
+		{
+			name:          "kv tag",
+			input:         "foo:bar",
+			expectedKey:   "foo",
+			expectedValue: "bar",
+		},
+		{
+			name:          "unnamed tag",
+			input:         "foo",
+			expectedKey:   "unnamed_foo",
+			expectedValue: "foo",
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			key, value := translateDatadogTagToKeyValuePair(c.input)
+			assert.Equal(t, c.expectedKey, key, "Expected key %s, got %s", c.expectedKey, key)
+			assert.Equal(t, c.expectedValue, value, "Expected value %s, got %s", c.expectedValue, value)
+		})
+	}
+
+}
+
+func TestTranslateDataDogKeyToOtel(t *testing.T) {
+	// make sure all known keys are translated
+	for k, v := range datadogKnownResourceAttributes {
+		t.Run(k, func(t *testing.T) {
+			assert.Equal(t, v, translateDatadogKeyToOTel(k))
+		})
+	}
+}

receiver/datadogreceiver/traces_translator.go

@@ -73,7 +73,7 @@ func toTraces(payload *pb.TracerPayload, req *http.Request) ptrace.Traces {
 	}
 
 	for k, v := range payload.Tags {
-		if k = translateDataDogKeyToOtel(k); v != "" {
+		if k = translateDatadogKeyToOTel(k); v != "" {
 			sharedAttributes.PutStr(k, v)
 		}
 	}
@@ -110,7 +110,7 @@ func toTraces(payload *pb.TracerPayload, req *http.Request) ptrace.Traces {
 			newSpan.Attributes().PutStr(attributeDatadogSpanID, strconv.FormatUint(span.SpanID, 10))
 			newSpan.Attributes().PutStr(attributeDatadogTraceID, strconv.FormatUint(span.TraceID, 10))
 			for k, v := range span.GetMeta() {
-				if k = translateDataDogKeyToOtel(k); len(k) > 0 {
+				if k = translateDatadogKeyToOTel(k); len(k) > 0 {
 					newSpan.Attributes().PutStr(k, v)
 				}
 			}
@@ -155,32 +155,6 @@ func toTraces(payload *pb.TracerPayload, req *http.Request) ptrace.Traces {
 	return results
 }
 
-func translateDataDogKeyToOtel(k string) string {
-	switch strings.ToLower(k) {
-	case "env":
-		return semconv.AttributeDeploymentEnvironment
-	case "version":
-		return semconv.AttributeServiceVersion
-	case "container_id":
-		return semconv.AttributeContainerID
-	case "container_name":
-		return semconv.AttributeContainerName
-	case "image_name":
-		return semconv.AttributeContainerImageName
-	case "image_tag":
-		return semconv.AttributeContainerImageTag
-	case "process_id":
-		return semconv.AttributeProcessPID
-	case "error.stacktrace":
-		return semconv.AttributeExceptionStacktrace
-	case "error.msg":
-		return semconv.AttributeExceptionMessage
-	default:
-		return k
-	}
-
-}
-
 var bufferPool = sync.Pool{
 	New: func() any {
 		return new(bytes.Buffer)