[processor/transform] Add support for profiles

Author: rockdaboot

internal/filter/filterottl/filter.go

@@ -10,6 +10,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/ottlresource"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlscope"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
@@ -121,6 +122,27 @@ func NewBoolExprForLogWithOptions(conditions []string, functions map[string]ottl
 	return &c, nil
 }
 
+// NewBoolExprForProfile creates a BoolExpr[ottlprofile.TransformContext] that will return true if any of the given OTTL conditions evaluate to true.
+// The passed in functions should use the ottlprofile.TransformContext.
+// If a function named `match` is not present in the function map it will be added automatically so that parsing works as expected
+func NewBoolExprForProfile(conditions []string, functions map[string]ottl.Factory[ottlprofile.TransformContext], errorMode ottl.ErrorMode, set component.TelemetrySettings) (*ottl.ConditionSequence[ottlprofile.TransformContext], error) {
+	return NewBoolExprForProfileWithOptions(conditions, functions, errorMode, set, nil)
+}
+
+// NewBoolExprForLogWithOptions is like NewBoolExprForLog, but with additional options.
+func NewBoolExprForProfileWithOptions(conditions []string, functions map[string]ottl.Factory[ottlprofile.TransformContext], errorMode ottl.ErrorMode, set component.TelemetrySettings, parserOptions []ottl.Option[ottlprofile.TransformContext]) (*ottl.ConditionSequence[ottlprofile.TransformContext], error) {
+	parser, err := ottlprofile.NewParser(functions, set, parserOptions...)
+	if err != nil {
+		return nil, err
+	}
+	statements, err := parser.ParseConditions(conditions)
+	if err != nil {
+		return nil, err
+	}
+	c := ottlprofile.NewConditionSequence(statements, set, ottlprofile.WithConditionSequenceErrorMode(errorMode))
+	return &c, nil
+}
+
 // NewBoolExprForResource creates a BoolExpr[ottlresource.TransformContext] that will return true if any of the given OTTL conditions evaluate to true.
 // The passed in functions should use the ottlresource.TransformContext.
 // If a function named `match` is not present in the function map it will be added automatically so that parsing works as expected

internal/filter/filterottl/functions.go

@@ -13,6 +13,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/ottlresource"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlscope"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
@@ -52,6 +53,10 @@ func StandardLogFuncs() map[string]ottl.Factory[ottllog.TransformContext] {
 	return ottlfuncs.StandardConverters[ottllog.TransformContext]()
 }
 
+func StandardProfileFuncs() map[string]ottl.Factory[ottlprofile.TransformContext] {
+	return ottlfuncs.StandardConverters[ottlprofile.TransformContext]()
+}
+
 func StandardResourceFuncs() map[string]ottl.Factory[ottlresource.TransformContext] {
 	return ottlfuncs.StandardConverters[ottlresource.TransformContext]()
 }

internal/filter/go.mod

@@ -62,7 +62,7 @@ require (
 	golang.org/x/net v0.37.0 // indirect
 	golang.org/x/sys v0.31.0 // indirect
 	golang.org/x/text v0.23.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250227231956-55c901821b1e // indirect
 	google.golang.org/grpc v1.71.0 // indirect
 	google.golang.org/protobuf v1.36.5 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

internal/filter/go.sum

@@ -22,6 +22,9 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/internal/logprofile"
 )
 
+// Experimental: *NOTE* this constant is subject to change or removal in the future.
+const ContextName = ctxprofile.Name
+
 var (
 	_ ctxresource.Context     = TransformContext{}
 	_ ctxscope.Context        = TransformContext{}
@@ -46,8 +49,6 @@ type TransformContext struct {
 	resourceProfiles     pprofile.ResourceProfiles
 }
 
-type Option func(*ottl.Parser[TransformContext])
-
 type TransformContextOption func(*TransformContext)
 
 func NewTransformContext(profile pprofile.Profile, instrumentationScope pcommon.InstrumentationScope, resource pcommon.Resource, scopeProfiles pprofile.ScopeProfiles, resourceProfiles pprofile.ResourceProfiles, options ...TransformContextOption) TransformContext {
@@ -107,7 +108,7 @@ type pathExpressionParser struct {
 	cacheGetSetter    ottl.PathExpressionParser[TransformContext]
 }
 
-func NewParser(functions map[string]ottl.Factory[TransformContext], telemetrySettings component.TelemetrySettings, options ...Option) (ottl.Parser[TransformContext], error) {
+func NewParser(functions map[string]ottl.Factory[TransformContext], telemetrySettings component.TelemetrySettings, options ...ottl.Option[TransformContext]) (ottl.Parser[TransformContext], error) {
 	pep := pathExpressionParser{
 		telemetrySettings: telemetrySettings,
 		cacheGetSetter:    ctxcache.PathExpressionParser(getCache),
@@ -131,7 +132,7 @@ func NewParser(functions map[string]ottl.Factory[TransformContext], telemetrySet
 // otherwise an error is reported.
 //
 // Experimental: *NOTE* this option is subject to change or removal in the future.
-func EnablePathContextNames() Option {
+func EnablePathContextNames() ottl.Option[TransformContext] {
 	return func(p *ottl.Parser[TransformContext]) {
 		ottl.WithPathContextNames[TransformContext]([]string{
 			ctxprofile.Name,

pkg/ottl/contexts/ottlprofile/profile.go

@@ -9,26 +9,31 @@ 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),
 	)
 }
 
@@ -104,3 +109,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.MetricStatements, 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))
+}

processor/transformprocessor/factory.go

@@ -28,7 +28,11 @@ require (
 	go.opentelemetry.io/collector/component/componenttest v0.122.2-0.20250319144947-41a9ea7f7402
 	go.opentelemetry.io/collector/confmap/xconfmap v0.122.2-0.20250319144947-41a9ea7f7402
 	go.opentelemetry.io/collector/consumer/consumertest v0.122.2-0.20250319144947-41a9ea7f7402
+	go.opentelemetry.io/collector/consumer/xconsumer v0.122.2-0.20250319144947-41a9ea7f7402
+	go.opentelemetry.io/collector/pdata/pprofile v0.122.2-0.20250319144947-41a9ea7f7402
+	go.opentelemetry.io/collector/processor/processorhelper/xprocessorhelper v0.122.1
 	go.opentelemetry.io/collector/processor/processortest v0.122.2-0.20250319144947-41a9ea7f7402
+	go.opentelemetry.io/collector/processor/xprocessor v0.122.2-0.20250319144947-41a9ea7f7402
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842
 )
 
@@ -67,18 +71,15 @@ require (
 	github.com/ua-parser/uap-go v0.0.0-20240611065828-3a4781585db6 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/collector/component/componentstatus v0.122.2-0.20250319144947-41a9ea7f7402 // indirect
-	go.opentelemetry.io/collector/consumer/xconsumer v0.122.2-0.20250319144947-41a9ea7f7402 // indirect
-	go.opentelemetry.io/collector/pdata/pprofile v0.122.2-0.20250319144947-41a9ea7f7402 // indirect
 	go.opentelemetry.io/collector/pdata/testdata v0.122.2-0.20250319144947-41a9ea7f7402 // indirect
 	go.opentelemetry.io/collector/pipeline v0.122.2-0.20250319144947-41a9ea7f7402 // indirect
-	go.opentelemetry.io/collector/processor/xprocessor v0.122.2-0.20250319144947-41a9ea7f7402 // indirect
 	go.opentelemetry.io/otel v1.35.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.35.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.35.0 // indirect
 	golang.org/x/net v0.37.0 // indirect
 	golang.org/x/sys v0.31.0 // indirect
 	golang.org/x/text v0.23.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250227231956-55c901821b1e // indirect
 	google.golang.org/grpc v1.71.0 // indirect
 	google.golang.org/protobuf v1.36.5 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

processor/transformprocessor/go.mod

@@ -22,12 +22,13 @@ const (
 	Metric    ContextID = "metric"
 	DataPoint ContextID = "datapoint"
 	Log       ContextID = "log"
+	Profile   ContextID = "profile"
 )
 
 func (c *ContextID) UnmarshalText(text []byte) error {
 	str := ContextID(strings.ToLower(string(text)))
 	switch str {
-	case Resource, Scope, Span, SpanEvent, Metric, DataPoint, Log:
+	case Resource, Scope, Span, SpanEvent, Metric, DataPoint, Log, Profile:
 		*c = str
 		return nil
 	default:

processor/transformprocessor/go.sum

@@ -10,6 +10,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/pprofile"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/expr"
@@ -84,6 +85,24 @@ func (r resourceStatements) ConsumeLogs(ctx context.Context, ld plog.Logs, cache
 	return nil
 }
 
+func (r resourceStatements) ConsumeProfiles(ctx context.Context, ld pprofile.Profiles, cache *pcommon.Map) error {
+	for i := 0; i < ld.ResourceProfiles().Len(); i++ {
+		rprofiles := ld.ResourceProfiles().At(i)
+		tCtx := ottlresource.NewTransformContext(rprofiles.Resource(), rprofiles, ottlresource.WithCache(cache))
+		condition, err := r.BoolExpr.Eval(ctx, tCtx)
+		if err != nil {
+			return err
+		}
+		if condition {
+			err := r.Execute(ctx, tCtx)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 var _ baseContext = &scopeStatements{}
 
 type scopeStatements struct {
@@ -158,10 +177,32 @@ func (s scopeStatements) ConsumeLogs(ctx context.Context, ld plog.Logs, cache *p
 	return nil
 }
 
+func (s scopeStatements) ConsumeProfiles(ctx context.Context, ld pprofile.Profiles, cache *pcommon.Map) error {
+	for i := 0; i < ld.ResourceProfiles().Len(); i++ {
+		rprofiles := ld.ResourceProfiles().At(i)
+		for j := 0; j < rprofiles.ScopeProfiles().Len(); j++ {
+			sprofiles := rprofiles.ScopeProfiles().At(j)
+			tCtx := ottlscope.NewTransformContext(sprofiles.Scope(), rprofiles.Resource(), sprofiles, ottlscope.WithCache(cache))
+			condition, err := s.BoolExpr.Eval(ctx, tCtx)
+			if err != nil {
+				return err
+			}
+			if condition {
+				err := s.Execute(ctx, tCtx)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
 type baseContext interface {
 	TracesConsumer
 	MetricsConsumer
 	LogsConsumer
+	ProfilesConsumer
 }
 
 func withCommonContextParsers[R any]() ottl.ParserCollectionOption[R] {