Add otel.sdk.processor.log.processed metric

src/OpenTelemetry/.publicApi/Stable/PublicAPI.Unshipped.txt

@@ -1,3 +1,6 @@
 OpenTelemetry.Resources.Resource.Resource(System.Collections.Generic.IEnumerable<System.Collections.Generic.KeyValuePair<string!, object!>>! attributes, string? schemaUrl) -> void
 OpenTelemetry.Resources.Resource.SchemaUrl.get -> string?
 static OpenTelemetry.Resources.ResourceBuilderExtensions.AddAttributes(this OpenTelemetry.Resources.ResourceBuilder! resourceBuilder, System.Collections.Generic.IEnumerable<System.Collections.Generic.KeyValuePair<string!, object!>>! attributes, string? schemaUrl) -> OpenTelemetry.Resources.ResourceBuilder!
+override OpenTelemetry.SimpleLogRecordExportProcessor.OnEnd(OpenTelemetry.Logs.LogRecord! data) -> void
+override OpenTelemetry.BatchLogRecordExportProcessor.OnShutdown(int timeoutMilliseconds) -> bool
+override OpenTelemetry.SimpleLogRecordExportProcessor.OnShutdown(int timeoutMilliseconds) -> bool

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/Internal/MeterFactory.cs

@@ -0,0 +1,60 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+using System.Diagnostics.Metrics;
+using OpenTelemetry.Internal;
+
+namespace OpenTelemetry.Metrics;
+
+internal static class MeterFactory
+{
+    /// <summary>
+    /// Creates a new <see cref="Meter"/> for the assembly associated
+    /// with the specified type and semantic conventions version.
+    /// </summary>
+    /// <typeparam name="T">The type for which the <see cref="Meter"/> is created for its assembly.</typeparam>
+    /// <param name="semanticConventionsVersion">The version of the semantic conventions.</param>
+    /// <param name="tags">The optional tags to associate with the created <see cref="Meter"/>.</param>
+    /// <returns>A new <see cref="Meter"/> instance.</returns>
+    public static Meter Create<T>(Version? semanticConventionsVersion, IEnumerable<KeyValuePair<string, object?>>? tags = null)
+        => Create(typeof(T), semanticConventionsVersion, tags);
+
+    /// <summary>
+    /// Creates a new <see cref="Meter"/> for the assembly associated
+    /// with the specified type and semantic conventions version.
+    /// </summary>
+    /// <param name="type">The type for which the <see cref="Meter"/> is created for its assembly.</param>
+    /// <param name="semanticConventionsVersion">The version of the semantic conventions.</param>
+    /// <param name="tags">The optional tags to associate with the created <see cref="Meter"/>.</param>
+    /// <param name="name">The optional name to use for the <see cref="Meter"/> instead of the assembly name associated with <paramref name="type"/>.</param>
+    /// <returns>A new <see cref="Meter"/> instance.</returns>
+    public static Meter Create(Type type, Version? semanticConventionsVersion, IEnumerable<KeyValuePair<string, object?>>? tags = null, string? name = null)
+    {
+        Guard.ThrowIfNull(type);
+
+        var assembly = type.Assembly;
+        var assemblyName = assembly.GetName();
+        var version = assembly.GetPackageVersion();
+
+#pragma warning disable IDE0370 // Suppression is unnecessary
+        name ??= assemblyName.Name!;
+#pragma warning restore IDE0370 // Suppression is unnecessary
+
+        var options = new MeterOptions(name)
+        {
+            Version = version,
+        };
+
+        if (tags is not null)
+        {
+            options.Tags = tags;
+        }
+
+        if (semanticConventionsVersion is not null)
+        {
+            options.TelemetrySchemaUrl = $"https://opentelemetry.io/schemas/{semanticConventionsVersion.ToString(3)}";
+        }
+
+        return new(options);
+    }
+}

src/OpenTelemetry/Internal/SdkSelfObservability.cs

@@ -0,0 +1,21 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+using System.Diagnostics.Metrics;
+using OpenTelemetry.Metrics;
+
+namespace OpenTelemetry;
+
+/// <summary>
+/// Shared infrastructure for SDK self-observability metrics.
+/// </summary>
+internal static class SdkSelfObservability
+{
+    internal static readonly Meter Meter = MeterFactory.Create(
+        typeof(SdkSelfObservability), semanticConventionsVersion: null, name: "otel.sdk.experimental");
+
+    internal static readonly Counter<long> LogProcessedCounter = Meter.CreateCounter<long>(
+        "otel.sdk.processor.log.processed",
+        "{log_record}",
+        "The number of log records for which the processing has finished, either successful or failed.");
+}

src/OpenTelemetry/Logs/Processor/BatchLogRecordExportProcessor.cs

@@ -11,6 +11,14 @@ namespace OpenTelemetry;
 /// </summary>
 public class BatchLogRecordExportProcessor : BatchExportProcessor<LogRecord>
 {
+    private static int instanceCounter = -1;
+
+    private readonly KeyValuePair<string, object?>[] successTags;
+    private readonly KeyValuePair<string, object?>[] queueFullTags;
+    private readonly KeyValuePair<string, object?>[] alreadyShutdownTags;
+    private int activeOnEndCount;
+    private int isShutdown;
+
     /// <summary>
     /// Initializes a new instance of the <see cref="BatchLogRecordExportProcessor"/> class.
     /// </summary>
@@ -32,38 +40,94 @@ public BatchLogRecordExportProcessor(
             exporterTimeoutMilliseconds,
             maxExportBatchSize)
     {
+        var index = Interlocked.Increment(ref instanceCounter);
+        var componentName = "batching_log_processor/" + index.ToString(System.Globalization.CultureInfo.InvariantCulture);
+        var baseTags = new KeyValuePair<string, object?>[]
+        {
+            new("otel.component.type", "batching_log_processor"),
+            new("otel.component.name", componentName),
+        };
+        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)
     {
-        // Note: Intentionally not using Guard.ThrowIfNull to save prod cycles
+        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
 #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;
+            }
+
+            if (!enqueued)
+            {
+                SdkSelfObservability.LogProcessedCounter.Add(1, this.queueFullTags);
+            }
+        }
+        finally
+        {
+            Interlocked.Decrement(ref this.activeOnEndCount);
+        }
+    }
+
+    /// <inheritdoc/>
+    protected override bool OnShutdown(int timeoutMilliseconds)
+    {
+        Interlocked.Exchange(ref this.isShutdown, 1);
+
+        SpinWait spinner = default;
+        while (Volatile.Read(ref this.activeOnEndCount) != 0)
         {
-            case LogRecord.LogRecordSource.FromSharedPool:
-                data.Buffer();
-                data.AddReference();
-                if (!this.TryExport(data))
-                {
-                    LogRecordSharedPool.Current.Return(data);
-                }
-
-                break;
-
-            case LogRecord.LogRecordSource.CreatedManually:
-                data.Buffer();
-                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.
-                this.TryExport(data.Copy());
-                break;
+            spinner.SpinOnce();
         }
+
+        return base.OnShutdown(timeoutMilliseconds);
     }
+
+    private void RecordSuccessfulProcessing(long count)
+        => SdkSelfObservability.LogProcessedCounter.Add(count, this.successTags);
 }