feat(logs): add Buffering and Batching

src/Sentry/Internal/BatchBuffer.cs

@@ -0,0 +1,80 @@
+namespace Sentry.Internal;
+
+/// <summary>
+/// A slim wrapper over an <see cref="System.Array"/>,
+/// intended for buffering.
+/// </summary>
+internal sealed class BatchBuffer<T>
+{
+    private readonly T[] _array;
+    private int _count;
+
+    public BatchBuffer(int capacity)
+    {
+        ThrowIfNegativeOrZero(capacity, nameof(capacity));
+
+        _array = new T[capacity];
+        _count = 0;
+    }
+
+    internal int Count => _count;
+    internal int Capacity => _array.Length;
+    internal bool IsEmpty => _count == 0 && _array.Length != 0;
+    internal bool IsFull => _count == _array.Length;
+
+    internal bool TryAdd(T item)
+    {
+        if (_count < _array.Length)
+        {
+            _array[_count] = item;
+            _count++;
+            return true;
+        }
+
+        return false;
+    }
+
+    internal T[] ToArray()
+    {
+        if (_count == 0)
+        {
+            return Array.Empty<T>();
+        }
+
+        var array = new T[_count];
+        Array.Copy(_array, array, _count);
+        return array;
+    }
+
+    internal void Clear()
+    {
+        if (_count == 0)
+        {
+            return;
+        }
+
+        var count = _count;
+        _count = 0;
+        Array.Clear(_array, 0, count);
+    }
+
+    internal T[] ToArrayAndClear()
+    {
+        var array = ToArray();
+        Clear();
+        return array;
+    }
+
+    private static void ThrowIfNegativeOrZero(int capacity, string paramName)
+    {
+        if (capacity <= 0)
+        {
+            ThrowNegativeOrZero(capacity, paramName);
+        }
+    }
+
+    private static void ThrowNegativeOrZero(int capacity, string paramName)
+    {
+        throw new ArgumentOutOfRangeException(paramName, capacity, "Argument must neither be negative nor zero.");
+    }
+}

src/Sentry/Internal/BatchProcessor.cs

@@ -0,0 +1,93 @@
+using System.Timers;
+using Sentry.Protocol;
+using Sentry.Protocol.Envelopes;
+
+namespace Sentry.Internal;
+
+/// <summary>
+/// The Sentry Batch Processor.
+/// This implementation is not complete yet.
+/// Also, the specification is still work in progress.
+/// </summary>
+/// <remarks>
+/// Sentry Specification: <see href="https://develop.sentry.dev/sdk/telemetry/spans/batch-processor/"/>.
+/// OpenTelemetry spec: <see href="https://github.com/open-telemetry/opentelemetry-collector/blob/main/processor/batchprocessor/README.md"/>.
+/// </remarks>
+internal sealed class BatchProcessor : IDisposable
+{
+    private readonly IHub _hub;
+    private readonly BatchProcessorTimer _timer;
+    private readonly BatchBuffer<SentryLog> _logs;
+    private readonly object _lock;
+
+    private DateTime _lastFlush = DateTime.MinValue;
+
+    public BatchProcessor(IHub hub, int batchCount, TimeSpan batchInterval)
+        : this(hub, batchCount, new TimersBatchProcessorTimer(batchInterval))
+    {
+    }
+
+    public BatchProcessor(IHub hub, int batchCount, BatchProcessorTimer timer)
+    {
+        _hub = hub;
+
+        _timer = timer;
+        _timer.Elapsed += OnIntervalElapsed;
+
+        _logs = new BatchBuffer<SentryLog>(batchCount);
+        _lock = new object();
+    }
+
+    internal void Enqueue(SentryLog log)
+    {
+        lock (_lock)
+        {
+            EnqueueCore(log);
+        }
+    }
+
+    private void EnqueueCore(SentryLog log)
+    {
+        var isFirstLog = _logs.IsEmpty;
+        var added = _logs.TryAdd(log);
+        Debug.Assert(added, $"Since we currently have no lock-free scenario, it's unexpected to exceed the {nameof(BatchBuffer<SentryLog>)}'s capacity.");
+
+        if (isFirstLog && !_logs.IsFull)
+        {
+            _timer.Enabled = true;
+        }
+        else if (_logs.IsFull)
+        {
+            _timer.Enabled = false;
+            Flush();
+        }
+    }
+
+    private void Flush()
+    {
+        _lastFlush = DateTime.UtcNow;
+
+        var logs = _logs.ToArrayAndClear();
+        _ = _hub.CaptureEnvelope(Envelope.FromLog(new StructuredLog(logs)));
+    }
+
+    private void OnIntervalElapsed(object? sender, ElapsedEventArgs e)
+    {
+        _timer.Enabled = false;
+
+        lock (_lock)
+        {
+            if (!_logs.IsEmpty && e.SignalTime > _lastFlush)
+            {
+                Flush();
+            }
+        }
+    }
+
+    public void Dispose()
+    {
+        _timer.Enabled = false;
+        _timer.Elapsed -= OnIntervalElapsed;
+        _timer.Dispose();
+    }
+}

src/Sentry/Internal/BatchProcessorTimer.cs

@@ -0,0 +1,61 @@
+using System.Timers;
+
+namespace Sentry.Internal;
+
+internal abstract class BatchProcessorTimer : IDisposable
+{
+    protected BatchProcessorTimer()
+    {
+    }
+
+    public abstract bool Enabled { get; set; }
+
+    public abstract event EventHandler<ElapsedEventArgs> Elapsed;
+
+    public void Dispose()
+    {
+        Dispose(true);
+        GC.SuppressFinalize(this);
+    }
+
+    protected virtual void Dispose(bool disposing)
+    {
+    }
+}
+
+internal sealed class TimersBatchProcessorTimer : BatchProcessorTimer
+{
+    private readonly System.Timers.Timer _timer;
+
+    public TimersBatchProcessorTimer(TimeSpan interval)
+    {
+        _timer = new System.Timers.Timer(interval.TotalMilliseconds)
+        {
+            AutoReset = false,
+            Enabled = false,
+        };
+        _timer.Elapsed += OnElapsed;
+    }
+
+    public override bool Enabled
+    {
+        get => _timer.Enabled;
+        set => _timer.Enabled = value;
+    }
+
+    public override event EventHandler<ElapsedEventArgs>? Elapsed;
+
+    private void OnElapsed(object? sender, ElapsedEventArgs e)
+    {
+        Elapsed?.Invoke(sender, e);
+    }
+
+    protected override void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            _timer.Elapsed -= OnElapsed;
+            _timer.Dispose();
+        }
+    }
+}

src/Sentry/Internal/DefaultSentryStructuredLogger.cs

@@ -1,6 +1,5 @@
 using Sentry.Extensibility;
 using Sentry.Infrastructure;
-using Sentry.Protocol.Envelopes;
 
 namespace Sentry.Internal;
 
@@ -10,13 +9,33 @@ internal sealed class DefaultSentryStructuredLogger : SentryStructuredLogger
     private readonly SentryOptions _options;
     private readonly ISystemClock _clock;
 
+    private readonly BatchProcessor _batchProcessor;
+
     internal DefaultSentryStructuredLogger(IHub hub, SentryOptions options, ISystemClock clock)
     {
         Debug.Assert(options is { Experimental.EnableLogs: true });
 
         _hub = hub;
         _options = options;
         _clock = clock;
+
+        _batchProcessor = new BatchProcessor(hub, ClampBatchCount(options.Experimental.InternalBatchSize), ClampBatchInterval(options.Experimental.InternalBatchTimeout));
+    }
+
+    private static int ClampBatchCount(int batchCount)
+    {
+        return batchCount <= 0
+            ? 1
+            : batchCount > 1_000_000
+                ? 1_000_000
+                : batchCount;
+    }
+
+    private static TimeSpan ClampBatchInterval(TimeSpan batchInterval)
+    {
+        return batchInterval.TotalMilliseconds is <= 0 or > int.MaxValue
+            ? TimeSpan.FromMilliseconds(int.MaxValue)
+            : batchInterval;
     }
 
     private protected override void CaptureLog(SentryLogLevel level, string template, object[]? parameters, Action<SentryLog>? configureLog)
@@ -71,9 +90,17 @@ private protected override void CaptureLog(SentryLogLevel level, string template
 
         if (configuredLog is not null)
         {
-            //TODO: enqueue in Batch-Processor / Background-Worker
-            // see https://github.com/getsentry/sentry-dotnet/issues/4132
-            _ = _hub.CaptureEnvelope(Envelope.FromLog(configuredLog));
+            _batchProcessor.Enqueue(configuredLog);
         }
     }
+
+    protected override void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            _batchProcessor.Dispose();
+        }
+
+        base.Dispose(disposing);
+    }
 }

src/Sentry/Internal/Hub.cs

@@ -778,6 +778,8 @@ public void Dispose()
         _memoryMonitor?.Dispose();
 #endif
 
+        Logger.Dispose();
+
         try
         {
             CurrentClient.FlushAsync(_options.ShutdownTimeout).ConfigureAwait(false).GetAwaiter().GetResult();

src/Sentry/Protocol/Envelopes/Envelope.cs

@@ -445,17 +445,14 @@ internal static Envelope FromClientReport(ClientReport clientReport)
         return new Envelope(header, items);
     }
 
-    // TODO: This is temporary. We don't expect single log messages to become an envelope by themselves since batching is needed
     [Experimental(DiagnosticId.ExperimentalFeature)]
-    internal static Envelope FromLog(SentryLog log)
+    internal static Envelope FromLog(StructuredLog log)
     {
-        //TODO: allow batching Sentry logs
-        //see https://github.com/getsentry/sentry-dotnet/issues/4132
         var header = DefaultHeader;
 
         var items = new[]
         {
-            EnvelopeItem.FromLog(log)
+            EnvelopeItem.FromLog(log),
         };
 
         return new Envelope(header, items);

src/Sentry/Protocol/Envelopes/EnvelopeItem.cs

@@ -372,14 +372,12 @@ internal static EnvelopeItem FromClientReport(ClientReport report)
     }
 
     [Experimental(Infrastructure.DiagnosticId.ExperimentalFeature)]
-    internal static EnvelopeItem FromLog(SentryLog log)
+    internal static EnvelopeItem FromLog(StructuredLog log)
     {
-        //TODO: allow batching Sentry logs
-        //see https://github.com/getsentry/sentry-dotnet/issues/4132
         var header = new Dictionary<string, object?>(3, StringComparer.Ordinal)
         {
             [TypeKey] = TypeValueLog,
-            ["item_count"] = 1,
+            ["item_count"] = log.Length,
             ["content_type"] = "application/vnd.sentry.items.log+json",
         };
 

src/Sentry/Protocol/StructuredLog.cs

@@ -0,0 +1,42 @@
+using Sentry.Extensibility;
+
+namespace Sentry.Protocol;
+
+/// <summary>
+/// Represents the Sentry Log protocol.
+/// </summary>
+/// <remarks>
+/// Sentry Docs: <see href="https://docs.sentry.io/product/explore/logs/"/>.
+/// Sentry Developer Documentation: <see href="https://develop.sentry.dev/sdk/telemetry/logs/"/>.
+/// </remarks>
+internal sealed class StructuredLog : ISentryJsonSerializable
+{
+    private readonly SentryLog[] _items;
+
+    public StructuredLog(SentryLog log)
+    {
+        _items = [log];
+    }
+
+    public StructuredLog(SentryLog[] logs)
+    {
+        _items = logs;
+    }
+
+    public int Length => _items.Length;
+    public ReadOnlySpan<SentryLog> Items => _items;
+
+    public void WriteTo(Utf8JsonWriter writer, IDiagnosticLogger? logger)
+    {
+        writer.WriteStartObject();
+        writer.WriteStartArray("items");
+
+        foreach (var log in _items)
+        {
+            log.WriteTo(writer, logger);
+        }
+
+        writer.WriteEndArray();
+        writer.WriteEndObject();
+    }
+}