Add warning log on dropped spans in batch span processor

CHANGELOG.md

@@ -21,6 +21,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
    A non-default registerer can be used by passing the `WithRegisterer` option. (#3239)
 - Added the `WithAggregationSelector` option to the `go.opentelemetry.io/otel/exporters/prometheus` package to change the default `AggregationSelector` used. (#3341)
 - The Prometheus exporter in `go.opentelemetry.io/otel/exporters/prometheus` converts the `Resource` associated with metric exports into a `target_info` metric. (#3285)
+- Batch span processor will now keep track of known dropped span count and log a warning during batch export on newly dropped spans. (#3289)
 
 ### Changed
 

internal/global/internal_logging.go

@@ -48,6 +48,13 @@ func Info(msg string, keysAndValues ...interface{}) {
 	globalLogger.V(1).Info(msg, keysAndValues...)
 }
 
+// Warn prints messages about potential issues with the API or SDK.
+func Warn(msg string, keysAndValues ...interface{}) {
+	globalLoggerLock.RLock()
+	defer globalLoggerLock.RUnlock()
+	globalLogger.V(0).Info(msg, keysAndValues...)
+}
+
 // Error prints messages about exceptional states of the API or SDK.
 func Error(err error, msg string, keysAndValues ...interface{}) {
 	globalLoggerLock.RLock()

sdk/trace/batch_span_processor.go

@@ -75,8 +75,9 @@ type batchSpanProcessor struct {
 	e SpanExporter
 	o BatchSpanProcessorOptions
 
-	queue   chan ReadOnlySpan
-	dropped uint32
+	queue        chan ReadOnlySpan
+	dropped      uint32
+	knownDropped uint32
 
 	batch      []ReadOnlySpan
 	batchMutex sync.Mutex
@@ -265,7 +266,14 @@ func (bsp *batchSpanProcessor) exportSpans(ctx context.Context) error {
 	}
 
 	if l := len(bsp.batch); l > 0 {
-		global.Debug("exporting spans", "count", len(bsp.batch), "total_dropped", atomic.LoadUint32(&bsp.dropped))
+		global.Debug("exporting spans", "count", len(bsp.batch))
+
+		dropped := atomic.LoadUint32(&bsp.dropped)
+		knownDropped := atomic.SwapUint32(&bsp.knownDropped, dropped)
+		if dropped > knownDropped {
+			global.Warn("dropped spans", "total_dropped", dropped, "known_dropped", knownDropped)
+		}
+
 		err := bsp.e.ExportSpans(ctx, bsp.batch)
 
 		// A new batch is always created after exporting, even if the batch failed to be exported.

sdk/trace/batch_span_processor_test.go

@@ -19,7 +19,11 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"github.com/go-logr/logr"
+	"github.com/go-logr/logr/testr"
+	"go.opentelemetry.io/otel"
 	"os"
+	"regexp"
 	"sync"
 	"testing"
 	"time"
@@ -49,6 +53,16 @@ type testBatchExporter struct {
 	err           error
 }
 
+type logCounter struct {
+	logr.LogSink
+	logs []string
+}
+
+func (l *logCounter) Info(level int, msg string, keysAndValues ...interface{}) {
+	l.logs = append(l.logs, fmt.Sprint(msg, keysAndValues))
+	l.LogSink.Info(level, msg, keysAndValues...)
+}
+
 func (t *testBatchExporter) ExportSpans(ctx context.Context, spans []sdktrace.ReadOnlySpan) error {
 	t.mu.Lock()
 	defer t.mu.Unlock()
@@ -350,6 +364,12 @@ func createAndRegisterBatchSP(option testOption, te *testBatchExporter) sdktrace
 	return sdktrace.NewBatchSpanProcessor(te, options...)
 }
 
+func createAndRegisterBatchSPNonBlocking(option testOption, te *testBatchExporter) sdktrace.SpanProcessor {
+	// Blocking span processor should be preferred to avoid flaky test
+	// To test dropped spans warn logging the blocking span processor bypasses drop logic
+	return sdktrace.NewBatchSpanProcessor(te, option.o...)
+}
+
 func generateSpan(t *testing.T, tr trace.Tracer, option testOption) {
 	sc := getSpanContext()
 
@@ -474,6 +494,38 @@ func TestBatchSpanProcessorForceFlushSucceeds(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestBatchSpanProcessorLogsWarningOnNewDropSpans(t *testing.T) {
+	tLog := testr.NewWithOptions(t, testr.Options{Verbosity: 0})
+	l := &logCounter{LogSink: tLog.GetSink()}
+	otel.SetLogger(logr.New(l))
+	global.SetLogger(logr.New(l))
+
+	te := testBatchExporter{}
+	tp := basicTracerProvider(t)
+	option := testOption{
+		name: "default BatchSpanProcessorOptions",
+		o: []sdktrace.BatchSpanProcessorOption{
+			sdktrace.WithMaxQueueSize(0),
+			sdktrace.WithMaxExportBatchSize(10),
+		},
+		genNumSpans: 100,
+	}
+	ssp := createAndRegisterBatchSPNonBlocking(option, &te)
+	if ssp == nil {
+		t.Fatalf("%s: Error creating new instance of BatchSpanProcessor\n", option.name)
+	}
+	tp.RegisterSpanProcessor(ssp)
+	tr := tp.Tracer("BatchSpanProcessorWithOption")
+
+	// Force flush any held span batches
+	generateSpan(t, tr, option)
+	ssp.ForceFlush(context.Background())
+
+	assert.Equal(t, len(l.logs), 1, "expected log message")
+	droppedLogs := l.logs[0]
+	assert.Regexp(t, regexp.MustCompile(`dropped spans\[total_dropped \d+ known_dropped \d+\]`), droppedLogs)
+}
+
 func TestBatchSpanProcessorDropBatchIfFailed(t *testing.T) {
 	te := testBatchExporter{
 		errors: []error{errors.New("fail to export")},