Merge pull request #377 from Azure/chgagnon/syncLA

Sync Logic Apps to latest

Author: Charles-Gagnon

performance/SqlBindingBenchmarks.cs

@@ -11,6 +11,7 @@ public static void Main()
         {
             BenchmarkRunner.Run<SqlInputBindingPerformance>();
             BenchmarkRunner.Run<SqlOutputBindingPerformance>();
+            BenchmarkRunner.Run<SqlTriggerBindingPerformance>();
         }
     }
 }

performance/SqlTriggerBindingPerformance.cs

@@ -0,0 +1,51 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System.Threading.Tasks;
+using Microsoft.Azure.WebJobs.Extensions.Sql.Samples.TriggerBindingSamples;
+using Microsoft.Azure.WebJobs.Extensions.Sql.Tests.Common;
+using Microsoft.Azure.WebJobs.Extensions.Sql.Tests.Integration;
+using BenchmarkDotNet.Attributes;
+
+namespace Microsoft.Azure.WebJobs.Extensions.Sql.Performance
+{
+    public class SqlTriggerBindingPerformance : SqlTriggerBindingIntegrationTests
+    {
+        [GlobalSetup]
+        public void GlobalSetup()
+        {
+            this.EnableChangeTrackingForTable("Products");
+            this.StartFunctionHost(nameof(ProductsTrigger), SupportedLanguages.CSharp);
+        }
+
+        [Benchmark]
+        [Arguments(1)]
+        [Arguments(10)]
+        [Arguments(100)]
+        [Arguments(1000)]
+        public async Task ProductsTriggerTest(int count)
+        {
+            await this.WaitForProductChanges(
+                1,
+                count,
+                SqlChangeOperation.Insert,
+                () => { this.InsertProducts(1, count); return Task.CompletedTask; },
+                id => $"Product {id}",
+                id => id * 100,
+                GetBatchProcessingTimeout(1, count));
+        }
+
+        [IterationCleanup]
+        public void IterationCleanup()
+        {
+            // Delete all rows in Products table after each iteration
+            this.ExecuteNonQuery("TRUNCATE TABLE Products");
+        }
+
+        [GlobalCleanup]
+        public void GlobalCleanup()
+        {
+            this.Dispose();
+        }
+    }
+}

src/Telemetry/Telemetry.cs

@@ -346,6 +346,7 @@ public enum TelemetryEventName
         TableInfoCacheMiss,
         TriggerFunctionEnd,
         TriggerFunctionStart,
+        TriggerMonitorStart,
         UpsertEnd,
         UpsertStart,
     }
@@ -359,6 +360,8 @@ public enum TelemetryPropertyName
         ErrorName,
         ExceptionType,
         HasIdentityColumn,
+        HasConfiguredBatchSize,
+        HasConfiguredPollingInterval,
         LeasesTableName,
         QueryType,
         ServerVersion,
@@ -373,6 +376,7 @@ public enum TelemetryMeasureName
     {
         AcquireLeasesDurationMs,
         BatchCount,
+        BatchSize,
         CommandDurationMs,
         CreatedSchemaDurationMs,
         CreateGlobalStateTableDurationMs,
@@ -383,6 +387,7 @@ public enum TelemetryMeasureName
         GetColumnDefinitionsDurationMs,
         GetPrimaryKeysDurationMs,
         InsertGlobalStateTableRowDurationMs,
+        PollingIntervalMs,
         ReleaseLeasesDurationMs,
         RetryAttemptNumber,
         SetLastSyncVersionDurationMs,

src/TriggerBinding/SqlTableChangeMonitor.cs

@@ -43,6 +43,9 @@ internal sealed class SqlTableChangeMonitor<T> : IDisposable
         private const int LeaseIntervalInSeconds = 60;
         private const int LeaseRenewalIntervalInSeconds = 15;
         private const int MaxRetryReleaseLeases = 3;
+
+        public const int DefaultBatchSize = 100;
+        public const int DefaultPollingIntervalMs = 1000;
         #endregion Constants
 
         private readonly string _connectionString;
@@ -58,11 +61,11 @@ internal sealed class SqlTableChangeMonitor<T> : IDisposable
         /// <summary>
         /// Number of changes to process in each iteration of the loop
         /// </summary>
-        private readonly int _batchSize = 10;
+        private readonly int _batchSize = DefaultBatchSize;
         /// <summary>
         /// Delay in ms between processing each batch of changes
         /// </summary>
-        private readonly int _pollingIntervalInMs = 5000;
+        private readonly int _pollingIntervalInMs = DefaultPollingIntervalMs;
 
         private readonly CancellationTokenSource _cancellationTokenSourceCheckForChanges = new CancellationTokenSource();
         private readonly CancellationTokenSource _cancellationTokenSourceRenewLeases = new CancellationTokenSource();
@@ -117,16 +120,31 @@ public SqlTableChangeMonitor(
             this._logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
             this._userTableId = userTableId;
+            this._telemetryProps = telemetryProps ?? new Dictionary<TelemetryPropertyName, string>();
+
             // Check if there's config settings to override the default batch size/polling interval values
-            this._batchSize = configuration.GetValue<int?>(SqlTriggerConstants.ConfigKey_SqlTrigger_BatchSize) ?? this._batchSize;
-            this._pollingIntervalInMs = configuration.GetValue<int?>(SqlTriggerConstants.ConfigKey_SqlTrigger_PollingInterval) ?? this._pollingIntervalInMs;
+            int? configuredBatchSize = configuration.GetValue<int?>(SqlTriggerConstants.ConfigKey_SqlTrigger_BatchSize);
+            int? configuredPollingInterval = configuration.GetValue<int?>(SqlTriggerConstants.ConfigKey_SqlTrigger_BatchSize);
+            this._batchSize = configuredBatchSize ?? this._batchSize;
+            this._pollingIntervalInMs = configuredPollingInterval ?? this._pollingIntervalInMs;
+            var monitorStartProps = new Dictionary<TelemetryPropertyName, string>(telemetryProps)
+            {
+                { TelemetryPropertyName.HasConfiguredBatchSize, (configuredBatchSize != null).ToString() },
+                { TelemetryPropertyName.HasConfiguredPollingInterval, (configuredPollingInterval != null).ToString() },
+            };
+            TelemetryInstance.TrackEvent(
+                TelemetryEventName.TriggerMonitorStart,
+                monitorStartProps,
+                new Dictionary<TelemetryMeasureName, double>() {
+                    { TelemetryMeasureName.BatchSize, this._batchSize },
+                    { TelemetryMeasureName.PollingIntervalMs, this._pollingIntervalInMs }
+            });
+
             // Prep search-conditions that will be used besides WHERE clause to match table rows.
             this._rowMatchConditions = Enumerable.Range(0, this._batchSize)
                 .Select(rowIndex => string.Join(" AND ", this._primaryKeyColumns.Select((col, colIndex) => $"{col.AsBracketQuotedString()} = @{rowIndex}_{colIndex}")))
                 .ToList();
 
-            this._telemetryProps = telemetryProps ?? new Dictionary<TelemetryPropertyName, string>();
-
 #pragma warning disable CS4014 // Queue the below tasks and exit. Do not wait for their completion.
             _ = Task.Run(() =>
             {
@@ -591,8 +609,8 @@ private long RecomputeLastSyncVersion()
             // have leases acquired on them by another worker.
             // Therefore, if there are more than one version numbers in the set, return the second highest one. Otherwise, return
             // the only version number in the set.
-            // Also this LastSyncVersion is actually updated in the GlobalState table only after verifying that the changes with  
-            // changeVersion <= newLastSyncVersion have been processed in BuildUpdateTablesPostInvocation query. 
+            // Also this LastSyncVersion is actually updated in the GlobalState table only after verifying that the changes with
+            // changeVersion <= newLastSyncVersion have been processed in BuildUpdateTablesPostInvocation query.
             long lastSyncVersion = changeVersionSet.ElementAt(changeVersionSet.Count > 1 ? changeVersionSet.Count - 2 : 0);
             this._logger.LogDebugWithThreadId($"RecomputeLastSyncVersion. LastSyncVersion={lastSyncVersion} ChangeVersionSet={string.Join(",", changeVersionSet)}");
             return lastSyncVersion;

src/TriggerBinding/SqlTriggerConstants.cs

@@ -15,6 +15,17 @@ internal static class SqlTriggerConstants
         public const string LeasesTableAttemptCountColumnName = "_az_func_AttemptCount";
         public const string LeasesTableLeaseExpirationTimeColumnName = "_az_func_LeaseExpirationTime";
 
+        /// <summary>
+        /// The column names that are used in internal state tables and so can't exist in the target table
+        /// since that shares column names with the primary keys from each user table being monitored.
+        /// </summary>
+        public static readonly string[] ReservedColumnNames = new string[]
+        {
+                    LeasesTableChangeVersionColumnName,
+                    LeasesTableAttemptCountColumnName,
+                    LeasesTableLeaseExpirationTimeColumnName
+        };
+
         public const string ConfigKey_SqlTrigger_BatchSize = "Sql_Trigger_BatchSize";
         public const string ConfigKey_SqlTrigger_PollingInterval = "Sql_Trigger_PollingIntervalMs";
     }

src/TriggerBinding/SqlTriggerListener.cs

@@ -257,22 +257,6 @@ FROM sys.indexes AS i
                     throw new InvalidOperationException($"Could not find primary key created in table: '{this._userTable.FullName}'.");
                 }
 
-                string[] reservedColumnNames = new[]
-                {
-                    SqlTriggerConstants.LeasesTableChangeVersionColumnName,
-                    SqlTriggerConstants.LeasesTableAttemptCountColumnName,
-                    SqlTriggerConstants.LeasesTableLeaseExpirationTimeColumnName
-                };
-
-                var conflictingColumnNames = primaryKeyColumns.Select(col => col.name).Intersect(reservedColumnNames).ToList();
-
-                if (conflictingColumnNames.Count > 0)
-                {
-                    string columnNames = string.Join(", ", conflictingColumnNames.Select(col => $"'{col}'"));
-                    throw new InvalidOperationException($"Found reserved column name(s): {columnNames} in table: '{this._userTable.FullName}'." +
-                        " Please rename them to be able to use trigger binding.");
-                }
-
                 this._logger.LogDebugWithThreadId($"END GetPrimaryKeyColumns ColumnNames(types) = {string.Join(", ", primaryKeyColumns.Select(col => $"'{col.name}({col.type})'"))}.");
                 return primaryKeyColumns;
             }
@@ -317,6 +301,15 @@ FROM sys.columns AS c
                     throw new InvalidOperationException($"Found column(s) with unsupported type(s): {columnNamesAndTypes} in table: '{this._userTable.FullName}'.");
                 }
 
+                var conflictingColumnNames = userTableColumns.Intersect(SqlTriggerConstants.ReservedColumnNames).ToList();
+
+                if (conflictingColumnNames.Count > 0)
+                {
+                    string columnNames = string.Join(", ", conflictingColumnNames.Select(col => $"'{col}'"));
+                    throw new InvalidOperationException($"Found reserved column name(s): {columnNames} in table: '{this._userTable.FullName}'." +
+                        " Please rename them to be able to use trigger binding.");
+                }
+
                 this._logger.LogDebugWithThreadId($"END GetUserTableColumns ColumnNames = {string.Join(", ", userTableColumns.Select(col => $"'{col}'"))}.");
                 return userTableColumns;
             }

test/Common/TestUtils.cs

@@ -6,6 +6,7 @@
 using System.Data;
 using System.Diagnostics;
 using System.Threading;
+using System.Threading.Tasks;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Sql.Tests.Common
 {
@@ -159,5 +160,22 @@ public static string CleanJsonString(string jsonStr)
         {
             return jsonStr.Trim().Replace(" ", "").Replace(Environment.NewLine, "");
         }
+
+        public static async Task<TResult> TimeoutAfter<TResult>(this Task<TResult> task, TimeSpan timeout, string message = "The operation has timed out.")
+        {
+
+            using var timeoutCancellationTokenSource = new CancellationTokenSource();
+
+            Task completedTask = await Task.WhenAny(task, Task.Delay(timeout, timeoutCancellationTokenSource.Token));
+            if (completedTask == task)
+            {
+                timeoutCancellationTokenSource.Cancel();
+                return await task;  // Very important in order to propagate exceptions
+            }
+            else
+            {
+                throw new TimeoutException(message);
+            }
+        }
     }
 }

test/Integration/IntegrationTestBase.cs

@@ -18,6 +18,7 @@
 using Microsoft.AspNetCore.WebUtilities;
 using System.Collections.Generic;
 using System.Linq;
+using static Microsoft.Azure.WebJobs.Extensions.Sql.Telemetry.Telemetry;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Sql.Tests.Integration
 {
@@ -199,6 +200,10 @@ protected void StartFunctionHost(string functionName, SupportedLanguages languag
             {
                 environmentVariables.ToList().ForEach(ev => startInfo.EnvironmentVariables[ev.Key] = ev.Value);
             }
+
+            // Always disable telemetry during test runs
+            startInfo.EnvironmentVariables[TelemetryOptoutEnvVar] = "1";
+
             this.LogOutput($"Starting {startInfo.FileName} {startInfo.Arguments} in {startInfo.WorkingDirectory}");
             this.FunctionHost = new Process
             {