cumulative_to_delta_processor: Add Cumulative to Delta Conversion logic

processor/cumulativetodeltaprocessor/go.mod

@@ -3,8 +3,11 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/processor/cumul
 go 1.16
 
 require (
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/metrics v0.0.0-00010101000000-000000000000
 	github.com/stretchr/testify v1.7.0
 	go.opentelemetry.io/collector v0.30.2-0.20210727185145-88b2935343aa
 	go.opentelemetry.io/collector/model v0.30.2-0.20210727185145-88b2935343aa
 	go.uber.org/zap v1.18.1
 )
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/metrics => ./../../internal/aws/metrics

processor/cumulativetodeltaprocessor/processor.go

@@ -16,35 +16,97 @@ package cumulativetodeltaprocessor
 
 import (
 	"context"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/model/pdata"
 	"go.uber.org/zap"
+
+	awsmetrics "github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/metrics"
 )
 
 type cumulativeToDeltaProcessor struct {
-	metrics []string
-	logger  *zap.Logger
+	metrics         map[string]bool
+	logger          *zap.Logger
+	deltaCalculator awsmetrics.MetricCalculator
 }
 
 func newCumulativeToDeltaProcessor(config *Config, logger *zap.Logger) *cumulativeToDeltaProcessor {
+	inputMetricSet := make(map[string]bool, len(config.Metrics))
+	for _, name := range config.Metrics {
+		inputMetricSet[name] = true
+	}
+
 	return &cumulativeToDeltaProcessor{
-		metrics: config.Metrics,
-		logger:  logger,
+		metrics:         inputMetricSet,
+		logger:          logger,
+		deltaCalculator: newDeltaCalculator(),
 	}
 }
 
 // Start is invoked during service startup.
-func (mgp *cumulativeToDeltaProcessor) Start(context.Context, component.Host) error {
+func (ctdp *cumulativeToDeltaProcessor) Start(context.Context, component.Host) error {
 	return nil
 }
 
 // processMetrics implements the ProcessMetricsFunc type.
-func (mgp *cumulativeToDeltaProcessor) processMetrics(_ context.Context, md pdata.Metrics) (pdata.Metrics, error) {
+func (ctdp *cumulativeToDeltaProcessor) processMetrics(_ context.Context, md pdata.Metrics) (pdata.Metrics, error) {
+	resourceMetricsSlice := md.ResourceMetrics()
+	for i := 0; i < resourceMetricsSlice.Len(); i++ {
+		rm := resourceMetricsSlice.At(i)
+		ilms := rm.InstrumentationLibraryMetrics()
+		for j := 0; j < ilms.Len(); j++ {
+			ilm := ilms.At(j)
+			metricSlice := ilm.Metrics()
+			for k := 0; k < metricSlice.Len(); k++ {
+				metric := metricSlice.At(k)
+				if ctdp.metrics[metric.Name()] {
+					if metric.DataType() == pdata.MetricDataTypeSum && metric.Sum().AggregationTemporality() == pdata.AggregationTemporalityCumulative {
+						dataPoints := metric.Sum().DataPoints()
+
+						for l := 0; l < dataPoints.Len(); l++ {
+							fromDataPoint := dataPoints.At(l)
+							labelMap := make(map[string]string)
+
+							fromDataPoint.LabelsMap().Range(func(k string, v string) bool {
+								labelMap[k] = v
+								return true
+							})
+
+							result, _ := ctdp.deltaCalculator.Calculate(metric.Name(), labelMap, fromDataPoint.Value(), fromDataPoint.Timestamp().AsTime())
+
+							fromDataPoint.SetValue(result.(delta).value)
+							fromDataPoint.SetStartTimestamp(pdata.TimestampFromTime(result.(delta).prevTimestamp))
+						}
+						metric.Sum().SetAggregationTemporality(pdata.AggregationTemporalityDelta)
+					}
+				}
+			}
+		}
+	}
 	return md, nil
 }
 
 // Shutdown is invoked during service shutdown.
-func (mgp *cumulativeToDeltaProcessor) Shutdown(context.Context) error {
+func (ctdp *cumulativeToDeltaProcessor) Shutdown(context.Context) error {
 	return nil
 }
+
+func newDeltaCalculator() awsmetrics.MetricCalculator {
+	return awsmetrics.NewMetricCalculator(func(prev *awsmetrics.MetricValue, val interface{}, timestamp time.Time) (interface{}, bool) {
+		result := delta{value: val.(float64), prevTimestamp: timestamp}
+
+		if prev != nil {
+			deltaValue := val.(float64) - prev.RawValue.(float64)
+			result.value = deltaValue
+			result.prevTimestamp = prev.Timestamp
+			return result, true
+		}
+		return result, false
+	})
+}
+
+type delta struct {
+	value         float64
+	prevTimestamp time.Time
+}

processor/cumulativetodeltaprocessor/processor_test.go

@@ -0,0 +1,175 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cumulativetodeltaprocessor
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/model/pdata"
+)
+
+type testMetric struct {
+	metricNames  []string
+	metricValues [][]float64
+	isCumulative []bool
+}
+
+type cumulativeToDeltaTest struct {
+	name       string
+	metrics    []string
+	inMetrics  pdata.Metrics
+	outMetrics pdata.Metrics
+}
+
+var (
+	testCases = []cumulativeToDeltaTest{
+		{
+			name:    "cumulative_to_delta_expect_same",
+			metrics: nil,
+			inMetrics: generateTestMetrics(testMetric{
+				metricNames:  []string{"metric_1", "metric_2"},
+				metricValues: [][]float64{{100}, {4}},
+				isCumulative: []bool{true, true},
+			}),
+			outMetrics: generateTestMetrics(testMetric{
+				metricNames:  []string{"metric_1", "metric_2"},
+				metricValues: [][]float64{{100}, {4}},
+				isCumulative: []bool{true, true},
+			}),
+		},
+		{
+			name:    "cumulative_to_delta_one_positive",
+			metrics: []string{"metric_1"},
+			inMetrics: generateTestMetrics(testMetric{
+				metricNames:  []string{"metric_1", "metric_2"},
+				metricValues: [][]float64{{100, 200, 500}, {4}},
+				isCumulative: []bool{true, true},
+			}),
+			outMetrics: generateTestMetrics(testMetric{
+				metricNames:  []string{"metric_1", "metric_2"},
+				metricValues: [][]float64{{100, 100, 300}, {4}},
+				isCumulative: []bool{false, true},
+			}),
+		},
+	}
+)
+
+func TestCumulativeToDeltaProcessor(t *testing.T) {
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+			// next stores the results of the filter metric processor
+			next := new(consumertest.MetricsSink)
+			cfg := &Config{
+				ProcessorSettings: config.NewProcessorSettings(config.NewID(typeStr)),
+				Metrics:           test.metrics,
+			}
+			factory := NewFactory()
+			mgp, err := factory.CreateMetricsProcessor(
+				context.Background(),
+				componenttest.NewNopProcessorCreateSettings(),
+				cfg,
+				next,
+			)
+			assert.NotNil(t, mgp)
+			assert.Nil(t, err)
+
+			caps := mgp.Capabilities()
+			assert.True(t, caps.MutatesData)
+			ctx := context.Background()
+			require.NoError(t, mgp.Start(ctx, nil))
+
+			cErr := mgp.ConsumeMetrics(context.Background(), test.inMetrics)
+			assert.Nil(t, cErr)
+			got := next.AllMetrics()
+
+			require.Equal(t, 1, len(got))
+			require.Equal(t, test.outMetrics.ResourceMetrics().Len(), got[0].ResourceMetrics().Len())
+
+			expectedMetrics := test.outMetrics.ResourceMetrics().At(0).InstrumentationLibraryMetrics().At(0).Metrics()
+			actualMetrics := got[0].ResourceMetrics().At(0).InstrumentationLibraryMetrics().At(0).Metrics()
+
+			require.Equal(t, expectedMetrics.Len(), actualMetrics.Len())
+
+			for i := 0; i < expectedMetrics.Len(); i++ {
+				eM := expectedMetrics.At(i)
+				aM := actualMetrics.At(i)
+
+				require.Equal(t, eM.Name(), aM.Name())
+
+				if eM.DataType() == pdata.MetricDataTypeGauge {
+					eDataPoints := eM.Gauge().DataPoints()
+					aDataPoints := aM.Gauge().DataPoints()
+					require.Equal(t, eDataPoints.Len(), aDataPoints.Len())
+
+					for j := 0; j < eDataPoints.Len(); j++ {
+						require.Equal(t, eDataPoints.At(j).Value(), aDataPoints.At(j).Value())
+					}
+				}
+
+				if eM.DataType() == pdata.MetricDataTypeSum {
+					eDataPoints := eM.Sum().DataPoints()
+					aDataPoints := aM.Sum().DataPoints()
+
+					require.Equal(t, eDataPoints.Len(), aDataPoints.Len())
+					require.Equal(t, eM.Sum().AggregationTemporality(), aM.Sum().AggregationTemporality())
+
+					for j := 0; j < eDataPoints.Len(); j++ {
+						require.Equal(t, eDataPoints.At(j).Value(), aDataPoints.At(j).Value())
+					}
+				}
+
+			}
+
+			require.NoError(t, mgp.Shutdown(ctx))
+		})
+	}
+}
+
+func generateTestMetrics(tm testMetric) pdata.Metrics {
+	md := pdata.NewMetrics()
+	now := time.Now()
+
+	rm := md.ResourceMetrics().AppendEmpty()
+	ms := rm.InstrumentationLibraryMetrics().AppendEmpty().Metrics()
+	for i, name := range tm.metricNames {
+		m := ms.AppendEmpty()
+		m.SetName(name)
+		m.SetDataType(pdata.MetricDataTypeSum)
+
+		sum := m.Sum()
+		sum.SetIsMonotonic(true)
+
+		if tm.isCumulative[i] {
+			sum.SetAggregationTemporality(pdata.AggregationTemporalityCumulative)
+		} else {
+			sum.SetAggregationTemporality(pdata.AggregationTemporalityDelta)
+		}
+
+		for _, value := range tm.metricValues[i] {
+			dp := m.Sum().DataPoints().AppendEmpty()
+			dp.SetTimestamp(pdata.TimestampFromTime(now.Add(10 * time.Second)))
+			dp.SetValue(value)
+		}
+	}
+
+	return md
+}