Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[processor/transform] Fix issue where ErrorMode wasn't used. #19629

Merged
merged 3 commits into from
Mar 14, 2023
Merged

[processor/transform] Fix issue where ErrorMode wasn't used. #19629

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1df24f8
Fix issue where ErrorMode wasn't used.
TylerHelmuth Mar 13, 2023
8aec8cb
add changelog
TylerHelmuth Mar 13, 2023
285b172
Add more tests
TylerHelmuth Mar 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions .chloggen/tp-fix-error-mode-usage.yaml
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Doesn't need its own version, it is provided by parserCollection.

}

type LogParserCollectionOption func(*LogParserCollection) error
Expand Down
1 change: 0 additions & 1 deletion processor/transformprocessor/internal/common/traces.go
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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