[processor/transform] Add support for profiles (open-telemetry#39974)

#### Description
PR for profiles support in the transform processor.
It's a split-out from open-telemetry#39036 and contains only basic functionality, that
makes this PR independent from open-telemetry#39681, open-telemetry#39416 and
open-telemetry/opentelemetry-collector#12798.
For this, several tests were commented out.

The reason for this PR is the hope to get it merged so that users and
developers can start experimenting with profiles functionality in the
transformprocessor.

Attributes can't currently can't be used with profiles (due to the above
mentioned unmerged PRs).

#### Link to tracking issue
Fixes open-telemetry#39009

---------

Co-authored-by: Edmo Vamerlatti Costa <[email protected]>
Co-authored-by: Tyler Helmuth <[email protected]>

Author: 3 people

.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.
 
@@ -114,6 +115,10 @@ transform:
     - replace_all_matches(log.attributes, "/user/*/list/*", "/user/{userId}/list/{listId}")
     - replace_all_patterns(log.attributes, "value", "/account/\\d{4}", "/account/{accountId}")
     - set(log.body, log.attributes["http.route"])
+  profile_statements:
+    - keep_keys(resource.attributes, ["host.name"])
+    - set(profile.attributes["tag"], "profile#23")
+    - set(profile.original_payload_format, "json")
 ```
 
 In some situations a combination of Paths, functions, or enums is not allowed, and the solution 
@@ -133,7 +138,7 @@ Format:
 ```yaml
 transform:
   error_mode: ignore
-  <trace|metric|log>_statements:
+  <trace|metric|log|profile>_statements:
     - context: string
       error_mode: propagate
       conditions: 

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))
+}