diff --git a/translator/config/sampleSchema/opentelemetry/invalidOpenTelemetryPrometheus.json b/translator/config/sampleSchema/opentelemetry/invalidOpenTelemetryPrometheus.json index 0026d88f4d4..b7aacb3a602 100644 --- a/translator/config/sampleSchema/opentelemetry/invalidOpenTelemetryPrometheus.json +++ b/translator/config/sampleSchema/opentelemetry/invalidOpenTelemetryPrometheus.json @@ -3,9 +3,9 @@ "region": "us-west-2" }, "opentelemetry": { + "cluster_name": "my-cluster", "collect": { "prometheus": { - "cluster_name": "my-cluster" } } } diff --git a/translator/config/sampleSchema/opentelemetry/validCombinedV1V2Config.json b/translator/config/sampleSchema/opentelemetry/validCombinedV1V2Config.json index 4cbf9873247..27e7c56cb9f 100644 --- a/translator/config/sampleSchema/opentelemetry/validCombinedV1V2Config.json +++ b/translator/config/sampleSchema/opentelemetry/validCombinedV1V2Config.json @@ -34,12 +34,12 @@ } }, "opentelemetry": { + "cluster_name": "TestCluster", "collect": { "host_metrics": { - "metrics_collection_interval": 60 + "collection_interval": 60 }, "container_insights": { - "cluster_name": "TestCluster", "collection_interval": 30 }, "database_insights": { diff --git a/translator/config/schema.json b/translator/config/schema.json index 13daebe2d97..9875a6f2d2a 100644 --- a/translator/config/schema.json +++ b/translator/config/schema.json @@ -1806,6 +1806,10 @@ "opentelemetryDefinition": { "type": "object", "properties": { + "cluster_name": { + "description": "The name of the K8s cluster. Applied to all telemetry collected under opentelemetry.", + "type": "string" + }, "collect": { "type": "object", "properties": { @@ -1840,7 +1844,7 @@ "host_metrics": { "type": "object", "properties": { - "metrics_collection_interval": { + "collection_interval": { "description": "How often the metrics defined will be collected", "$ref": "#/definitions/timeIntervalDefinition" } @@ -1850,16 +1854,12 @@ "container_insights": { "type": "object", "properties": { - "cluster_name": { - "description": "The name of the EKS cluster", - "type": "string" - }, "collection_interval": { "description": "How often metrics are collected", "$ref": "#/definitions/timeIntervalDefinition" }, - "mode": { - "description": "Pipeline mode: node (daemonset), cluster (deployment). If omitted, falls back to CWAGENT_ROLE env var", + "role": { + "description": "Pipeline role: node (daemonset), cluster (deployment). If omitted, falls back to CWAGENT_ROLE env var", "type": "string", "enum": ["node", "cluster"] }, @@ -1874,7 +1874,6 @@ "additionalProperties": false } }, - "required": ["cluster_name"], "additionalProperties": false }, "prometheus": { @@ -1883,10 +1882,6 @@ "config_path": { "description": "Path to the Prometheus scrape configuration file", "type": "string" - }, - "cluster_name": { - "description": "The name of the EKS cluster", - "type": "string" } }, "required": ["config_path"], diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_ec2_config.json b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_ec2_config.json index d63dc6fcb97..69a53e56534 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_ec2_config.json +++ b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_ec2_config.json @@ -125,9 +125,10 @@ "endpoint_override": "https://x-ray-endpoint.us-west-2.amazonaws.com" }, "opentelemetry": { + "cluster_name": "TestCluster", "collect": { "host_metrics": { - "metrics_collection_interval": 60 + "collection_interval": 60 }, "database_insights": { "postgresql": [ diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_ec2_config.yaml b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_ec2_config.yaml index 4eb25ca8ecd..0b090f80b2d 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_ec2_config.yaml +++ b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_ec2_config.yaml @@ -542,9 +542,9 @@ processors: ec2_instance_tag_keys: - AutoScalingGroupName ec2_metadata_tags: - - ImageId - InstanceId - InstanceType + - ImageId imds_retries: 1 middleware: agenthealth/statuscode refresh_tags_interval: 0s @@ -1027,6 +1027,24 @@ processors: - set(scope.attributes["cloudwatch.source"], "cloudwatch-agent") - set(scope.attributes["cloudwatch.solution"], "otel-prometheus") trace_statements: [] + transform/set_cluster_name: + error_mode: ignore + flatten_data: false + log_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") + metric_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") + trace_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") receivers: awsxray: endpoint: 0.0.0.0:2000 @@ -1562,6 +1580,7 @@ service: - otlphttp/logs processors: - resourcedetection/opentelemetry + - transform/set_cluster_name - transform/identity - transform/logs_routing - attributestocontext/opentelemetry @@ -1596,13 +1615,13 @@ service: - ec2tagger - awsentity/resource receivers: - - telegraf_cpu - - telegraf_netstat - - telegraf_mem - telegraf_procstat/1917393364 - - telegraf_swap - - telegraf_processes - telegraf_disk + - telegraf_processes + - telegraf_mem + - telegraf_swap + - telegraf_netstat + - telegraf_cpu metrics/host_metrics: exporters: - forward/opentelemetry @@ -1617,8 +1636,8 @@ service: - ec2tagger - awsentity/service/telegraf receivers: - - telegraf_statsd - telegraf_socket_listener + - telegraf_statsd metrics/hostDeltaMetrics: exporters: - awscloudwatch @@ -1627,13 +1646,14 @@ service: - ec2tagger - awsentity/resource receivers: - - telegraf_diskio - telegraf_net + - telegraf_diskio metrics/opentelemetry: exporters: - otlphttp/metrics processors: - resourcedetection/opentelemetry + - transform/set_cluster_name - transform/identity - batch/opentelemetry_metrics receivers: @@ -1658,6 +1678,7 @@ service: - otlphttp/traces processors: - resourcedetection/opentelemetry + - transform/set_cluster_name - transform/identity - batch/opentelemetry_traces receivers: diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.json b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.json index 0dd61edc922..86ab573d2e5 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.json +++ b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.json @@ -125,12 +125,12 @@ "endpoint_override": "https://x-ray-endpoint.us-west-2.amazonaws.com" }, "opentelemetry": { + "cluster_name": "TestCluster", "collect": { "host_metrics": { - "metrics_collection_interval": 60 + "collection_interval": 60 }, "container_insights": { - "cluster_name": "TestCluster", "collection_interval": 30 }, "database_insights": { @@ -147,8 +147,7 @@ ] }, "prometheus": { - "config_path": "./sampleConfig/opentelemetry/prometheus_otel_scrape_config.yaml", - "cluster_name": "TestCluster" + "config_path": "./sampleConfig/opentelemetry/prometheus_otel_scrape_config.yaml" }, "otlp": { "grpc_endpoint": "127.0.0.1:4327", diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.yaml b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.yaml index 06c2155f93d..62bc1bd4af7 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.yaml +++ b/translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_eks_config.yaml @@ -1561,13 +1561,21 @@ processors: transform/set_cluster_name: error_mode: ignore flatten_data: false - log_statements: [] + log_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") metric_statements: - context: resource error_mode: ignore statements: - set(resource.attributes["k8s.cluster.name"], "TestCluster") - trace_statements: [] + trace_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") receivers: awsefareceiver/cw_k8s_ci_v0: collection_interval: 30s @@ -2260,6 +2268,7 @@ service: processors: - resourcedetection/opentelemetry - k8sattributes/opentelemetry + - transform/set_cluster_name - transform/identity - transform/logs_routing - attributestocontext/opentelemetry @@ -2462,13 +2471,13 @@ service: - ec2tagger - awsentity/resource receivers: - - telegraf_cpu - - telegraf_processes + - telegraf_netstat - telegraf_swap - telegraf_procstat/1917393364 - - telegraf_mem - - telegraf_netstat - telegraf_disk + - telegraf_mem + - telegraf_processes + - telegraf_cpu metrics/host_metrics: exporters: - forward/opentelemetry @@ -2483,8 +2492,8 @@ service: - ec2tagger - awsentity/service/telegraf receivers: - - telegraf_statsd - telegraf_socket_listener + - telegraf_statsd metrics/hostDeltaMetrics: exporters: - awscloudwatch @@ -2493,14 +2502,15 @@ service: - ec2tagger - awsentity/resource receivers: - - telegraf_net - telegraf_diskio + - telegraf_net metrics/opentelemetry: exporters: - otlphttp/metrics processors: - resourcedetection/opentelemetry - k8sattributes/opentelemetry + - transform/set_cluster_name - transform/identity - batch/opentelemetry_metrics receivers: @@ -2510,7 +2520,6 @@ service: - forward/opentelemetry processors: - transform/prometheus_scope - - transform/set_cluster_name receivers: - prometheus/opentelemetry metrics/otlp: @@ -2527,6 +2536,7 @@ service: processors: - resourcedetection/opentelemetry - k8sattributes/opentelemetry + - transform/set_cluster_name - transform/identity - batch/opentelemetry_traces receivers: diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_config.json b/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_config.json index 97b4f4f11b2..82b3b7ec01a 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_config.json +++ b/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_config.json @@ -3,9 +3,9 @@ "region": "us-east-1" }, "opentelemetry": { + "cluster_name": "TestCluster", "collect": { "container_insights": { - "cluster_name": "TestCluster", "collection_interval": 30 } } diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_config.yaml b/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_config.yaml index d73f49e26de..b8cf369ca70 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_config.yaml +++ b/translator/tocwconfig/sampleConfig/opentelemetry/container_insights_config.yaml @@ -891,6 +891,24 @@ processors: - delete_key(resource.attributes, "app.kubernetes.io/instance") - delete_key(resource.attributes, "app.kubernetes.io/name") - delete_key(resource.attributes, "app.kubernetes.io/version") + transform/set_cluster_name: + error_mode: ignore + flatten_data: false + log_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") + metric_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") + trace_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") receivers: awsefareceiver/cw_k8s_ci_v0: collection_interval: 30s @@ -1228,6 +1246,7 @@ service: processors: - resourcedetection/opentelemetry - k8sattributes/opentelemetry + - transform/set_cluster_name - transform/identity - batch/opentelemetry_metrics receivers: diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_aks_config.json b/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_aks_config.json index b0f6f9f64d6..5a64ceecf21 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_aks_config.json +++ b/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_aks_config.json @@ -5,7 +5,7 @@ "opentelemetry": { "collect": { "host_metrics": { - "metrics_collection_interval": 10 + "collection_interval": 10 } } } diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_azurevm_config.json b/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_azurevm_config.json index 8363781c290..efba5f6a923 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_azurevm_config.json +++ b/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_azurevm_config.json @@ -8,7 +8,7 @@ "opentelemetry": { "collect": { "host_metrics": { - "metrics_collection_interval": 10 + "collection_interval": 10 } } } diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_azurevm_sharedcreds_config.json b/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_azurevm_sharedcreds_config.json index 8363781c290..efba5f6a923 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_azurevm_sharedcreds_config.json +++ b/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_azurevm_sharedcreds_config.json @@ -8,7 +8,7 @@ "opentelemetry": { "collect": { "host_metrics": { - "metrics_collection_interval": 10 + "collection_interval": 10 } } } diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_config.json b/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_config.json index b0f6f9f64d6..5a64ceecf21 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_config.json +++ b/translator/tocwconfig/sampleConfig/opentelemetry/host_metrics_config.json @@ -5,7 +5,7 @@ "opentelemetry": { "collect": { "host_metrics": { - "metrics_collection_interval": 10 + "collection_interval": 10 } } } diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_config.json b/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_config.json index ae4cf31acef..5d937d90764 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_config.json +++ b/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_config.json @@ -3,6 +3,7 @@ "region": "us-east-1" }, "opentelemetry": { + "cluster_name": "TestCluster", "collect": { "prometheus": { "config_path": "./sampleConfig/opentelemetry/prometheus_otel_scrape_config.yaml" diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_config.yaml b/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_config.yaml index 859ad8b950e..40d0b8b505d 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_config.yaml +++ b/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_config.yaml @@ -370,6 +370,24 @@ processors: - set(scope.attributes["cloudwatch.source"], "cloudwatch-agent") - set(scope.attributes["cloudwatch.solution"], "otel-prometheus") trace_statements: [] + transform/set_cluster_name: + error_mode: ignore + flatten_data: false + log_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") + metric_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") + trace_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") receivers: prometheus/opentelemetry: config: @@ -431,6 +449,7 @@ service: - otlphttp/metrics processors: - resourcedetection/opentelemetry + - transform/set_cluster_name - transform/identity - batch/opentelemetry_metrics receivers: diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_eks_config.json b/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_eks_config.json index ae4cf31acef..5d937d90764 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_eks_config.json +++ b/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_eks_config.json @@ -3,6 +3,7 @@ "region": "us-east-1" }, "opentelemetry": { + "cluster_name": "TestCluster", "collect": { "prometheus": { "config_path": "./sampleConfig/opentelemetry/prometheus_otel_scrape_config.yaml" diff --git a/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_eks_config.yaml b/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_eks_config.yaml index dd18496ceb3..389f4de6608 100644 --- a/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_eks_config.yaml +++ b/translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_eks_config.yaml @@ -535,6 +535,24 @@ processors: - set(scope.attributes["cloudwatch.source"], "cloudwatch-agent") - set(scope.attributes["cloudwatch.solution"], "otel-prometheus") trace_statements: [] + transform/set_cluster_name: + error_mode: ignore + flatten_data: false + log_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") + metric_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") + trace_statements: + - context: resource + error_mode: ignore + statements: + - set(resource.attributes["k8s.cluster.name"], "TestCluster") receivers: prometheus/opentelemetry: config: @@ -598,6 +616,7 @@ service: processors: - resourcedetection/opentelemetry - k8sattributes/opentelemetry + - transform/set_cluster_name - transform/identity - batch/opentelemetry_metrics receivers: diff --git a/translator/tocwconfig/tocwconfig_unix_test.go b/translator/tocwconfig/tocwconfig_unix_test.go index 4037321582b..d52a89c7c57 100644 --- a/translator/tocwconfig/tocwconfig_unix_test.go +++ b/translator/tocwconfig/tocwconfig_unix_test.go @@ -180,7 +180,6 @@ func TestCombinedV1V2EKSConfig(t *testing.T) { yamlConfig, err := mapstructure.Marshal(cfg) require.NoError(t, err) yamlStr := toyamlconfig.ToYamlConfig(yamlConfig) - // assert.NoError(t, os.WriteFile("./sampleConfig/opentelemetry/combined_v1_v2_eks_config.yaml", []byte(yamlStr), 0644)) // useful for regenerating YAML require.NoError(t, yaml.Unmarshal([]byte(yamlStr), &actual)) opt := cmpopts.SortSlices(func(x, y interface{}) bool { diff --git a/translator/translate/otel/common/common.go b/translator/translate/otel/common/common.go index 54d23413ad9..a164351b24f 100644 --- a/translator/translate/otel/common/common.go +++ b/translator/translate/otel/common/common.go @@ -8,6 +8,7 @@ import ( "fmt" "os" "reflect" + "regexp" "strconv" "strings" "time" @@ -58,6 +59,8 @@ const ( RoleARNKey = "role_arn" SigV4Auth = "sigv4auth" MetricsCollectionIntervalKey = "metrics_collection_interval" + CollectionIntervalKey = "collection_interval" + ClusterNameKey = "cluster_name" AggregationDimensionsKey = "aggregation_dimensions" MeasurementKey = "measurement" DropOriginalMetricsKey = "drop_original_metrics" @@ -554,9 +557,9 @@ func SanitizeName(input string) string { }, strings.ToLower(input)) } -func GetClusterName(conf *confmap.Conf) string { - val, ok := GetString(conf, ConfigKey(LogsKey, MetricsCollectedKey, KubernetesKey, "cluster_name")) - if ok && val != "" { +func GetClusterName(conf *confmap.Conf, key string) string { + // Check any config keys passed + if val, ok := GetString(conf, key); ok && val != "" { return val } @@ -593,3 +596,20 @@ func EscapeDollarDigit(s string) string { } return string(out) } + +// OtelClusterNameKey is the config key for the root-level cluster name under opentelemetry. +var OtelClusterNameKey = ConfigKey(OpenTelemetryKey, ClusterNameKey) + +// LegacyClusterNameKey is the config key for the cluster name in the V1 config path. +var LegacyClusterNameKey = ConfigKey(LogsKey, MetricsCollectedKey, KubernetesKey, ClusterNameKey) + +// ClusterNameRegex validates cluster names. +var ClusterNameRegex = regexp.MustCompile(`^[0-9A-Za-z][A-Za-z0-9\-_]*$`) + +// ValidateClusterName returns an error if the cluster name does not match the expected pattern. +func ValidateClusterName(name string) error { + if !ClusterNameRegex.MatchString(name) { + return fmt.Errorf("cluster_name %q is invalid: must match pattern %s", name, ClusterNameRegex.String()) + } + return nil +} diff --git a/translator/translate/otel/exporter/awsemf/prometheus.go b/translator/translate/otel/exporter/awsemf/prometheus.go index 20a4a324e40..86c967b39c4 100644 --- a/translator/translate/otel/exporter/awsemf/prometheus.go +++ b/translator/translate/otel/exporter/awsemf/prometheus.go @@ -39,7 +39,7 @@ func setPrometheusLogGroup(conf *confmap.Conf, cfg *awsemfexporter.Config) error } } else { - if clusterName := common.GetClusterName(conf); clusterName != "" { + if clusterName := common.GetClusterName(conf, common.LegacyClusterNameKey); clusterName != "" { cfg.LogGroupName = fmt.Sprintf(eksDefaultLogGroupFormat, clusterName) } } diff --git a/translator/translate/otel/pipeline/opentelemetry/containerinsights/common.go b/translator/translate/otel/pipeline/opentelemetry/containerinsights/common.go index 2cfa4d31ffe..2cf68cac8b0 100644 --- a/translator/translate/otel/pipeline/opentelemetry/containerinsights/common.go +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/common.go @@ -6,7 +6,6 @@ package containerinsights import ( "fmt" "os" - "regexp" "strings" "time" @@ -20,9 +19,8 @@ import ( const ( ciPrefix = "cw_k8s_ci_v0" defaultCollectionInterval = 30 * time.Second - - modeNode = "node" - modeCluster = "cluster" + roleNode = "node" + roleCluster = "cluster" ) var ciConfigKey = common.ConfigKey(common.OpenTelemetryKey, common.CollectKey, common.OtelContainerInsightsKey) @@ -76,28 +74,21 @@ func (t *yamlComponentTranslator) Translate(_ *confmap.Conf) (component.Config, return t.cfg, nil } -// clusterNameRegex restricts cluster_name to safe characters, preventing -// OTTL injection and template metacharacter issues in YAML templates. -var clusterNameRegex = regexp.MustCompile(`^[a-zA-Z0-9._-]+$`) - func getClusterName(conf *confmap.Conf) (string, error) { - key := common.ConfigKey(ciConfigKey, "cluster_name") - name, ok := common.GetString(conf, key) - if !ok || name == "" { - return "", fmt.Errorf("cluster_name is required for container_insights") + name := common.GetClusterName(conf, common.OtelClusterNameKey) + if name == "" { + return "", fmt.Errorf("cluster_name is required for container_insights: set opentelemetry::cluster_name in config or K8S_CLUSTER_NAME environment variable") } - if !clusterNameRegex.MatchString(name) { - return "", fmt.Errorf("cluster_name contains invalid characters: %q (must match %s)", name, clusterNameRegex.String()) + if err := common.ValidateClusterName(name); err != nil { + return "", err } return name, nil } func getCollectionInterval(conf *confmap.Conf) time.Duration { - key := common.ConfigKey(ciConfigKey, "collection_interval") - if v, ok := common.GetNumber(conf, key); ok && v > 0 { - return time.Duration(v) * time.Second - } - return defaultCollectionInterval + return common.GetOrDefaultDuration(conf, []string{ + common.ConfigKey(ciConfigKey, common.CollectionIntervalKey), + }, defaultCollectionInterval) } // logsEnabled returns true if container_insights.logs.enabled is set to true. @@ -109,14 +100,14 @@ func logsEnabled(conf *confmap.Conf) bool { return common.GetOrDefaultBool(conf, key, false) } -// getMode resolves the container insights pipeline mode using the following +// getRole resolves the container insights pipeline role using the following // priority order: // 1. JSON config field // 2. Environment variable // 3. Default: "node" (DaemonSet) -func getMode(conf *confmap.Conf) string { +func getRole(conf *confmap.Conf) string { if conf != nil { - key := common.ConfigKey(ciConfigKey, "mode") + key := common.ConfigKey(ciConfigKey, "role") if v, ok := common.GetString(conf, key); ok && v != "" { return v } @@ -124,12 +115,12 @@ func getMode(conf *confmap.Conf) string { if role := strings.ToUpper(os.Getenv(envconfig.CWAGENT_ROLE)); role != "" { switch role { case envconfig.NODE: - return modeNode + return roleNode case envconfig.LEADER: - return modeCluster + return roleCluster } } - return modeNode + return roleNode } type pipelineSpec struct { diff --git a/translator/translate/otel/pipeline/opentelemetry/containerinsights/common_test.go b/translator/translate/otel/pipeline/opentelemetry/containerinsights/common_test.go index 4485a6622bc..dacc4164438 100644 --- a/translator/translate/otel/pipeline/opentelemetry/containerinsights/common_test.go +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/common_test.go @@ -12,20 +12,20 @@ import ( "github.com/aws/amazon-cloudwatch-agent/cfg/envconfig" ) -func TestGetMode_JSONConfig(t *testing.T) { +func TestGetRole_JSONConfig(t *testing.T) { cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ "collect": map[string]interface{}{ "container_insights": map[string]interface{}{ - "mode": "cluster", + "role": "cluster", }, }, }, }) - assert.Equal(t, modeCluster, getMode(cfg)) + assert.Equal(t, roleCluster, getRole(cfg)) } -func TestGetMode_EnvVarFallback(t *testing.T) { +func TestGetRole_EnvVarFallback(t *testing.T) { cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ "collect": map[string]interface{}{ @@ -35,13 +35,13 @@ func TestGetMode_EnvVarFallback(t *testing.T) { }) t.Setenv(envconfig.CWAGENT_ROLE, envconfig.LEADER) - assert.Equal(t, modeCluster, getMode(cfg)) + assert.Equal(t, roleCluster, getRole(cfg)) t.Setenv(envconfig.CWAGENT_ROLE, envconfig.NODE) - assert.Equal(t, modeNode, getMode(cfg)) + assert.Equal(t, roleNode, getRole(cfg)) } -func TestGetMode_DefaultsToNode(t *testing.T) { +func TestGetRole_DefaultsToNode(t *testing.T) { cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ "collect": map[string]interface{}{ @@ -49,10 +49,10 @@ func TestGetMode_DefaultsToNode(t *testing.T) { }, }, }) - assert.Equal(t, modeNode, getMode(cfg)) + assert.Equal(t, roleNode, getRole(cfg)) } -func TestGetMode_EnvVarCaseInsensitive(t *testing.T) { +func TestGetRole_EnvVarCaseInsensitive(t *testing.T) { cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ "collect": map[string]interface{}{ @@ -62,27 +62,27 @@ func TestGetMode_EnvVarCaseInsensitive(t *testing.T) { }) t.Setenv(envconfig.CWAGENT_ROLE, "leader") // lowercase - assert.Equal(t, modeCluster, getMode(cfg)) + assert.Equal(t, roleCluster, getRole(cfg)) t.Setenv(envconfig.CWAGENT_ROLE, "node") // lowercase - assert.Equal(t, modeNode, getMode(cfg)) + assert.Equal(t, roleNode, getRole(cfg)) t.Setenv(envconfig.CWAGENT_ROLE, "Leader") // mixed case - assert.Equal(t, modeCluster, getMode(cfg)) + assert.Equal(t, roleCluster, getRole(cfg)) } -func TestGetMode_JSONOverridesEnv(t *testing.T) { +func TestGetRole_JSONOverridesEnv(t *testing.T) { t.Setenv(envconfig.CWAGENT_ROLE, envconfig.NODE) cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ "collect": map[string]interface{}{ "container_insights": map[string]interface{}{ - "mode": "cluster", + "role": "cluster", }, }, }, }) - assert.Equal(t, modeCluster, getMode(cfg)) + assert.Equal(t, roleCluster, getRole(cfg)) } func TestLogsEnabled(t *testing.T) { diff --git a/translator/translate/otel/pipeline/opentelemetry/containerinsights/translator.go b/translator/translate/otel/pipeline/opentelemetry/containerinsights/translator.go index 76e614692d2..bf463ccdd82 100644 --- a/translator/translate/otel/pipeline/opentelemetry/containerinsights/translator.go +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/translator.go @@ -74,7 +74,7 @@ var apiserverYAML string var kubeStateMetricsYAML string // NewTranslators returns all container insights pipeline translators. -// The pipelines generated depend on the resolved mode (see getMode for priority): +// The pipelines generated depend on the resolved role (see getRole for priority): // - "node": daemonset pipelines (per-node metrics + logs) // - "cluster": deployment pipelines (cluster-wide metrics) func NewTranslators(conf *confmap.Conf) common.PipelineTranslatorMap { @@ -85,10 +85,10 @@ func NewTranslators(conf *confmap.Conf) common.PipelineTranslatorMap { return translators } - mode := getMode(conf) + role := getRole(conf) // Daemonset metrics pipelines - if mode == modeNode { + if role == roleNode { translators.Set(newYAMLPipeline("kubeletstats", pipeline.SignalMetrics, kubeletstatsYAML)) translators.Set(newYAMLPipeline("cadvisor", pipeline.SignalMetrics, cadvisorYAML)) translators.Set(newYAMLPipeline("node_exporter", pipeline.SignalMetrics, nodeExporterYAML)) @@ -106,7 +106,7 @@ func NewTranslators(conf *confmap.Conf) common.PipelineTranslatorMap { } // Deployment metrics pipelines - if mode == modeCluster { + if role == roleCluster { translators.Set(newYAMLPipeline("apiserver", pipeline.SignalMetrics, apiserverYAML)) translators.Set(newYAMLPipeline("kube_state_metrics", pipeline.SignalMetrics, kubeStateMetricsYAML)) } diff --git a/translator/translate/otel/pipeline/opentelemetry/containerinsights/translators_test.go b/translator/translate/otel/pipeline/opentelemetry/containerinsights/translators_test.go index 526356860ff..122be82b78a 100644 --- a/translator/translate/otel/pipeline/opentelemetry/containerinsights/translators_test.go +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/translators_test.go @@ -19,29 +19,29 @@ func TestNewTranslators_MissingKey(t *testing.T) { assert.Equal(t, 0, NewTranslators(confmap.NewFromStringMap(map[string]interface{}{})).Len()) } -func TestNewTranslators_ModeNode(t *testing.T) { +func TestNewTranslators_RoleNode(t *testing.T) { cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ "container_insights": map[string]interface{}{ - "cluster_name": "test-cluster", - "mode": "node", + "role": "node", }, }, }, }) translators := NewTranslators(cfg) - // node mode: kubeletstats, cadvisor, node_exporter, dcgm, neuron, efa, ebs_csi, lis_csi = 8 pipelines + // node role: kubeletstats, cadvisor, node_exporter, dcgm, neuron, efa, ebs_csi, lis_csi = 8 pipelines assert.Equal(t, 8, translators.Len()) } -func TestNewTranslators_ModeNodeWithLogs(t *testing.T) { +func TestNewTranslators_RoleNodeWithLogs(t *testing.T) { cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ "container_insights": map[string]interface{}{ - "cluster_name": "test-cluster", - "mode": "node", + "role": "node", "logs": map[string]interface{}{ "enabled": true, }, @@ -50,73 +50,70 @@ func TestNewTranslators_ModeNodeWithLogs(t *testing.T) { }, }) translators := NewTranslators(cfg) - // node mode + logs: 8 metric pipelines + 2 log pipelines = 10 + // node role + logs: 8 metric pipelines + 2 log pipelines = 10 assert.Equal(t, 10, translators.Len()) } -func TestNewTranslators_ModeCluster(t *testing.T) { +func TestNewTranslators_RoleCluster(t *testing.T) { cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ "container_insights": map[string]interface{}{ - "cluster_name": "test-cluster", - "mode": "cluster", + "role": "cluster", }, }, }, }) translators := NewTranslators(cfg) - // cluster mode: apiserver, kube_state_metrics = 2 pipelines + // cluster role: apiserver, kube_state_metrics = 2 pipelines assert.Equal(t, 2, translators.Len()) } -func TestNewTranslators_DefaultMode(t *testing.T) { - // No mode specified, no env var - should default to node +func TestNewTranslators_DefaultRole(t *testing.T) { + // No role specified, no env var - should default to node cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ - "container_insights": map[string]interface{}{ - "cluster_name": "test-cluster", - }, + "container_insights": map[string]interface{}{}, }, }, }) translators := NewTranslators(cfg) - // defaults to node mode: 8 pipelines + // defaults to node role: 8 pipelines assert.Equal(t, 8, translators.Len()) } func TestNewTranslators_EnvVarFallback_Node(t *testing.T) { - // No mode in config, CWAGENT_ROLE=NODE + // No role in config, CWAGENT_ROLE=NODE t.Setenv(envconfig.CWAGENT_ROLE, envconfig.NODE) cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ - "container_insights": map[string]interface{}{ - "cluster_name": "test-cluster", - }, + "container_insights": map[string]interface{}{}, }, }, }) translators := NewTranslators(cfg) - // env var NODE -> node mode: 8 pipelines + // env var NODE -> node role: 8 pipelines assert.Equal(t, 8, translators.Len()) } func TestNewTranslators_EnvVarFallback_Leader(t *testing.T) { - // No mode in config, CWAGENT_ROLE=LEADER + // No role in config, CWAGENT_ROLE=LEADER t.Setenv(envconfig.CWAGENT_ROLE, envconfig.LEADER) cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ - "container_insights": map[string]interface{}{ - "cluster_name": "test-cluster", - }, + "container_insights": map[string]interface{}{}, }, }, }) translators := NewTranslators(cfg) - // env var LEADER -> cluster mode: 2 pipelines + // env var LEADER -> cluster role: 2 pipelines assert.Equal(t, 2, translators.Len()) } @@ -125,15 +122,15 @@ func TestNewTranslators_JSONConfigOverridesEnvVar(t *testing.T) { t.Setenv(envconfig.CWAGENT_ROLE, envconfig.NODE) cfg := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ "container_insights": map[string]interface{}{ - "cluster_name": "test-cluster", - "mode": "cluster", + "role": "cluster", }, }, }, }) translators := NewTranslators(cfg) - // JSON config wins: cluster mode = 2 pipelines + // JSON config wins: cluster role = 2 pipelines assert.Equal(t, 2, translators.Len()) } diff --git a/translator/translate/otel/pipeline/opentelemetry/prometheus/translator.go b/translator/translate/otel/pipeline/opentelemetry/prometheus/translator.go index 60bf17756cc..9dbea64bed7 100644 --- a/translator/translate/otel/pipeline/opentelemetry/prometheus/translator.go +++ b/translator/translate/otel/pipeline/opentelemetry/prometheus/translator.go @@ -7,7 +7,6 @@ import ( "fmt" "os" "path/filepath" - "regexp" "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver" "go.opentelemetry.io/collector/component" @@ -24,11 +23,8 @@ const ( pipelineName = "otel_prometheus" ) -var clusterNameRegex = regexp.MustCompile(`^[a-zA-Z0-9._-]+$`) - var prometheusKey = common.ConfigKey(common.OpenTelemetryKey, common.CollectKey, common.PrometheusKey) var configPathKey = common.ConfigKey(prometheusKey, "config_path") -var clusterNameKey = common.ConfigKey(prometheusKey, "cluster_name") type translator struct{} @@ -55,17 +51,6 @@ func (t *translator) Translate(conf *confmap.Conf) (*common.ComponentTranslators transformprocessor.WithErrorMode("ignore"), transformprocessor.WithMetricScopeStatements(common.ScopeStatementsForSolution("otel-prometheus")), )) - if clusterName, ok := common.GetString(conf, clusterNameKey); ok && clusterName != "" { - if !clusterNameRegex.MatchString(clusterName) { - return nil, fmt.Errorf("cluster_name contains invalid characters: %q", clusterName) - } - processors.Set(transformprocessor.NewTranslatorWithName("set_cluster_name", - transformprocessor.WithMetricResourceStatements([]string{ - fmt.Sprintf(`set(resource.attributes["k8s.cluster.name"], "%s")`, clusterName), - }), - )) - } - return &common.ComponentTranslators{ Receivers: common.NewTranslatorMap[component.Config, component.ID](receiver), Processors: processors, diff --git a/translator/translate/otel/pipeline/opentelemetry/prometheus/translator_test.go b/translator/translate/otel/pipeline/opentelemetry/prometheus/translator_test.go index e9fcc842351..8a1de7e3a77 100644 --- a/translator/translate/otel/pipeline/opentelemetry/prometheus/translator_test.go +++ b/translator/translate/otel/pipeline/opentelemetry/prometheus/translator_test.go @@ -36,6 +36,7 @@ func TestPrometheusTranslator(t *testing.T) { "WithValidConfig": { input: map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ "prometheus": map[string]interface{}{ "config_path": createTempPromConfig(t), @@ -48,32 +49,20 @@ func TestPrometheusTranslator(t *testing.T) { "WithClusterName": { input: map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "my-cluster", "collect": map[string]interface{}{ "prometheus": map[string]interface{}{ - "config_path": createTempPromConfig(t), - "cluster_name": "my-cluster", + "config_path": createTempPromConfig(t), }, }, }, }, wantErr: false, }, - "WithInvalidClusterName": { - input: map[string]interface{}{ - "opentelemetry": map[string]interface{}{ - "collect": map[string]interface{}{ - "prometheus": map[string]interface{}{ - "config_path": createTempPromConfig(t), - "cluster_name": `bad"name`, - }, - }, - }, - }, - wantErr: true, - }, "WithMissingConfigFile": { input: map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ "prometheus": map[string]interface{}{ "config_path": "/nonexistent/path.yml", @@ -110,10 +99,10 @@ func TestPrometheusTranslator(t *testing.T) { func TestPrometheusTranslatorClusterNameProcessor(t *testing.T) { conf := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ "prometheus": map[string]interface{}{ - "config_path": createTempPromConfig(t), - "cluster_name": "test-cluster", + "config_path": createTempPromConfig(t), }, }, }, @@ -122,9 +111,8 @@ func TestPrometheusTranslatorClusterNameProcessor(t *testing.T) { tt := NewTranslator() got, err := tt.Translate(conf) require.NoError(t, err) - assert.Equal(t, 2, got.Processors.Len()) + assert.Equal(t, 1, got.Processors.Len()) assert.Equal(t, "transform/prometheus_scope", got.Processors.Keys()[0].String()) - assert.Equal(t, "transform/set_cluster_name", got.Processors.Keys()[1].String()) } func TestPrometheusTranslatorNoClusterNameProcessor(t *testing.T) { @@ -276,10 +264,10 @@ func TestPrometheusTranslatorK8sMode(t *testing.T) { conf := confmap.NewFromStringMap(map[string]interface{}{ "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", "collect": map[string]interface{}{ "prometheus": map[string]interface{}{ - "config_path": createTempPromConfig(t), - "cluster_name": "test-cluster", + "config_path": createTempPromConfig(t), }, }, }, @@ -288,11 +276,10 @@ func TestPrometheusTranslatorK8sMode(t *testing.T) { tt := NewTranslator() got, err := tt.Translate(conf) require.NoError(t, err) - assert.Equal(t, 2, got.Processors.Len()) // scope + set_cluster_name + assert.Equal(t, 1, got.Processors.Len()) // scope only, cluster_name in export pipeline keys := make([]string, 0, got.Processors.Len()) for _, k := range got.Processors.Keys() { keys = append(keys, k.String()) } assert.Contains(t, keys, "transform/prometheus_scope") - assert.Contains(t, keys, "transform/set_cluster_name") } diff --git a/translator/translate/otel/pipeline/opentelemetry/translator_logs.go b/translator/translate/otel/pipeline/opentelemetry/translator_logs.go index ddb56eda433..6ec52312a58 100644 --- a/translator/translate/otel/pipeline/opentelemetry/translator_logs.go +++ b/translator/translate/otel/pipeline/opentelemetry/translator_logs.go @@ -112,6 +112,19 @@ func (t *baseLogsTranslator) Translate(conf *confmap.Conf) (*common.ComponentTra if context.CurrentContext().KubernetesMode() != "" { processors.Set(k8sattributesprocessor.NewTranslator(common.OpenTelemetryKey)) } + // Apply root-level cluster name if set + clusterName := common.GetClusterName(conf, common.OtelClusterNameKey) + if clusterName != "" { + if err := common.ValidateClusterName(clusterName); err != nil { + return nil, err + } + stmt := fmt.Sprintf(`set(resource.attributes["k8s.cluster.name"], "%s")`, clusterName) + processors.Set(transformprocessor.NewTranslatorWithName("set_cluster_name", + transformprocessor.WithMetricResourceStatements([]string{stmt}), + transformprocessor.WithLogResourceStatements([]string{stmt}), + transformprocessor.WithTraceResourceStatements([]string{stmt}), + )) + } processors.Set(transformprocessor.NewTranslatorWithName(common.Identity)) processors.Set(logsRouting) processors.Set(attrCtx) diff --git a/translator/translate/otel/pipeline/opentelemetry/translator_metrics.go b/translator/translate/otel/pipeline/opentelemetry/translator_metrics.go index d4a0c002ac1..f0969a24c4e 100644 --- a/translator/translate/otel/pipeline/opentelemetry/translator_metrics.go +++ b/translator/translate/otel/pipeline/opentelemetry/translator_metrics.go @@ -63,6 +63,19 @@ func (t *baseMetricsTranslator) Translate(conf *confmap.Conf) (*common.Component if context.CurrentContext().KubernetesMode() != "" { processors.Set(k8sattributesprocessor.NewTranslator(common.OpenTelemetryKey)) } + // Apply root-level cluster name if set + clusterName := common.GetClusterName(conf, common.OtelClusterNameKey) + if clusterName != "" { + if err := common.ValidateClusterName(clusterName); err != nil { + return nil, err + } + stmt := fmt.Sprintf(`set(resource.attributes["k8s.cluster.name"], "%s")`, clusterName) + processors.Set(transformprocessor.NewTranslatorWithName("set_cluster_name", + transformprocessor.WithMetricResourceStatements([]string{stmt}), + transformprocessor.WithLogResourceStatements([]string{stmt}), + transformprocessor.WithTraceResourceStatements([]string{stmt}), + )) + } processors.Set(transformprocessor.NewTranslatorWithName(common.Identity)) processors.Set(batchprocessor.NewTranslator(common.WithName("opentelemetry_metrics"), batchprocessor.WithSendBatchSize(common.MaxMetricsPerRequest), batchprocessor.WithSendBatchMaxSize(common.MaxMetricsPerRequest), batchprocessor.WithTimeout(common.BatchTimeout))) diff --git a/translator/translate/otel/pipeline/opentelemetry/translator_metrics_test.go b/translator/translate/otel/pipeline/opentelemetry/translator_metrics_test.go index da57983a653..de154ab0c07 100644 --- a/translator/translate/otel/pipeline/opentelemetry/translator_metrics_test.go +++ b/translator/translate/otel/pipeline/opentelemetry/translator_metrics_test.go @@ -110,3 +110,50 @@ func TestBaseMetricsTranslatorEmptyRegion(t *testing.T) { assert.Nil(t, got) assert.Contains(t, err.Error(), "region is required") } + +func TestBaseMetricsTranslatorClusterName(t *testing.T) { + agent.Global_Config.Region = "us-east-1" + tt := NewBaseMetricsTranslator() + + conf := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "cluster_name": "test-cluster", + "collect": map[string]interface{}{ + "host_metrics": map[string]interface{}{}, + }, + }, + }) + + got, err := tt.Translate(conf) + require.NoError(t, err) + + // Verify set_cluster_name processor is present + keys := make([]string, 0, got.Processors.Len()) + for _, k := range got.Processors.Keys() { + keys = append(keys, k.String()) + } + assert.Contains(t, keys, "transform/set_cluster_name") +} + +func TestBaseMetricsTranslatorNoClusterName(t *testing.T) { + agent.Global_Config.Region = "us-east-1" + tt := NewBaseMetricsTranslator() + + conf := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "collect": map[string]interface{}{ + "host_metrics": map[string]interface{}{}, + }, + }, + }) + + got, err := tt.Translate(conf) + require.NoError(t, err) + + // Verify set_cluster_name processor is NOT present + keys := make([]string, 0, got.Processors.Len()) + for _, k := range got.Processors.Keys() { + keys = append(keys, k.String()) + } + assert.NotContains(t, keys, "transform/set_cluster_name") +} diff --git a/translator/translate/otel/pipeline/opentelemetry/translator_traces.go b/translator/translate/otel/pipeline/opentelemetry/translator_traces.go index bcb442c15d2..ef9f28ff916 100644 --- a/translator/translate/otel/pipeline/opentelemetry/translator_traces.go +++ b/translator/translate/otel/pipeline/opentelemetry/translator_traces.go @@ -58,6 +58,19 @@ func (t *baseTracesTranslator) Translate(conf *confmap.Conf) (*common.ComponentT if context.CurrentContext().KubernetesMode() != "" { processors.Set(k8sattributesprocessor.NewTranslator(common.OpenTelemetryKey)) } + // Apply root-level cluster name if set + clusterName := common.GetClusterName(conf, common.OtelClusterNameKey) + if clusterName != "" { + if err := common.ValidateClusterName(clusterName); err != nil { + return nil, err + } + stmt := fmt.Sprintf(`set(resource.attributes["k8s.cluster.name"], "%s")`, clusterName) + processors.Set(transformprocessor.NewTranslatorWithName("set_cluster_name", + transformprocessor.WithMetricResourceStatements([]string{stmt}), + transformprocessor.WithLogResourceStatements([]string{stmt}), + transformprocessor.WithTraceResourceStatements([]string{stmt}), + )) + } processors.Set(transformprocessor.NewTranslatorWithName(common.Identity)) processors.Set(batchprocessor.NewTranslator(common.WithName("opentelemetry_traces"), batchprocessor.WithSendBatchSize(common.MaxSpansPerRequest), batchprocessor.WithSendBatchMaxSize(common.MaxSpansPerRequest), batchprocessor.WithTimeout(common.BatchTimeout))) diff --git a/translator/translate/otel/processor/awsapplicationsignals/translator.go b/translator/translate/otel/processor/awsapplicationsignals/translator.go index ca0143c6d85..2dad5d5785b 100644 --- a/translator/translate/otel/processor/awsapplicationsignals/translator.go +++ b/translator/translate/otel/processor/awsapplicationsignals/translator.go @@ -67,7 +67,7 @@ func (t *translator) Translate(conf *confmap.Conf) (component.Config, error) { hostedIn, hostedInConfigured := common.GetHostedIn(conf) if common.IsAppSignalsKubernetes() { if !hostedInConfigured { - hostedIn = common.GetClusterName(conf) + hostedIn = common.GetClusterName(conf, common.LegacyClusterNameKey) } } diff --git a/translator/translate/otel/processor/awsentity/translator.go b/translator/translate/otel/processor/awsentity/translator.go index db3825de6fb..cdbb0361c95 100644 --- a/translator/translate/otel/processor/awsentity/translator.go +++ b/translator/translate/otel/processor/awsentity/translator.go @@ -103,7 +103,7 @@ func (t *translator) Translate(conf *confmap.Conf) (component.Config, error) { clusterName, clusterNameConfigured := common.GetHostedIn(conf) if !clusterNameConfigured { - clusterName = common.GetClusterName(conf) + clusterName = common.GetClusterName(conf, common.LegacyClusterNameKey) } cfg.ClusterName = clusterName diff --git a/translator/translate/otel/processor/resourceprocessor/translator.go b/translator/translate/otel/processor/resourceprocessor/translator.go index c526245aa63..ff3ef551365 100644 --- a/translator/translate/otel/processor/resourceprocessor/translator.go +++ b/translator/translate/otel/processor/resourceprocessor/translator.go @@ -137,7 +137,7 @@ func (t *translator) getJMXAttributes(conf *confmap.Conf) []any { } func (t *translator) getContainerInsightsJMXAttributes(conf *confmap.Conf) []any { - clusterName := common.GetClusterName(conf) + clusterName := common.GetClusterName(conf, common.LegacyClusterNameKey) nodeName := os.Getenv(config.HOST_NAME) return []any{ map[string]any{ diff --git a/translator/translate/otel/processor/transformprocessor/translator.go b/translator/translate/otel/processor/transformprocessor/translator.go index 1b9cac998a8..101ebef84c4 100644 --- a/translator/translate/otel/processor/transformprocessor/translator.go +++ b/translator/translate/otel/processor/transformprocessor/translator.go @@ -66,6 +66,13 @@ func WithMetricResourceStatements(statements []string) Option { } } +// WithTraceResourceStatements sets OTTL statements to execute in the "resource" context for traces. +func WithTraceResourceStatements(statements []string) Option { + return func(t *translator) { + t.traceStatements = statements + } +} + // WithErrorMode sets the error mode for dynamic statements. Defaults to "propagate". func WithErrorMode(mode string) Option { return func(t *translator) { @@ -107,6 +114,7 @@ type translator struct { logStatements []string logContextStatements []string metricStatements []string + traceStatements []string scopeStatements []string logScopeStatements []string metricScopeStatements []string @@ -131,6 +139,7 @@ func (t *translator) hasDynamicStatements() bool { return len(t.logStatements) > 0 || len(t.logContextStatements) > 0 || len(t.metricStatements) > 0 || + len(t.traceStatements) > 0 || len(t.scopeStatements) > 0 || len(t.logScopeStatements) > 0 || len(t.metricScopeStatements) > 0 @@ -153,6 +162,9 @@ func (t *translator) Translate(conf *confmap.Conf) (component.Config, error) { if len(t.logStatements) > 0 { cfgMap["log_statements"] = []any{buildResourceStatements(t.logStatements, errorMode)} } + if len(t.traceStatements) > 0 { + cfgMap["trace_statements"] = []any{buildResourceStatements(t.traceStatements, errorMode)} + } if len(t.scopeStatements) > 0 { scopeBlock := buildScopeStatements(t.scopeStatements, errorMode) cfgMap["metric_statements"] = appendStatements(cfgMap["metric_statements"], scopeBlock) diff --git a/translator/translate/otel/receiver/awscontainerinsight/translator.go b/translator/translate/otel/receiver/awscontainerinsight/translator.go index f83dc32fe4b..c6f49ddb50a 100644 --- a/translator/translate/otel/receiver/awscontainerinsight/translator.go +++ b/translator/translate/otel/receiver/awscontainerinsight/translator.go @@ -130,7 +130,7 @@ func (t *translator) Translate(conf *confmap.Conf) (component.Config, error) { } func (t *translator) setClusterName(conf *confmap.Conf, cfg *awscontainerinsightreceiver.Config) error { - cfg.ClusterName = common.GetClusterName(conf) + cfg.ClusterName = common.GetClusterName(conf, common.LegacyClusterNameKey) if cfg.ClusterName == "" { return errors.New("cluster name is not provided and was not auto-detected from EC2 tags") diff --git a/translator/translate/otel/receiver/awscontainerinsightskueue/translator.go b/translator/translate/otel/receiver/awscontainerinsightskueue/translator.go index e62600380fa..594d0675f2c 100644 --- a/translator/translate/otel/receiver/awscontainerinsightskueue/translator.go +++ b/translator/translate/otel/receiver/awscontainerinsightskueue/translator.go @@ -59,7 +59,7 @@ func (t *translator) Translate(conf *confmap.Conf) (component.Config, error) { } func (t *translator) setClusterName(conf *confmap.Conf, cfg *awscontainerinsightskueuereceiver.Config) error { - cfg.ClusterName = common.GetClusterName(conf) + cfg.ClusterName = common.GetClusterName(conf, common.LegacyClusterNameKey) if cfg.ClusterName == "" { return errors.New("cluster name is not provided and was not auto-detected from EC2 tags") diff --git a/translator/translate/otel/receiver/hostmetrics/translator.go b/translator/translate/otel/receiver/hostmetrics/translator.go index 31a2a4cdf1c..a24a8ffa3b2 100644 --- a/translator/translate/otel/receiver/hostmetrics/translator.go +++ b/translator/translate/otel/receiver/hostmetrics/translator.go @@ -94,7 +94,7 @@ func (t *translator) Translate(conf *confmap.Conf) (component.Config, error) { } intervalKeyChain := []string{ - common.ConfigKey(common.OpenTelemetryKey, common.CollectKey, common.HostMetricsKey, common.MetricsCollectionIntervalKey), + common.ConfigKey(common.OpenTelemetryKey, common.CollectKey, common.HostMetricsKey, common.CollectionIntervalKey), common.ConfigKey(common.AgentKey, common.MetricsCollectionIntervalKey), } return &Config{ diff --git a/translator/translate/otel/receiver/hostmetrics/translator_test.go b/translator/translate/otel/receiver/hostmetrics/translator_test.go index 11eb24c322a..232876437bf 100644 --- a/translator/translate/otel/receiver/hostmetrics/translator_test.go +++ b/translator/translate/otel/receiver/hostmetrics/translator_test.go @@ -45,7 +45,7 @@ func TestTranslate(t *testing.T) { "opentelemetry": map[string]interface{}{ "collect": map[string]interface{}{ "host_metrics": map[string]interface{}{ - "metrics_collection_interval": 10, + "collection_interval": 10, }, }, }, @@ -60,7 +60,7 @@ func TestTranslate(t *testing.T) { "opentelemetry": map[string]interface{}{ "collect": map[string]interface{}{ "host_metrics": map[string]interface{}{ - "metrics_collection_interval": 10, + "collection_interval": 10, }, }, },