Merge pull request #29 from bartelink/feature/queue-monitoring

Add queue inspector facility

Author: nblumhardt

README.md

@@ -19,24 +19,53 @@ Log.Logger = new LoggerConfiguration()
     .WriteTo.Async(a => a.File("logs/myapp.log"))
     // Other logger configuration
     .CreateLogger()
-    
+
 Log.Information("This will be written to disk on the worker thread");
 
-// At application shutdown
+// At application shutdown (results in monitors getting StopMonitoring calls)
 Log.CloseAndFlush();
 ```
 
 The wrapped sink (`File` in this case) will be invoked on a worker thread while your application's thread gets on with more important stuff.
 
 Because the memory buffer may contain events that have not yet been written to the target sink, it is important to call `Log.CloseAndFlush()` or `Logger.Dispose()` when the application exits.
 
-### Buffering
+### Buffering & Dropping
+
+The default memory buffer feeding the worker thread is capped to 10,000 items, after which arriving events will be dropped. To increase or decrease this limit, specify it when configuring the async sink. One can determine whether events have been dropped via `Serilog.Async.IAsyncLogEventSinkInspector.DroppedMessagesCount` (see Sink State Inspection interface below).
+
+```csharp
+// Reduce the buffer to 500 events
+.WriteTo.Async(a => a.File("logs/myapp.log"), bufferSize: 500)
+```
+
+### Health Monitoring via the Monitor and Inspector interfaces
+
+The `Async` wrapper is primarily intended to allow one to achieve minimal logging latency at all times, even when writing to sinks that may momentarily block during the course of their processing (e.g., a `File` Sink might block for a low number of ms while flushing). The dropping behavior is an important failsafe; it avoids having an unbounded buffering behaviour should logging throughput overwhelm the sink, or the sink ingestion throughput degrade.
 
-The default memory buffer feeding the worker thread is capped to 10,000 items, after which arriving events will be dropped. To increase or decrease this limit, specify it when configuring the async sink.
+In practice, this configuration (assuming one provisions an adequate `bufferSize`) achieves an efficient and resilient logging configuration that can safely handle load without impacting processing throughput. The risk is of course that events get be dropped if the buffer threshold gets breached. The inspection interface, `IAsyncLogEventSinkInspector` (obtained by providing an `IAsyncLogEventSinkMonitor` when configuring the `Async` Sink), enables a health monitoring mechanism to actively validate that the buffer allocation is not being exceeded in practice.
 
 ```csharp
-    // Reduce the buffer to 500 events
-    .WriteTo.Async(a => a.File("logs/myapp.log"), bufferSize: 500)
+// Example check: log message to an out of band alarm channel if logging is showing signs of getting overwhelmed
+void ExecuteAsyncBufferCheck(IAsyncLogEventSinkInspector inspector)
+{
+    var usagePct = inspector.Count * 100 / inspector.BoundedCapacity;
+    if (usagePct > 50) SelfLog.WriteLine("Log buffer exceeded {0:p0} usage (limit: {1})", usagePct, inspector.BoundedCapacity);
+}
+
+class MonitorConfiguration : IAsyncLogEventSinkMonitor
+{
+    public void StartMonitoring(IAsyncLogEventSinkInspector inspector) =>
+        HealthMonitor.AddPeriodicCheck(() => ExecuteAsyncBufferCheck(inspector));
+
+    public void StopMonitoring(IAsyncLogEventSinkInspector inspector) 
+    { /* reverse of StartMonitoring */ }
+}
+
+// Provide monitor so we can wire the health check to the inspector
+var monitor = new MonitorConfiguration();
+// Use default config (drop events if >10,000 backlog)
+.WriteTo.Async(a => a.File("logs/myapp.log"), monitor: monitor) ...
 ```
 
 ### Blocking
@@ -46,9 +75,9 @@ Warning: For the same reason one typically does not want exceptions from logging
 When the buffer size limit is reached, the default behavior is to drop any further attempted writes until the queue abates, reporting each such failure to the `Serilog.Debugging.SelfLog`. To replace this with a blocking behaviour, set `blockWhenFull` to `true`.
 
 ```csharp
-    // Wait for any queued event to be accepted by the `File` log before allowing the calling thread
-    // to resume its application work after a logging call when there are 10,000 LogEvents waiting
-    .WriteTo.Async(a => a.File("logs/myapp.log"), blockWhenFull: true)
+// Wait for any queued event to be accepted by the `File` log before allowing the calling thread to resume its
+// application work after a logging call when there are 10,000 LogEvents awaiting ingestion by the pipeline
+.WriteTo.Async(a => a.File("logs/myapp.log"), blockWhenFull: true)
 ```
 
 ### XML `<appSettings>` and JSON configuration

src/Serilog.Sinks.Async/LoggerConfigurationAsyncExtensions.cs

@@ -44,12 +44,34 @@ public static LoggerConfiguration Async(
             Action<LoggerSinkConfiguration> configure,
             int bufferSize = 10000,
             bool blockWhenFull = false)
+        {
+            return loggerSinkConfiguration.Async(configure, null, bufferSize, blockWhenFull);
+        }
+
+        /// <summary>
+        /// Configure a sink to be invoked asynchronously, on a background worker thread.
+        /// Accepts a reference to a <paramref name="monitor"/> that will be supplied the internal state interface for health monitoring purposes.
+        /// </summary>
+        /// <param name="loggerSinkConfiguration">The <see cref="LoggerSinkConfiguration"/> being configured.</param>
+        /// <param name="configure">An action that configures the wrapped sink.</param>
+        /// <param name="bufferSize">The size of the concurrent queue used to feed the background worker thread. If
+        /// the thread is unable to process events quickly enough and the queue is filled, depending on
+        /// <paramref name="blockWhenFull"/> the queue will block or subsequent events will be dropped until
+        /// room is made in the queue.</param>
+        /// <param name="blockWhenFull">Block when the queue is full, instead of dropping events.</param>
+        /// <param name="monitor">Monitor to supply buffer information to.</param>
+        /// <returns>A <see cref="LoggerConfiguration"/> allowing configuration to continue.</returns>
+        public static LoggerConfiguration Async(
+            this LoggerSinkConfiguration loggerSinkConfiguration,
+            Action<LoggerSinkConfiguration> configure,
+            IAsyncLogEventSinkMonitor monitor,
+            int bufferSize = 10000,
+            bool blockWhenFull = false)
         {
             return LoggerSinkConfiguration.Wrap(
                 loggerSinkConfiguration,
-                wrappedSink => new BackgroundWorkerSink(wrappedSink, bufferSize, blockWhenFull),
+                wrappedSink => new BackgroundWorkerSink(wrappedSink, bufferSize, blockWhenFull, monitor),
                 configure);
         }
-
     }
-}
+}

src/Serilog.Sinks.Async/Sinks/Async/BackgroundWorkerSink.cs

@@ -8,21 +8,25 @@
 
 namespace Serilog.Sinks.Async
 {
-    sealed class BackgroundWorkerSink : ILogEventSink, IDisposable
+    sealed class BackgroundWorkerSink : ILogEventSink, IAsyncLogEventSinkInspector, IDisposable
     {
         readonly ILogEventSink _pipeline;
         readonly bool _blockWhenFull;
         readonly BlockingCollection<LogEvent> _queue;
         readonly Task _worker;
+        readonly IAsyncLogEventSinkMonitor _monitor;
 
-        public BackgroundWorkerSink(ILogEventSink pipeline, int bufferCapacity, bool blockWhenFull)
+        long _droppedMessages;
+
+        public BackgroundWorkerSink(ILogEventSink pipeline, int bufferCapacity, bool blockWhenFull, IAsyncLogEventSinkMonitor monitor = null)
         {
-            if (pipeline == null) throw new ArgumentNullException(nameof(pipeline));
             if (bufferCapacity <= 0) throw new ArgumentOutOfRangeException(nameof(bufferCapacity));
-            _pipeline = pipeline;
+            _pipeline = pipeline ?? throw new ArgumentNullException(nameof(pipeline));
             _blockWhenFull = blockWhenFull;
             _queue = new BlockingCollection<LogEvent>(bufferCapacity);
             _worker = Task.Factory.StartNew(Pump, CancellationToken.None, TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);
+            _monitor = monitor;
+            monitor?.StartMonitoring(this);
         }
 
         public void Emit(LogEvent logEvent)
@@ -39,7 +43,10 @@ public void Emit(LogEvent logEvent)
                 else
                 {
                     if (!_queue.TryAdd(logEvent))
+                    {
+                        Interlocked.Increment(ref _droppedMessages);
                         SelfLog.WriteLine("{0} unable to enqueue, capacity {1}", typeof(BackgroundWorkerSink), _queue.BoundedCapacity);
+                    }
                 }
             }
             catch (InvalidOperationException)
@@ -55,9 +62,11 @@ public void Dispose()
             _queue.CompleteAdding();
 
             // Allow queued events to be flushed
-            _worker.Wait();      
-            
+            _worker.Wait();
+
             (_pipeline as IDisposable)?.Dispose();
+
+            _monitor?.StopMonitoring(this);
         }
 
         void Pump()
@@ -74,5 +83,11 @@ void Pump()
                 SelfLog.WriteLine("{0} fatal error in worker thread: {1}", typeof(BackgroundWorkerSink), ex);
             }
         }
+
+        int IAsyncLogEventSinkInspector.BufferSize => _queue.BoundedCapacity;
+
+        int IAsyncLogEventSinkInspector.Count => _queue.Count;
+
+        long IAsyncLogEventSinkInspector.DroppedMessagesCount => _droppedMessages;
     }
-}
+}

src/Serilog.Sinks.Async/Sinks/Async/IAsyncLogEventSinkInspector.cs

@@ -0,0 +1,25 @@
+namespace Serilog.Sinks.Async
+{
+    /// <summary>
+    /// Provides a way to inspect the state of Async wrapper's ingestion queue.
+    /// </summary>
+    public interface IAsyncLogEventSinkInspector
+    {
+        /// <summary>
+        /// Configured maximum number of items permitted to be held in the buffer awaiting ingestion.
+        /// </summary>
+        /// <exception cref="T:System.ObjectDisposedException">The Sink has been disposed.</exception>
+        int BufferSize { get; }
+
+        /// <summary>
+        /// Current moment-in-time Count of items currently awaiting ingestion.
+        /// </summary>
+        /// <exception cref="T:System.ObjectDisposedException">The Sink has been disposed.</exception>
+        int Count { get; }
+
+        /// <summary>
+        /// Accumulated number of messages dropped due to breaches of <see cref="BufferSize"/> limit.
+        /// </summary>
+        long DroppedMessagesCount { get; }
+    }
+}

src/Serilog.Sinks.Async/Sinks/Async/IAsyncLogEventSinkMonitor.cs

@@ -0,0 +1,20 @@
+namespace Serilog.Sinks.Async
+{
+    /// <summary>
+    /// Defines a mechanism for the Async Sink to afford Health Checks a buffer metadata inspection mechanism.
+    /// </summary>
+    public interface IAsyncLogEventSinkMonitor
+    {
+        /// <summary>
+        /// Invoked by Sink to supply the inspector to the monitor.
+        /// </summary>
+        /// <param name="inspector">The Async Sink's inspector.</param>
+        void StartMonitoring(IAsyncLogEventSinkInspector inspector);
+
+        /// <summary>
+        /// Invoked by Sink to indicate that it is being Disposed.
+        /// </summary>
+        /// <param name="inspector">The Async Sink's inspector.</param>
+        void StopMonitoring(IAsyncLogEventSinkInspector inspector);
+    }
+}

test/Serilog.Sinks.Async.Tests/BackgroundWorkerSinkSpec.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Serilog.Core;
 using Serilog.Events;
@@ -25,7 +26,7 @@ public BackgroundWorkerSinkSpec()
         [Fact]
         public void WhenCtorWithNullSink_ThenThrows()
         {
-            Assert.Throws<ArgumentNullException>(() => new BackgroundWorkerSink(null, 10000, false));
+            Assert.Throws<ArgumentNullException>(() => new BackgroundWorkerSink(null, 10000, false, null));
         }
 
         [Fact]
@@ -84,12 +85,12 @@ public async Task WhenEmitMultipleTimes_ThenRelaysToInnerSink()
         }
 
         [Fact]
-        public async Task GivenDefaultConfig_WhenQueueOverCapacity_DoesNotBlock()
+        public async Task GivenDefaultConfig_WhenRequestsExceedCapacity_DoesNotBlock()
         {
             var batchTiming = Stopwatch.StartNew();
             using (var sink = new BackgroundWorkerSink(_logger, 1, blockWhenFull: false /*default*/))
             {
-                // Cause a delay when emmitting to the inner sink, allowing us to easily fill the queue to capacity
+                // Cause a delay when emitting to the inner sink, allowing us to easily fill the queue to capacity
                 // while the first event is being propagated
                 var acceptInterval = TimeSpan.FromMilliseconds(500);
                 _innerSink.DelayEmit = acceptInterval;
@@ -106,6 +107,7 @@ public async Task GivenDefaultConfig_WhenQueueOverCapacity_DoesNotBlock()
 
                 // Allow at least one to propagate
                 await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
+                Assert.NotEqual(0, ((IAsyncLogEventSinkInspector)sink).DroppedMessagesCount);
             }
             // Sanity check the overall timing
             batchTiming.Stop();
@@ -114,7 +116,7 @@ public async Task GivenDefaultConfig_WhenQueueOverCapacity_DoesNotBlock()
         }
 
         [Fact]
-        public async Task GivenDefaultConfig_WhenRequestsOverCapacity_ThenDropsEventsAndRecovers()
+        public async Task GivenDefaultConfig_WhenRequestsExceedCapacity_ThenDropsEventsAndRecovers()
         {
             using (var sink = new BackgroundWorkerSink(_logger, 1, blockWhenFull: false /*default*/))
             {
@@ -144,6 +146,7 @@ from e in _innerSink.Events
                 Assert.InRange(2, 2 * 3 / 2 - 1, propagatedExcludingFinal.Count());
                 // Final event should have made it through
                 Assert.Contains(_innerSink.Events, x => Object.ReferenceEquals(finalEvent, x));
+                Assert.NotEqual(0, ((IAsyncLogEventSinkInspector)sink).DroppedMessagesCount);
             }
         }
 
@@ -182,9 +185,51 @@ public async Task GivenConfiguredToBlock_WhenQueueFilled_ThenBlocks()
 
                 // No events should be dropped
                 Assert.Equal(3, _innerSink.Events.Count);
+                Assert.Equal(0, ((IAsyncLogEventSinkInspector)sink).DroppedMessagesCount);
             }
         }
 
+        [Fact]
+        public void MonitorParameterAffordsSinkInspectorSuitableForHealthChecking()
+        {
+            var collector = new MemorySink { DelayEmit = TimeSpan.FromSeconds(2) };
+            // 2 spaces in queue; 1 would make the second log entry eligible for dropping if consumer does not activate instantaneously
+            var bufferSize = 2;
+            var monitor = new DummyMonitor();
+            using (var logger = new LoggerConfiguration()
+                .WriteTo.Async(w => w.Sink(collector), bufferSize: 2, monitor: monitor)
+                .CreateLogger())
+            {
+                // Construction of BackgroundWorkerSink triggers StartMonitoring
+                var inspector = monitor.Inspector;
+                Assert.Equal(bufferSize, inspector.BufferSize);
+                Assert.Equal(0, inspector.Count);
+                Assert.Equal(0, inspector.DroppedMessagesCount);
+                logger.Information("Something to freeze the processing for 2s");
+                // Can be taken from queue either instantanously or be awaiting consumer to take
+                Assert.InRange(inspector.Count, 0, 1);
+                Assert.Equal(0, inspector.DroppedMessagesCount);
+                logger.Information("Something that will sit in the queue");
+                Assert.InRange(inspector.Count, 1, 2);
+                logger.Information("Something that will probably also sit in the queue (but could get dropped if first message has still not been picked up)");
+                Assert.InRange(inspector.Count, 1, 2);
+                logger.Information("Something that will get dropped unless we get preempted for 2s during our execution");
+                const string droppedMessage = "Something that will definitely get dropped";
+                logger.Information(droppedMessage);
+                Assert.InRange(inspector.Count, 1, 2);
+                // Unless we are put to sleep for a Rip Van Winkle period, either:
+                // a) the BackgroundWorker will be emitting the item [and incurring the 2s delay we established], leaving a single item in the buffer
+                // or b) neither will have been picked out of the buffer yet.
+                Assert.InRange(inspector.Count, 1, 2);
+                Assert.Equal(bufferSize, inspector.BufferSize);
+                Assert.DoesNotContain(collector.Events, x => x.MessageTemplate.Text == droppedMessage);
+                // Because messages wait 2 seconds, the only real way to get one into the buffer is with a debugger breakpoint or a sleep
+                Assert.InRange(collector.Events.Count, 0, 3);
+            }
+            // Dispose should trigger a StopMonitoring call
+            Assert.Null(monitor.Inspector);
+        }
+
         private BackgroundWorkerSink CreateSinkWithDefaultOptions()
         {
             return new BackgroundWorkerSink(_logger, 10000, false);

test/Serilog.Sinks.Async.Tests/BackgroundWorkerSinkTests.cs

@@ -34,5 +34,21 @@ public void DisposeCompletesWithoutWorkPerformed()
 
             Assert.Empty(collector.Events);
         }
+
+        [Fact]
+        public void CtorAndDisposeInformMonitor()
+        {
+            var collector = new MemorySink();
+            var monitor = new DummyMonitor();
+
+            using (new LoggerConfiguration()
+                .WriteTo.Async(w => w.Sink(collector), monitor: monitor)
+                .CreateLogger())
+            {
+                Assert.NotNull(monitor.Inspector);
+            }
+
+            Assert.Null(monitor.Inspector);
+        }
     }
-}
+}

test/Serilog.Sinks.Async.Tests/Support/DummyMonitor.cs

@@ -0,0 +1,14 @@
+namespace Serilog.Sinks.Async.Tests.Support
+{
+    class DummyMonitor : IAsyncLogEventSinkMonitor
+    {
+        IAsyncLogEventSinkInspector inspector;
+        public IAsyncLogEventSinkInspector Inspector => inspector;
+
+        void IAsyncLogEventSinkMonitor.StartMonitoring(IAsyncLogEventSinkInspector inspector) =>
+            this.inspector = inspector;
+
+        void IAsyncLogEventSinkMonitor.StopMonitoring(IAsyncLogEventSinkInspector inspector) =>
+            System.Threading.Interlocked.CompareExchange(ref this.inspector, null, inspector);
+    }
+}