Skip to content

Commit d2b097a

Browse files
PaamickyPaamicky
authored
Gate cluster-name resolution behind KubernetesMode (#2259)
1 parent a2fbeb6 commit d2b097a

8 files changed

Lines changed: 272 additions & 81 deletions

File tree

‎translator/tocwconfig/sampleConfig/opentelemetry/combined_v1_v2_ec2_config.yaml‎

Lines changed: 4 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -544,9 +544,9 @@ processors:
544544
ec2_instance_tag_keys:
545545
- AutoScalingGroupName
546546
ec2_metadata_tags:
547-
- ImageId
548547
- InstanceId
549548
- InstanceType
549+
- ImageId
550550
imds_retries: 1
551551
middleware: agenthealth/statuscode
552552
refresh_tags_interval: 0s
@@ -1029,24 +1029,6 @@ processors:
10291029
- set(scope.attributes["cloudwatch.source"], "cloudwatch-agent")
10301030
- set(scope.attributes["cloudwatch.solution"], "otel-prometheus")
10311031
trace_statements: []
1032-
transform/set_cluster_name:
1033-
error_mode: ignore
1034-
flatten_data: false
1035-
log_statements:
1036-
- context: resource
1037-
error_mode: ignore
1038-
statements:
1039-
- set(resource.attributes["k8s.cluster.name"], "TestCluster")
1040-
metric_statements:
1041-
- context: resource
1042-
error_mode: ignore
1043-
statements:
1044-
- set(resource.attributes["k8s.cluster.name"], "TestCluster")
1045-
trace_statements:
1046-
- context: resource
1047-
error_mode: ignore
1048-
statements:
1049-
- set(resource.attributes["k8s.cluster.name"], "TestCluster")
10501032
receivers:
10511033
awsxray:
10521034
endpoint: 0.0.0.0:2000
@@ -1695,7 +1677,6 @@ service:
16951677
- otlphttp/logs
16961678
processors:
16971679
- resourcedetection/opentelemetry
1698-
- transform/set_cluster_name
16991680
- transform/identity
17001681
- transform/logs_routing
17011682
- attributestocontext/opentelemetry
@@ -1731,13 +1712,13 @@ service:
17311712
- ec2tagger
17321713
- awsentity/resource
17331714
receivers:
1715+
- telegraf_processes
17341716
- telegraf_swap
1717+
- telegraf_netstat
1718+
- telegraf_disk
17351719
- telegraf_mem
1736-
- telegraf_processes
17371720
- telegraf_procstat/1917393364
1738-
- telegraf_netstat
17391721
- telegraf_cpu
1740-
- telegraf_disk
17411722
metrics/host_metrics:
17421723
exporters:
17431724
- forward/opentelemetry
@@ -1769,7 +1750,6 @@ service:
17691750
- otlphttp/metrics
17701751
processors:
17711752
- resourcedetection/opentelemetry
1772-
- transform/set_cluster_name
17731753
- transform/identity
17741754
- awsattributelimit/opentelemetry_metrics
17751755
- batch/opentelemetry_metrics
@@ -1795,7 +1775,6 @@ service:
17951775
- otlphttp/traces
17961776
processors:
17971777
- resourcedetection/opentelemetry
1798-
- transform/set_cluster_name
17991778
- transform/identity
18001779
- batch/opentelemetry_traces
18011780
receivers:

‎translator/tocwconfig/sampleConfig/opentelemetry/prometheus_otel_pipeline_config.yaml‎

Lines changed: 0 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -372,24 +372,6 @@ processors:
372372
- set(scope.attributes["cloudwatch.source"], "cloudwatch-agent")
373373
- set(scope.attributes["cloudwatch.solution"], "otel-prometheus")
374374
trace_statements: []
375-
transform/set_cluster_name:
376-
error_mode: ignore
377-
flatten_data: false
378-
log_statements:
379-
- context: resource
380-
error_mode: ignore
381-
statements:
382-
- set(resource.attributes["k8s.cluster.name"], "TestCluster")
383-
metric_statements:
384-
- context: resource
385-
error_mode: ignore
386-
statements:
387-
- set(resource.attributes["k8s.cluster.name"], "TestCluster")
388-
trace_statements:
389-
- context: resource
390-
error_mode: ignore
391-
statements:
392-
- set(resource.attributes["k8s.cluster.name"], "TestCluster")
393375
receivers:
394376
prometheus/opentelemetry:
395377
config:
@@ -451,7 +433,6 @@ service:
451433
- otlphttp/metrics
452434
processors:
453435
- resourcedetection/opentelemetry
454-
- transform/set_cluster_name
455436
- transform/identity
456437
- awsattributelimit/opentelemetry_metrics
457438
- batch/opentelemetry_metrics

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

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -118,19 +118,19 @@ func (t *baseLogsTranslator) Translate(conf *confmap.Conf) (*common.ComponentTra
118118
processors.Set(resourcedetection.NewTranslator(resourcedetection.WithName(common.OpenTelemetryKey)))
119119
if context.CurrentContext().KubernetesMode() != "" {
120120
processors.Set(k8sattributesprocessor.NewTranslator(common.OpenTelemetryKey))
121-
}
122-
// Apply root-level cluster name if set
123-
clusterName := common.GetClusterName(conf, common.OtelClusterNameKey)
124-
if clusterName != "" {
125-
if err := common.ValidateClusterName(clusterName); err != nil {
126-
return nil, err
121+
// Apply root-level cluster name if set
122+
clusterName := common.GetClusterName(conf, common.OtelClusterNameKey)
123+
if clusterName != "" {
124+
if err := common.ValidateClusterName(clusterName); err != nil {
125+
return nil, err
126+
}
127+
stmt := fmt.Sprintf(`set(resource.attributes["k8s.cluster.name"], "%s")`, clusterName)
128+
processors.Set(transformprocessor.NewTranslatorWithName("set_cluster_name",
129+
transformprocessor.WithMetricResourceStatements([]string{stmt}),
130+
transformprocessor.WithLogResourceStatements([]string{stmt}),
131+
transformprocessor.WithTraceResourceStatements([]string{stmt}),
132+
))
127133
}
128-
stmt := fmt.Sprintf(`set(resource.attributes["k8s.cluster.name"], "%s")`, clusterName)
129-
processors.Set(transformprocessor.NewTranslatorWithName("set_cluster_name",
130-
transformprocessor.WithMetricResourceStatements([]string{stmt}),
131-
transformprocessor.WithLogResourceStatements([]string{stmt}),
132-
transformprocessor.WithTraceResourceStatements([]string{stmt}),
133-
))
134134
}
135135
processors.Set(transformprocessor.NewTranslatorWithName(common.Identity))
136136
// resourcedetection/opentelemetry re-stamps schema_url post-fan-in; clear it here for CI.

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

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,16 @@ import (
88
"strings"
99
"testing"
1010

11+
"github.com/aws/aws-sdk-go/service/ec2"
1112
"github.com/stretchr/testify/assert"
1213
"github.com/stretchr/testify/require"
1314
"go.opentelemetry.io/collector/confmap"
1415

16+
"github.com/aws/amazon-cloudwatch-agent/translator/config"
17+
"github.com/aws/amazon-cloudwatch-agent/translator/context"
1518
"github.com/aws/amazon-cloudwatch-agent/translator/translate/agent"
1619
"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
20+
"github.com/aws/amazon-cloudwatch-agent/translator/util/tagutil"
1721
)
1822

1923
func TestBaseLogsTranslatorActivatedByContainerInsightsLogs(t *testing.T) {
@@ -159,3 +163,89 @@ func TestBaseLogsTranslatorEmptyRegion(t *testing.T) {
159163
assert.Nil(t, got)
160164
assert.Contains(t, err.Error(), "region is required")
161165
}
166+
167+
func TestBaseLogsTranslatorClusterName(t *testing.T) {
168+
agent.Global_Config.Region = "us-east-1"
169+
// Cluster name is only applied in a Kubernetes environment.
170+
context.CurrentContext().SetKubernetesMode(config.ModeEKS)
171+
t.Cleanup(func() { context.CurrentContext().SetKubernetesMode("") })
172+
tt := NewBaseLogsTranslator()
173+
174+
conf := confmap.NewFromStringMap(map[string]interface{}{
175+
"opentelemetry": map[string]interface{}{
176+
"cluster_name": "test-cluster",
177+
"collect": map[string]interface{}{
178+
"logs": map[string]interface{}{},
179+
},
180+
},
181+
})
182+
183+
got, err := tt.Translate(conf)
184+
require.NoError(t, err)
185+
186+
// Verify set_cluster_name processor is present
187+
keys := make([]string, 0, got.Processors.Len())
188+
for _, k := range got.Processors.Keys() {
189+
keys = append(keys, k.String())
190+
}
191+
assert.Contains(t, keys, "transform/set_cluster_name")
192+
}
193+
194+
// TestLogsClusterNameSkippedNonK8s verifies the cluster name is gated on Kubernetes mode
195+
func TestLogsClusterNameSkippedNonK8s(t *testing.T) {
196+
agent.Global_Config.Region = "us-east-1"
197+
context.CurrentContext().SetKubernetesMode("")
198+
t.Cleanup(func() { context.CurrentContext().SetKubernetesMode("") })
199+
tt := NewBaseLogsTranslator()
200+
201+
conf := confmap.NewFromStringMap(map[string]interface{}{
202+
"opentelemetry": map[string]interface{}{
203+
"cluster_name": "test-cluster",
204+
"collect": map[string]interface{}{
205+
"logs": map[string]interface{}{},
206+
},
207+
},
208+
})
209+
210+
got, err := tt.Translate(conf)
211+
require.NoError(t, err)
212+
213+
keys := make([]string, 0, got.Processors.Len())
214+
for _, k := range got.Processors.Keys() {
215+
keys = append(keys, k.String())
216+
}
217+
assert.NotContains(t, keys, "transform/set_cluster_name")
218+
}
219+
220+
func TestBaseLogsTranslatorNoClusterName(t *testing.T) {
221+
agent.Global_Config.Region = "us-east-1"
222+
// Force the ec2 tag lookup with no network to return no cluster name.
223+
context.CurrentContext().SetKubernetesMode(config.ModeEKS)
224+
t.Cleanup(func() { context.CurrentContext().SetKubernetesMode("") })
225+
tagutil.SetEC2APIProviderForTesting(func() interface {
226+
DescribeTags(input *ec2.DescribeTagsInput) (*ec2.DescribeTagsOutput, error)
227+
} {
228+
return noTagsEC2Client{}
229+
})
230+
t.Cleanup(tagutil.ResetEC2APIProvider)
231+
t.Cleanup(tagutil.ResetTagsCache)
232+
233+
tt := NewBaseLogsTranslator()
234+
235+
conf := confmap.NewFromStringMap(map[string]interface{}{
236+
"opentelemetry": map[string]interface{}{
237+
"collect": map[string]interface{}{
238+
"logs": map[string]interface{}{},
239+
},
240+
},
241+
})
242+
243+
got, err := tt.Translate(conf)
244+
require.NoError(t, err)
245+
246+
keys := make([]string, 0, got.Processors.Len())
247+
for _, k := range got.Processors.Keys() {
248+
keys = append(keys, k.String())
249+
}
250+
assert.NotContains(t, keys, "transform/set_cluster_name")
251+
}

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

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -67,19 +67,19 @@ func (t *baseMetricsTranslator) Translate(conf *confmap.Conf) (*common.Component
6767
processors.Set(resourcedetection.NewTranslator(resourcedetection.WithName(common.OpenTelemetryKey)))
6868
if context.CurrentContext().KubernetesMode() != "" {
6969
processors.Set(k8sattributesprocessor.NewTranslator(common.OpenTelemetryKey))
70-
}
71-
// Apply root-level cluster name if set
72-
clusterName := common.GetClusterName(conf, common.OtelClusterNameKey)
73-
if clusterName != "" {
74-
if err := common.ValidateClusterName(clusterName); err != nil {
75-
return nil, err
70+
// Apply root-level cluster name if set
71+
clusterName := common.GetClusterName(conf, common.OtelClusterNameKey)
72+
if clusterName != "" {
73+
if err := common.ValidateClusterName(clusterName); err != nil {
74+
return nil, err
75+
}
76+
stmt := fmt.Sprintf(`set(resource.attributes["k8s.cluster.name"], "%s")`, clusterName)
77+
processors.Set(transformprocessor.NewTranslatorWithName("set_cluster_name",
78+
transformprocessor.WithMetricResourceStatements([]string{stmt}),
79+
transformprocessor.WithLogResourceStatements([]string{stmt}),
80+
transformprocessor.WithTraceResourceStatements([]string{stmt}),
81+
))
7682
}
77-
stmt := fmt.Sprintf(`set(resource.attributes["k8s.cluster.name"], "%s")`, clusterName)
78-
processors.Set(transformprocessor.NewTranslatorWithName("set_cluster_name",
79-
transformprocessor.WithMetricResourceStatements([]string{stmt}),
80-
transformprocessor.WithLogResourceStatements([]string{stmt}),
81-
transformprocessor.WithTraceResourceStatements([]string{stmt}),
82-
))
8383
}
8484
processors.Set(transformprocessor.NewTranslatorWithName(common.Identity))
8585
// Cluster-scoped Container Insights metrics (marked by transform/cw_k8s_ci_v0_mark_cluster)

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

Lines changed: 52 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,15 +8,27 @@ import (
88
"testing"
99
"time"
1010

11+
"github.com/aws/aws-sdk-go/service/ec2"
1112
"github.com/stretchr/testify/assert"
1213
"github.com/stretchr/testify/require"
1314
"go.opentelemetry.io/collector/confmap"
1415
"go.opentelemetry.io/collector/processor/batchprocessor"
1516

17+
"github.com/aws/amazon-cloudwatch-agent/translator/config"
18+
"github.com/aws/amazon-cloudwatch-agent/translator/context"
1619
"github.com/aws/amazon-cloudwatch-agent/translator/translate/agent"
1720
"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
21+
"github.com/aws/amazon-cloudwatch-agent/translator/util/tagutil"
1822
)
1923

24+
// noTagsEC2Client returns no tags, used to deterministically exercise the "cluster name
25+
// looked up but none found" path without any network calls.
26+
type noTagsEC2Client struct{}
27+
28+
func (noTagsEC2Client) DescribeTags(*ec2.DescribeTagsInput) (*ec2.DescribeTagsOutput, error) {
29+
return &ec2.DescribeTagsOutput{}, nil
30+
}
31+
2032
func TestBaseMetricsTranslator(t *testing.T) {
2133
tt := NewBaseMetricsTranslator()
2234
assert.EqualValues(t, "metrics/opentelemetry", tt.ID().String())
@@ -140,6 +152,9 @@ func TestBaseMetricsTranslatorEmptyRegion(t *testing.T) {
140152

141153
func TestBaseMetricsTranslatorClusterName(t *testing.T) {
142154
agent.Global_Config.Region = "us-east-1"
155+
// Cluster name is only applied in a Kubernetes environment.
156+
context.CurrentContext().SetKubernetesMode(config.ModeEKS)
157+
t.Cleanup(func() { context.CurrentContext().SetKubernetesMode("") })
143158
tt := NewBaseMetricsTranslator()
144159

145160
conf := confmap.NewFromStringMap(map[string]interface{}{
@@ -162,8 +177,45 @@ func TestBaseMetricsTranslatorClusterName(t *testing.T) {
162177
assert.Contains(t, keys, "transform/set_cluster_name")
163178
}
164179

180+
// TestClusterNameSkippedNonK8s verifies the cluster name is gated on Kubernetes mode
181+
func TestClusterNameSkippedNonK8s(t *testing.T) {
182+
agent.Global_Config.Region = "us-east-1"
183+
context.CurrentContext().SetKubernetesMode("") // non-Kubernetes (EC2 host)
184+
t.Cleanup(func() { context.CurrentContext().SetKubernetesMode("") })
185+
tt := NewBaseMetricsTranslator()
186+
187+
conf := confmap.NewFromStringMap(map[string]interface{}{
188+
"opentelemetry": map[string]interface{}{
189+
"cluster_name": "test-cluster",
190+
"collect": map[string]interface{}{
191+
"host_metrics": map[string]interface{}{},
192+
},
193+
},
194+
})
195+
196+
got, err := tt.Translate(conf)
197+
require.NoError(t, err)
198+
199+
keys := make([]string, 0, got.Processors.Len())
200+
for _, k := range got.Processors.Keys() {
201+
keys = append(keys, k.String())
202+
}
203+
assert.NotContains(t, keys, "transform/set_cluster_name")
204+
}
205+
165206
func TestBaseMetricsTranslatorNoClusterName(t *testing.T) {
166207
agent.Global_Config.Region = "us-east-1"
208+
// Force the ec2 tag lookup with no network to return no cluster name
209+
context.CurrentContext().SetKubernetesMode(config.ModeEKS)
210+
t.Cleanup(func() { context.CurrentContext().SetKubernetesMode("") })
211+
tagutil.SetEC2APIProviderForTesting(func() interface {
212+
DescribeTags(input *ec2.DescribeTagsInput) (*ec2.DescribeTagsOutput, error)
213+
} {
214+
return noTagsEC2Client{}
215+
})
216+
t.Cleanup(tagutil.ResetEC2APIProvider)
217+
t.Cleanup(tagutil.ResetTagsCache)
218+
167219
tt := NewBaseMetricsTranslator()
168220

169221
conf := confmap.NewFromStringMap(map[string]interface{}{
@@ -177,7 +229,6 @@ func TestBaseMetricsTranslatorNoClusterName(t *testing.T) {
177229
got, err := tt.Translate(conf)
178230
require.NoError(t, err)
179231

180-
// Verify set_cluster_name processor is NOT present
181232
keys := make([]string, 0, got.Processors.Len())
182233
for _, k := range got.Processors.Keys() {
183234
keys = append(keys, k.String())

0 commit comments

Comments
 (0)