[pkg/ottl] Add support for scaling values

.chloggen/add-scale-function.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: pkg/ottl
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Added support for scaling values
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [16214]
+
+# (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: []

pkg/ottl/e2e/e2e_test.go

@@ -12,10 +12,13 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottllog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
 )
 
 var (
@@ -54,6 +57,7 @@ func Test_e2e_editors(t *testing.T) {
 				tCtx.GetLogRecord().Attributes().Remove("flags")
 				tCtx.GetLogRecord().Attributes().Remove("total.string")
 				tCtx.GetLogRecord().Attributes().Remove("foo")
+				tCtx.GetLogRecord().Attributes().Remove("double_value")
 			},
 		},
 		{
@@ -81,6 +85,7 @@ func Test_e2e_editors(t *testing.T) {
 				m.PutStr("test.foo.flags", "pass")
 				m.PutStr("test.foo.slice.0", "val")
 				m.PutStr("test.foo.nested.test", "pass")
+				m.PutDouble("test.double_value", 10.5)
 				m.CopyTo(tCtx.GetLogRecord().Attributes())
 			},
 		},
@@ -102,6 +107,7 @@ func Test_e2e_editors(t *testing.T) {
 				m.PutStr("foo.bar", "pass")
 				m.PutStr("foo.flags", "pass")
 				m.PutStr("foo.slice.0", "val")
+				m.PutDouble("double_value", 10.5)
 				m2 := m.PutEmptyMap("foo.nested")
 				m2.PutStr("test", "pass")
 				m.CopyTo(tCtx.GetLogRecord().Attributes())
@@ -114,6 +120,7 @@ func Test_e2e_editors(t *testing.T) {
 				tCtx.GetLogRecord().Attributes().Remove("http.path")
 				tCtx.GetLogRecord().Attributes().Remove("http.url")
 				tCtx.GetLogRecord().Attributes().Remove("foo")
+				tCtx.GetLogRecord().Attributes().Remove("double_value")
 			},
 		},
 		{
@@ -128,6 +135,7 @@ func Test_e2e_editors(t *testing.T) {
 				tCtx.GetLogRecord().Attributes().Remove("http.url")
 				tCtx.GetLogRecord().Attributes().Remove("flags")
 				tCtx.GetLogRecord().Attributes().Remove("foo")
+				tCtx.GetLogRecord().Attributes().Remove("double_value")
 			},
 		},
 		{
@@ -297,6 +305,38 @@ func Test_e2e_editors(t *testing.T) {
 	}
 }
 
+func Test_e2e_metricEditors(t *testing.T) {
+	tests := []struct {
+		statement string
+		want      func(tCtx ottlmetric.TransformContext)
+	}{
+		{
+			statement: `scale_metric(data_points,0.1)`,
+			want: func(tCtx ottlmetric.TransformContext) {
+				tCtx.GetMetric().Gauge().DataPoints().At(0).SetDoubleValue(1.0)
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.statement, func(t *testing.T) {
+			settings := componenttest.NewNopTelemetrySettings()
+			metricParser, err := ottlmetric.NewParser(ottlfuncs.StandardFuncs[ottlmetric.TransformContext](), settings)
+			assert.NoError(t, err)
+			logStatements, err := metricParser.ParseStatement(tt.statement)
+			assert.NoError(t, err)
+
+			tCtx := constructMetricTransformContext()
+			_, _, _ = logStatements.Execute(context.Background(), tCtx)
+
+			exTCtx := constructMetricTransformContext()
+			tt.want(exTCtx)
+
+			assert.NoError(t, pmetrictest.CompareResourceMetrics(newResourceMetrics(exTCtx), newResourceMetrics(tCtx)))
+		})
+	}
+}
+
 func Test_e2e_converters(t *testing.T) {
 	tests := []struct {
 		statement string
@@ -816,6 +856,19 @@ func Test_e2e_ottl_features(t *testing.T) {
 	}
 }
 
+func constructMetricTransformContext() ottlmetric.TransformContext {
+	resource := pcommon.NewResource()
+	resource.Attributes().PutStr("host.name", "localhost")
+
+	scope := pcommon.NewInstrumentationScope()
+	scope.SetName("scope")
+
+	metric := pmetric.NewMetric()
+	metric.SetEmptyGauge().DataPoints().AppendEmpty().SetDoubleValue(10.0)
+
+	return ottlmetric.NewTransformContext(metric, pmetric.NewMetricSlice(), scope, resource)
+}
+
 func constructLogTransformContext() ottllog.TransformContext {
 	resource := pcommon.NewResource()
 	resource.Attributes().PutStr("host.name", "localhost")
@@ -837,6 +890,7 @@ func constructLogTransformContext() ottllog.TransformContext {
 	logRecord.Attributes().PutStr("http.url", "http://localhost/health")
 	logRecord.Attributes().PutStr("flags", "A|B|C")
 	logRecord.Attributes().PutStr("total.string", "123456789")
+	logRecord.Attributes().PutDouble("double_value", 10.5)
 	m := logRecord.Attributes().PutEmptyMap("foo")
 	m.PutStr("bar", "pass")
 	m.PutStr("flags", "pass")
@@ -858,3 +912,13 @@ func newResourceLogs(tCtx ottllog.TransformContext) plog.ResourceLogs {
 	tCtx.GetLogRecord().CopyTo(l)
 	return rl
 }
+
+func newResourceMetrics(tCtx ottlmetric.TransformContext) pmetric.ResourceMetrics {
+	rl := pmetric.NewResourceMetrics()
+	tCtx.GetResource().CopyTo(rl.Resource())
+	sl := rl.ScopeMetrics().AppendEmpty()
+	tCtx.GetInstrumentationScope().CopyTo(sl.Scope())
+	l := sl.Metrics().AppendEmpty()
+	tCtx.GetMetric().CopyTo(l)
+	return rl
+}

pkg/ottl/ottlfuncs/README.md

@@ -59,6 +59,7 @@ Available Editors:
 - [replace_pattern](#replace_pattern)
 - [set](#set)
 - [truncate_all](#truncate_all)
+- [scale_metric](#scale_metric)
 
 ### append
 
@@ -402,6 +403,22 @@ Examples:
 
 - `truncate_all(resource.attributes, 50)`
 
+### scale_metric
+
+`scale_metric(value, factor)`
+
+The `Scale` function multiplies the original `value` by the `factor`.
+The supported data types are:
+
+- `data_points` - Supported metric types are `Gauge`, `Sum` and `Histogram`.
+  To scale a metric of these types, the `data_points` property of the respective metric needs to be passed to the function,
+  as indicated in the examples below.
+
+Examples:
+
+- `scale_metric(10.0, 0.1)`: Trivial example
+- `scale_metric(data_points, 10.0)`: Modifies the metric's `data_points` by multiplying them with the factor `10.0`.
+
 ## Converters
 
 Converters are pure functions that take OTTL values as input and output a single value for use within a statement.

pkg/ottl/ottlfuncs/func_scale.go

@@ -0,0 +1,124 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package ottlfuncs // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs"
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"go.opentelemetry.io/collector/pdata/pmetric"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+)
+
+type ScaleArguments[K any] struct {
+	Value      ottl.GetSetter[K]
+	Multiplier float64
+}
+
+func NewScaleFactory[K any]() ottl.Factory[K] {
+	return ottl.NewFactory("scale_metric", &ScaleArguments[K]{}, createScaleFunction[K])
+}
+
+func createScaleFunction[K any](_ ottl.FunctionContext, oArgs ottl.Arguments) (ottl.ExprFunc[K], error) {
+	args, ok := oArgs.(*ScaleArguments[K])
+
+	if !ok {
+		return nil, fmt.Errorf("ScaleFactory args must be of type *ScaleArguments[K]")
+	}
+
+	return Scale(args.Value, args.Multiplier)
+}
+
+func Scale[K any](getSetter ottl.GetSetter[K], multiplier float64) (ottl.ExprFunc[K], error) {
+	return func(ctx context.Context, tCtx K) (any, error) {
+		got, err := getSetter.Get(ctx, tCtx)
+		if err != nil {
+			return nil, err
+		}
+
+		switch value := got.(type) {
+		case pmetric.NumberDataPointSlice:
+			scaleMetric(value, multiplier)
+			return nil, nil
+		case pmetric.HistogramDataPointSlice:
+			scaleHistogram(value, multiplier)
+			return nil, nil
+		case pmetric.SummaryDataPointValueAtQuantileSlice:
+			scaleSummaryDataPointValueAtQuantileSlice(value, multiplier)
+			return nil, nil
+		case pmetric.ExemplarSlice:
+			scaleExemplarSlice(value, multiplier)
+			return nil, nil
+		case pmetric.ExponentialHistogramDataPointSlice:
+			return nil, errors.New("exponential histograms are not supported by the 'scale_metric' function")
+		default:
+			return nil, fmt.Errorf("unsupported data type: '%T'", value)
+		}
+	}, nil
+}
+
+func scaleExemplarSlice(values pmetric.ExemplarSlice, multiplier float64) {
+	for i := 0; i < values.Len(); i++ {
+		ex := values.At(i)
+		scaleExemplar(&ex, multiplier)
+	}
+}
+
+func scaleExemplar(ex *pmetric.Exemplar, multiplier float64) {
+	switch ex.ValueType() {
+	case pmetric.ExemplarValueTypeInt:
+		ex.SetIntValue(int64(float64(ex.IntValue()) * multiplier))
+	case pmetric.ExemplarValueTypeDouble:
+		ex.SetDoubleValue(ex.DoubleValue() * multiplier)
+	}
+}
+
+func scaleSummaryDataPointValueAtQuantileSlice(values pmetric.SummaryDataPointValueAtQuantileSlice, multiplier float64) {
+	for i := 0; i < values.Len(); i++ {
+		dp := values.At(i)
+
+		dp.SetValue(dp.Value() * multiplier)
+	}
+}
+
+func scaleHistogram(datapoints pmetric.HistogramDataPointSlice, multiplier float64) {
+	for i := 0; i < datapoints.Len(); i++ {
+		dp := datapoints.At(i)
+
+		if dp.HasSum() {
+			dp.SetSum(dp.Sum() * multiplier)
+		}
+		if dp.HasMin() {
+			dp.SetMin(dp.Min() * multiplier)
+		}
+		if dp.HasMax() {
+			dp.SetMax(dp.Max() * multiplier)
+		}
+
+		for bounds, bi := dp.ExplicitBounds(), 0; bi < bounds.Len(); bi++ {
+			bounds.SetAt(bi, bounds.At(bi)*multiplier)
+		}
+
+		for exemplars, ei := dp.Exemplars(), 0; ei < exemplars.Len(); ei++ {
+			exemplar := exemplars.At(ei)
+			scaleExemplar(&exemplar, multiplier)
+		}
+	}
+}
+
+func scaleMetric(points pmetric.NumberDataPointSlice, multiplier float64) {
+	for i := 0; i < points.Len(); i++ {
+		dp := points.At(i)
+		switch dp.ValueType() {
+		case pmetric.NumberDataPointValueTypeInt:
+			dp.SetIntValue(int64(float64(dp.IntValue()) * multiplier))
+
+		case pmetric.NumberDataPointValueTypeDouble:
+			dp.SetDoubleValue(dp.DoubleValue() * multiplier)
+		default:
+		}
+	}
+}