[processor/transform] Allow specifying functions per context

.chloggen/tp-functions-per-context.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# 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: Fix issue where the metric context was using datapoint functions.
+
+# One or more tracking issues related to the change
+issues: [16251]
+
+# (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:

processor/transformprocessor/config.go

@@ -60,15 +60,15 @@ func (c *Config) Validate() error {
 	}
 
 	if len(c.Traces.Statements) > 0 {
-		ottlspanp := ottlspan.NewParser(traces.Functions(), component.TelemetrySettings{Logger: zap.NewNop()})
+		ottlspanp := ottlspan.NewParser(traces.SpanFunctions(), component.TelemetrySettings{Logger: zap.NewNop()})
 		_, err := ottlspanp.ParseStatements(c.Traces.Statements)
 		if err != nil {
 			return err
 		}
 	}
 
 	if len(c.TraceStatements) > 0 {
-		pc, err := common.NewTraceParserCollection(traces.Functions(), component.TelemetrySettings{Logger: zap.NewNop()})
+		pc, err := common.NewTraceParserCollection(component.TelemetrySettings{Logger: zap.NewNop()}, common.WithSpanParser(traces.SpanFunctions()), common.WithSpanEventParser(traces.SpanEventFunctions()))
 		if err != nil {
 			return err
 		}
@@ -81,15 +81,15 @@ func (c *Config) Validate() error {
 	}
 
 	if len(c.Metrics.Statements) > 0 {
-		ottldatapointp := ottldatapoint.NewParser(metrics.Functions(), component.TelemetrySettings{Logger: zap.NewNop()})
-		_, err := ottldatapointp.ParseStatements(c.Metrics.Statements)
+		ottlmetricsp := ottldatapoint.NewParser(metrics.DataPointFunctions(), component.TelemetrySettings{Logger: zap.NewNop()})
+		_, err := ottlmetricsp.ParseStatements(c.Metrics.Statements)
 		if err != nil {
 			return err
 		}
 	}
 
 	if len(c.MetricStatements) > 0 {
-		pc, err := common.NewMetricParserCollection(metrics.Functions(), component.TelemetrySettings{Logger: zap.NewNop()})
+		pc, err := common.NewMetricParserCollection(component.TelemetrySettings{Logger: zap.NewNop()}, common.WithMetricParser(metrics.MetricFunctions()), common.WithDataPointParser(metrics.DataPointFunctions()))
 		if err != nil {
 			return err
 		}
@@ -102,15 +102,15 @@ func (c *Config) Validate() error {
 	}
 
 	if len(c.Logs.Statements) > 0 {
-		ottllogsp := ottllog.NewParser(logs.Functions(), component.TelemetrySettings{Logger: zap.NewNop()})
+		ottllogsp := ottllog.NewParser(logs.LogFunctions(), component.TelemetrySettings{Logger: zap.NewNop()})
 		_, err := ottllogsp.ParseStatements(c.Logs.Statements)
 		if err != nil {
 			return err
 		}
 	}
 
 	if len(c.LogStatements) > 0 {
-		pc, err := common.NewLogParserCollection(logs.Functions(), component.TelemetrySettings{Logger: zap.NewNop()})
+		pc, err := common.NewLogParserCollection(component.TelemetrySettings{Logger: zap.NewNop()}, common.WithLogParser(logs.LogFunctions()))
 		if err != nil {
 			return err
 		}

processor/transformprocessor/internal/common/logs.go

@@ -64,14 +64,20 @@ type LogParserCollection struct {
 
 type LogParserCollectionOption func(*LogParserCollection) error
 
-func NewLogParserCollection(functions map[string]interface{}, settings component.TelemetrySettings, options ...LogParserCollectionOption) (*LogParserCollection, error) {
+func WithLogParser(functions map[string]interface{}) LogParserCollectionOption {
+	return func(lp *LogParserCollection) error {
+		lp.logParser = ottllog.NewParser(functions, lp.settings)
+		return nil
+	}
+}
+
+func NewLogParserCollection(settings component.TelemetrySettings, options ...LogParserCollectionOption) (*LogParserCollection, error) {
 	lpc := &LogParserCollection{
 		parserCollection: parserCollection{
 			settings:       settings,
 			resourceParser: ottlresource.NewParser(ResourceFunctions(), settings),
 			scopeParser:    ottlscope.NewParser(ScopeFunctions(), settings),
 		},
-		logParser: ottllog.NewParser(functions, settings),
 	}
 
 	for _, op := range options {

processor/transformprocessor/internal/common/metrics.go

@@ -161,15 +161,27 @@ type MetricParserCollection struct {
 
 type MetricParserCollectionOption func(*MetricParserCollection) error
 
-func NewMetricParserCollection(functions map[string]interface{}, settings component.TelemetrySettings, options ...MetricParserCollectionOption) (*MetricParserCollection, error) {
+func WithMetricParser(functions map[string]interface{}) MetricParserCollectionOption {
+	return func(mp *MetricParserCollection) error {
+		mp.metricParser = ottlmetric.NewParser(functions, mp.settings)
+		return nil
+	}
+}
+
+func WithDataPointParser(functions map[string]interface{}) MetricParserCollectionOption {
+	return func(mp *MetricParserCollection) error {
+		mp.dataPointParser = ottldatapoint.NewParser(functions, mp.settings)
+		return nil
+	}
+}
+
+func NewMetricParserCollection(settings component.TelemetrySettings, options ...MetricParserCollectionOption) (*MetricParserCollection, error) {
 	mpc := &MetricParserCollection{
 		parserCollection: parserCollection{
 			settings:       settings,
 			resourceParser: ottlresource.NewParser(ResourceFunctions(), settings),
 			scopeParser:    ottlscope.NewParser(ScopeFunctions(), settings),
 		},
-		metricParser:    ottlmetric.NewParser(functions, settings),
-		dataPointParser: ottldatapoint.NewParser(functions, settings),
 	}
 
 	for _, op := range options {

processor/transformprocessor/internal/common/traces.go

@@ -100,14 +100,27 @@ type TraceParserCollection struct {
 
 type TraceParserCollectionOption func(*TraceParserCollection) error
 
-func NewTraceParserCollection(functions map[string]interface{}, settings component.TelemetrySettings, options ...TraceParserCollectionOption) (*TraceParserCollection, error) {
+func WithSpanParser(functions map[string]interface{}) TraceParserCollectionOption {
+	return func(tp *TraceParserCollection) error {
+		tp.spanParser = ottlspan.NewParser(functions, tp.settings)
+		return nil
+	}
+}
+
+func WithSpanEventParser(functions map[string]interface{}) TraceParserCollectionOption {
+	return func(tp *TraceParserCollection) error {
+		tp.spanEventParser = ottlspanevent.NewParser(functions, tp.settings)
+		return nil
+	}
+}
+
+func NewTraceParserCollection(settings component.TelemetrySettings, options ...TraceParserCollectionOption) (*TraceParserCollection, error) {
 	tpc := &TraceParserCollection{
 		parserCollection: parserCollection{
 			settings:       settings,
 			resourceParser: ottlresource.NewParser(ResourceFunctions(), settings),
 			scopeParser:    ottlscope.NewParser(ScopeFunctions(), settings),
 		},
-		spanParser: ottlspan.NewParser(functions, settings),
 	}
 
 	for _, op := range options {

processor/transformprocessor/internal/logs/functions.go

@@ -19,7 +19,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/common"
 )
 
-func Functions() map[string]interface{} {
+func LogFunctions() map[string]interface{} {
 	// No logs-only functions yet.
 	return common.Functions[ottllog.TransformContext]()
 }

processor/transformprocessor/internal/logs/functions_test.go

@@ -24,9 +24,9 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/common"
 )
 
-func Test_DefaultFunctions(t *testing.T) {
+func Test_LogFunctions(t *testing.T) {
 	expected := common.Functions[ottllog.TransformContext]()
-	actual := Functions()
+	actual := LogFunctions()
 	require.Equal(t, len(expected), len(actual))
 	for k := range actual {
 		assert.Contains(t, expected, k)

processor/transformprocessor/internal/logs/processor.go

@@ -34,7 +34,7 @@ type Processor struct {
 
 func NewProcessor(statements []string, contextStatements []common.ContextStatements, settings component.TelemetrySettings) (*Processor, error) {
 	if len(statements) > 0 {
-		ottlp := ottllog.NewParser(Functions(), settings)
+		ottlp := ottllog.NewParser(LogFunctions(), settings)
 		parsedStatements, err := ottlp.ParseStatements(statements)
 		if err != nil {
 			return nil, err
@@ -44,7 +44,7 @@ func NewProcessor(statements []string, contextStatements []common.ContextStateme
 		}, nil
 	}
 
-	pc, err := common.NewLogParserCollection(Functions(), settings)
+	pc, err := common.NewLogParserCollection(settings, common.WithLogParser(LogFunctions()))
 	if err != nil {
 		return nil, err
 	}

processor/transformprocessor/internal/metrics/functions.go

@@ -16,11 +16,12 @@ package metrics // import "github.com/open-telemetry/opentelemetry-collector-con
 
 import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/common"
 )
 
 // registry is a map of names to functions for metrics pipelines
-var registry = map[string]interface{}{
+var datapointRegistry = map[string]interface{}{
 	"convert_sum_to_gauge":             convertSumToGauge,
 	"convert_gauge_to_sum":             convertGaugeToSum,
 	"convert_summary_sum_val_to_sum":   convertSummarySumValToSum,
@@ -30,10 +31,14 @@ var registry = map[string]interface{}{
 func init() {
 	// Init metrics registry with default functions common to all signals
 	for k, v := range common.Functions[ottldatapoint.TransformContext]() {
-		registry[k] = v
+		datapointRegistry[k] = v
 	}
 }
 
-func Functions() map[string]interface{} {
-	return registry
+func DataPointFunctions() map[string]interface{} {
+	return datapointRegistry
+}
+
+func MetricFunctions() map[string]interface{} {
+	return common.Functions[ottlmetric.TransformContext]()
 }

processor/transformprocessor/internal/metrics/functions_test.go

@@ -21,20 +21,30 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/common"
 )
 
-func Test_DefaultFunctions(t *testing.T) {
+func Test_DataPointFunctions(t *testing.T) {
 	expected := common.Functions[ottldatapoint.TransformContext]()
 	expected["convert_sum_to_gauge"] = convertSumToGauge
 	expected["convert_gauge_to_sum"] = convertGaugeToSum
 	expected["convert_summary_sum_val_to_sum"] = convertSummarySumValToSum
 	expected["convert_summary_count_val_to_sum"] = convertSummaryCountValToSum
 
-	actual := Functions()
+	actual := DataPointFunctions()
 
 	require.Equal(t, len(expected), len(actual))
 	for k := range actual {
 		assert.Contains(t, expected, k)
 	}
 }
+
+func Test_MetricFunctions(t *testing.T) {
+	expected := common.Functions[ottlmetric.TransformContext]()
+	actual := MetricFunctions()
+	require.Equal(t, len(expected), len(actual))
+	for k := range actual {
+		assert.Contains(t, expected, k)
+	}
+}

processor/transformprocessor/internal/metrics/processor.go

@@ -35,7 +35,7 @@ type Processor struct {
 
 func NewProcessor(statements []string, contextStatements []common.ContextStatements, settings component.TelemetrySettings) (*Processor, error) {
 	if len(statements) > 0 {
-		ottlp := ottldatapoint.NewParser(Functions(), settings)
+		ottlp := ottldatapoint.NewParser(DataPointFunctions(), settings)
 		parsedStatements, err := ottlp.ParseStatements(statements)
 		if err != nil {
 			return nil, err
@@ -45,7 +45,7 @@ func NewProcessor(statements []string, contextStatements []common.ContextStateme
 		}, nil
 	}
 
-	pc, err := common.NewMetricParserCollection(Functions(), settings)
+	pc, err := common.NewMetricParserCollection(settings, common.WithMetricParser(MetricFunctions()), common.WithDataPointParser(DataPointFunctions()))
 	if err != nil {
 		return nil, err
 	}

processor/transformprocessor/internal/traces/functions.go

@@ -16,10 +16,16 @@ package traces // import "github.com/open-telemetry/opentelemetry-collector-cont
 
 import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspanevent"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/common"
 )
 
-func Functions() map[string]interface{} {
+func SpanFunctions() map[string]interface{} {
 	// No trace-only functions yet.
 	return common.Functions[ottlspan.TransformContext]()
 }
+
+func SpanEventFunctions() map[string]interface{} {
+	// No trace-only functions yet.
+	return common.Functions[ottlspanevent.TransformContext]()
+}

processor/transformprocessor/internal/traces/functions_test.go

@@ -21,12 +21,22 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspanevent"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/common"
 )
 
-func Test_DefaultFunctions(t *testing.T) {
+func Test_SpanFunctions(t *testing.T) {
 	expected := common.Functions[ottlspan.TransformContext]()
-	actual := Functions()
+	actual := SpanFunctions()
+	require.Equal(t, len(expected), len(actual))
+	for k := range actual {
+		assert.Contains(t, expected, k)
+	}
+}
+
+func Test_SpanEventFunctions(t *testing.T) {
+	expected := common.Functions[ottlspanevent.TransformContext]()
+	actual := SpanEventFunctions()
 	require.Equal(t, len(expected), len(actual))
 	for k := range actual {
 		assert.Contains(t, expected, k)

processor/transformprocessor/internal/traces/processor.go

@@ -34,7 +34,7 @@ type Processor struct {
 
 func NewProcessor(statements []string, contextStatements []common.ContextStatements, settings component.TelemetrySettings) (*Processor, error) {
 	if len(statements) > 0 {
-		ottlp := ottlspan.NewParser(Functions(), settings)
+		ottlp := ottlspan.NewParser(SpanFunctions(), settings)
 		parsedStatements, err := ottlp.ParseStatements(statements)
 		if err != nil {
 			return nil, err
@@ -44,7 +44,7 @@ func NewProcessor(statements []string, contextStatements []common.ContextStateme
 		}, nil
 	}
 
-	pc, err := common.NewTraceParserCollection(Functions(), settings)
+	pc, err := common.NewTraceParserCollection(settings, common.WithSpanParser(SpanFunctions()), common.WithSpanEventParser(SpanEventFunctions()))
 	if err != nil {
 		return nil, err
 	}