send syslog datagrams through BackgroundSender #314

The appender's own pump held an unbounded queue, so a syslog server that stopped
accepting datagrams grew it until the process ran out of memory. Shutdown waited
five seconds and then abandoned a drain that had no limit of its own.

It now holds SendQueueSize datagrams (500), a logging call waits at most
EnqueueTimeoutMillis (5000) for room, losses are counted, and Flush honours its
timeout: this appender sends asynchronously, so unlike a buffering one the timeout
means something here.

Author: FreeAndNil

src/changelog/3.5.0/314-remote-syslog-background-sender.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<entry xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns="https://logging.apache.org/xml/ns"
+       xsi:schemaLocation="https://logging.apache.org/xml/ns https://logging.apache.org/xml/ns/log4j-changelog-0.xsd"
+       type="fixed">
+  <issue id="314" link="https://github.com/apache/logging-log4net/pull/314"/>
+  <description format="asciidoc">
+    bound the queue `RemoteSyslogAppender` sends from. It was unbounded, so a syslog server that
+ stopped accepting datagrams grew it until the process ran out of memory, and shutdown waited five
+ seconds and then abandoned a drain that had no limit of its own. It now holds `sendQueueSize`
+ datagrams (500), a logging call waits at most `enqueueTimeoutMillis` (5000) for room, losses are
+ counted and reported, and `Flush` honours its timeout (implemented by @FreeAndNil)
+  </description>
+</entry>

src/log4net.Tests/Appender/RemoteSyslogAppenderTest.cs

@@ -21,7 +21,6 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
-using System.Threading;
 using log4net.Appender;
 using log4net.Appender.Internal;
 using log4net.Core;
@@ -68,6 +67,31 @@ private sealed class RecordingErrorHandler : IErrorHandler
     public void Error(string message) => Messages.Add(message);
   }
 
+  private const int FlushTimeoutMillis = 30_000;
+
+  /// <summary>Bounded, so an unreachable server cannot grow the queue without limit.</summary>
+  [Test]
+  public void SendQueueSizeDefaultsTo500()
+    => Assert.That(new RemoteSyslogAppender().SendQueueSize, Is.EqualTo(500));
+
+  /// <summary>The longest a logging call waits when the queue is full.</summary>
+  [Test]
+  public void EnqueueTimeoutMillisDefaultsTo5000()
+    => Assert.That(new RemoteSyslogAppender().EnqueueTimeoutMillis, Is.EqualTo(5_000));
+
+  /// <summary>A queue with no room for anything cannot work.</summary>
+  [TestCase(0)]
+  [TestCase(-1)]
+  public void SendQueueSizeRejectsValuesThatAreNotPositive(int value)
+    => Assert.That(() => new RemoteSyslogAppender().SendQueueSize = value,
+      Throws.TypeOf<ArgumentOutOfRangeException>());
+
+  /// <summary>0 is allowed and never waits, a negative wait is meaningless.</summary>
+  [Test]
+  public void EnqueueTimeoutMillisRejectsANegativeValue()
+    => Assert.That(() => new RemoteSyslogAppender().EnqueueTimeoutMillis = -1,
+      Throws.TypeOf<ArgumentOutOfRangeException>());
+
   /// <summary>
   /// This appender sends through the connection its pump owns, so the inherited one would be a
   /// second socket, holding a second binding of LocalPort, that nothing ever uses.
@@ -262,13 +286,7 @@ private static List<byte[]> ExecuteAppend(string message,
       Domain = "TestDomain",
     });
     appender.DoAppend(loggingEvent);
-    for (int i = 0; i < 20; i++)
-    {
-      if (appender.Mock.Sent.Count == 0)
-      {
-        Thread.Sleep(10);
-      }
-    }
+    Assert.That(appender.Flush(FlushTimeoutMillis), Is.True);
     appender.Close();
     Assert.That(appender.Mock.ConnectedTo, Is.EqualTo((0, ipAddress, 514)));
     Assert.That(appender.Mock.WasDisposed, Is.True);

src/log4net/Appender/RemoteSyslogAppender.cs

@@ -18,10 +18,8 @@
 #endregion
 
 using System;
-using System.Collections.Concurrent;
 using System.Text;
 using System.Threading;
-using System.Threading.Tasks;
 using log4net.Appender.Internal;
 using log4net.Core;
 using log4net.Layout;
@@ -279,9 +277,11 @@ public enum SyslogNewLineHandling
     Keep
   }
 
-  private readonly BlockingCollection<byte[]> _sendQueue = new();
-  private CancellationTokenSource? _cancellationTokenSource;
-  private Task? _pumpTask;
+  private const int CloseTimeoutMillis = 5_000;
+  private IUdpConnection? _connection;
+  private BackgroundSender<byte[]>? _sender;
+  private int _sendQueueSize = 500;
+  private int _enqueueTimeoutMillis = 5_000;
 
   /// <summary>
   /// Initializes a new instance of the <see cref="RemoteSyslogAppender" /> class.
@@ -389,7 +389,8 @@ protected override void Append(LoggingEvent loggingEvent)
         // Grab as a byte array
         byte[] buffer = Encoding.GetBytes(builder.ToString());
 
-        _sendQueue.Add(buffer);
+        // A full queue is reported by the sender itself.
+        _sender?.TryEnqueue(buffer, EnqueueTimeoutMillis);
       }
     }
     catch (Exception e) when (!e.IsFatal())
@@ -524,10 +525,89 @@ public override void ActivateOptions()
         $"The NewLineHandling is not {SyslogNewLineHandling.Escape} or {SyslogNewLineHandling.Keep} or {SyslogNewLineHandling.Split}.");
     }
     _levelMapping.ActivateOptions();
-    // Start the background pump
-    _cancellationTokenSource = new();
-    _pumpTask = Task.Factory.StartNew(() => ProcessQueueAsync(_cancellationTokenSource.Token), CancellationToken.None,
-      TaskCreationOptions.LongRunning, TaskScheduler.Default);
+    StopSending();
+
+    IUdpConnection connection = CreateUdpConnection();
+    try
+    {
+      connection.Connect(LocalPort, RemoteAddress.EnsureNotNull(), RemotePort);
+    }
+    catch (Exception e) when (!e.IsFatal())
+    {
+      ErrorHandler.Error(
+        $"Unable to connect to remote syslog {RemoteAddress} on port {RemotePort} from local port {LocalPort}. "
+        + "No logging event will be sent.",
+        e, ErrorCode.GenericFailure);
+      connection.Dispose();
+      return;
+    }
+
+    _connection = connection;
+    _sender = new(nameof(RemoteSyslogAppender), SendQueueSize, Send, Report);
+  }
+
+  /// <summary>
+  /// How many datagrams may wait to be sent. Defaults to 500.
+  /// </summary>
+  /// <exception cref="ArgumentOutOfRangeException">The value specified is not positive.</exception>
+  public int SendQueueSize
+  {
+    get => _sendQueueSize;
+    set
+    {
+      if (value <= 0)
+      {
+        throw SystemInfo.CreateArgumentOutOfRangeException(nameof(value), value,
+          "The value specified for SendQueueSize is not positive.");
+      }
+      _sendQueueSize = value;
+    }
+  }
+
+  /// <summary>
+  /// How long a logging call waits for room in a full queue. Defaults to 5000, 0 never waits.
+  /// </summary>
+  /// <exception cref="ArgumentOutOfRangeException">The value specified is negative.</exception>
+  public int EnqueueTimeoutMillis
+  {
+    get => _enqueueTimeoutMillis;
+    set
+    {
+      if (value < 0)
+      {
+        throw SystemInfo.CreateArgumentOutOfRangeException(nameof(value), value,
+          "The value specified for EnqueueTimeoutMillis is negative.");
+      }
+      _enqueueTimeoutMillis = value;
+    }
+  }
+
+  private void Send(byte[] datagram, CancellationToken cancellationToken)
+    => _connection?.SendAsync(datagram, datagram.Length).GetAwaiter().GetResult();
+
+  private void Report(string message, Exception? exception)
+  {
+    if (exception is null)
+    {
+      ErrorHandler.Error(message);
+    }
+    else
+    {
+      ErrorHandler.Error(message, exception);
+    }
+  }
+
+  private void StopSending()
+  {
+    if (_sender is BackgroundSender<byte[]> sender)
+    {
+      _sender = null;
+      sender.Close(CloseTimeoutMillis);
+      sender.Dispose();
+    }
+
+    _connection?.Dispose();
+    _connection = null;
   }
 
   /// <summary>
@@ -636,12 +716,17 @@ public class LevelSeverity : LevelMappingEntry
     public SyslogSeverity Severity { get; set; }
   }
 
+  /// <summary>
+  /// Waits until everything queued so far has been sent.
+  /// </summary>
+  /// <param name="millisecondsTimeout">The maximum time to wait.</param>
+  /// <returns><see langword="false"/> on timeout, or when the sender is no longer running.</returns>
+  public override bool Flush(int millisecondsTimeout) => _sender?.Flush(millisecondsTimeout) ?? true;
+
   /// <inheritdoc/>
   protected override void OnClose()
   {
-    // Signal shutdown and wait for the pump to drain
-    _cancellationTokenSource?.Cancel();
-    _pumpTask?.Wait(TimeSpan.FromSeconds(5));
+    StopSending();
     base.OnClose();
   }
 
@@ -653,58 +738,4 @@ protected override void OnClose()
   protected override void InitializeClientConnection()
   {
   }
-
-  private async Task ProcessQueueAsync(CancellationToken token)
-  {
-    // We create our own UdpClient here, so that client lifetime is tied to this task
-    IUdpConnection udpClient;
-    try
-    {
-      udpClient = CreateUdpConnection();
-      udpClient.Connect(LocalPort, RemoteAddress.EnsureNotNull(), RemotePort);
-    }
-    catch (Exception e) when (!e.IsFatal())
-    {
-      // Outside the loop below, so this would otherwise fault the pump unobserved and leave
-      // every later event queued behind a sender that is no longer running.
-      ErrorHandler.Error(
-        $"Unable to connect to remote syslog {RemoteAddress} on port {RemotePort} from local port {LocalPort}. "
-        + "No logging event will be sent.",
-        e, ErrorCode.GenericFailure);
-      return;
-    }
-
-    using IUdpConnection connection = udpClient;
-    try
-    {
-      while (!token.IsCancellationRequested)
-      {
-        // Take next message or throw when canceled
-        byte[] datagram = _sendQueue.Take(token);
-        try
-        {
-          await udpClient.SendAsync(datagram, datagram.Length).ConfigureAwait(false);
-        }
-        catch (Exception ex) when (!ex.IsFatal())
-        {
-          ErrorHandler.Error("RemoteSyslogAppender: send failed", ex, ErrorCode.WriteFailure);
-        }
-      }
-    }
-    catch (OperationCanceledException)
-    {
-      // Clean shutdown: drain remaining items if desired
-      while (_sendQueue.TryTake(out byte[]? leftover))
-      {
-        try
-        {
-          await udpClient.SendAsync(leftover, leftover.Length).ConfigureAwait(false);
-        }
-        catch (Exception ex) when (!ex.IsFatal())
-        {
-          ErrorHandler.Error("RemoteSyslogAppender: send failed during shutdown", ex, ErrorCode.FlushFailure);
-        }
-      }
-    }
-  }
 }

src/site/antora/modules/ROOT/pages/manual/configuration/appenders/remotesyslogappender.adoc

@@ -42,6 +42,13 @@ You can also specify:
 
 * Facility (default: user)
 * Identity (default: application name)
+* SendQueueSize (default: 500), how many datagrams may wait to be sent
+* EnqueueTimeoutMillis (default: 5000), how long a logging call waits for room in a full queue
+
+Datagrams are handed to a background thread, so a slow or unreachable syslog server does not hold
+up logging. Once the queue is full, a logging call waits `EnqueueTimeoutMillis` for room and the
+datagram is then discarded and counted, rather than growing the queue without limit. `Flush` waits
+for the queue to drain.
 
 [source,xml]
 ----