Add cumulative to delta conversion for JMX metrics. (aws#1264)

translator/tocwconfig/sampleConfig/complete_linux_config.yaml

@@ -117,6 +117,13 @@ processors:
             match_type: ""
         initial_value: 0
         max_staleness: 0s
+    cumulativetodelta/jmx:
+        exclude:
+            match_type: ""
+        include:
+            match_type: ""
+        initial_value: 0
+        max_staleness: 0s
     ec2tagger:
         ec2_instance_tag_keys:
             - AutoScalingGroupName
@@ -384,6 +391,7 @@ service:
             processors:
                 - filter/jmx/0
                 - resource/jmx
+                - cumulativetodelta/jmx
                 - transform/jmx/0
                 - ec2tagger
             receivers:
@@ -394,6 +402,7 @@ service:
             processors:
                 - filter/jmx/1
                 - resource/jmx
+                - cumulativetodelta/jmx
                 - transform/jmx/1
                 - ec2tagger
             receivers:

translator/tocwconfig/sampleConfig/jmx_config_linux.yaml

@@ -21,6 +21,13 @@ extensions:
                 mode: EC2
                 region_type: ACJ
 processors:
+    cumulativetodelta/jmx:
+        exclude:
+            match_type: ""
+        include:
+            match_type: ""
+        initial_value: 0
+        max_staleness: 0s
     filter/jmx:
         error_mode: propagate
         logs: {}
@@ -106,6 +113,7 @@ service:
             processors:
                 - filter/jmx
                 - resource/jmx
+                - cumulativetodelta/jmx
                 - transform/jmx
             receivers:
                 - jmx

translator/translate/otel/common/common.go

@@ -392,3 +392,13 @@ func GetMeasurements(m map[string]any) []string {
 	}
 	return results
 }
+
+// IsAnySet checks if any of the provided keys are present in the configuration.
+func IsAnySet(conf *confmap.Conf, keys []string) bool {
+	for _, key := range keys {
+		if conf.IsSet(key) {
+			return true
+		}
+	}
+	return false
+}

translator/translate/otel/common/common_test.go

@@ -34,7 +34,7 @@ func TestConfigKeys(t *testing.T) {
 }
 
 func TestGetString(t *testing.T) {
-	conf := confmap.NewFromStringMap(map[string]interface{}{"int": 10, "string": "test"})
+	conf := confmap.NewFromStringMap(map[string]any{"int": 10, "string": "test"})
 	got, ok := GetString(conf, "int")
 	require.True(t, ok)
 	// converts int to string
@@ -69,7 +69,7 @@ func TestGetArray(t *testing.T) {
 }
 
 func TestGetBool(t *testing.T) {
-	conf := confmap.NewFromStringMap(map[string]interface{}{"int": 10, "string": "test", "bool1": false, "bool2": true})
+	conf := confmap.NewFromStringMap(map[string]any{"int": 10, "string": "test", "bool1": false, "bool2": true})
 	got, ok := GetBool(conf, "int")
 	require.False(t, ok)
 	require.False(t, got)
@@ -88,7 +88,7 @@ func TestGetBool(t *testing.T) {
 }
 
 func TestGetOrDefaultBool(t *testing.T) {
-	conf := confmap.NewFromStringMap(map[string]interface{}{"int": 10, "string": "test", "bool1": false, "bool2": true})
+	conf := confmap.NewFromStringMap(map[string]any{"int": 10, "string": "test", "bool1": false, "bool2": true})
 	got := GetOrDefaultBool(conf, "int", false)
 	require.False(t, got)
 
@@ -106,10 +106,10 @@ func TestGetOrDefaultBool(t *testing.T) {
 }
 
 func TestGetNumber(t *testing.T) {
-	test := map[string]interface{}{"int": 10, "string": "test", "bool": false, "float": 1.3}
+	test := map[string]any{"int": 10, "string": "test", "bool": false, "float": 1.3}
 	marshalled, err := json.Marshal(test)
 	require.NoError(t, err)
-	var unmarshalled map[string]interface{}
+	var unmarshalled map[string]any
 	require.NoError(t, json.Unmarshal(marshalled, &unmarshalled))
 
 	conf := confmap.NewFromStringMap(unmarshalled)
@@ -131,7 +131,7 @@ func TestGetNumber(t *testing.T) {
 }
 
 func TestGetDuration(t *testing.T) {
-	conf := confmap.NewFromStringMap(map[string]interface{}{"invalid": "invalid", "valid": 1, "zero": 0})
+	conf := confmap.NewFromStringMap(map[string]any{"invalid": "invalid", "valid": 1, "zero": 0})
 	got, ok := GetDuration(conf, "invalid")
 	require.False(t, ok)
 	require.Equal(t, time.Duration(0), got)
@@ -145,7 +145,7 @@ func TestGetDuration(t *testing.T) {
 
 func TestParseDuration(t *testing.T) {
 	testCases := map[string]struct {
-		input   interface{}
+		input   any
 		want    time.Duration
 		wantErr bool
 	}{
@@ -201,50 +201,50 @@ func TestMissingKeyError(t *testing.T) {
 func TestGetOrDefaultDuration(t *testing.T) {
 	sectionKeys := []string{"section::metrics_collection_interval", "backup::metrics_collection_interval"}
 	testCases := map[string]struct {
-		input map[string]interface{}
+		input map[string]any
 		want  time.Duration
 	}{
 		"WithDefault": {
-			input: map[string]interface{}{},
+			input: map[string]any{},
 			want:  time.Minute,
 		},
 		"WithZeroInterval": {
-			input: map[string]interface{}{
-				"backup": map[string]interface{}{
+			input: map[string]any{
+				"backup": map[string]any{
 					"metrics_collection_interval": 0,
 				},
-				"section": map[string]interface{}{
+				"section": map[string]any{
 					"metrics_collection_interval": 0,
 				},
 			},
 			want: time.Minute,
 		},
 		"WithoutSectionOverride": {
-			input: map[string]interface{}{
-				"backup": map[string]interface{}{
+			input: map[string]any{
+				"backup": map[string]any{
 					"metrics_collection_interval": 10,
 				},
-				"section": map[string]interface{}{},
+				"section": map[string]any{},
 			},
 			want: 10 * time.Second,
 		},
 		"WithInvalidSectionOverride": {
-			input: map[string]interface{}{
-				"backup": map[string]interface{}{
+			input: map[string]any{
+				"backup": map[string]any{
 					"metrics_collection_interval": 10,
 				},
-				"section": map[string]interface{}{
+				"section": map[string]any{
 					"metrics_collection_interval": "invalid",
 				},
 			},
 			want: 10 * time.Second,
 		},
 		"WithSectionOverride": {
-			input: map[string]interface{}{
-				"backup": map[string]interface{}{
+			input: map[string]any{
+				"backup": map[string]any{
 					"metrics_collection_interval": 10,
 				},
-				"section": map[string]interface{}{
+				"section": map[string]any{
 					"metrics_collection_interval": 120,
 				},
 			},
@@ -357,3 +357,30 @@ func TestGetMeasurements(t *testing.T) {
 		})
 	}
 }
+
+func TestIsAnySet(t *testing.T) {
+	conf := confmap.NewFromStringMap(map[string]any{
+		"one": map[string]any{
+			"endpoint": "test",
+		},
+		"two": map[string]any{},
+	})
+	testCases := map[string]struct {
+		keys []string
+		want bool
+	}{
+		"NotSet": {
+			keys: []string{"test"},
+			want: false,
+		},
+		"Set": {
+			keys: []string{"test", "one::endpoint"},
+			want: true,
+		},
+	}
+	for name, testCase := range testCases {
+		t.Run(name, func(t *testing.T) {
+			assert.Equal(t, testCase.want, IsAnySet(conf, testCase.keys))
+		})
+	}
+}

translator/translate/otel/common/options.go

@@ -0,0 +1,30 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package common
+
+type TranslatorOption func(any)
+
+type NameSetter interface {
+	SetName(string)
+}
+
+func WithName(name string) TranslatorOption {
+	return func(target any) {
+		if setter, ok := target.(NameSetter); ok {
+			setter.SetName(name)
+		}
+	}
+}
+
+type NameProvider struct {
+	name string
+}
+
+func (p *NameProvider) Name() string {
+	return p.name
+}
+
+func (p *NameProvider) SetName(name string) {
+	p.name = name
+}

translator/translate/otel/common/options_test.go

@@ -0,0 +1,17 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestWithName(t *testing.T) {
+	p := &NameProvider{name: "a"}
+	opt := WithName("b")
+	opt(p)
+	assert.Equal(t, "b", p.Name())
+}

translator/translate/otel/pipeline/host/translator.go

@@ -74,7 +74,7 @@ func (t translator) Translate(conf *confmap.Conf) (*common.ComponentTranslators,
 	// we need to add delta processor because (only) diskio and net input plugins report delta metric
 	if common.PipelineNameHostDeltaMetrics == t.name {
 		log.Printf("D! delta processor required because metrics with diskio or net are set")
-		translators.Processors.Set(cumulativetodeltaprocessor.NewTranslatorWithName(t.name))
+		translators.Processors.Set(cumulativetodeltaprocessor.NewTranslator(common.WithName(t.name), cumulativetodeltaprocessor.WithDiskIONetKeys()))
 	}
 
 	if conf.IsSet(common.ConfigKey(common.MetricsKey, common.AppendDimensionsKey)) {

translator/translate/otel/pipeline/jmx/translator.go

@@ -13,6 +13,7 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/exporter/awscloudwatch"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/agenthealth"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/cumulativetodeltaprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/ec2taggerprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/filterprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/metricsdecorator"
@@ -76,6 +77,7 @@ func (t *translator) Translate(conf *confmap.Conf) (*common.ComponentTranslators
 		Processors: common.NewTranslatorMap(
 			filterprocessor.NewTranslator(filterprocessor.WithName(common.PipelineNameJmx), filterprocessor.WithIndex(t.index)),
 			resourceprocessor.NewTranslator(resourceprocessor.WithName(common.PipelineNameJmx)),
+			cumulativetodeltaprocessor.NewTranslator(common.WithName(common.PipelineNameJmx), cumulativetodeltaprocessor.WithConfigKeys(common.JmxConfigKey)),
 		),
 		Exporters:  common.NewTranslatorMap(awscloudwatch.NewTranslator()),
 		Extensions: common.NewTranslatorMap(agenthealth.NewTranslator(component.DataTypeMetrics, []string{agenthealth.OperationPutMetricData})),

translator/translate/otel/pipeline/jmx/translator_test.go

@@ -123,7 +123,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineID: "metrics/jmx",
 				receivers:  []string{"jmx"},
-				processors: []string{"filter/jmx", "resource/jmx"},
+				processors: []string{"filter/jmx", "resource/jmx", "cumulativetodelta/jmx"},
 				exporters:  []string{"awscloudwatch"},
 				extensions: []string{"agenthealth/metrics"},
 			},
@@ -151,7 +151,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineID: "metrics/jmx",
 				receivers:  []string{"jmx"},
-				processors: []string{"filter/jmx", "resource/jmx", "transform/jmx"},
+				processors: []string{"filter/jmx", "resource/jmx", "cumulativetodelta/jmx", "transform/jmx"},
 				exporters:  []string{"awscloudwatch"},
 				extensions: []string{"agenthealth/metrics"},
 			},
@@ -185,7 +185,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineID: "metrics/jmx/0",
 				receivers:  []string{"jmx/0"},
-				processors: []string{"filter/jmx/0", "resource/jmx", "transform/jmx/0", "ec2tagger"},
+				processors: []string{"filter/jmx/0", "resource/jmx", "cumulativetodelta/jmx", "transform/jmx/0", "ec2tagger"},
 				exporters:  []string{"awscloudwatch"},
 				extensions: []string{"agenthealth/metrics"},
 			},