[exporter/loadbalancingexporter] batch before exporting in load balancer

.chloggen/improve-loadbalancer-performance.yaml

@@ -0,0 +1,17 @@
+# 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: loadbalancingexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add batching for each endpoint in the loadbalancingexporter.
+
+# One or more tracking issues related to the change
+issues: [17173]
+
+# (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/loadbalancingexporter/log_exporter.go

@@ -76,25 +76,45 @@ func (e *logExporterImp) Shutdown(context.Context) error {
 
 func (e *logExporterImp) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 	var errs error
+	endpointToLogData := make(map[string]plog.Logs)
 	batches := batchpersignal.SplitLogs(ld)
-	for _, batch := range batches {
-		errs = multierr.Append(errs, e.consumeLog(ctx, batch))
+
+	// Map the logs to their respective endpoints.
+	for batch := range batches {
+		// Each batch contains at most one trace ID.
+		traceID := traceIDFromLogs(ld)
+		balancingKey := traceID
+		if traceID == pcommon.NewTraceIDEmpty() {
+			// every log may not contain a traceID
+			// generate a random traceID as balancingKey
+			// so the log can be routed to a random backend
+			balancingKey = random()
+		}
+
+		endpoint := e.loadBalancer.Endpoint(balancingKey[:])
+		if _, ok := endpointToLogData[endpoint]; ok {
+			// append
+			for i := 0; i < batches[batch].ResourceLogs().Len(); i++ {
+				batches[batch].ResourceLogs().At(i).CopyTo(endpointToLogData[endpoint].ResourceLogs().AppendEmpty())
+			}
+		} else {
+			newLog := plog.NewLogs()
+			for i := 0; i < batches[batch].ResourceLogs().Len(); i++ {
+				batches[batch].ResourceLogs().At(i).CopyTo(newLog.ResourceLogs().AppendEmpty())
+			}
+			endpointToLogData[endpoint] = newLog
+		}
+	}
+
+	// Send the logs off by endpoint.
+	for endpoint, logs := range endpointToLogData {
+		errs = multierr.Append(errs, e.consumeLog(ctx, logs, endpoint))
 	}
 
 	return errs
 }
 
-func (e *logExporterImp) consumeLog(ctx context.Context, ld plog.Logs) error {
-	traceID := traceIDFromLogs(ld)
-	balancingKey := traceID
-	if traceID == pcommon.NewTraceIDEmpty() {
-		// every log may not contain a traceID
-		// generate a random traceID as balancingKey
-		// so the log can be routed to a random backend
-		balancingKey = random()
-	}
-
-	endpoint := e.loadBalancer.Endpoint(balancingKey[:])
+func (e *logExporterImp) consumeLog(ctx context.Context, ld plog.Logs, endpoint string) error {
 	exp, err := e.loadBalancer.Exporter(endpoint)
 	if err != nil {
 		return err

exporter/loadbalancingexporter/log_exporter_test.go

@@ -231,7 +231,8 @@ func TestLogBatchWithTwoTraces(t *testing.T) {
 
 	// verify
 	assert.NoError(t, err)
-	assert.Len(t, sink.AllLogs(), 2)
+	assert.Len(t, sink.AllLogs(), 1)
+	assert.Equal(t, sink.LogRecordCount(), 2)
 }
 
 func TestNoLogsInBatch(t *testing.T) {

exporter/loadbalancingexporter/trace_exporter.go

@@ -90,46 +90,59 @@ func (e *traceExporterImp) Shutdown(context.Context) error {
 func (e *traceExporterImp) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
 	var errs error
 	batches := batchpersignal.SplitTraces(td)
-	for _, batch := range batches {
-		errs = multierr.Append(errs, e.consumeTrace(ctx, batch))
+	endpointToTraceData := make(map[string]ptrace.Traces)
+
+	// Map the trace data to their respective endpoints.
+	for batch := range batches {
+		routingIDs, err := routingIdentifiersFromTraces(batches[batch], e.routingKey)
+		errs = multierr.Append(errs, err)
+		for rid := range routingIDs {
+			endpoint := e.loadBalancer.Endpoint([]byte(rid))
+			if _, ok := endpointToTraceData[endpoint]; ok {
+				// append
+				for i := 0; i < batches[batch].ResourceSpans().Len(); i++ {
+					batches[batch].ResourceSpans().At(i).CopyTo(endpointToTraceData[endpoint].ResourceSpans().AppendEmpty())
+				}
+			} else {
+				newTrace := ptrace.NewTraces()
+				for i := 0; i < batches[batch].ResourceSpans().Len(); i++ {
+					batches[batch].ResourceSpans().At(i).CopyTo(newTrace.ResourceSpans().AppendEmpty())
+				}
+				endpointToTraceData[endpoint] = newTrace
+			}
+		}
 	}
 
+	// Send the trace data off by endpoint.
+	for endpoint, traces := range endpointToTraceData {
+		errs = multierr.Append(errs, e.consumeTrace(ctx, traces, endpoint))
+	}
 	return errs
 }
 
-func (e *traceExporterImp) consumeTrace(ctx context.Context, td ptrace.Traces) error {
-	var exp component.Component
-	routingIds, err := routingIdentifiersFromTraces(td, e.routingKey)
+func (e *traceExporterImp) consumeTrace(ctx context.Context, td ptrace.Traces, endpoint string) error {
+	exp, err := e.loadBalancer.Exporter(endpoint)
 	if err != nil {
 		return err
 	}
-	for rid := range routingIds {
-		endpoint := e.loadBalancer.Endpoint([]byte(rid))
-		exp, err = e.loadBalancer.Exporter(endpoint)
-		if err != nil {
-			return err
-		}
 
-		te, ok := exp.(exporter.Traces)
-		if !ok {
-			return fmt.Errorf("unable to export traces, unexpected exporter type: expected exporter.Traces but got %T", exp)
-		}
-
-		start := time.Now()
-		err = te.ConsumeTraces(ctx, td)
-		duration := time.Since(start)
-
-		if err == nil {
-			_ = stats.RecordWithTags(
-				ctx,
-				[]tag.Mutator{tag.Upsert(endpointTagKey, endpoint), successTrueMutator},
-				mBackendLatency.M(duration.Milliseconds()))
-		} else {
-			_ = stats.RecordWithTags(
-				ctx,
-				[]tag.Mutator{tag.Upsert(endpointTagKey, endpoint), successFalseMutator},
-				mBackendLatency.M(duration.Milliseconds()))
-		}
+	te, ok := exp.(exporter.Traces)
+	if !ok {
+		return fmt.Errorf("unable to export traces, unexpected exporter type: expected exporter.Traces but got %T", exp)
+	}
+	start := time.Now()
+	err = te.ConsumeTraces(ctx, td)
+	duration := time.Since(start)
+	if err == nil {
+		_ = stats.RecordWithTags(
+			ctx,
+			[]tag.Mutator{tag.Upsert(endpointTagKey, endpoint), successTrueMutator},
+			mBackendLatency.M(duration.Milliseconds()))
+	} else {
+		_ = stats.RecordWithTags(
+			ctx,
+			[]tag.Mutator{tag.Upsert(endpointTagKey, endpoint), successFalseMutator},
+			mBackendLatency.M(duration.Milliseconds()))
 	}
 	return err
 }

exporter/loadbalancingexporter/trace_exporter_test.go

@@ -354,7 +354,8 @@ func TestBatchWithTwoTraces(t *testing.T) {
 
 	// verify
 	assert.NoError(t, err)
-	assert.Len(t, sink.AllTraces(), 2)
+	assert.Len(t, sink.AllTraces(), 1)
+	assert.Equal(t, sink.SpanCount(), 2)
 }
 
 func TestNoTracesInBatch(t *testing.T) {