diff --git a/.github/workflows/dotnet.yml b/.github/workflows/dotnet.yml index fbfafb9..f83f630 100644 --- a/.github/workflows/dotnet.yml +++ b/.github/workflows/dotnet.yml @@ -59,7 +59,9 @@ jobs: - name: Setup .NET uses: actions/setup-dotnet@v6 with: - dotnet-version: 10.0.x + # Pinned while Microsoft's download origin returns 409 for SDK 10.0.403 / runtime 10.0.14 (October 2026). + # Go back to 10.0.x once that release downloads. + dotnet-version: 10.0.402 - name: Setup GitVersion uses: gittools/actions/gitversion/setup@v4 with: diff --git a/Obvs.RabbitMQ.Tests/Hosting/TestRabbitMqTracing.cs b/Obvs.RabbitMQ.Tests/Hosting/TestRabbitMqTracing.cs new file mode 100644 index 0000000..12cfd9e --- /dev/null +++ b/Obvs.RabbitMQ.Tests/Hosting/TestRabbitMqTracing.cs @@ -0,0 +1,104 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Obvs.Hosting; +using Obvs.Serialization.Json; +using Xunit; + +namespace Obvs.RabbitMQ.Tests.Hosting +{ + // Two hosts connected only through RabbitMQ stand in for two processes + public class TestRabbitMqTracing + { + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10); + + [Fact, Trait("Category", "Explicit")] + public async Task ShouldContinueTheSendersTraceAcrossRabbitMq() + { + const string serviceName = "obvs-tracing-commands"; + using var activities = new ActivityRecorder(); + using var receiver = await StartHostAsync(serviceName, service => service.AsServer() + .HandleCommand((command, context, ct) => Task.CompletedTask)); + using var sender = await StartHostAsync(serviceName, service => service.AsClient()); + + await sender.Services.GetRequiredService().SendAsync(new PlaceOrder { OrderId = "o-1" }); + + var send = await WaitForAsync(() => activities.For(serviceName).SingleOrDefault(activity => activity.DisplayName == $"send {serviceName}")); + var process = await WaitForAsync(() => activities.For(serviceName).SingleOrDefault(activity => activity.DisplayName == $"process {serviceName}")); + Assert.Equal(send.TraceId, process.TraceId); + Assert.Equal(send.SpanId, process.ParentSpanId); + } + + [Fact, Trait("Category", "Explicit")] + public async Task ShouldCarryHeadersTheAppSetsOverRabbitMq() + { + const string serviceName = "obvs-tracing-headers"; + var received = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); + using var receiver = await StartHostAsync(serviceName, service => service.AsServer() + .HandleCommand((command, context, ct) => + { + received.TrySetResult(ObvsMessageHeaders.Get(command)); + return Task.CompletedTask; + })); + using var sender = await StartHostAsync(serviceName, service => service.AsClient()); + var placeOrder = new PlaceOrder { OrderId = "o-1" }; + ObvsMessageHeaders.Set(placeOrder, new Dictionary { ["x-tenant"] = "acme" }); + + await sender.Services.GetRequiredService().SendAsync(placeOrder); + + var headers = await received.Task.WaitAsync(Timeout, TestContext.Current.CancellationToken); + Assert.Equal("acme", headers["x-tenant"]); + } + + private static async Task StartHostAsync(string serviceName, Action configure) + { + var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true }); + builder.Services.AddObvs(obvs => obvs.AddService(serviceName, service => configure(service + .UseRabbitMq(options => options.Uri = new Uri(TestBroker.Uri)) + .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)))))); + var host = builder.Build(); + await host.StartAsync(TestContext.Current.CancellationToken); + return host; + } + + 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, TestContext.Current.CancellationToken); + } + return found ?? throw new TimeoutException($"The expected {typeof(T).Name} wasn't recorded in time."); + } + + private 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(); + } + } +} diff --git a/Obvs.RabbitMQ/MessagePublisher.cs b/Obvs.RabbitMQ/MessagePublisher.cs index e1d5bdc..4e5530f 100644 --- a/Obvs.RabbitMQ/MessagePublisher.cs +++ b/Obvs.RabbitMQ/MessagePublisher.cs @@ -1,4 +1,6 @@ using System; +using System.Collections.Generic; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Obvs.Serialization; @@ -31,12 +33,19 @@ public MessagePublisher(Lazy connection, IMessageSerializer seriali public async Task PublishAsync(TMessage message) { var body = _serializer.Serialize(message); - var properties = new BasicProperties {ContentType = "bytes"}; + var properties = new BasicProperties {ContentType = "bytes", Headers = GetHeaders(message)}; var routingKey = GetRoutingKey(message); var channel = await _channel.Value.ConfigureAwait(false); await channel.BasicPublishAsync(_exchange, routingKey, false, properties, body).ConfigureAwait(false); } + // The message's Obvs headers (such as trace context) as AMQP headers + private static IDictionary GetHeaders(TMessage message) + { + var headers = ObvsMessageHeaders.Get(message); + return headers.Count == 0 ? null : headers.ToDictionary(header => header.Key, header => (object)header.Value); + } + private string GetRoutingKey(TMessage message) { return string.Format("{0}.{1}", _routingKeyPrefix, message.GetType().Name); diff --git a/Obvs.RabbitMQ/MessageSource.cs b/Obvs.RabbitMQ/MessageSource.cs index a9aca68..0785f7b 100644 --- a/Obvs.RabbitMQ/MessageSource.cs +++ b/Obvs.RabbitMQ/MessageSource.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Diagnostics; using System.Linq; +using System.Text; using System.Reactive.Concurrency; using System.Reactive.Disposables; using System.Reactive.Linq; @@ -122,7 +123,29 @@ private TMessage Deserialize(BasicDeliverEventArgs deliverEventArgs) { var deserializer = GetDeserializer(deliverEventArgs); // The body's buffer is only valid until the delivery handler returns, so copy it - return deserializer.Deserialize(deliverEventArgs.Body.ToArray()); + var message = deserializer.Deserialize(deliverEventArgs.Body.ToArray()); + SetHeaders(message, deliverEventArgs.BasicProperties.Headers); + return message; + } + + // AMQP string headers arrive as UTF-8 bytes + private static void SetHeaders(TMessage message, IDictionary amqpHeaders) + { + if (amqpHeaders == null || amqpHeaders.Count == 0) + { + return; + } + + var headers = new Dictionary(); + foreach (var header in amqpHeaders) + { + var value = header.Value is byte[] bytes ? Encoding.UTF8.GetString(bytes) : header.Value?.ToString(); + if (value != null) + { + headers[header.Key] = value; + } + } + ObvsMessageHeaders.Set(message, headers); } private IMessageDeserializer GetDeserializer(BasicDeliverEventArgs deliverEventArgs) diff --git a/Obvs.Tests/Hosting/Telemetry/TestTracePropagation.cs b/Obvs.Tests/Hosting/Telemetry/TestTracePropagation.cs new file mode 100644 index 0000000..f58f359 --- /dev/null +++ b/Obvs.Tests/Hosting/Telemetry/TestTracePropagation.cs @@ -0,0 +1,142 @@ +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.Hosting; +using Obvs.Hosting; +using Obvs.Testing; +using Obvs.Tests.Hosting.Handlers; +using Xunit; + +namespace Obvs.Tests.Hosting.Telemetry +{ + public class TestMessageHeaders + { + [Fact] + public void ShouldReturnTheHeadersSetOnAMessage() + { + var message = new PlaceOrder("o-1"); + + ObvsMessageHeaders.Set(message, new Dictionary { ["x-tenant"] = "acme" }); + + Assert.Equal("acme", ObvsMessageHeaders.Get(message)["x-tenant"]); + } + + [Fact] + public void ShouldReturnNoHeadersForAMessageWithoutAny() + { + Assert.Empty(ObvsMessageHeaders.Get(new PlaceOrder("o-1"))); + } + + [Fact] + public void ShouldKeepHeadersSeparateForEqualMessages() + { + // Records with equal values are equal, but headers belong to the instance + var first = new PlaceOrder("o-1"); + var second = new PlaceOrder("o-1"); + + ObvsMessageHeaders.Set(first, new Dictionary { ["x-tenant"] = "acme" }); + + Assert.Empty(ObvsMessageHeaders.Get(second)); + } + } + + // Two hosts sharing an in-memory broker stand in for two processes + public class TestTracePropagation + { + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private readonly string _service = "orders-" + Guid.NewGuid().ToString("N"); + + [Fact] + public async Task ShouldPutTheSendSpansTraceContextInTheMessagesHeaders() + { + using var activities = new ActivityRecorder(); + await using var harness = await ObvsTestHarness.StartAsync(services => services.AddObvs(obvs => obvs + .AddService(_service, service => service.UseInMemory())), cancellationToken: Ct); + var command = new PlaceOrder("o-1"); + + await harness.Client.SendAsync(command); + + var send = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"send {_service}")); + Assert.Equal($"00-{send.TraceId}-{send.SpanId}-01", ObvsMessageHeaders.Get(command)[ObvsMessageHeaders.TraceParent]); + } + + [Fact] + public async Task ShouldKeepHeadersTheAppSetWhenAddingTraceContext() + { + using var activities = new ActivityRecorder(); + await using var harness = await ObvsTestHarness.StartAsync(services => services.AddObvs(obvs => obvs + .AddService(_service, service => service.UseInMemory())), cancellationToken: Ct); + var command = new PlaceOrder("o-1"); + ObvsMessageHeaders.Set(command, new Dictionary { ["x-tenant"] = "acme" }); + + await harness.Client.SendAsync(command); + + var headers = ObvsMessageHeaders.Get(command); + Assert.Equal("acme", headers["x-tenant"]); + Assert.True(headers.ContainsKey(ObvsMessageHeaders.TraceParent)); + } + + [Fact] + public async Task ShouldContinueTheSendersTraceWhenHandling() + { + using var activities = new ActivityRecorder(); + var broker = new InMemoryBroker(); + using var sender = await StartHostAsync(service => service.UseInMemory(broker).AsClient()); + using var receiver = await StartHostAsync(service => service.UseInMemory(broker).AsServer() + .HandleCommand((command, context, ct) => Task.CompletedTask)); + + await sender.Services.GetRequiredService().SendAsync(new PlaceOrder("o-1")); + + var send = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"send {_service}")); + var process = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"process {_service}")); + Assert.Equal(send.TraceId, process.TraceId); + Assert.Equal(send.SpanId, process.ParentSpanId); + } + + [Fact] + public async Task ShouldContinueTheTraceThroughARequestAndItsReply() + { + using var activities = new ActivityRecorder(); + var broker = new InMemoryBroker(); + using var client = await StartHostAsync(service => service.UseInMemory(broker).AsClient()); + using var server = await StartHostAsync(service => service.UseInMemory(broker).AsServer() + .HandleRequest((request, context, ct) => Task.FromResult(new Quote(request.Sku, 9.99m)))); + + await client.Services.GetRequiredService().RequestAsync(new GetQuote("sku-1"), Timeout, Ct); + + var request = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"request {_service}")); + var process = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"process {_service}")); + var reply = await WaitForAsync(() => activities.For(_service).SingleOrDefault(activity => activity.DisplayName == $"reply {_service}")); + Assert.Equal(request.SpanId, process.ParentSpanId); + Assert.Equal(process.SpanId, reply.ParentSpanId); + Assert.Equal(request.TraceId, reply.TraceId); + } + + private async Task StartHostAsync(Action configure) + { + var builder = Host.CreateApplicationBuilder(new HostApplicationBuilderSettings { DisableDefaults = true }); + builder.Services.AddObvs(obvs => obvs.AddService(_service, configure)); + var host = builder.Build(); + await host.StartAsync(Ct); + return host; + } + + 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 d367fb3..fa2d3b6 100644 --- a/Obvs/Hosting/HandlerRunner.cs +++ b/Obvs/Hosting/HandlerRunner.cs @@ -67,7 +67,7 @@ async Task Handle(IMessage message, CancellationToken cancellationToken) 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())) + using (var activity = ObvsDiagnostics.StartProcess(serviceName, message.GetType(), ObvsDiagnostics.ExtractParent(message))) { var scope = scopes.CreateAsyncScope(); try diff --git a/Obvs/Hosting/ObvsDiagnostics.cs b/Obvs/Hosting/ObvsDiagnostics.cs index ad00189..7d65c89 100644 --- a/Obvs/Hosting/ObvsDiagnostics.cs +++ b/Obvs/Hosting/ObvsDiagnostics.cs @@ -39,9 +39,10 @@ public static Activity StartProducer(string operation, string serviceName, Type return activity; } - public static Activity StartProcess(string serviceName, Type messageType) + // A child of the sender's span when the message carries trace context, so the trace continues + public static Activity StartProcess(string serviceName, Type messageType, ActivityContext parent) { - var activity = Source.StartActivity($"process {serviceName}", ActivityKind.Consumer); + var activity = Source.StartActivity($"process {serviceName}", ActivityKind.Consumer, parent); if (activity != null) { activity.SetTag("messaging.system", System); @@ -53,6 +54,41 @@ public static Activity StartProcess(string serviceName, Type messageType) return activity; } + // Adds the span's trace context to the message's headers, keeping any the app set. Falls back to the + // ambient span (an incoming web request, say) when nobody listens to Obvs's own spans. + public static void InjectTraceContext(Activity activity, object message) + { + activity = activity ?? Activity.Current; + if (activity == null) + { + return; + } + + var headers = new Dictionary(); + foreach (var header in ObvsMessageHeaders.Get(message)) + { + headers[header.Key] = header.Value; + } + DistributedContextPropagator.Current.Inject(activity, headers, + (carrier, name, value) => ((Dictionary)carrier)[name] = value); + ObvsMessageHeaders.Set(message, headers); + } + + // The sender's span from the message's headers, if it has trace context + public static ActivityContext ExtractParent(object message) + { + var headers = ObvsMessageHeaders.Get(message); + DistributedContextPropagator.Current.ExtractTraceIdAndState(headers, + (object carrier, string name, out string value, out IEnumerable values) => + { + values = null; + ((IReadOnlyDictionary)carrier).TryGetValue(name, out value); + }, + out var traceParent, out var traceState); + + return ActivityContext.TryParse(traceParent, traceState, out var parent) ? parent : default(ActivityContext); + } + public static void RecordFailure(Activity activity, Exception exception) { if (activity == null) diff --git a/Obvs/Hosting/TelemetryEndpoints.cs b/Obvs/Hosting/TelemetryEndpoints.cs index dd9e82f..e702a01 100644 --- a/Obvs/Hosting/TelemetryEndpoints.cs +++ b/Obvs/Hosting/TelemetryEndpoints.cs @@ -69,6 +69,7 @@ public IObservable GetResponses(IRequest request) { using (var activity = ObvsDiagnostics.StartProducer("request", _serviceName, request.GetType())) { + ObvsDiagnostics.InjectTraceContext(activity, request); Telemetry.LogAndCount("request", _serviceName, request, _logger); try { @@ -96,6 +97,7 @@ public static async Task SendAsync(string operation, string serviceName, IMessag { using (var activity = ObvsDiagnostics.StartProducer(operation, serviceName, message.GetType())) { + ObvsDiagnostics.InjectTraceContext(activity, message); LogAndCount(operation, serviceName, message, logger); try { diff --git a/Obvs/ObvsMessageHeaders.cs b/Obvs/ObvsMessageHeaders.cs new file mode 100644 index 0000000..8157059 --- /dev/null +++ b/Obvs/ObvsMessageHeaders.cs @@ -0,0 +1,62 @@ +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; + +namespace Obvs +{ + /// + /// Headers that travel with a message without being part of it, such as trace context. + /// Transports read a message's headers when sending it and set them on each message they receive. + /// + public static class ObvsMessageHeaders + { + /// The W3C trace-context header holding the sender's trace and span IDs. + public const string TraceParent = "traceparent"; + + /// The W3C trace-context header holding vendor-specific trace state. + public const string TraceState = "tracestate"; + + private static readonly IReadOnlyDictionary None = new Dictionary(); + + // Keyed by the message instance, not its value, and weak, so headers go when their message does + private static readonly ConditionalWeakTable Headers = new ConditionalWeakTable(); + + /// The headers attached to , or an empty set if none are. + public static IReadOnlyDictionary Get(object message) + { + if (message == null) + { + throw new ArgumentNullException(nameof(message)); + } + + return Headers.TryGetValue(message, out var holder) ? holder.Headers : None; + } + + /// Attaches to , replacing any already attached. + public static void Set(object message, IReadOnlyDictionary headers) + { + if (message == null) + { + throw new ArgumentNullException(nameof(message)); + } + if (headers == null) + { + throw new ArgumentNullException(nameof(headers)); + } + + // A copy, so later changes to the caller's dictionary don't change what's sent + var copy = new Dictionary(); + foreach (var header in headers) + { + copy[header.Key] = header.Value; + } + Headers.GetValue(message, _ => new Holder()).Headers = copy; + } + + // Mutable, because ConditionalWeakTable can't replace an entry's value on every target framework + private sealed class Holder + { + public volatile IReadOnlyDictionary Headers = None; + } + } +}