Avoid extra copy of all data during routing (#44387)

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

.chloggen/avoid-extra-copy.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. receiver/filelog)
+component: connector/routing
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Avoid extra copy of all data during routing
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [44387]
+
+# (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: [user]

connector/routingconnector/logs.go

@@ -62,17 +62,17 @@ func (*logsConnector) Capabilities() consumer.Capabilities {
 func (c *logsConnector) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 	groups := make(map[consumer.Logs]plog.Logs)
 	var errs error
+	matched := plog.NewLogs()
 	for i := 0; i < len(c.router.routeSlice) && ld.ResourceLogs().Len() > 0; i++ {
 		route := c.router.routeSlice[i]
-		matchedLogs := plog.NewLogs()
 		switch route.statementContext {
 		case "request":
 			if route.requestCondition.matchRequest(ctx) {
-				groupAllLogs(groups, route.consumer, ld)
-				ld = plog.NewLogs() // all logs have been routed
+				// all logs are routed
+				ld.MoveTo(matched)
 			}
 		case "", "resource":
-			plogutil.MoveResourcesIf(ld, matchedLogs,
+			plogutil.MoveResourcesIf(ld, matched,
 				func(rl plog.ResourceLogs) bool {
 					rtx := ottlresource.NewTransformContext(rl.Resource(), rl)
 					_, isMatch, err := route.resourceStatement.Execute(ctx, rtx)
@@ -81,7 +81,7 @@ func (c *logsConnector) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 				},
 			)
 		case "log":
-			plogutil.MoveRecordsWithContextIf(ld, matchedLogs,
+			plogutil.MoveRecordsWithContextIf(ld, matched,
 				func(rl plog.ResourceLogs, sl plog.ScopeLogs, lr plog.LogRecord) bool {
 					ltx := ottllog.NewTransformContext(lr, sl.Scope(), rl.Resource(), sl, rl)
 					_, isMatch, err := route.logStatement.Execute(ctx, ltx)
@@ -94,9 +94,9 @@ func (c *logsConnector) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 			if c.config.ErrorMode == ottl.PropagateError {
 				return errs
 			}
-			groupAllLogs(groups, c.router.defaultConsumer, matchedLogs)
+			groupAllLogs(groups, c.router.defaultConsumer, matched)
 		}
-		groupAllLogs(groups, route.consumer, matchedLogs)
+		groupAllLogs(groups, route.consumer, matched)
 	}
 	// anything left wasn't matched by any route. Send to default consumer
 	groupAllLogs(groups, c.router.defaultConsumer, ld)
@@ -110,24 +110,17 @@ func groupAllLogs(
 	groups map[consumer.Logs]plog.Logs,
 	cons consumer.Logs,
 	logs plog.Logs,
-) {
-	for i := 0; i < logs.ResourceLogs().Len(); i++ {
-		groupLogs(groups, cons, logs.ResourceLogs().At(i))
-	}
-}
-
-func groupLogs(
-	groups map[consumer.Logs]plog.Logs,
-	cons consumer.Logs,
-	logs plog.ResourceLogs,
 ) {
 	if cons == nil {
 		return
 	}
+	if logs.ResourceLogs().Len() == 0 {
+		return
+	}
 	group, ok := groups[cons]
 	if !ok {
 		group = plog.NewLogs()
+		groups[cons] = group
 	}
-	logs.CopyTo(group.ResourceLogs().AppendEmpty())
-	groups[cons] = group
+	logs.ResourceLogs().MoveAndAppendTo(group.ResourceLogs())
 }

connector/routingconnector/metrics.go

@@ -63,17 +63,17 @@ func (*metricsConnector) Capabilities() consumer.Capabilities {
 func (c *metricsConnector) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
 	groups := make(map[consumer.Metrics]pmetric.Metrics)
 	var errs error
+	matched := pmetric.NewMetrics()
 	for i := 0; i < len(c.router.routeSlice) && md.ResourceMetrics().Len() > 0; i++ {
 		route := c.router.routeSlice[i]
-		matchedMetrics := pmetric.NewMetrics()
 		switch route.statementContext {
 		case "request":
 			if route.requestCondition.matchRequest(ctx) {
-				groupAllMetrics(groups, route.consumer, md)
-				md = pmetric.NewMetrics() // all metrics have been routed
+				// all metrics are routed
+				md.MoveTo(matched)
 			}
 		case "", "resource":
-			pmetricutil.MoveResourcesIf(md, matchedMetrics,
+			pmetricutil.MoveResourcesIf(md, matched,
 				func(rs pmetric.ResourceMetrics) bool {
 					rtx := ottlresource.NewTransformContext(rs.Resource(), rs)
 					_, isMatch, err := route.resourceStatement.Execute(ctx, rtx)
@@ -82,7 +82,7 @@ func (c *metricsConnector) ConsumeMetrics(ctx context.Context, md pmetric.Metric
 				},
 			)
 		case "metric":
-			pmetricutil.MoveMetricsWithContextIf(md, matchedMetrics,
+			pmetricutil.MoveMetricsWithContextIf(md, matched,
 				func(rm pmetric.ResourceMetrics, sm pmetric.ScopeMetrics, m pmetric.Metric) bool {
 					mtx := ottlmetric.NewTransformContext(m, sm.Metrics(), sm.Scope(), rm.Resource(), sm, rm)
 					_, isMatch, err := route.metricStatement.Execute(ctx, mtx)
@@ -91,7 +91,7 @@ func (c *metricsConnector) ConsumeMetrics(ctx context.Context, md pmetric.Metric
 				},
 			)
 		case "datapoint":
-			pmetricutil.MoveDataPointsWithContextIf(md, matchedMetrics,
+			pmetricutil.MoveDataPointsWithContextIf(md, matched,
 				func(rm pmetric.ResourceMetrics, sm pmetric.ScopeMetrics, m pmetric.Metric, dp any) bool {
 					dptx := ottldatapoint.NewTransformContext(dp, m, sm.Metrics(), sm.Scope(), rm.Resource(), sm, rm)
 					_, isMatch, err := route.dataPointStatement.Execute(ctx, dptx)
@@ -104,9 +104,9 @@ func (c *metricsConnector) ConsumeMetrics(ctx context.Context, md pmetric.Metric
 			if c.config.ErrorMode == ottl.PropagateError {
 				return errs
 			}
-			groupAllMetrics(groups, c.router.defaultConsumer, matchedMetrics)
+			groupAllMetrics(groups, c.router.defaultConsumer, matched)
 		}
-		groupAllMetrics(groups, route.consumer, matchedMetrics)
+		groupAllMetrics(groups, route.consumer, matched)
 	}
 	// anything left wasn't matched by any route. Send to default consumer
 	groupAllMetrics(groups, c.router.defaultConsumer, md)
@@ -121,23 +121,16 @@ func groupAllMetrics(
 	cons consumer.Metrics,
 	metrics pmetric.Metrics,
 ) {
-	for i := 0; i < metrics.ResourceMetrics().Len(); i++ {
-		groupMetrics(groups, cons, metrics.ResourceMetrics().At(i))
+	if cons == nil {
+		return
 	}
-}
-
-func groupMetrics(
-	groups map[consumer.Metrics]pmetric.Metrics,
-	consumer consumer.Metrics,
-	metrics pmetric.ResourceMetrics,
-) {
-	if consumer == nil {
+	if metrics.ResourceMetrics().Len() == 0 {
 		return
 	}
-	group, ok := groups[consumer]
+	group, ok := groups[cons]
 	if !ok {
 		group = pmetric.NewMetrics()
+		groups[cons] = group
 	}
-	metrics.CopyTo(group.ResourceMetrics().AppendEmpty())
-	groups[consumer] = group
+	metrics.ResourceMetrics().MoveAndAppendTo(group.ResourceMetrics())
 }

connector/routingconnector/traces.go

@@ -62,17 +62,17 @@ func (*tracesConnector) Capabilities() consumer.Capabilities {
 func (c *tracesConnector) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
 	groups := make(map[consumer.Traces]ptrace.Traces)
 	var errs error
+	matched := ptrace.NewTraces()
 	for i := 0; i < len(c.router.routeSlice) && td.ResourceSpans().Len() > 0; i++ {
 		route := c.router.routeSlice[i]
-		matchedSpans := ptrace.NewTraces()
 		switch route.statementContext {
 		case "request":
 			if route.requestCondition.matchRequest(ctx) {
-				groupAllTraces(groups, route.consumer, td)
-				td = ptrace.NewTraces() // all traces have been routed
+				// all traces are routed
+				td.MoveTo(matched)
 			}
 		case "", "resource":
-			ptraceutil.MoveResourcesIf(td, matchedSpans,
+			ptraceutil.MoveResourcesIf(td, matched,
 				func(rs ptrace.ResourceSpans) bool {
 					rtx := ottlresource.NewTransformContext(rs.Resource(), rs)
 					_, isMatch, err := route.resourceStatement.Execute(ctx, rtx)
@@ -81,7 +81,7 @@ func (c *tracesConnector) ConsumeTraces(ctx context.Context, td ptrace.Traces) e
 				},
 			)
 		case "span":
-			ptraceutil.MoveSpansWithContextIf(td, matchedSpans,
+			ptraceutil.MoveSpansWithContextIf(td, matched,
 				func(rs ptrace.ResourceSpans, ss ptrace.ScopeSpans, s ptrace.Span) bool {
 					mtx := ottlspan.NewTransformContext(s, ss.Scope(), rs.Resource(), ss, rs)
 					_, isMatch, err := route.spanStatement.Execute(ctx, mtx)
@@ -94,9 +94,9 @@ func (c *tracesConnector) ConsumeTraces(ctx context.Context, td ptrace.Traces) e
 			if c.config.ErrorMode == ottl.PropagateError {
 				return errs
 			}
-			groupAllTraces(groups, c.router.defaultConsumer, matchedSpans)
+			groupAllTraces(groups, c.router.defaultConsumer, matched)
 		}
-		groupAllTraces(groups, route.consumer, matchedSpans)
+		groupAllTraces(groups, route.consumer, matched)
 	}
 	// anything left wasn't matched by any route. Send to default consumer
 	groupAllTraces(groups, c.router.defaultConsumer, td)
@@ -110,24 +110,17 @@ func groupAllTraces(
 	groups map[consumer.Traces]ptrace.Traces,
 	cons consumer.Traces,
 	traces ptrace.Traces,
-) {
-	for i := 0; i < traces.ResourceSpans().Len(); i++ {
-		groupTraces(groups, cons, traces.ResourceSpans().At(i))
-	}
-}
-
-func groupTraces(
-	groups map[consumer.Traces]ptrace.Traces,
-	cons consumer.Traces,
-	spans ptrace.ResourceSpans,
 ) {
 	if cons == nil {
 		return
 	}
+	if traces.ResourceSpans().Len() == 0 {
+		return
+	}
 	group, ok := groups[cons]
 	if !ok {
 		group = ptrace.NewTraces()
+		groups[cons] = group
 	}
-	spans.CopyTo(group.ResourceSpans().AppendEmpty())
-	groups[cons] = group
+	traces.ResourceSpans().MoveAndAppendTo(group.ResourceSpans())
 }