[connector/routing] Avoid unnecessary copy of the data

.chloggen/rm-copies-routing.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: routingconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Avoid unnecessary copy of the data in routing connector
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37946]
+
+# (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/internal/plogutil/logs.go

@@ -14,7 +14,7 @@ func MoveResourcesIf(from, to plog.Logs, f func(plog.ResourceLogs) bool) {
 		if !f(rl) {
 			return false
 		}
-		rl.CopyTo(to.ResourceLogs().AppendEmpty())
+		rl.MoveTo(to.ResourceLogs().AppendEmpty())
 		return true
 	})
 }
@@ -25,12 +25,10 @@ func MoveResourcesIf(from, to plog.Logs, f func(plog.ResourceLogs) bool) {
 // 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)
+	rls.RemoveIf(func(rl plog.ResourceLogs) bool {
 		sls := rl.ScopeLogs()
 		var rlCopy *plog.ResourceLogs
-		for j := 0; j < sls.Len(); j++ {
-			sl := sls.At(j)
+		sls.RemoveIf(func(sl plog.ScopeLogs) bool {
 			lrs := sl.LogRecords()
 			var slCopy *plog.ScopeLogs
 			lrs.RemoveIf(func(lr plog.LogRecord) bool {
@@ -49,15 +47,11 @@ func MoveRecordsWithContextIf(from, to plog.Logs, f func(plog.ResourceLogs, plog
 					sl.Scope().CopyTo(slCopy.Scope())
 					slCopy.SetSchemaUrl(sl.SchemaUrl())
 				}
-				lr.CopyTo(slCopy.LogRecords().AppendEmpty())
+				lr.MoveTo(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
 	})
 }

connector/routingconnector/internal/plogutil/logs_test.go

@@ -233,3 +233,16 @@ func TestMoveRecordsWithContextIf(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMoveResourcesIfLogs(b *testing.B) {
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		from := plogutiltest.NewLogs("AB", "CD", "EF")
+		to := plog.NewLogs()
+		plogutil.MoveResourcesIf(from, to, func(plog.ResourceLogs) bool {
+			return true
+		})
+		assert.Equal(b, 0, from.LogRecordCount())
+		assert.Equal(b, 8, to.LogRecordCount())
+	}
+}

connector/routingconnector/internal/pmetricutil/metrics.go

@@ -12,7 +12,7 @@ func MoveResourcesIf(from, to pmetric.Metrics, f func(pmetric.ResourceMetrics) b
 		if !f(rs) {
 			return false
 		}
-		rs.CopyTo(to.ResourceMetrics().AppendEmpty())
+		rs.MoveTo(to.ResourceMetrics().AppendEmpty())
 		return true
 	})
 }
@@ -23,39 +23,27 @@ func MoveResourcesIf(from, to pmetric.Metrics, f func(pmetric.ResourceMetrics) b
 // Resources or Scopes are removed from the original if they become empty. All ordering is preserved.
 func MoveMetricsWithContextIf(from, to pmetric.Metrics, f func(pmetric.ResourceMetrics, pmetric.ScopeMetrics, pmetric.Metric) bool) {
 	rms := from.ResourceMetrics()
-	for i := 0; i < rms.Len(); i++ {
-		rm := rms.At(i)
+	rms.RemoveIf(func(rm pmetric.ResourceMetrics) bool {
 		sms := rm.ScopeMetrics()
 		var rmCopy *pmetric.ResourceMetrics
-		for j := 0; j < sms.Len(); j++ {
-			sm := sms.At(j)
+		sms.RemoveIf(func(sm pmetric.ScopeMetrics) bool {
 			ms := sm.Metrics()
 			var smCopy *pmetric.ScopeMetrics
 			ms.RemoveIf(func(m pmetric.Metric) bool {
 				if !f(rm, sm, m) {
 					return false
 				}
 				if rmCopy == nil {
-					rmc := to.ResourceMetrics().AppendEmpty()
-					rmCopy = &rmc
-					rm.Resource().CopyTo(rmCopy.Resource())
-					rmCopy.SetSchemaUrl(rm.SchemaUrl())
+					rmCopy = copyResourceMetrics(rm, to.ResourceMetrics())
 				}
 				if smCopy == nil {
-					smc := rmCopy.ScopeMetrics().AppendEmpty()
-					smCopy = &smc
-					sm.Scope().CopyTo(smCopy.Scope())
-					smCopy.SetSchemaUrl(sm.SchemaUrl())
+					smCopy = copyScopeMetrics(sm, rmCopy.ScopeMetrics())
 				}
-				m.CopyTo(smCopy.Metrics().AppendEmpty())
+				m.MoveTo(smCopy.Metrics().AppendEmpty())
 				return true
 			})
-		}
-		sms.RemoveIf(func(sm pmetric.ScopeMetrics) bool {
 			return sm.Metrics().Len() == 0
 		})
-	}
-	rms.RemoveIf(func(rm pmetric.ResourceMetrics) bool {
 		return rm.ScopeMetrics().Len() == 0
 	})
 }
@@ -66,16 +54,13 @@ func MoveMetricsWithContextIf(from, to pmetric.Metrics, f func(pmetric.ResourceM
 // Resources, Scopes, or Metrics are removed from the original if they become empty. All ordering is preserved.
 func MoveDataPointsWithContextIf(from, to pmetric.Metrics, f func(pmetric.ResourceMetrics, pmetric.ScopeMetrics, pmetric.Metric, any) bool) {
 	rms := from.ResourceMetrics()
-	for i := 0; i < rms.Len(); i++ {
-		rm := rms.At(i)
+	rms.RemoveIf(func(rm pmetric.ResourceMetrics) bool {
 		sms := rm.ScopeMetrics()
 		var rmCopy *pmetric.ResourceMetrics
-		for j := 0; j < sms.Len(); j++ {
-			sm := sms.At(j)
+		sms.RemoveIf(func(sm pmetric.ScopeMetrics) bool {
 			ms := sm.Metrics()
 			var smCopy *pmetric.ScopeMetrics
-			for k := 0; k < ms.Len(); k++ {
-				m := ms.At(k)
+			ms.RemoveIf(func(m pmetric.Metric) bool {
 				var mCopy *pmetric.Metric
 
 				// TODO condense this code
@@ -87,168 +72,127 @@ func MoveDataPointsWithContextIf(from, to pmetric.Metrics, f func(pmetric.Resour
 							return false
 						}
 						if rmCopy == nil {
-							rmc := to.ResourceMetrics().AppendEmpty()
-							rmCopy = &rmc
-							rm.Resource().CopyTo(rmCopy.Resource())
-							rmCopy.SetSchemaUrl(rm.SchemaUrl())
+							rmCopy = copyResourceMetrics(rm, to.ResourceMetrics())
 						}
 						if smCopy == nil {
-							smc := rmCopy.ScopeMetrics().AppendEmpty()
-							smCopy = &smc
-							sm.Scope().CopyTo(smCopy.Scope())
-							smCopy.SetSchemaUrl(sm.SchemaUrl())
+							smCopy = copyScopeMetrics(sm, rmCopy.ScopeMetrics())
 						}
 						if mCopy == nil {
-							mc := smCopy.Metrics().AppendEmpty()
-							mCopy = &mc
-							mCopy.SetName(m.Name())
-							mCopy.SetDescription(m.Description())
-							mCopy.SetUnit(m.Unit())
+							mCopy = copyMetricDescription(m, smCopy.Metrics())
 							mCopy.SetEmptyGauge()
 						}
-						dp.CopyTo(mCopy.Gauge().DataPoints().AppendEmpty())
+						dp.MoveTo(mCopy.Gauge().DataPoints().AppendEmpty())
 						return true
 					})
+					return dps.Len() == 0
 				case pmetric.MetricTypeSum:
 					dps := m.Sum().DataPoints()
 					dps.RemoveIf(func(dp pmetric.NumberDataPoint) bool {
 						if !f(rm, sm, m, dp) {
 							return false
 						}
 						if rmCopy == nil {
-							rmc := to.ResourceMetrics().AppendEmpty()
-							rmCopy = &rmc
-							rm.Resource().CopyTo(rmCopy.Resource())
-							rmCopy.SetSchemaUrl(rm.SchemaUrl())
+							rmCopy = copyResourceMetrics(rm, to.ResourceMetrics())
 						}
 						if smCopy == nil {
-							smc := rmCopy.ScopeMetrics().AppendEmpty()
-							smCopy = &smc
-							sm.Scope().CopyTo(smCopy.Scope())
-							smCopy.SetSchemaUrl(sm.SchemaUrl())
+							smCopy = copyScopeMetrics(sm, rmCopy.ScopeMetrics())
 						}
 						if mCopy == nil {
-							mc := smCopy.Metrics().AppendEmpty()
-							mCopy = &mc
-							mCopy.SetName(m.Name())
-							mCopy.SetDescription(m.Description())
-							mCopy.SetUnit(m.Unit())
+							mCopy = copyMetricDescription(m, smCopy.Metrics())
 							mCopy.SetEmptySum()
 						}
-						dp.CopyTo(mCopy.Sum().DataPoints().AppendEmpty())
+						dp.MoveTo(mCopy.Sum().DataPoints().AppendEmpty())
 						return true
 					})
+					return dps.Len() == 0
 				case pmetric.MetricTypeHistogram:
 					dps := m.Histogram().DataPoints()
 					dps.RemoveIf(func(dp pmetric.HistogramDataPoint) bool {
 						if !f(rm, sm, m, dp) {
 							return false
 						}
 						if rmCopy == nil {
-							rmc := to.ResourceMetrics().AppendEmpty()
-							rmCopy = &rmc
-							rm.Resource().CopyTo(rmCopy.Resource())
-							rmCopy.SetSchemaUrl(rm.SchemaUrl())
+							rmCopy = copyResourceMetrics(rm, to.ResourceMetrics())
 						}
 						if smCopy == nil {
-							smc := rmCopy.ScopeMetrics().AppendEmpty()
-							smCopy = &smc
-							sm.Scope().CopyTo(smCopy.Scope())
-							smCopy.SetSchemaUrl(sm.SchemaUrl())
+							smCopy = copyScopeMetrics(sm, rmCopy.ScopeMetrics())
 						}
 						if mCopy == nil {
-							mc := smCopy.Metrics().AppendEmpty()
-							mCopy = &mc
-							mCopy.SetName(m.Name())
-							mCopy.SetDescription(m.Description())
-							mCopy.SetUnit(m.Unit())
+							mCopy = copyMetricDescription(m, smCopy.Metrics())
 							mCopy.SetEmptyHistogram()
 						}
-						dp.CopyTo(mCopy.Histogram().DataPoints().AppendEmpty())
+						dp.MoveTo(mCopy.Histogram().DataPoints().AppendEmpty())
 						return true
 					})
+					return dps.Len() == 0
 				case pmetric.MetricTypeExponentialHistogram:
 					dps := m.ExponentialHistogram().DataPoints()
 					dps.RemoveIf(func(dp pmetric.ExponentialHistogramDataPoint) bool {
 						if !f(rm, sm, m, dp) {
 							return false
 						}
 						if rmCopy == nil {
-							rmc := to.ResourceMetrics().AppendEmpty()
-							rmCopy = &rmc
-							rm.Resource().CopyTo(rmCopy.Resource())
-							rmCopy.SetSchemaUrl(rm.SchemaUrl())
+							rmCopy = copyResourceMetrics(rm, to.ResourceMetrics())
 						}
 						if smCopy == nil {
-							smc := rmCopy.ScopeMetrics().AppendEmpty()
-							smCopy = &smc
-							sm.Scope().CopyTo(smCopy.Scope())
-							smCopy.SetSchemaUrl(sm.SchemaUrl())
+							smCopy = copyScopeMetrics(sm, rmCopy.ScopeMetrics())
 						}
 						if mCopy == nil {
-							mc := smCopy.Metrics().AppendEmpty()
-							mCopy = &mc
-							mCopy.SetName(m.Name())
-							mCopy.SetDescription(m.Description())
-							mCopy.SetUnit(m.Unit())
+							mCopy = copyMetricDescription(m, smCopy.Metrics())
 							mCopy.SetEmptyExponentialHistogram()
 						}
-						dp.CopyTo(mCopy.ExponentialHistogram().DataPoints().AppendEmpty())
+						dp.MoveTo(mCopy.ExponentialHistogram().DataPoints().AppendEmpty())
 						return true
 					})
+					return dps.Len() == 0
 				case pmetric.MetricTypeSummary:
 					dps := m.Summary().DataPoints()
 					dps.RemoveIf(func(dp pmetric.SummaryDataPoint) bool {
 						if !f(rm, sm, m, dp) {
 							return false
 						}
 						if rmCopy == nil {
-							rmc := to.ResourceMetrics().AppendEmpty()
-							rmCopy = &rmc
-							rm.Resource().CopyTo(rmCopy.Resource())
-							rmCopy.SetSchemaUrl(rm.SchemaUrl())
+							rmCopy = copyResourceMetrics(rm, to.ResourceMetrics())
 						}
 						if smCopy == nil {
-							smc := rmCopy.ScopeMetrics().AppendEmpty()
-							smCopy = &smc
-							sm.Scope().CopyTo(smCopy.Scope())
-							smCopy.SetSchemaUrl(sm.SchemaUrl())
+							smCopy = copyScopeMetrics(sm, rmCopy.ScopeMetrics())
 						}
 						if mCopy == nil {
-							mc := smCopy.Metrics().AppendEmpty()
-							mCopy = &mc
-							mCopy.SetName(m.Name())
-							mCopy.SetDescription(m.Description())
-							mCopy.SetUnit(m.Unit())
+							mCopy = copyMetricDescription(m, smCopy.Metrics())
 							mCopy.SetEmptySummary()
 						}
-						dp.CopyTo(mCopy.Summary().DataPoints().AppendEmpty())
+						dp.MoveTo(mCopy.Summary().DataPoints().AppendEmpty())
 						return true
 					})
+					return dps.Len() == 0
 				}
-			}
-			ms.RemoveIf(func(m pmetric.Metric) bool {
-				var numDPs int
-				switch m.Type() {
-				case pmetric.MetricTypeGauge:
-					numDPs = m.Gauge().DataPoints().Len()
-				case pmetric.MetricTypeSum:
-					numDPs = m.Sum().DataPoints().Len()
-				case pmetric.MetricTypeHistogram:
-					numDPs = m.Histogram().DataPoints().Len()
-				case pmetric.MetricTypeExponentialHistogram:
-					numDPs = m.ExponentialHistogram().DataPoints().Len()
-				case pmetric.MetricTypeSummary:
-					numDPs = m.Summary().DataPoints().Len()
-				}
-				return numDPs == 0
+				// Do not remove unknown type.
+				return false
 			})
-		}
-		sms.RemoveIf(func(sm pmetric.ScopeMetrics) bool {
 			return sm.Metrics().Len() == 0
 		})
-	}
-	rms.RemoveIf(func(rm pmetric.ResourceMetrics) bool {
 		return rm.ScopeMetrics().Len() == 0
 	})
 }
+
+func copyResourceMetrics(from pmetric.ResourceMetrics, to pmetric.ResourceMetricsSlice) *pmetric.ResourceMetrics {
+	rmc := to.AppendEmpty()
+	from.Resource().CopyTo(rmc.Resource())
+	rmc.SetSchemaUrl(from.SchemaUrl())
+	return &rmc
+}
+
+func copyScopeMetrics(from pmetric.ScopeMetrics, to pmetric.ScopeMetricsSlice) *pmetric.ScopeMetrics {
+	smc := to.AppendEmpty()
+	from.Scope().CopyTo(smc.Scope())
+	smc.SetSchemaUrl(from.SchemaUrl())
+	return &smc
+}
+
+func copyMetricDescription(from pmetric.Metric, to pmetric.MetricSlice) *pmetric.Metric {
+	mc := to.AppendEmpty()
+	mc.SetName(from.Name())
+	mc.SetDescription(from.Description())
+	mc.SetUnit(from.Unit())
+	return &mc
+}

connector/routingconnector/internal/pmetricutil/metrics_test.go

@@ -1506,3 +1506,16 @@ func TestMoveDataPointsWithContextIf(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkMoveResourcesIfMetrics(b *testing.B) {
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		from := pmetricutiltest.NewGauges("AB", "CD", "EF", "GH")
+		to := pmetric.NewMetrics()
+		pmetricutil.MoveResourcesIf(from, to, func(pmetric.ResourceMetrics) bool {
+			return true
+		})
+		assert.Equal(b, 0, from.DataPointCount())
+		assert.Equal(b, 16, to.DataPointCount())
+	}
+}