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 @@ +