Add more debug logging (#324)

* Add more debug logging

* Move startlistener event

* Move comment

Author: Charles-Gagnon

src/TriggerBinding/SqlTableChangeMonitor.cs

@@ -145,25 +145,27 @@ public void Dispose()
         /// </summary>
         private async Task RunChangeConsumptionLoopAsync()
         {
-            this._logger.LogInformation("Starting change consumption loop.");
+            this._logger.LogDebugWithThreadId("Starting change consumption loop.");
 
             try
             {
                 CancellationToken token = this._cancellationTokenSourceCheckForChanges.Token;
 
                 using (var connection = new SqlConnection(this._connectionString))
                 {
+                    this._logger.LogDebugWithThreadId("BEGIN OpenChangeConsumptionConnection");
                     await connection.OpenAsync(token);
-
+                    this._logger.LogDebugWithThreadId("END OpenChangeConsumptionConnection");
                     // Check for cancellation request only after a cycle of checking and processing of changes completes.
                     while (!token.IsCancellationRequested)
                     {
+                        this._logger.LogDebugWithThreadId($"BEGIN CheckingForChanges State={this._state}");
                         if (this._state == State.CheckingForChanges)
                         {
                             await this.GetTableChangesAsync(connection, token);
                             await this.ProcessTableChangesAsync(connection, token);
                         }
-
+                        this._logger.LogDebugWithThreadId("END CheckingForChanges");
                         await Task.Delay(TimeSpan.FromSeconds(PollingIntervalInSeconds), token);
                     }
                 }
@@ -195,7 +197,7 @@ private async Task RunChangeConsumptionLoopAsync()
         private async Task GetTableChangesAsync(SqlConnection connection, CancellationToken token)
         {
             TelemetryInstance.TrackEvent(TelemetryEventName.GetChangesStart, this._telemetryProps);
-
+            this._logger.LogDebugWithThreadId("BEGIN GetTableChanges");
             try
             {
                 var transactionSw = Stopwatch.StartNew();
@@ -208,14 +210,17 @@ private async Task GetTableChangesAsync(SqlConnection connection, CancellationTo
                         // Update the version number stored in the global state table if necessary before using it.
                         using (SqlCommand updateTablesPreInvocationCommand = this.BuildUpdateTablesPreInvocation(connection, transaction))
                         {
+                            this._logger.LogDebugWithThreadId($"BEGIN UpdateTablesPreInvocation Query={updateTablesPreInvocationCommand.CommandText}");
                             var commandSw = Stopwatch.StartNew();
                             await updateTablesPreInvocationCommand.ExecuteNonQueryAsync(token);
                             setLastSyncVersionDurationMs = commandSw.ElapsedMilliseconds;
                         }
+                        this._logger.LogDebugWithThreadId($"END UpdateTablesPreInvocation Duration={setLastSyncVersionDurationMs}ms");
 
                         // Use the version number to query for new changes.
                         using (SqlCommand getChangesCommand = this.BuildGetChangesCommand(connection, transaction))
                         {
+                            this._logger.LogDebugWithThreadId($"BEGIN GetChanges Query={getChangesCommand.CommandText}");
                             var commandSw = Stopwatch.StartNew();
                             var rows = new List<IReadOnlyDictionary<string, string>>();
 
@@ -230,18 +235,19 @@ private async Task GetTableChangesAsync(SqlConnection connection, CancellationTo
                             this._rows = rows;
                             getChangesDurationMs = commandSw.ElapsedMilliseconds;
                         }
-
-                        this._logger.LogDebug($"Changed rows count: {this._rows.Count}.");
+                        this._logger.LogDebugWithThreadId($"END GetChanges Duration={getChangesDurationMs}ms ChangedRows={this._rows.Count}");
 
                         // If changes were found, acquire leases on them.
                         if (this._rows.Count > 0)
                         {
                             using (SqlCommand acquireLeasesCommand = this.BuildAcquireLeasesCommand(connection, transaction))
                             {
+                                this._logger.LogDebugWithThreadId($"BEGIN AcquireLeases Query={acquireLeasesCommand.CommandText}");
                                 var commandSw = Stopwatch.StartNew();
                                 await acquireLeasesCommand.ExecuteNonQueryAsync(token);
                                 acquireLeasesDurationMs = commandSw.ElapsedMilliseconds;
                             }
+                            this._logger.LogDebugWithThreadId($"END AcquireLeases Duration={acquireLeasesDurationMs}ms");
                         }
 
                         transaction.Commit();
@@ -282,10 +288,12 @@ private async Task GetTableChangesAsync(SqlConnection connection, CancellationTo
                 this._logger.LogError($"Failed to check for changes in table '{this._userTable.FullName}' due to exception: {e.GetType()}. Exception message: {e.Message}");
                 TelemetryInstance.TrackException(TelemetryErrorName.GetChanges, e, this._telemetryProps);
             }
+            this._logger.LogDebugWithThreadId("END GetTableChanges");
         }
 
         private async Task ProcessTableChangesAsync(SqlConnection connection, CancellationToken token)
         {
+            this._logger.LogDebugWithThreadId("BEGIN ProcessTableChanges");
             if (this._rows.Count > 0)
             {
                 this._state = State.ProcessingChanges;
@@ -297,7 +305,7 @@ private async Task ProcessTableChangesAsync(SqlConnection connection, Cancellati
                     // thing. We could still try to trigger on the correctly processed changes, but that adds additional
                     // complication because we don't want to release the leases on the incorrectly processed changes.
                     // For now, just give up I guess?
-                    changes = this.GetChanges();
+                    changes = this.ProcessChanges();
                 }
                 catch (Exception e)
                 {
@@ -311,18 +319,20 @@ private async Task ProcessTableChangesAsync(SqlConnection connection, Cancellati
                     var input = new TriggeredFunctionData() { TriggerValue = changes };
 
                     TelemetryInstance.TrackEvent(TelemetryEventName.TriggerFunctionStart, this._telemetryProps);
+                    this._logger.LogDebugWithThreadId("Executing triggered function");
                     var stopwatch = Stopwatch.StartNew();
 
                     FunctionResult result = await this._executor.TryExecuteAsync(input, this._cancellationTokenSourceExecutor.Token);
-
+                    long durationMs = stopwatch.ElapsedMilliseconds;
                     var measures = new Dictionary<TelemetryMeasureName, double>
                     {
-                        [TelemetryMeasureName.DurationMs] = stopwatch.ElapsedMilliseconds,
+                        [TelemetryMeasureName.DurationMs] = durationMs,
                         [TelemetryMeasureName.BatchCount] = this._rows.Count,
                     };
 
                     if (result.Succeeded)
                     {
+                        this._logger.LogDebugWithThreadId($"Successfully triggered function. Duration={durationMs}ms");
                         TelemetryInstance.TrackEvent(TelemetryEventName.TriggerFunctionEnd, this._telemetryProps, measures);
                         await this.ReleaseLeasesAsync(connection, token);
                     }
@@ -337,6 +347,7 @@ private async Task ProcessTableChangesAsync(SqlConnection connection, Cancellati
                     }
                 }
             }
+            this._logger.LogDebugWithThreadId("END ProcessTableChanges");
         }
 
         /// <summary>
@@ -353,11 +364,14 @@ private async void RunLeaseRenewalLoopAsync()
 
                 using (var connection = new SqlConnection(this._connectionString))
                 {
+                    this._logger.LogDebugWithThreadId("BEGIN OpenLeaseRenewalLoopConnection");
                     await connection.OpenAsync(token);
-
+                    this._logger.LogDebugWithThreadId("END OpenLeaseRenewalLoopConnection");
                     while (!token.IsCancellationRequested)
                     {
+                        this._logger.LogDebugWithThreadId("BEGIN WaitRowsLock - LeaseRenewal");
                         await this._rowsLock.WaitAsync(token);
+                        this._logger.LogDebugWithThreadId("END WaitRowsLock - LeaseRenewal");
                         await this.RenewLeasesAsync(connection, token);
                         await Task.Delay(TimeSpan.FromSeconds(LeaseRenewalIntervalInSeconds), token);
                     }
@@ -391,13 +405,16 @@ private async Task RenewLeasesAsync(SqlConnection connection, CancellationToken
                     using (SqlCommand renewLeasesCommand = this.BuildRenewLeasesCommand(connection))
                     {
                         TelemetryInstance.TrackEvent(TelemetryEventName.RenewLeasesStart, this._telemetryProps);
+                        this._logger.LogDebugWithThreadId($"BEGIN RenewLeases Query={renewLeasesCommand.CommandText}");
                         var stopwatch = Stopwatch.StartNew();
 
                         await renewLeasesCommand.ExecuteNonQueryAsync(token);
 
+                        long durationMs = stopwatch.ElapsedMilliseconds;
+                        this._logger.LogDebugWithThreadId($"END RenewLeases Duration={durationMs}ms");
                         var measures = new Dictionary<TelemetryMeasureName, double>
                         {
-                            [TelemetryMeasureName.DurationMs] = stopwatch.ElapsedMilliseconds,
+                            [TelemetryMeasureName.DurationMs] = durationMs,
                         };
 
                         TelemetryInstance.TrackEvent(TelemetryEventName.RenewLeasesEnd, this._telemetryProps, measures);
@@ -437,6 +454,7 @@ private async Task RenewLeasesAsync(SqlConnection connection, CancellationToken
                 }
 
                 // Want to always release the lock at the end, even if renewing the leases failed.
+                this._logger.LogDebugWithThreadId("ReleaseRowLock - RenewLeases");
                 this._rowsLock.Release();
             }
         }
@@ -449,12 +467,15 @@ private async Task ClearRowsAsync(bool acquireLock)
         {
             if (acquireLock)
             {
+                this._logger.LogDebugWithThreadId("BEGIN WaitRowsLock - ClearRows");
                 await this._rowsLock.WaitAsync();
+                this._logger.LogDebugWithThreadId("END WaitRowsLock - ClearRows");
             }
 
             this._leaseRenewalCount = 0;
             this._state = State.CheckingForChanges;
             this._rows = new List<IReadOnlyDictionary<string, string>>();
+            this._logger.LogDebugWithThreadId("ReleaseRowLock - ClearRows");
             this._rowsLock.Release();
         }
 
@@ -465,9 +486,10 @@ private async Task ClearRowsAsync(bool acquireLock)
         private async Task ReleaseLeasesAsync(SqlConnection connection, CancellationToken token)
         {
             TelemetryInstance.TrackEvent(TelemetryEventName.ReleaseLeasesStart, this._telemetryProps);
-
+            this._logger.LogDebugWithThreadId("BEGIN WaitRowsLock - ReleaseLeases");
             // Don't want to change the "_rows" while another thread is attempting to renew leases on them.
             await this._rowsLock.WaitAsync(token);
+            this._logger.LogDebugWithThreadId("END WaitRowsLock - ReleaseLeases");
             long newLastSyncVersion = this.RecomputeLastSyncVersion();
 
             try
@@ -482,18 +504,22 @@ private async Task ReleaseLeasesAsync(SqlConnection connection, CancellationToke
                         // Release the leases held on "_rows".
                         using (SqlCommand releaseLeasesCommand = this.BuildReleaseLeasesCommand(connection, transaction))
                         {
+                            this._logger.LogDebugWithThreadId($"BEGIN ReleaseLeases Query={releaseLeasesCommand.CommandText}");
                             var commandSw = Stopwatch.StartNew();
                             await releaseLeasesCommand.ExecuteNonQueryAsync(token);
                             releaseLeasesDurationMs = commandSw.ElapsedMilliseconds;
+                            this._logger.LogDebugWithThreadId($"END ReleaseLeases Duration={releaseLeasesDurationMs}ms");
                         }
 
                         // Update the global state table if we have processed all changes with ChangeVersion <= newLastSyncVersion,
                         // and clean up the leases table to remove all rows with ChangeVersion <= newLastSyncVersion.
                         using (SqlCommand updateTablesPostInvocationCommand = this.BuildUpdateTablesPostInvocation(connection, transaction, newLastSyncVersion))
                         {
+                            this._logger.LogDebugWithThreadId($"BEGIN UpdateTablesPostInvocation Query={updateTablesPostInvocationCommand.CommandText}");
                             var commandSw = Stopwatch.StartNew();
                             await updateTablesPostInvocationCommand.ExecuteNonQueryAsync(token);
                             updateLastSyncVersionDurationMs = commandSw.ElapsedMilliseconds;
+                            this._logger.LogDebugWithThreadId($"END UpdateTablesPostInvocation Duration={updateLastSyncVersionDurationMs}ms");
                         }
 
                         transaction.Commit();
@@ -550,10 +576,11 @@ private long RecomputeLastSyncVersion()
                 string changeVersion = row["SYS_CHANGE_VERSION"];
                 changeVersionSet.Add(long.Parse(changeVersion, CultureInfo.InvariantCulture));
             }
-
             // 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.
-            return changeVersionSet.ElementAt(changeVersionSet.Count > 1 ? changeVersionSet.Count - 2 : 0);
+            long lastSyncVersion = changeVersionSet.ElementAt(changeVersionSet.Count > 1 ? changeVersionSet.Count - 2 : 0);
+            this._logger.LogDebugWithThreadId($"RecomputeLastSyncVersion. LastSyncVersion={lastSyncVersion} ChangeVersionSet={string.Join(",", changeVersionSet)}");
+            return lastSyncVersion;
         }
 
         /// <summary>
@@ -562,8 +589,9 @@ private long RecomputeLastSyncVersion()
         /// will be populated with only the primary key values of the deleted row.
         /// </summary>
         /// <returns>The list of changes</returns>
-        private IReadOnlyList<SqlChange<T>> GetChanges()
+        private IReadOnlyList<SqlChange<T>> ProcessChanges()
         {
+            this._logger.LogDebugWithThreadId("BEGIN ProcessChanges");
             var changes = new List<SqlChange<T>>();
             foreach (IReadOnlyDictionary<string, string> row in this._rows)
             {
@@ -577,7 +605,7 @@ private IReadOnlyList<SqlChange<T>> GetChanges()
 
                 changes.Add(new SqlChange<T>(operation, JsonConvert.DeserializeObject<T>(JsonConvert.SerializeObject(item))));
             }
-
+            this._logger.LogDebugWithThreadId("END ProcessChanges");
             return changes;
         }
 
@@ -616,7 +644,7 @@ private SqlCommand BuildUpdateTablesPreInvocation(SqlConnection connection, SqlT
                 SELECT @last_sync_version = LastSyncVersion
                 FROM {SqlTriggerConstants.GlobalStateTableName}
                 WHERE UserFunctionID = '{this._userFunctionId}' AND UserTableID = {this._userTableId};
-                
+
                 IF @last_sync_version < @min_valid_version
                     UPDATE {SqlTriggerConstants.GlobalStateTableName}
                     SET LastSyncVersion = @min_valid_version