Add deltatocumulativeprocessor for AMP destinations (#1624)

Author: dricross

go.mod

@@ -211,6 +211,7 @@ require (
 
 require (
 	github.com/aws/aws-sdk-go-v2 v1.32.6
+	github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor v0.115.0
 	go.opentelemetry.io/collector/component/componenttest v0.115.0
 	go.opentelemetry.io/collector/config/configtelemetry v0.115.0
 	go.opentelemetry.io/collector/confmap/converter/expandconverter v0.113.0
@@ -425,6 +426,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/xray v0.115.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.115.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.115.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics v0.115.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter v0.115.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig v0.115.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka v0.115.0 // indirect

go.sum

@@ -1206,6 +1206,8 @@ github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/metrics v
 github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/metrics v0.115.0/go.mod h1:G56rS4nL0VypkD7a94UaQmIjO5t0kffVcjbhpvSogww=
 github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.115.0 h1:vRQQFD4YpasQFUAdF030UWtaflSYFXK542bfWMGhOK0=
 github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.115.0/go.mod h1:BZ7DT+0VkKR7P3I9PGEDfVa0GdB0ty41eEcejIUXF9A=
+github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics v0.115.0 h1:tFUm48xxdtuk3AgY5AY90DJ6UnxRW5k/HBpA24blCAo=
+github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics v0.115.0/go.mod h1:EI5GXHQVRNLx78DSyqSU8ZzIxQayUN7KlaeVChk5rJc=
 github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter v0.115.0 h1:h6zEsBtuZalQu7lKYf6ZCcj8fTocT+zxdmuOou9515Q=
 github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter v0.115.0/go.mod h1:6QU/K0dGCGYorkOvJmhbDFCspy4RPxRkFjf9I64y6I0=
 github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8stest v0.115.0 h1:vXDJE8YHfAoYIAlPRtODchlqb6lWnGhJxPaT2ljvN7I=
@@ -1240,6 +1242,8 @@ github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributespr
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor v0.115.0/go.mod h1:z8XdvlhXSYVboxS3TPGembE9kfxLAYH2PxPLMvf8wTk=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/cumulativetodeltaprocessor v0.115.0 h1:t3BGnPpmeuxW51vISSu51PrAs49ACBCa1Yl1NfZGE5Y=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/cumulativetodeltaprocessor v0.115.0/go.mod h1:jQLYyroEYEV1kWJApmGBgVuGUd73v+Q6EUJ6Wy7N508=
+github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor v0.115.0 h1:a0LC50FmNTyWI/vIhKWSyDRw8GLlq2XFIY7GezlK3/w=
+github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor v0.115.0/go.mod h1:hNQwpth9RZ+nS9oCDDKRo56XwNc2TGhZGcAKc4CF1AI=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatorateprocessor v0.115.0 h1:X6rEs7IxDpcDDBOCmkA3xHmc373UxHchH7BykK3Ao+o=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatorateprocessor v0.115.0/go.mod h1:fmLLh7jL0uK/t8op9TieOz7pwxItl4hdFo2fX7U0Etg=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor v0.115.0 h1:ficXJmB6l6kfiu+R6CmggtnlQWMHUNzu2csDYA4CFSs=

service/defaultcomponents/components.go

@@ -16,6 +16,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/storage/filestorage"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/cumulativetodeltaprocessor"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatorateprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbytraceprocessor"
@@ -96,6 +97,7 @@ func Factories() (otelcol.Factories, error) {
 		awsentity.NewFactory(),
 		batchprocessor.NewFactory(),
 		cumulativetodeltaprocessor.NewFactory(),
+		deltatocumulativeprocessor.NewFactory(),
 		deltatorateprocessor.NewFactory(),
 		ec2tagger.NewFactory(),
 		filterprocessor.NewFactory(),

service/defaultcomponents/components_test.go

@@ -45,6 +45,7 @@ func TestComponents(t *testing.T) {
 		"attributes",
 		"batch",
 		"cumulativetodelta",
+		"deltatocumulative",
 		"deltatorate",
 		"ec2tagger",
 		"metricsgeneration",

translator/tocwconfig/sampleConfig/amp_config_linux.yaml

@@ -99,6 +99,9 @@ processors:
         send_batch_max_size: 0
         send_batch_size: 8192
         timeout: 1m0s
+    deltatocumulative/host/amp:
+        max_stale: 336h0m0s
+        max_streams: 9223372036854775807
     ec2tagger:
         ec2_instance_tag_keys:
             - AutoScalingGroupName
@@ -152,6 +155,7 @@ service:
                 - transform
                 - rollup
                 - batch/host/amp
+                - deltatocumulative/host/amp
             receivers:
                 - telegraf_cpu
         metrics/host/cloudwatch:

translator/tocwconfig/sampleConfig/jmx_config_linux.yaml

@@ -105,6 +105,12 @@ processors:
             match_type: ""
         initial_value: 2
         max_staleness: 0s
+    deltatocumulative/host/amp:
+        max_stale: 336h0m0s
+        max_streams: 9223372036854775807
+    deltatocumulative/jmx/amp:
+        max_stale: 336h0m0s
+        max_streams: 9223372036854775807
     filter/jmx:
         error_mode: propagate
         logs: {}
@@ -187,6 +193,7 @@ service:
             processors:
                 - transform
                 - batch/host/amp
+                - deltatocumulative/host/amp
             receivers:
                 - telegraf_cpu
                 - telegraf_disk
@@ -207,6 +214,7 @@ service:
                 - resource/jmx
                 - transform/jmx
                 - batch/jmx/amp
+                - deltatocumulative/jmx/amp
             receivers:
                 - jmx
         metrics/jmx/cloudwatch:

translator/tocwconfig/sampleConfig/jmx_eks_config_linux.yaml

@@ -92,6 +92,12 @@ processors:
             match_type: ""
         initial_value: 2
         max_staleness: 0s
+    deltatocumulative/jmx/amp/0:
+        max_stale: 336h0m0s
+        max_streams: 9223372036854775807
+    deltatocumulative/jmx/amp/1:
+        max_stale: 336h0m0s
+        max_streams: 9223372036854775807
     filter/jmx/0:
         error_mode: propagate
         logs: {}
@@ -240,6 +246,7 @@ service:
                 - transform/jmx/drop
                 - transform/jmx/0
                 - batch/jmx/amp/0
+                - deltatocumulative/jmx/amp/0
             receivers:
                 - otlp/jmx
         metrics/jmx/amp/1:
@@ -252,6 +259,7 @@ service:
                 - transform/jmx/drop
                 - transform/jmx/1
                 - batch/jmx/amp/1
+                - deltatocumulative/jmx/amp/1
             receivers:
                 - otlp/jmx
         metrics/jmx/cloudwatch/0:

translator/tocwconfig/sampleConfig/prometheus_combined_config_linux.yaml

@@ -151,6 +151,9 @@ processors:
         send_batch_max_size: 0
         send_batch_size: 8192
         timeout: 30s
+    deltatocumulative/prometheus/amp:
+        max_stale: 336h0m0s
+        max_streams: 9223372036854775807
 receivers:
     prometheus:
         config:
@@ -202,6 +205,7 @@ service:
                 - prometheusremotewrite/amp
             processors:
                 - batch/prometheus/amp
+                - deltatocumulative/prometheus/amp
             receivers:
                 - prometheus
         metrics/prometheus/cloudwatchlogs:

translator/tocwconfig/sampleConfig/prometheus_otel_config_linux.yaml

@@ -62,6 +62,9 @@ processors:
         send_batch_max_size: 0
         send_batch_size: 8192
         timeout: 1m0s
+    deltatocumulative/prometheus/amp:
+        max_stale: 336h0m0s
+        max_streams: 9223372036854775807
 receivers:
     prometheus:
         config:
@@ -107,6 +110,7 @@ service:
                 - prometheusremotewrite/amp
             processors:
                 - batch/prometheus/amp
+                - deltatocumulative/prometheus/amp
             receivers:
                 - prometheus
     telemetry:

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

@@ -24,6 +24,7 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/awsentity"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/batchprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/cumulativetodeltaprocessor"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/deltatocumulativeprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/ec2taggerprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/metricsdecorator"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/rollupprocessor"
@@ -136,6 +137,8 @@ func (t translator) Translate(conf *confmap.Conf) (*common.ComponentTranslators,
 			translators.Processors.Set(rollupprocessor.NewTranslator())
 		}
 		translators.Processors.Set(batchprocessor.NewTranslatorWithNameAndSection(t.name, common.MetricsKey))
+		// prometheusremotewrite doesn't support delta metrics so convert them to cumulative metrics
+		translators.Processors.Set(deltatocumulativeprocessor.NewTranslator(common.WithName(t.name)))
 		translators.Exporters.Set(prometheusremotewrite.NewTranslatorWithName(common.AMPKey))
 		translators.Extensions.Set(sigv4auth.NewTranslator())
 	case common.CloudWatchLogsKey:

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

@@ -307,7 +307,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineID: "metrics/host/amp",
 				receivers:  []string{"nop", "other"},
-				processors: []string{"rollup", "batch/host/amp"},
+				processors: []string{"rollup", "batch/host/amp", "deltatocumulative/host/amp"},
 				exporters:  []string{"prometheusremotewrite/amp"},
 				extensions: []string{"sigv4auth"},
 			},
@@ -322,7 +322,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineID: "metrics/host/amp",
 				receivers:  []string{"nop", "other"},
-				processors: []string{"batch/host/amp"},
+				processors: []string{"batch/host/amp", "deltatocumulative/host/amp"},
 				exporters:  []string{"prometheusremotewrite/amp"},
 				extensions: []string{"sigv4auth"},
 			},

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

@@ -19,6 +19,7 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/sigv4auth"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/batchprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/cumulativetodeltaprocessor"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/deltatocumulativeprocessor"
 	"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"
@@ -120,6 +121,8 @@ func (t *translator) Translate(conf *confmap.Conf) (*common.ComponentTranslators
 		if conf.IsSet(common.MetricsAggregationDimensionsKey) {
 			translators.Processors.Set(rollupprocessor.NewTranslator())
 		}
+		// prometheusremotewrite doesn't support delta metrics so convert them to cumulative metrics
+		translators.Processors.Set(deltatocumulativeprocessor.NewTranslator(common.WithName(t.name)))
 		translators.Exporters.Set(prometheusremotewrite.NewTranslatorWithName(common.AMPKey))
 		translators.Extensions.Set(sigv4auth.NewTranslator())
 	default:

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

@@ -208,7 +208,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineID: "metrics/jmx/amp",
 				receivers:  []string{"jmx"},
-				processors: []string{"filter/jmx", "resource/jmx", "batch/jmx/amp"},
+				processors: []string{"filter/jmx", "resource/jmx", "batch/jmx/amp", "deltatocumulative/jmx/amp"},
 				exporters:  []string{"prometheusremotewrite/amp"},
 				extensions: []string{"sigv4auth"},
 			},
@@ -238,7 +238,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineID: "metrics/jmx/amp",
 				receivers:  []string{"otlp/jmx"},
-				processors: []string{"filter/jmx", "metricstransform/jmx", "transform/jmx/drop", "batch/jmx/amp"},
+				processors: []string{"filter/jmx", "metricstransform/jmx", "transform/jmx/drop", "batch/jmx/amp", "deltatocumulative/jmx/amp"},
 				exporters:  []string{"prometheusremotewrite/amp"},
 				extensions: []string{"sigv4auth"},
 			},
@@ -338,7 +338,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineID: "metrics/jmx/amp/0",
 				receivers:  []string{"otlp/jmx"},
-				processors: []string{"filter/jmx/0", "metricstransform/jmx", "transform/jmx/drop", "transform/jmx/0", "batch/jmx/amp/0"},
+				processors: []string{"filter/jmx/0", "metricstransform/jmx", "transform/jmx/drop", "transform/jmx/0", "batch/jmx/amp/0", "deltatocumulative/jmx/amp/0"},
 				exporters:  []string{"prometheusremotewrite/amp"},
 				extensions: []string{"sigv4auth"},
 			},

translator/translate/otel/pipeline/prometheus/translator.go

@@ -17,6 +17,7 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/agenthealth"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/sigv4auth"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/batchprocessor"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/deltatocumulativeprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/rollupprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/receiver/adapter"
 	otelprom "github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/receiver/prometheus"
@@ -79,8 +80,12 @@ func (t *translator) Translate(conf *confmap.Conf) (*common.ComponentTranslators
 			return nil, fmt.Errorf("pipeline (%s) is missing prometheus configuration under metrics section with destination (%s)", t.name, t.Destination())
 		}
 		translators := &common.ComponentTranslators{
-			Receivers:  common.NewTranslatorMap(otelprom.NewTranslator()),
-			Processors: common.NewTranslatorMap(batchprocessor.NewTranslatorWithNameAndSection(t.name, common.MetricsKey)),
+			Receivers: common.NewTranslatorMap(otelprom.NewTranslator()),
+			Processors: common.NewTranslatorMap(
+				batchprocessor.NewTranslatorWithNameAndSection(t.name, common.MetricsKey),
+				// prometheusremotewrite doesn't support delta metrics so convert them to cumulative metrics
+				deltatocumulativeprocessor.NewTranslator(common.WithName(t.name)),
+			),
 			Exporters:  common.NewTranslatorMap(prometheusremotewrite.NewTranslatorWithName(common.AMPKey)),
 			Extensions: common.NewTranslatorMap(sigv4auth.NewTranslator()),
 		}

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

@@ -107,7 +107,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineID: "metrics/prometheus/amp",
 				receivers:  []string{"prometheus"},
-				processors: []string{"batch/prometheus/amp"},
+				processors: []string{"batch/prometheus/amp", "deltatocumulative/prometheus/amp"},
 				exporters:  []string{"prometheusremotewrite/amp"},
 				extensions: []string{"sigv4auth"},
 			},

translator/translate/otel/processor/deltatocumulativeprocessor/translator.go

@@ -0,0 +1,42 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package deltatocumulativeprocessor
+
+import (
+	"time"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap"
+	"go.opentelemetry.io/collector/processor"
+
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
+)
+
+type translator struct {
+	factory processor.Factory
+	common.NameProvider
+}
+
+var _ common.ComponentTranslator = (*translator)(nil)
+var _ common.NameSetter = (*translator)(nil)
+
+func NewTranslator(opts ...common.TranslatorOption) common.ComponentTranslator {
+	t := &translator{factory: deltatocumulativeprocessor.NewFactory()}
+	for _, opt := range opts {
+		opt(t)
+	}
+	return t
+}
+
+func (t *translator) ID() component.ID {
+	return component.NewIDWithName(t.factory.Type(), t.Name())
+}
+
+// Translate creates a processor config
+func (t *translator) Translate(_ *confmap.Conf) (component.Config, error) {
+	cfg := t.factory.CreateDefaultConfig().(*deltatocumulativeprocessor.Config)
+	cfg.MaxStale = 14 * 24 * time.Hour // two weeks
+	return cfg, nil
+}

translator/translate/otel/processor/deltatocumulativeprocessor/translator_test.go

@@ -0,0 +1,51 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package deltatocumulativeprocessor
+
+import (
+	"math"
+	"testing"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/confmap"
+
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
+)
+
+func TestTranslator(t *testing.T) {
+	dcpTranslator := NewTranslator(common.WithName("test"))
+	require.EqualValues(t, "deltatocumulative/test", dcpTranslator.ID().String())
+	testCases := map[string]struct {
+		input   map[string]any
+		want    map[string]any
+		wantErr error
+	}{
+		"EmptyConfig": {
+			input: map[string]any{},
+			want: map[string]any{
+				"max_stale":   1209600000000000, // 2 weeks, in minutes
+				"max_streams": math.MaxInt64,
+			},
+		},
+	}
+	factory := deltatocumulativeprocessor.NewFactory()
+	for name, testCase := range testCases {
+		t.Run(name, func(t *testing.T) {
+			conf := confmap.NewFromStringMap(testCase.input)
+			got, err := dcpTranslator.Translate(conf)
+			require.Equal(t, testCase.wantErr, err)
+			if err == nil {
+				require.NotNil(t, got)
+				gotCfg, ok := got.(*deltatocumulativeprocessor.Config)
+				require.True(t, ok)
+				wantCfg := factory.CreateDefaultConfig()
+				wantConf := confmap.NewFromStringMap(testCase.want)
+				require.NoError(t, wantConf.Unmarshal(&wantCfg))
+				assert.Equal(t, wantCfg, gotCfg)
+			}
+		})
+	}
+}