Add resource_attributes support to opentelemetry section (#2207)

Author: movence

translator/cmdutil/translatorutil_test.go

@@ -431,6 +431,13 @@ func TestOpenTelemetryOtlpSchemaValidation(t *testing.T) {
 	checkIfSchemaValidateAsExpected(t, "../../translator/config/sampleSchema/opentelemetry/validOpenTelemetryOtlp.json", true, map[string]int{})
 }
 
+func TestOpenTelemetryResourceAttributesSchemaValidation(t *testing.T) {
+	checkIfSchemaValidateAsExpected(t, "../../translator/config/sampleSchema/opentelemetry/validOpenTelemetryResourceAttributes.json", true, map[string]int{})
+	checkIfSchemaValidateAsExpected(t, "../../translator/config/sampleSchema/opentelemetry/invalidOpenTelemetryResourceAttributes.json", false, map[string]int{
+		"invalid_type": 1,
+	})
+}
+
 func TestDefaultOtelConfigSchemaValidation(t *testing.T) {
 	cfg, ok := config.DefaultJSONConfigFor("otel")
 	require.True(t, ok)

translator/config/sampleSchema/opentelemetry/invalidOpenTelemetryResourceAttributes.json

@@ -0,0 +1,15 @@
+{
+  "agent": {
+    "region": "us-west-2"
+  },
+  "opentelemetry": {
+    "resource_attributes": {
+      "team": 123
+    },
+    "collect": {
+      "otlp": {
+        "grpc_endpoint": "127.0.0.1:4317"
+      }
+    }
+  }
+}

translator/config/sampleSchema/opentelemetry/validOpenTelemetryResourceAttributes.json

@@ -0,0 +1,17 @@
+{
+  "agent": {
+    "region": "us-west-2"
+  },
+  "opentelemetry": {
+    "resource_attributes": {
+      "team": "cloudwatch",
+      "environment": "prod"
+    },
+    "collect": {
+      "otlp": {
+        "grpc_endpoint": "127.0.0.1:4317",
+        "http_endpoint": "127.0.0.1:4318"
+      }
+    }
+  }
+}

translator/config/schema.json

@@ -1810,6 +1810,16 @@
           "description": "The name of the K8s cluster. Applied to all telemetry collected under opentelemetry.",
           "type": "string"
         },
+        "resource_attributes": {
+          "type": "object",
+          "description": "Additional resource attributes to add to every metric, log, and trace exported by the opentelemetry section. Applied via a resource processor at the front of each export pipeline. Attributes are added before resource detection, so for keys the agent auto-detects (e.g. cloud.region, host.id, cloud.provider) the detected value takes precedence over the configured one; use this for attributes the agent does not detect. Keys managed internally for log routing (aws.log.group.name, aws.log.stream.name, aws.log.source) are reserved and rejected.",
+          "maxProperties": 30,
+          "additionalProperties": {
+            "type": "string",
+            "minLength": 1,
+            "maxLength": 1024
+          }
+        },
         "collect": {
           "type": "object",
           "properties": {

translator/tocwconfig/sampleConfig/opentelemetry/otlp_otel_config.json

@@ -1,6 +1,10 @@
 {
   "agent": {"region": "us-west-2"},
   "opentelemetry": {
+    "resource_attributes": {
+      "team": "cloudwatch",
+      "deployment.environment": "prod"
+    },
     "collect": {
       "otlp": {
         "grpc_endpoint": "127.0.0.1:4317",

translator/tocwconfig/sampleConfig/opentelemetry/otlp_otel_config.yaml

@@ -176,6 +176,22 @@ processors:
         send_batch_max_size: 10000
         send_batch_size: 10000
         timeout: 30s
+    resource/opentelemetry:
+        attributes:
+            - action: upsert
+              converted_type: ""
+              from_attribute: ""
+              from_context: ""
+              key: deployment.environment
+              pattern: ""
+              value: prod
+            - action: upsert
+              converted_type: ""
+              from_attribute: ""
+              from_context: ""
+              key: team
+              pattern: ""
+              value: cloudwatch
     resourcedetection/opentelemetry:
         aks:
             resource_attributes:
@@ -582,6 +598,7 @@ service:
             exporters:
                 - otlphttp/logs
             processors:
+                - resource/opentelemetry
                 - resourcedetection/opentelemetry
                 - transform/identity
                 - transform/logs_routing
@@ -603,6 +620,7 @@ service:
             exporters:
                 - otlphttp/metrics
             processors:
+                - resource/opentelemetry
                 - resourcedetection/opentelemetry
                 - transform/identity
                 - batch/opentelemetry_metrics
@@ -620,6 +638,7 @@ service:
             exporters:
                 - otlphttp/traces
             processors:
+                - resource/opentelemetry
                 - resourcedetection/opentelemetry
                 - transform/identity
                 - batch/opentelemetry_traces

translator/translate/otel/common/common.go

@@ -32,6 +32,7 @@ const (
 	TracesCollectedKey                             = "traces_collected"
 	OpenTelemetryKey                               = "opentelemetry"
 	CollectKey                                     = "collect"
+	ResourceAttributesKey                          = "resource_attributes"
 	HostMetricsKey                                 = "host_metrics"
 	OtelContainerInsightsKey                       = "container_insights"
 	MetricsDestinationsKey                         = "metrics_destinations"
@@ -169,6 +170,8 @@ var (
 	OtelSpanMetricsEnabledKey   = ConfigKey(OpenTelemetryKey, CollectKey, OtlpKey, "span_metrics_enabled")
 	WindowsEventsConfigKey      = ConfigKey(OpenTelemetryKey, CollectKey, WindowsEventsKey)
 	FilesConfigKey              = ConfigKey(OpenTelemetryKey, CollectKey, FilesKey)
+	// OtelResourceAttributesKey holds customer-supplied resource attributes added to every opentelemetry export pipeline.
+	OtelResourceAttributesKey = ConfigKey(OpenTelemetryKey, ResourceAttributesKey)
 )
 
 const (
@@ -375,6 +378,26 @@ func GetString(conf *confmap.Conf, key string) (string, bool) {
 	return "", false
 }
 
+// GetStringMap gets the key/value pairs for the key as a map[string]string,
+// coercing non-string values to their string form. Returns nil if the key is
+// missing or is not a map.
+func GetStringMap(conf *confmap.Conf, key string) map[string]string {
+	value := conf.Get(key)
+	raw, ok := value.(map[string]any)
+	if !ok {
+		return nil
+	}
+	result := make(map[string]string, len(raw))
+	for k, v := range raw {
+		if s, ok := v.(string); ok {
+			result[k] = s
+		} else {
+			result[k] = fmt.Sprintf("%v", v)
+		}
+	}
+	return result
+}
+
 // GetArray gets the array value for the key. If the key is missing,
 // the return value will be nil
 func GetArray[C any](conf *confmap.Conf, key string) []C {

translator/translate/otel/common/common_test.go

@@ -49,6 +49,18 @@ func TestGetString(t *testing.T) {
 	require.Equal(t, "", got)
 }
 
+func TestGetStringMap(t *testing.T) {
+	conf := confmap.NewFromStringMap(map[string]any{
+		"attrs":   map[string]any{"team": "cloudwatch", "count": 3},
+		"notAMap": "value",
+	})
+	got := GetStringMap(conf, "attrs")
+	require.Equal(t, map[string]string{"team": "cloudwatch", "count": "3"}, got)
+
+	require.Nil(t, GetStringMap(conf, "notAMap"))
+	require.Nil(t, GetStringMap(conf, "missing"))
+}
+
 func TestGetArray(t *testing.T) {
 	conf := confmap.NewFromStringMap(map[string]any{
 		"int":    []any{5, 8, 10},

translator/translate/otel/pipeline/opentelemetry/resource_attributes.go

@@ -0,0 +1,42 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package opentelemetry
+
+import (
+	"go.opentelemetry.io/collector/confmap"
+
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/resourceprocessor"
+)
+
+// reservedResourceAttributeKeys are attributes the agent manages internally for
+// log routing; customers must not override them via resource_attributes.
+var reservedResourceAttributeKeys = []string{
+	"aws.log.group.name",
+	"aws.log.stream.name",
+	"aws.log.source",
+}
+
+// resourceAttributesProcessor returns a resource processor that upserts the
+// customer-supplied opentelemetry.resource_attributes onto every record, or nil
+// if none are configured. Callers place it at the front of the processor list so
+// the attributes are present before any downstream processing.
+//
+// Note: it runs before resourcedetection (override: true), so for keys the agent
+// also auto-detects (e.g. cloud.region, host.id) the detected value wins. This is
+// intentional; the field is for adding attributes the agent does not detect.
+func resourceAttributesProcessor(conf *confmap.Conf) common.ComponentTranslator {
+	if conf == nil {
+		return nil
+	}
+	attrs := common.GetStringMap(conf, common.OtelResourceAttributesKey)
+	if len(attrs) == 0 {
+		return nil
+	}
+	return resourceprocessor.NewTranslator(
+		common.WithName(common.OpenTelemetryKey),
+		resourceprocessor.WithAttributes(attrs),
+		resourceprocessor.WithReservedKeys(reservedResourceAttributeKeys...),
+	)
+}

translator/translate/otel/pipeline/opentelemetry/resource_attributes_test.go

@@ -0,0 +1,72 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package opentelemetry
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+func TestResourceAttributesProcessor(t *testing.T) {
+	// nil conf -> no processor
+	assert.Nil(t, resourceAttributesProcessor(nil))
+
+	// no resource_attributes key -> no processor
+	conf := confmap.NewFromStringMap(map[string]interface{}{
+		"opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"otlp": map[string]interface{}{}}},
+	})
+	assert.Nil(t, resourceAttributesProcessor(conf))
+
+	// empty map -> no processor
+	conf = confmap.NewFromStringMap(map[string]interface{}{
+		"opentelemetry": map[string]interface{}{"resource_attributes": map[string]interface{}{}},
+	})
+	assert.Nil(t, resourceAttributesProcessor(conf))
+
+	// populated map -> processor present, Translate succeeds
+	conf = confmap.NewFromStringMap(map[string]interface{}{
+		"opentelemetry": map[string]interface{}{"resource_attributes": map[string]interface{}{"team": "cloudwatch"}},
+	})
+	tr := resourceAttributesProcessor(conf)
+	require.NotNil(t, tr)
+	assert.Equal(t, "resource/opentelemetry", tr.ID().String())
+	cfg, err := tr.Translate(conf)
+	require.NoError(t, err)
+	assert.NotNil(t, cfg)
+}
+
+func TestResourceAttributesProcessor_ReservedKeyRejected(t *testing.T) {
+	for _, key := range reservedResourceAttributeKeys {
+		t.Run(key, func(t *testing.T) {
+			conf := confmap.NewFromStringMap(map[string]interface{}{
+				"opentelemetry": map[string]interface{}{
+					"resource_attributes": map[string]interface{}{key: "/hijacked"},
+				},
+			})
+			tr := resourceAttributesProcessor(conf)
+			require.NotNil(t, tr)
+			cfg, err := tr.Translate(conf)
+			require.Error(t, err)
+			assert.Nil(t, cfg)
+			assert.Contains(t, err.Error(), "reserved")
+		})
+	}
+}
+
+func TestResourceAttributesProcessor_EmptyKeyRejected(t *testing.T) {
+	conf := confmap.NewFromStringMap(map[string]interface{}{
+		"opentelemetry": map[string]interface{}{
+			"resource_attributes": map[string]interface{}{"": "value"},
+		},
+	})
+	tr := resourceAttributesProcessor(conf)
+	require.NotNil(t, tr)
+	cfg, err := tr.Translate(conf)
+	require.Error(t, err)
+	assert.Nil(t, cfg)
+	assert.Contains(t, err.Error(), "must not be empty")
+}