Merge pull request #25 from rebus-org/forward-exceptions-to-activity

Forward exceptions and add a bit more data to traces

Author: mookid8000

Rebus.Diagnostics.Tests/Incoming/HandlerInvokerWrapperTests.cs

@@ -1,4 +1,6 @@
-using System.Diagnostics;
+using System;
+using System.Diagnostics;
+using System.Linq;
 using System.Threading.Tasks;
 using NUnit.Framework;
 using Rebus.Diagnostics.Incoming;
@@ -41,6 +43,40 @@ public async Task CreatesNewSubActivityIfThereIsAnActiveActivity()
             Assert.That(hadActivity);
         }
 
+        [Test]
+        public void MarksActivityAsFailedIfHandlerThrows()
+        {
+            using var activity = new Activity("MyActivity");
+
+            Activity? innerActivity = null;
+            var innerInvoker = new TestInvoker(() =>
+            {
+                innerActivity = Activity.Current;
+                throw new Exception("Look im failing");
+            });
+
+            var wrapper = new HandlerInvokerWrapper(innerInvoker, "MyMessage");
+
+            activity.Start();
+
+            Assume.That(activity, Is.SameAs(Activity.Current));
+
+            Assert.That(async () =>
+            {
+                await wrapper.Invoke();
+            }, Throws.Exception);
+
+            
+            Assert.That(innerActivity, Is.Not.Null);
+
+            Assert.That(innerActivity!.Status, Is.EqualTo(ActivityStatusCode.Error));
+            Assert.That(innerActivity!.StatusDescription, Is.EqualTo("Look im failing"));
+            Assert.That(innerActivity.Events, Has.Exactly(1).Items);
+            var ev = innerActivity.Events.Single();
+            Assert.That(ev.Tags.FirstOrDefault(t => t.Key == "exception.type").Value, Is.EqualTo("System.Exception"));
+            Assert.That(ev.Tags.FirstOrDefault(t => t.Key == "exception.message").Value, Is.EqualTo("Look im failing"));
+        }
+
         [Test]
         public async Task CreatesNoNewActivityIfThereIsntAlreadyAnActiveActivity()
         {

Rebus.Diagnostics.Tests/Incoming/IncomingDiagnosticsStepTests.cs

@@ -331,7 +331,7 @@ private string RenderString(string message, object[] objs)
         /// <summary>
         ///     Formatter function that is invoked for each object value to be rendered into a string while interpolating log lines
         /// </summary>
-        private string FormatObject(object obj, string format)
+        private string FormatObject(object obj, string? format)
         {
             if (obj is string) return $@"""{obj}""";
             if (obj is IEnumerable)
@@ -345,7 +345,7 @@ private string FormatObject(object obj, string format)
             if (obj is DateTimeOffset) return ((DateTimeOffset) obj).ToString(format ?? "O");
             if (obj is IFormattable) return ((IFormattable) obj).ToString(format, CultureInfo.InvariantCulture);
             if (obj is IConvertible) return ((IConvertible) obj).ToString(CultureInfo.InvariantCulture);
-            return obj.ToString();
+            return obj.ToString()!;
         }
 
         #endregion

Rebus.Diagnostics.Tests/Outgoing/OutgoingDiagnosticsStepTests.cs

@@ -15,7 +15,7 @@ namespace Rebus.Diagnostics.Tests.Outgoing
     [TestFixture]
     public class OutgoingDiagnosticsStepTests : FixtureBase
     {
-        RebusTransactionScope? _scope;
+        RebusTransactionScope _scope = null!;
 
         [OneTimeSetUp]
         public static void ListenForRebus()

Rebus.Diagnostics/Diagnostics/Helpers/ActivityExtensions.cs

@@ -0,0 +1,65 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+
+namespace Rebus.Diagnostics.Helpers;
+
+internal static class ActivityExtensions
+{
+    
+    const string ExceptionEventName = "exception";
+    const string ExceptionMessageTag = "exception.message";
+    const string ExceptionStackTraceTag = "exception.stacktrace";
+    const string ExceptionTypeTag = "exception.type";
+
+    
+    // "Borrowed" from https://github.com/tarekgh/runtime/blob/d4464d7ae6d99d75ff89309fb56fd2a5a8f0c845/src/libraries/System.Diagnostics.DiagnosticSource/src/System/Diagnostics/Activity.cs#L535
+    // to avoid requiring version 9 of System.Diagnostics.DiagnosticSource
+    public static Activity AddException(this Activity activity, Exception exception,
+        ActivityTagsCollection? tags = null, DateTimeOffset timestamp = default)
+    {
+        if (exception == null)
+        {
+            throw new ArgumentNullException(nameof(exception));
+        }
+
+        var exceptionTags = tags ?? new();
+
+        var hasMessage = false;
+        var hasStackTrace = false;
+        var hasType = false;
+
+        foreach (var pair in exceptionTags)
+        {
+            if (pair.Key == ExceptionMessageTag)
+            {
+                hasMessage = true;
+            }
+            else if (pair.Key == ExceptionStackTraceTag)
+            {
+                hasStackTrace = true;
+            }
+            else if (pair.Key == ExceptionTypeTag)
+            {
+                hasType = true;
+            }
+        }
+
+        if (!hasMessage)
+        {
+            exceptionTags.Add(new KeyValuePair<string, object?>(ExceptionMessageTag, exception.Message));
+        }
+
+        if (!hasStackTrace)
+        {
+            exceptionTags.Add(new KeyValuePair<string, object?>(ExceptionStackTraceTag, exception.ToString()));
+        }
+
+        if (!hasType)
+        {
+            exceptionTags.Add(new KeyValuePair<string, object?>(ExceptionTypeTag, exception.GetType().ToString()));
+        }
+
+        return activity.AddEvent(new ActivityEvent(ExceptionEventName, timestamp, exceptionTags));
+    }
+}

Rebus.Diagnostics/Diagnostics/Helpers/MessageWrapper.cs

@@ -25,6 +25,11 @@ internal string GetCorrectionId()
     {
         return Headers.GetValueOrNull(Rebus.Messages.Headers.Intent) ?? GetMessageId();
     }
+
+    internal string GetMessageType()
+    {
+        return Headers.GetValueOrNull(Rebus.Messages.Headers.Type) ?? "";
+    }
 }
 
 internal class TransportMessageWrapper : MessageWrapper

Rebus.Diagnostics/Diagnostics/Helpers/TagHelper.cs

@@ -18,6 +18,7 @@ internal static ActivityTagsCollection ExtractInitialTags(MessageWrapper message
         initialTags.Add("messaging.destination_kind", kind);
         initialTags.Add("messaging.message_id", message.GetMessageId());
         initialTags.Add("messaging.conversation_id", message.GetCorrectionId());
+        initialTags.Add("rebus.message.type", message.GetMessageType());
 
         return initialTags;
     }

Rebus.Diagnostics/Diagnostics/Incoming/HandlerInvokerWrapper.cs

@@ -1,4 +1,5 @@
-using System.Diagnostics;
+using System;
+using System.Diagnostics;
 using System.Threading.Tasks;
 using Rebus.Diagnostics.Helpers;
 using Rebus.Pipeline.Receive;
@@ -34,12 +35,22 @@ public override async Task Invoke()
                 initialTags.Add(tag.Key, tag.Value);
         }
         initialTags["messaging.operation"] = "process";
-            
+        initialTags["rebus.handler.type"] = _handlerInvokerImplementation.Handler?.GetType().FullName ?? "Unknown handler type";
+
         using var activity = RebusDiagnosticConstants.ActivitySource.StartActivity($"{_messageType} process", ActivityKind.Internal, parentActivity.Context, initialTags);
 
         TagHelper.CopyBaggage(parentActivity, activity);
-            
-        await _handlerInvokerImplementation.Invoke();
+
+        try
+        {
+            await _handlerInvokerImplementation.Invoke();
+        }
+        catch (Exception e)
+        {
+            activity?.AddException(e);
+            activity?.SetStatus(ActivityStatusCode.Error, e.Message);
+            throw;
+        }
     }
 
     public override void SetSagaData(ISagaData sagaData)