[exporter/splunkhec] apply multimetric metric merge for the whole batch

.chloggen/splunkhec-exporter-multimetric-batch.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: splunkhecexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Apply multi-metric merge at the level of the whole batch rather than within events emitted for one metric.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [23365]
+
+# (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:

exporter/splunkhecexporter/client.go

@@ -4,7 +4,6 @@
 package splunkhecexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/splunkhecexporter"
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -243,55 +242,107 @@ func (c *client) fillMetricsBuffer(metrics pmetric.Metrics, buf buffer, is iterS
 	jsonStream := jsonStreamPool.Get().(*jsoniter.Stream)
 	defer jsonStreamPool.Put(jsonStream)
 
-	for i := is.resource; i < metrics.ResourceMetrics().Len(); i++ {
+	var events []*splunk.Event
+
+	for i := 0; i < metrics.ResourceMetrics().Len(); i++ {
 		rm := metrics.ResourceMetrics().At(i)
-		for j := is.library; j < rm.ScopeMetrics().Len(); j++ {
-			is.library = 0 // Reset library index for next resource.
+		for j := 0; j < rm.ScopeMetrics().Len(); j++ {
 			sm := rm.ScopeMetrics().At(j)
-			for k := is.record; k < sm.Metrics().Len(); k++ {
-				is.record = 0 // Reset record index for next library.
+			for k := 0; k < sm.Metrics().Len(); k++ {
 				metric := sm.Metrics().At(k)
 
 				// Parsing metric record to Splunk event.
-				events := mapMetricToSplunkEvent(rm.Resource(), metric, c.config, c.logger)
-				tempBuf := bytes.NewBuffer(make([]byte, 0, c.config.MaxContentLengthMetrics))
-				if c.config.UseMultiMetricFormat {
-					merged, err := mergeEventsToMultiMetricFormat(events)
-					if err != nil {
-						permanentErrors = append(permanentErrors, consumererror.NewPermanent(fmt.Errorf(
-							"error merging events: %w", err)))
-					} else {
-						events = merged
-					}
-				}
-				for _, event := range events {
-					// JSON encoding event and writing to buffer.
-					b, err := marshalEvent(event, c.config.MaxEventSize, jsonStream)
-					if err != nil {
-						permanentErrors = append(permanentErrors, consumererror.NewPermanent(fmt.Errorf("dropped metric event: %v, error: %w", event, err)))
-						continue
-					}
-					tempBuf.Write(b)
-				}
+				events = append(events, mapMetricToSplunkEvent(rm.Resource(), metric, c.config, c.logger)...)
+			}
+		}
+	}
 
-				// Continue adding events to buffer up to capacity.
-				b := tempBuf.Bytes()
-				_, err := buf.Write(b)
-				if err == nil {
-					continue
-				}
-				if errors.Is(err, errOverCapacity) {
-					if !buf.Empty() {
-						return iterState{i, j, k, false}, permanentErrors
-					}
-					permanentErrors = append(permanentErrors, consumererror.NewPermanent(
-						fmt.Errorf("dropped metric event: error: event size %d bytes larger than configured max"+
-							" content length %d bytes", len(b), c.config.MaxContentLengthMetrics)))
-					return iterState{i, j, k + 1, false}, permanentErrors
-				}
-				permanentErrors = append(permanentErrors, consumererror.NewPermanent(fmt.Errorf(
-					"error writing the event: %w", err)))
+	for i := is.record; i < len(events); i++ {
+		event := events[i]
+		// JSON encoding event and writing to buffer.
+		b, jsonErr := marshalEvent(event, c.config.MaxEventSize, jsonStream)
+		if jsonErr != nil {
+			permanentErrors = append(permanentErrors, consumererror.NewPermanent(fmt.Errorf("dropped metric event: %v, error: %w", event, jsonErr)))
+			continue
+		}
+		_, err := buf.Write(b)
+		if errors.Is(err, errOverCapacity) {
+			if !buf.Empty() {
+				return iterState{
+					record: i,
+					done:   false,
+				}, permanentErrors
+			}
+			permanentErrors = append(permanentErrors, consumererror.NewPermanent(
+				fmt.Errorf("dropped metric event: error: event size %d bytes larger than configured max"+
+					" content length %d bytes", len(b), c.config.MaxContentLengthMetrics)))
+			return iterState{
+				record: i + 1,
+				done:   i+1 != len(events),
+			}, permanentErrors
+		} else if err != nil {
+			permanentErrors = append(permanentErrors, consumererror.NewPermanent(fmt.Errorf(
+				"error writing the event: %w", err)))
+
+		}
+	}
+
+	return iterState{done: true}, permanentErrors
+}
+
+func (c *client) fillMetricsBufferMultiMetrics(metrics pmetric.Metrics, buf buffer, is iterState) (iterState, []error) {
+	var permanentErrors []error
+	jsonStream := jsonStreamPool.Get().(*jsoniter.Stream)
+	defer jsonStreamPool.Put(jsonStream)
+
+	var events []*splunk.Event
+
+	for i := 0; i < metrics.ResourceMetrics().Len(); i++ {
+		rm := metrics.ResourceMetrics().At(i)
+		for j := 0; j < rm.ScopeMetrics().Len(); j++ {
+			sm := rm.ScopeMetrics().At(j)
+			for k := 0; k < sm.Metrics().Len(); k++ {
+				metric := sm.Metrics().At(k)
+
+				// Parsing metric record to Splunk event.
+				events = append(events, mapMetricToSplunkEvent(rm.Resource(), metric, c.config, c.logger)...)
+			}
+		}
+	}
+
+	merged, err := mergeEventsToMultiMetricFormat(events)
+	if err != nil {
+		permanentErrors = append(permanentErrors, consumererror.NewPermanent(fmt.Errorf(
+			"error merging events: %w", err)))
+	}
+
+	for i := is.record; i < len(merged); i++ {
+		event := merged[i]
+		// JSON encoding event and writing to buffer.
+		b, jsonErr := marshalEvent(event, c.config.MaxEventSize, jsonStream)
+		if jsonErr != nil {
+			permanentErrors = append(permanentErrors, consumererror.NewPermanent(fmt.Errorf("dropped metric event: %v, error: %w", event, jsonErr)))
+			continue
+		}
+		_, err = buf.Write(b)
+		if errors.Is(err, errOverCapacity) {
+			if !buf.Empty() {
+				return iterState{
+					record: i,
+					done:   false,
+				}, permanentErrors
 			}
+			permanentErrors = append(permanentErrors, consumererror.NewPermanent(
+				fmt.Errorf("dropped metric event: error: event size %d bytes larger than configured max"+
+					" content length %d bytes", len(b), c.config.MaxContentLengthMetrics)))
+			return iterState{
+				record: i + 1,
+				done:   i+1 != len(merged),
+			}, permanentErrors
+		} else if err != nil {
+			permanentErrors = append(permanentErrors, consumererror.NewPermanent(fmt.Errorf(
+				"error writing the event: %w", err)))
+
 		}
 	}
 
@@ -356,11 +407,17 @@ func (c *client) pushMetricsDataInBatches(ctx context.Context, md pmetric.Metric
 
 	for !is.done {
 		buf.Reset()
-		latestIterState, batchPermanentErrors := c.fillMetricsBuffer(md, buf, is)
+		var latestIterState iterState
+		var batchPermanentErrors []error
+		if c.config.UseMultiMetricFormat {
+			latestIterState, batchPermanentErrors = c.fillMetricsBufferMultiMetrics(md, buf, is)
+		} else {
+			latestIterState, batchPermanentErrors = c.fillMetricsBuffer(md, buf, is)
+		}
 		permanentErrors = append(permanentErrors, batchPermanentErrors...)
 		if !buf.Empty() {
 			if err := c.postEvents(ctx, buf, headers); err != nil {
-				return consumererror.NewMetrics(err, subMetrics(md, is))
+				return consumererror.NewMetrics(err, md)
 			}
 		}
 		is = latestIterState
@@ -443,49 +500,6 @@ func subLogs(src plog.Logs, state iterState) plog.Logs {
 	return dst
 }
 
-// subMetrics returns a subset of metrics starting from the state.
-func subMetrics(src pmetric.Metrics, state iterState) pmetric.Metrics {
-	if state.empty() {
-		return src
-	}
-
-	dst := pmetric.NewMetrics()
-	resources := src.ResourceMetrics()
-	resourcesSub := dst.ResourceMetrics()
-
-	for i := state.resource; i < resources.Len(); i++ {
-		newSub := resourcesSub.AppendEmpty()
-		resources.At(i).Resource().CopyTo(newSub.Resource())
-
-		libraries := resources.At(i).ScopeMetrics()
-		librariesSub := newSub.ScopeMetrics()
-
-		j := 0
-		if i == state.resource {
-			j = state.library
-		}
-		for ; j < libraries.Len(); j++ {
-			lib := libraries.At(j)
-
-			newLibSub := librariesSub.AppendEmpty()
-			lib.Scope().CopyTo(newLibSub.Scope())
-
-			metrics := lib.Metrics()
-			metricsSub := newLibSub.Metrics()
-
-			k := 0
-			if i == state.resource && j == state.library {
-				k = state.record
-			}
-			for ; k < metrics.Len(); k++ {
-				metrics.At(k).CopyTo(metricsSub.AppendEmpty())
-			}
-		}
-	}
-
-	return dst
-}
-
 func subTraces(src ptrace.Traces, state iterState) ptrace.Traces {
 	if state.empty() {
 		return src