Merge pull request #1530 from Azure/dev

Promote to master for 2.3.1 release - 3rd instance

Author: davidmrdavid

src/WebJobs.Extensions.DurableTask.Analyzers/SyntaxNodeUtils.cs

@@ -57,7 +57,7 @@ public static bool TryGetISymbol(SemanticModel semanticModel, SyntaxNode node, o
         {
             if (node != null && TryGetSemanticModelForSyntaxTree(semanticModel, node, out SemanticModel newModel))
             {
-                symbol = semanticModel.GetSymbolInfo(node).Symbol;
+                symbol = newModel.GetSymbolInfo(node).Symbol;
                 return symbol != null;
             }
 

src/WebJobs.Extensions.DurableTask.Analyzers/WebJobs.Extensions.DurableTask.Analyzers.csproj

@@ -8,7 +8,7 @@
 
   <PropertyGroup>
     <PackageId>Microsoft.Azure.WebJobs.Extensions.DurableTask.Analyzers</PackageId>
-    <PackageVersion>0.3.1</PackageVersion>
+    <PackageVersion>0.3.2</PackageVersion>
     <Authors>Microsoft</Authors>
     <PackageLicenseUrl>https://go.microsoft.com/fwlink/?linkid=2028464</PackageLicenseUrl>
     <PackageProjectUrl>https://github.com/Azure/azure-functions-durable-extension</PackageProjectUrl>

src/WebJobs.Extensions.DurableTask/DurableTaskExtension.cs

@@ -221,8 +221,11 @@ private MessagePayloadDataConverter CreateErrorDataConverter(IErrorSerializerSet
         /// <param name="context">Extension context provided by WebJobs.</param>
         void IExtensionConfigProvider.Initialize(ExtensionConfigContext context)
         {
+#if !FUNCTIONS_V1
+            // .NET461 is not supported in linux, so this is conditionally compiled
             // We initialize linux logging early on in case any initialization steps below were to trigger a log event.
             this.InitializeLinuxLogging();
+#endif
 
             ConfigureLoaderHooks();
 
@@ -342,15 +345,21 @@ private void InitializeLinuxLogging()
                 // The logging service for linux works by capturing EventSource messages,
                 // which our linux platform does not recognize, and logging them via a
                 // different strategy such as writing to console or to a file.
-                this.eventSourceListener = new EventSourceListener(linuxLogger);
+
+                // Since our logging payload can be quite large, linux telemetry by default
+                // disables verbose-level telemetry to avoid a performance hit.
+                bool enableVerbose = this.Options.Tracing.AllowVerboseLinuxTelemetry;
+                this.eventSourceListener = new EventSourceListener(linuxLogger, enableVerbose);
             }
         }
 
         /// <inheritdoc />
         public void Dispose()
         {
             // Not flushing the linux logger may lead to lost logs
-            Serilog.Log.CloseAndFlush();
+            // 40 seconds timeout because we write normally every 30 seconds, so we're just
+            // adding an extra 10 seconds to flush.
+            LinuxAppServiceLogger.Logger?.Stop(TimeSpan.FromSeconds(40));
             this.HttpApiHandler?.Dispose();
             this.eventSourceListener?.Dispose();
         }

src/WebJobs.Extensions.DurableTask/EndToEndTraceHelper.cs

@@ -14,15 +14,18 @@ namespace Microsoft.Azure.WebJobs.Extensions.DurableTask
     internal class EventSourceListener : EventListener
     {
         private readonly LinuxAppServiceLogger logger;
+        private readonly bool disableVerbose;
 
         /// <summary>
         /// Create an EventSourceListener to capture and log Durable EventSource
         /// data in Linux.
         /// </summary>
         /// <param name="logger">A LinuxAppService logger configured for the current linux host.</param>
-        public EventSourceListener(LinuxAppServiceLogger logger)
+        /// <param name="enableVerbose">If true, durableTask.Core verbose logs are enabled. The opposite if false.</param>
+        public EventSourceListener(LinuxAppServiceLogger logger, bool enableVerbose)
         {
             this.logger = logger;
+            this.disableVerbose = !enableVerbose; // We set the opposite to simply logic later
         }
 
         /// <summary>
@@ -52,7 +55,13 @@ protected override void OnEventSourceCreated(EventSource eventSource)
         /// <param name="eventData">The EventSource event data, for logging.</param>
         protected override void OnEventWritten(EventWrittenEventArgs eventData)
         {
-            this.logger.Log(eventData);
+            // When disabling verbose logs, we skip Verbose DurableTask-Core telemetry
+            if (!(this.disableVerbose
+               && eventData.EventSource.Name == "DurableTask-Core"
+               && eventData.Level == EventLevel.Verbose))
+            {
+                this.logger.Log(eventData);
+            }
         }
     }
 }

src/WebJobs.Extensions.DurableTask/EventSourceListener.cs

@@ -0,0 +1,181 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in the project root for license information.
+
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+#if !FUNCTIONS_V1
+using Mono.Unix.Native;
+#endif
+
+namespace Microsoft.Azure.WebJobs.Extensions.DurableTask
+{
+    /// <summary>
+    /// The File logger for linux dedicated. Manages file rolling and is concurrency-safe.
+    /// This is copied over from the azure-funtions-host codebase here:
+    /// https://github.com/Azure/azure-functions-host/blob/35cf323fa3464a08b410a518bcab006e801301fe/src/WebJobs.Script.WebHost/Diagnostics/LinuxAppServiceFileLogger.cs
+    /// We have modified their implementation to utilize syscall.rename instead of File.Move during file rolling.
+    /// This change is necessary for older versions of fluent-bit, our logging infrastructure in linux dedicated, to properly deal with logfile archiving.
+    /// </summary>
+    public class LinuxAppServiceFileLogger
+    {
+        private readonly string logFileName;
+        private readonly string logFileDirectory;
+        private readonly string logFilePath;
+        private readonly string archiveFilePath;
+        private readonly BlockingCollection<string> buffer;
+        private readonly List<string> currentBatch;
+        private readonly CancellationTokenSource cancellationTokenSource;
+        private Task outputTask;
+
+        /// <summary>
+        /// The File logger for linux dedicated. Manages file rolling and is concurrency-safe.
+        /// </summary>
+        /// <param name="logFileName">Name of target logfile.</param>
+        /// <param name="logFileDirectory">Directory of target logfile.</param>
+        /// <param name="startOnCreate">Whether or not to start monitoring the write buffer at initialization time.</param>
+        public LinuxAppServiceFileLogger(string logFileName, string logFileDirectory, bool startOnCreate = true)
+        {
+            this.logFileName = logFileName;
+            this.logFileDirectory = logFileDirectory;
+            this.logFilePath = Path.Combine(this.logFileDirectory, this.logFileName);
+            this.archiveFilePath = this.logFilePath + "1";
+            this.buffer = new BlockingCollection<string>(new ConcurrentQueue<string>());
+            this.currentBatch = new List<string>();
+            this.cancellationTokenSource = new CancellationTokenSource();
+
+            if (startOnCreate)
+            {
+                this.Start();
+            }
+        }
+
+        // Maximum size of individual log file in MB
+        private int MaxFileSizeMb { get; set; } = 10;
+
+        // Maximum time between successive flushes (seconds)
+        private int FlushFrequencySeconds { get; set; } = 30;
+
+        /// <summary>
+        /// Log a string.
+        /// </summary>
+        /// <param name="message">Message to log.</param>
+        public virtual void Log(string message)
+        {
+            try
+            {
+                this.buffer.Add(message);
+            }
+            catch (Exception)
+            {
+                // ignored
+            }
+        }
+
+        private void Start()
+        {
+            if (this.outputTask == null)
+            {
+                this.outputTask = Task.Factory.StartNew(this.ProcessLogQueue, null, TaskCreationOptions.LongRunning);
+            }
+        }
+
+        /// <summary>
+        /// Flushes the write buffer, stops writing to logfile afterwards.
+        /// </summary>
+        /// <param name="timeSpan">Timeout in milliseconds for flushing task.</param>
+        public void Stop(TimeSpan timeSpan)
+        {
+            this.cancellationTokenSource.Cancel();
+
+            try
+            {
+                this.outputTask?.Wait(timeSpan);
+            }
+            catch (Exception)
+            {
+                // ignored
+            }
+        }
+
+        private async Task ProcessLogQueue(object state)
+        {
+            while (!this.cancellationTokenSource.IsCancellationRequested)
+            {
+                await this.InternalProcessLogQueue();
+                await Task.Delay(TimeSpan.FromSeconds(this.FlushFrequencySeconds), this.cancellationTokenSource.Token).ContinueWith(task => { });
+            }
+
+            await this.InternalProcessLogQueue();
+
+            // ReSharper disable once FunctionNeverReturns
+        }
+
+        // internal for unittests (in func host)
+        internal async Task InternalProcessLogQueue()
+        {
+            string currentMessage;
+            while (this.buffer.TryTake(out currentMessage))
+            {
+                this.currentBatch.Add(currentMessage);
+            }
+
+            if (this.currentBatch.Any())
+            {
+                try
+                {
+                    await this.WriteLogs(this.currentBatch);
+                }
+                catch (Exception)
+                {
+                    // Ignored
+                }
+
+                this.currentBatch.Clear();
+            }
+        }
+
+        private async Task WriteLogs(IEnumerable<string> currentBatch)
+        {
+            // If the directory already exists, this does nothing
+            Directory.CreateDirectory(this.logFileDirectory);
+
+            var fileInfo = new FileInfo(this.logFilePath);
+            if (fileInfo.Exists)
+            {
+                if (fileInfo.Length / (1024 * 1024) >= this.MaxFileSizeMb)
+                {
+                    this.RollFiles();
+                }
+            }
+
+            await this.AppendLogs(this.logFilePath, currentBatch);
+        }
+
+        private async Task AppendLogs(string filePath, IEnumerable<string> logs)
+        {
+            using (var streamWriter = File.AppendText(filePath))
+            {
+                foreach (var log in logs)
+                {
+                    await streamWriter.WriteLineAsync(log);
+                }
+            }
+        }
+
+        private void RollFiles()
+        {
+            // Rename current file to older file.
+
+#if !FUNCTIONS_V1
+            Syscall.rename(this.logFilePath, this.archiveFilePath);
+#endif
+
+        }
+    }
+}