[processor/transform] Add support for profiles

.chloggen/profiles-transformprocessor.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: transformprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add profiles support to transformprocessor.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39009]
+
+# (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/README.md

@@ -46,19 +46,20 @@ transform:
   <trace|metric|log>_statements: []
 ```
 
-The Transform Processor's primary configuration section is broken down by signal (traces, metrics, and logs)
+The Transform Processor's primary configuration section is broken down by signal (traces, metrics, logs, and profiles)
 and allows you to configure a list of statements for the processor to execute. The list can be made of:
 
 - OTTL statements. This option will meet most user's needs. See [Basic Config](#basic-config) for more details.
 - Objects, which allows users to apply configuration options to a specific list of statements. See [Advanced Config](#advanced-config) for more details.
 
 Within each `<signal_statements>` list, only certain OTTL Path prefixes can be used:
 
-| Signal            | Path Prefix Values                             |
-|-------------------|------------------------------------------------|
-| trace_statements  | `resource`, `scope`, `span`, and `spanevent`   |
-| metric_statements | `resource`, `scope`, `metric`, and `datapoint` |
-| log_statements    | `resource`, `scope`, and `log`                 |
+| Signal             | Path Prefix Values                             |
+|--------------------|------------------------------------------------|
+| trace_statements   | `resource`, `scope`, `span`, and `spanevent`   |
+| metric_statements  | `resource`, `scope`, `metric`, and `datapoint` |
+| log_statements     | `resource`, `scope`, and `log`                 |
+| profile_statements | `resource`, `scope`, and `profile`             |
 
 This means, for example, that you cannot use the Path `span.attributes` within the `log_statements` configuration section.
 

processor/transformprocessor/config.go

@@ -18,6 +18,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
 	"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/contexts/ottlprofile"
 	"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"
@@ -41,9 +42,10 @@ type Config struct {
 	// The default value is `propagate`.
 	ErrorMode ottl.ErrorMode `mapstructure:"error_mode"`
 
-	TraceStatements  []common.ContextStatements `mapstructure:"trace_statements"`
-	MetricStatements []common.ContextStatements `mapstructure:"metric_statements"`
-	LogStatements    []common.ContextStatements `mapstructure:"log_statements"`
+	TraceStatements   []common.ContextStatements `mapstructure:"trace_statements"`
+	MetricStatements  []common.ContextStatements `mapstructure:"metric_statements"`
+	LogStatements     []common.ContextStatements `mapstructure:"log_statements"`
+	ProfileStatements []common.ContextStatements `mapstructure:"profile_statements"`
 
 	FlattenData bool `mapstructure:"flatten_data"`
 	logger      *zap.Logger
@@ -53,6 +55,7 @@ type Config struct {
 	metricFunctions    map[string]ottl.Factory[ottlmetric.TransformContext]
 	spanEventFunctions map[string]ottl.Factory[ottlspanevent.TransformContext]
 	spanFunctions      map[string]ottl.Factory[ottlspan.TransformContext]
+	profileFunctions   map[string]ottl.Factory[ottlprofile.TransformContext]
 }
 
 // Unmarshal is used internally by mapstructure to parse the transformprocessor configuration (Config),
@@ -80,9 +83,10 @@ func (c *Config) Unmarshal(conf *confmap.Conf) error {
 	}
 
 	contextStatementsFields := map[string]*[]common.ContextStatements{
-		"trace_statements":  &c.TraceStatements,
-		"metric_statements": &c.MetricStatements,
-		"log_statements":    &c.LogStatements,
+		"trace_statements":   &c.TraceStatements,
+		"metric_statements":  &c.MetricStatements,
+		"log_statements":     &c.LogStatements,
+		"profile_statements": &c.ProfileStatements,
 	}
 
 	contextStatementsPatch := map[string]any{}
@@ -179,6 +183,19 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if len(c.ProfileStatements) > 0 {
+		pc, err := common.NewProfileParserCollection(component.TelemetrySettings{Logger: zap.NewNop()}, common.WithProfileParser(c.profileFunctions))
+		if err != nil {
+			return err
+		}
+		for _, cs := range c.ProfileStatements {
+			_, err = pc.ParseContextStatements(cs)
+			if err != nil {
+				errors = multierr.Append(errors, err)
+			}
+		}
+	}
+
 	if c.FlattenData && !flatLogsFeatureGate.IsEnabled() {
 		errors = multierr.Append(errors, errFlatLogsGateDisabled)
 	}

processor/transformprocessor/config_test.go

@@ -75,6 +75,20 @@ func TestLoadConfig(t *testing.T) {
 						},
 					},
 				},
+				ProfileStatements: []common.ContextStatements{
+					{
+						Context: "profile",
+						Statements: []string{
+							`set(original_payload_format, "bear") where original_payload_format == "/animal"`,
+						},
+					},
+					{
+						Context: "resource",
+						Statements: []string{
+							`set(attributes["name"], "bear")`,
+						},
+					},
+				},
 			},
 		},
 		{
@@ -108,6 +122,15 @@ func TestLoadConfig(t *testing.T) {
 						},
 					},
 				},
+				ProfileStatements: []common.ContextStatements{
+					{
+						Context:    "profile",
+						Conditions: []string{`original_payload_format == "/animal"`},
+						Statements: []string{
+							`set(original_payload_format, "bear")`,
+						},
+					},
+				},
 			},
 		},
 		{
@@ -122,8 +145,9 @@ func TestLoadConfig(t *testing.T) {
 						},
 					},
 				},
-				MetricStatements: []common.ContextStatements{},
-				LogStatements:    []common.ContextStatements{},
+				MetricStatements:  []common.ContextStatements{},
+				LogStatements:     []common.ContextStatements{},
+				ProfileStatements: []common.ContextStatements{},
 			},
 		},
 		{
@@ -144,6 +168,12 @@ func TestLoadConfig(t *testing.T) {
 		{
 			id: component.NewIDWithName(metadata.Type, "unknown_function_log"),
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "bad_syntax_profile"),
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "unknown_function_profile"),
+		},
 		{
 			id: component.NewIDWithName(metadata.Type, "bad_syntax_multi_signal"),
 			errors: []error{
@@ -174,6 +204,12 @@ func TestLoadConfig(t *testing.T) {
 						Statements: []string{`set(log.body, "bear") where log.attributes["http.path"] == "/animal"`},
 					},
 				},
+				ProfileStatements: []common.ContextStatements{
+					{
+						Context:    "profile",
+						Statements: []string{`set(profile.original_payload_format, "bear") where profile.original_payload_format == "/animal"`},
+					},
+				},
 			},
 		},
 		{
@@ -204,6 +240,14 @@ func TestLoadConfig(t *testing.T) {
 						},
 					},
 				},
+				ProfileStatements: []common.ContextStatements{
+					{
+						Statements: []string{
+							`set(profile.original_payload_format, "bear") where profile.original_payload_format == "/animal"`,
+							`set(resource.attributes["name"], "bear")`,
+						},
+					},
+				},
 			},
 		},
 		{
@@ -234,6 +278,14 @@ func TestLoadConfig(t *testing.T) {
 						},
 					},
 				},
+				ProfileStatements: []common.ContextStatements{
+					{
+						Statements: []string{
+							`set(profile.original_payload_format, "bear") where profile.original_payload_format == "/animal"`,
+							`set(resource.attributes["name"], "bear")`,
+						},
+					},
+				},
 			},
 		},
 		{
@@ -270,6 +322,16 @@ func TestLoadConfig(t *testing.T) {
 						ErrorMode:  "",
 					},
 				},
+				ProfileStatements: []common.ContextStatements{
+					{
+						Statements: []string{`set(resource.attributes["name"], "propagate")`},
+						ErrorMode:  ottl.PropagateError,
+					},
+					{
+						Statements: []string{`set(resource.attributes["name"], "ignore")`},
+						ErrorMode:  "",
+					},
+				},
 			},
 		},
 	}

processor/transformprocessor/factory.go

@@ -9,20 +9,25 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/xconsumer"
 	"go.opentelemetry.io/collector/processor"
 	"go.opentelemetry.io/collector/processor/processorhelper"
+	"go.opentelemetry.io/collector/processor/processorhelper/xprocessorhelper"
+	"go.opentelemetry.io/collector/processor/xprocessor"
 	"go.uber.org/zap"
 
 	"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/ottllog"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlprofile"
 	"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"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/logs"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/metadata"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/metrics"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/profiles"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/traces"
 )
 
@@ -34,11 +39,13 @@ type transformProcessorFactory struct {
 	metricFunctions                     map[string]ottl.Factory[ottlmetric.TransformContext]
 	spanEventFunctions                  map[string]ottl.Factory[ottlspanevent.TransformContext]
 	spanFunctions                       map[string]ottl.Factory[ottlspan.TransformContext]
+	profileFunctions                    map[string]ottl.Factory[ottlprofile.TransformContext]
 	defaultDataPointFunctionsOverridden bool
 	defaultLogFunctionsOverridden       bool
 	defaultMetricFunctionsOverridden    bool
 	defaultSpanEventFunctionsOverridden bool
 	defaultSpanFunctionsOverridden      bool
+	defaultProfileFunctionsOverridden   bool
 }
 
 // FactoryOption applies changes to transformProcessorFactory.
@@ -104,6 +111,18 @@ func WithSpanFunctions(spanFunctions []ottl.Factory[ottlspan.TransformContext])
 	}
 }
 
+// WithProfileFunctions will override the default OTTL profile context functions with the provided profileFunctions in the resulting processor.
+// Subsequent uses of WithProfileFunctions will merge the provided profileFunctions with the previously registered functions.
+func WithProfileFunctions(profileFunctions []ottl.Factory[ottlprofile.TransformContext]) FactoryOption {
+	return func(factory *transformProcessorFactory) {
+		if !factory.defaultProfileFunctionsOverridden {
+			factory.profileFunctions = map[string]ottl.Factory[ottlprofile.TransformContext]{}
+			factory.defaultProfileFunctionsOverridden = true
+		}
+		factory.profileFunctions = mergeFunctionsToMap(factory.profileFunctions, profileFunctions)
+	}
+}
+
 func NewFactory() processor.Factory {
 	return NewFactoryWithOptions()
 }
@@ -116,17 +135,19 @@ func NewFactoryWithOptions(options ...FactoryOption) processor.Factory {
 		metricFunctions:    defaultMetricFunctionsMap(),
 		spanEventFunctions: defaultSpanEventFunctionsMap(),
 		spanFunctions:      defaultSpanFunctionsMap(),
+		profileFunctions:   defaultProfileFunctionsMap(),
 	}
 	for _, o := range options {
 		o(f)
 	}
 
-	return processor.NewFactory(
+	return xprocessor.NewFactory(
 		metadata.Type,
 		f.createDefaultConfig,
-		processor.WithLogs(f.createLogsProcessor, metadata.LogsStability),
-		processor.WithTraces(f.createTracesProcessor, metadata.TracesStability),
-		processor.WithMetrics(f.createMetricsProcessor, metadata.MetricsStability),
+		xprocessor.WithLogs(f.createLogsProcessor, metadata.LogsStability),
+		xprocessor.WithTraces(f.createTracesProcessor, metadata.TracesStability),
+		xprocessor.WithMetrics(f.createMetricsProcessor, metadata.MetricsStability),
+		xprocessor.WithProfiles(f.createProfilesProcessor, metadata.ProfilesStability),
 	)
 }
 
@@ -136,11 +157,13 @@ func (f *transformProcessorFactory) createDefaultConfig() component.Config {
 		TraceStatements:    []common.ContextStatements{},
 		MetricStatements:   []common.ContextStatements{},
 		LogStatements:      []common.ContextStatements{},
+		ProfileStatements:  []common.ContextStatements{},
 		dataPointFunctions: f.dataPointFunctions,
 		logFunctions:       f.logFunctions,
 		metricFunctions:    f.metricFunctions,
 		spanEventFunctions: f.spanEventFunctions,
 		spanFunctions:      f.spanFunctions,
+		profileFunctions:   f.profileFunctions,
 	}
 }
 
@@ -219,3 +242,28 @@ func (f *transformProcessorFactory) createMetricsProcessor(
 		proc.ProcessMetrics,
 		processorhelper.WithCapabilities(processorCapabilities))
 }
+
+func (f *transformProcessorFactory) createProfilesProcessor(
+	ctx context.Context,
+	set processor.Settings,
+	cfg component.Config,
+	nextConsumer xconsumer.Profiles,
+) (xprocessor.Profiles, error) {
+	oCfg := cfg.(*Config)
+	oCfg.logger = set.Logger
+
+	if f.defaultProfileFunctionsOverridden {
+		set.Logger.Debug("non-default OTTL profile functions have been registered in the \"transform\" processor", zap.Bool("profile", f.defaultProfileFunctionsOverridden))
+	}
+	proc, err := profiles.NewProcessor(oCfg.ProfileStatements, oCfg.ErrorMode, set.TelemetrySettings, f.profileFunctions)
+	if err != nil {
+		return nil, fmt.Errorf("invalid config for \"transform\" processor %w", err)
+	}
+	return xprocessorhelper.NewProfiles(
+		ctx,
+		set,
+		cfg,
+		nextConsumer,
+		proc.ProcessProfiles,
+		xprocessorhelper.WithCapabilities(processorCapabilities))
+}