Console log buffer PR feedback and improvements:

- implement flush on dispose, so that logs are flushed when process shuts down
- specify reader/writer usage of the channel
- Change the write to console path to use WriteAsync instead of WaitToWriteAsync
- Use TryComplete instead of Complete so that repeated completes do not throw
- fix bug where a line could be read from buffer but not written to output

Author: paulbatum

src/WebJobs.Script.WebHost/Diagnostics/ConsoleWriter.cs

@@ -3,24 +3,27 @@
 
 using System;
 using System.Text;
+using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
 
 namespace Microsoft.Azure.WebJobs.Script.WebHost.Diagnostics
 {
-    internal class ConsoleWriter
+    internal class ConsoleWriter : IDisposable
     {
         // A typical out-of-proc function execution will generate 8 log lines.
         // A single core container can potentially get around 1K RPS at the higher end, and a typical log line is around 300 bytes
         // So in the extreme case, this is about 1 second of buffer and should be less than 3MB
         private const int DefaultBufferSize = 8000;
 
+        private static readonly TimeSpan DisposeTimeout = TimeSpan.FromSeconds(5);
         private static readonly TimeSpan DefaultConsoleBufferTimeout = TimeSpan.FromSeconds(1);
         private readonly Channel<string> _consoleBuffer;
         private readonly TimeSpan _consoleBufferTimeout;
         private readonly Action<Exception> _exceptionhandler;
         private Task _consoleBufferReadLoop;
         private Action<string> _writeEvent;
+        private bool _disposed;
 
         public ConsoleWriter(IEnvironment environment, Action<Exception> unhandledExceptionHandler)
             : this(environment, unhandledExceptionHandler, consoleBufferTimeout: DefaultConsoleBufferTimeout, autoStart: true)
@@ -36,10 +39,10 @@ internal ConsoleWriter(IEnvironment environment, Action<Exception> exceptionHand
                 // We are going to used stdout, but do we write directly or use a buffer?
                 _consoleBuffer = environment.GetEnvironmentVariable(EnvironmentSettingNames.ConsoleLoggingBufferSize) switch
                 {
-                    "-1" => Channel.CreateUnbounded<string>(),              // buffer size of -1 indicates that buffer should be enabled but unbounded
-                    "0" => null,                                            // buffer size of 0 indicates that buffer should be disabled
+                    "-1" => Channel.CreateUnbounded<string>(new UnboundedChannelOptions() { SingleReader = true, SingleWriter = false }),           // buffer size of -1 indicates that buffer should be enabled but unbounded
+                    "0" => null,                                                                                                                    // buffer size of 0 indicates that buffer should be disabled
                     var s when int.TryParse(s, out int i) && i > 0 => Channel.CreateBounded<string>(i),
-                    _ => Channel.CreateBounded<string>(DefaultBufferSize),  // default behavior is to use buffer with default size
+                    _ => Channel.CreateBounded<string>(new BoundedChannelOptions(DefaultBufferSize) { SingleReader = true, SingleWriter = false }), // default behavior is to use buffer with default size
                 };
 
                 if (_consoleBuffer == null)
@@ -79,17 +82,20 @@ private void WriteToConsoleBuffer(string evt)
         {
             try
             {
-                while (_consoleBuffer.Writer.TryWrite(evt) == false)
+                if (_consoleBuffer.Writer.TryWrite(evt) == false)
                 {
                     // Buffer is currently full, wait until writing is permitted.
-                    // This is the downside of using channels, we are on a sync code path and so we have to block on this task
-                    var writeTask = _consoleBuffer.Writer.WaitToWriteAsync().AsTask();
-                    if (writeTask.WaitAsync(_consoleBufferTimeout).Result == false)
+                    using var source = new CancellationTokenSource(_consoleBufferTimeout);
+                    var writeTask = _consoleBuffer.Writer.WriteAsync(evt, source.Token);
+
+                    // This is the downside of using channels, we are on a sync code path and so we have to block on this task if we want to wait for the buffer to clear.
+                    if (writeTask.IsCompleted)
                     {
-                        // The buffer has been completed and does not allow further writes.
-                        // Currently this should not be possible, but the safest thing to do is just write directly to the console.
-                        Console.WriteLine(evt);
-                        break;
+                        writeTask.GetAwaiter().GetResult();
+                    }
+                    else
+                    {
+                        writeTask.AsTask().GetAwaiter().GetResult();
                     }
                 }
             }
@@ -133,7 +139,7 @@ private async Task ProcessConsoleBufferAsync(bool batched)
             {
                 // if this has failed for any reason, fall everything back to console
                 _writeEvent = Console.WriteLine;
-                _consoleBuffer.Writer.Complete();
+                _consoleBuffer.Writer.TryComplete();
             }
         }
 
@@ -149,15 +155,8 @@ private async Task ProcessConsoleBufferBatchedAsync()
         {
             var builder = new StringBuilder();
 
-            while (true)
+            while (await _consoleBuffer.Reader.WaitToReadAsync())
             {
-                if (await _consoleBuffer.Reader.WaitToReadAsync() == false)
-                {
-                    // The buffer has been completed and does not allow further reads.
-                    // Currently this should not be possible, but safest thing to do is break out of the loop.
-                    break;
-                }
-
                 if (_consoleBuffer.Reader.TryRead(out string line1))
                 {
                     // Can we synchronously read multiple lines?
@@ -169,7 +168,7 @@ private async Task ProcessConsoleBufferBatchedAsync()
                         builder.AppendLine(line2);
                         int lines = 2;
 
-                        while (_consoleBuffer.Reader.TryRead(out string nextLine) && lines < DefaultBufferSize)
+                        while (lines < DefaultBufferSize && _consoleBuffer.Reader.TryRead(out string nextLine))
                         {
                             builder.AppendLine(nextLine);
                             lines++;
@@ -185,5 +184,28 @@ private async Task ProcessConsoleBufferBatchedAsync()
                 }
             }
         }
+
+        protected virtual void Dispose(bool disposing)
+        {
+            if (!_disposed)
+            {
+                if (disposing)
+                {
+                    if (_consoleBuffer != null)
+                    {
+                        _consoleBuffer.Writer.TryComplete();
+                        _consoleBufferReadLoop.Wait(DisposeTimeout);
+                    }
+                }
+
+                _disposed = true;
+            }
+        }
+
+        public void Dispose()
+        {
+            Dispose(disposing: true);
+            GC.SuppressFinalize(this);
+        }
     }
 }

src/WebJobs.Script.WebHost/Diagnostics/LinuxContainerEventGenerator.cs

@@ -2,15 +2,11 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
-using System.Text;
-using System.Threading;
-using System.Threading.Channels;
-using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 
 namespace Microsoft.Azure.WebJobs.Script.WebHost.Diagnostics
 {
-    internal class LinuxContainerEventGenerator : LinuxEventGenerator
+    internal class LinuxContainerEventGenerator : LinuxEventGenerator, IDisposable
     {
         private const int MaxDetailsLength = 10000;
         private static readonly Lazy<LinuxContainerEventGenerator> _Lazy = new Lazy<LinuxContainerEventGenerator>(() => new LinuxContainerEventGenerator(SystemEnvironment.Instance, Console.WriteLine));
@@ -20,6 +16,7 @@ internal class LinuxContainerEventGenerator : LinuxEventGenerator
         private string _containerName;
         private string _stampName;
         private string _tenantId;
+        private bool _disposed;
 
         public LinuxContainerEventGenerator(IEnvironment environment, Action<string> writeEvent = null)
         {
@@ -152,5 +149,23 @@ public static void LogEvent(string message, Exception e = null, LogLevel logLeve
                 slotName: SystemEnvironment.Instance.GetSlotName() ?? string.Empty,
                 eventTimestamp: DateTime.UtcNow);
         }
+
+        protected virtual void Dispose(bool disposing)
+        {
+            if (!_disposed)
+            {
+                if (disposing)
+                {
+                    _consoleWriter?.Dispose();
+                }
+                _disposed = true;
+            }
+        }
+
+        public void Dispose()
+        {
+            Dispose(disposing: true);
+            GC.SuppressFinalize(this);
+        }
     }
 }

test/WebJobs.Script.Tests/Diagnostics/LinuxContainerEventGeneratorWithConsoleOutputTests.cs

@@ -4,6 +4,7 @@
 using System;
 using System.IO;
 using System.IO.Abstractions.TestingHelpers;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Azure.WebJobs.Script.WebHost.Diagnostics;
 using Microsoft.Extensions.Logging;
@@ -153,7 +154,7 @@ public void MultipleEventsWithTinyBuffer_WritesDirectlyToConsoleOnTimeout(bool b
             // The first two messages are still stuck in the buffer. The third message will have been written to the console.
             // We should also have a log for the timeout exception that occurred while waiting for the buffer to become available.
             Assert.Equal(2, output.Length);
-            Assert.StartsWith("MS_FUNCTION_LOGS 2,,,,,LogUnhandledException,\"System.AggregateException: One or more errors occurred. (The operation has timed out.)", output[0]);
+            Assert.StartsWith("MS_FUNCTION_LOGS 2,,,,,LogUnhandledException,\"System.OperationCanceledException: The operation was canceled.", output[0]);
             Assert.Equal($"MS_FUNCTION_LOGS 4,C37E3412-86D1-4B93-BC5A-A2AE09D26C2D,TestApp,TestFunction3,TestEvent,TestSource,\"These are the details, lots of details\",\"This is the summary, a great summary\",{ScriptHost.Version},{timestamp.ToString("O")},TestExceptionType,\"Test exception message, with details\",E2D5A6ED-4CE3-4CFD-8878-FD4814F0A1F3,3AD41658-1C4E-4C9D-B0B9-24F2BDAE2829,F0AAA9AD-C3A6-48B9-A75E-57BB280EBB53,TEST-CONTAINER,test-stamp,test-tenant,TestRuntimeSiteName,TestSlotName", output[1]);
         }
 
@@ -175,7 +176,7 @@ public async Task MultipleEventsBatchedWithTinyBuffer_BlocksOnWriteOnBufferFull_
             Assert.False(logTask.IsCompleted);
 
             consoleWriter.StartProcessingBuffer(batched: true);
-            await Task.Delay(TimeSpan.FromMilliseconds(10));
+            await Task.Delay(TimeSpan.FromMilliseconds(50));
             Assert.True(logTask.IsCompleted);
 
             using var sr = new StreamReader(_consoleOut);
@@ -188,5 +189,72 @@ public async Task MultipleEventsBatchedWithTinyBuffer_BlocksOnWriteOnBufferFull_
             Assert.Equal($"MS_FUNCTION_LOGS 4,C37E3412-86D1-4B93-BC5A-A2AE09D26C2D,TestApp,TestFunction2,TestEvent,TestSource,\"These are the details, lots of details\",\"This is the summary, a great summary\",{ScriptHost.Version},{timestamp.ToString("O")},TestExceptionType,\"Test exception message, with details\",E2D5A6ED-4CE3-4CFD-8878-FD4814F0A1F3,3AD41658-1C4E-4C9D-B0B9-24F2BDAE2829,F0AAA9AD-C3A6-48B9-A75E-57BB280EBB53,TEST-CONTAINER,test-stamp,test-tenant,TestRuntimeSiteName,TestSlotName", output[1]);
             Assert.Equal($"MS_FUNCTION_LOGS 4,C37E3412-86D1-4B93-BC5A-A2AE09D26C2D,TestApp,TestFunction3,TestEvent,TestSource,\"These are the details, lots of details\",\"This is the summary, a great summary\",{ScriptHost.Version},{timestamp.ToString("O")},TestExceptionType,\"Test exception message, with details\",E2D5A6ED-4CE3-4CFD-8878-FD4814F0A1F3,3AD41658-1C4E-4C9D-B0B9-24F2BDAE2829,F0AAA9AD-C3A6-48B9-A75E-57BB280EBB53,TEST-CONTAINER,test-stamp,test-tenant,TestRuntimeSiteName,TestSlotName", output[2]);
         }
+
+        [Fact]
+        public async Task FlushesBufferOnDispose()
+        {
+            var env = CreateEnvironment(bufferSize: 10, batched: true);
+            var consoleWriter = new ConsoleWriter(env, LinuxContainerEventGenerator.LogUnhandledException, consoleBufferTimeout: TimeSpan.FromMilliseconds(500), autoStart: true);
+            var generator = new LinuxContainerEventGenerator(env, consoleWriter);
+
+            // Setup console output that will block until we release the semaphore.
+            var controlledWriter = new ControlledWriter(_consoleOut);
+            await controlledWriter.Semaphore.WaitAsync();
+            Console.SetOut(controlledWriter);
+
+            var timestamp = DateTime.Parse("2023-04-19T14:12:00.0000000Z");
+            generator.LogFunctionTraceEvent(LogLevel.Information, "C37E3412-86D1-4B93-BC5A-A2AE09D26C2D", "TestApp", "TestFunction1", "TestEvent", "TestSource", "These are the details, lots of details", "This is the summary, a great summary", "TestExceptionType", "Test exception message, with details", "E2D5A6ED-4CE3-4CFD-8878-FD4814F0A1F3", "3AD41658-1C4E-4C9D-B0B9-24F2BDAE2829", "F0AAA9AD-C3A6-48B9-A75E-57BB280EBB53", "TestRuntimeSiteName", "TestSlotName", timestamp);
+            generator.LogFunctionTraceEvent(LogLevel.Information, "C37E3412-86D1-4B93-BC5A-A2AE09D26C2D", "TestApp", "TestFunction2", "TestEvent", "TestSource", "These are the details, lots of details", "This is the summary, a great summary", "TestExceptionType", "Test exception message, with details", "E2D5A6ED-4CE3-4CFD-8878-FD4814F0A1F3", "3AD41658-1C4E-4C9D-B0B9-24F2BDAE2829", "F0AAA9AD-C3A6-48B9-A75E-57BB280EBB53", "TestRuntimeSiteName", "TestSlotName", timestamp);
+            await Task.Delay(TimeSpan.FromMilliseconds(20));
+
+            // We haven't released the semaphore yet, so nothing should have been written.
+            Assert.Equal(0, _consoleOut.Length);
+
+            var disposeTask = Task.Run(consoleWriter.Dispose);
+            await Task.Delay(TimeSpan.FromMilliseconds(20));
+
+            // Dispose should be blocked on flushing the buffer, which is blocked on the semaphore.
+            Assert.False(disposeTask.IsCompleted);
+
+            controlledWriter.Semaphore.Release();
+            await disposeTask;
+
+            // after being disposed, console writer should fall back to direct console write, just so that we don't get errors on logging paths
+            generator.LogFunctionTraceEvent(LogLevel.Information, "C37E3412-86D1-4B93-BC5A-A2AE09D26C2D", "TestApp", "TestFunction3", "TestEvent", "TestSource", "These are the details, lots of details", "This is the summary, a great summary", "TestExceptionType", "Test exception message, with details", "E2D5A6ED-4CE3-4CFD-8878-FD4814F0A1F3", "3AD41658-1C4E-4C9D-B0B9-24F2BDAE2829", "F0AAA9AD-C3A6-48B9-A75E-57BB280EBB53", "TestRuntimeSiteName", "TestSlotName", timestamp);
+
+            using var sr = new StreamReader(_consoleOut);
+            _consoleOut.Position = 0;
+            var output = sr.ReadToEnd().Trim().SplitLines();
+
+            Assert.Equal(3, output.Length);
+
+            Assert.Equal($"MS_FUNCTION_LOGS 4,C37E3412-86D1-4B93-BC5A-A2AE09D26C2D,TestApp,TestFunction1,TestEvent,TestSource,\"These are the details, lots of details\",\"This is the summary, a great summary\",{ScriptHost.Version},{timestamp.ToString("O")},TestExceptionType,\"Test exception message, with details\",E2D5A6ED-4CE3-4CFD-8878-FD4814F0A1F3,3AD41658-1C4E-4C9D-B0B9-24F2BDAE2829,F0AAA9AD-C3A6-48B9-A75E-57BB280EBB53,TEST-CONTAINER,test-stamp,test-tenant,TestRuntimeSiteName,TestSlotName", output[0]);
+            Assert.Equal($"MS_FUNCTION_LOGS 4,C37E3412-86D1-4B93-BC5A-A2AE09D26C2D,TestApp,TestFunction2,TestEvent,TestSource,\"These are the details, lots of details\",\"This is the summary, a great summary\",{ScriptHost.Version},{timestamp.ToString("O")},TestExceptionType,\"Test exception message, with details\",E2D5A6ED-4CE3-4CFD-8878-FD4814F0A1F3,3AD41658-1C4E-4C9D-B0B9-24F2BDAE2829,F0AAA9AD-C3A6-48B9-A75E-57BB280EBB53,TEST-CONTAINER,test-stamp,test-tenant,TestRuntimeSiteName,TestSlotName", output[1]);
+            Assert.Equal($"MS_FUNCTION_LOGS 4,C37E3412-86D1-4B93-BC5A-A2AE09D26C2D,TestApp,TestFunction3,TestEvent,TestSource,\"These are the details, lots of details\",\"This is the summary, a great summary\",{ScriptHost.Version},{timestamp.ToString("O")},TestExceptionType,\"Test exception message, with details\",E2D5A6ED-4CE3-4CFD-8878-FD4814F0A1F3,3AD41658-1C4E-4C9D-B0B9-24F2BDAE2829,F0AAA9AD-C3A6-48B9-A75E-57BB280EBB53,TEST-CONTAINER,test-stamp,test-tenant,TestRuntimeSiteName,TestSlotName", output[2]);
+        }
+
+        private class ControlledWriter : StreamWriter
+        {
+            public ControlledWriter(Stream stream) : base(stream)
+            {
+                this.AutoFlush = true;
+            }
+
+            public SemaphoreSlim Semaphore { get; set; } = new SemaphoreSlim(1);
+
+            public override void Write(string value)
+            {
+                Semaphore.Wait();
+                base.Write(value);
+                Semaphore.Release();
+            }
+
+            public override void WriteLine(string value)
+            {
+                Semaphore.Wait();
+                base.WriteLine(value);
+                Semaphore.Release();
+            }
+        }
     }
 }