[pdatautil] Add function for splitting log records while preserving c…

…ontext

.chloggen/split-log-records.yaml

@@ -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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: pkg/pdatautil
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add `MoveLogRecordsWithContextIf` function to allow separation of log records while preserving context.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35623]
+
+# (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: [api]

exporter/opensearchexporter/go.mod

@@ -82,3 +82,5 @@ require (
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil => ../../pkg/pdatautil
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden => ../../pkg/golden
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest => ../../pkg/pdatatest

pkg/golden/go.mod

@@ -28,3 +28,5 @@ require (
 )
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil => ../pdatautil
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest => ../pdatatest

pkg/pdatautil/go.mod

@@ -4,6 +4,8 @@ go 1.22.0
 
 require (
 	github.com/cespare/xxhash/v2 v2.3.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.111.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.111.0
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector/pdata v1.17.0
 	go.uber.org/goleak v1.3.0
@@ -12,7 +14,10 @@ require (
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/kr/text v0.2.0 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/net v0.28.0 // indirect
@@ -28,3 +33,7 @@ retract (
 	v0.76.2
 	v0.76.1
 )
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest => ../pdatatest
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden => ../golden

pkg/pdatautil/logs.go

@@ -0,0 +1,59 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package pdatautil // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil"
+
+import "go.opentelemetry.io/collector/pdata/plog"
+
+// MoveLogRecordsWithContextIf calls f sequentially for each LogRecord present in the plog.Logs.
+// If f returns true, the element is removed from the original plog.Logs and added to the returned plog.Logs.
+// Notably, the Resource and Scope associated with the LogRecord are recreated in the returned plog.Logs as necessary.
+// Resources or Scopes are added to the result only if necessary, and removed from the original if they become empty.
+// All ordering is preserved.
+func MoveLogRecordsWithContextIf(ld plog.Logs, f func(plog.LogRecord) bool) plog.Logs {
+	result := plog.NewLogs()
+	rls := ld.ResourceLogs()
+	for i := 0; i < rls.Len(); i++ {
+		if rls.Len() == 0 {
+			continue // Don't remove empty
+		}
+		rl := rls.At(i)
+		sls := rl.ScopeLogs()
+		var rlCopy *plog.ResourceLogs
+		for j := 0; j < sls.Len(); j++ {
+			if sls.Len() == 0 {
+				continue // Don't remove empty
+			}
+			sl := sls.At(j)
+			lrs := sl.LogRecords()
+			var slCopy *plog.ScopeLogs
+			moveWithContextIf := func(lr plog.LogRecord) bool {
+				if !f(lr) {
+					return false
+				}
+				if rlCopy == nil {
+					rlc := result.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
+			}
+			lrs.RemoveIf(moveWithContextIf)
+		}
+		sls.RemoveIf(func(sl plog.ScopeLogs) bool {
+			return sl.LogRecords().Len() == 0
+		})
+	}
+	rls.RemoveIf(func(rl plog.ResourceLogs) bool {
+		return rl.ScopeLogs().Len() == 0
+	})
+	return result
+}

pkg/pdatautil/logs_test.go

@@ -0,0 +1,87 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package pdatautil_test
+
+import (
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/plog"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/plogtest"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil"
+)
+
+func TestMoveLogRecordsWithContextIf(t *testing.T) {
+	testCases := []struct {
+		name      string
+		condition func(plog.LogRecord) bool
+	}{
+		{
+			name: "move_none",
+			condition: func(plog.LogRecord) bool {
+				return false
+			},
+		},
+		{
+			name: "move_all",
+			condition: func(plog.LogRecord) bool {
+				return true
+			},
+		},
+		{
+			name: "move_one",
+			condition: func(lr plog.LogRecord) bool {
+				return lr.Body().AsString() == "resourceA, scopeB, logB"
+			},
+		},
+		{
+			name: "move_all_from_one_scope",
+			condition: func(lr plog.LogRecord) bool {
+				return strings.HasPrefix(lr.Body().AsString(), "resourceB, scopeA")
+			},
+		},
+		{
+			name: "move_all_from_one_resource",
+			condition: func(lr plog.LogRecord) bool {
+				return strings.HasPrefix(lr.Body().AsString(), "resourceB")
+			},
+		},
+		{
+			name: "move_one_from_each_scope",
+			condition: func(lr plog.LogRecord) bool {
+				return strings.HasSuffix(lr.Body().AsString(), "logA")
+			},
+		},
+		{
+			name: "move_all_from_one_scope_in_each_resource",
+			condition: func(lr plog.LogRecord) bool {
+				return strings.Contains(lr.Body().AsString(), "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.
+			input, err := golden.ReadLogs(filepath.Join("testdata", "logs", "input.yaml"))
+			require.NoError(t, err)
+
+			expectModified, err := golden.ReadLogs(filepath.Join("testdata", "logs", tt.name, "modified.yaml"))
+			require.NoError(t, err)
+
+			expectResult, err := golden.ReadLogs(filepath.Join("testdata", "logs", tt.name, "returned.yaml"))
+			require.NoError(t, err)
+
+			result := pdatautil.MoveLogRecordsWithContextIf(input, tt.condition)
+
+			assert.NoError(t, plogtest.CompareLogs(expectModified, input), "input not modified as expected")
+			assert.NoError(t, plogtest.CompareLogs(expectResult, result), "result not as expected")
+		})
+	}
+}

pkg/pdatautil/testdata/logs/input.yaml

@@ -0,0 +1,141 @@
+resourceLogs:
+  - resource:
+      attributes:
+        - key: resourceAttrOne
+          value:
+            stringValue: resourceA
+        - key: resourceAttrTwo
+          value:
+            stringValue: resourceA
+    schemaUrl: https://opentelemetry.io/schemas/1.6.1
+    scopeLogs:
+      - attributes:
+          - key: scopeAttrOne
+            value:
+              stringValue: scopeA
+          - key: scopeAttrTwo
+            value:
+              stringValue: scopeA
+        logRecords:
+          - attributes:
+              - key: logAttrOne
+                value:
+                  stringValue: logA
+              - key: logAttrTwo
+                value:
+                  stringValue: logA
+            body:
+              stringValue: resourceA, scopeA, logA
+          - attributes:
+              - key: logAttrOne
+                value:
+                  stringValue: logB
+              - key: logAttrTwo
+                value:
+                  stringValue: logB
+            body:
+              stringValue: resourceA, scopeA, logB
+        schemaUrl: https://opentelemetry.io/schemas/1.6.1
+        scope:
+          name: resourceA, scopeA
+          version: v0.1.0
+      - attributes:
+          - key: scopeAttrOne
+            value:
+              stringValue: scopeB
+          - key: scopeAttrTwo
+            value:
+              stringValue: scopeB
+        logRecords:
+          - attributes:
+              - key: logAttrOne
+                value:
+                  stringValue: logA
+              - key: logAttrTwo
+                value:
+                  stringValue: logA
+            body:
+              stringValue: resourceA, scopeB, logA
+          - attributes:
+              - key: logAttrOne
+                value:
+                  stringValue: logB
+              - key: logAttrTwo
+                value:
+                  stringValue: logB
+            body:
+              stringValue: resourceA, scopeB, logB
+        schemaUrl: https://opentelemetry.io/schemas/1.6.1
+        scope:
+          name: resourceA, scopeB
+          version: v0.1.0
+  - resource:
+      attributes:
+        - key: resourceAttrOne
+          value:
+            stringValue: resourceB
+        - key: resourceAttrTwo
+          value:
+            stringValue: resourceB
+    schemaUrl: https://opentelemetry.io/schemas/1.6.1
+    scopeLogs:
+      - attributes:
+          - key: scopeAttrOne
+            value:
+              stringValue: scopeA
+          - key: scopeAttrTwo
+            value:
+              stringValue: scopeA
+        logRecords:
+          - attributes:
+              - key: logAttrOne
+                value:
+                  stringValue: logA
+              - key: logAttrTwo
+                value:
+                  stringValue: logA
+            body:
+              stringValue: resourceB, scopeA, logA
+          - attributes:
+              - key: logAttrOne
+                value:
+                  stringValue: logB
+              - key: logAttrTwo
+                value:
+                  stringValue: logB
+            body:
+              stringValue: resourceB, scopeA, logB
+        schemaUrl: https://opentelemetry.io/schemas/1.6.1
+        scope:
+          name: resourceB, scopeA
+          version: v0.1.0
+      - attributes:
+          - key: scopeAttrOne
+            value:
+              stringValue: scopeB
+          - key: scopeAttrTwo
+            value:
+              stringValue: scopeB
+        logRecords:
+          - attributes:
+              - key: logAttrOne
+                value:
+                  stringValue: logA
+              - key: logAttrTwo
+                value:
+                  stringValue: logA
+            body:
+              stringValue: resourceB, scopeB, logA
+          - attributes:
+              - key: logAttrOne
+                value:
+                  stringValue: logB
+              - key: logAttrTwo
+                value:
+                  stringValue: logB
+            body:
+              stringValue: resourceB, scopeB, logB
+        schemaUrl: https://opentelemetry.io/schemas/1.6.1
+        scope:
+          name: resourceB, scopeB
+          version: v0.1.0

pkg/pdatautil/testdata/logs/move_all/modified.yaml

@@ -0,0 +1 @@
+