Support CWAGENT_ROLE env var for Container Insights mode in JSON v2 config

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"]
                 },

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 {

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))
+		})
+	}
+}

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))
 	}

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())
+}