Fix metrics split logic to consider metrics description into the size

.chloggen/metrics-tests.yaml

@@ -0,0 +1,25 @@
+# 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. otlpreceiver)
+component: exporterhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix metrics split logic to consider metrics description into the size.
+
+# One or more tracking issues or pull requests related to the change
+issues: [13418]
+
+# (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:
+
+# 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]

exporter/exporterhelper/logs_batch_test.go

@@ -246,6 +246,9 @@ func TestMergeSplitLogsBasedOnByteSize(t *testing.T) {
 			assert.Len(t, res, len(tt.expected))
 			for i := range res {
 				assert.Equal(t, tt.expected[i].(*logsRequest).ld, res[i].(*logsRequest).ld)
+				assert.Equal(t,
+					logsMarshaler.LogsSize(tt.expected[i].(*logsRequest).ld),
+					logsMarshaler.LogsSize(res[i].(*logsRequest).ld))
 			}
 		})
 	}

exporter/exporterhelper/metrics_batch.go

@@ -165,7 +165,7 @@ func extractScopeMetrics(srcSM pmetric.ScopeMetrics, capacity int, sz sizer.Metr
 		rawRmSize := sz.MetricSize(srcSM)
 		rmSize := sz.DeltaSize(rawRmSize)
 		if rmSize > capacityLeft {
-			extSrcSM, extRmSize := extractMetricDataPoints(srcSM, capacityLeft, sz)
+			extSrcDP, extRmSize := extractMetricDataPoints(srcSM, capacityLeft, sz)
 			// This cannot make it to exactly 0 for the bytes,
 			// force it to be 0 since that is the stopping condition.
 			capacityLeft = 0
@@ -174,10 +174,10 @@ func extractScopeMetrics(srcSM pmetric.ScopeMetrics, capacity int, sz sizer.Metr
 			removedSize += rmSize - rawRmSize - (sz.DeltaSize(rawRmSize-extRmSize) - (rawRmSize - extRmSize))
 			// It is possible that for the bytes scenario, the extracted field contains no datapoints.
 			// Do not add it to the destination if that is the case.
-			if sz.MetricSize(extSrcSM) > 0 {
-				extSrcSM.MoveTo(destSM.Metrics().AppendEmpty())
+			if dataPointsLen(extSrcDP) > 0 {
+				extSrcDP.MoveTo(destSM.Metrics().AppendEmpty())
 			}
-			return sz.MetricSize(extSrcSM) != 0
+			return dataPointsLen(extSrcDP) != 0
 		}
 		capacityLeft -= rmSize
 		removedSize += rmSize
@@ -188,37 +188,53 @@ func extractScopeMetrics(srcSM pmetric.ScopeMetrics, capacity int, sz sizer.Metr
 }
 
 func extractMetricDataPoints(srcMetric pmetric.Metric, capacity int, sz sizer.MetricsSizer) (pmetric.Metric, int) {
-	var destMetric pmetric.Metric
+	destMetric := pmetric.NewMetric()
+	destMetric.SetName(srcMetric.Name())
+	destMetric.SetDescription(srcMetric.Description())
+	destMetric.SetUnit(srcMetric.Unit())
+	srcMetric.Metadata().CopyTo(destMetric.Metadata())
+
 	var removedSize int
 	switch srcMetric.Type() {
 	case pmetric.MetricTypeGauge:
-		destMetric, removedSize = extractGaugeDataPoints(srcMetric.Gauge(), capacity, sz)
+		removedSize = extractGaugeDataPoints(srcMetric.Gauge(), destMetric, capacity, sz)
 	case pmetric.MetricTypeSum:
-		destMetric, removedSize = extractSumDataPoints(srcMetric.Sum(), capacity, sz)
+		removedSize = extractSumDataPoints(srcMetric.Sum(), destMetric, capacity, sz)
 		destMetric.Sum().SetIsMonotonic(srcMetric.Sum().IsMonotonic())
 		destMetric.Sum().SetAggregationTemporality(srcMetric.Sum().AggregationTemporality())
 	case pmetric.MetricTypeHistogram:
-		destMetric, removedSize = extractHistogramDataPoints(srcMetric.Histogram(), capacity, sz)
+		removedSize = extractHistogramDataPoints(srcMetric.Histogram(), destMetric, capacity, sz)
 		destMetric.Histogram().SetAggregationTemporality(srcMetric.Histogram().AggregationTemporality())
 	case pmetric.MetricTypeExponentialHistogram:
-		destMetric, removedSize = extractExponentialHistogramDataPoints(srcMetric.ExponentialHistogram(), capacity, sz)
+		removedSize = extractExponentialHistogramDataPoints(srcMetric.ExponentialHistogram(), destMetric, capacity, sz)
 		destMetric.ExponentialHistogram().SetAggregationTemporality(srcMetric.ExponentialHistogram().AggregationTemporality())
 	case pmetric.MetricTypeSummary:
-		destMetric, removedSize = extractSummaryDataPoints(srcMetric.Summary(), capacity, sz)
+		removedSize = extractSummaryDataPoints(srcMetric.Summary(), destMetric, capacity, sz)
 	}
-	destMetric.SetName(srcMetric.Name())
-	destMetric.SetDescription(srcMetric.Description())
-	destMetric.SetUnit(srcMetric.Unit())
-	srcMetric.Metadata().CopyTo(destMetric.Metadata())
 	return destMetric, removedSize
 }
 
-func extractGaugeDataPoints(srcGauge pmetric.Gauge, capacity int, sz sizer.MetricsSizer) (pmetric.Metric, int) {
-	m := pmetric.NewMetric()
-	destGauge := m.SetEmptyGauge()
+func dataPointsLen(m pmetric.Metric) int {
+	switch m.Type() {
+	case pmetric.MetricTypeGauge:
+		return m.Gauge().DataPoints().Len()
+	case pmetric.MetricTypeSum:
+		return m.Sum().DataPoints().Len()
+	case pmetric.MetricTypeHistogram:
+		return m.Histogram().DataPoints().Len()
+	case pmetric.MetricTypeExponentialHistogram:
+		return m.ExponentialHistogram().DataPoints().Len()
+	case pmetric.MetricTypeSummary:
+		return m.Summary().DataPoints().Len()
+	}
+	return 0
+}
+
+func extractGaugeDataPoints(srcGauge pmetric.Gauge, destMetric pmetric.Metric, capacity int, sz sizer.MetricsSizer) int {
+	destGauge := destMetric.SetEmptyGauge()
 
 	// Take into account that this can have max "capacity", so when added to the parent will need space for the extra delta size.
-	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(m)
+	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(destMetric)
 	removedSize := 0
 
 	srcGauge.DataPoints().RemoveIf(func(srcDP pmetric.NumberDataPoint) bool {
@@ -239,14 +255,13 @@ func extractGaugeDataPoints(srcGauge pmetric.Gauge, capacity int, sz sizer.Metri
 		srcDP.MoveTo(destGauge.DataPoints().AppendEmpty())
 		return true
 	})
-	return m, removedSize
+	return removedSize
 }
 
-func extractSumDataPoints(srcSum pmetric.Sum, capacity int, sz sizer.MetricsSizer) (pmetric.Metric, int) {
-	m := pmetric.NewMetric()
-	destSum := m.SetEmptySum()
+func extractSumDataPoints(srcSum pmetric.Sum, destMetric pmetric.Metric, capacity int, sz sizer.MetricsSizer) int {
+	destSum := destMetric.SetEmptySum()
 	// Take into account that this can have max "capacity", so when added to the parent will need space for the extra delta size.
-	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(m)
+	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(destMetric)
 	removedSize := 0
 	srcSum.DataPoints().RemoveIf(func(srcDP pmetric.NumberDataPoint) bool {
 		// If the no more capacity left just return.
@@ -266,14 +281,13 @@ func extractSumDataPoints(srcSum pmetric.Sum, capacity int, sz sizer.MetricsSize
 		srcDP.MoveTo(destSum.DataPoints().AppendEmpty())
 		return true
 	})
-	return m, removedSize
+	return removedSize
 }
 
-func extractHistogramDataPoints(srcHistogram pmetric.Histogram, capacity int, sz sizer.MetricsSizer) (pmetric.Metric, int) {
-	m := pmetric.NewMetric()
-	destHistogram := m.SetEmptyHistogram()
+func extractHistogramDataPoints(srcHistogram pmetric.Histogram, destMetric pmetric.Metric, capacity int, sz sizer.MetricsSizer) int {
+	destHistogram := destMetric.SetEmptyHistogram()
 	// Take into account that this can have max "capacity", so when added to the parent will need space for the extra delta size.
-	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(m)
+	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(destMetric)
 	removedSize := 0
 	srcHistogram.DataPoints().RemoveIf(func(srcDP pmetric.HistogramDataPoint) bool {
 		// If the no more capacity left just return.
@@ -293,14 +307,13 @@ func extractHistogramDataPoints(srcHistogram pmetric.Histogram, capacity int, sz
 		srcDP.MoveTo(destHistogram.DataPoints().AppendEmpty())
 		return true
 	})
-	return m, removedSize
+	return removedSize
 }
 
-func extractExponentialHistogramDataPoints(srcExponentialHistogram pmetric.ExponentialHistogram, capacity int, sz sizer.MetricsSizer) (pmetric.Metric, int) {
-	m := pmetric.NewMetric()
-	destExponentialHistogram := m.SetEmptyExponentialHistogram()
+func extractExponentialHistogramDataPoints(srcExponentialHistogram pmetric.ExponentialHistogram, destMetric pmetric.Metric, capacity int, sz sizer.MetricsSizer) int {
+	destExponentialHistogram := destMetric.SetEmptyExponentialHistogram()
 	// Take into account that this can have max "capacity", so when added to the parent will need space for the extra delta size.
-	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(m)
+	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(destMetric)
 	removedSize := 0
 	srcExponentialHistogram.DataPoints().RemoveIf(func(srcDP pmetric.ExponentialHistogramDataPoint) bool {
 		// If the no more capacity left just return.
@@ -320,14 +333,13 @@ func extractExponentialHistogramDataPoints(srcExponentialHistogram pmetric.Expon
 		srcDP.MoveTo(destExponentialHistogram.DataPoints().AppendEmpty())
 		return true
 	})
-	return m, removedSize
+	return removedSize
 }
 
-func extractSummaryDataPoints(srcSummary pmetric.Summary, capacity int, sz sizer.MetricsSizer) (pmetric.Metric, int) {
-	m := pmetric.NewMetric()
-	destSummary := m.SetEmptySummary()
+func extractSummaryDataPoints(srcSummary pmetric.Summary, destMetric pmetric.Metric, capacity int, sz sizer.MetricsSizer) int {
+	destSummary := destMetric.SetEmptySummary()
 	// Take into account that this can have max "capacity", so when added to the parent will need space for the extra delta size.
-	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(m)
+	capacityLeft := capacity - (sz.DeltaSize(capacity) - capacity) - sz.MetricSize(destMetric)
 	removedSize := 0
 	srcSummary.DataPoints().RemoveIf(func(srcDP pmetric.SummaryDataPoint) bool {
 		// If the no more capacity left just return.
@@ -347,5 +359,5 @@ func extractSummaryDataPoints(srcSummary pmetric.Summary, capacity int, sz sizer
 		srcDP.MoveTo(destSummary.DataPoints().AppendEmpty())
 		return true
 	})
-	return m, removedSize
+	return removedSize
 }