Add exception telemetry recording

Records the standard properties recommended by OpenTelemetry.

Update sample console app with AI reporting using user-secrets locally (plus envvars).

Closes #76

src/Directory.Packages.props

@@ -1,7 +1,9 @@
 <Project>
   <ItemGroup>
     <PackageVersion Include="AutoMapper" Version="12.0.0" />
+    <PackageVersion Include="Azure.Monitor.OpenTelemetry.Exporter" Version="1.0.0-beta.5" />
     <PackageVersion Include="Devlooped.Dynamically" Version="1.0.1" />
+    <PackageVersion Include="Microsoft.Extensions.Configuration.UserSecrets" Version="7.0.0" />
     <PackageVersion Include="Microsoft.SourceLink.GitHub" Version="1.1.1" />
     <PackageVersion Include="Microsoft.TestPlatform.ObjectModel" Version="17.3.2" />
     <PackageVersion Include="NuGetizer" Version="0.9.1" />

src/Merq.Core/MessageBus.cs

@@ -189,15 +189,23 @@ public void Execute(ICommand command)
         var type = GetCommandType(command);
         using var activity = StartActivity(type);
 
-        if (type.IsPublic)
-            // For public types, we can use the faster dynamic dispatch approach
-            ExecuteCore((dynamic)command);
-        else
-            voidExecutors.GetOrAdd(type, type
-                => (VoidDispatcher)Activator.CreateInstance(
-                    typeof(VoidDispatcher<>).MakeGenericType(type),
-                    this))
-            .Execute(command);
+        try
+        {
+            if (type.IsPublic)
+                // For public types, we can use the faster dynamic dispatch approach
+                ExecuteCore((dynamic)command);
+            else
+                voidExecutors.GetOrAdd(type, type
+                    => (VoidDispatcher)Activator.CreateInstance(
+                        typeof(VoidDispatcher<>).MakeGenericType(type),
+                        this))
+                .Execute(command);
+        }
+        catch (Exception e)
+        {
+            activity.RecordException(e);
+            throw;
+        }
     }
 
     /// <summary>
@@ -211,15 +219,23 @@ public TResult Execute<TResult>(ICommand<TResult> command)
         var type = GetCommandType(command);
         using var activity = StartActivity(type);
 
-        if (type.IsPublic)
-            // For public types, we can use the faster dynamic dispatch approach
-            return WithResult<TResult>().Execute((dynamic)command);
+        try
+        {
+            if (type.IsPublic)
+                // For public types, we can use the faster dynamic dispatch approach
+                return WithResult<TResult>().Execute((dynamic)command);
 
-        return (TResult)resultExecutors.GetOrAdd(type, type
-                => (ResultDispatcher)Activator.CreateInstance(
-                   typeof(ResultDispatcher<,>).MakeGenericType(type, typeof(TResult)),
-                   this))
-            .Execute(command)!;
+            return (TResult)resultExecutors.GetOrAdd(type, type
+                    => (ResultDispatcher)Activator.CreateInstance(
+                       typeof(ResultDispatcher<,>).MakeGenericType(type, typeof(TResult)),
+                       this))
+                .Execute(command)!;
+        }
+        catch (Exception e)
+        {
+            activity.RecordException(e);
+            throw;
+        }
     }
 
     /// <summary>
@@ -232,15 +248,23 @@ public Task ExecuteAsync(IAsyncCommand command, CancellationToken cancellation =
         var type = GetCommandType(command);
         using var activity = StartActivity(type);
 
-        if (type.IsPublic)
-            // For public types, we can use the faster dynamic dispatch approach
-            return ExecuteAsyncCore((dynamic)command, cancellation);
+        try
+        {
+            if (type.IsPublic)
+                // For public types, we can use the faster dynamic dispatch approach
+                return ExecuteAsyncCore((dynamic)command, cancellation);
 
-        return voidAsyncExecutors.GetOrAdd(type, type
-            => (VoidAsyncDispatcher)Activator.CreateInstance(
-                typeof(VoidAsyncDispatcher<>).MakeGenericType(type),
-                this))
-            .ExecuteAsync(command, cancellation);
+            return voidAsyncExecutors.GetOrAdd(type, type
+                => (VoidAsyncDispatcher)Activator.CreateInstance(
+                    typeof(VoidAsyncDispatcher<>).MakeGenericType(type),
+                    this))
+                .ExecuteAsync(command, cancellation);
+        }
+        catch (Exception e)
+        {
+            activity.RecordException(e);
+            throw;
+        }
     }
 
     /// <summary>
@@ -255,15 +279,23 @@ public Task<TResult> ExecuteAsync<TResult>(IAsyncCommand<TResult> command, Cance
         var type = GetCommandType(command);
         using var activity = StartActivity(type);
 
-        if (type.IsPublic)
-            // For public types, we can use the faster dynamic dispatch approach
-            return WithResult<TResult>().ExecuteAsync((dynamic)command, cancellation);
+        try
+        {
+            if (type.IsPublic)
+                // For public types, we can use the faster dynamic dispatch approach
+                return WithResult<TResult>().ExecuteAsync((dynamic)command, cancellation);
 
-        return (Task<TResult>)resultAsyncExecutors.GetOrAdd(type, type
-            => (ResultAsyncDispatcher)Activator.CreateInstance(
-                typeof(ResultAsyncDispatcher<,>).MakeGenericType(type, typeof(TResult)),
-                this))
-            .ExecuteAsync(command, cancellation);
+            return (Task<TResult>)resultAsyncExecutors.GetOrAdd(type, type
+                => (ResultAsyncDispatcher)Activator.CreateInstance(
+                    typeof(ResultAsyncDispatcher<,>).MakeGenericType(type, typeof(TResult)),
+                    this))
+                .ExecuteAsync(command, cancellation);
+        }
+        catch (Exception e)
+        {
+            activity.RecordException(e);
+            throw;
+        }
     }
 
     /// <summary>
@@ -294,7 +326,17 @@ public void Notify<TEvent>(TEvent e)
 
         foreach (var subject in compatible)
         {
-            subject.OnNext(e);
+            try
+            {
+                subject.OnNext(e);
+            }
+            catch (Exception ex)
+            {
+                activity.RecordException(ex);
+                // TODO: should we swallow the exception and remove the 
+                // failing subscribers?
+                throw;
+            }
         }
     }
 
@@ -304,7 +346,6 @@ public void Notify<TEvent>(TEvent e)
     public IObservable<TEvent> Observe<TEvent>()
     {
         var eventType = typeof(TEvent);
-        using var activity = StartActivity(eventType, "observe");
 
         // NOTE: in order for the base event subscription to work properly for external
         // producers, they must register the service for each T in the TEvent hierarchy.

src/Merq.Core/Telemetry.cs

@@ -19,4 +19,15 @@ static class Telemetry
             ?.SetTag("messaging.operation", operation)
             ?.SetTag("messaging.protocol", type.Assembly.GetName().Name)
             ?.SetTag("messaging.protocol_version", type.Assembly.GetName().Version?.ToString() ?? "unknown");
+
+    public static void RecordException(this Activity? activity, Exception e)
+    {
+        // See https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/exceptions.md
+        activity?.AddEvent(new ActivityEvent("exception", tags: new()
+        {
+            { "exception.message", e.Message },
+            { "exception.type", e.GetType().FullName },
+            { "exception.stacktrace", e.ToString() },
+        }));
+    }
 }

src/Samples/ConsoleApp/ConsoleApp.csproj

@@ -7,17 +7,18 @@
     <!-- Allow inspection of generated code under obj -->
     <EmitCompilerGeneratedFiles>true</EmitCompilerGeneratedFiles>
     <GenerateDocumentationFile>false</GenerateDocumentationFile>
+    <UserSecretsId>55E4443F-538A-4BBF-898D-26F7E13F8508</UserSecretsId>
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Azure.Monitor.OpenTelemetry.Exporter" />
+    <PackageReference Include="Microsoft.Extensions.Configuration.UserSecrets" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" />
     <PackageReference Include="OpenTelemetry.Exporter.Console" />
     <PackageReference Include="OpenTelemetry.Exporter.Zipkin" />
-    <PackageReference Include="RxFree">
-      <PrivateAssets>all</PrivateAssets>
-      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
-    </PackageReference>
+    <PackageReference Include="RxFree" />
     <PackageReference Include="Spectre.Console" />
+    <PackageReference Include="ThisAssembly.Project" />
   </ItemGroup>
 
   <ItemGroup>
@@ -26,6 +27,10 @@
     <ProjectReference Include="..\Library2\Library2.csproj" Aliases="Library2" />
   </ItemGroup>
 
+  <ItemGroup>
+    <ProjectProperty Include="UserSecretsId" />
+  </ItemGroup>
+
   <!-- Item fixes to support project references vs package references -->
   <ItemGroup Label="All items in this group aren't needed when referencing the nuget packages instead of project references">    
     <!-- Analyzers and code fixes otherwise automatically added by Merq package -->

src/Samples/ConsoleApp/Program.cs

@@ -1,12 +1,21 @@
 extern alias Library1;
 extern alias Library2;
+using System.Diagnostics;
+using Azure.Monitor.OpenTelemetry.Exporter;
 using Merq;
+using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using OpenTelemetry;
 using OpenTelemetry.Resources;
 using OpenTelemetry.Trace;
 using static Spectre.Console.AnsiConsole;
 
+var source = new ActivitySource("ConsoleApp");
+var config = new ConfigurationBuilder()
+    .AddUserSecrets(ThisAssembly.Project.UserSecretsId)
+    .AddEnvironmentVariables()
+    .Build();
+
 // Initialize services
 var collection = new ServiceCollection();
 // Library1 contains [Service]-annotated classes, which will be automatically registered here.
@@ -19,10 +28,12 @@
 using var tracer = Sdk
     .CreateTracerProviderBuilder()
     .SetResourceBuilder(ResourceBuilder.CreateDefault().AddService("ConsoleApp"))
+    .AddSource(source.Name)
     .AddSource("Merq.Core")
     .AddSource("Merq.AutoMapper")
     .AddConsoleExporter()
     .AddZipkinExporter()
+    .AddAzureMonitorTraceExporter(o => o.ConnectionString = config["AppInsights"])
     .Build();
 
 MarkupLine("[yellow]Executing with command from same assembly[/]");
@@ -48,6 +59,16 @@
 
 WriteLine(message);
 
+try
+{
+    using var _ = source.StartActivity("Error");
+    // Showcase error telemetry
+    bus.Execute(new Library1::Library.Echo(""));
+}
+catch (NotSupportedException)
+{
+}
+
 // Test rapid fire messages
 //Parallel.For(0, 10, i 
 //    => bus.Execute(new Library2::Library.Echo($"Hello World ({i})")));

src/Samples/Library1/Commands.cs

@@ -29,7 +29,7 @@ public class EchoHandler : ICommandHandler<Echo, string>
     readonly IMessageBus? bus;
 
     public EchoHandler() { }
-    
+
     public EchoHandler(IMessageBus bus) => this.bus = bus;
 
     public bool CanExecute(Echo command) => !string.IsNullOrEmpty(command.Message);
@@ -38,7 +38,7 @@ public string Execute(Echo command)
     {
         if (string.IsNullOrEmpty(command.Message))
             throw new NotSupportedException("Cannot echo an empty or null message");
-        
+
         bus?.Notify(new OnDidSay(command.Message));
         return command.Message;
     }