[processor/transform] Convert between sum and gauge in metric context

.chloggen/metric-conversion-context.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: processor/transform
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Convert between sum and gauge in metric context when alpha feature gate `processor.transform.ConvertBetweenSumAndGaugeMetricContext` enabled
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [20773]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

processor/transformprocessor/go.mod

@@ -3,11 +3,13 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/processor/trans
 go 1.20
 
 require (
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.89.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl v0.89.0
 	github.com/stretchr/testify v1.8.4
 	go.opentelemetry.io/collector/component v0.89.0
 	go.opentelemetry.io/collector/confmap v0.89.0
 	go.opentelemetry.io/collector/consumer v0.89.0
+	go.opentelemetry.io/collector/featuregate v1.0.0-rcv0018
 	go.opentelemetry.io/collector/pdata v1.0.0-rcv0018
 	go.opentelemetry.io/collector/processor v0.89.0
 	go.uber.org/multierr v1.11.0
@@ -37,7 +39,6 @@ require (
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/collector v0.89.0 // indirect
 	go.opentelemetry.io/collector/config/configtelemetry v0.89.0 // indirect
-	go.opentelemetry.io/collector/featuregate v1.0.0-rcv0018 // indirect
 	go.opentelemetry.io/otel v1.21.0 // indirect
 	go.opentelemetry.io/otel/metric v1.21.0 // indirect
 	go.opentelemetry.io/otel/trace v1.21.0 // indirect
@@ -66,3 +67,5 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest => ../../pkg/pdatatest
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden => ../../pkg/golden
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/common => ../../internal/common

processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum.go

@@ -10,19 +10,19 @@ import (
 	"go.opentelemetry.io/collector/pdata/pmetric"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
 )
 
 type convertGaugeToSumArguments struct {
 	StringAggTemp string
 	Monotonic     bool
 }
 
-func newConvertGaugeToSumFactory() ottl.Factory[ottldatapoint.TransformContext] {
+func newConvertGaugeToSumFactory() ottl.Factory[ottlmetric.TransformContext] {
 	return ottl.NewFactory("convert_gauge_to_sum", &convertGaugeToSumArguments{}, createConvertGaugeToSumFunction)
 }
 
-func createConvertGaugeToSumFunction(_ ottl.FunctionContext, oArgs ottl.Arguments) (ottl.ExprFunc[ottldatapoint.TransformContext], error) {
+func createConvertGaugeToSumFunction(_ ottl.FunctionContext, oArgs ottl.Arguments) (ottl.ExprFunc[ottlmetric.TransformContext], error) {
 	args, ok := oArgs.(*convertGaugeToSumArguments)
 
 	if !ok {
@@ -32,7 +32,7 @@ func createConvertGaugeToSumFunction(_ ottl.FunctionContext, oArgs ottl.Argument
 	return convertGaugeToSum(args.StringAggTemp, args.Monotonic)
 }
 
-func convertGaugeToSum(stringAggTemp string, monotonic bool) (ottl.ExprFunc[ottldatapoint.TransformContext], error) {
+func convertGaugeToSum(stringAggTemp string, monotonic bool) (ottl.ExprFunc[ottlmetric.TransformContext], error) {
 	var aggTemp pmetric.AggregationTemporality
 	switch stringAggTemp {
 	case "delta":
@@ -43,7 +43,7 @@ func convertGaugeToSum(stringAggTemp string, monotonic bool) (ottl.ExprFunc[ottl
 		return nil, fmt.Errorf("unknown aggregation temporality: %s", stringAggTemp)
 	}
 
-	return func(_ context.Context, tCtx ottldatapoint.TransformContext) (any, error) {
+	return func(_ context.Context, tCtx ottlmetric.TransformContext) (any, error) {
 		metric := tCtx.GetMetric()
 		if metric.Type() != pmetric.MetricTypeGauge {
 			return nil, nil

processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum_datapoint.go

@@ -0,0 +1,58 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package metrics // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/metrics"
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/collector/pdata/pmetric"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+)
+
+func newConvertDatapointGaugeToSumFactory() ottl.Factory[ottldatapoint.TransformContext] {
+	return ottl.NewFactory("convert_gauge_to_sum", &convertGaugeToSumArguments{}, createConvertDatapointGaugeToSumFunction)
+}
+
+func createConvertDatapointGaugeToSumFunction(_ ottl.FunctionContext, oArgs ottl.Arguments) (ottl.ExprFunc[ottldatapoint.TransformContext], error) {
+	// use the same args as in metric context
+	args, ok := oArgs.(*convertGaugeToSumArguments)
+
+	if !ok {
+		return nil, fmt.Errorf("ConvertGaugeToSumFactory args must be of type *ConvertGaugeToSumArguments")
+	}
+
+	return convertDatapointGaugeToSum(args.StringAggTemp, args.Monotonic)
+}
+
+func convertDatapointGaugeToSum(stringAggTemp string, monotonic bool) (ottl.ExprFunc[ottldatapoint.TransformContext], error) {
+	var aggTemp pmetric.AggregationTemporality
+	switch stringAggTemp {
+	case "delta":
+		aggTemp = pmetric.AggregationTemporalityDelta
+	case "cumulative":
+		aggTemp = pmetric.AggregationTemporalityCumulative
+	default:
+		return nil, fmt.Errorf("unknown aggregation temporality: %s", stringAggTemp)
+	}
+
+	return func(_ context.Context, tCtx ottldatapoint.TransformContext) (any, error) {
+		metric := tCtx.GetMetric()
+		if metric.Type() != pmetric.MetricTypeGauge {
+			return nil, nil
+		}
+
+		dps := metric.Gauge().DataPoints()
+
+		metric.SetEmptySum().SetAggregationTemporality(aggTemp)
+		metric.Sum().SetIsMonotonic(monotonic)
+
+		// Setting the data type removed all the data points, so we must copy them back to the metric.
+		dps.CopyTo(metric.Sum().DataPoints())
+
+		return nil, nil
+	}, nil
+}

processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum_datapoint_test.go

@@ -0,0 +1,149 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package metrics
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+)
+
+func Test_convertDatapointGaugeToSum(t *testing.T) {
+	gaugeInput := pmetric.NewMetric()
+
+	dp1 := gaugeInput.SetEmptyGauge().DataPoints().AppendEmpty()
+	dp1.SetIntValue(10)
+
+	dp2 := gaugeInput.Gauge().DataPoints().AppendEmpty()
+	dp2.SetDoubleValue(14.5)
+
+	sumInput := pmetric.NewMetric()
+	sumInput.SetEmptySum()
+
+	histogramInput := pmetric.NewMetric()
+	histogramInput.SetEmptyHistogram()
+
+	expoHistogramInput := pmetric.NewMetric()
+	expoHistogramInput.SetEmptyHistogram()
+
+	summaryInput := pmetric.NewMetric()
+	summaryInput.SetEmptySummary()
+
+	tests := []struct {
+		name          string
+		stringAggTemp string
+		monotonic     bool
+		input         pmetric.Metric
+		want          func(pmetric.Metric)
+	}{
+		{
+			name:          "convert gauge to cumulative sum",
+			stringAggTemp: "cumulative",
+			monotonic:     false,
+			input:         gaugeInput,
+			want: func(metric pmetric.Metric) {
+				gaugeInput.CopyTo(metric)
+
+				dps := gaugeInput.Gauge().DataPoints()
+
+				metric.SetEmptySum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
+				metric.Sum().SetIsMonotonic(false)
+
+				dps.CopyTo(metric.Sum().DataPoints())
+			},
+		},
+		{
+			name:          "convert gauge to delta sum",
+			stringAggTemp: "delta",
+			monotonic:     true,
+			input:         gaugeInput,
+			want: func(metric pmetric.Metric) {
+				gaugeInput.CopyTo(metric)
+
+				dps := gaugeInput.Gauge().DataPoints()
+
+				metric.SetEmptySum().SetAggregationTemporality(pmetric.AggregationTemporalityDelta)
+				metric.Sum().SetIsMonotonic(true)
+
+				dps.CopyTo(metric.Sum().DataPoints())
+			},
+		},
+		{
+			name:          "noop for sum",
+			stringAggTemp: "delta",
+			monotonic:     true,
+			input:         sumInput,
+			want: func(metric pmetric.Metric) {
+				sumInput.CopyTo(metric)
+			},
+		},
+		{
+			name:          "noop for histogram",
+			stringAggTemp: "delta",
+			monotonic:     true,
+			input:         histogramInput,
+			want: func(metric pmetric.Metric) {
+				histogramInput.CopyTo(metric)
+			},
+		},
+		{
+			name:          "noop for exponential histogram",
+			stringAggTemp: "delta",
+			monotonic:     true,
+			input:         expoHistogramInput,
+			want: func(metric pmetric.Metric) {
+				expoHistogramInput.CopyTo(metric)
+			},
+		},
+		{
+			name:          "noop for summary",
+			stringAggTemp: "delta",
+			monotonic:     true,
+			input:         summaryInput,
+			want: func(metric pmetric.Metric) {
+				summaryInput.CopyTo(metric)
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			metric := pmetric.NewMetric()
+			tt.input.CopyTo(metric)
+
+			ctx := ottldatapoint.NewTransformContext(pmetric.NewNumberDataPoint(), metric, pmetric.NewMetricSlice(), pcommon.NewInstrumentationScope(), pcommon.NewResource())
+
+			exprFunc, _ := convertDatapointGaugeToSum(tt.stringAggTemp, tt.monotonic)
+
+			_, err := exprFunc(nil, ctx)
+			assert.Nil(t, err)
+
+			expected := pmetric.NewMetric()
+			tt.want(expected)
+
+			assert.Equal(t, expected, metric)
+		})
+	}
+}
+
+func Test_convertDatapointGaugeToSum_validation(t *testing.T) {
+	tests := []struct {
+		name          string
+		stringAggTemp string
+	}{
+		{
+			name:          "invalid aggregation temporality",
+			stringAggTemp: "not a real aggregation temporality",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := convertDatapointGaugeToSum(tt.stringAggTemp, true)
+			assert.Error(t, err, "unknown aggregation temporality: not a real aggregation temporality")
+		})
+	}
+}

processor/transformprocessor/internal/metrics/func_convert_gauge_to_sum_test.go

@@ -10,7 +10,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
 )
 
 func Test_convertGaugeToSum(t *testing.T) {
@@ -115,7 +115,7 @@ func Test_convertGaugeToSum(t *testing.T) {
 			metric := pmetric.NewMetric()
 			tt.input.CopyTo(metric)
 
-			ctx := ottldatapoint.NewTransformContext(pmetric.NewNumberDataPoint(), metric, pmetric.NewMetricSlice(), pcommon.NewInstrumentationScope(), pcommon.NewResource())
+			ctx := ottlmetric.NewTransformContext(metric, pmetric.NewMetricSlice(), pcommon.NewInstrumentationScope(), pcommon.NewResource())
 
 			exprFunc, _ := convertGaugeToSum(tt.stringAggTemp, tt.monotonic)
 

processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge.go

@@ -9,19 +9,19 @@ import (
 	"go.opentelemetry.io/collector/pdata/pmetric"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
 )
 
-func newConvertSumToGaugeFactory() ottl.Factory[ottldatapoint.TransformContext] {
+func newConvertSumToGaugeFactory() ottl.Factory[ottlmetric.TransformContext] {
 	return ottl.NewFactory("convert_sum_to_gauge", nil, createConvertSumToGaugeFunction)
 }
 
-func createConvertSumToGaugeFunction(_ ottl.FunctionContext, _ ottl.Arguments) (ottl.ExprFunc[ottldatapoint.TransformContext], error) {
+func createConvertSumToGaugeFunction(_ ottl.FunctionContext, _ ottl.Arguments) (ottl.ExprFunc[ottlmetric.TransformContext], error) {
 	return convertSumToGauge()
 }
 
-func convertSumToGauge() (ottl.ExprFunc[ottldatapoint.TransformContext], error) {
-	return func(_ context.Context, tCtx ottldatapoint.TransformContext) (any, error) {
+func convertSumToGauge() (ottl.ExprFunc[ottlmetric.TransformContext], error) {
+	return func(_ context.Context, tCtx ottlmetric.TransformContext) (any, error) {
 		metric := tCtx.GetMetric()
 		if metric.Type() != pmetric.MetricTypeSum {
 			return nil, nil

processor/transformprocessor/internal/metrics/func_convert_sum_to_gauge_datapoint.go

@@ -0,0 +1,37 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package metrics // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/metrics"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/pdata/pmetric"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+)
+
+func newConvertDatapointSumToGaugeFactory() ottl.Factory[ottldatapoint.TransformContext] {
+	return ottl.NewFactory("convert_sum_to_gauge", nil, createDatapointConvertSumToGaugeFunction)
+}
+
+func createDatapointConvertSumToGaugeFunction(_ ottl.FunctionContext, _ ottl.Arguments) (ottl.ExprFunc[ottldatapoint.TransformContext], error) {
+	return convertDatapointSumToGauge()
+}
+
+func convertDatapointSumToGauge() (ottl.ExprFunc[ottldatapoint.TransformContext], error) {
+	return func(_ context.Context, tCtx ottldatapoint.TransformContext) (any, error) {
+		metric := tCtx.GetMetric()
+		if metric.Type() != pmetric.MetricTypeSum {
+			return nil, nil
+		}
+
+		dps := metric.Sum().DataPoints()
+
+		// Setting the data type removed all the data points, so we must copy them back to the metric.
+		dps.CopyTo(metric.SetEmptyGauge().DataPoints())
+
+		return nil, nil
+	}, nil
+}