Add key metrics to improve observability

kzu · kzu · commit 0595fcc44be8 · 2023-07-07T16:49:20.000-03:00
Adds events/commands metrics, and duration histograms for both processing (commands) and publishing (events).
diff --git a/readme.md b/readme.md
@@ -231,6 +231,41 @@ negatively impacting run-time app startup, via the dependency on
 This also ensures that all proper service interfaces are registered for the various 
 components.
 
+### Telemetry and Monitoring
+
+The core implementation of the `IMessageBus` is instrumented with `ActivitySource` and 
+`Metric`, providing out of the box support for [Open Telemetry](https://learn.microsoft.com/en-us/dotnet/core/diagnostics/distributed-tracing-instrumentation-walkthroughs)-based monitoring, as well 
+as via [dotnet trace](https://learn.microsoft.com/en-us/dotnet/core/diagnostics/dotnet-trace) 
+and [dotnet counters](https://learn.microsoft.com/en-us/dotnet/core/diagnostics/dotnet-counters).
+
+To export telemetry using [Open Telemetry](https://learn.microsoft.com/en-us/dotnet/core/diagnostics/distributed-tracing-instrumentation-walkthroughs), 
+for example:
+
+```csharp
+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();
+```
+
+Collecting traces via [dotnet-trace](https://learn.microsoft.com/en-us/dotnet/core/diagnostics/dotnet-trace):
+
+```shell
+dotnet trace collect --name [PROCESS_NAME] --providers="Microsoft-Diagnostics-DiagnosticSource:::FilterAndPayloadSpecs=[AS]Merq,System.Diagnostics.Metrics:::Metrics=Merq"
+```
+
+Monitoring metrics via [dotnet-counters](https://learn.microsoft.com/en-us/dotnet/core/diagnostics/dotnet-counters):
+
+```shell
+dotnet counters monitor --process-id [PROCESS_ID] --counters Merq
+```
+
 ## Duck Typing Support
 
 <!-- #duck -->
diff --git a/src/Merq.Core/MessageBus.cs b/src/Merq.Core/MessageBus.cs
@@ -1,5 +1,6 @@
 ﻿using System;
 using System.Collections.Concurrent;
+using System.Diagnostics;
 using System.Linq;
 using System.Reactive.Disposables;
 using System.Reactive.Subjects;
@@ -9,6 +10,7 @@
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
 using static Merq.Telemetry;
+using Tag = System.Collections.Generic.KeyValuePair<string, object?>;
 
 namespace Merq;
 
@@ -188,7 +190,7 @@ public bool CanHandle(IExecutable command) => canHandleMap.GetOrAdd(GetCommandTy
     public void Execute(ICommand command)
     {
         var type = GetCommandType(command);
-        using var activity = StartActivity(type, Process);
+        using var activity = StartActivity(type, Telemetry.Process);
 
         try
         {
@@ -220,7 +222,7 @@ public void Execute(ICommand command)
     public TResult Execute<TResult>(ICommand<TResult> command)
     {
         var type = GetCommandType(command);
-        using var activity = StartActivity(type, Process);
+        using var activity = StartActivity(type, Telemetry.Process);
 
         try
         {
@@ -251,7 +253,7 @@ public TResult Execute<TResult>(ICommand<TResult> command)
     public Task ExecuteAsync(IAsyncCommand command, CancellationToken cancellation = default)
     {
         var type = GetCommandType(command);
-        using var activity = StartActivity(type, Process);
+        using var activity = StartActivity(type, Telemetry.Process);
 
         try
         {
@@ -284,7 +286,7 @@ public Task ExecuteAsync(IAsyncCommand command, CancellationToken cancellation =
     public Task<TResult> ExecuteAsync<TResult>(IAsyncCommand<TResult> command, CancellationToken cancellation = default)
     {
         var type = GetCommandType(command);
-        using var activity = StartActivity(type, Process);
+        using var activity = StartActivity(type, Telemetry.Process);
 
         try
         {
@@ -314,41 +316,49 @@ public void Notify<TEvent>(TEvent e)
     {
         var type = (e ?? throw new ArgumentNullException(nameof(e))).GetType();
         using var activity = StartActivity(type, Publish);
+        var watch = Stopwatch.StartNew();
 
-        // TODO: if we prevent Notify for externally produced events, we won't be 
-        // able to notify base event subscribers when those events are produced. 
-        //var producer = services.GetService<IObservable<TEvent>>();
-        //if (producer != null)
-        //    throw new NotSupportedException($"Cannot explicitly notify event {type} because it is externally produced by {producer.GetType()}.");
-
-        // We call all subjects that are compatible with
-        // the event type, not just concrete event type subscribers.
-        // Also adds as compatible the dynamic conversion ones.
-        var compatible = compatibleSubjects.GetOrAdd(type, eventType => subjects.Keys
-            .Where(subjectEventType => subjectEventType.IsAssignableFrom(eventType))
-            .Select(subjectEventType => subjects[subjectEventType])
-            .Concat(dynamicSubjects
-                .GetOrAdd(type.FullName, _ => new())
-                .Where(pair => pair.Key != type && pair.Value != null)
-                .Select(pair => pair.Value!))
-            .ToArray());
-
-        foreach (var subject in compatible)
+        try
         {
-            try
+            // TODO: if we prevent Notify for externally produced events, we won't be 
+            // able to notify base event subscribers when those events are produced. 
+            //var producer = services.GetService<IObservable<TEvent>>();
+            //if (producer != null)
+            //    throw new NotSupportedException($"Cannot explicitly notify event {type} because it is externally produced by {producer.GetType()}.");
+
+            // We call all subjects that are compatible with
+            // the event type, not just concrete event type subscribers.
+            // Also adds as compatible the dynamic conversion ones.
+            var compatible = compatibleSubjects.GetOrAdd(type, eventType => subjects.Keys
+                .Where(subjectEventType => subjectEventType.IsAssignableFrom(eventType))
+                .Select(subjectEventType => subjects[subjectEventType])
+                .Concat(dynamicSubjects
+                    .GetOrAdd(type.FullName, _ => new())
+                    .Where(pair => pair.Key != type && pair.Value != null)
+                    .Select(pair => pair.Value!))
+                .ToArray());
+
+            foreach (var subject in compatible)
             {
-                subject.OnNext(e);
-            }
-            catch (Exception ex)
-            {
-                activity.RecordException(ex);
-                // TODO: should we swallow the exception and remove the 
-                // failing subscribers?
-                // Rethrow original exception to preserve stacktrace.
-                ExceptionDispatchInfo.Capture(ex).Throw();
-                throw;
+                try
+                {
+                    subject.OnNext(e);
+                }
+                catch (Exception ex)
+                {
+                    activity.RecordException(ex);
+                    // TODO: should we swallow the exception and remove the 
+                    // failing subscribers?
+                    // Rethrow original exception to preserve stacktrace.
+                    ExceptionDispatchInfo.Capture(ex).Throw();
+                    throw;
+                }
             }
         }
+        finally
+        {
+            Publishing.Record(watch.ElapsedMilliseconds, new Tag("Event", type.FullName));
+        }
     }
 
     /// <summary>
@@ -487,8 +497,18 @@ void ExecuteCore<TCommand>(TCommand command) where TCommand : ICommand
         var handler = services.GetService<ICommandHandler<TCommand>>();
         if (handler != null)
         {
-            handler.Execute(command);
-            return;
+            var watch = Stopwatch.StartNew();
+            try
+            {
+                handler.Execute(command);
+                return;
+            }
+            finally
+            {
+                Processing.Record(watch.ElapsedMilliseconds,
+                    new Tag("Command", typeof(TCommand).FullName),
+                    new Tag("Handler", handler.GetType().FullName));
+            }
         }
 
         // See if we can convert from the TCommand to a compatible type with 
@@ -509,7 +529,17 @@ Task ExecuteAsyncCore<TCommand>(TCommand command, CancellationToken cancellation
         var handler = services.GetService<IAsyncCommandHandler<TCommand>>();
         if (handler != null)
         {
-            return handler.ExecuteAsync(command, cancellation);
+            var watch = Stopwatch.StartNew();
+            try
+            {
+                return handler.ExecuteAsync(command, cancellation);
+            }
+            finally
+            {
+                Processing.Record(watch.ElapsedMilliseconds,
+                    new Tag("Command", typeof(TCommand).FullName),
+                    new Tag("Handler", handler.GetType().FullName));
+            }
         }
 
         // See if we can convert from the TCommand to a compatible type with 
@@ -528,7 +558,19 @@ TResult ExecuteCore<TCommand, TResult>(TCommand command) where TCommand : IComma
     {
         var handler = services.GetService<ICommandHandler<TCommand, TResult>>();
         if (handler != null)
-            return handler.Execute(command);
+        {
+            var watch = Stopwatch.StartNew();
+            try
+            {
+                return handler.Execute(command);
+            }
+            finally
+            {
+                Processing.Record(watch.ElapsedMilliseconds,
+                    new Tag("Command", typeof(TCommand).FullName),
+                    new Tag("Handler", handler.GetType().FullName));
+            }
+        }
 
         // See if we can convert from the TCommand to a compatible type with 
         // a registered command handler
@@ -546,7 +588,19 @@ Task<TResult> ExecuteAsyncCore<TCommand, TResult>(TCommand command, Cancellation
     {
         var handler = services.GetService<IAsyncCommandHandler<TCommand, TResult>>();
         if (handler != null)
-            return handler.ExecuteAsync(command, cancellation);
+        {
+            var watch = Stopwatch.StartNew();
+            try
+            {
+                return handler.ExecuteAsync(command, cancellation);
+            }
+            finally
+            {
+                Processing.Record(watch.ElapsedMilliseconds,
+                    new Tag("Command", typeof(TCommand).FullName),
+                    new Tag("Handler", handler.GetType().FullName));
+            }
+        }
 
         // See if we can convert from the TCommand to a compatible type with 
         // a registered command handler
diff --git a/src/Merq.Core/Subject.cs b/src/Merq.Core/Subject.cs
@@ -1,5 +1,4 @@
 ﻿using System.Collections.Concurrent;
-using System.Diagnostics;
 using static Merq.Telemetry;
 
 namespace System.Reactive.Subjects;
diff --git a/src/Merq.Core/Telemetry.cs b/src/Merq.Core/Telemetry.cs
@@ -1,12 +1,36 @@
 ﻿using System;
+using System.Collections.Generic;
 using System.Diagnostics;
+using System.Diagnostics.Metrics;
 using System.Runtime.CompilerServices;
 
 namespace Merq;
 
 static class Telemetry
 {
-    static readonly ActivitySource tracer = new(ThisAssembly.Project.AssemblyName, ThisAssembly.Project.Version);
+    static readonly ActivitySource tracer = new("Merq", ThisAssembly.Project.Version);
+    static Meter Meter { get; } = new Meter("Merq", ThisAssembly.Project.Version);
+
+    static readonly Counter<long> commands;
+    static readonly Counter<long> events;
+
+    /// <summary>
+    /// Duration of commands executed by the Merq message bus.
+    /// </summary>
+    public static Histogram<long> Processing { get; } =
+        Meter.CreateHistogram<long>("Processing", unit: "ms", description: "Duration of commands executed by the Merq message bus.");
+
+    /// <summary>
+    /// Duration of event publishing by the Merq message bus.
+    /// </summary>
+    public static Histogram<long> Publishing { get; } =
+        Meter.CreateHistogram<long>("Publishing", unit: "ms", description: "Duration of event publishing by the Merq message bus.");
+
+    static Telemetry()
+    {
+        commands = Meter.CreateCounter<long>("Commands", description: "Commands executed by the Merq message bus.");
+        events = Meter.CreateCounter<long>("Events", description: "Events published to the Merq message bus.");
+    }
 
     // See https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/messaging.md#operation-names
     // publish:	A message is sent to a destination by a message producer/client.
@@ -21,27 +45,37 @@ static class Telemetry
     public const string Process = nameof(Process);
 
     public static Activity? StartActivity(Type type, string operation, [CallerMemberName] string? member = default, [CallerFilePath] string? file = default, [CallerLineNumber] int? line = default)
+    {
+        if (operation == Publish)
+            events.Add(1, new KeyValuePair<string, object?>("Name", type.FullName));
+        else if (operation == Process)
+            commands.Add(1, new KeyValuePair<string, object?>("Name", type.FullName));
+
         // Span name convention should be: <destination> <operation> (see https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/messaging.md#span-name)
         // Requirement is that the destination has low cardinality. In our case, the destination is 
         // the logical operation being performed, such as "Execute", "Notify" or "Deliver". The 
         // operation is actually the type being acted on (such as CreateUser -a command- or UserCreated -event).
-        => tracer.StartActivity(ActivityKind.Producer, name: $"{operation} {type.FullName}")
+        return tracer.StartActivity(ActivityKind.Producer, name: $"{operation}/{type.FullName}")
             ?.SetTag("code.function", member)
             ?.SetTag("code.filepath", file)
             ?.SetTag("code.lineno", line)
             ?.SetTag("messaging.system", "merq")
             ?.SetTag("messaging.destination.name", type.FullName)
             ?.SetTag("messaging.destination.kind", "topic")
-            ?.SetTag("messaging.operation", operation)
+            ?.SetTag("messaging.operation", operation.ToLowerInvariant())
             ?.SetTag("messaging.protocol.name", type.Assembly.GetName().Name)
             ?.SetTag("messaging.protocol.version", type.Assembly.GetName().Version?.ToString() ?? "unknown");
+    }
 
     public static void RecordException(this Activity? activity, Exception e)
     {
         if (activity != null)
         {
             activity.SetStatus(ActivityStatusCode.Error, e.Message);
 
+            activity.SetTag("otel.status_code", "ERROR");
+            activity.SetTag("otel.status_description", e.Message);
+
             // See https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/exceptions.md
             activity.AddEvent(new ActivityEvent("exception", tags: new()
             {
diff --git a/src/Packages.props b/src/Packages.props
@@ -29,6 +29,7 @@
     <PackageReference Update="Devlooped.Extensions.DependencyInjection.Attributed" Version="1.3.2" />
     <PackageReference Update="RxFree" Version="1.1.2" />
     <PackageReference Update="Devlooped.SponsorLink" Version="v0.11.0" />
+    <PackageReference Update="Microsoft.Extensions.Logging.EventSource" Version="6.0.0" />
   </ItemGroup>
   <ItemGroup Label="Tests">
     <PackageReference Update="Microsoft.NET.Test.Sdk" Version="17.6.3" />
diff --git a/src/Samples/ConsoleApp/ConsoleApp.csproj b/src/Samples/ConsoleApp/ConsoleApp.csproj
@@ -15,6 +15,7 @@
     <PackageReference Include="Azure.Monitor.OpenTelemetry.Exporter" />
     <PackageReference Include="Microsoft.Extensions.Configuration.UserSecrets" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" />
+    <PackageReference Include="Microsoft.Extensions.Logging.EventSource" />
     <PackageReference Include="OpenTelemetry.Exporter.Console" />
     <PackageReference Include="OpenTelemetry.Exporter.Zipkin" />
     <PackageReference Include="RxFree" />
diff --git a/src/Samples/ConsoleApp/Program.cs b/src/Samples/ConsoleApp/Program.cs

Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,4 @@`
`1`	`1`	`using System.Collections.Concurrent;`
`2`		`-using System.Diagnostics;`
`3`	`2`	`using static Merq.Telemetry;`
`4`	`3`
`5`	`4`	`namespace System.Reactive.Subjects;`