Add target-based scaling support for Azure Storage

src/WebJobs.Extensions.DurableTask/AzureStorageDurabilityProvider.cs

@@ -10,6 +10,7 @@
 using DurableTask.AzureStorage.Tracking;
 using DurableTask.Core;
 using Microsoft.Extensions.Logging;
+using Microsoft.WindowsAzure.Storage;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Converters;
 using Newtonsoft.Json.Linq;
@@ -200,6 +201,16 @@ internal static OrchestrationInstanceStatusQueryCondition ConvertWebjobsDurableC
         }
 
 #if !FUNCTIONS_V1
+
+        internal DurableTaskMetricsProvider GetMetricsProvider(
+            string functionName,
+            string hubName,
+            CloudStorageAccount storageAccount,
+            ILogger logger)
+        {
+            return new DurableTaskMetricsProvider(functionName, hubName, logger, performanceMonitor: null, storageAccount);
+        }
+
         /// <inheritdoc/>
         public override bool TryGetScaleMonitor(
             string functionId,
@@ -208,12 +219,30 @@ public override bool TryGetScaleMonitor(
             string connectionName,
             out IScaleMonitor scaleMonitor)
         {
+            CloudStorageAccount storageAccount = this.storageAccountProvider.GetStorageAccountDetails(connectionName).ToCloudStorageAccount();
+            DurableTaskMetricsProvider metricsProvider = this.GetMetricsProvider(functionName, hubName, storageAccount, this.logger);
             scaleMonitor = new DurableTaskScaleMonitor(
                 functionId,
                 functionName,
                 hubName,
-                this.storageAccountProvider.GetStorageAccountDetails(connectionName).ToCloudStorageAccount(),
-                this.logger);
+                storageAccount,
+                this.logger,
+                metricsProvider);
+            return true;
+        }
+
+#endif
+#if FUNCTIONS_V3_OR_GREATER
+        public override bool TryGetTargetScaler(
+            string functionId,
+            string functionName,
+            string hubName,
+            string connectionName,
+            out ITargetScaler targetScaler)
+        {
+            CloudStorageAccount storageAccount = this.storageAccountProvider.GetStorageAccountDetails(connectionName).ToCloudStorageAccount();
+            DurableTaskMetricsProvider metricsProvider = this.GetMetricsProvider(functionName, hubName, storageAccount, this.logger);
+            targetScaler = new DurableTaskTargetScaler(functionId, metricsProvider, this, this.logger);
             return true;
         }
 #endif

src/WebJobs.Extensions.DurableTask/DurabilityProvider.cs

@@ -565,5 +565,27 @@ public virtual bool TryGetScaleMonitor(
             return false;
         }
 #endif
+
+#if FUNCTIONS_V3_OR_GREATER
+        /// <summary>
+        /// Tries to obtain a scaler for target based scaling.
+        /// </summary>
+        /// <param name="functionId">Function id.</param>
+        /// <param name="functionName">Function name.</param>
+        /// <param name="hubName">Task hub name.</param>
+        /// <param name="connectionName">The name of the storage-specific connection settings.</param>
+        /// <param name="targetScaler">The target-based scaler.</param>
+        /// <returns>True if target-based scaling is supported, false otherwise.</returns>
+        public virtual bool TryGetTargetScaler(
+            string functionId,
+            string functionName,
+            string hubName,
+            string connectionName,
+            out ITargetScaler targetScaler)
+        {
+            targetScaler = null;
+            return false;
+        }
+#endif
     }
 }

src/WebJobs.Extensions.DurableTask/DurableTaskExtension.cs

@@ -1589,10 +1589,52 @@ internal IScaleMonitor GetScaleMonitor(string functionId, FunctionName functionN
             {
                 // 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());
+                return new NoOpScaleMonitor($"{functionId}-DurableTaskTrigger-{this.Options.HubName}".ToLower(), functionId);
             }
         }
+#endif
+#if FUNCTIONS_V3_OR_GREATER
+
+        internal ITargetScaler GetTargetScaler(string functionId, FunctionName functionName, string connectionName)
+        {
+            if (this.defaultDurabilityProvider.TryGetTargetScaler(
+                    functionId,
+                    functionName.Name,
+                    this.Options.HubName,
+                    connectionName,
+                    out ITargetScaler targetScaler))
+            {
+                return targetScaler;
+            }
+            else
+            {
+                // the durability provider does not support target-based scaling.
+                // Create an empty target scaler to avoid exceptions (unless target-based scaling is actually turned on).
+                return new NoOpTargetScaler(functionId);
+            }
+        }
+
+        private sealed class NoOpTargetScaler : ITargetScaler
+        {
+            /// <summary>
+            /// Construct a placeholder target scaler.
+            /// </summary>
+            /// <param name="functionId">The function ID.</param>
+            public NoOpTargetScaler(string functionId)
+            {
+                this.TargetScalerDescriptor = new TargetScalerDescriptor(functionId);
+            }
+
+            public TargetScalerDescriptor TargetScalerDescriptor { get; }
+
+            public Task<TargetScalerResult> GetScaleResultAsync(TargetScalerContext context)
+            {
+                throw new NotImplementedException("The current DurableTask backend configuration does not support target-based scaling");
+            }
+        }
+#endif
 
+#if !FUNCTIONS_V1
         /// <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
@@ -1604,9 +1646,16 @@ private sealed class NoOpScaleMonitor : IScaleMonitor
             /// Construct a placeholder scale monitor.
             /// </summary>
             /// <param name="name">A descriptive name.</param>
-            public NoOpScaleMonitor(string name)
+            /// <param name="functionId">The function ID.</param>
+            public NoOpScaleMonitor(string name, string functionId)
             {
+#if FUNCTIONS_V3_OR_GREATER
+                this.Descriptor = new ScaleMonitorDescriptor(name, functionId);
+#else
+#pragma warning disable CS0618 // Type or member is obsolete
                 this.Descriptor = new ScaleMonitorDescriptor(name);
+#pragma warning restore CS0618 // Type or member is obsolete
+#endif
             }
 
             /// <summary>

src/WebJobs.Extensions.DurableTask/DurableTaskJobHostConfigurationExtensions.cs

@@ -2,8 +2,6 @@
 // Licensed under the MIT License. See LICENSE in the project root for license information.
 
 using System;
-using System.Net.Http;
-using System.Threading;
 #if !FUNCTIONS_V1
 using Microsoft.Azure.WebJobs.Extensions.DurableTask.Auth;
 using Microsoft.Azure.WebJobs.Extensions.DurableTask.ContextImplementations;

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

@@ -2,20 +2,18 @@
 // Licensed under the MIT License. See LICENSE in the project root for license information.
 
 using System;
-using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
-using DurableTask.AzureStorage.Monitoring;
-using Microsoft.Azure.WebJobs.Host.Executors;
 using Microsoft.Azure.WebJobs.Host.Listeners;
 #if !FUNCTIONS_V1
 using Microsoft.Azure.WebJobs.Host.Scale;
 #endif
 
 namespace Microsoft.Azure.WebJobs.Extensions.DurableTask
 {
-#if !FUNCTIONS_V1
+#if FUNCTIONS_V3_OR_GREATER
+    internal sealed class DurableTaskListener : IListener, IScaleMonitorProvider, ITargetScalerProvider
+#elif FUNCTIONS_V2_OR_GREATER
     internal sealed class DurableTaskListener : IListener, IScaleMonitorProvider
 #else
     internal sealed class DurableTaskListener : IListener
@@ -26,10 +24,15 @@ internal sealed class DurableTaskListener : IListener
         private readonly FunctionName functionName;
         private readonly FunctionType functionType;
         private readonly string connectionName;
+
 #if !FUNCTIONS_V1
         private readonly Lazy<IScaleMonitor> scaleMonitor;
 #endif
 
+#if FUNCTIONS_V3_OR_GREATER
+        private readonly Lazy<ITargetScaler> targetScaler;
+#endif
+
         public DurableTaskListener(
             DurableTaskExtension config,
             string functionId,
@@ -49,11 +52,20 @@ public DurableTaskListener(
             this.functionType = functionType;
             this.connectionName = connectionName;
 #if !FUNCTIONS_V1
+
             this.scaleMonitor = new Lazy<IScaleMonitor>(() =>
                 this.config.GetScaleMonitor(
                     this.functionId,
                     this.functionName,
                     this.connectionName));
+
+#endif
+#if FUNCTIONS_V3_OR_GREATER
+            this.targetScaler = new Lazy<ITargetScaler>(() =>
+                this.config.GetTargetScaler(
+                    this.functionId,
+                    this.functionName,
+                    this.connectionName));
 #endif
         }
 
@@ -98,5 +110,12 @@ public IScaleMonitor GetMonitor()
             return this.scaleMonitor.Value;
         }
 #endif
+
+#if FUNCTIONS_V3_OR_GREATER
+        public ITargetScaler GetTargetScaler()
+        {
+            return this.targetScaler.Value;
+        }
+#endif
     }
 }

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

@@ -0,0 +1,79 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in the project root for license information.
+
+#if !FUNCTIONS_V1
+using System;
+using System.Threading.Tasks;
+using DurableTask.AzureStorage.Monitoring;
+using Microsoft.Extensions.Logging;
+using Microsoft.WindowsAzure.Storage;
+using Newtonsoft.Json;
+
+namespace Microsoft.Azure.WebJobs.Extensions.DurableTask
+{
+    internal class DurableTaskMetricsProvider
+    {
+        private readonly string functionName;
+        private readonly string hubName;
+        private readonly ILogger logger;
+        private readonly CloudStorageAccount storageAccount;
+
+        private DisconnectedPerformanceMonitor performanceMonitor;
+
+        public DurableTaskMetricsProvider(string functionName, string hubName, ILogger logger, DisconnectedPerformanceMonitor performanceMonitor, CloudStorageAccount storageAccount)
+        {
+            this.functionName = functionName;
+            this.hubName = hubName;
+            this.logger = logger;
+            this.performanceMonitor = performanceMonitor;
+            this.storageAccount = storageAccount;
+        }
+
+        public virtual async Task<DurableTaskTriggerMetrics> GetMetricsAsync()
+        {
+            DurableTaskTriggerMetrics metrics = new DurableTaskTriggerMetrics();
+
+            // Durable stores its own metrics, so we just collect them here
+            PerformanceHeartbeat heartbeat = null;
+            try
+            {
+                DisconnectedPerformanceMonitor performanceMonitor = this.GetPerformanceMonitor();
+                heartbeat = await performanceMonitor.PulseAsync();
+            }
+            catch (StorageException e)
+            {
+                this.logger.LogWarning("{details}. Function: {functionName}. HubName: {hubName}.", e.ToString(), this.functionName, this.hubName);
+            }
+
+            if (heartbeat != null)
+            {
+                metrics.PartitionCount = heartbeat.PartitionCount;
+                metrics.ControlQueueLengths = JsonConvert.SerializeObject(heartbeat.ControlQueueLengths);
+                metrics.ControlQueueLatencies = JsonConvert.SerializeObject(heartbeat.ControlQueueLatencies);
+                metrics.WorkItemQueueLength = heartbeat.WorkItemQueueLength;
+                if (heartbeat.WorkItemQueueLatency > TimeSpan.Zero)
+                {
+                    metrics.WorkItemQueueLatency = heartbeat.WorkItemQueueLatency.ToString();
+                }
+            }
+
+            return metrics;
+        }
+
+        internal DisconnectedPerformanceMonitor GetPerformanceMonitor()
+        {
+            if (this.performanceMonitor == null)
+            {
+                if (this.storageAccount == null)
+                {
+                    throw new ArgumentNullException(nameof(this.storageAccount));
+                }
+
+                this.performanceMonitor = new DisconnectedPerformanceMonitor(this.storageAccount, this.hubName);
+            }
+
+            return this.performanceMonitor;
+        }
+    }
+}
+#endif