Add AttemptNumber and OperationId to WorkItemExecuted custom events, improve exception handling/logging (#4514)

<!-- Link the GitHub or AzDO issue this pull request is associated with.
Please copy and paste the full URL rather than using the
dotnet/arcade-services# syntax -->
#4198

---------

Co-authored-by: Přemek Vysoký <[email protected]>

Author: dkurepa

src/Microsoft.DotNet.Darc/DarcLib/TelemetryRecorder.cs

@@ -2,6 +2,7 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 
 using System;
+using System.Collections.Generic;
 
 namespace Microsoft.DotNet.DarcLib;
 
@@ -12,17 +13,27 @@ public enum TrackedGitOperation
     Push,
 }
 
+public enum CustomEventType
+{
+    PullRequestUpdateFailed
+}
+
 public interface ITelemetryRecorder
 {
     /// <summary>
     /// Records work item duration and result in the customEvents table, with the `workItem.{type}` name
     /// </summary>
-    ITelemetryScope RecordWorkItemCompletion(string workItemName);
+    ITelemetryScope RecordWorkItemCompletion(string workItemName, long attemptNumber, string operationId);
 
     /// <summary>
     /// Records git operation duration and result.
     /// </summary>
     ITelemetryScope RecordGitOperation(TrackedGitOperation operation, string repoUri);
+
+    /// <summary>
+    /// Records a custom event with the given name and custom properties.
+    /// </summary>
+    void RecordCustomEvent(CustomEventType customEvent, Dictionary<string, string> customProperties);
 }
 
 public interface ITelemetryScope : IDisposable
@@ -40,8 +51,9 @@ public class NoTelemetryRecorder : ITelemetryRecorder
 {
     private static readonly NoTelemetryScope _instance = new();
 
-    public ITelemetryScope RecordWorkItemCompletion(string workItemName) => _instance;
+    public ITelemetryScope RecordWorkItemCompletion(string workItemName, long attemptNumber, string operationId) => _instance;
     public ITelemetryScope RecordGitOperation(TrackedGitOperation operation, string repoUri) => _instance;
+    public void RecordCustomEvent(CustomEventType eventName, Dictionary<string, string> customProperties) { }
 
     public class NoTelemetryScope : ITelemetryScope
     {

src/ProductConstructionService/ProductConstructionService.Api/Telemetry/TelemetryRecorder.cs

@@ -15,12 +15,22 @@ public class TelemetryRecorder(
     private readonly ILogger<TelemetryRecorder> _logger = logger;
     private readonly TelemetryClient _telemetryClient = telemetryClient;
 
-    public ITelemetryScope RecordWorkItemCompletion(string workItemType)
-        => CreateScope("WorkItemExecuted", new() {{ "WorkItemType", workItemType }});
+    public ITelemetryScope RecordWorkItemCompletion(string workItemType, long attemptNumber, string operationId)
+        => CreateScope("WorkItemExecuted", new()
+            {
+                { "WorkItemType", workItemType },
+                { "Attempt", attemptNumber.ToString()},
+                { "OperationId", operationId }
+            });
 
     public ITelemetryScope RecordGitOperation(TrackedGitOperation operation, string repoUri)
         => CreateScope($"Git{operation}", new() { { "Uri", repoUri } });
 
+    public void RecordCustomEvent(CustomEventType eventName, Dictionary<string, string> customProperties)
+    {
+        _telemetryClient.TrackEvent(eventName.ToString(), customProperties);
+    }
+
     private TelemetryScope CreateScope(string metricName, Dictionary<string, string> customDimensions)
     {
         return new TelemetryScope(metricName, _logger, _telemetryClient, customDimensions, []);

src/ProductConstructionService/ProductConstructionService.DependencyFlow/PullRequestUpdater.cs

@@ -998,15 +998,14 @@ await HandlePrUpdateConflictAsync(
             }
             return;
         }
-        catch (Exception e)
+        catch (Exception)
         {
-            _logger.LogError(e, "Failed to flow source changes for build {buildId} in subscription {subscriptionId}",
+            _logger.LogError("Failed to flow source changes for build {buildId} in subscription {subscriptionId}",
                 build.Id,
                 subscription.Id);
             throw;
         }
 
-
         if (codeFlowRes.hadUpdates)
         {
             _logger.LogInformation("Code changes for {subscriptionId} ready in local branch {branch}",
@@ -1031,18 +1030,8 @@ await HandlePrUpdateConflictAsync(
         }
         else if (pr != null)
         {
-            try
-            {
-                await UpdateCodeFlowPullRequestAsync(update, pr, previousSourceSha, isForwardFlow, subscription, localRepoPath);
-                _logger.LogInformation("Code flow update processed for pull request {prUrl}", pr.Url);
-            }
-            catch (Exception e)
-            {
-                // TODO https://github.com/dotnet/arcade-services/issues/4198: Notify us about these kind of failures
-                _logger.LogError(e, "Failed to update PR {url} of subscription {subscriptionId}",
-                    pr.Url,
-                    update.SubscriptionId);
-            }
+            await UpdateCodeFlowPullRequestAsync(update, pr, previousSourceSha, isForwardFlow, subscription, localRepoPath);
+            _logger.LogInformation("Code flow update processed for pull request {prUrl}", pr.Url);
         }
     }
 
@@ -1077,18 +1066,38 @@ private async Task UpdateCodeFlowPullRequestAsync(
 
         var description = _pullRequestBuilder.GenerateCodeFlowPRDescription(update, build, previousSourceSha, realPR.Description);
 
-        await remote.UpdatePullRequestAsync(pullRequest.Url, new PullRequest
+        try
         {
-            Title = title,
-            Description = description
-        });
-
-        pullRequest.SourceSha = update.SourceSha;
-        pullRequest.LastUpdate = DateTime.UtcNow;
-        pullRequest.MergeState = InProgressPullRequestState.Mergeable;
-        pullRequest.NextBuildsToProcess.Remove(update.SubscriptionId);
-        await SetPullRequestCheckReminder(pullRequest, isCodeFlow: true);
-        await _pullRequestUpdateReminders.UnsetReminderAsync(isCodeFlow: true);
+            await remote.UpdatePullRequestAsync(pullRequest.Url, new PullRequest
+            {
+                Title = title,
+                Description = description
+            });
+        }
+        catch (Exception e)
+        {
+            // If we get here, we already pushed the code updates, but failed to update things like the PR title and description
+            // and enqueue a PullRequestCheck, so we'll just log a custom event for it
+            _telemetryRecorder.RecordCustomEvent(CustomEventType.PullRequestUpdateFailed, new Dictionary<string, string>
+                {
+                    { "SubscriptionId", update.SubscriptionId.ToString() },
+                    { "PullRequestUrl", pullRequest.Url }
+                });
+            _logger.LogError(e, "Failed to update PR {url} of subscription {subscriptionId}",
+                pullRequest.Url,
+                update.SubscriptionId);
+        }
+        finally
+        {
+            // Even if we fail to update the PR title and description, the changes already got pushed, so we want to enqueue a
+            // PullRequestCheck
+            pullRequest.SourceSha = update.SourceSha;
+            pullRequest.LastUpdate = DateTime.UtcNow;
+            pullRequest.MergeState = InProgressPullRequestState.Mergeable;
+            pullRequest.NextBuildsToProcess.Remove(update.SubscriptionId);
+            await SetPullRequestCheckReminder(pullRequest, isCodeFlow: true);
+            await _pullRequestUpdateReminders.UnsetReminderAsync(isCodeFlow: true);
+        }
     }
 
     private async Task CreateCodeFlowPullRequestAsync(
@@ -1147,9 +1156,9 @@ await AddDependencyFlowEventsAsync(
 
             _logger.LogInformation("Code flow pull request created: {prUrl}", prUrl);
         }
-        catch (Exception e)
+        catch (Exception)
         {
-            _logger.LogError(e, "Failed to create code flow pull request for subscription {subscriptionId}",
+            _logger.LogError("Failed to create code flow pull request for subscription {subscriptionId}",
                 update.SubscriptionId);
             await darcRemote.DeleteBranchAsync(targetRepository, prBranch);
             throw;

src/ProductConstructionService/ProductConstructionService.DependencyFlow/WorkItemProcessors/SubscriptionTriggerProcessor.cs

@@ -115,10 +115,10 @@ private async Task<bool> UpdateSubscriptionAsync(Guid subscriptionId, int buildI
                 await triggerer.UpdateSubscriptionAsync(buildId);
                 return true;
             }
-            catch (Exception e)
+            catch (Exception)
             {
-                _logger.LogError(e, "Failed to update subscription '{subscriptionId}' with build '{buildId}'", subscriptionId, buildId);
-                return false;
+                _logger.LogError("Failed to update subscription '{subscriptionId}' with build '{buildId}'", subscriptionId, buildId);
+                throw;
             }
         }
     }

src/ProductConstructionService/ProductConstructionService.WorkItems/WorkItemConsumer.cs

@@ -4,6 +4,9 @@
 using System.Text.Json.Nodes;
 using Azure.Storage.Queues;
 using Azure.Storage.Queues.Models;
+using Microsoft.ApplicationInsights;
+using Microsoft.ApplicationInsights.DataContracts;
+using Microsoft.DotNet.DarcLib;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
@@ -18,7 +21,8 @@ internal class WorkItemConsumer(
         IOptions<WorkItemConsumerOptions> options,
         WorkItemScopeManager scopeManager,
         QueueServiceClient queueServiceClient,
-        IMetricRecorder metricRecorder)
+        IMetricRecorder metricRecorder,
+        ITelemetryRecorder telemetryRecorder)
     : BackgroundService
 {
     private readonly string _consumerId = consumerId;
@@ -93,28 +97,38 @@ private async Task ReadAndProcessWorkItemAsync(QueueClient queueClient, WorkItem
 
         _metricRecorder.QueueMessageReceived(message, delay ?? 0);
 
-        try
-        {
-            _logger.LogInformation("Starting attempt {attemptNumber} for {workItemType}", message.DequeueCount, workItemType);
-            await workItemScope.RunWorkItemAsync(node, cancellationToken);
-            await queueClient.DeleteMessageAsync(message.MessageId, message.PopReceipt, cancellationToken);
-        }
-        // If the cancellation token gets cancelled, don't retry, just exit without deleting the message, we'll handle it later
-        catch (OperationCanceledException)
-        {
-            throw;
-        }
-        catch (Exception ex)
+        TelemetryClient telemetryClient = workItemScope.GetRequiredService<TelemetryClient>();
+
+        using (var operation = telemetryClient.StartOperation<RequestTelemetry>(workItemType))
+        using (ITelemetryScope telemetryScope = telemetryRecorder.RecordWorkItemCompletion(
+            workItemType,
+            message.DequeueCount,
+            operation.Telemetry.Context.Operation.Id))
         {
-            _logger.LogError(ex, "Processing work item {workItemType} attempt {attempt}/{maxAttempts} failed",
-                workItemType, message.DequeueCount, _options.Value.MaxWorkItemRetries);
-            // Let the workItem retry a few times. If it fails a few times, delete it from the queue, it's a bad work item
-            if (message.DequeueCount == _options.Value.MaxWorkItemRetries || ex is NonRetriableException)
+            try
             {
-                _logger.LogError("Work item {type} has failed {maxAttempts} times. Discarding the message {message} from the queue",
-                    workItemType, _options.Value.MaxWorkItemRetries, message.Body.ToString());
+                _logger.LogInformation("Starting attempt {attemptNumber} for {workItemType}", message.DequeueCount, workItemType);
+                await workItemScope.RunWorkItemAsync(node, telemetryScope, cancellationToken);
                 await queueClient.DeleteMessageAsync(message.MessageId, message.PopReceipt, cancellationToken);
             }
+            // If the cancellation token gets cancelled, don't retry, just exit without deleting the message, we'll handle it later
+            catch (OperationCanceledException)
+            {
+                throw;
+            }
+            catch (Exception ex)
+            {
+                operation.Telemetry.Success = false;
+                _logger.LogError(ex, "Processing work item {workItemType} attempt {attempt}/{maxAttempts} failed",
+                    workItemType, message.DequeueCount, _options.Value.MaxWorkItemRetries);
+                // Let the workItem retry a few times. If it fails a few times, delete it from the queue, it's a bad work item
+                if (message.DequeueCount == _options.Value.MaxWorkItemRetries || ex is NonRetriableException)
+                {
+                    _logger.LogError("Work item {type} has failed {maxAttempts} times. Discarding the message {message} from the queue",
+                        workItemType, _options.Value.MaxWorkItemRetries, message.Body.ToString());
+                    await queueClient.DeleteMessageAsync(message.MessageId, message.PopReceipt, cancellationToken);
+                }
+            }
         }
     }
 }

src/ProductConstructionService/ProductConstructionService.WorkItems/WorkItemScope.cs

@@ -38,7 +38,7 @@ public async ValueTask DisposeAsync()
         _serviceScope.Dispose();
     }
 
-    public async Task RunWorkItemAsync(JsonNode node, CancellationToken cancellationToken)
+    public async Task RunWorkItemAsync(JsonNode node, ITelemetryScope telemetryScope, CancellationToken cancellationToken)
     {
         var type = node["type"]!.ToString();
 
@@ -60,29 +60,18 @@ public async Task RunWorkItemAsync(JsonNode node, CancellationToken cancellation
 
         async Task ProcessWorkItemAsync()
         {
-            using (var operation = telemetryClient.StartOperation<RequestTelemetry>(type))
-            using (ITelemetryScope telemetryScope = _telemetryRecorder.RecordWorkItemCompletion(type))
             using (logger.BeginScope(processor.GetLoggingContextData(workItem)))
             {
-                try
+                logger.LogInformation("Processing work item {type}", type);
+                var success = await processor.ProcessWorkItemAsync(workItem, cancellationToken);
+                if (success)
                 {
-                    logger.LogInformation("Processing work item {type}", type);
-                    var success = await processor.ProcessWorkItemAsync(workItem, cancellationToken);
-                    if (success)
-                    {
-                        telemetryScope.SetSuccess();
-                        logger.LogInformation("Work item {type} processed successfully", type);
-                    }
-                    else
-                    {
-                        logger.LogInformation("Work item {type} processed unsuccessfully", type);
-                    }
+                    telemetryScope.SetSuccess();
+                    logger.LogInformation("Work item {type} processed successfully", type);
                 }
-                catch (Exception e)
+                else
                 {
-                    operation.Telemetry.Success = false;
-                    logger.LogError(e, "Failed to process work item {type}", type);
-                    throw;
+                    logger.LogInformation("Work item {type} processed unsuccessfully", type);
                 }
             }
         }
@@ -107,4 +96,9 @@ async Task ProcessWorkItemAsync()
             }
         } while (@lock == null);
     }
+
+    public T GetRequiredService<T>() where T : notnull
+    {
+        return _serviceScope.ServiceProvider.GetRequiredService<T>();
+    }
 }

test/ProductConstructionService.DependencyFlow.Tests/PullRequestUpdaterTests.cs

@@ -598,7 +598,7 @@ protected void ThenUpdateReminderIsRemoved()
     protected IPullRequestUpdater CreatePullRequestActor(IServiceProvider context)
     {
         var updaterFactory = context.GetRequiredService<IPullRequestUpdaterFactory>();
-        return updaterFactory.CreatePullRequestUpdater(GetPullRequestUpdaterId());
+        return updaterFactory.CreatePullRequestUpdater(GetPullRequestUpdaterId()); 
     }
 
     protected SubscriptionUpdateWorkItem CreateSubscriptionUpdate(Build forBuild, bool isCodeFlow = false)