From a32e800c7f61a6bfc16523e0b80c42e006bfec3e Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 11:45:24 -0700 Subject: [PATCH] OpenTelemetry tracing and metrics, and Debug logging of outgoing messages Obvs now traces under the ActivitySource and measures under the Meter named in ObvsTelemetry ("Obvs"), following OpenTelemetry's messaging conventions with the service as the destination: - send/publish/request/reply Producer spans for outgoing messages, and a process Consumer span for each handler run, with messaging.system=obvs and obvs.message.type; a failing handler's span records the error and exception, and messages a handler sends are child spans of it. - messaging.client.sent.messages counts outgoing messages by operation, and messaging.process.duration measures handler runs, with error.type when they fail. Outgoing messages are logged at Debug through ILogger. The telemetry wraps each service's endpoints, which know the service name, and the handler runner. Trace context across processes needs message headers in every transport, and comes next. Part of #90. Co-Authored-By: Claude Opus 5.5 --- Directory.Packages.props | 1 + .../Hosting/Telemetry/TelemetryRecorders.cs | 77 +++++++ Obvs.Tests/Hosting/Telemetry/TestTelemetry.cs | 192 ++++++++++++++++++ Obvs/Hosting/HandlerRunner.cs | 40 ++-- Obvs/Hosting/ObvsBuilder.cs | 7 +- Obvs/Hosting/ObvsDiagnostics.cs | 93 +++++++++ Obvs/Hosting/ObvsTelemetry.cs | 20 ++ Obvs/Hosting/TelemetryEndpoints.cs | 118 +++++++++++ Obvs/Obvs.csproj | 1 + 9 files changed, 532 insertions(+), 17 deletions(-) create mode 100644 Obvs.Tests/Hosting/Telemetry/TelemetryRecorders.cs create mode 100644 Obvs.Tests/Hosting/Telemetry/TestTelemetry.cs create mode 100644 Obvs/Hosting/ObvsDiagnostics.cs create mode 100644 Obvs/Hosting/ObvsTelemetry.cs create mode 100644 Obvs/Hosting/TelemetryEndpoints.cs diff --git a/Directory.Packages.props b/Directory.Packages.props index 76adacd..14df808 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -24,6 +24,7 @@ + diff --git a/Obvs.Tests/Hosting/Telemetry/TelemetryRecorders.cs b/Obvs.Tests/Hosting/Telemetry/TelemetryRecorders.cs new file mode 100644 index 0000000..732ee15 --- /dev/null +++ b/Obvs.Tests/Hosting/Telemetry/TelemetryRecorders.cs @@ -0,0 +1,77 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.Diagnostics.Metrics; +using System.Linq; +using Obvs.Hosting; + +namespace Obvs.Tests.Hosting.Telemetry +{ + // Listeners are process-wide and tests run in parallel, so each test reads only its own service's telemetry + public sealed class ActivityRecorder : IDisposable + { + private readonly ConcurrentQueue _stopped = new ConcurrentQueue(); + private readonly ActivityListener _listener; + + public ActivityRecorder() + { + _listener = new ActivityListener + { + ShouldListenTo = source => source.Name == ObvsTelemetry.ActivitySourceName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = activity => _stopped.Enqueue(activity), + }; + ActivitySource.AddActivityListener(_listener); + } + + public IReadOnlyList For(string serviceName) + { + return _stopped.Where(activity => (activity.GetTagItem("messaging.destination.name") as string) == serviceName).ToList(); + } + + public void Dispose() => _listener.Dispose(); + } + + public sealed record Measurement(string Instrument, double Value, IReadOnlyDictionary Tags); + + public sealed class MetricRecorder : IDisposable + { + private readonly ConcurrentQueue _measurements = new ConcurrentQueue(); + private readonly MeterListener _listener = new MeterListener(); + + public MetricRecorder() + { + _listener.InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == ObvsTelemetry.MeterName) + { + listener.EnableMeasurementEvents(instrument); + } + }; + _listener.SetMeasurementEventCallback((instrument, value, tags, _) => Record(instrument, value, tags)); + _listener.SetMeasurementEventCallback((instrument, value, tags, _) => Record(instrument, value, tags)); + _listener.Start(); + } + + public IReadOnlyList For(string serviceName, string instrument) + { + return _measurements + .Where(measurement => measurement.Instrument == instrument && + measurement.Tags.TryGetValue("messaging.destination.name", out var service) && (service as string) == serviceName) + .ToList(); + } + + public void Dispose() => _listener.Dispose(); + + private void Record(Instrument instrument, double value, ReadOnlySpan> tags) + { + var tagDictionary = new Dictionary(); + foreach (var tag in tags) + { + tagDictionary[tag.Key] = tag.Value; + } + _measurements.Enqueue(new Measurement(instrument.Name, value, tagDictionary)); + } + } +} diff --git a/Obvs.Tests/Hosting/Telemetry/TestTelemetry.cs b/Obvs.Tests/Hosting/Telemetry/TestTelemetry.cs new file mode 100644 index 0000000..150273a --- /dev/null +++ b/Obvs.Tests/Hosting/Telemetry/TestTelemetry.cs @@ -0,0 +1,192 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Obvs.Hosting; +using Obvs.Testing; +using Obvs.Tests.Hosting.Handlers; +using Xunit; + +namespace Obvs.Tests.Hosting.Telemetry +{ + public class TestTelemetry + { + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + // Unique per test, so a test sees only its own spans, measurements and logs + private readonly string _service = "orders-" + Guid.NewGuid().ToString("N"); + + [Fact] + public async Task ShouldTraceSendingACommand() + { + using var activities = new ActivityRecorder(); + await using var harness = await StartAsync(service => { }); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + + var span = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"send {_service}")); + Assert.Equal(ActivityKind.Producer, span.Kind); + Assert.Equal("obvs", span.GetTagItem("messaging.system")); + Assert.Equal("send", span.GetTagItem("messaging.operation.name")); + Assert.Equal(nameof(PlaceOrder), span.GetTagItem("obvs.message.type")); + } + + [Fact] + public async Task ShouldTracePublishingAnEvent() + { + using var activities = new ActivityRecorder(); + await using var harness = await StartAsync(service => { }); + + await harness.Bus.PublishAsync(new OrderPlaced("o-1")); + + var span = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"publish {_service}")); + Assert.Equal(ActivityKind.Producer, span.Kind); + Assert.Equal("publish", span.GetTagItem("messaging.operation.name")); + Assert.Equal(nameof(OrderPlaced), span.GetTagItem("obvs.message.type")); + } + + [Fact] + public async Task ShouldTraceRequestsAndReplies() + { + using var activities = new ActivityRecorder(); + await using var harness = await StartAsync(service => service + .HandleRequest((request, context, ct) => Task.FromResult(new Quote(request.Sku, 9.99m)))); + + await harness.Client.RequestAsync(new GetQuote("sku-1"), Timeout, Ct); + + var request = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"request {_service}")); + var reply = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"reply {_service}")); + Assert.Equal(nameof(GetQuote), request.GetTagItem("obvs.message.type")); + Assert.Equal(nameof(Quote), reply.GetTagItem("obvs.message.type")); + } + + [Fact] + public async Task ShouldTraceHandlingAMessage() + { + using var activities = new ActivityRecorder(); + await using var harness = await StartAsync(service => service + .HandleCommand((command, context, ct) => Task.CompletedTask)); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + + var span = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"process {_service}")); + Assert.Equal(ActivityKind.Consumer, span.Kind); + Assert.Equal("process", span.GetTagItem("messaging.operation.type")); + Assert.Equal(nameof(PlaceOrder), span.GetTagItem("obvs.message.type")); + Assert.NotEqual(ActivityStatusCode.Error, span.Status); + } + + [Fact] + public async Task ShouldMarkTheHandlingSpanAsFailedWhenAHandlerThrows() + { + using var activities = new ActivityRecorder(); + await using var harness = await StartAsync(service => service + .HandleCommand((command, context, ct) => throw new InvalidOperationException("boom"))); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + + var span = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"process {_service}")); + Assert.Equal(ActivityStatusCode.Error, span.Status); + Assert.Equal(typeof(InvalidOperationException).FullName, span.GetTagItem("error.type")); + Assert.Contains(span.Events, activityEvent => activityEvent.Name == "exception"); + } + + [Fact] + public async Task ShouldMakeMessagesSentByAHandlerChildrenOfItsSpan() + { + using var activities = new ActivityRecorder(); + await using var harness = await StartAsync(service => service + .HandleCommand((command, context, ct) => context.PublishAsync(new OrderPlaced(command.OrderId), ct))); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + + var process = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"process {_service}")); + var publish = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"publish {_service}")); + Assert.Equal(process.TraceId, publish.TraceId); + Assert.Equal(process.SpanId, publish.ParentSpanId); + } + + [Fact] + public async Task ShouldCountSentMessages() + { + using var metrics = new MetricRecorder(); + await using var harness = await StartAsync(service => { }); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + await harness.Client.SendAsync(new PlaceOrder("o-2")); + await harness.Bus.PublishAsync(new OrderPlaced("o-1")); + + var sent = await WaitForAsync(() => + { + var measurements = metrics.For(_service, "messaging.client.sent.messages"); + return measurements.Count == 3 ? measurements : null; + }); + Assert.Equal(2, sent.Where(m => (string)m.Tags["messaging.operation.name"] == "send").Sum(m => m.Value)); + Assert.Equal(1, sent.Where(m => (string)m.Tags["messaging.operation.name"] == "publish").Sum(m => m.Value)); + Assert.All(sent, m => Assert.Equal("obvs", m.Tags["messaging.system"])); + } + + [Fact] + public async Task ShouldMeasureHandlingDuration() + { + using var metrics = new MetricRecorder(); + await using var harness = await StartAsync(service => service + .HandleCommand((command, context, ct) => command.OrderId == "o-2" + ? throw new InvalidOperationException("boom") + : Task.CompletedTask)); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + await harness.Client.SendAsync(new PlaceOrder("o-2")); + + var durations = await WaitForAsync(() => + { + var measurements = metrics.For(_service, "messaging.process.duration"); + return measurements.Count == 2 ? measurements : null; + }); + Assert.All(durations, m => Assert.True(m.Value >= 0)); + Assert.All(durations, m => Assert.Equal(nameof(PlaceOrder), m.Tags["obvs.message.type"])); + Assert.Single(durations, m => !m.Tags.ContainsKey("error.type")); + Assert.Single(durations, m => m.Tags.TryGetValue("error.type", out var errorType) && (string)errorType == typeof(InvalidOperationException).FullName); + } + + [Fact] + public async Task ShouldLogSentMessagesAtDebugLevelOnly() + { + var logs = new ListLoggerProvider(); + await using var harness = await ObvsTestHarness.StartAsync(services => services + .AddLogging(logging => logging.AddProvider(logs).SetMinimumLevel(LogLevel.Debug)) + .AddObvs(obvs => obvs.AddService(_service, service => service.UseInMemory())), + cancellationToken: Ct); + + await harness.Client.SendAsync(new PlaceOrder("o-1")); + + var entry = await WaitForAsync(() => logs.Entries.FirstOrDefault(e => e.Message.Contains(_service) && e.Message.Contains(nameof(PlaceOrder)))); + Assert.Equal(LogLevel.Debug, entry.Level); + Assert.DoesNotContain(logs.Entries, e => e.Level >= LogLevel.Information && e.Message.Contains(_service)); + } + + private Task StartAsync(Action configure) + { + return ObvsTestHarness.StartAsync(services => services.AddObvs(obvs => obvs + .AddService(_service, service => configure(service.UseInMemory()))), + cancellationToken: Ct); + } + + private static async Task WaitForAsync(Func find) where T : class + { + var deadline = DateTime.UtcNow + Timeout; + T found; + while ((found = find()) == null && DateTime.UtcNow < deadline) + { + await Task.Delay(20, Ct); + } + return found ?? throw new TimeoutException($"The expected {typeof(T).Name} wasn't recorded in time."); + } + } +} diff --git a/Obvs/Hosting/HandlerRunner.cs b/Obvs/Hosting/HandlerRunner.cs index 07b74c2..d367fb3 100644 --- a/Obvs/Hosting/HandlerRunner.cs +++ b/Obvs/Hosting/HandlerRunner.cs @@ -1,4 +1,5 @@ using System; +using System.Diagnostics; using System.Linq; using System.Threading; using System.Threading.Channels; @@ -63,22 +64,31 @@ public static HandlerRunner Start(IObservable messages, Func async Task Handle(IMessage message, CancellationToken cancellationToken) { - var scope = scopes.CreateAsyncScope(); - try + var started = Stopwatch.GetTimestamp(); + Exception failure = null; + // The span is current while the handler runs, so anything it sends becomes a child span + using (var activity = ObvsDiagnostics.StartProcess(serviceName, message.GetType())) { - await handler((TMessage)message, new MessageContext(serviceName, scope.ServiceProvider, bus), cancellationToken).ConfigureAwait(false); - } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) - { - // Shutdown ran out of time and cancelled the handler - } - catch (Exception exception) - { - logger.LogError(exception, "Handling {MessageType} on service '{ServiceName}' failed.", message.GetType().Name, serviceName); - } - finally - { - await scope.DisposeAsync().ConfigureAwait(false); + var scope = scopes.CreateAsyncScope(); + try + { + await handler((TMessage)message, new MessageContext(serviceName, scope.ServiceProvider, bus), cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Shutdown ran out of time and cancelled the handler + } + catch (Exception exception) + { + failure = exception; + ObvsDiagnostics.RecordFailure(activity, exception); + logger.LogError(exception, "Handling {MessageType} on service '{ServiceName}' failed.", message.GetType().Name, serviceName); + } + finally + { + await scope.DisposeAsync().ConfigureAwait(false); + ObvsDiagnostics.RecordProcessed(serviceName, message.GetType(), started, failure); + } } } diff --git a/Obvs/Hosting/ObvsBuilder.cs b/Obvs/Hosting/ObvsBuilder.cs index 5883239..7e31798 100644 --- a/Obvs/Hosting/ObvsBuilder.cs +++ b/Obvs/Hosting/ObvsBuilder.cs @@ -2,6 +2,8 @@ using System.Collections.Generic; using System.Linq; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; using Obvs.Configuration; using Obvs.Types; @@ -68,6 +70,7 @@ internal ServiceBus Build(IServiceProvider serviceProvider) { Validate(serviceProvider); + var logger = serviceProvider.GetService()?.CreateLogger("Obvs.Hosting.Messages") ?? NullLogger.Instance; var endpoints = new List>(); var endpointClients = new List>(); foreach (var service in _services) @@ -75,11 +78,11 @@ internal ServiceBus Build(IServiceProvider serviceProvider) var provider = service.EndpointProviderFactory(serviceProvider); if (service.IsServer) { - endpoints.Add(provider.CreateEndpoint()); + endpoints.Add(new TelemetryServiceEndpoint(provider.CreateEndpoint(), service.Name, logger)); } if (service.IsClient) { - endpointClients.Add(provider.CreateEndpointClient()); + endpointClients.Add(new TelemetryServiceEndpointClient(provider.CreateEndpointClient(), service.Name, logger)); } } diff --git a/Obvs/Hosting/ObvsDiagnostics.cs b/Obvs/Hosting/ObvsDiagnostics.cs new file mode 100644 index 0000000..ad00189 --- /dev/null +++ b/Obvs/Hosting/ObvsDiagnostics.cs @@ -0,0 +1,93 @@ +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Diagnostics.Metrics; + +namespace Obvs.Hosting +{ + // Spans and metrics, named after OpenTelemetry's messaging conventions. The service is the destination, + // as that's the name Obvs users work with; broker-level names stay inside the transports. + internal static class ObvsDiagnostics + { + private static readonly string Version = typeof(ObvsDiagnostics).Assembly.GetName().Version?.ToString(); + + public static readonly ActivitySource Source = new ActivitySource(ObvsTelemetry.ActivitySourceName, Version); + + private static readonly Meter Meter = new Meter(ObvsTelemetry.MeterName, Version); + + private static readonly Counter SentMessages = Meter.CreateCounter( + "messaging.client.sent.messages", "{message}", "Messages sent, published, requested or replied."); + + private static readonly Histogram ProcessDuration = Meter.CreateHistogram( + "messaging.process.duration", "s", "How long handlers took to process messages."); + + public const string System = "obvs"; + + // operation is send, publish, request or reply + public static Activity StartProducer(string operation, string serviceName, Type messageType) + { + var activity = Source.StartActivity($"{operation} {serviceName}", ActivityKind.Producer); + if (activity != null) + { + activity.SetTag("messaging.system", System); + // The conventions' type for every outgoing message; the name says which Obvs operation it was + activity.SetTag("messaging.operation.type", "send"); + activity.SetTag("messaging.operation.name", operation); + activity.SetTag("messaging.destination.name", serviceName); + activity.SetTag("obvs.message.type", messageType.Name); + } + return activity; + } + + public static Activity StartProcess(string serviceName, Type messageType) + { + var activity = Source.StartActivity($"process {serviceName}", ActivityKind.Consumer); + if (activity != null) + { + activity.SetTag("messaging.system", System); + activity.SetTag("messaging.operation.type", "process"); + activity.SetTag("messaging.operation.name", "process"); + activity.SetTag("messaging.destination.name", serviceName); + activity.SetTag("obvs.message.type", messageType.Name); + } + return activity; + } + + public static void RecordFailure(Activity activity, Exception exception) + { + if (activity == null) + { + return; + } + activity.SetStatus(ActivityStatusCode.Error, exception.Message); + activity.SetTag("error.type", exception.GetType().FullName); + activity.AddException(exception); + } + + public static void RecordSent(string operation, string serviceName, Type messageType) + { + SentMessages.Add(1, + new KeyValuePair("messaging.system", System), + new KeyValuePair("messaging.operation.name", operation), + new KeyValuePair("messaging.destination.name", serviceName), + new KeyValuePair("obvs.message.type", messageType.Name)); + } + + public static void RecordProcessed(string serviceName, Type messageType, long startTimestamp, Exception failure) + { + var seconds = (Stopwatch.GetTimestamp() - startTimestamp) / (double)Stopwatch.Frequency; + var tags = new TagList + { + { "messaging.system", System }, + { "messaging.operation.name", "process" }, + { "messaging.destination.name", serviceName }, + { "obvs.message.type", messageType.Name }, + }; + if (failure != null) + { + tags.Add("error.type", failure.GetType().FullName); + } + ProcessDuration.Record(seconds, tags); + } + } +} diff --git a/Obvs/Hosting/ObvsTelemetry.cs b/Obvs/Hosting/ObvsTelemetry.cs new file mode 100644 index 0000000..e257b9d --- /dev/null +++ b/Obvs/Hosting/ObvsTelemetry.cs @@ -0,0 +1,20 @@ +namespace Obvs.Hosting +{ + /// + /// The names Obvs traces and measures under. Add them to OpenTelemetry to collect them: + /// tracing.AddSource(ObvsTelemetry.ActivitySourceName) and metrics.AddMeter(ObvsTelemetry.MeterName). + /// + public static class ObvsTelemetry + { + /// + /// The ActivitySource for spans: send, publish, request and reply for + /// outgoing messages, and process for each handler run, following OpenTelemetry's messaging conventions. + /// + public const string ActivitySourceName = "Obvs"; + + /// + /// The Meter for metrics: messaging.client.sent.messages and messaging.process.duration. + /// + public const string MeterName = "Obvs"; + } +} diff --git a/Obvs/Hosting/TelemetryEndpoints.cs b/Obvs/Hosting/TelemetryEndpoints.cs new file mode 100644 index 0000000..dd9e82f --- /dev/null +++ b/Obvs/Hosting/TelemetryEndpoints.cs @@ -0,0 +1,118 @@ +using System; +using System.Reactive.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using Obvs.Types; + +namespace Obvs.Hosting +{ + // Wraps a service's endpoints, which know which service a message belongs to, to trace, count and + // log (at Debug) every outgoing message + internal sealed class TelemetryServiceEndpoint : IServiceEndpoint + { + private readonly IServiceEndpoint _inner; + private readonly string _serviceName; + private readonly ILogger _logger; + + public TelemetryServiceEndpoint(IServiceEndpoint inner, string serviceName, ILogger logger) + { + _inner = inner; + _serviceName = serviceName; + _logger = logger; + } + + public IObservable Requests => _inner.Requests; + + public IObservable Commands => _inner.Commands; + + public Task PublishAsync(IEvent ev) + { + return Telemetry.SendAsync("publish", _serviceName, ev, _logger, () => _inner.PublishAsync(ev)); + } + + public Task ReplyAsync(IRequest request, IResponse response) + { + return Telemetry.SendAsync("reply", _serviceName, response, _logger, () => _inner.ReplyAsync(request, response)); + } + + public bool CanHandle(IMessage message) => _inner.CanHandle(message); + + public string Name => _inner.Name; + + public void Dispose() => _inner.Dispose(); + } + + internal sealed class TelemetryServiceEndpointClient : IServiceEndpointClient + { + private readonly IServiceEndpointClient _inner; + private readonly string _serviceName; + private readonly ILogger _logger; + + public TelemetryServiceEndpointClient(IServiceEndpointClient inner, string serviceName, ILogger logger) + { + _inner = inner; + _serviceName = serviceName; + _logger = logger; + } + + public IObservable Events => _inner.Events; + + public Task SendAsync(ICommand command) + { + return Telemetry.SendAsync("send", _serviceName, command, _logger, () => _inner.SendAsync(command)); + } + + // The inner endpoint sends the request when subscribed to, so the span covers the subscription + public IObservable GetResponses(IRequest request) + { + return Observable.Create(observer => + { + using (var activity = ObvsDiagnostics.StartProducer("request", _serviceName, request.GetType())) + { + Telemetry.LogAndCount("request", _serviceName, request, _logger); + try + { + return _inner.GetResponses(request).Subscribe(observer); + } + catch (Exception exception) + { + ObvsDiagnostics.RecordFailure(activity, exception); + throw; + } + } + }); + } + + public bool CanHandle(IMessage message) => _inner.CanHandle(message); + + public string Name => _inner.Name; + + public void Dispose() => _inner.Dispose(); + } + + internal static class Telemetry + { + public static async Task SendAsync(string operation, string serviceName, IMessage message, ILogger logger, Func send) + { + using (var activity = ObvsDiagnostics.StartProducer(operation, serviceName, message.GetType())) + { + LogAndCount(operation, serviceName, message, logger); + try + { + await send().ConfigureAwait(false); + } + catch (Exception exception) + { + ObvsDiagnostics.RecordFailure(activity, exception); + throw; + } + } + } + + public static void LogAndCount(string operation, string serviceName, IMessage message, ILogger logger) + { + ObvsDiagnostics.RecordSent(operation, serviceName, message.GetType()); + logger.LogDebug("Obvs {Operation} {MessageType} on service '{ServiceName}'.", operation, message.GetType().Name, serviceName); + } + } +} diff --git a/Obvs/Obvs.csproj b/Obvs/Obvs.csproj index 6f5c8e6..27d8fab 100644 --- a/Obvs/Obvs.csproj +++ b/Obvs/Obvs.csproj @@ -15,6 +15,7 @@ +