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

[processor/routing] Fix statement not eval in order #34999

Open
wants to merge 7 commits into
base: main
Choose a base branch
from
Open

[processor/routing] Fix statement not eval in order #34999

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b3d9211
fix statement eval not in order
Frapschen Sep 4, 2024
92b6178
fix lint
Frapschen Sep 4, 2024
6a62c1b
git rebase
Frapschen Oct 8, 2024
b34879e
update
Frapschen Oct 8, 2024
ab30438
add benchmark
Frapschen Oct 8, 2024
72877ae
fix pipeline
Frapschen Oct 8, 2024
00caceb
Merge branch 'main' into fix-router-random-statements
codeboten Oct 31, 2024
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
27 changes: 27 additions & 0 deletions .chloggen/fix-router-random-statements.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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: bug_fix

# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
component: routingprocessor

# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
note: Fix OTTL statement not eval in order

# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
issues: [34860]

# (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: []
10 changes: 5 additions & 5 deletions processor/routingprocessor/logs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,21 +117,21 @@ func (p *logProcessor) route(ctx context.Context, l plog.Logs) error {
)

matchCount := len(p.router.routes)
for key, route := range p.router.routes {
for _, route := range p.router.routes {
_, isMatch, err := route.statement.Execute(ctx, ltx)
if err != nil {
if p.config.ErrorMode == ottl.PropagateError {
return err
}
p.group("", groups, p.router.defaultExporters, rlogs)
p.recordNonRoutedResourceLogs(ctx, key, rlogs)
p.recordNonRoutedResourceLogs(ctx, route.key, rlogs)
continue
}
if !isMatch {
matchCount--
continue
}
p.group(key, groups, route.exporters, rlogs)
p.group(route.key, groups, route.exporters, rlogs)
}

if matchCount == 0 {
Expand All @@ -152,14 +152,14 @@ func (p *logProcessor) group(
key string,
groups map[string]logsGroup,
exporters []exporter.Logs,
spans plog.ResourceLogs,
logs plog.ResourceLogs,
) {
group, ok := groups[key]
if !ok {
group.logs = plog.NewLogs()
group.exporters = exporters
}
spans.CopyTo(group.logs.ResourceLogs().AppendEmpty())
logs.CopyTo(group.logs.ResourceLogs().AppendEmpty())
groups[key] = group
}

Expand Down
222 changes: 222 additions & 0 deletions processor/routingprocessor/logs_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ package routingprocessor

import (
"context"
"fmt"
"testing"

"github.com/stretchr/testify/assert"
Expand Down Expand Up @@ -128,6 +129,63 @@ func TestLogs_RoutingWorks_Context(t *testing.T) {
})
}

func TestLogs_RoutingWorks_Context_Ordered(t *testing.T) {
defaultExp := &mockLogsExporter{}
lExpFirst := &mockLogsExporter{}
lExpSecond := &mockLogsExporter{}
lExpThird := &mockLogsExporter{}

host := newMockHost(map[pipeline.Signal]map[component.ID]component.Component{
pipeline.SignalLogs: {
component.MustNewID("otlp"): defaultExp,
component.MustNewIDWithName("otlp", "first"): lExpFirst,
component.MustNewIDWithName("otlp", "second"): lExpSecond,
component.MustNewIDWithName("otlp", "third"): lExpThird,
},
})

exp, err := newLogProcessor(noopTelemetrySettings, &Config{
FromAttribute: "X-Tenant",
AttributeSource: contextAttributeSource,
DefaultExporters: []string{"otlp"},
Table: []RoutingTableItem{
{
Value: "order-second",
Exporters: []string{"otlp/second"},
},
{
Value: "order-first",
Exporters: []string{"otlp/first"},
},
{
Value: "order-third",
Exporters: []string{"otlp/third"},
},
},
})
require.NoError(t, err)
require.NoError(t, exp.Start(context.Background(), host))

for i := 1; i <= 5; i++ {
t.Run(fmt.Sprintf("run %d time", i), func(t *testing.T) {
l := plog.NewLogs()
ll := l.ResourceLogs().AppendEmpty().ScopeLogs().AppendEmpty().LogRecords()
ll.AppendEmpty().Body().SetStr("this is a log")

assert.NoError(t, exp.ConsumeLogs(
metadata.NewIncomingContext(context.Background(), metadata.New(map[string]string{
"X-Tenant": "order-third",
})),
l,
))
assert.Empty(t, defaultExp.AllLogs())
assert.Empty(t, lExpFirst.AllLogs())
assert.Empty(t, lExpSecond.AllLogs())
assert.Len(t, lExpThird.AllLogs(), i, "log should only be routed to lExpThird exporter")
})
}
}

func TestLogs_RoutingWorks_ResourceAttribute(t *testing.T) {
defaultExp := &mockLogsExporter{}
lExp := &mockLogsExporter{}
Expand Down Expand Up @@ -183,6 +241,59 @@ func TestLogs_RoutingWorks_ResourceAttribute(t *testing.T) {
})
}

func TestLogs_RoutingWorks_ResourceAttribute_Ordered(t *testing.T) {
defaultExp := &mockLogsExporter{}
lExpFirst := &mockLogsExporter{}
lExpSecond := &mockLogsExporter{}
lExpThird := &mockLogsExporter{}

host := newMockHost(map[pipeline.Signal]map[component.ID]component.Component{
pipeline.SignalLogs: {
component.MustNewID("otlp"): defaultExp,
component.MustNewIDWithName("otlp", "first"): lExpFirst,
component.MustNewIDWithName("otlp", "second"): lExpSecond,
component.MustNewIDWithName("otlp", "third"): lExpThird,
},
})

exp, err := newLogProcessor(noopTelemetrySettings, &Config{
FromAttribute: "X-Tenant",
AttributeSource: resourceAttributeSource,
DefaultExporters: []string{"otlp"},
Table: []RoutingTableItem{
{
Value: "order-second",
Exporters: []string{"otlp/second"},
},
{
Value: "order-first",
Exporters: []string{"otlp/first"},
},
{
Value: "order-third",
Exporters: []string{"otlp/third"},
},
},
})
require.NoError(t, err)
require.NoError(t, exp.Start(context.Background(), host))

for i := 1; i <= 5; i++ {
t.Run(fmt.Sprintf("run %d time", i), func(t *testing.T) {
l := plog.NewLogs()
rl := l.ResourceLogs().AppendEmpty()
rl.Resource().Attributes().PutStr("X-Tenant", "order-third")
rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty().Body().SetStr("this is a log")

assert.NoError(t, exp.ConsumeLogs(context.Background(), l))
assert.Empty(t, defaultExp.AllLogs())
assert.Empty(t, lExpFirst.AllLogs())
assert.Empty(t, lExpSecond.AllLogs())
assert.Len(t, lExpThird.AllLogs(), i, "log should only be routed to lExpThird exporter")
})
}
}

func TestLogs_RoutingWorks_ResourceAttribute_DropsRoutingAttribute(t *testing.T) {
defaultExp := &mockLogsExporter{}
lExp := &mockLogsExporter{}
Expand Down Expand Up @@ -401,6 +512,117 @@ func TestLogsAreCorrectlySplitPerResourceAttributeWithOTTL(t *testing.T) {
})
}

func TestLogs_RoutingWorks_ResourceAttribute_WithOTTL_Ordered(t *testing.T) {
defaultExp := &mockLogsExporter{}
lExpFirst := &mockLogsExporter{}
lExpSecond := &mockLogsExporter{}

host := newMockHost(map[pipeline.Signal]map[component.ID]component.Component{
pipeline.SignalLogs: {
component.MustNewID("otlp"): defaultExp,
component.MustNewIDWithName("otlp", "first"): lExpFirst,
component.MustNewIDWithName("otlp", "second"): lExpSecond,
},
})

exp, err := newLogProcessor(noopTelemetrySettings, &Config{
FromAttribute: "__otel_enabled__",
AttributeSource: resourceAttributeSource,
DefaultExporters: []string{"otlp"},
Table: []RoutingTableItem{
Copy link
Member

Choose a reason for hiding this comment

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

To be honest, I find this test a bit confusing: the first two conditions will never be true for different reasons and the third is only resulting in true if the deletion happens based on the value of the key used in the where clause.

The statements could be just simple route statements?

route() where resource.attributes["__otel_enabled__"] == nil
route() where resource.attributes["__otel_enabled__"] == "true"
route() where resource.attributes["__otel_enabled__"] == "false" // this is the only one resulting in true

Or even better:

route() where resource.attributes["non-matching"] != nil // non-matching is not set, this is never true
route() where resource.attributes["non-matching"] == "true" // non-matching is not set, therefore, not "true"
route() where resource.attributes["matching"] == "true" // matches!

And in the main test code, have this attribute instead:

rl.Resource().Attributes().PutStr("matching", "true")

Copy link
Contributor Author

@Frapschen Frapschen Sep 6, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

@jpkrohling The test is designed to run 5 times, and I expected it to produce the same result each time. This is how expected statements work. If the statements are executed randomly (map[string]routingItem[E, K]), they can't yield consistent results.

BWT, the statements is from the issue's description. It can demonstrate that statements are executed in order or randomly will affect the routing results. so I used it in the test.

Copy link
Member

Choose a reason for hiding this comment

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

Running them 5 times is fine, my concern is about the readability of the test. While the statements on the issue description were enough to reproduce the problem, the tests should be easily understandable to our future selves without having to come back to this issue to get the full picture, IMO.

{
Statement: `route() where resource.attributes["non-matching"] != nil`,
Exporters: []string{"otlp/first"},
},
{
Statement: `route() where resource.attributes["non-matching"] == "true"`,
Exporters: []string{"otlp/first"},
},
{
Statement: `route() where resource.attributes["matching"] == "true"`,
Exporters: []string{"otlp/second"},
},
},
})
require.NoError(t, err)
require.NoError(t, exp.Start(context.Background(), host))

for i := 1; i <= 5; i++ {
t.Run(fmt.Sprintf("run %d time", i), func(t *testing.T) {
l := plog.NewLogs()
rl := l.ResourceLogs().AppendEmpty()
rl.Resource().Attributes().PutStr("matching", "true")
rl.ScopeLogs().AppendEmpty().LogRecords().AppendEmpty().Body().SetStr("this is a log")

assert.NoError(t, exp.ConsumeLogs(context.Background(), l))

assert.Empty(t, defaultExp.AllLogs())
assert.Empty(t, lExpFirst.AllLogs())

// we expect Statement eval in order and log should only be routed to lExpSecond exporter
assert.Len(t, lExpSecond.AllLogs(), i, "log should only be routed to lExpSecond exporter")
})
}
}

/*
# before
BenchmarkLogs_Routing
BenchmarkLogs_Routing-8 18951 63102 ns/op

# after
BenchmarkLogs_Routing
BenchmarkLogs_Routing-8 21171 57387 ns/op
*/
func BenchmarkLogs_Routing(t *testing.B) {
defaultExp := &mockLogsExporter{}
lExpFirst := &mockLogsExporter{}
lExpSecond := &mockLogsExporter{}

host := newMockHost(map[pipeline.Signal]map[component.ID]component.Component{
pipeline.SignalLogs: {
component.MustNewID("otlp"): defaultExp,
component.MustNewIDWithName("otlp", "first"): lExpFirst,
component.MustNewIDWithName("otlp", "second"): lExpSecond,
},
})

exp, err := newLogProcessor(noopTelemetrySettings, &Config{
FromAttribute: "__otel_enabled__",
AttributeSource: resourceAttributeSource,
DefaultExporters: []string{"otlp"},
Table: []RoutingTableItem{
{
Statement: `route() where resource.attributes["non-matching"] != nil`,
Exporters: []string{"otlp/first"},
},
{
Statement: `route() where resource.attributes["non-matching"] == "true"`,
Exporters: []string{"otlp/first"},
},
{
Statement: `route() where resource.attributes["matching"] == "true"`,
Exporters: []string{"otlp/second"},
},
},
})
require.NoError(t, err)
require.NoError(t, exp.Start(context.Background(), host))

mockLogs := plog.NewLogs()
for i := 0; i < 100; i++ {
rl := mockLogs.ResourceLogs().AppendEmpty()
rl.Resource().Attributes().PutStr("matching", "true")
}

for i := 0; i < t.N; i++ {
l := plog.NewLogs()
mockLogs.CopyTo(l)

assert.NoError(t, exp.ConsumeLogs(context.Background(), l))
}
}

// see https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/26462
func TestLogsAttributeWithOTTLDoesNotCauseCrash(t *testing.T) {
// prepare
Expand Down
4 changes: 2 additions & 2 deletions processor/routingprocessor/metrics.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,7 @@ func (p *metricsProcessor) route(ctx context.Context, tm pmetric.Metrics) error
)

matchCount := len(p.router.routes)
for key, route := range p.router.routes {
for _, route := range p.router.routes {
_, isMatch, err := route.statement.Execute(ctx, mtx)
if err != nil {
if p.config.ErrorMode == ottl.PropagateError {
Expand All @@ -129,7 +129,7 @@ func (p *metricsProcessor) route(ctx context.Context, tm pmetric.Metrics) error
matchCount--
continue
}
p.group(key, groups, route.exporters, rmetrics)
p.group(route.key, groups, route.exporters, rmetrics)
}

if matchCount == 0 {
Expand Down
Loading