Skip to content

Commit

Permalink
[processor/transform] Fix issue where ErrorMode wasn't used. (#19629)
Browse files Browse the repository at this point in the history 
* Fix issue where ErrorMode wasn't used.

* add changelog

* Add more tests
  • Loading branch information
@TylerHelmuth
TylerHelmuth committed Mar 14, 2023
1 parent f775944 commit dc2f6e1
Show file tree
Hide file tree
Showing 8 changed files with 147 additions and 26 deletions.
16 changes: 16 additions & 0 deletions .chloggen/tp-fix-error-mode-usage.yaml
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
change_type: bug_fix

# 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: Fixes bug where the value for `error_mode` was ignored.

# One or more tracking issues related to the change
issues: [19629]

# (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:
6 changes: 3 additions & 3 deletions processor/transformprocessor/factory.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ func createLogsProcessor(
) (processor.Logs, error) {
oCfg := cfg.(*Config)

proc, err := logs.NewProcessor(oCfg.LogStatements, ottl.PropagateError, set.TelemetrySettings)
proc, err := logs.NewProcessor(oCfg.LogStatements, oCfg.ErrorMode, set.TelemetrySettings)
if err != nil {
return nil, fmt.Errorf("invalid config for \"transform\" processor %w", err)
}
Expand All @@ -85,7 +85,7 @@ func createTracesProcessor(
) (processor.Traces, error) {
oCfg := cfg.(*Config)

proc, err := traces.NewProcessor(oCfg.TraceStatements, ottl.PropagateError, set.TelemetrySettings)
proc, err := traces.NewProcessor(oCfg.TraceStatements, oCfg.ErrorMode, set.TelemetrySettings)
if err != nil {
return nil, fmt.Errorf("invalid config for \"transform\" processor %w", err)
}
Expand All @@ -106,7 +106,7 @@ func createMetricsProcessor(
) (processor.Metrics, error) {
oCfg := cfg.(*Config)

proc, err := metrics.NewProcessor(oCfg.MetricStatements, ottl.PropagateError, set.TelemetrySettings)
proc, err := metrics.NewProcessor(oCfg.MetricStatements, oCfg.ErrorMode, set.TelemetrySettings)
if err != nil {
return nil, fmt.Errorf("invalid config for \"transform\" processor %w", err)
}
Expand Down
25 changes: 19 additions & 6 deletions processor/transformprocessor/factory_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,10 +74,14 @@ func TestFactoryCreateTracesProcessor(t *testing.T) {
factory := NewFactory()
cfg := factory.CreateDefaultConfig()
oCfg := cfg.(*Config)
oCfg.ErrorMode = ottl.IgnoreError
oCfg.TraceStatements = []common.ContextStatements{
{
Context: "span",
Statements: []string{`set(attributes["test"], "pass") where name == "operationA"`},
Context: "span",
Statements: []string{
`set(attributes["test"], "pass") where name == "operationA"`,
`set(attributes["test error mode"], ParseJSON(1)) where name == "operationA"`,
},
},
}
tp, err := factory.CreateTracesProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg, consumertest.NewNop())
Expand All @@ -103,6 +107,7 @@ func TestFactoryCreateMetricsProcessor_InvalidActions(t *testing.T) {
factory := NewFactory()
cfg := factory.CreateDefaultConfig()
oCfg := cfg.(*Config)
oCfg.ErrorMode = ottl.IgnoreError
oCfg.MetricStatements = []common.ContextStatements{
{
Context: "datapoint",
Expand All @@ -118,10 +123,14 @@ func TestFactoryCreateMetricsProcessor(t *testing.T) {
factory := NewFactory()
cfg := factory.CreateDefaultConfig()
oCfg := cfg.(*Config)
oCfg.ErrorMode = ottl.IgnoreError
oCfg.MetricStatements = []common.ContextStatements{
{
Context: "datapoint",
Statements: []string{`set(attributes["test"], "pass") where metric.name == "operationA"`},
Context: "datapoint",
Statements: []string{
`set(attributes["test"], "pass") where metric.name == "operationA"`,
`set(attributes["test error mode"], ParseJSON(1)) where metric.name == "operationA"`,
},
},
}
metricsProcessor, err := factory.CreateMetricsProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg, consumertest.NewNop())
Expand All @@ -147,10 +156,14 @@ func TestFactoryCreateLogsProcessor(t *testing.T) {
factory := NewFactory()
cfg := factory.CreateDefaultConfig()
oCfg := cfg.(*Config)
oCfg.ErrorMode = ottl.IgnoreError
oCfg.LogStatements = []common.ContextStatements{
{
Context: "log",
Statements: []string{`set(attributes["test"], "pass") where body == "operationA"`},
Context: "log",
Statements: []string{
`set(attributes["test"], "pass") where body == "operationA"`,
`set(attributes["test error mode"], ParseJSON(1)) where body == "operationA"`,
},
},
}
lp, err := factory.CreateLogsProcessor(context.Background(), processortest.NewNopCreateSettings(), cfg, consumertest.NewNop())
Expand Down
1 change: 0 additions & 1 deletion processor/transformprocessor/internal/common/logs.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,6 @@ func (l logStatements) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
type LogParserCollection struct {
parserCollection
logParser ottl.Parser[ottllog.TransformContext]
errorMode ottl.ErrorMode
}

type LogParserCollectionOption func(*LogParserCollection) error
Expand Down
1 change: 0 additions & 1 deletion processor/transformprocessor/internal/common/traces.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,6 @@ type TraceParserCollection struct {
parserCollection
spanParser ottl.Parser[ottlspan.TransformContext]
spanEventParser ottl.Parser[ottlspanevent.TransformContext]
errorMode ottl.ErrorMode
}

type TraceParserCollectionOption func(*TraceParserCollection) error
Expand Down
36 changes: 32 additions & 4 deletions processor/transformprocessor/internal/logs/processor_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ func Test_ProcessLogs_ResourceContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructLogs()
processor, err := NewProcessor([]common.ContextStatements{{Context: "resource", Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "resource", Statements: []string{tt.statement}}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessLogs(context.Background(), td)
Expand Down Expand Up @@ -95,7 +95,7 @@ func Test_ProcessLogs_ScopeContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructLogs()
processor, err := NewProcessor([]common.ContextStatements{{Context: "scope", Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "scope", Statements: []string{tt.statement}}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessLogs(context.Background(), td)
Expand Down Expand Up @@ -333,7 +333,7 @@ func Test_ProcessLogs_LogContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructLogs()
processor, err := NewProcessor([]common.ContextStatements{{Context: "log", Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "log", Statements: []string{tt.statement}}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessLogs(context.Background(), td)
Expand Down Expand Up @@ -450,7 +450,7 @@ func Test_ProcessLogs_MixContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
td := constructLogs()
processor, err := NewProcessor(tt.contextStatments, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor(tt.contextStatments, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessLogs(context.Background(), td)
Expand All @@ -464,6 +464,34 @@ func Test_ProcessLogs_MixContext(t *testing.T) {
}
}

func Test_ProcessTraces_Error(t *testing.T) {
tests := []struct {
statement string
context common.ContextID
}{
{
context: "resource",
},
{
context: "scope",
},
{
context: "log",
},
}

for _, tt := range tests {
t.Run(string(tt.context), func(t *testing.T) {
td := constructLogs()
processor, err := NewProcessor([]common.ContextStatements{{Context: tt.context, Statements: []string{`set(attributes["test"], ParseJSON(1))`}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessLogs(context.Background(), td)
assert.Error(t, err)
})
}
}

func constructLogs() plog.Logs {
td := plog.NewLogs()
rs0 := td.ResourceLogs().AppendEmpty()
Expand Down
43 changes: 39 additions & 4 deletions processor/transformprocessor/internal/metrics/processor_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ func Test_ProcessMetrics_ResourceContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructMetrics()
processor, err := NewProcessor([]common.ContextStatements{{Context: "resource", Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "resource", Statements: []string{tt.statement}}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessMetrics(context.Background(), td)
Expand Down Expand Up @@ -91,7 +91,7 @@ func Test_ProcessMetrics_ScopeContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructMetrics()
processor, err := NewProcessor([]common.ContextStatements{{Context: "scope", Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "scope", Statements: []string{tt.statement}}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessMetrics(context.Background(), td)
Expand Down Expand Up @@ -510,7 +510,7 @@ func Test_ProcessMetrics_DataPointContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statements[0], func(t *testing.T) {
td := constructMetrics()
processor, err := NewProcessor([]common.ContextStatements{{Context: "datapoint", Statements: tt.statements}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "datapoint", Statements: tt.statements}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessMetrics(context.Background(), td)
Expand Down Expand Up @@ -642,7 +642,7 @@ func Test_ProcessMetrics_MixContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
td := constructMetrics()
processor, err := NewProcessor(tt.contextStatments, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor(tt.contextStatments, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessMetrics(context.Background(), td)
Expand All @@ -656,6 +656,41 @@ func Test_ProcessMetrics_MixContext(t *testing.T) {
}
}

func Test_ProcessMetrics_Error(t *testing.T) {
tests := []struct {
statement string
context common.ContextID
}{
{
statement: `set(attributes["test"], ParseJSON(1))`,
context: "resource",
},
{
statement: `set(attributes["test"], ParseJSON(1))`,
context: "scope",
},
{
statement: `set(name, ParseJSON(1))`,
context: "metric",
},
{
statement: `set(attributes["test"], ParseJSON(1))`,
context: "datapoint",
},
}

for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructMetrics()
processor, err := NewProcessor([]common.ContextStatements{{Context: tt.context, Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessMetrics(context.Background(), td)
assert.Error(t, err)
})
}
}

func constructMetrics() pmetric.Metrics {
td := pmetric.NewMetrics()
rm0 := td.ResourceMetrics().AppendEmpty()
Expand Down
45 changes: 38 additions & 7 deletions processor/transformprocessor/internal/traces/processor_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ func Test_ProcessTraces_ResourceContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructTraces()
processor, err := NewProcessor([]common.ContextStatements{{Context: "resource", Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "resource", Statements: []string{tt.statement}}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessTraces(context.Background(), td)
Expand Down Expand Up @@ -96,7 +96,7 @@ func Test_ProcessTraces_ScopeContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructTraces()
processor, err := NewProcessor([]common.ContextStatements{{Context: "scope", Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "scope", Statements: []string{tt.statement}}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessTraces(context.Background(), td)
Expand Down Expand Up @@ -378,7 +378,7 @@ func Test_ProcessTraces_TraceContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructTraces()
processor, err := NewProcessor([]common.ContextStatements{{Context: "span", Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "span", Statements: []string{tt.statement}}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessTraces(context.Background(), td)
Expand Down Expand Up @@ -408,7 +408,7 @@ func Test_ProcessTraces_SpanEventContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.statement, func(t *testing.T) {
td := constructTraces()
processor, err := NewProcessor([]common.ContextStatements{{Context: "spanevent", Statements: []string{tt.statement}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "spanevent", Statements: []string{tt.statement}}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessTraces(context.Background(), td)
Expand Down Expand Up @@ -525,7 +525,7 @@ func Test_ProcessTraces_MixContext(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
td := constructTraces()
processor, err := NewProcessor(tt.contextStatments, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor(tt.contextStatments, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessTraces(context.Background(), td)
Expand All @@ -539,6 +539,37 @@ func Test_ProcessTraces_MixContext(t *testing.T) {
}
}

func Test_ProcessTraces_Error(t *testing.T) {
tests := []struct {
statement string
context common.ContextID
}{
{
context: "resource",
},
{
context: "scope",
},
{
context: "span",
},
{
context: "spanevent",
},
}

for _, tt := range tests {
t.Run(string(tt.context), func(t *testing.T) {
td := constructTraces()
processor, err := NewProcessor([]common.ContextStatements{{Context: tt.context, Statements: []string{`set(attributes["test"], ParseJSON(1))`}}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
assert.NoError(t, err)

_, err = processor.ProcessTraces(context.Background(), td)
assert.Error(t, err)
})
}
}

func BenchmarkTwoSpans(b *testing.B) {
tests := []struct {
name string
Expand Down Expand Up @@ -575,7 +606,7 @@ func BenchmarkTwoSpans(b *testing.B) {

for _, tt := range tests {
b.Run(tt.name, func(b *testing.B) {
processor, err := NewProcessor([]common.ContextStatements{{Context: "span", Statements: tt.statements}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "span", Statements: tt.statements}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(b, err)
b.ResetTimer()
for n := 0; n < b.N; n++ {
Expand Down Expand Up @@ -617,7 +648,7 @@ func BenchmarkHundredSpans(b *testing.B) {
}
for _, tt := range tests {
b.Run(tt.name, func(b *testing.B) {
processor, err := NewProcessor([]common.ContextStatements{{Context: "span", Statements: tt.statements}}, ottl.PropagateError, componenttest.NewNopTelemetrySettings())
processor, err := NewProcessor([]common.ContextStatements{{Context: "span", Statements: tt.statements}}, ottl.IgnoreError, componenttest.NewNopTelemetrySettings())
assert.NoError(b, err)
b.ResetTimer()
for n := 0; n < b.N; n++ {
Expand Down

0 comments on commit dc2f6e1

Please sign in to comment.