Optimize Logs SDK BatchProcessor

CHANGELOG.md

@@ -36,6 +36,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - Initialize map with `len(keys)` in `NewAllowKeysFilter` and `NewDenyKeysFilter` to avoid unnecessary allocations in `go.opentelemetry.io/otel/attribute`. (#6455)
 - `go.opentelemetry.io/otel/log/logtest` is now a separate Go module. (#6465)
 - `go.opentelemetry.io/otel/sdk/log/logtest` is now a separate Go module. (#6466)
+- `BatchProcessor` in `go.opentelemetry.io/otel/sdk/log` now drops the most recently emitted log records instead of the oldest ones when the queue reaches its capacity. (#6569)
 
 ### Deprecated
 

sdk/log/batch.go

@@ -156,13 +156,17 @@ func (b *BatchProcessor) poll(interval time.Duration) (done chan struct{}) {
 				global.Warn("dropped log records", "dropped", d)
 			}
 
-			qLen := b.q.TryDequeue(buf, func(r []Record) bool {
-				ok := b.exporter.EnqueueExport(r)
-				if ok {
-					buf = slices.Clone(buf)
-				}
-				return ok
-			})
+			qLen := b.q.Len()
+			// don't copy data from queue unless exporter can accept more, it is very expensive
+			if !b.exporter.IsQueueFull() {
+				qLen = b.q.TryDequeue(buf, func(r []Record) bool {
+					ok := b.exporter.EnqueueExport(r)
+					if ok {
+						buf = slices.Clone(buf)
+					}
+					return ok
+				})
+			}
 			if qLen >= b.batchSize {
 				// There is another full batch ready. Immediately trigger
 				// another export attempt.
@@ -272,6 +276,13 @@ func newQueue(size int) *queue {
 	}
 }
 
+func (q *queue) Len() int {
+	q.Lock()
+	defer q.Unlock()
+
+	return q.len
+}
+
 // Dropped returns the number of Records dropped during enqueueing since the
 // last time Dropped was called.
 func (q *queue) Dropped() uint64 {

sdk/log/bench_test.go

@@ -13,6 +13,17 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+type mockDelayExporter struct{}
+
+func (mockDelayExporter) Export(context.Context, []Record) error {
+	time.Sleep(time.Millisecond * 5)
+	return nil
+}
+
+func (mockDelayExporter) Shutdown(context.Context) error { return nil }
+
+func (mockDelayExporter) ForceFlush(context.Context) error { return nil }
+
 func BenchmarkProcessor(b *testing.B) {
 	for _, tc := range []struct {
 		name string
@@ -30,6 +41,12 @@ func BenchmarkProcessor(b *testing.B) {
 				return []LoggerProviderOption{WithProcessor(NewBatchProcessor(noopExporter{}))}
 			},
 		},
+		{
+			name: "BatchSimulateExport",
+			f: func() []LoggerProviderOption {
+				return []LoggerProviderOption{WithProcessor(NewBatchProcessor(mockDelayExporter{}))}
+			},
+		},
 		{
 			name: "SetTimestampSimple",
 			f: func() []LoggerProviderOption {

sdk/log/exporter.go

@@ -186,11 +186,10 @@ type bufferExporter struct {
 
 // newBufferExporter returns a new bufferExporter that wraps exporter. The
 // returned bufferExporter will buffer at most size number of export requests.
-// If size is less than zero, zero will be used (i.e. only synchronous
-// exporting will be supported).
+// If size is less than 1, 1 will be used.
 func newBufferExporter(exporter Exporter, size int) *bufferExporter {
-	if size < 0 {
-		size = 0
+	if size < 1 {
+		size = 1
 	}
 	input := make(chan exportData, size)
 	return &bufferExporter{
@@ -201,6 +200,10 @@ func newBufferExporter(exporter Exporter, size int) *bufferExporter {
 	}
 }
 
+func (e *bufferExporter) IsQueueFull() bool {
+	return len(e.input) == cap(e.input)
+}
+
 var errStopped = errors.New("exporter stopped")
 
 func (e *bufferExporter) enqueue(ctx context.Context, records []Record, rCh chan<- error) error {