Log records dropped by the BatchProcessor (#5276)

* Add dropped count to queue

* Log dropped records

* Add changelog entry

---------

Co-authored-by: Robert Pająk <pellared@hotmail.com>

sdk/log/batch_test.go

@@ -4,7 +4,9 @@
 package log // import "go.opentelemetry.io/otel/sdk/log"
 
 import (
+	"bytes"
 	"context"
+	stdlog "log"
 	"slices"
 	"strconv"
 	"sync"
@@ -12,10 +14,12 @@ import (
 	"time"
 	"unsafe"
 
+	"github.com/go-logr/stdr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/internal/global"
 	"go.opentelemetry.io/otel/log"
 )
 
@@ -413,6 +417,41 @@ func TestBatchProcessor(t *testing.T) {
 		})
 	})
 
+	t.Run("DroppedLogs", func(t *testing.T) {
+		orig := global.GetLogger()
+		t.Cleanup(func() { global.SetLogger(orig) })
+		buf := new(bytes.Buffer)
+		stdr.SetVerbosity(1)
+		global.SetLogger(stdr.New(stdlog.New(buf, "", 0)))
+
+		e := newTestExporter(nil)
+		e.ExportTrigger = make(chan struct{})
+
+		b := NewBatchProcessor(
+			e,
+			WithMaxQueueSize(1),
+			WithExportMaxBatchSize(1),
+			WithExportInterval(time.Hour),
+			WithExportTimeout(time.Hour),
+		)
+		var r Record
+		assert.NoError(t, b.OnEmit(ctx, r), "queued")
+		assert.NoError(t, b.OnEmit(ctx, r), "dropped")
+
+		var n int
+		require.Eventually(t, func() bool {
+			n = e.ExportN()
+			return n > 0
+		}, 2*time.Second, time.Microsecond, "blocked export not attempted")
+
+		got := buf.String()
+		want := `"level"=1 "msg"="dropped log records" "dropped"=1`
+		assert.Contains(t, got, want)
+
+		close(e.ExportTrigger)
+		_ = b.Shutdown(ctx)
+	})
+
 	t.Run("ConcurrentSafe", func(t *testing.T) {
 		const goRoutines = 10
 
@@ -488,6 +527,18 @@ func TestQueue(t *testing.T) {
 		assert.Equal(t, []Record{r, r}, q.Flush(), "flushed Records")
 	})
 
+	t.Run("Dropped", func(t *testing.T) {
+		q := newQueue(1)
+
+		_ = q.Enqueue(r)
+		_ = q.Enqueue(r)
+		assert.Equal(t, uint64(1), q.Dropped(), "fist")
+
+		_ = q.Enqueue(r)
+		_ = q.Enqueue(r)
+		assert.Equal(t, uint64(2), q.Dropped(), "second")
+	})
+
 	t.Run("Flush", func(t *testing.T) {
 		const size = 2
 		q := newQueue(size)