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/routing] Fix statement not eval in order #34999

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
+222 −27
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
48d676d
fix statement eval not in order
Frapschen Sep 4, 2024
471a0e4
fix lint
Frapschen Sep 4, 2024
c583bab
Merge branch 'main' into fix-router-random-statements
djaglowski Sep 5, 2024
1f87da8
make genotelcontribcol
Frapschen Sep 5, 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: []
2 changes: 1 addition & 1 deletion cmd/otelcontribcol/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -671,7 +671,7 @@ require (
github.com/open-telemetry/otel-arrow v0.25.0 // indirect
github.com/opencontainers/go-digest v1.0.0 // indirect
github.com/opencontainers/image-spec v1.1.0 // indirect
github.com/opencontainers/runc v1.1.14 // indirect
github.com/opencontainers/runc v1.1.13 // indirect
Copy link
Contributor

Choose a reason for hiding this comment

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

was this change intended?

github.com/opencontainers/runtime-spec v1.2.0 // indirect
github.com/opencontainers/selinux v1.10.0 // indirect
github.com/opensearch-project/opensearch-go/v2 v2.3.0 // indirect
Expand Down
4 changes: 2 additions & 2 deletions cmd/otelcontribcol/go.sum
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

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 @@ -116,21 +116,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 @@ -151,14 +151,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
164 changes: 164 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 @@ -127,6 +128,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[component.DataType]map[component.ID]component.Component{
component.DataTypeLogs: {
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 @@ -182,6 +240,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[component.DataType]map[component.ID]component.Component{
component.DataTypeLogs: {
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 @@ -400,6 +511,59 @@ func TestLogsAreCorrectlySplitPerResourceAttributeWithOTTL(t *testing.T) {
})
}

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

host := newMockHost(map[component.DataType]map[component.ID]component.Component{
component.DataTypeLogs: {
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["__otel_enabled__"] == nil`,
Exporters: []string{"otlp/first"},
},
{
Statement: `delete_key(resource.attributes, "__otel_enabled__") where resource.attributes["__otel_enabled__"] == "true"`,
Exporters: []string{"otlp/first"},
},
{
Statement: `delete_key(resource.attributes, "__otel_enabled__") where resource.attributes["__otel_enabled__"] == "false"`,
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("__otel_enabled__", "false")
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")
})
}
}

// 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 @@ -114,7 +114,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 @@ -128,7 +128,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
32 changes: 18 additions & 14 deletions processor/routingprocessor/router.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ type router[E component.Component, K any] struct {
table []RoutingTableItem

defaultExporters []E
routes map[string]routingItem[E, K]
routes []routingItem[E, K]
}

// newRouter creates a new router instance with its type parameter constrained
Expand All @@ -44,12 +44,11 @@ func newRouter[E component.Component, K any](

table: table,
defaultExporterIDs: defaultExporterIDs,

routes: make(map[string]routingItem[E, K]),
}
}

type routingItem[E component.Component, K any] struct {
key string
exporters []E
statement *ottl.Statement[K]
}
Expand Down Expand Up @@ -90,17 +89,14 @@ func (r *router[E, K]) registerDefaultExporters(available map[component.ID]compo
// registerRouteExporters registers route exporters using the provided
// available exporters map to check if they were available.
func (r *router[E, K]) registerRouteExporters(available map[component.ID]component.Component) error {
var routes []routingItem[E, K]
for _, item := range r.table {
statement, err := r.getStatementFrom(item)
if err != nil {
return err
}

route, ok := r.routes[key(item)]
if !ok {
route.statement = statement
}

var exporters []E
for _, name := range item.Exporters {
e, err := r.extractExporter(name, available)
if errors.Is(err, errExporterNotFound) {
Expand All @@ -109,10 +105,16 @@ func (r *router[E, K]) registerRouteExporters(available map[component.ID]compone
if err != nil {
return err
}
route.exporters = append(route.exporters, e)
exporters = append(exporters, e)
}
r.routes[key(item)] = route
routes = append(routes, routingItem[E, K]{
key: key(item),
exporters: exporters,
statement: statement,
})
}
r.routes = routes

return nil
}

Expand Down Expand Up @@ -171,9 +173,11 @@ func (r *router[E, K]) extractExporter(name string, available map[component.ID]c
}

func (r *router[E, K]) getExporters(key string) []E {
e, ok := r.routes[key]
if !ok {
return r.defaultExporters
for _, route := range r.routes {
Copy link
Member

Choose a reason for hiding this comment

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

I wonder what are the performance implications of this change. This code is in the hot path. If you are using this change in production already, are you able to provide some insights here?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I haven't tested it in production yet, but according to the changed code, it does not break the loop behavior. Whether routes is a map or a slice, it always loops through all the elements in the routes. So, I would say it will not affect the performance negatively.

Copy link
Member

Choose a reason for hiding this comment

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

I'm not concerned about the behavior, I'm concerned about the performance. I just double-checked, and this doesn't seem like it's in the hot path, called only on the Start path. In any case, it would make me more comfortable if we had a benchmark comparing how it was before and how it is now.

if route.key == key {
return route.exporters
}
}
return e.exporters

return r.defaultExporters
}
Loading