Add target based scaling support for MSSQL

src/DurableTask.SqlServer.AzureFunctions/DurableTask.SqlServer.AzureFunctions.csproj

@@ -4,7 +4,7 @@
   <Import Project="../common.props" />
 
   <PropertyGroup>
-    <TargetFrameworks>netstandard2.0</TargetFrameworks>
+    <TargetFrameworks>netstandard2.0;netcoreapp3.1</TargetFrameworks>
   </PropertyGroup>
 
   <!-- NuGet package settings -->

src/DurableTask.SqlServer.AzureFunctions/SqlDurabilityProvider.cs

@@ -197,8 +197,24 @@ public override bool TryGetScaleMonitor(
             string storageConnectionString,
             out IScaleMonitor scaleMonitor)
         {
-            scaleMonitor = this.scaleMonitor ??= new SqlScaleMonitor(this.service, hubName);
+            SqlMetricsProvider sqlMetricsProvider = new SqlMetricsProvider(this.service);
+            scaleMonitor = this.scaleMonitor ??= new SqlScaleMonitor(this.service, hubName, sqlMetricsProvider);
             return true;
         }
+
+#if NETCOREAPP
+        public override bool TryGetTargetScaler(
+            string functionId,
+            string functionName,
+            string hubName,
+            string connectionName,
+            out ITargetScaler targetScaler)
+        {
+            SqlMetricsProvider sqlMetricsProvider = new SqlMetricsProvider(this.service);
+            targetScaler = new SqlTargetScaler(functionId, sqlMetricsProvider);
+
+            return true;
+        }
+#endif
     }
 }

src/DurableTask.SqlServer.AzureFunctions/SqlMetricsProvider.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace DurableTask.SqlServer.AzureFunctions
+{
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    public class SqlMetricsProvider
+    {
+        readonly SqlOrchestrationService service;
+
+        public SqlMetricsProvider(SqlOrchestrationService service, int? previousWorkerCount = null)
+        {
+            this.service = service;
+        }
+
+        public virtual async Task<SqlScaleMetric> GetMetricsAsync(int? previousWorkerCount = null)
+        {
+            // GetRecommendedReplicaCountAsync will write a trace if the recommendation results
+            // in a worker count that is different from the worker count we pass in as an argument.
+            int recommendedReplicaCount = await this.service.GetRecommendedReplicaCountAsync(
+                previousWorkerCount,
+                CancellationToken.None);
+
+            return new SqlScaleMetric { RecommendedReplicaCount = recommendedReplicaCount };
+        }
+    }
+}

src/DurableTask.SqlServer.AzureFunctions/SqlScaleMetric.cs

@@ -5,7 +5,7 @@ namespace DurableTask.SqlServer.AzureFunctions
 {
     using Microsoft.Azure.WebJobs.Host.Scale;
 
-    class SqlScaleMetric : ScaleMetrics
+    public class SqlScaleMetric : ScaleMetrics
     {
         public int RecommendedReplicaCount { get; set; }
     }

src/DurableTask.SqlServer.AzureFunctions/SqlScaleMonitor.cs

@@ -20,13 +20,15 @@ class SqlScaleMonitor : IScaleMonitor<SqlScaleMetric>
         static readonly ScaleStatus ScaleOutVote = new ScaleStatus { Vote = ScaleVote.ScaleOut };
 
         readonly SqlOrchestrationService service;
+        readonly SqlMetricsProvider metricsProvider;
 
         int? previousWorkerCount = -1;
 
-        public SqlScaleMonitor(SqlOrchestrationService service, string taskHubName)
+        public SqlScaleMonitor(SqlOrchestrationService service, string taskHubName, SqlMetricsProvider sqlMetricsProvider)
         {
             this.service = service ?? throw new ArgumentNullException(nameof(service));
             this.Descriptor = new ScaleMonitorDescriptor($"DurableTask-SqlServer:{taskHubName ?? "default"}");
+            this.metricsProvider = sqlMetricsProvider ?? throw new ArgumentNullException(nameof(sqlMetricsProvider));
         }
 
         /// <inheritdoc />
@@ -38,13 +40,7 @@ public SqlScaleMonitor(SqlOrchestrationService service, string taskHubName)
         /// <inheritdoc />
         public async Task<SqlScaleMetric> GetMetricsAsync()
         {
-            // GetRecommendedReplicaCountAsync will write a trace if the recommendation results
-            // in a worker count that is different from the worker count we pass in as an argument.
-            int recommendedReplicaCount = await this.service.GetRecommendedReplicaCountAsync(
-                this.previousWorkerCount,
-                CancellationToken.None);
-
-            return new SqlScaleMetric { RecommendedReplicaCount = recommendedReplicaCount };
+            return await this.metricsProvider.GetMetricsAsync(this.previousWorkerCount);
         }
 
         /// <inheritdoc />

src/DurableTask.SqlServer.AzureFunctions/SqlTargetScaler.cs

@@ -0,0 +1,32 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace DurableTask.SqlServer.AzureFunctions
+{
+    using System.Threading.Tasks;
+    using Microsoft.Azure.WebJobs.Host.Scale;
+
+#if !NETSTANDARD
+    public class SqlTargetScaler : ITargetScaler
+    {
+        readonly SqlMetricsProvider sqlMetricsProvider;
+        readonly TargetScalerResult scaleResult;
+
+        public SqlTargetScaler(string functionId, SqlMetricsProvider sqlMetricsProvider)
+        {
+            this.sqlMetricsProvider = sqlMetricsProvider;
+            this.scaleResult = new TargetScalerResult();
+            this.TargetScalerDescriptor = new TargetScalerDescriptor(functionId);
+        }
+
+        public TargetScalerDescriptor TargetScalerDescriptor { get; private set; }
+
+        public async Task<TargetScalerResult> GetScaleResultAsync(TargetScalerContext context)
+        {
+            SqlScaleMetric sqlScaleMetric = await this.sqlMetricsProvider.GetMetricsAsync();
+            this.scaleResult.TargetWorkerCount = sqlScaleMetric.RecommendedReplicaCount;
+            return this.scaleResult;
+        }
+    }
+#endif
+}

test/DurableTask.SqlServer.AzureFunctions.Tests/TargetBasedScalingTests.cs

@@ -0,0 +1,59 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace DurableTask.SqlServer.AzureFunctions.Tests
+{
+    using System;
+    using DurableTask.Core;
+    using Microsoft.Azure.WebJobs.Extensions.DurableTask;
+    using Microsoft.Azure.WebJobs.Host.Scale;
+    using Microsoft.Extensions.Logging;
+    using Moq;
+    using Xunit;
+    using Xunit.Abstractions;
+
+    public class TargetBasedScalingTests
+    {
+        readonly Mock<SqlMetricsProvider> metricsProviderMock;
+        readonly Mock<IOrchestrationService> orchestrationServiceMock;
+
+        public TargetBasedScalingTests()
+        {
+            this.orchestrationServiceMock = new Mock<IOrchestrationService>(MockBehavior.Strict);
+
+            this.metricsProviderMock = new Mock<SqlMetricsProvider>(
+                MockBehavior.Strict,
+                null,
+                null);
+        }
+
+        [Theory]
+        [InlineData(0)]
+        [InlineData(10)]
+        [InlineData(20)]
+        public async void TargetBasedScalingTest(int expectedTargetWorkerCount)
+        {
+            var durabilityProviderMock = new Mock<DurabilityProvider>(
+                MockBehavior.Strict,
+                "storageProviderName",
+                this.orchestrationServiceMock.Object,
+                new Mock<IOrchestrationServiceClient>().Object,
+                "connectionName");
+
+            SqlScaleMetric sqlScaleMetric = new SqlScaleMetric()
+            {
+                RecommendedReplicaCount = expectedTargetWorkerCount,
+            };
+
+            this.metricsProviderMock.Setup(m => m.GetMetricsAsync(null)).ReturnsAsync(sqlScaleMetric);
+
+            SqlTargetScaler targetScaler = new SqlTargetScaler(
+                "functionId",
+                this.metricsProviderMock.Object);
+
+            TargetScalerResult result = await targetScaler.GetScaleResultAsync(new TargetScalerContext());
+
+            Assert.Equal(expectedTargetWorkerCount, result.TargetWorkerCount);
+        }
+    }
+}