-
Notifications
You must be signed in to change notification settings - Fork 2.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[chore][connector/routing] Add internal package plogutil for splittin…
…g logs
- Loading branch information
1 parent
34c776f
commit dd69a63
Showing
56 changed files
with
4,378 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package plogutil // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/routingconnector/internal/plogutil" | ||
|
|
||
| import ( | ||
| "go.opentelemetry.io/collector/pdata/plog" | ||
| ) | ||
|
|
||
| // MoveResourcesIf calls f sequentially for each ResourceLogs present in the first plog.Logs. | ||
| // If f returns true, the element is removed from the first plog.Logs and added to the second plog.Logs. | ||
| func MoveResourcesIf(from, to plog.Logs, f func(plog.ResourceLogs) bool) { | ||
| from.ResourceLogs().RemoveIf(func(rl plog.ResourceLogs) bool { | ||
| if !f(rl) { | ||
| return false | ||
| } | ||
| rl.CopyTo(to.ResourceLogs().AppendEmpty()) | ||
| return true | ||
| }) | ||
| } | ||
|
|
||
| // MoveRecordsWithContextIf calls f sequentially for each LogRecord present in the first plog.Logs. | ||
| // If f returns true, the element is removed from the first plog.Logs and added to the second plog.Logs. | ||
| // Notably, the Resource and Scope associated with the LogRecord are created in the second plog.Logs only once. | ||
| // Resources or Scopes are removed from the original if they become empty. All ordering is preserved. | ||
| func MoveRecordsWithContextIf(from, to plog.Logs, f func(plog.ResourceLogs, plog.ScopeLogs, plog.LogRecord) bool) { | ||
| rls := from.ResourceLogs() | ||
| for i := 0; i < rls.Len(); i++ { | ||
| rl := rls.At(i) | ||
| sls := rl.ScopeLogs() | ||
| var rlCopy *plog.ResourceLogs | ||
| for j := 0; j < sls.Len(); j++ { | ||
| sl := sls.At(j) | ||
| lrs := sl.LogRecords() | ||
| var slCopy *plog.ScopeLogs | ||
| lrs.RemoveIf(func(lr plog.LogRecord) bool { | ||
| if !f(rl, sl, lr) { | ||
| return false | ||
| } | ||
| if rlCopy == nil { | ||
| rlc := to.ResourceLogs().AppendEmpty() | ||
| rlCopy = &rlc | ||
| rl.Resource().CopyTo(rlCopy.Resource()) | ||
| rlCopy.SetSchemaUrl(rl.SchemaUrl()) | ||
| } | ||
| if slCopy == nil { | ||
| slc := rlCopy.ScopeLogs().AppendEmpty() | ||
| slCopy = &slc | ||
| sl.Scope().CopyTo(slCopy.Scope()) | ||
| slCopy.SetSchemaUrl(sl.SchemaUrl()) | ||
| } | ||
| lr.CopyTo(slCopy.LogRecords().AppendEmpty()) | ||
| return true | ||
| }) | ||
| } | ||
| sls.RemoveIf(func(sl plog.ScopeLogs) bool { | ||
| return sl.LogRecords().Len() == 0 | ||
| }) | ||
| } | ||
| rls.RemoveIf(func(rl plog.ResourceLogs) bool { | ||
| return rl.ScopeLogs().Len() == 0 | ||
| }) | ||
| } |
161 changes: 161 additions & 0 deletions
161
connector/routingconnector/internal/plogutil/logs_test.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,161 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package plogutil_test | ||
|
|
||
| import ( | ||
| "path/filepath" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| "go.opentelemetry.io/collector/pdata/plog" | ||
|
|
||
| "github.com/open-telemetry/opentelemetry-collector-contrib/connector/routingconnector/internal/plogutil" | ||
| "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden" | ||
| "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest" | ||
| ) | ||
|
|
||
| func TestMoveResourcesIf(t *testing.T) { | ||
| testCases := []struct { | ||
| name string | ||
| condition func(plog.ResourceLogs) bool | ||
| }{ | ||
| { | ||
| name: "move_none", | ||
| condition: func(rl plog.ResourceLogs) bool { | ||
| return false | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_all", | ||
| condition: func(rl plog.ResourceLogs) bool { | ||
| return true | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_one", | ||
| condition: func(rl plog.ResourceLogs) bool { | ||
| rname, ok := rl.Resource().Attributes().Get("resourceName") | ||
| return ok && rname.AsString() == "resourceA" | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_to_preexisting", | ||
| condition: func(rl plog.ResourceLogs) bool { | ||
| rname, ok := rl.Resource().Attributes().Get("resourceName") | ||
| return ok && rname.AsString() == "resourceB" | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| for _, tt := range testCases { | ||
| t.Run(tt.name, func(t *testing.T) { | ||
| // Load up a fresh copy of the input for each test, since it may be modified in place. | ||
| from, err := golden.ReadLogs(filepath.Join("testdata", "resource", tt.name, "from.yaml")) | ||
| require.NoError(t, err) | ||
|
|
||
| to, err := golden.ReadLogs(filepath.Join("testdata", "resource", tt.name, "to.yaml")) | ||
| require.NoError(t, err) | ||
|
|
||
| fromModifed, err := golden.ReadLogs(filepath.Join("testdata", "resource", tt.name, "from_modified.yaml")) | ||
| require.NoError(t, err) | ||
|
|
||
| toModified, err := golden.ReadLogs(filepath.Join("testdata", "resource", tt.name, "to_modified.yaml")) | ||
| require.NoError(t, err) | ||
|
|
||
| plogutil.MoveResourcesIf(from, to, tt.condition) | ||
|
|
||
| assert.NoError(t, plogtest.CompareLogs(fromModifed, from), "from not modified as expected") | ||
| assert.NoError(t, plogtest.CompareLogs(toModified, to), "to not as expected") | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| func TestMoveRecordsWithContextIf(t *testing.T) { | ||
| testCases := []struct { | ||
| name string | ||
| condition func(plog.ResourceLogs, plog.ScopeLogs, plog.LogRecord) bool | ||
| }{ | ||
| { | ||
| name: "move_none", | ||
| condition: func(plog.ResourceLogs, plog.ScopeLogs, plog.LogRecord) bool { | ||
| return false | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_all", | ||
| condition: func(plog.ResourceLogs, plog.ScopeLogs, plog.LogRecord) bool { | ||
| return true | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_all_from_one_resource", | ||
| condition: func(rl plog.ResourceLogs, _ plog.ScopeLogs, _ plog.LogRecord) bool { | ||
| rname, ok := rl.Resource().Attributes().Get("resourceName") | ||
| return ok && rname.AsString() == "resourceB" | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_all_from_one_scope", | ||
| condition: func(rl plog.ResourceLogs, sl plog.ScopeLogs, _ plog.LogRecord) bool { | ||
| rname, ok := rl.Resource().Attributes().Get("resourceName") | ||
| return ok && rname.AsString() == "resourceB" && sl.Scope().Name() == "scopeA" | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_all_from_one_scope_in_each_resource", | ||
| condition: func(_ plog.ResourceLogs, sl plog.ScopeLogs, _ plog.LogRecord) bool { | ||
| return sl.Scope().Name() == "scopeB" | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_one", | ||
| condition: func(rl plog.ResourceLogs, sl plog.ScopeLogs, lr plog.LogRecord) bool { | ||
| rname, ok := rl.Resource().Attributes().Get("resourceName") | ||
| return ok && rname.AsString() == "resourceA" && sl.Scope().Name() == "scopeB" && lr.Body().AsString() == "logB" | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_one_from_each_scope", | ||
| condition: func(_ plog.ResourceLogs, _ plog.ScopeLogs, lr plog.LogRecord) bool { | ||
| return lr.Body().AsString() == "logA" | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_one_from_each_scope_in_one_resource", | ||
| condition: func(rl plog.ResourceLogs, _ plog.ScopeLogs, lr plog.LogRecord) bool { | ||
| rname, ok := rl.Resource().Attributes().Get("resourceName") | ||
| return ok && rname.AsString() == "resourceB" && lr.Body().AsString() == "logA" | ||
| }, | ||
| }, | ||
| { | ||
| name: "move_some_to_preexisting", | ||
| condition: func(_ plog.ResourceLogs, sl plog.ScopeLogs, _ plog.LogRecord) bool { | ||
| return sl.Scope().Name() == "scopeB" | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| for _, tt := range testCases { | ||
| t.Run(tt.name, func(t *testing.T) { | ||
| // Load up a fresh copy of the input for each test, since it may be modified in place. | ||
| from, err := golden.ReadLogs(filepath.Join("testdata", "record", tt.name, "from.yaml")) | ||
| require.NoError(t, err) | ||
|
|
||
| to, err := golden.ReadLogs(filepath.Join("testdata", "record", tt.name, "to.yaml")) | ||
| require.NoError(t, err) | ||
|
|
||
| fromModifed, err := golden.ReadLogs(filepath.Join("testdata", "record", tt.name, "from_modified.yaml")) | ||
| require.NoError(t, err) | ||
|
|
||
| toModified, err := golden.ReadLogs(filepath.Join("testdata", "record", tt.name, "to_modified.yaml")) | ||
| require.NoError(t, err) | ||
|
|
||
| plogutil.MoveRecordsWithContextIf(from, to, tt.condition) | ||
|
|
||
| assert.NoError(t, plogtest.CompareLogs(fromModifed, from), "from not modified as expected") | ||
| assert.NoError(t, plogtest.CompareLogs(toModified, to), "to not as expected") | ||
| }) | ||
| } | ||
| } |
Oops, something went wrong.