diff --git a/translator/config/schema.json b/translator/config/schema.json index 680eed8111..648b6fdb03 100644 --- a/translator/config/schema.json +++ b/translator/config/schema.json @@ -1859,7 +1859,7 @@ "$ref": "#/definitions/timeIntervalDefinition" }, "mode": { - "description": "Pipeline mode: node (daemonset), cluster (deployment), or omit for all", + "description": "Pipeline mode: node (daemonset), cluster (deployment). If omitted, falls back to CWAGENT_ROLE env var", "type": "string", "enum": ["node", "cluster"] }, diff --git a/translator/translate/otel/pipeline/opentelemetry/containerinsights/common.go b/translator/translate/otel/pipeline/opentelemetry/containerinsights/common.go index 912d5c3645..4aeb66d17a 100644 --- a/translator/translate/otel/pipeline/opentelemetry/containerinsights/common.go +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/common.go @@ -5,18 +5,24 @@ package containerinsights import ( "fmt" + "os" "regexp" + "strings" "time" "go.opentelemetry.io/collector/component" "go.opentelemetry.io/collector/confmap" + "github.com/aws/amazon-cloudwatch-agent/cfg/envconfig" "github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common" ) const ( ciPrefix = "cw_k8s_ci_v0" defaultCollectionInterval = 30 * time.Second + + modeNode = "node" + modeCluster = "cluster" ) var ciConfigKey = common.ConfigKey(common.OpenTelemetryKey, common.CollectKey, common.OtelContainerInsightsKey) @@ -117,16 +123,27 @@ func logsEnabled(conf *confmap.Conf) bool { return common.GetOrDefaultBool(conf, key, false) } -// getMode returns the container_insights.mode value ("node", "cluster", or "" for all). +// getMode resolves the container insights pipeline mode using the following +// priority order: +// 1. JSON config field +// 2. Environment variable +// 3. Default: "node" (DaemonSet) func getMode(conf *confmap.Conf) string { - if conf == nil { - return "" + if conf != nil { + key := common.ConfigKey(ciConfigKey, "mode") + if v, ok := common.GetString(conf, key); ok && v != "" { + return v + } } - key := common.ConfigKey(ciConfigKey, "mode") - if v, ok := common.GetString(conf, key); ok { - return v + if role := strings.ToUpper(os.Getenv(envconfig.CWAGENT_ROLE)); role != "" { + switch role { + case envconfig.NODE: + return modeNode + case envconfig.LEADER: + return modeCluster + } } - return "" + return modeNode } 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 57a7e3de73..bcc8c86b02 100644 --- a/translator/translate/otel/pipeline/opentelemetry/containerinsights/common_test.go +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/common_test.go @@ -3,7 +3,14 @@ package containerinsights -import "testing" +import ( + "testing" + + "github.com/stretchr/testify/assert" + "go.opentelemetry.io/collector/confmap" + + "github.com/aws/amazon-cloudwatch-agent/cfg/envconfig" +) func TestEscapeDollarDigit(t *testing.T) { tests := []struct { @@ -35,3 +42,100 @@ func TestEscapeDollarDigit(t *testing.T) { }) } } + +func TestGetMode_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", + }, + }, + }, + }) + assert.Equal(t, modeCluster, getMode(cfg)) +} + +func TestGetMode_EnvVarFallback(t *testing.T) { + cfg := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "collect": map[string]interface{}{ + "container_insights": map[string]interface{}{}, + }, + }, + }) + + t.Setenv(envconfig.CWAGENT_ROLE, envconfig.LEADER) + assert.Equal(t, modeCluster, getMode(cfg)) + + t.Setenv(envconfig.CWAGENT_ROLE, envconfig.NODE) + assert.Equal(t, modeNode, getMode(cfg)) +} + +func TestGetMode_DefaultsToNode(t *testing.T) { + cfg := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "collect": map[string]interface{}{ + "container_insights": map[string]interface{}{}, + }, + }, + }) + assert.Equal(t, modeNode, getMode(cfg)) +} + +func TestGetMode_EnvVarCaseInsensitive(t *testing.T) { + cfg := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "collect": map[string]interface{}{ + "container_insights": map[string]interface{}{}, + }, + }, + }) + + t.Setenv(envconfig.CWAGENT_ROLE, "leader") // lowercase + assert.Equal(t, modeCluster, getMode(cfg)) + + t.Setenv(envconfig.CWAGENT_ROLE, "node") // lowercase + assert.Equal(t, modeNode, getMode(cfg)) + + t.Setenv(envconfig.CWAGENT_ROLE, "Leader") // mixed case + assert.Equal(t, modeCluster, getMode(cfg)) +} + +func TestGetMode_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", + }, + }, + }, + }) + assert.Equal(t, modeCluster, getMode(cfg)) +} + +func TestLogsEnabled(t *testing.T) { + tests := []struct { + name string + cfg *confmap.Conf + want bool + }{ + {"nil config", nil, false}, + {"not set", confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"container_insights": map[string]interface{}{}}}, + }), false}, + {"enabled true", confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"container_insights": map[string]interface{}{"logs": map[string]interface{}{"enabled": true}}}}, + }), true}, + {"enabled false", confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"container_insights": map[string]interface{}{"logs": map[string]interface{}{"enabled": false}}}}, + }), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, logsEnabled(tt.cfg)) + }) + } +} diff --git a/translator/translate/otel/pipeline/opentelemetry/containerinsights/translator.go b/translator/translate/otel/pipeline/opentelemetry/containerinsights/translator.go index a6eb6c1115..cbd309655a 100644 --- a/translator/translate/otel/pipeline/opentelemetry/containerinsights/translator.go +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/translator.go @@ -74,16 +74,21 @@ var apiserverYAML string var kubeStateMetricsYAML string // NewTranslators returns all container insights pipeline translators. -// The pipelines generated depend on the "mode" config field: +// The pipelines generated depend on the resolved mode (see getMode for priority): // - "node": daemonset pipelines (per-node metrics + logs) // - "cluster": deployment pipelines (cluster-wide metrics) -// - omitted: all pipelines func NewTranslators(conf *confmap.Conf) common.PipelineTranslatorMap { translators := common.NewTranslatorMap[*common.ComponentTranslators, pipeline.ID]() + + // Guard: no container_insights config key means no pipelines to build. + if conf == nil || !conf.IsSet(ciConfigKey) { + return translators + } + mode := getMode(conf) // Daemonset metrics pipelines - if mode == "" || mode == "node" { + if mode == modeNode { translators.Set(newYAMLPipeline("kubeletstats", pipeline.SignalMetrics, kubeletstatsYAML)) translators.Set(newYAMLPipeline("cadvisor", pipeline.SignalMetrics, cadvisorYAML)) translators.Set(newYAMLPipeline("node_exporter", pipeline.SignalMetrics, nodeExporterYAML)) @@ -101,7 +106,7 @@ func NewTranslators(conf *confmap.Conf) common.PipelineTranslatorMap { } // Deployment metrics pipelines - if mode == "cluster" { + if mode == modeCluster { 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 new file mode 100644 index 0000000000..526356860f --- /dev/null +++ b/translator/translate/otel/pipeline/opentelemetry/containerinsights/translators_test.go @@ -0,0 +1,139 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package containerinsights + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "go.opentelemetry.io/collector/confmap" + + "github.com/aws/amazon-cloudwatch-agent/cfg/envconfig" +) + +func TestNewTranslators_MissingKey(t *testing.T) { + // nil config - should return 0 translators (no container_insights key present) + assert.Equal(t, 0, NewTranslators(nil).Len()) + // empty config - should return 0 translators + assert.Equal(t, 0, NewTranslators(confmap.NewFromStringMap(map[string]interface{}{})).Len()) +} + +func TestNewTranslators_ModeNode(t *testing.T) { + cfg := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "collect": map[string]interface{}{ + "container_insights": map[string]interface{}{ + "cluster_name": "test-cluster", + "mode": "node", + }, + }, + }, + }) + translators := NewTranslators(cfg) + // node mode: 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) { + cfg := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "collect": map[string]interface{}{ + "container_insights": map[string]interface{}{ + "cluster_name": "test-cluster", + "mode": "node", + "logs": map[string]interface{}{ + "enabled": true, + }, + }, + }, + }, + }) + translators := NewTranslators(cfg) + // node mode + logs: 8 metric pipelines + 2 log pipelines = 10 + assert.Equal(t, 10, translators.Len()) +} + +func TestNewTranslators_ModeCluster(t *testing.T) { + cfg := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "collect": map[string]interface{}{ + "container_insights": map[string]interface{}{ + "cluster_name": "test-cluster", + "mode": "cluster", + }, + }, + }, + }) + translators := NewTranslators(cfg) + // cluster mode: 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 + cfg := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "collect": map[string]interface{}{ + "container_insights": map[string]interface{}{ + "cluster_name": "test-cluster", + }, + }, + }, + }) + translators := NewTranslators(cfg) + // defaults to node mode: 8 pipelines + assert.Equal(t, 8, translators.Len()) +} + +func TestNewTranslators_EnvVarFallback_Node(t *testing.T) { + // No mode in config, CWAGENT_ROLE=NODE + 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{}{ + "cluster_name": "test-cluster", + }, + }, + }, + }) + translators := NewTranslators(cfg) + // env var NODE -> node mode: 8 pipelines + assert.Equal(t, 8, translators.Len()) +} + +func TestNewTranslators_EnvVarFallback_Leader(t *testing.T) { + // No mode in config, CWAGENT_ROLE=LEADER + t.Setenv(envconfig.CWAGENT_ROLE, envconfig.LEADER) + cfg := confmap.NewFromStringMap(map[string]interface{}{ + "opentelemetry": map[string]interface{}{ + "collect": map[string]interface{}{ + "container_insights": map[string]interface{}{ + "cluster_name": "test-cluster", + }, + }, + }, + }) + translators := NewTranslators(cfg) + // env var LEADER -> cluster mode: 2 pipelines + assert.Equal(t, 2, translators.Len()) +} + +func TestNewTranslators_JSONConfigOverridesEnvVar(t *testing.T) { + // JSON says cluster, env var says NODE -> JSON wins + 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{}{ + "cluster_name": "test-cluster", + "mode": "cluster", + }, + }, + }, + }) + translators := NewTranslators(cfg) + // JSON config wins: cluster mode = 2 pipelines + assert.Equal(t, 2, translators.Len()) +}