Introduce Target Based Scaling

Directory.Packages.props

@@ -4,8 +4,8 @@
     <PackageVersion Include="Microsoft.Azure.WebJobs.Extensions.Storage" Version="5.0.1" />
     <PackageVersion Include="Microsoft.NET.Sdk.Functions" Version="4.1.3" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Abstractions" Version="1.1.0" />
+    <PackageVersion Include="Microsoft.Azure.WebJobs" Version="3.0.36" />
     <PackageVersion Include="Microsoft.ApplicationInsights" Version="2.21.0" />
-    <PackageVersion Include="Microsoft.Azure.WebJobs" Version="3.0.33" />
     <PackageVersion Include="Microsoft.Data.SqlClient" Version="5.0.1" />
     <PackageVersion Include="Microsoft.SourceLink.GitHub" Version="1.1.1" />
     <PackageVersion Include="morelinq" Version="3.3.2" />

docs/BindingsOverview.md

@@ -173,9 +173,16 @@ This controls the upper limit on the number of pending changes in the user table
 
 ### Scaling for Trigger Bindings
 
-If your application containing functions with SQL trigger bindings is running as an Azure function app, it will be scaled automatically based on the amount of changes that are pending to be processed in the user table. As of today, we only support scaling of function apps running in Elastic Premium plan. To enable scaling, you will need to go the function app resource's page on Azure Portal, then to Configuration > 'Function runtime settings' and turn on 'Runtime Scale Monitoring'. For more information, check documentation on [Runtime Scaling](https://learn.microsoft.com/azure/azure-functions/event-driven-scaling#runtime-scaling). You can configure scaling parameters by going to 'Scale out (App Service plan)' setting on the function app's page. To understand various scale settings, please check the respective sections in [Azure Functions Premium plan](https://learn.microsoft.com/azure/azure-functions/functions-premium-plan?tabs=portal#eliminate-cold-starts)'s documentation.
+If your application containing functions with SQL trigger bindings is running as an Azure function app, it will be scaled automatically based on the amount of changes that are pending to be processed in the user table. As of today, we only support scaling of function apps running in Elastic Premium plan with 'Runtime Scale Monitoring' enabled. To enable scaling, you will need to go the function app resource's page on Azure Portal, then to Configuration > 'Function runtime settings' and turn on 'Runtime Scale Monitoring'.
+
+There are two types of scaling available:
+
+- Incremental scaling - This scales the application serially, increasing or decreasing the workers by 1. There are a couple of checks made to decide on whether the host application needs to be scaled in or out. The rationale behind these checks is to ensure that the count of pending changes per application-worker stays below a certain maximum limit, controlled by [Sql_Trigger_MaxChangesPerWorker](#sql_trigger_maxchangesperworker), while also ensuring that the number of workers running stays minimal. The scaling decision is made based on the latest count of the pending changes and whether the last 5 samples we took were continually increasing or decreasing.
+
+- Target Based Scaling - This type of scaling depends on the pending change count and the value of [dynamic concurrency](https://learn.microsoft.com/azure/azure-functions/functions-concurrency#dynamic-concurrency) which if not enabled is defaulted to [Sql_Trigger_MaxChangesPerWorker](#sql_trigger_maxchangesperworker). The target worker count is decided by dividing the pending changes by the concurrency value. The application scales out to the number of instances specified by the target worker count.
+
+For more information, check documentation on [Runtime Scaling](https://learn.microsoft.com/azure/azure-functions/event-driven-scaling#runtime-scaling). You can configure scaling parameters by going to 'Scale out (App Service plan)' setting on the function app's page. To understand various scale settings, please check the respective sections in [Azure Functions Premium plan](https://learn.microsoft.com/azure/azure-functions/functions-premium-plan?tabs=portal#eliminate-cold-starts)'s documentation.
 
-There are a couple of checks made to decide on whether the host application needs to be scaled in or out. The rationale behind these checks is to ensure that the count of pending changes per application-worker stays below a certain maximum limit, which is defaulted to 1000, while also ensuring that the number of workers running stays minimal. The scaling decision is made based on the latest count of the pending changes and whether the last 5 times we checked the count, we found it to be continuously increasing or decreasing.
 
 ### Retry support for Trigger Bindings
 

nuget.config

@@ -19,4 +19,4 @@
     </packageSource>
   </packageSourceMapping>
 
-</configuration>
+</configuration>

performance/packages.lock.json

@@ -277,11 +277,12 @@
       },
       "Microsoft.Azure.WebJobs.Core": {
         "type": "Transitive",
-        "resolved": "3.0.33",
-        "contentHash": "4Rp5of6KFEGisQL7OJV2/8v3IQGFvTgZ9xSbB3bfkw5VHoNKn3Yllx7Qk/oUUX7Zey7jNu5mQFIR6dW6xNMX5Q==",
+        "resolved": "3.0.36",
+        "contentHash": "4ht/u677qxDL5rviXafYI5K9qtxee8peXQd8JDqG5KBRr5VXct2pgpEs5SnNu+gDO8dAVp+83N1IRiGG2Fl7Nw==",
         "dependencies": {
           "System.ComponentModel.Annotations": "4.4.0",
-          "System.Diagnostics.TraceSource": "4.3.0"
+          "System.Diagnostics.TraceSource": "4.3.0",
+          "System.Memory.Data": "1.0.1"
         }
       },
       "Microsoft.Azure.WebJobs.Extensions": {
@@ -1860,7 +1861,7 @@
         "dependencies": {
           "Microsoft.ApplicationInsights": "[2.21.0, )",
           "Microsoft.AspNetCore.Http": "[2.2.2, )",
-          "Microsoft.Azure.WebJobs": "[3.0.33, )",
+          "Microsoft.Azure.WebJobs": "[3.0.36, )",
           "Microsoft.Data.SqlClient": "[5.0.1, )",
           "Newtonsoft.Json": "[13.0.2, )",
           "System.Runtime.Caching": "[5.0.0, )",
@@ -1950,11 +1951,11 @@
       },
       "Microsoft.Azure.WebJobs": {
         "type": "CentralTransitive",
-        "requested": "[3.0.33, )",
-        "resolved": "3.0.33",
-        "contentHash": "eSv858ll+GsLYxM2+RKBiTC34CvGnEgLtnrzRljzrociIXsosadHDQLxvvqu0eyIQRRf5kc4MHII/wc0HRNvyg==",
+        "requested": "[3.0.36, )",
+        "resolved": "3.0.36",
+        "contentHash": "S5wppmL8sUfanGmo/zDzpoWrcM+IOvGNFjpkD5XJFCFzrBqktZTEX6MpP/vF3RN6VZm2sFKegZYyce+RNhgE4Q==",
         "dependencies": {
-          "Microsoft.Azure.WebJobs.Core": "3.0.33",
+          "Microsoft.Azure.WebJobs.Core": "3.0.36",
           "Microsoft.Extensions.Configuration": "2.1.1",
           "Microsoft.Extensions.Configuration.Abstractions": "2.1.1",
           "Microsoft.Extensions.Configuration.EnvironmentVariables": "2.1.0",
@@ -1963,8 +1964,8 @@
           "Microsoft.Extensions.Logging": "2.1.1",
           "Microsoft.Extensions.Logging.Abstractions": "2.1.1",
           "Microsoft.Extensions.Logging.Configuration": "2.1.0",
-          "Newtonsoft.Json": "11.0.2",
-          "System.Memory.Data": "1.0.1",
+          "Newtonsoft.Json": "13.0.1",
+          "System.Memory.Data": "1.0.2",
           "System.Threading.Tasks.Dataflow": "4.8.0"
         }
       },

samples/samples-csharp/packages.lock.json

@@ -277,11 +277,12 @@
       },
       "Microsoft.Azure.WebJobs.Core": {
         "type": "Transitive",
-        "resolved": "3.0.33",
-        "contentHash": "4Rp5of6KFEGisQL7OJV2/8v3IQGFvTgZ9xSbB3bfkw5VHoNKn3Yllx7Qk/oUUX7Zey7jNu5mQFIR6dW6xNMX5Q==",
+        "resolved": "3.0.36",
+        "contentHash": "4ht/u677qxDL5rviXafYI5K9qtxee8peXQd8JDqG5KBRr5VXct2pgpEs5SnNu+gDO8dAVp+83N1IRiGG2Fl7Nw==",
         "dependencies": {
           "System.ComponentModel.Annotations": "4.4.0",
-          "System.Diagnostics.TraceSource": "4.3.0"
+          "System.Diagnostics.TraceSource": "4.3.0",
+          "System.Memory.Data": "1.0.1"
         }
       },
       "Microsoft.Azure.WebJobs.Extensions": {
@@ -1720,7 +1721,7 @@
         "dependencies": {
           "Microsoft.ApplicationInsights": "[2.21.0, )",
           "Microsoft.AspNetCore.Http": "[2.2.2, )",
-          "Microsoft.Azure.WebJobs": "[3.0.33, )",
+          "Microsoft.Azure.WebJobs": "[3.0.36, )",
           "Microsoft.Data.SqlClient": "[5.0.1, )",
           "Newtonsoft.Json": "[13.0.2, )",
           "System.Runtime.Caching": "[5.0.0, )",
@@ -1761,11 +1762,11 @@
       },
       "Microsoft.Azure.WebJobs": {
         "type": "CentralTransitive",
-        "requested": "[3.0.33, )",
-        "resolved": "3.0.33",
-        "contentHash": "eSv858ll+GsLYxM2+RKBiTC34CvGnEgLtnrzRljzrociIXsosadHDQLxvvqu0eyIQRRf5kc4MHII/wc0HRNvyg==",
+        "requested": "[3.0.36, )",
+        "resolved": "3.0.36",
+        "contentHash": "S5wppmL8sUfanGmo/zDzpoWrcM+IOvGNFjpkD5XJFCFzrBqktZTEX6MpP/vF3RN6VZm2sFKegZYyce+RNhgE4Q==",
         "dependencies": {
-          "Microsoft.Azure.WebJobs.Core": "3.0.33",
+          "Microsoft.Azure.WebJobs.Core": "3.0.36",
           "Microsoft.Extensions.Configuration": "2.1.1",
           "Microsoft.Extensions.Configuration.Abstractions": "2.1.1",
           "Microsoft.Extensions.Configuration.EnvironmentVariables": "2.1.0",
@@ -1774,8 +1775,8 @@
           "Microsoft.Extensions.Logging": "2.1.1",
           "Microsoft.Extensions.Logging.Abstractions": "2.1.1",
           "Microsoft.Extensions.Logging.Configuration": "2.1.0",
-          "Newtonsoft.Json": "11.0.2",
-          "System.Memory.Data": "1.0.1",
+          "Newtonsoft.Json": "13.0.1",
+          "System.Memory.Data": "1.0.2",
           "System.Threading.Tasks.Dataflow": "4.8.0"
         }
       },

samples/samples-csharpscript/packages.lock.json

@@ -130,11 +130,12 @@
       },
       "Microsoft.Azure.WebJobs.Core": {
         "type": "Transitive",
-        "resolved": "3.0.33",
-        "contentHash": "4Rp5of6KFEGisQL7OJV2/8v3IQGFvTgZ9xSbB3bfkw5VHoNKn3Yllx7Qk/oUUX7Zey7jNu5mQFIR6dW6xNMX5Q==",
+        "resolved": "3.0.36",
+        "contentHash": "4ht/u677qxDL5rviXafYI5K9qtxee8peXQd8JDqG5KBRr5VXct2pgpEs5SnNu+gDO8dAVp+83N1IRiGG2Fl7Nw==",
         "dependencies": {
           "System.ComponentModel.Annotations": "4.4.0",
-          "System.Diagnostics.TraceSource": "4.3.0"
+          "System.Diagnostics.TraceSource": "4.3.0",
+          "System.Memory.Data": "1.0.1"
         }
       },
       "Microsoft.Azure.WebJobs.Extensions.Storage.Blobs": {
@@ -1574,7 +1575,7 @@
         "dependencies": {
           "Microsoft.ApplicationInsights": "[2.21.0, )",
           "Microsoft.AspNetCore.Http": "[2.2.2, )",
-          "Microsoft.Azure.WebJobs": "[3.0.33, )",
+          "Microsoft.Azure.WebJobs": "[3.0.36, )",
           "Microsoft.Data.SqlClient": "[5.0.1, )",
           "Newtonsoft.Json": "[13.0.2, )",
           "System.Runtime.Caching": "[5.0.0, )",
@@ -1592,11 +1593,11 @@
       },
       "Microsoft.Azure.WebJobs": {
         "type": "CentralTransitive",
-        "requested": "[3.0.33, )",
-        "resolved": "3.0.33",
-        "contentHash": "eSv858ll+GsLYxM2+RKBiTC34CvGnEgLtnrzRljzrociIXsosadHDQLxvvqu0eyIQRRf5kc4MHII/wc0HRNvyg==",
+        "requested": "[3.0.36, )",
+        "resolved": "3.0.36",
+        "contentHash": "S5wppmL8sUfanGmo/zDzpoWrcM+IOvGNFjpkD5XJFCFzrBqktZTEX6MpP/vF3RN6VZm2sFKegZYyce+RNhgE4Q==",
         "dependencies": {
-          "Microsoft.Azure.WebJobs.Core": "3.0.33",
+          "Microsoft.Azure.WebJobs.Core": "3.0.36",
           "Microsoft.Extensions.Configuration": "2.1.1",
           "Microsoft.Extensions.Configuration.Abstractions": "2.1.1",
           "Microsoft.Extensions.Configuration.EnvironmentVariables": "2.1.0",
@@ -1605,8 +1606,8 @@
           "Microsoft.Extensions.Logging": "2.1.1",
           "Microsoft.Extensions.Logging.Abstractions": "2.1.1",
           "Microsoft.Extensions.Logging.Configuration": "2.1.0",
-          "Newtonsoft.Json": "11.0.2",
-          "System.Memory.Data": "1.0.1",
+          "Newtonsoft.Json": "13.0.1",
+          "System.Memory.Data": "1.0.2",
           "System.Threading.Tasks.Dataflow": "4.8.0"
         }
       },

samples/samples-outofproc/packages.lock.json

@@ -846,9 +846,9 @@
       "microsoft.azure.functions.worker.extensions.sql": {
         "type": "Project",
         "dependencies": {
-          "Microsoft.Azure.Functions.Worker.Extensions.Abstractions": "[1.1.0, )",
-          "Microsoft.Data.SqlClient": "[5.0.1, )",
-          "System.Drawing.Common": "[5.0.3, )"
+          "Microsoft.Azure.Functions.Worker.Extensions.Abstractions": "1.1.0",
+          "Microsoft.Data.SqlClient": "5.0.1",
+          "System.Drawing.Common": "5.0.3"
         }
       },
       "Microsoft.Azure.Functions.Worker.Extensions.Abstractions": {

src/TriggerBinding/SqlTableChangeMonitor.cs

@@ -180,69 +180,6 @@ public void Dispose()
             this._cancellationTokenSourceCheckForChanges.Cancel();
         }
 
-        public async Task<long> GetUnprocessedChangeCountAsync()
-        {
-            long unprocessedChangeCount = 0L;
-
-            try
-            {
-                long getUnprocessedChangesDurationMs = 0L;
-
-                using (var connection = new SqlConnection(this._connectionString))
-                {
-                    this._logger.LogDebugWithThreadId("BEGIN OpenGetUnprocessedChangesConnection");
-                    await connection.OpenAsync();
-                    this._logger.LogDebugWithThreadId("END OpenGetUnprocessedChangesConnection");
-
-                    // Use a transaction to automatically release the app lock when we're done executing the query
-                    using (SqlTransaction transaction = connection.BeginTransaction(IsolationLevel.RepeatableRead))
-                    {
-                        try
-                        {
-                            using (SqlCommand getUnprocessedChangesCommand = this.BuildGetUnprocessedChangesCommand(connection, transaction))
-                            {
-                                this._logger.LogInformation("Getting change count");
-                                this._logger.LogDebugWithThreadId($"BEGIN GetUnprocessedChangeCount Query={getUnprocessedChangesCommand.CommandText}");
-                                var commandSw = Stopwatch.StartNew();
-                                unprocessedChangeCount = (long)await getUnprocessedChangesCommand.ExecuteScalarAsync();
-                                getUnprocessedChangesDurationMs = commandSw.ElapsedMilliseconds;
-                            }
-
-                            this._logger.LogDebugWithThreadId($"END GetUnprocessedChangeCount Duration={getUnprocessedChangesDurationMs}ms Count={unprocessedChangeCount}");
-                            transaction.Commit();
-                        }
-                        catch (Exception)
-                        {
-                            try
-                            {
-                                transaction.Rollback();
-                            }
-                            catch (Exception ex2)
-                            {
-                                this._logger.LogError($"GetUnprocessedChangeCount : Failed to rollback transaction due to exception: {ex2.GetType()}. Exception message: {ex2.Message}");
-                                TelemetryInstance.TrackException(TelemetryErrorName.GetUnprocessedChangeCountRollback, ex2, this._telemetryProps);
-                            }
-                            throw;
-                        }
-                    }
-                }
-
-                var measures = new Dictionary<TelemetryMeasureName, double>
-                {
-                    [TelemetryMeasureName.GetUnprocessedChangesDurationMs] = getUnprocessedChangesDurationMs,
-                    [TelemetryMeasureName.UnprocessedChangeCount] = unprocessedChangeCount,
-                };
-            }
-            catch (Exception ex)
-            {
-                this._logger.LogError($"Failed to query count of unprocessed changes for table '{this._userTable.FullName}' due to exception: {ex.GetType()}. Exception message: {ex.Message}");
-                TelemetryInstance.TrackException(TelemetryErrorName.GetUnprocessedChangeCount, ex, this._telemetryProps);
-                throw;
-            }
-
-            return unprocessedChangeCount;
-        }
-
         /// <summary>
         /// Executed once every <see cref="_pollingIntervalInMs"/> period. Each iteration will go through a series of state
         /// changes, if an error occurs in any of them then it will skip the rest of the states and try again in the next
@@ -904,38 +841,6 @@ LEFT OUTER JOIN {this._userTable.BracketQuotedFullName} AS u ON {userTableJoinCo
             return new SqlCommand(getChangesQuery, connection, transaction);
         }
 
-        /// <summary>
-        /// Builds the query to get count of unprocessed changes in the user's table. This one mimics the query that is
-        /// used by workers to get the changes for processing.
-        /// </summary>
-        /// <param name="connection">The connection to add to the returned SqlCommand</param>
-        /// <param name="transaction">The transaction to add to the returned SqlCommand</param>
-        /// <returns>The SqlCommand populated with the query and appropriate parameters</returns>
-        private SqlCommand BuildGetUnprocessedChangesCommand(SqlConnection connection, SqlTransaction transaction)
-        {
-            string leasesTableJoinCondition = string.Join(" AND ", this._primaryKeyColumns.Select(col => $"c.{col.name.AsBracketQuotedString()} = l.{col.name.AsBracketQuotedString()}"));
-
-            string getUnprocessedChangesQuery = $@"
-                {AppLockStatements}
-
-                DECLARE @last_sync_version bigint;
-                SELECT @last_sync_version = LastSyncVersion
-                FROM {GlobalStateTableName}
-                WHERE UserFunctionID = '{this._userFunctionId}' AND UserTableID = {this._userTableId};
-
-                SELECT COUNT_BIG(*)
-                FROM CHANGETABLE(CHANGES {this._userTable.BracketQuotedFullName}, @last_sync_version) AS c
-                LEFT OUTER JOIN {this._leasesTableName} AS l ON {leasesTableJoinCondition}
-                WHERE
-                    (l.{LeasesTableLeaseExpirationTimeColumnName} IS NULL AND
-                       (l.{LeasesTableChangeVersionColumnName} IS NULL OR l.{LeasesTableChangeVersionColumnName} < c.{SysChangeVersionColumnName}) OR
-                        l.{LeasesTableLeaseExpirationTimeColumnName} < SYSDATETIME()) AND
-                    (l.{LeasesTableAttemptCountColumnName} IS NULL OR l.{LeasesTableAttemptCountColumnName} < {MaxChangeProcessAttemptCount});
-            ";
-
-            return new SqlCommand(getUnprocessedChangesQuery, connection, transaction);
-        }
-
         /// <summary>
         /// Builds the query to acquire leases on the rows in "_rows" if changes are detected in the user's table
         /// (<see cref="RunChangeConsumptionLoopAsync()"/>).