diff --git a/Worker.Extensions.Sql/src/SqlTriggerAttribute.cs b/Worker.Extensions.Sql/src/SqlTriggerAttribute.cs index db9549ca1..42523923c 100644 --- a/Worker.Extensions.Sql/src/SqlTriggerAttribute.cs +++ b/Worker.Extensions.Sql/src/SqlTriggerAttribute.cs @@ -13,12 +13,21 @@ public sealed class SqlTriggerAttribute : TriggerBindingAttribute /// /// Name of the table to watch for changes. /// The name of the app setting where the SQL connection string is stored - public SqlTriggerAttribute(string tableName, string connectionStringSetting) + /// Optional - The name of the table used to store leases. If not specified, the leases table name will be Leases_{FunctionId}_{TableId} + public SqlTriggerAttribute(string tableName, string connectionStringSetting, string leasesTableName = null) { this.TableName = tableName ?? throw new ArgumentNullException(nameof(tableName)); this.ConnectionStringSetting = connectionStringSetting ?? throw new ArgumentNullException(nameof(connectionStringSetting)); + this.LeasesTableName = leasesTableName; } + /// + /// Initializes a new instance of the class with null value for LeasesTableName. + /// + /// Name of the table to watch for changes. + /// The name of the app setting where the SQL connection string is stored + public SqlTriggerAttribute(string tableName, string connectionStringSetting) : this(tableName, connectionStringSetting, null) { } + /// /// Name of the app setting containing the SQL connection string. /// @@ -28,5 +37,13 @@ public SqlTriggerAttribute(string tableName, string connectionStringSetting) /// Name of the table to watch for changes. /// public string TableName { get; } + + /// + /// Name of the table used to store leases. + /// If not specified, the leases table name will be Leases_{FunctionId}_{TableId} + /// More information on how this is generated can be found here + /// https://github.com/Azure/azure-functions-sql-extension/blob/release/trigger/docs/TriggerBinding.md#az_funcleases_ + /// + public string LeasesTableName { get; } } } \ No newline at end of file diff --git a/samples/samples-csharp/TriggerBindingSamples/ProductsTriggerLeasesTableName.cs b/samples/samples-csharp/TriggerBindingSamples/ProductsTriggerLeasesTableName.cs new file mode 100644 index 000000000..94f01458e --- /dev/null +++ b/samples/samples-csharp/TriggerBindingSamples/ProductsTriggerLeasesTableName.cs @@ -0,0 +1,22 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. See License.txt in the project root for license information. + +using System.Collections.Generic; +using Microsoft.Azure.WebJobs.Extensions.Sql.Samples.Common; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; + +namespace Microsoft.Azure.WebJobs.Extensions.Sql.Samples.TriggerBindingSamples +{ + public static class ProductsTriggerLeasesTableName + { + [FunctionName(nameof(ProductsTriggerLeasesTableName))] + public static void Run( + [SqlTrigger("[dbo].[Products]", "SqlConnectionString", "Leases")] + IReadOnlyList> changes, + ILogger logger) + { + logger.LogInformation("SQL Changes: " + JsonConvert.SerializeObject(changes)); + } + } +} diff --git a/samples/samples-csx/ProductsTriggerLeasesTableName/function.json b/samples/samples-csx/ProductsTriggerLeasesTableName/function.json new file mode 100644 index 000000000..8b9d204b1 --- /dev/null +++ b/samples/samples-csx/ProductsTriggerLeasesTableName/function.json @@ -0,0 +1,13 @@ +{ + "bindings": [ + { + "name": "changes", + "type": "sqlTrigger", + "direction": "in", + "tableName": "dbo.Products", + "connectionStringSetting": "SqlConnectionString", + "leasesTableName": "Leases" + } + ], + "disabled": false +} \ No newline at end of file diff --git a/samples/samples-csx/ProductsTriggerLeasesTableName/run.csx b/samples/samples-csx/ProductsTriggerLeasesTableName/run.csx new file mode 100644 index 000000000..4a0cb01cf --- /dev/null +++ b/samples/samples-csx/ProductsTriggerLeasesTableName/run.csx @@ -0,0 +1,14 @@ +#load "../Common/product.csx" +#r "Newtonsoft.Json" +#r "Microsoft.Azure.WebJobs.Extensions.Sql" + +using System.Net; +using Microsoft.AspNetCore.Mvc; +using Microsoft.Extensions.Primitives; +using Newtonsoft.Json; +using Microsoft.Azure.WebJobs.Extensions.Sql; + +public static void Run(IReadOnlyList> changes, ILogger log) +{ + log.LogInformation("SQL Changes: " + JsonConvert.SerializeObject(changes)); +} \ No newline at end of file diff --git a/samples/samples-js/ProductsTriggerLeasesTableName/function.json b/samples/samples-js/ProductsTriggerLeasesTableName/function.json new file mode 100644 index 000000000..075d638d9 --- /dev/null +++ b/samples/samples-js/ProductsTriggerLeasesTableName/function.json @@ -0,0 +1,13 @@ +{ + "bindings": [ + { + "name": "changes", + "type": "sqlTrigger", + "direction": "in", + "tableName": "dbo.Products", + "connectionStringSetting": "SqlConnectionString", + "leasesTableName": "Leases" + } + ], + "disabled": false + } \ No newline at end of file diff --git a/samples/samples-js/ProductsTriggerLeasesTableName/index.js b/samples/samples-js/ProductsTriggerLeasesTableName/index.js new file mode 100644 index 000000000..5886e5c82 --- /dev/null +++ b/samples/samples-js/ProductsTriggerLeasesTableName/index.js @@ -0,0 +1,6 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. See License.txt in the project root for license information. + +module.exports = async function (context, changes) { + context.log(`SQL Changes: ${JSON.stringify(changes)}`) +} \ No newline at end of file diff --git a/samples/samples-outofproc/TriggerBindingSamples/ProductsTriggerLeasesTableName.cs b/samples/samples-outofproc/TriggerBindingSamples/ProductsTriggerLeasesTableName.cs new file mode 100644 index 000000000..82943bd9b --- /dev/null +++ b/samples/samples-outofproc/TriggerBindingSamples/ProductsTriggerLeasesTableName.cs @@ -0,0 +1,29 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. See License.txt in the project root for license information. + +using System.Collections.Generic; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Azure.Functions.Worker.Extensions.Sql; +using Microsoft.Azure.WebJobs.Extensions.Sql.SamplesOutOfProc.Common; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; +using System; + +namespace Microsoft.Azure.WebJobs.Extensions.Sql.SamplesOutOfProc.TriggerBindingSamples +{ + public class ProductsTriggerLeasesTableName + { + private static readonly Action _loggerMessage = LoggerMessage.Define(LogLevel.Information, eventId: new EventId(0, "INFO"), formatString: "{Message}"); + + [Function("ProductsTriggerLeasesTableName")] + public static void Run( + [SqlTrigger("[dbo].[Products]", "SqlConnectionString", "Leases")] + IReadOnlyList> changes, FunctionContext context) + { + if (changes != null && changes.Count > 0) + { + _loggerMessage(context.GetLogger("ProductsTriggerLeasesTableName"), "SQL Changes: " + JsonConvert.SerializeObject(changes), null); + } + } + } +} diff --git a/samples/samples-powershell/ProductsTriggerLeasesTableName/function.json b/samples/samples-powershell/ProductsTriggerLeasesTableName/function.json new file mode 100644 index 000000000..075d638d9 --- /dev/null +++ b/samples/samples-powershell/ProductsTriggerLeasesTableName/function.json @@ -0,0 +1,13 @@ +{ + "bindings": [ + { + "name": "changes", + "type": "sqlTrigger", + "direction": "in", + "tableName": "dbo.Products", + "connectionStringSetting": "SqlConnectionString", + "leasesTableName": "Leases" + } + ], + "disabled": false + } \ No newline at end of file diff --git a/samples/samples-powershell/ProductsTriggerLeasesTableName/run.ps1 b/samples/samples-powershell/ProductsTriggerLeasesTableName/run.ps1 new file mode 100644 index 000000000..c990058f5 --- /dev/null +++ b/samples/samples-powershell/ProductsTriggerLeasesTableName/run.ps1 @@ -0,0 +1,9 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for license information. + +using namespace System.Net + +param($changes) + +$changesJson = $changes | ConvertTo-Json -Compress +Write-Host "SQL Changes: $changesJson" \ No newline at end of file diff --git a/samples/samples-python/ProductsTriggerLeasesTableName/__init__.py b/samples/samples-python/ProductsTriggerLeasesTableName/__init__.py new file mode 100644 index 000000000..4ef0f7b40 --- /dev/null +++ b/samples/samples-python/ProductsTriggerLeasesTableName/__init__.py @@ -0,0 +1,8 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +import json +import logging + +def main(changes): + logging.info("SQL Changes: %s", json.loads(changes)) diff --git a/samples/samples-python/ProductsTriggerLeasesTableName/function.json b/samples/samples-python/ProductsTriggerLeasesTableName/function.json new file mode 100644 index 000000000..075d638d9 --- /dev/null +++ b/samples/samples-python/ProductsTriggerLeasesTableName/function.json @@ -0,0 +1,13 @@ +{ + "bindings": [ + { + "name": "changes", + "type": "sqlTrigger", + "direction": "in", + "tableName": "dbo.Products", + "connectionStringSetting": "SqlConnectionString", + "leasesTableName": "Leases" + } + ], + "disabled": false + } \ No newline at end of file diff --git a/src/TriggerBinding/SqlTriggerAttribute.cs b/src/TriggerBinding/SqlTriggerAttribute.cs index 92e8ed790..b1cc84497 100644 --- a/src/TriggerBinding/SqlTriggerAttribute.cs +++ b/src/TriggerBinding/SqlTriggerAttribute.cs @@ -18,12 +18,21 @@ public sealed class SqlTriggerAttribute : Attribute /// /// Name of the table to watch for changes. /// The name of the app setting where the SQL connection string is stored - public SqlTriggerAttribute(string tableName, string connectionStringSetting) + /// Optional - The name of the table used to store leases. If not specified, the leases table name will be Leases_{FunctionId}_{TableId} + public SqlTriggerAttribute(string tableName, string connectionStringSetting, string leasesTableName = null) { this.TableName = tableName ?? throw new ArgumentNullException(nameof(tableName)); this.ConnectionStringSetting = connectionStringSetting ?? throw new ArgumentNullException(nameof(connectionStringSetting)); + this.LeasesTableName = leasesTableName; } + /// + /// Initializes a new instance of the class with null value for LeasesTableName. + /// + /// Name of the table to watch for changes. + /// The name of the app setting where the SQL connection string is stored + public SqlTriggerAttribute(string tableName, string connectionStringSetting) : this(tableName, connectionStringSetting, null) { } + /// /// Name of the app setting containing the SQL connection string. /// @@ -34,5 +43,13 @@ public SqlTriggerAttribute(string tableName, string connectionStringSetting) /// Name of the table to watch for changes. /// public string TableName { get; } + + /// + /// Name of the table used to store leases. + /// If not specified, the leases table name will be Leases_{FunctionId}_{TableId} + /// More information on how this is generated can be found here + /// https://github.com/Azure/azure-functions-sql-extension/blob/release/trigger/docs/TriggerBinding.md#az_funcleases_ + /// + public string LeasesTableName { get; } } } \ No newline at end of file diff --git a/src/TriggerBinding/SqlTriggerBinding.cs b/src/TriggerBinding/SqlTriggerBinding.cs index 841c8a79e..4e8415de1 100644 --- a/src/TriggerBinding/SqlTriggerBinding.cs +++ b/src/TriggerBinding/SqlTriggerBinding.cs @@ -27,6 +27,7 @@ internal sealed class SqlTriggerBinding : ITriggerBinding { private readonly string _connectionString; private readonly string _tableName; + private readonly string _leasesTableName; private readonly ParameterInfo _parameter; private readonly IHostIdProvider _hostIdProvider; private readonly ILogger _logger; @@ -40,14 +41,16 @@ internal sealed class SqlTriggerBinding : ITriggerBinding /// /// SQL connection string used to connect to user database /// Name of the user table + /// Optional - Name of the leases table /// Trigger binding parameter information /// Provider of unique host identifier /// Facilitates logging of messages /// Provides configuration values - public SqlTriggerBinding(string connectionString, string tableName, ParameterInfo parameter, IHostIdProvider hostIdProvider, ILogger logger, IConfiguration configuration) + public SqlTriggerBinding(string connectionString, string tableName, string leasesTableName, ParameterInfo parameter, IHostIdProvider hostIdProvider, ILogger logger, IConfiguration configuration) { this._connectionString = connectionString ?? throw new ArgumentNullException(nameof(connectionString)); this._tableName = tableName ?? throw new ArgumentNullException(nameof(tableName)); + this._leasesTableName = leasesTableName; this._parameter = parameter ?? throw new ArgumentNullException(nameof(parameter)); this._hostIdProvider = hostIdProvider ?? throw new ArgumentNullException(nameof(hostIdProvider)); this._logger = logger ?? throw new ArgumentNullException(nameof(logger)); @@ -72,7 +75,7 @@ public async Task CreateListenerAsync(ListenerFactoryContext context) _ = context ?? throw new ArgumentNullException(nameof(context), "Missing listener context"); string userFunctionId = await this.GetUserFunctionIdAsync(); - return new SqlTriggerListener(this._connectionString, this._tableName, userFunctionId, context.Executor, this._logger, this._configuration); + return new SqlTriggerListener(this._connectionString, this._tableName, this._leasesTableName, userFunctionId, context.Executor, this._logger, this._configuration); } public ParameterDescriptor ToParameterDescriptor() diff --git a/src/TriggerBinding/SqlTriggerBindingProvider.cs b/src/TriggerBinding/SqlTriggerBindingProvider.cs index 5c139c299..26414775d 100644 --- a/src/TriggerBinding/SqlTriggerBindingProvider.cs +++ b/src/TriggerBinding/SqlTriggerBindingProvider.cs @@ -87,10 +87,10 @@ public Task TryCreateAsync(TriggerBindingProviderContext contex bindingType = typeof(SqlTriggerBinding<>).MakeGenericType(userType); } - var constructorParameterTypes = new Type[] { typeof(string), typeof(string), typeof(ParameterInfo), typeof(IHostIdProvider), typeof(ILogger), typeof(IConfiguration) }; + var constructorParameterTypes = new Type[] { typeof(string), typeof(string), typeof(string), typeof(ParameterInfo), typeof(IHostIdProvider), typeof(ILogger), typeof(IConfiguration) }; ConstructorInfo bindingConstructor = bindingType.GetConstructor(constructorParameterTypes); - object[] constructorParameterValues = new object[] { connectionString, attribute.TableName, parameter, this._hostIdProvider, this._logger, this._configuration }; + object[] constructorParameterValues = new object[] { connectionString, attribute.TableName, attribute.LeasesTableName, parameter, this._hostIdProvider, this._logger, this._configuration }; var triggerBinding = (ITriggerBinding)bindingConstructor.Invoke(constructorParameterValues); return Task.FromResult(triggerBinding); diff --git a/src/TriggerBinding/SqlTriggerConstants.cs b/src/TriggerBinding/SqlTriggerConstants.cs index 16fbcdf3e..4da60a0e0 100644 --- a/src/TriggerBinding/SqlTriggerConstants.cs +++ b/src/TriggerBinding/SqlTriggerConstants.cs @@ -11,6 +11,8 @@ internal static class SqlTriggerConstants public const string LeasesTableNameFormat = "[" + SchemaName + "].[Leases_{0}]"; + public const string UserDefinedLeasesTableNameFormat = "[" + SchemaName + "].{0}"; + public const string LeasesTableChangeVersionColumnName = "_az_func_ChangeVersion"; public const string LeasesTableAttemptCountColumnName = "_az_func_AttemptCount"; public const string LeasesTableLeaseExpirationTimeColumnName = "_az_func_LeaseExpirationTime"; diff --git a/src/TriggerBinding/SqlTriggerListener.cs b/src/TriggerBinding/SqlTriggerListener.cs index bf578ff60..7d928a80b 100644 --- a/src/TriggerBinding/SqlTriggerListener.cs +++ b/src/TriggerBinding/SqlTriggerListener.cs @@ -4,7 +4,6 @@ using System; using System.Collections.Generic; using System.Diagnostics; -using System.Globalization; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -40,6 +39,7 @@ internal sealed class SqlTriggerListener : IListener, IScaleMonitorProvider, private readonly SqlObject _userTable; private readonly string _connectionString; + private readonly string _userDefinedLeasesTableName; private readonly string _userFunctionId; private readonly ITriggeredFunctionExecutor _executor; private readonly ILogger _logger; @@ -61,14 +61,16 @@ internal sealed class SqlTriggerListener : IListener, IScaleMonitorProvider, /// /// SQL connection string used to connect to user database /// Name of the user table + /// Optional - Name of the leases table /// Unique identifier for the user function /// Defines contract for triggering user function /// Facilitates logging of messages /// Provides configuration values - public SqlTriggerListener(string connectionString, string tableName, string userFunctionId, ITriggeredFunctionExecutor executor, ILogger logger, IConfiguration configuration) + public SqlTriggerListener(string connectionString, string tableName, string userDefinedLeasesTableName, string userFunctionId, ITriggeredFunctionExecutor executor, ILogger logger, IConfiguration configuration) { this._connectionString = !string.IsNullOrEmpty(connectionString) ? connectionString : throw new ArgumentNullException(nameof(connectionString)); this._userTable = !string.IsNullOrEmpty(tableName) ? new SqlObject(tableName) : throw new ArgumentNullException(nameof(tableName)); + this._userDefinedLeasesTableName = userDefinedLeasesTableName; this._userFunctionId = !string.IsNullOrEmpty(userFunctionId) ? userFunctionId : throw new ArgumentNullException(nameof(userFunctionId)); this._executor = executor ?? throw new ArgumentNullException(nameof(executor)); this._logger = logger ?? throw new ArgumentNullException(nameof(logger)); @@ -82,8 +84,8 @@ public SqlTriggerListener(string connectionString, string tableName, string user } this._hasConfiguredMaxChangesPerWorker = configuredMaxChangesPerWorker != null; - this._scaleMonitor = new SqlTriggerScaleMonitor(this._userFunctionId, this._userTable, this._connectionString, this._maxChangesPerWorker, this._logger); - this._targetScaler = new SqlTriggerTargetScaler(this._userFunctionId, this._userTable, this._connectionString, this._maxChangesPerWorker, this._logger); + this._scaleMonitor = new SqlTriggerScaleMonitor(this._userFunctionId, this._userTable, this._userDefinedLeasesTableName, this._connectionString, this._maxChangesPerWorker, this._logger); + this._targetScaler = new SqlTriggerTargetScaler(this._userFunctionId, this._userTable, this._userDefinedLeasesTableName, this._connectionString, this._maxChangesPerWorker, this._logger); } public void Cancel() @@ -123,7 +125,7 @@ public async Task StartAsync(CancellationToken cancellationToken) IReadOnlyList<(string name, string type)> primaryKeyColumns = await GetPrimaryKeyColumnsAsync(connection, userTableId, this._logger, this._userTable.FullName, cancellationToken); IReadOnlyList userTableColumns = await this.GetUserTableColumnsAsync(connection, userTableId, cancellationToken); - string leasesTableName = string.Format(CultureInfo.InvariantCulture, LeasesTableNameFormat, $"{this._userFunctionId}_{userTableId}"); + string leasesTableName = GetLeasesTableName(this._userDefinedLeasesTableName, this._userFunctionId, userTableId); this._telemetryProps[TelemetryPropertyName.LeasesTableName] = leasesTableName; var transactionSw = Stopwatch.StartNew(); diff --git a/src/TriggerBinding/SqlTriggerMetricsProvider.cs b/src/TriggerBinding/SqlTriggerMetricsProvider.cs index 67ad5ea6b..1a1c60c8c 100644 --- a/src/TriggerBinding/SqlTriggerMetricsProvider.cs +++ b/src/TriggerBinding/SqlTriggerMetricsProvider.cs @@ -5,7 +5,6 @@ using System.Collections.Generic; using System.Data; using System.Diagnostics; -using System.Globalization; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -27,13 +26,15 @@ internal class SqlTriggerMetricsProvider private readonly ILogger _logger; private readonly SqlObject _userTable; private readonly string _userFunctionId; + private readonly string _userDefinedLeasesTableName; - public SqlTriggerMetricsProvider(string connectionString, ILogger logger, SqlObject userTable, string userFunctionId) + public SqlTriggerMetricsProvider(string connectionString, ILogger logger, SqlObject userTable, string userFunctionId, string userDefinedLeasesTableName) { this._connectionString = !string.IsNullOrEmpty(connectionString) ? connectionString : throw new ArgumentNullException(nameof(connectionString)); this._logger = logger ?? throw new ArgumentNullException(nameof(logger)); this._userTable = userTable ?? throw new ArgumentNullException(nameof(userTable)); this._userFunctionId = !string.IsNullOrEmpty(userFunctionId) ? userFunctionId : throw new ArgumentNullException(nameof(userFunctionId)); + this._userDefinedLeasesTableName = userDefinedLeasesTableName; } public async Task GetMetricsAsync() { @@ -99,7 +100,7 @@ private async Task GetUnprocessedChangeCountAsync() private SqlCommand BuildGetUnprocessedChangesCommand(SqlConnection connection, SqlTransaction transaction, IReadOnlyList<(string name, string type)> primaryKeyColumns, int userTableId) { string leasesTableJoinCondition = string.Join(" AND ", primaryKeyColumns.Select(col => $"c.{col.name.AsBracketQuotedString()} = l.{col.name.AsBracketQuotedString()}")); - string leasesTableName = string.Format(CultureInfo.InvariantCulture, LeasesTableNameFormat, $"{this._userFunctionId}_{userTableId}"); + string leasesTableName = GetLeasesTableName(this._userDefinedLeasesTableName, this._userFunctionId, userTableId); string getUnprocessedChangesQuery = $@" {AppLockStatements} diff --git a/src/TriggerBinding/SqlTriggerScaleMonitor.cs b/src/TriggerBinding/SqlTriggerScaleMonitor.cs index c30f54e30..d5f1fbf14 100644 --- a/src/TriggerBinding/SqlTriggerScaleMonitor.cs +++ b/src/TriggerBinding/SqlTriggerScaleMonitor.cs @@ -25,7 +25,7 @@ internal sealed class SqlTriggerScaleMonitor : IScaleMonitor private readonly IDictionary _telemetryProps = new Dictionary(); private readonly int _maxChangesPerWorker; - public SqlTriggerScaleMonitor(string userFunctionId, SqlObject userTable, string connectionString, int maxChangesPerWorker, ILogger logger) + public SqlTriggerScaleMonitor(string userFunctionId, SqlObject userTable, string userDefinedLeasesTableName, string connectionString, int maxChangesPerWorker, ILogger logger) { _ = !string.IsNullOrEmpty(userFunctionId) ? true : throw new ArgumentNullException(userFunctionId); _ = userTable != null ? true : throw new ArgumentNullException(nameof(userTable)); @@ -33,7 +33,7 @@ public SqlTriggerScaleMonitor(string userFunctionId, SqlObject userTable, string // Do not convert the scale-monitor ID to lower-case string since SQL table names can be case-sensitive // depending on the collation of the current database. this.Descriptor = new ScaleMonitorDescriptor($"{userFunctionId}-SqlTrigger-{this._userTable.FullName}", userFunctionId); - this._metricsProvider = new SqlTriggerMetricsProvider(connectionString, logger, this._userTable, userFunctionId); + this._metricsProvider = new SqlTriggerMetricsProvider(connectionString, logger, this._userTable, userFunctionId, userDefinedLeasesTableName); this._logger = logger ?? throw new ArgumentNullException(nameof(logger)); this._maxChangesPerWorker = maxChangesPerWorker; } diff --git a/src/TriggerBinding/SqlTriggerTargetScaler.cs b/src/TriggerBinding/SqlTriggerTargetScaler.cs index b1aabf5eb..7c8bda82a 100644 --- a/src/TriggerBinding/SqlTriggerTargetScaler.cs +++ b/src/TriggerBinding/SqlTriggerTargetScaler.cs @@ -15,9 +15,9 @@ internal sealed class SqlTriggerTargetScaler : ITargetScaler private readonly SqlTriggerMetricsProvider _metricsProvider; private readonly int _maxChangesPerWorker; - public SqlTriggerTargetScaler(string userFunctionId, SqlObject userTable, string connectionString, int maxChangesPerWorker, ILogger logger) + public SqlTriggerTargetScaler(string userFunctionId, SqlObject userTable, string userDefinedLeasesTableName, string connectionString, int maxChangesPerWorker, ILogger logger) { - this._metricsProvider = new SqlTriggerMetricsProvider(connectionString, logger, userTable, userFunctionId); + this._metricsProvider = new SqlTriggerMetricsProvider(connectionString, logger, userTable, userFunctionId, userDefinedLeasesTableName); this.TargetScalerDescriptor = new TargetScalerDescriptor(userFunctionId); this._maxChangesPerWorker = maxChangesPerWorker; } diff --git a/src/TriggerBinding/SqlTriggerUtils.cs b/src/TriggerBinding/SqlTriggerUtils.cs index 9d704be45..43eb87f11 100644 --- a/src/TriggerBinding/SqlTriggerUtils.cs +++ b/src/TriggerBinding/SqlTriggerUtils.cs @@ -3,11 +3,13 @@ using System; using System.Collections.Generic; +using System.Globalization; using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Data.SqlClient; using Microsoft.Extensions.Logging; +using static Microsoft.Azure.WebJobs.Extensions.Sql.SqlTriggerConstants; namespace Microsoft.Azure.WebJobs.Extensions.Sql { @@ -111,5 +113,17 @@ internal static async Task GetUserTableIdAsync(SqlConnection connection, Sq return (int)userTableId; } } + + /// + /// Returns the formatted leases table name. If userDefinedLeasesTableName is null, the default name Leases_{FunctionId}_{TableId} is used. + /// + /// Leases table name defined by the user + /// SQL object ID of the user table + /// Unique identifier for the user function + internal static string GetLeasesTableName(string userDefinedLeasesTableName, string userFunctionId, int userTableId) + { + return string.IsNullOrEmpty(userDefinedLeasesTableName) ? string.Format(CultureInfo.InvariantCulture, LeasesTableNameFormat, $"{userFunctionId}_{userTableId}") : + string.Format(CultureInfo.InvariantCulture, UserDefinedLeasesTableNameFormat, $"{userDefinedLeasesTableName.AsBracketQuotedString()}"); + } } } \ No newline at end of file diff --git a/test/Integration/SqlTriggerBindingIntegrationTests.cs b/test/Integration/SqlTriggerBindingIntegrationTests.cs index f5f93b26d..56f1d6cab 100644 --- a/test/Integration/SqlTriggerBindingIntegrationTests.cs +++ b/test/Integration/SqlTriggerBindingIntegrationTests.cs @@ -544,11 +544,11 @@ public async void GetMetricsTest() this.SetChangeTrackingForTable("Products"); string userFunctionId = "func-id"; IConfiguration configuration = new ConfigurationBuilder().Build(); - var listener = new SqlTriggerListener(this.DbConnectionString, "dbo.Products", userFunctionId, Mock.Of(), Mock.Of(), configuration); + var listener = new SqlTriggerListener(this.DbConnectionString, "dbo.Products", "", userFunctionId, Mock.Of(), Mock.Of(), configuration); await listener.StartAsync(CancellationToken.None); // Cancel immediately so the listener doesn't start processing the changes await listener.StopAsync(CancellationToken.None); - var metricsProvider = new SqlTriggerMetricsProvider(this.DbConnectionString, Mock.Of(), new SqlObject("dbo.Products"), userFunctionId); + var metricsProvider = new SqlTriggerMetricsProvider(this.DbConnectionString, Mock.Of(), new SqlObject("dbo.Products"), userFunctionId, ""); SqlTriggerMetrics metrics = await metricsProvider.GetMetricsAsync(); Assert.True(metrics.UnprocessedChangeCount == 0, "There should initially be 0 unprocessed changes"); this.InsertProducts(1, 5); @@ -625,7 +625,7 @@ public async void LastAccessTimeColumn_Created_OnStartup() this.SetChangeTrackingForTable("Products"); string userFunctionId = "func-id"; IConfiguration configuration = new ConfigurationBuilder().Build(); - var listener = new SqlTriggerListener(this.DbConnectionString, "dbo.Products", userFunctionId, Mock.Of(), Mock.Of(), configuration); + var listener = new SqlTriggerListener(this.DbConnectionString, "dbo.Products", "", userFunctionId, Mock.Of(), Mock.Of(), configuration); await listener.StartAsync(CancellationToken.None); // Cancel immediately so the listener doesn't start processing the changes await listener.StopAsync(CancellationToken.None); @@ -805,5 +805,22 @@ void MonitorOutputData(object sender, DataReceivedEventArgs e) functionHost.OutputDataReceived -= MonitorOutputData; } } + + /// + /// Ensures that the user defined leasesTableName is used to create the leases table. + /// + [Theory] + [SqlInlineData()] + [UnsupportedLanguages(SupportedLanguages.Java)] + public void LeasesTableNameTest(SupportedLanguages lang) + { + this.ExecuteNonQuery("DROP TABLE IF EXISTS [az_func].[Leases]"); + this.SetChangeTrackingForTable("Products"); + int count = (int)this.ExecuteScalar("SELECT COUNT(*) FROM sys.tables WHERE [name] = 'Leases'"); + Assert.Equal(0, count); + this.StartFunctionHost(nameof(ProductsTriggerLeasesTableName), lang); + Thread.Sleep(5000); + Assert.Equal(1, (int)this.ExecuteScalar("SELECT COUNT(*) FROM sys.tables WHERE [name] = 'Leases'")); + } } } \ No newline at end of file diff --git a/test/Unit/TriggerBinding/SqlTriggerBindingProviderTests.cs b/test/Unit/TriggerBinding/SqlTriggerBindingProviderTests.cs index 727a5083a..97cd321f0 100644 --- a/test/Unit/TriggerBinding/SqlTriggerBindingProviderTests.cs +++ b/test/Unit/TriggerBinding/SqlTriggerBindingProviderTests.cs @@ -78,6 +78,18 @@ public async Task TryCreateAsync_ValidTriggerParameterType_ReturnsTriggerBinding Assert.IsType>(binding); } + /// + /// Verifies that is returned if the has all + /// required and optional properties set and it is applied on the trigger parameter of supported type. + /// + [Fact] + public async Task TryCreateAsync_LeasesTableName_ReturnsTriggerBinding() + { + Type parameterType = typeof(IReadOnlyList>); + ITriggerBinding binding = await CreateTriggerBindingAsync(parameterType, nameof(UserFunctionWithLeasesTableName)); + Assert.IsType>(binding); + } + private static async Task CreateTriggerBindingAsync(Type parameterType, string methodName) { var provider = new SqlTriggerBindingProvider( @@ -99,5 +111,7 @@ private static void UserFunctionWithoutAttribute(T _) { } private static void UserFunctionWithoutConnectionString([SqlTrigger("testTableName", null)] T _) { } private static void UserFunctionWithAttribute([SqlTrigger("testTableName", "testConnectionStringSetting")] T _) { } + + private static void UserFunctionWithLeasesTableName([SqlTrigger("testTableName", "testConnectionStringSetting", "testLeasesTableName")] T _) { } } } \ No newline at end of file diff --git a/test/Unit/TriggerBinding/SqlTriggerScaleMonitorTests.cs b/test/Unit/TriggerBinding/SqlTriggerScaleMonitorTests.cs index b7d5db856..c2fb1793e 100644 --- a/test/Unit/TriggerBinding/SqlTriggerScaleMonitorTests.cs +++ b/test/Unit/TriggerBinding/SqlTriggerScaleMonitorTests.cs @@ -226,7 +226,7 @@ public void InvalidUserConfiguredMaxChangesPerWorker(string maxChangesPerWorker) (Mock mockLogger, List logMessages) = CreateMockLogger(); Mock mockConfiguration = CreateMockConfiguration(maxChangesPerWorker); - Assert.Throws(() => new SqlTriggerListener("testConnectionString", "testTableName", "testUserFunctionId", Mock.Of(), mockLogger.Object, mockConfiguration.Object)); + Assert.Throws(() => new SqlTriggerListener("testConnectionString", "testTableName", "", "testUserFunctionId", Mock.Of(), mockLogger.Object, mockConfiguration.Object)); } private static IScaleMonitor GetScaleMonitor(string tableName, string userFunctionId) @@ -234,6 +234,7 @@ private static IScaleMonitor GetScaleMonitor(string tableName return new SqlTriggerScaleMonitor( userFunctionId, new SqlObject(tableName), + "testUserDefinedLeasesTableName", "testConnectionString", SqlTriggerListener.DefaultMaxChangesPerWorker, Mock.Of()); @@ -246,6 +247,7 @@ private static (IScaleMonitor monitor, List logMessag IScaleMonitor monitor = new SqlTriggerScaleMonitor( "testUserFunctionId", new SqlObject("testTableName"), + "testUserDefinedLeasesTableName", "testConnectionString", maxChangesPerWorker, mockLogger.Object);