[processor/transform] Add support for profiles

Author: rockdaboot

.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/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/metrics"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/profiles"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/traces"
 )
 
@@ -39,9 +40,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
@@ -72,9 +74,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{}
@@ -171,6 +174,19 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if len(c.ProfileStatements) > 0 {
+		pc, err := common.NewProfileParserCollection(component.TelemetrySettings{Logger: zap.NewNop()}, common.WithProfileParser(profiles.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,35 +9,41 @@ 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"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 	"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"
 )
 
 var processorCapabilities = consumer.Capabilities{MutatesData: true}
 
 func NewFactory() processor.Factory {
-	return processor.NewFactory(
+	return xprocessor.NewFactory(
 		metadata.Type,
 		createDefaultConfig,
-		processor.WithLogs(createLogsProcessor, metadata.LogsStability),
-		processor.WithTraces(createTracesProcessor, metadata.TracesStability),
-		processor.WithMetrics(createMetricsProcessor, metadata.MetricsStability),
+		xprocessor.WithLogs(createLogsProcessor, metadata.LogsStability),
+		xprocessor.WithTraces(createTracesProcessor, metadata.TracesStability),
+		xprocessor.WithMetrics(createMetricsProcessor, metadata.MetricsStability),
+		xprocessor.WithProfiles(createProfilesProcessor, metadata.ProfilesStability),
 	)
 }
 
 func createDefaultConfig() component.Config {
 	return &Config{
-		ErrorMode:        ottl.PropagateError,
-		TraceStatements:  []common.ContextStatements{},
-		MetricStatements: []common.ContextStatements{},
-		LogStatements:    []common.ContextStatements{},
+		ErrorMode:         ottl.PropagateError,
+		TraceStatements:   []common.ContextStatements{},
+		MetricStatements:  []common.ContextStatements{},
+		LogStatements:     []common.ContextStatements{},
+		ProfileStatements: []common.ContextStatements{},
 	}
 }
 
@@ -104,3 +110,25 @@ func createMetricsProcessor(
 		proc.ProcessMetrics,
 		processorhelper.WithCapabilities(processorCapabilities))
 }
+
+func createProfilesProcessor(
+	ctx context.Context,
+	set processor.Settings,
+	cfg component.Config,
+	nextConsumer xconsumer.Profiles,
+) (xprocessor.Profiles, error) {
+	oCfg := cfg.(*Config)
+	oCfg.logger = set.Logger
+
+	proc, err := profiles.NewProcessor(oCfg.ProfileStatements, oCfg.ErrorMode, set.TelemetrySettings)
+	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))
+}