Add OpenTelemetry error recording with Activity.AddException polyfill (#446)

* Add OpenTelemetry error recording with Activity.AddException polyfill

Addresses #429 - telemetry activities were not consistently marked as errors when exceptions occurred. This adds a net8 polyfill for the .NET 10 Activity.AddException API, a SetErrorStatus extension that both sets ActivityStatusCode.Error and records exception events per OTel semantic conventions, and consistent error recording across all activity scopes (jobs, queues, work items, messaging, locks).

Co-authored-by: Cursor <cursoragent@cursor.com>

* Remove redundant Activity.Current error status from WorkItemJob

The job runner already sets error status on the parent activity for any non-success, non-cancelled JobResult. This call was setting it on Activity.Current before the child activity was created, duplicating work the runner handles.

Co-authored-by: Cursor <cursoragent@cursor.com>

* PR Feedback

* Apply suggestion from @niemyjski

* PR feedback

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: niemyjski

src/Foundatio.TestHarness/Jobs/HelloWorldJob.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 using Foundatio.Jobs;

src/Foundatio/Jobs/JobResult.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using Microsoft.Extensions.Logging;
 
 namespace Foundatio.Jobs;

src/Foundatio/Jobs/JobWithLockBase.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 using Foundatio.Lock;
@@ -45,7 +45,16 @@ public virtual async Task<JobResult> RunAsync(CancellationToken cancellationToke
         {
             lockActivity?.AddTag("job.id", JobId);
 
-            lockValue = await GetLockAsync(cancellationToken).AnyContext();
+            try
+            {
+                lockValue = await GetLockAsync(cancellationToken).AnyContext();
+            }
+            catch (Exception ex)
+            {
+                lockActivity?.SetErrorStatus(ex);
+                throw;
+            }
+
             if (lockValue is null)
             {
                 return JobResult.CancelledWithMessage("Unable to acquire job lock");

src/Foundatio/Jobs/QueueJobBase.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
@@ -55,9 +55,9 @@ public virtual async Task<JobResult> RunAsync(CancellationToken cancellationToke
         using var linkedCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
         linkedCancellationTokenSource.CancelAfter(TimeSpan.FromSeconds(30));
 
+        using var dequeueActivity = StartDequeueActivity();
         try
         {
-            using var dequeueActivity = StartDequeueActivity();
             queueEntry = await _queue.Value.DequeueAsync(linkedCancellationTokenSource.Token).AnyContext();
             EnrichDequeueActivity(dequeueActivity, queueEntry);
         }
@@ -67,6 +67,7 @@ public virtual async Task<JobResult> RunAsync(CancellationToken cancellationToke
         }
         catch (Exception ex)
         {
+            dequeueActivity?.SetErrorStatus(ex, $"Error trying to dequeue message: {ex.Message}");
             return JobResult.FromException(ex, $"Error trying to dequeue message: {ex.Message}");
         }
 
@@ -132,6 +133,7 @@ public async Task<JobResult> ProcessAsync(IQueueEntry<T> queueEntry, Cancellatio
         }
         catch (Exception ex)
         {
+            activity?.SetErrorStatus(ex);
             _logger.LogError(ex, "Error processing {QueueName} queue entry: {QueueEntryId}", _queueName, queueEntry.Id);
 
             if (!queueEntry.IsCompleted && !queueEntry.IsAbandoned)

src/Foundatio/Jobs/WorkItemJob/WorkItemJob.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Threading;
@@ -93,6 +93,7 @@ public async Task<JobResult> ProcessAsync(IQueueEntry<WorkItemData> queueEntry,
         }
         catch (Exception ex)
         {
+            activity?.SetErrorStatus(ex, $"Abandoning {queueEntry.Value.Type} work item: {queueEntry.Id}: Failed to parse {workItemDataType.Name} work item data");
             await queueEntry.AbandonAsync().AnyContext();
             return JobResult.FromException(ex, $"Abandoning {queueEntry.Value.Type} work item: {queueEntry.Id}: Failed to parse {workItemDataType.Name} work item data");
         }
@@ -101,7 +102,9 @@ public async Task<JobResult> ProcessAsync(IQueueEntry<WorkItemData> queueEntry,
         if (handler == null)
         {
             await queueEntry.CompleteAsync().AnyContext();
-            return JobResult.FailedWithMessage($"Completing {queueEntry.Value.Type} work item: {queueEntry.Id}: Handler for type {workItemDataType.Name} not registered");
+            var result = JobResult.FailedWithMessage($"Completing {queueEntry.Value.Type} work item: {queueEntry.Id}: Handler for type {workItemDataType.Name} not registered");
+            activity?.SetErrorStatus(message: result.Message);
+            return result;
         }
 
         if (queueEntry.Value.SendProgressReports)
@@ -165,6 +168,8 @@ await Task.WhenAll(
         }
         catch (Exception ex)
         {
+            activity?.SetErrorStatus(ex);
+
             if (queueEntry.Value.SendProgressReports)
                 await ReportProgressAsync(handler, queueEntry, -1, $"Failed: {ex.Message}").AnyContext();
 

src/Foundatio/Lock/CacheLockProvider.cs

@@ -183,7 +183,7 @@ public async Task<ILock> AcquireAsync(string resource, TimeSpan? timeUntilExpire
             _lockTimeoutCounter.Add(1);
 
             if (cancellationToken.IsCancellationRequested)
-                _logger.LogTrace("Cancellation requested for lock {Resource} ({LockId}) after {Duration:g}", resource, lockId, sw.Elapsed);
+                _logger.LogDebug("Cancellation requested for lock {Resource} ({LockId}) after {Duration:g}", resource, lockId, sw.Elapsed);
             else
                 _logger.LogWarning("Failed to acquire lock {Resource} ({LockId}) after {Duration:g}", resource, lockId, sw.Elapsed);
 

src/Foundatio/Messaging/MessageBusBase.cs

@@ -274,26 +274,34 @@ protected async Task SendMessageToSubscribersAsync(IMessage message)
                 _logger.LogTrace("Calling subscriber action: {SubscriberId}", subscriber.Id);
                 using var activity = StartHandleMessageActivity(message);
 
-                using (_logger.BeginScope(s => s
-                           .PropertyIf("UniqueId", message.UniqueId, !String.IsNullOrEmpty(message.UniqueId))
-                           .PropertyIf("CorrelationId", message.CorrelationId, !String.IsNullOrEmpty(message.CorrelationId))))
+                try
                 {
-                    if (subscriber.Type == typeof(IMessage))
-                    {
-                        await subscriber.Action(message, subscriber.CancellationToken).AnyContext();
-                    }
-                    else if (subscriber.GenericType != null)
-                    {
-                        object typedMessage = Activator.CreateInstance(subscriber.GenericType, message);
-                        await subscriber.Action(typedMessage, subscriber.CancellationToken).AnyContext();
-                    }
-                    else
+                    using (_logger.BeginScope(s => s
+                               .PropertyIf("UniqueId", message.UniqueId, !String.IsNullOrEmpty(message.UniqueId))
+                               .PropertyIf("CorrelationId", message.CorrelationId, !String.IsNullOrEmpty(message.CorrelationId))))
                     {
-                        await subscriber.Action(message.GetBody(), subscriber.CancellationToken).AnyContext();
+                        if (subscriber.Type == typeof(IMessage))
+                        {
+                            await subscriber.Action(message, subscriber.CancellationToken).AnyContext();
+                        }
+                        else if (subscriber.GenericType != null)
+                        {
+                            object typedMessage = Activator.CreateInstance(subscriber.GenericType, message);
+                            await subscriber.Action(typedMessage, subscriber.CancellationToken).AnyContext();
+                        }
+                        else
+                        {
+                            await subscriber.Action(message.GetBody(), subscriber.CancellationToken).AnyContext();
+                        }
                     }
-                }
 
-                _logger.LogTrace("Finished calling subscriber action: {SubscriberId}", subscriber.Id);
+                    _logger.LogTrace("Finished calling subscriber action: {SubscriberId}", subscriber.Id);
+                }
+                catch (Exception ex)
+                {
+                    activity?.SetErrorStatus(ex);
+                    throw;
+                }
             }, DisposedCancellationToken);
         });
 

src/Foundatio/Queues/QueueBase.cs

@@ -83,14 +83,15 @@ protected QueueBase(TOptions options) : base(options?.TimeProvider, options?.Log
 
             _nextQueueStatsUpdate = _timeProvider.GetUtcNow().Add(_options.MetricsPollingInterval);
             _logger.LogTrace("Getting metrics queue stats for {QueueName} ({QueueId}): Next update scheduled for {NextQueueStatsUpdate:O}", _options.Name, QueueId, _nextQueueStatsUpdate);
+            using var activity = FoundatioDiagnostics.ActivitySource.StartActivity("Queue Stats: " + _options.Name);
             try
             {
-                using var _ = FoundatioDiagnostics.ActivitySource.StartActivity("Queue Stats: " + _options.Name);
                 _queueStats = GetMetricsQueueStats();
                 return (_queueStats.Queued, _queueStats.Working, _queueStats.Deadletter);
             }
             catch (Exception ex)
             {
+                activity?.SetErrorStatus(ex);
                 _logger.LogError(ex, "Error getting queue metrics for {QueueName} ({QueueId}): {Message}", _options.Name, QueueId, ex.Message);
                 return (0, 0, 0);
             }

src/Foundatio/Utility/FoundatioDiagnostics.cs

@@ -1,4 +1,5 @@
-using System.Diagnostics;
+using System;
+using System.Diagnostics;
 using System.Diagnostics.Metrics;
 using System.Reflection;
 
@@ -10,4 +11,62 @@ public static class FoundatioDiagnostics
     internal static readonly string AssemblyVersion = typeof(FoundatioDiagnostics).Assembly.GetCustomAttribute<AssemblyInformationalVersionAttribute>()?.InformationalVersion ?? AssemblyName.Version.ToString();
     public static readonly ActivitySource ActivitySource = new(AssemblyName.Name, AssemblyVersion);
     public static readonly Meter Meter = new("Foundatio", AssemblyVersion);
+
+    /// <summary>
+    /// Sets the activity status to Error and records the exception details.
+    /// </summary>
+    /// <param name="activity">The activity to set error status on.</param>
+    /// <param name="exception">The exception that caused the error (optional).</param>
+    /// <param name="message">A custom error message (optional, defaults to exception message).</param>
+    public static void SetErrorStatus(this Activity activity, Exception exception = null, string message = null)
+    {
+        if (activity is null)
+            return;
+
+        activity.SetStatus(ActivityStatusCode.Error, message ?? exception?.Message);
+
+        if (exception is not null)
+            activity.AddException(exception);
+    }
+}
+
+#if !NET10_0_OR_GREATER
+internal static class ActivityAddExceptionPolyfill
+{
+    private const string ExceptionEventName = "exception";
+    private const string ExceptionMessageTag = "exception.message";
+    private const string ExceptionStackTraceTag = "exception.stacktrace";
+    private const string ExceptionTypeTag = "exception.type";
+
+    /// <summary>
+    /// Polyfill for Activity.AddException available in .NET 10+.
+    /// Add an <see cref="ActivityEvent" /> containing exception information to the <see cref="Activity.Events" /> list.
+    /// </summary>
+    /// <param name="activity">The activity to record the exception on.</param>
+    /// <param name="exception">The exception to add to the attached events list.</param>
+    /// <param name="tags">The tags to add to the exception event.</param>
+    /// <param name="timestamp">The timestamp to add to the exception event.</param>
+    /// <returns><see langword="this" /> for convenient chaining.</returns>
+    public static Activity AddException(this Activity activity, Exception exception, in TagList tags = default, DateTimeOffset timestamp = default)
+    {
+        ArgumentNullException.ThrowIfNull(activity);
+        ArgumentNullException.ThrowIfNull(exception);
+
+        var exceptionTags = new ActivityTagsCollection();
+
+        for (int i = 0; i < tags.Count; i++)
+            exceptionTags.Add(tags[i]);
+
+        if (!exceptionTags.ContainsKey(ExceptionMessageTag))
+            exceptionTags.Add(ExceptionMessageTag, exception.Message);
+
+        if (!exceptionTags.ContainsKey(ExceptionStackTraceTag))
+            exceptionTags.Add(ExceptionStackTraceTag, exception.ToString());
+
+        if (!exceptionTags.ContainsKey(ExceptionTypeTag))
+            exceptionTags.Add(ExceptionTypeTag, exception.GetType().ToString());
+
+        return activity.AddEvent(new ActivityEvent(ExceptionEventName, timestamp, exceptionTags));
+    }
 }
+#endif

tests/Foundatio.Tests/Jobs/JobTests.cs

@@ -204,4 +204,31 @@ public async Task JobLoopPerf()
         await job.RunContinuousAsync(null, iterations, TestCancellationToken);
         sw.Stop();
     }
+
+    [Fact]
+    public async Task RunContinuousAsync_SuccessfulJob_DoesNotSetActivityErrorStatus()
+    {
+        // Arrange
+        Activity capturedActivity = null;
+        using var listener = new ActivityListener
+        {
+            ShouldListenTo = s => s.Name == "Foundatio",
+            Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllDataAndRecorded,
+            ActivityStopped = a =>
+            {
+                if (a.OperationName.StartsWith("Job:"))
+                    capturedActivity = a;
+            }
+        };
+        ActivitySource.AddActivityListener(listener);
+
+        var job = new HelloWorldJob(null, Log);
+
+        // Act
+        await job.RunContinuousAsync(iterationLimit: 1, cancellationToken: TestCancellationToken);
+
+        // Assert
+        Assert.NotNull(capturedActivity);
+        Assert.Equal(ActivityStatusCode.Unset, capturedActivity.Status);
+    }
 }