[processor/transform] Add support for expressing a statement's context via path names

.chloggen/transformprocessor-add-statements-with-context-support.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: processor/transformprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for expressing statement's context via path names.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [29017]
+
+# (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/config.go

@@ -5,8 +5,11 @@ package transformprocessor // import "github.com/open-telemetry/opentelemetry-co
 
 import (
 	"errors"
+	"fmt"
+	"reflect"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/featuregate"
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
@@ -24,7 +27,8 @@ var (
 		featuregate.WithRegisterFromVersion("v0.103.0"),
 		featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/32080#issuecomment-2120764953"),
 	)
-	errFlatLogsGateDisabled = errors.New("'flatten_data' requires the 'transform.flatten.logs' feature gate to be enabled")
+	errFlatLogsGateDisabled       = errors.New("'flatten_data' requires the 'transform.flatten.logs' feature gate to be enabled")
+	configContextStatementsFields = []string{"trace_statements", "metric_statements", "log_statements"}
 )
 
 // Config defines the configuration for the processor.
@@ -44,6 +48,56 @@ type Config struct {
 	logger      *zap.Logger
 }
 
+// Unmarshal is used internally by mapstructure to parse the transformprocessor configuration (Config),
+// adding support to structured and flat configuration styles (array of statements strings).
+// When the flat configuration style is used, each statement becomes a new common.ContextStatements
+// object, with empty [common.ContextStatements.Context] value.
+// On the other hand, structured configurations are parsed following the mapstructure Config format.
+// Mixed configuration styles are also supported.
+func (c *Config) Unmarshal(component *confmap.Conf) error {
+	if component == nil {
+		return nil
+	}
+
+	contextStatementsPatch := map[string]any{}
+	for _, fieldName := range configContextStatementsFields {
+		if !component.IsSet(fieldName) {
+			continue
+		}
+
+		rawVal := component.Get(fieldName)
+		values, ok := rawVal.([]any)
+		if !ok {
+			return fmt.Errorf("invalid %s type, expected: array, got: %t", fieldName, rawVal)
+		}
+
+		if len(values) == 0 {
+			continue
+		}
+
+		stmts := make([]any, 0, len(values))
+		for _, value := range values {
+			// Array of strings means it's a flat configuration style
+			if reflect.TypeOf(value).Kind() == reflect.String {
+				stmts = append(stmts, map[string]any{"statements": []any{value}})
+			} else {
+				stmts = append(stmts, value)
+			}
+		}
+
+		contextStatementsPatch[fieldName] = stmts
+	}
+
+	if len(contextStatementsPatch) > 0 {
+		err := component.Merge(confmap.NewFromStringMap(contextStatementsPatch))
+		if err != nil {
+			return err
+		}
+	}
+
+	return component.Unmarshal(c)
+}
+
 var _ component.Config = (*Config)(nil)
 
 func (c *Config) Validate() error {

processor/transformprocessor/config_test.go

@@ -147,9 +147,81 @@ func TestLoadConfig(t *testing.T) {
 			id:       component.NewIDWithName(metadata.Type, "bad_syntax_multi_signal"),
 			errorLen: 3,
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "flat_configuration"),
+			expected: &Config{
+				ErrorMode: ottl.PropagateError,
+				TraceStatements: []common.ContextStatements{
+					{
+						Statements: []string{`set(span.name, "bear") where span.attributes["http.path"] == "/animal"`},
+					},
+					{
+						Statements: []string{`set(resource.attributes["name"], "bear")`},
+					},
+				},
+				MetricStatements: []common.ContextStatements{
+					{
+						Statements: []string{`set(metric.name, "bear") where resource.attributes["http.path"] == "/animal"`},
+					},
+					{
+						Statements: []string{`set(resource.attributes["name"], "bear")`},
+					},
+				},
+				LogStatements: []common.ContextStatements{
+					{
+						Statements: []string{`set(log.body, "bear") where log.attributes["http.path"] == "/animal"`},
+					},
+					{
+						Statements: []string{`set(resource.attributes["name"], "bear")`},
+					},
+				},
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "mixed_configuration_styles"),
+			expected: &Config{
+				ErrorMode: ottl.PropagateError,
+				TraceStatements: []common.ContextStatements{
+					{
+						Statements: []string{`set(span.name, "bear") where span.attributes["http.path"] == "/animal"`},
+					},
+					{
+						Context: "span",
+						Statements: []string{
+							`set(attributes["name"], "bear")`,
+							`keep_keys(attributes, ["http.method", "http.path"])`,
+						},
+					},
+				},
+				MetricStatements: []common.ContextStatements{
+					{
+						Statements: []string{`set(metric.name, "bear") where resource.attributes["http.path"] == "/animal"`},
+					},
+					{
+						Context: "resource",
+						Statements: []string{
+							`set(attributes["name"], "bear")`,
+							`keep_keys(attributes, ["http.method", "http.path"])`,
+						},
+					},
+				},
+				LogStatements: []common.ContextStatements{
+					{
+						Statements: []string{`set(log.body, "bear") where log.attributes["http.path"] == "/animal"`},
+					},
+					{
+						Context: "resource",
+						Statements: []string{
+							`set(attributes["name"], "bear")`,
+							`keep_keys(attributes, ["http.method", "http.path"])`,
+						},
+					},
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
-		t.Run(tt.id.String(), func(t *testing.T) {
+		t.Run(tt.id.Name(), func(t *testing.T) {
 			cm, err := confmaptest.LoadConf(filepath.Join("testdata", "config.yaml"))
 			assert.NoError(t, err)
 

processor/transformprocessor/internal/common/config.go

@@ -6,8 +6,12 @@ package common // import "github.com/open-telemetry/opentelemetry-collector-cont
 import (
 	"fmt"
 	"strings"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 )
 
+var _ ottl.StatementsGetter = (*ContextStatements)(nil)
+
 type ContextID string
 
 const (
@@ -36,3 +40,15 @@ type ContextStatements struct {
 	Conditions []string  `mapstructure:"conditions"`
 	Statements []string  `mapstructure:"statements"`
 }
+
+func (c ContextStatements) GetStatements() []string {
+	return c.Statements
+}
+
+func toContextStatements(statements any) (*ContextStatements, error) {
+	contextStatements, ok := statements.(ContextStatements)
+	if !ok {
+		return nil, fmt.Errorf("invalid context statements type, expected: common.ContextStatements, got: %T", statements)
+	}
+	return &contextStatements, nil
+}

processor/transformprocessor/internal/common/logs.go

@@ -14,8 +14,6 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterottl"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottllog"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlresource"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlscope"
 )
 
 var _ consumer.Logs = &logStatements{}
@@ -55,76 +53,60 @@ func (l logStatements) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 	return nil
 }
 
-type LogParserCollection struct {
-	parserCollection
-	logParser ottl.Parser[ottllog.TransformContext]
-}
+type LogParserCollection ottl.ParserCollection[consumer.Logs]
 
-type LogParserCollectionOption func(*LogParserCollection) error
+type LogParserCollectionOption ottl.ParserCollectionOption[consumer.Logs]
 
 func WithLogParser(functions map[string]ottl.Factory[ottllog.TransformContext]) LogParserCollectionOption {
-	return func(lp *LogParserCollection) error {
-		logParser, err := ottllog.NewParser(functions, lp.settings)
+	return func(pc *ottl.ParserCollection[consumer.Logs]) error {
+		logParser, err := ottllog.NewParser(functions, pc.Settings, ottllog.EnablePathContextNames())
 		if err != nil {
 			return err
 		}
-		lp.logParser = logParser
-		return nil
+		return ottl.WithParserCollectionContext(ottllog.ContextName, &logParser, convertLogStatements)(pc)
 	}
 }
 
 func WithLogErrorMode(errorMode ottl.ErrorMode) LogParserCollectionOption {
-	return func(lp *LogParserCollection) error {
-		lp.errorMode = errorMode
-		return nil
-	}
+	return LogParserCollectionOption(ottl.WithParserCollectionErrorMode[consumer.Logs](errorMode))
 }
 
 func NewLogParserCollection(settings component.TelemetrySettings, options ...LogParserCollectionOption) (*LogParserCollection, error) {
-	rp, err := ottlresource.NewParser(ResourceFunctions(), settings)
+	pcOptions := []ottl.ParserCollectionOption[consumer.Logs]{
+		withCommonContextParsers[consumer.Logs](),
+		ottl.EnableParserCollectionModifiedStatementLogging[consumer.Logs](true),
+	}
+
+	for _, option := range options {
+		pcOptions = append(pcOptions, ottl.ParserCollectionOption[consumer.Logs](option))
+	}
+
+	pc, err := ottl.NewParserCollection(settings, pcOptions...)
 	if err != nil {
 		return nil, err
 	}
-	sp, err := ottlscope.NewParser(ScopeFunctions(), settings)
+
+	lpc := LogParserCollection(*pc)
+	return &lpc, nil
+}
+
+func convertLogStatements(pc *ottl.ParserCollection[consumer.Logs], _ *ottl.Parser[ottllog.TransformContext], _ string, statements ottl.StatementsGetter, parsedStatements []*ottl.Statement[ottllog.TransformContext]) (consumer.Logs, error) {
+	contextStatements, err := toContextStatements(statements)
 	if err != nil {
 		return nil, err
 	}
-	lpc := &LogParserCollection{
-		parserCollection: parserCollection{
-			settings:       settings,
-			resourceParser: rp,
-			scopeParser:    sp,
-		},
+	globalExpr, errGlobalBoolExpr := parseGlobalExpr(filterottl.NewBoolExprForLog, contextStatements.Conditions, pc.ErrorMode, pc.Settings, filterottl.StandardLogFuncs())
+	if errGlobalBoolExpr != nil {
+		return nil, errGlobalBoolExpr
 	}
-
-	for _, op := range options {
-		err := op(lpc)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	return lpc, nil
+	lStatements := ottllog.NewStatementSequence(parsedStatements, pc.Settings, ottllog.WithStatementSequenceErrorMode(pc.ErrorMode))
+	return logStatements{lStatements, globalExpr}, nil
 }
 
-func (pc LogParserCollection) ParseContextStatements(contextStatements ContextStatements) (consumer.Logs, error) {
-	switch contextStatements.Context {
-	case Log:
-		parsedStatements, err := pc.logParser.ParseStatements(contextStatements.Statements)
-		if err != nil {
-			return nil, err
-		}
-		globalExpr, errGlobalBoolExpr := parseGlobalExpr(filterottl.NewBoolExprForLog, contextStatements.Conditions, pc.parserCollection, filterottl.StandardLogFuncs())
-		if errGlobalBoolExpr != nil {
-			return nil, errGlobalBoolExpr
-		}
-		lStatements := ottllog.NewStatementSequence(parsedStatements, pc.settings, ottllog.WithStatementSequenceErrorMode(pc.errorMode))
-		return logStatements{lStatements, globalExpr}, nil
-	default:
-		statements, err := pc.parseCommonContextStatements(contextStatements)
-		if err != nil {
-			return nil, err
-		}
-		return statements, nil
+func (lpc *LogParserCollection) ParseContextStatements(contextStatements ContextStatements) (consumer.Logs, error) {
+	pc := ottl.ParserCollection[consumer.Logs](*lpc)
+	if contextStatements.Context != "" {
+		return pc.ParseStatementsWithContext(string(contextStatements.Context), contextStatements, true)
 	}
+	return pc.ParseStatements(contextStatements)
 }