Skip to content

Commit c6575bf

Browse files
Paamickyjefchien
Paamicky
and
jefchien
authored
Support CWAGENT_ROLE env var for Container Insights mode in JSON v2 config (#2185)
Co-authored-by: Jeffrey Chien <chienjef@amazon.com>
1 parent 5e34d71 commit c6575bf

5 files changed

Lines changed: 278 additions & 13 deletions

File tree

‎translator/config/schema.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1859,7 +1859,7 @@
18591859
"$ref": "#/definitions/timeIntervalDefinition"
18601860
},
18611861
"mode": {
1862-
"description": "Pipeline mode: node (daemonset), cluster (deployment), or omit for all",
1862+
"description": "Pipeline mode: node (daemonset), cluster (deployment). If omitted, falls back to CWAGENT_ROLE env var",
18631863
"type": "string",
18641864
"enum": ["node", "cluster"]
18651865
},

‎translator/translate/otel/pipeline/opentelemetry/containerinsights/common.go‎

Lines changed: 24 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,18 +5,24 @@ package containerinsights
55

66
import (
77
"fmt"
8+
"os"
89
"regexp"
10+
"strings"
911
"time"
1012

1113
"go.opentelemetry.io/collector/component"
1214
"go.opentelemetry.io/collector/confmap"
1315

16+
"github.com/aws/amazon-cloudwatch-agent/cfg/envconfig"
1417
"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
1518
)
1619

1720
const (
1821
ciPrefix = "cw_k8s_ci_v0"
1922
defaultCollectionInterval = 30 * time.Second
23+
24+
modeNode = "node"
25+
modeCluster = "cluster"
2026
)
2127

2228
var ciConfigKey = common.ConfigKey(common.OpenTelemetryKey, common.CollectKey, common.OtelContainerInsightsKey)
@@ -117,16 +123,27 @@ func logsEnabled(conf *confmap.Conf) bool {
117123
return common.GetOrDefaultBool(conf, key, false)
118124
}
119125

120-
// getMode returns the container_insights.mode value ("node", "cluster", or "" for all).
126+
// getMode resolves the container insights pipeline mode using the following
127+
// priority order:
128+
// 1. JSON config field
129+
// 2. Environment variable
130+
// 3. Default: "node" (DaemonSet)
121131
func getMode(conf *confmap.Conf) string {
122-
if conf == nil {
123-
return ""
132+
if conf != nil {
133+
key := common.ConfigKey(ciConfigKey, "mode")
134+
if v, ok := common.GetString(conf, key); ok && v != "" {
135+
return v
136+
}
124137
}
125-
key := common.ConfigKey(ciConfigKey, "mode")
126-
if v, ok := common.GetString(conf, key); ok {
127-
return v
138+
if role := strings.ToUpper(os.Getenv(envconfig.CWAGENT_ROLE)); role != "" {
139+
switch role {
140+
case envconfig.NODE:
141+
return modeNode
142+
case envconfig.LEADER:
143+
return modeCluster
144+
}
128145
}
129-
return ""
146+
return modeNode
130147
}
131148

132149
type pipelineSpec struct {

‎translator/translate/otel/pipeline/opentelemetry/containerinsights/common_test.go‎

Lines changed: 105 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,14 @@
33

44
package containerinsights
55

6-
import "testing"
6+
import (
7+
"testing"
8+
9+
"github.com/stretchr/testify/assert"
10+
"go.opentelemetry.io/collector/confmap"
11+
12+
"github.com/aws/amazon-cloudwatch-agent/cfg/envconfig"
13+
)
714

815
func TestEscapeDollarDigit(t *testing.T) {
916
tests := []struct {
@@ -35,3 +42,100 @@ func TestEscapeDollarDigit(t *testing.T) {
3542
})
3643
}
3744
}
45+
46+
func TestGetMode_JSONConfig(t *testing.T) {
47+
cfg := confmap.NewFromStringMap(map[string]interface{}{
48+
"opentelemetry": map[string]interface{}{
49+
"collect": map[string]interface{}{
50+
"container_insights": map[string]interface{}{
51+
"mode": "cluster",
52+
},
53+
},
54+
},
55+
})
56+
assert.Equal(t, modeCluster, getMode(cfg))
57+
}
58+
59+
func TestGetMode_EnvVarFallback(t *testing.T) {
60+
cfg := confmap.NewFromStringMap(map[string]interface{}{
61+
"opentelemetry": map[string]interface{}{
62+
"collect": map[string]interface{}{
63+
"container_insights": map[string]interface{}{},
64+
},
65+
},
66+
})
67+
68+
t.Setenv(envconfig.CWAGENT_ROLE, envconfig.LEADER)
69+
assert.Equal(t, modeCluster, getMode(cfg))
70+
71+
t.Setenv(envconfig.CWAGENT_ROLE, envconfig.NODE)
72+
assert.Equal(t, modeNode, getMode(cfg))
73+
}
74+
75+
func TestGetMode_DefaultsToNode(t *testing.T) {
76+
cfg := confmap.NewFromStringMap(map[string]interface{}{
77+
"opentelemetry": map[string]interface{}{
78+
"collect": map[string]interface{}{
79+
"container_insights": map[string]interface{}{},
80+
},
81+
},
82+
})
83+
assert.Equal(t, modeNode, getMode(cfg))
84+
}
85+
86+
func TestGetMode_EnvVarCaseInsensitive(t *testing.T) {
87+
cfg := confmap.NewFromStringMap(map[string]interface{}{
88+
"opentelemetry": map[string]interface{}{
89+
"collect": map[string]interface{}{
90+
"container_insights": map[string]interface{}{},
91+
},
92+
},
93+
})
94+
95+
t.Setenv(envconfig.CWAGENT_ROLE, "leader") // lowercase
96+
assert.Equal(t, modeCluster, getMode(cfg))
97+
98+
t.Setenv(envconfig.CWAGENT_ROLE, "node") // lowercase
99+
assert.Equal(t, modeNode, getMode(cfg))
100+
101+
t.Setenv(envconfig.CWAGENT_ROLE, "Leader") // mixed case
102+
assert.Equal(t, modeCluster, getMode(cfg))
103+
}
104+
105+
func TestGetMode_JSONOverridesEnv(t *testing.T) {
106+
t.Setenv(envconfig.CWAGENT_ROLE, envconfig.NODE)
107+
cfg := confmap.NewFromStringMap(map[string]interface{}{
108+
"opentelemetry": map[string]interface{}{
109+
"collect": map[string]interface{}{
110+
"container_insights": map[string]interface{}{
111+
"mode": "cluster",
112+
},
113+
},
114+
},
115+
})
116+
assert.Equal(t, modeCluster, getMode(cfg))
117+
}
118+
119+
func TestLogsEnabled(t *testing.T) {
120+
tests := []struct {
121+
name string
122+
cfg *confmap.Conf
123+
want bool
124+
}{
125+
{"nil config", nil, false},
126+
{"not set", confmap.NewFromStringMap(map[string]interface{}{
127+
"opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"container_insights": map[string]interface{}{}}},
128+
}), false},
129+
{"enabled true", confmap.NewFromStringMap(map[string]interface{}{
130+
"opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"container_insights": map[string]interface{}{"logs": map[string]interface{}{"enabled": true}}}},
131+
}), true},
132+
{"enabled false", confmap.NewFromStringMap(map[string]interface{}{
133+
"opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"container_insights": map[string]interface{}{"logs": map[string]interface{}{"enabled": false}}}},
134+
}), false},
135+
}
136+
for _, tt := range tests {
137+
t.Run(tt.name, func(t *testing.T) {
138+
assert.Equal(t, tt.want, logsEnabled(tt.cfg))
139+
})
140+
}
141+
}

‎translator/translate/otel/pipeline/opentelemetry/containerinsights/translator.go‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -74,16 +74,21 @@ var apiserverYAML string
7474
var kubeStateMetricsYAML string
7575

7676
// NewTranslators returns all container insights pipeline translators.
77-
// The pipelines generated depend on the "mode" config field:
77+
// The pipelines generated depend on the resolved mode (see getMode for priority):
7878
// - "node": daemonset pipelines (per-node metrics + logs)
7979
// - "cluster": deployment pipelines (cluster-wide metrics)
80-
// - omitted: all pipelines
8180
func NewTranslators(conf *confmap.Conf) common.PipelineTranslatorMap {
8281
translators := common.NewTranslatorMap[*common.ComponentTranslators, pipeline.ID]()
82+
83+
// Guard: no container_insights config key means no pipelines to build.
84+
if conf == nil || !conf.IsSet(ciConfigKey) {
85+
return translators
86+
}
87+
8388
mode := getMode(conf)
8489

8590
// Daemonset metrics pipelines
86-
if mode == "" || mode == "node" {
91+
if mode == modeNode {
8792
translators.Set(newYAMLPipeline("kubeletstats", pipeline.SignalMetrics, kubeletstatsYAML))
8893
translators.Set(newYAMLPipeline("cadvisor", pipeline.SignalMetrics, cadvisorYAML))
8994
translators.Set(newYAMLPipeline("node_exporter", pipeline.SignalMetrics, nodeExporterYAML))
@@ -101,7 +106,7 @@ func NewTranslators(conf *confmap.Conf) common.PipelineTranslatorMap {
101106
}
102107

103108
// Deployment metrics pipelines
104-
if mode == "cluster" {
109+
if mode == modeCluster {
105110
translators.Set(newYAMLPipeline("apiserver", pipeline.SignalMetrics, apiserverYAML))
106111
translators.Set(newYAMLPipeline("kube_state_metrics", pipeline.SignalMetrics, kubeStateMetricsYAML))
107112
}

‎translator/translate/otel/pipeline/opentelemetry/containerinsights/translators_test.go‎

Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,139 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: MIT
3+
4+
package containerinsights
5+
6+
import (
7+
"testing"
8+
9+
"github.com/stretchr/testify/assert"
10+
"go.opentelemetry.io/collector/confmap"
11+
12+
"github.com/aws/amazon-cloudwatch-agent/cfg/envconfig"
13+
)
14+
15+
func TestNewTranslators_MissingKey(t *testing.T) {
16+
// nil config - should return 0 translators (no container_insights key present)
17+
assert.Equal(t, 0, NewTranslators(nil).Len())
18+
// empty config - should return 0 translators
19+
assert.Equal(t, 0, NewTranslators(confmap.NewFromStringMap(map[string]interface{}{})).Len())
20+
}
21+
22+
func TestNewTranslators_ModeNode(t *testing.T) {
23+
cfg := confmap.NewFromStringMap(map[string]interface{}{
24+
"opentelemetry": map[string]interface{}{
25+
"collect": map[string]interface{}{
26+
"container_insights": map[string]interface{}{
27+
"cluster_name": "test-cluster",
28+
"mode": "node",
29+
},
30+
},
31+
},
32+
})
33+
translators := NewTranslators(cfg)
34+
// node mode: kubeletstats, cadvisor, node_exporter, dcgm, neuron, efa, ebs_csi, lis_csi = 8 pipelines
35+
assert.Equal(t, 8, translators.Len())
36+
}
37+
38+
func TestNewTranslators_ModeNodeWithLogs(t *testing.T) {
39+
cfg := confmap.NewFromStringMap(map[string]interface{}{
40+
"opentelemetry": map[string]interface{}{
41+
"collect": map[string]interface{}{
42+
"container_insights": map[string]interface{}{
43+
"cluster_name": "test-cluster",
44+
"mode": "node",
45+
"logs": map[string]interface{}{
46+
"enabled": true,
47+
},
48+
},
49+
},
50+
},
51+
})
52+
translators := NewTranslators(cfg)
53+
// node mode + logs: 8 metric pipelines + 2 log pipelines = 10
54+
assert.Equal(t, 10, translators.Len())
55+
}
56+
57+
func TestNewTranslators_ModeCluster(t *testing.T) {
58+
cfg := confmap.NewFromStringMap(map[string]interface{}{
59+
"opentelemetry": map[string]interface{}{
60+
"collect": map[string]interface{}{
61+
"container_insights": map[string]interface{}{
62+
"cluster_name": "test-cluster",
63+
"mode": "cluster",
64+
},
65+
},
66+
},
67+
})
68+
translators := NewTranslators(cfg)
69+
// cluster mode: apiserver, kube_state_metrics = 2 pipelines
70+
assert.Equal(t, 2, translators.Len())
71+
}
72+
73+
func TestNewTranslators_DefaultMode(t *testing.T) {
74+
// No mode specified, no env var - should default to node
75+
cfg := confmap.NewFromStringMap(map[string]interface{}{
76+
"opentelemetry": map[string]interface{}{
77+
"collect": map[string]interface{}{
78+
"container_insights": map[string]interface{}{
79+
"cluster_name": "test-cluster",
80+
},
81+
},
82+
},
83+
})
84+
translators := NewTranslators(cfg)
85+
// defaults to node mode: 8 pipelines
86+
assert.Equal(t, 8, translators.Len())
87+
}
88+
89+
func TestNewTranslators_EnvVarFallback_Node(t *testing.T) {
90+
// No mode in config, CWAGENT_ROLE=NODE
91+
t.Setenv(envconfig.CWAGENT_ROLE, envconfig.NODE)
92+
cfg := confmap.NewFromStringMap(map[string]interface{}{
93+
"opentelemetry": map[string]interface{}{
94+
"collect": map[string]interface{}{
95+
"container_insights": map[string]interface{}{
96+
"cluster_name": "test-cluster",
97+
},
98+
},
99+
},
100+
})
101+
translators := NewTranslators(cfg)
102+
// env var NODE -> node mode: 8 pipelines
103+
assert.Equal(t, 8, translators.Len())
104+
}
105+
106+
func TestNewTranslators_EnvVarFallback_Leader(t *testing.T) {
107+
// No mode in config, CWAGENT_ROLE=LEADER
108+
t.Setenv(envconfig.CWAGENT_ROLE, envconfig.LEADER)
109+
cfg := confmap.NewFromStringMap(map[string]interface{}{
110+
"opentelemetry": map[string]interface{}{
111+
"collect": map[string]interface{}{
112+
"container_insights": map[string]interface{}{
113+
"cluster_name": "test-cluster",
114+
},
115+
},
116+
},
117+
})
118+
translators := NewTranslators(cfg)
119+
// env var LEADER -> cluster mode: 2 pipelines
120+
assert.Equal(t, 2, translators.Len())
121+
}
122+
123+
func TestNewTranslators_JSONConfigOverridesEnvVar(t *testing.T) {
124+
// JSON says cluster, env var says NODE -> JSON wins
125+
t.Setenv(envconfig.CWAGENT_ROLE, envconfig.NODE)
126+
cfg := confmap.NewFromStringMap(map[string]interface{}{
127+
"opentelemetry": map[string]interface{}{
128+
"collect": map[string]interface{}{
129+
"container_insights": map[string]interface{}{
130+
"cluster_name": "test-cluster",
131+
"mode": "cluster",
132+
},
133+
},
134+
},
135+
})
136+
translators := NewTranslators(cfg)
137+
// JSON config wins: cluster mode = 2 pipelines
138+
assert.Equal(t, 2, translators.Len())
139+
}

0 commit comments

Comments
 (0)