Add otel.sdk.processor.log.processed metric

src/OpenTelemetry/BatchExportProcessor.cs

@@ -57,6 +57,8 @@ protected BatchExportProcessor(
         this.worker.Start();
     }
 
+    internal Action<long>? ExportStarted { get; set; }
+
     /// <summary>
     /// Gets the number of telemetry objects dropped by the processor.
     /// </summary>
@@ -153,7 +155,8 @@ private BatchExportWorker<T> CreateWorker()
                 this.exporter,
                 this.MaxExportBatchSize,
                 this.ScheduledDelayMilliseconds,
-                this.ExporterTimeoutMilliseconds);
+                this.ExporterTimeoutMilliseconds,
+                this.OnExportStarted);
         }
 #endif
 
@@ -163,6 +166,10 @@ private BatchExportWorker<T> CreateWorker()
             this.exporter,
             this.MaxExportBatchSize,
             this.ScheduledDelayMilliseconds,
-            this.ExporterTimeoutMilliseconds);
+            this.ExporterTimeoutMilliseconds,
+            this.OnExportStarted);
     }
+
+    private void OnExportStarted(long count)
+        => this.ExportStarted?.Invoke(count);
 }

src/OpenTelemetry/CHANGELOG.md

@@ -27,6 +27,9 @@ Notes](../../RELEASENOTES.md).
 * Added support for a Schema URL on `Resource` instances.
   ([#7472](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7472))
 
+* Added the `otel.sdk.processor.log.processed` SDK self-observability metric.
+  ([#7486](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7486))
+
 ## 1.16.0
 
 Released 2026-Jun-10

src/OpenTelemetry/Internal/BatchExportTaskWorker.cs

@@ -27,13 +27,15 @@ internal sealed class BatchExportTaskWorker<T> : BatchExportWorker<T>
     /// <param name="maxExportBatchSize">The maximum batch size for exports.</param>
     /// <param name="scheduledDelayMilliseconds">The delay between exports in milliseconds.</param>
     /// <param name="exporterTimeoutMilliseconds">The timeout for export operations in milliseconds.</param>
+    /// <param name="exportStarted">Callback invoked when a batch is submitted to the exporter.</param>
     public BatchExportTaskWorker(
         CircularBuffer<T> circularBuffer,
         BaseExporter<T> exporter,
         int maxExportBatchSize,
         int scheduledDelayMilliseconds,
-        int exporterTimeoutMilliseconds)
-        : base(circularBuffer, exporter, maxExportBatchSize, scheduledDelayMilliseconds, exporterTimeoutMilliseconds)
+        int exporterTimeoutMilliseconds,
+        Action<long>? exportStarted = null)
+        : base(circularBuffer, exporter, maxExportBatchSize, scheduledDelayMilliseconds, exporterTimeoutMilliseconds, exportStarted)
     {
     }
 

src/OpenTelemetry/Internal/BatchExportThreadWorker.cs

@@ -26,13 +26,15 @@ internal sealed class BatchExportThreadWorker<T> : BatchExportWorker<T>
     /// <param name="maxExportBatchSize">The maximum batch size for exports.</param>
     /// <param name="scheduledDelayMilliseconds">The delay between exports in milliseconds.</param>
     /// <param name="exporterTimeoutMilliseconds">The timeout for export operations in milliseconds.</param>
+    /// <param name="exportStarted">Callback invoked when a batch is submitted to the exporter.</param>
     public BatchExportThreadWorker(
         CircularBuffer<T> circularBuffer,
         BaseExporter<T> exporter,
         int maxExportBatchSize,
         int scheduledDelayMilliseconds,
-        int exporterTimeoutMilliseconds)
-        : base(circularBuffer, exporter, maxExportBatchSize, scheduledDelayMilliseconds, exporterTimeoutMilliseconds)
+        int exporterTimeoutMilliseconds,
+        Action<long>? exportStarted = null)
+        : base(circularBuffer, exporter, maxExportBatchSize, scheduledDelayMilliseconds, exporterTimeoutMilliseconds, exportStarted)
     {
         this.exporterThread = new Thread(this.ExporterProc)
         {

src/OpenTelemetry/Internal/BatchExportWorker.cs

@@ -10,6 +10,7 @@ namespace OpenTelemetry.Internal;
 internal abstract class BatchExportWorker<T> : IDisposable
     where T : class
 {
+    private readonly Action<long>? exportStarted;
     private long shutdownDrainTarget = long.MaxValue;
     private long droppedCount;
 
@@ -21,18 +22,21 @@ internal abstract class BatchExportWorker<T> : IDisposable
     /// <param name="maxExportBatchSize">The maximum batch size for exports.</param>
     /// <param name="scheduledDelayMilliseconds">The delay between exports in milliseconds.</param>
     /// <param name="exporterTimeoutMilliseconds">The timeout for export operations in milliseconds.</param>
+    /// <param name="exportStarted">Callback invoked when a batch is submitted to the exporter.</param>
     protected BatchExportWorker(
         CircularBuffer<T> circularBuffer,
         BaseExporter<T> exporter,
         int maxExportBatchSize,
         int scheduledDelayMilliseconds,
-        int exporterTimeoutMilliseconds)
+        int exporterTimeoutMilliseconds,
+        Action<long>? exportStarted = null)
     {
         this.CircularBuffer = circularBuffer;
         this.Exporter = exporter;
         this.MaxExportBatchSize = maxExportBatchSize;
         this.ScheduledDelayMilliseconds = scheduledDelayMilliseconds;
         this.ExporterTimeoutMilliseconds = exporterTimeoutMilliseconds;
+        this.exportStarted = exportStarted;
     }
 
     ~BatchExportWorker()
@@ -134,6 +138,7 @@ protected void PerformExport()
         {
             using (var batch = new Batch<T>(this.CircularBuffer, this.MaxExportBatchSize))
             {
+                this.exportStarted?.Invoke(batch.Count);
                 this.Exporter.Export(batch);
             }
         }

src/OpenTelemetry/Logs/Processor/BatchLogRecordExportProcessor.cs

@@ -16,7 +16,8 @@ public class BatchLogRecordExportProcessor : BatchExportProcessor<LogRecord>
     private readonly KeyValuePair<string, object?>[] successTags;
     private readonly KeyValuePair<string, object?>[] queueFullTags;
     private readonly KeyValuePair<string, object?>[] alreadyShutdownTags;
-    private volatile bool isShutdown;
+    private int activeOnEndCount;
+    private int isShutdown;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="BatchLogRecordExportProcessor"/> class.
@@ -49,69 +50,84 @@ public BatchLogRecordExportProcessor(
         this.successTags = baseTags;
         this.queueFullTags = [.. baseTags, new("error.type", "queue_full")];
         this.alreadyShutdownTags = [.. baseTags, new("error.type", "already_shutdown")];
+        this.ExportStarted = this.RecordSuccessfulProcessing;
     }
 
     /// <inheritdoc/>
     public override void OnEnd(LogRecord data)
     {
-        bool enqueued;
+        if (Volatile.Read(ref this.isShutdown) != 0)
+        {
+            SdkSelfObservability.LogProcessedCounter.Add(1, this.alreadyShutdownTags);
+            return;
+        }
+
+        Interlocked.Increment(ref this.activeOnEndCount);
+        try
+        {
+            if (Volatile.Read(ref this.isShutdown) != 0)
+            {
+                SdkSelfObservability.LogProcessedCounter.Add(1, this.alreadyShutdownTags);
+                return;
+            }
+
+            bool enqueued;
 
-        // Note: Intentionally not using Guard.ThrowIfNull to save prod cycles
+            // Note: Intentionally not using Guard.ThrowIfNull to save prod cycles
 #pragma warning disable CA1062 // Validate arguments of public methods
-        switch (data.Source)
+            switch (data.Source)
 #pragma warning restore CA1062 // Validate arguments of public methods
-        {
-            case LogRecord.LogRecordSource.FromSharedPool:
-                data.Buffer();
-                data.AddReference();
-                enqueued = this.TryExport(data);
-                if (!enqueued)
-                {
-                    LogRecordSharedPool.Current.Return(data);
-                }
-
-                break;
-
-            case LogRecord.LogRecordSource.CreatedManually:
-                data.Buffer();
-                enqueued = this.TryExport(data);
-                break;
-
-            case LogRecord.LogRecordSource.FromThreadStaticPool:
-            default:
-                Debug.Assert(data.Source == LogRecord.LogRecordSource.FromThreadStaticPool, "LogRecord source was something unexpected");
-
-                // Note: If we are using ThreadStatic pool we make a copy of the record.
-                enqueued = this.TryExport(data.Copy());
-                break;
-        }
+            {
+                case LogRecord.LogRecordSource.FromSharedPool:
+                    data.Buffer();
+                    data.AddReference();
+                    enqueued = this.TryExport(data);
+                    if (!enqueued)
+                    {
+                        LogRecordSharedPool.Current.Return(data);
+                    }
 
-        // TODO: Consider switching to an ObservableCounter that piggybacks on
-        // CircularBuffer.AddedCount and DroppedCount to eliminate per-item
-        // Counter.Add() overhead. This would require a registry pattern for
-        // multiple instances but avoids any hot-path cost when a listener is active.
-        KeyValuePair<string, object?>[] tags;
+                    break;
 
-        if (this.isShutdown)
-        {
-            tags = this.alreadyShutdownTags;
-        }
-        else if (!enqueued)
-        {
-            tags = this.queueFullTags;
+                case LogRecord.LogRecordSource.CreatedManually:
+                    data.Buffer();
+                    enqueued = this.TryExport(data);
+                    break;
+
+                case LogRecord.LogRecordSource.FromThreadStaticPool:
+                default:
+                    Debug.Assert(data.Source == LogRecord.LogRecordSource.FromThreadStaticPool, "LogRecord source was something unexpected");
+
+                    // Note: If we are using ThreadStatic pool we make a copy of the record.
+                    enqueued = this.TryExport(data.Copy());
+                    break;
+            }
+
+            if (!enqueued)
+            {
+                SdkSelfObservability.LogProcessedCounter.Add(1, this.queueFullTags);
+            }
         }
-        else
+        finally
         {
-            tags = this.successTags;
+            Interlocked.Decrement(ref this.activeOnEndCount);
         }
-
-        SdkSelfObservability.LogProcessedCounter.Add(1, tags);
     }
 
     /// <inheritdoc/>
     protected override bool OnShutdown(int timeoutMilliseconds)
     {
-        this.isShutdown = true;
+        Interlocked.Exchange(ref this.isShutdown, 1);
+
+        SpinWait spinner = default;
+        while (Volatile.Read(ref this.activeOnEndCount) != 0)
+        {
+            spinner.SpinOnce();
+        }
+
         return base.OnShutdown(timeoutMilliseconds);
     }
+
+    private void RecordSuccessfulProcessing(long count)
+        => SdkSelfObservability.LogProcessedCounter.Add(count, this.successTags);
 }

src/OpenTelemetry/Logs/Processor/SimpleLogRecordExportProcessor.cs

@@ -14,7 +14,8 @@ public class SimpleLogRecordExportProcessor : SimpleExportProcessor<LogRecord>
 
     private readonly KeyValuePair<string, object?>[] successTags;
     private readonly KeyValuePair<string, object?>[] alreadyShutdownTags;
-    private volatile bool isShutdown;
+    private int activeOnEndCount;
+    private int isShutdown;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="SimpleLogRecordExportProcessor"/> class.
@@ -37,15 +38,41 @@ public SimpleLogRecordExportProcessor(BaseExporter<LogRecord> exporter)
     /// <inheritdoc/>
     public override void OnEnd(LogRecord data)
     {
-        base.OnEnd(data);
-        SdkSelfObservability.LogProcessedCounter.Add(
-            1, this.isShutdown ? this.alreadyShutdownTags : this.successTags);
+        if (Volatile.Read(ref this.isShutdown) != 0)
+        {
+            SdkSelfObservability.LogProcessedCounter.Add(1, this.alreadyShutdownTags);
+            return;
+        }
+
+        Interlocked.Increment(ref this.activeOnEndCount);
+        try
+        {
+            if (Volatile.Read(ref this.isShutdown) != 0)
+            {
+                SdkSelfObservability.LogProcessedCounter.Add(1, this.alreadyShutdownTags);
+                return;
+            }
+
+            SdkSelfObservability.LogProcessedCounter.Add(1, this.successTags);
+            base.OnEnd(data);
+        }
+        finally
+        {
+            Interlocked.Decrement(ref this.activeOnEndCount);
+        }
     }
 
     /// <inheritdoc/>
     protected override bool OnShutdown(int timeoutMilliseconds)
     {
-        this.isShutdown = true;
+        Interlocked.Exchange(ref this.isShutdown, 1);
+
+        SpinWait spinner = default;
+        while (Volatile.Read(ref this.activeOnEndCount) != 0)
+        {
+            spinner.SpinOnce();
+        }
+
         return base.OnShutdown(timeoutMilliseconds);
     }
 }