Complete IAsyncLogEventSinkMonitor impl

Author: bartelink

README.md

@@ -19,10 +19,10 @@ 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();
 ```
 
@@ -32,32 +32,40 @@ Because the memory buffer may contain events that have not yet been written to t
 
 ### 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.IAsyncLogEventSinkState.DroppedMessagesCount` (see Sink State Inspection interface below).
+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)
+// Reduce the buffer to 500 events
+.WriteTo.Async(a => a.File("logs/myapp.log"), bufferSize: 500)
 ```
 
-### Health Monitoring via the Sink State Inspection interface
+### 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 frequency overwhelm the sink, or the sink ingestion throughput degrade to a major degree.
+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.
 
-In practice, this configuration (assuming one provisions an adequate `bufferSize`) achieves an efficient and resilient logging configuration that can handle load safely. The key risk is of course that events get be dropped when the buffer threshold gets breached. The `inspector` allows one to arrange for your Application's health monitoring mechanism to actively validate that the buffer allocation is not being exceeded in practice.
+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
-    // Example check: log message to an out of band alarm channel if logging is showing signs of getting overwhelmed
-    void PeriodicMonitorCheck(IQueueState inspector)
-    {
-        var usagePct = inspector.Count * 100 / inspector.BoundedCapacity;
-        if (usagePct > 50) SelfLog.WriteLine("Log buffer exceeded {0:p0} usage (limit: {1})", usagePct, inspector.BoundedCapacity);
-    }
-
-    // Allow a backlog of up to 10,000 items to be maintained (dropping extras if full)
-    .WriteTo.Async(a => a.File("logs/myapp.log"), inspector: out Async.IAsyncLogEventSinkState inspector) ...
-
-    // Wire the inspector through to application health monitoring and/or metrics in order to periodically emit a metric, raise an alarm, etc.
-    ... HealthMonitor.RegisterAsyncLogSink(inspector);
+// 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
@@ -67,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

@@ -59,59 +59,19 @@ public static LoggerConfiguration Async(
         /// <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. If the monitor implements <see cref="IDisposable"/>, <c>Dispose()</c> will be called to advise of the Sink being <c>Dispose()</c>d.</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,
-            bool blockWhenFull)
-        {
-            return LoggerSinkConfiguration.Wrap(
-                loggerSinkConfiguration,
-                wrappedSink =>
-                {
-                    var sink = new BackgroundWorkerSink(wrappedSink, bufferSize, blockWhenFull, monitor);
-                    monitor?.MonitorState(sink);
-                    return sink;
-                },
-                configure);
-        }
-
-        /// <summary>
-        /// Configure a sink to be invoked asynchronously, on a background worker thread.
-        /// Provides an <paramref name="inspector"/> that can be used to check the live state of the buffer 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="inspector">Provides a way to inspect the state of the queue for health monitoring purposes.</param>
-        /// <returns>A <see cref="LoggerConfiguration"/> allowing configuration to continue.</returns>
-        public static LoggerConfiguration Async(
-            this LoggerSinkConfiguration loggerSinkConfiguration,
-            Action<LoggerSinkConfiguration> configure,
-            out IAsyncLogEventSinkState inspector,
             int bufferSize = 10000,
             bool blockWhenFull = false)
         {
-            // Cannot assign directly to the out param from within the lambda, so we need a temp
-            IAsyncLogEventSinkState stateLens = null;
-            var result = LoggerSinkConfiguration.Wrap(
+            return LoggerSinkConfiguration.Wrap(
                 loggerSinkConfiguration,
-                wrappedSink =>
-                {
-                    var sink = new BackgroundWorkerSink(wrappedSink, bufferSize, blockWhenFull, null);
-                    stateLens = sink;
-                    return sink;
-                },
+                wrappedSink => new BackgroundWorkerSink(wrappedSink, bufferSize, blockWhenFull, monitor),
                 configure);
-            inspector = stateLens;
-            return result;
         }
     }
 }

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

@@ -8,7 +8,7 @@
 
 namespace Serilog.Sinks.Async
 {
-    sealed class BackgroundWorkerSink : ILogEventSink, IAsyncLogEventSinkState, IDisposable
+    sealed class BackgroundWorkerSink : ILogEventSink, IAsyncLogEventSinkInspector, IDisposable
     {
         readonly ILogEventSink _pipeline;
         readonly bool _blockWhenFull;
@@ -26,6 +26,7 @@ public BackgroundWorkerSink(ILogEventSink pipeline, int bufferCapacity, bool blo
             _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)
@@ -65,7 +66,7 @@ public void Dispose()
 
             (_pipeline as IDisposable)?.Dispose();
 
-            (_monitor as IDisposable)?.Dispose();
+            _monitor?.StopMonitoring(this);
         }
 
         void Pump()
@@ -83,10 +84,10 @@ void Pump()
             }
         }
 
-        int IAsyncLogEventSinkState.BufferSize => _queue.BoundedCapacity;
+        int IAsyncLogEventSinkInspector.BufferSize => _queue.BoundedCapacity;
 
-        int IAsyncLogEventSinkState.Count => _queue.Count;
+        int IAsyncLogEventSinkInspector.Count => _queue.Count;
 
-        long IAsyncLogEventSinkState.DroppedMessagesCount => _droppedMessages;
+        long IAsyncLogEventSinkInspector.DroppedMessagesCount => _droppedMessages;
     }
-}
+}

src/Serilog.Sinks.Async/Sinks/Async/IAsyncLogEventSinkState.cs renamed to src/Serilog.Sinks.Async/Sinks/Async/IAsyncLogEventSinkInspector.cs

@@ -1,9 +1,9 @@
 namespace Serilog.Sinks.Async
 {
     /// <summary>
-    /// Provides a way to monitor the state of Async wrapper's ingestion queue.
+    /// Provides a way to inspect the state of Async wrapper's ingestion queue.
     /// </summary>
-    public interface IAsyncLogEventSinkState
+    public interface IAsyncLogEventSinkInspector
     {
         /// <summary>
         /// Configured maximum number of items permitted to be held in the buffer awaiting ingestion.
@@ -18,7 +18,7 @@ public interface IAsyncLogEventSinkState
         int Count { get; }
 
         /// <summary>
-        /// Accumulated number of messages dropped due to breach of <see cref="BufferSize"/> limit.
+        /// 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

@@ -1,15 +1,20 @@
 namespace Serilog.Sinks.Async
 {
     /// <summary>
-    /// Defines a mechanism for the Async Sink to provide buffer metadata to facilitate integration into system health checking.
+    /// Defines a mechanism for the Async Sink to afford Health Checks a buffer metadata inspection mechanism.
     /// </summary>
-    /// <remarks>If the instance implements <see cref="System.IDisposable"/>, it will be <c>Dispose()</c>d at then time the Sink is.</remarks>
     public interface IAsyncLogEventSinkMonitor
     {
         /// <summary>
-        /// Invoked by Sink to supply the buffer state hook to the monitor.
+        /// Invoked by Sink to supply the inspector to the monitor.
         /// </summary>
-        /// <param name="state">The Async Sink's state information interface.</param>
-        void MonitorState(IAsyncLogEventSinkState state);
+        /// <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;
@@ -106,7 +107,7 @@ public async Task GivenDefaultConfig_WhenRequestsExceedCapacity_DoesNotBlock()
 
                 // Allow at least one to propagate
                 await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
-                Assert.NotEqual(0, ((IAsyncLogEventSinkState)sink).DroppedMessagesCount);
+                Assert.NotEqual(0, ((IAsyncLogEventSinkInspector)sink).DroppedMessagesCount);
             }
             // Sanity check the overall timing
             batchTiming.Stop();
@@ -145,7 +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, ((IAsyncLogEventSinkState)sink).DroppedMessagesCount);
+                Assert.NotEqual(0, ((IAsyncLogEventSinkInspector)sink).DroppedMessagesCount);
             }
         }
 
@@ -184,20 +185,23 @@ public async Task GivenConfiguredToBlock_WhenQueueFilled_ThenBlocks()
 
                 // No events should be dropped
                 Assert.Equal(3, _innerSink.Events.Count);
-                Assert.Equal(0, ((IAsyncLogEventSinkState)sink).DroppedMessagesCount);
+                Assert.Equal(0, ((IAsyncLogEventSinkInspector)sink).DroppedMessagesCount);
             }
         }
 
         [Fact]
-        public void InspectorOutParameterAffordsHealthMonitoringHook()
+        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, inspector: out IAsyncLogEventSinkState inspector)
+                .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);
@@ -222,6 +226,8 @@ public void InspectorOutParameterAffordsHealthMonitoringHook()
                 // 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()

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);
+    }
+}