Add a mechanism for providers to override the scale monitor (#1692)

This allows backends to plug in their own autoscaler for use in elastic premium deployments.

Adds a virtual TryGetScaleMonitor method to DurabilityProvider
If the provider does not implement that method, constructs a NoOpScaleMonitor to avoid exceptions when runtime scale is disabled
Allows the provider to trace scaling decisions in the same way as currently done by Azure Table Scaling monitor

Co-authored-by: Varshi Bachu <[email protected]>

Author: sebastianburckhardt

src/WebJobs.Extensions.DurableTask/AzureStorageDurabilityProvider.cs

@@ -9,10 +9,14 @@
 using DurableTask.AzureStorage;
 using DurableTask.AzureStorage.Tracking;
 using DurableTask.Core;
+using Microsoft.Extensions.Logging;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Converters;
 using Newtonsoft.Json.Linq;
 using Newtonsoft.Json.Serialization;
+#if !FUNCTIONS_V1
+using Microsoft.Azure.WebJobs.Host.Scale;
+#endif
 using AzureStorage = DurableTask.AzureStorage;
 
 namespace Microsoft.Azure.WebJobs.Extensions.DurableTask
@@ -25,11 +29,13 @@ internal class AzureStorageDurabilityProvider : DurabilityProvider
         private readonly AzureStorageOrchestrationService serviceClient;
         private readonly string connectionName;
         private readonly JObject storageOptionsJson;
+        private readonly ILogger logger;
 
         public AzureStorageDurabilityProvider(
             AzureStorageOrchestrationService service,
             string connectionName,
-            AzureStorageOptions options)
+            AzureStorageOptions options,
+            ILogger logger)
             : base("Azure Storage", service, service, connectionName)
         {
             this.serviceClient = service;
@@ -41,6 +47,7 @@ public AzureStorageDurabilityProvider(
                     Converters = { new StringEnumConverter() },
                     ContractResolver = new CamelCasePropertyNamesContractResolver(),
                 });
+            this.logger = logger;
         }
 
         public override bool SupportsEntities => true;
@@ -180,5 +187,24 @@ internal static OrchestrationInstanceStatusQueryCondition ConvertWebjobsDurableC
                 FetchInput = condition.ShowInput,
             };
         }
+
+#if !FUNCTIONS_V1
+        /// <inheritdoc/>
+        public override bool TryGetScaleMonitor(
+            string functionId,
+            string functionName,
+            string hubName,
+            string storageConnectionString,
+            out IScaleMonitor scaleMonitor)
+        {
+            scaleMonitor = new DurableTaskScaleMonitor(
+                functionId,
+                functionName,
+                hubName,
+                storageConnectionString,
+                this.logger);
+            return true;
+        }
+#endif
     }
 }

src/WebJobs.Extensions.DurableTask/AzureStorageDurabilityProviderFactory.cs

@@ -11,6 +11,7 @@ namespace Microsoft.Azure.WebJobs.Extensions.DurableTask
 {
     internal class AzureStorageDurabilityProviderFactory : IDurabilityProviderFactory
     {
+        private const string LoggerName = "Host.Triggers.DurableTask.AzureStorage";
         internal const string ProviderName = "AzureStorage";
 
         private readonly DurableTaskOptions options;
@@ -95,10 +96,12 @@ public virtual DurabilityProvider GetDurabilityProvider()
             if (this.defaultStorageProvider == null)
             {
                 var defaultService = new AzureStorageOrchestrationService(this.defaultSettings);
+                ILogger logger = this.loggerFactory.CreateLogger(LoggerName);
                 this.defaultStorageProvider = new AzureStorageDurabilityProvider(
                     defaultService,
                     this.defaultConnectionName,
-                    this.azureStorageOptions);
+                    this.azureStorageOptions,
+                    logger);
             }
 
             return this.defaultStorageProvider;
@@ -126,10 +129,12 @@ private AzureStorageDurabilityProvider GetAzureStorageStorageProvider(DurableCli
             }
             else
             {
+                ILogger logger = this.loggerFactory.CreateLogger(LoggerName);
                 innerClient = new AzureStorageDurabilityProvider(
                     new AzureStorageOrchestrationService(settings),
                     connectionName,
-                    this.azureStorageOptions);
+                    this.azureStorageOptions,
+                    logger);
             }
 
             return innerClient;

src/WebJobs.Extensions.DurableTask/DurabilityProvider.cs

@@ -6,8 +6,12 @@
 using System.Threading;
 using System.Threading.Tasks;
 using DurableTask.Core;
+using Microsoft.Extensions.Logging;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
+#if !FUNCTIONS_V1
+using Microsoft.Azure.WebJobs.Host.Scale;
+#endif
 
 namespace Microsoft.Azure.WebJobs.Extensions.DurableTask
 {
@@ -438,5 +442,27 @@ internal virtual bool ConnectionNameMatches(DurabilityProvider durabilityProvide
         {
             return this.ConnectionName.Equals(durabilityProvider.ConnectionName);
         }
+
+#if !FUNCTIONS_V1
+        /// <summary>
+        /// Tries to obtain a scale monitor for autoscaling.
+        /// </summary>
+        /// <param name="functionId">Function id.</param>
+        /// <param name="functionName">Function name.</param>
+        /// <param name="hubName">Task hub name.</param>
+        /// <param name="storageConnectionString">Storage account connection string, used for Azure Storage provider.</param>
+        /// <param name="scaleMonitor">The scale monitor.</param>
+        /// <returns>True if autoscaling is supported, false otherwise.</returns>
+        public virtual bool TryGetScaleMonitor(
+            string functionId,
+            string functionName,
+            string hubName,
+            string storageConnectionString,
+            out IScaleMonitor scaleMonitor)
+        {
+            scaleMonitor = null;
+            return false;
+        }
+#endif
     }
 }

src/WebJobs.Extensions.DurableTask/DurableTaskExtension.cs

@@ -20,6 +20,7 @@
 using Microsoft.Azure.WebJobs.Description;
 #if !FUNCTIONS_V1
 using Microsoft.Azure.WebJobs.Extensions.DurableTask.Correlation;
+using Microsoft.Azure.WebJobs.Host.Scale;
 #endif
 using Microsoft.Azure.WebJobs.Extensions.DurableTask.Listener;
 using Microsoft.Azure.WebJobs.Host;
@@ -1311,6 +1312,59 @@ internal static void TagActivityWithOrchestrationStatus(OrchestrationRuntimeStat
                 activity.AddTag("DurableFunctionsRuntimeStatus", statusStr);
             }
         }
+
+        internal IScaleMonitor GetScaleMonitor(string functionId, FunctionName functionName, string storageConnectionString)
+        {
+            if (this.defaultDurabilityProvider.TryGetScaleMonitor(
+                    functionId,
+                    functionName.Name,
+                    this.Options.HubName,
+                    storageConnectionString,
+                    out IScaleMonitor scaleMonitor))
+            {
+                return scaleMonitor;
+            }
+            else
+            {
+                // the durability provider does not support runtime scaling.
+                // Create an empty scale monitor to avoid exceptions (unless runtime scaling is actually turned on).
+                return new NoOpScaleMonitor($"{functionId}-DurableTaskTrigger-{this.Options.HubName}".ToLower());
+            }
+        }
+
+        /// <summary>
+        /// A placeholder scale monitor, can be used by durability providers that do not support runtime scaling.
+        /// This is required to allow operation of those providers even if runtime scaling is turned off
+        /// see discussion https://github.com/Azure/azure-functions-durable-extension/pull/1009/files#r341767018.
+        /// </summary>
+        private sealed class NoOpScaleMonitor : IScaleMonitor
+        {
+            /// <summary>
+            /// Construct a placeholder scale monitor.
+            /// </summary>
+            /// <param name="name">A descriptive name.</param>
+            public NoOpScaleMonitor(string name)
+            {
+                this.Descriptor = new ScaleMonitorDescriptor(name);
+            }
+
+            /// <summary>
+            /// A descriptive name.
+            /// </summary>
+            public ScaleMonitorDescriptor Descriptor { get; private set; }
+
+            /// <inheritdoc/>
+            Task<ScaleMetrics> IScaleMonitor.GetMetricsAsync()
+            {
+                throw new InvalidOperationException("The current DurableTask backend configuration does not support runtime scaling");
+            }
+
+            /// <inheritdoc/>
+            ScaleStatus IScaleMonitor.GetScaleStatus(ScaleStatusContext context)
+            {
+                throw new InvalidOperationException("The current DurableTask backend configuration does not support runtime scaling");
+            }
+        }
 #endif
     }
 }

src/WebJobs.Extensions.DurableTask/Listener/DurableTaskListener.cs

@@ -28,7 +28,7 @@ internal sealed class DurableTaskListener : IListener
         private readonly FunctionType functionType;
         private readonly string storageConnectionString;
 #if !FUNCTIONS_V1
-        private readonly Lazy<DurableTaskScaleMonitor> scaleMonitor;
+        private readonly Lazy<IScaleMonitor> scaleMonitor;
 #endif
 
         public DurableTaskListener(
@@ -52,13 +52,11 @@ public DurableTaskListener(
             this.functionType = functionType;
             this.storageConnectionString = storageConnectionString;
 #if !FUNCTIONS_V1
-            this.scaleMonitor = new Lazy<DurableTaskScaleMonitor>(() =>
-                new DurableTaskScaleMonitor(
+            this.scaleMonitor = new Lazy<IScaleMonitor>(() =>
+                this.config.GetScaleMonitor(
                     this.functionId,
                     this.functionName,
-                    this.config.Options.HubName,
-                    this.storageConnectionString,
-                    this.config.TraceHelper));
+                    this.storageConnectionString));
 #endif
         }
 

src/WebJobs.Extensions.DurableTask/Listener/DurableTaskScaleMonitor.cs

@@ -12,6 +12,7 @@
 using DurableTask.AzureStorage.Monitoring;
 using Dynamitey.DynamicObjects;
 using Microsoft.Azure.WebJobs.Host.Scale;
+using Microsoft.Extensions.Logging;
 using Microsoft.WindowsAzure.Storage;
 using Newtonsoft.Json;
 
@@ -20,28 +21,28 @@ namespace Microsoft.Azure.WebJobs.Extensions.DurableTask
     internal sealed class DurableTaskScaleMonitor : IScaleMonitor<DurableTaskTriggerMetrics>
     {
         private readonly string functionId;
-        private readonly FunctionName functionName;
+        private readonly string functionName;
         private readonly string hubName;
         private readonly string storageConnectionString;
-        private readonly EndToEndTraceHelper traceHelper;
         private readonly ScaleMonitorDescriptor scaleMonitorDescriptor;
+        private readonly ILogger logger;
 
         private DisconnectedPerformanceMonitor performanceMonitor;
 
         public DurableTaskScaleMonitor(
             string functionId,
-            FunctionName functionName,
+            string functionName,
             string hubName,
             string storageConnectionString,
-            EndToEndTraceHelper traceHelper,
+            ILogger logger,
             DisconnectedPerformanceMonitor performanceMonitor = null)
         {
-            this.functionId = functionId;
-            this.functionName = functionName;
-            this.hubName = hubName;
+            this.functionId = functionId;
+            this.functionName = functionName;
+            this.hubName = hubName;
             this.storageConnectionString = storageConnectionString;
+            this.logger = logger;
             this.performanceMonitor = performanceMonitor;
-            this.traceHelper = traceHelper;
             this.scaleMonitorDescriptor = new ScaleMonitorDescriptor($"{this.functionId}-DurableTaskTrigger-{this.hubName}".ToLower());
         }
 
@@ -86,7 +87,7 @@ public async Task<DurableTaskTriggerMetrics> GetMetricsAsync()
             }
             catch (StorageException e)
             {
-                this.traceHelper.ExtensionWarningEvent(this.hubName, this.functionName.Name, string.Empty, e.ToString());
+                this.logger.LogWarning("{details}. Function: {functionName}. HubName: {hubName}.", e.ToString(), this.functionName, this.hubName);
             }
 
             if (heartbeat != null)
@@ -114,7 +115,7 @@ public ScaleStatus GetScaleStatus(ScaleStatusContext<DurableTaskTriggerMetrics>
             return this.GetScaleStatusCore(context.WorkerCount, context.Metrics?.ToArray());
         }
 
-        private ScaleStatus GetScaleStatusCore(int workerCount, DurableTaskTriggerMetrics[] metrics)
+        private ScaleStatus GetScaleStatusCore(int workerCount, DurableTaskTriggerMetrics[] metrics)
         {
             var scaleStatus = new ScaleStatus() { Vote = ScaleVote.None };
             if (metrics == null)
@@ -173,12 +174,13 @@ private ScaleStatus GetScaleStatusCore(int workerCount, DurableTaskTriggerMetric
                     break;
             }
 
-            this.traceHelper.ExtensionInformationalEvent(
-                                    this.hubName,
-                                    string.Empty,
-                                    this.functionName.Name,
-                                    $"Durable Functions Trigger Scale Decision: {scaleStatus.Vote.ToString()}, Reason: {scaleRecommendation?.Reason}",
-                                    writeToUserLogs: writeToUserLogs);
+            if (writeToUserLogs)
+            {
+                this.logger.LogInformation(
+                    $"Durable Functions Trigger Scale Decision: {scaleStatus.Vote.ToString()}, Reason: {scaleRecommendation?.Reason}",
+                    this.hubName,
+                    this.functionName);
+            }
 
             return scaleStatus;
         }