Skip to content

Commit 2534ea1

Browse files
PaamickyPaamicky
committed
Add mode priority order for OTel Container Insights
1 parent 349abc0 commit 2534ea1

4 files changed

Lines changed: 263 additions & 12 deletions

File tree

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

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

66
import (
77
"fmt"
8+
"os"
89
"regexp"
910
"time"
1011

@@ -17,6 +18,16 @@ import (
1718
const (
1819
ciPrefix = "cw_k8s_ci_v0"
1920
defaultCollectionInterval = 30 * time.Second
21+
22+
// envCWAgentRole is the environment variable used by the helm chart to indicate
23+
// whether the agent runs as a DaemonSet (node-level) or Deployment (cluster-level).
24+
envCWAgentRole = "CWAGENT_ROLE"
25+
// Environment variable values for CWAGENT_ROLE
26+
envRoleNode = "NODE"
27+
envRoleLeader = "LEADER"
28+
// Mode values
29+
modeNode = "node"
30+
modeCluster = "cluster"
2031
)
2132

2233
var ciConfigKey = common.ConfigKey(common.OpenTelemetryKey, common.CollectKey, common.OtelContainerInsightsKey)
@@ -117,16 +128,27 @@ func logsEnabled(conf *confmap.Conf) bool {
117128
return common.GetOrDefaultBool(conf, key, false)
118129
}
119130

120-
// getMode returns the container_insights.mode value ("node", "cluster", or "" for all).
131+
// getMode resolves the container insights pipeline mode using the following
132+
// priority order:
133+
// 1. JSON config field
134+
// 2. Environment variable
135+
// 3. Default: "node" (DaemonSet)
121136
func getMode(conf *confmap.Conf) string {
122-
if conf == nil {
123-
return ""
137+
if conf != nil {
138+
key := common.ConfigKey(ciConfigKey, "mode")
139+
if v, ok := common.GetString(conf, key); ok && v != "" {
140+
return v
141+
}
124142
}
125-
key := common.ConfigKey(ciConfigKey, "mode")
126-
if v, ok := common.GetString(conf, key); ok {
127-
return v
143+
if role := os.Getenv(envCWAgentRole); role != "" {
144+
switch role {
145+
case envRoleNode:
146+
return modeNode
147+
case envRoleLeader:
148+
return modeCluster
149+
}
128150
}
129-
return ""
151+
return modeNode
130152
}
131153

132154
type pipelineSpec struct {

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

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

44
package containerinsights
55

6-
import "testing"
6+
import (
7+
"os"
8+
"testing"
9+
10+
"github.com/stretchr/testify/assert"
11+
"go.opentelemetry.io/collector/confmap"
12+
)
713

814
func TestEscapeDollarDigit(t *testing.T) {
915
tests := []struct {
@@ -35,3 +41,82 @@ func TestEscapeDollarDigit(t *testing.T) {
3541
})
3642
}
3743
}
44+
45+
func TestGetMode_JSONConfig(t *testing.T) {
46+
cfg := confmap.NewFromStringMap(map[string]interface{}{
47+
"opentelemetry": map[string]interface{}{
48+
"collect": map[string]interface{}{
49+
"container_insights": map[string]interface{}{
50+
"mode": "cluster",
51+
},
52+
},
53+
},
54+
})
55+
assert.Equal(t, modeCluster, getMode(cfg))
56+
}
57+
58+
func TestGetMode_EnvVarFallback(t *testing.T) {
59+
cfg := confmap.NewFromStringMap(map[string]interface{}{
60+
"opentelemetry": map[string]interface{}{
61+
"collect": map[string]interface{}{
62+
"container_insights": map[string]interface{}{},
63+
},
64+
},
65+
})
66+
67+
t.Setenv(envCWAgentRole, envRoleLeader)
68+
assert.Equal(t, modeCluster, getMode(cfg))
69+
70+
t.Setenv(envCWAgentRole, envRoleNode)
71+
assert.Equal(t, modeNode, getMode(cfg))
72+
}
73+
74+
func TestGetMode_DefaultsToNode(t *testing.T) {
75+
os.Unsetenv(envCWAgentRole)
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_JSONOverridesEnv(t *testing.T) {
87+
t.Setenv(envCWAgentRole, envRoleNode)
88+
cfg := confmap.NewFromStringMap(map[string]interface{}{
89+
"opentelemetry": map[string]interface{}{
90+
"collect": map[string]interface{}{
91+
"container_insights": map[string]interface{}{
92+
"mode": "cluster",
93+
},
94+
},
95+
},
96+
})
97+
assert.Equal(t, modeCluster, getMode(cfg))
98+
}
99+
100+
func TestLogsEnabled(t *testing.T) {
101+
tests := []struct {
102+
name string
103+
cfg *confmap.Conf
104+
want bool
105+
}{
106+
{"nil config", nil, false},
107+
{"not set", confmap.NewFromStringMap(map[string]interface{}{
108+
"opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"container_insights": map[string]interface{}{}}},
109+
}), false},
110+
{"enabled true", confmap.NewFromStringMap(map[string]interface{}{
111+
"opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"container_insights": map[string]interface{}{"logs": map[string]interface{}{"enabled": true}}}},
112+
}), true},
113+
{"enabled false", confmap.NewFromStringMap(map[string]interface{}{
114+
"opentelemetry": map[string]interface{}{"collect": map[string]interface{}{"container_insights": map[string]interface{}{"logs": map[string]interface{}{"enabled": false}}}},
115+
}), false},
116+
}
117+
for _, tt := range tests {
118+
t.Run(tt.name, func(t *testing.T) {
119+
assert.Equal(t, tt.want, logsEnabled(tt.cfg))
120+
})
121+
}
122+
}

‎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+
"os"
8+
"testing"
9+
10+
"github.com/stretchr/testify/assert"
11+
"go.opentelemetry.io/collector/confmap"
12+
)
13+
14+
func TestNewTranslators_MissingKey(t *testing.T) {
15+
// nil config - should return 0 translators (no container_insights key present)
16+
assert.Equal(t, 0, NewTranslators(nil).Len())
17+
// empty config - should return 0 translators
18+
assert.Equal(t, 0, NewTranslators(confmap.NewFromStringMap(map[string]interface{}{})).Len())
19+
}
20+
21+
func TestNewTranslators_ModeNode(t *testing.T) {
22+
cfg := confmap.NewFromStringMap(map[string]interface{}{
23+
"opentelemetry": map[string]interface{}{
24+
"collect": map[string]interface{}{
25+
"container_insights": map[string]interface{}{
26+
"cluster_name": "test-cluster",
27+
"mode": "node",
28+
},
29+
},
30+
},
31+
})
32+
translators := NewTranslators(cfg)
33+
// node mode: kubeletstats, cadvisor, node_exporter, dcgm, neuron, efa, ebs_csi, lis_csi = 8 pipelines
34+
assert.Equal(t, 8, translators.Len())
35+
}
36+
37+
func TestNewTranslators_ModeNodeWithLogs(t *testing.T) {
38+
cfg := confmap.NewFromStringMap(map[string]interface{}{
39+
"opentelemetry": map[string]interface{}{
40+
"collect": map[string]interface{}{
41+
"container_insights": map[string]interface{}{
42+
"cluster_name": "test-cluster",
43+
"mode": "node",
44+
"logs": map[string]interface{}{
45+
"enabled": true,
46+
},
47+
},
48+
},
49+
},
50+
})
51+
translators := NewTranslators(cfg)
52+
// node mode + logs: 8 metric pipelines + 2 log pipelines = 10
53+
assert.Equal(t, 10, translators.Len())
54+
}
55+
56+
func TestNewTranslators_ModeCluster(t *testing.T) {
57+
cfg := confmap.NewFromStringMap(map[string]interface{}{
58+
"opentelemetry": map[string]interface{}{
59+
"collect": map[string]interface{}{
60+
"container_insights": map[string]interface{}{
61+
"cluster_name": "test-cluster",
62+
"mode": "cluster",
63+
},
64+
},
65+
},
66+
})
67+
translators := NewTranslators(cfg)
68+
// cluster mode: apiserver, kube_state_metrics = 2 pipelines
69+
assert.Equal(t, 2, translators.Len())
70+
}
71+
72+
func TestNewTranslators_DefaultMode(t *testing.T) {
73+
// No mode specified, no env var - should default to node
74+
os.Unsetenv(envCWAgentRole)
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(envCWAgentRole, envRoleNode)
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(envCWAgentRole, envRoleLeader)
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(envCWAgentRole, envRoleNode)
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)