From f03dec2e0be23a9800d1fc83096e7b05390c33c6 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 11:55:24 -0700 Subject: [PATCH 1/2] Carry trace context in message headers so traces connect across processes ObvsMessageHeaders.Get/Set attach headers to a message instance without changing its type or any transport interface; they're held in a weak table, so they go when the message does. Sending adds the producer span's W3C trace context (traceparent, tracestate) to the outgoing message's headers, keeping any the app set, or the ambient span's when nobody listens to Obvs's spans. A handler's process span becomes a child of the span in the message's traceparent, so the receiving service continues the sender's trace. RabbitMQ sends a message's headers as AMQP headers and sets them on each message it receives. The in-memory transport passes the sender's object to the receiver, so headers already travel with it. Kafka, NATS, ActiveMQ and NetMQ will get header support with their Use() increments. Part of #90. Co-Authored-By: Claude Opus 5.5 --- .../Hosting/TestRabbitMqTracing.cs | 104 +++++++++++++ Obvs.RabbitMQ/MessagePublisher.cs | 11 +- Obvs.RabbitMQ/MessageSource.cs | 25 ++- .../Hosting/Telemetry/TestTracePropagation.cs | 142 ++++++++++++++++++ Obvs/Hosting/HandlerRunner.cs | 2 +- Obvs/Hosting/ObvsDiagnostics.cs | 40 ++++- Obvs/Hosting/TelemetryEndpoints.cs | 2 + Obvs/ObvsMessageHeaders.cs | 62 ++++++++ 8 files changed, 383 insertions(+), 5 deletions(-) create mode 100644 Obvs.RabbitMQ.Tests/Hosting/TestRabbitMqTracing.cs create mode 100644 Obvs.Tests/Hosting/Telemetry/TestTracePropagation.cs create mode 100644 Obvs/ObvsMessageHeaders.cs 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; + } + } +} From 44e096aaf5a1805888df47a033a9851119454b06 Mon Sep 17 00:00:00 2001 From: Christopher Read <1907605+christopherread@users.noreply.github.com> Date: Sun, 25 Oct 2026 12:17:52 -0700 Subject: [PATCH 2/2] CI: pin the .NET SDK to 10.0.402 Microsoft's download origin returns 409 for the new SDK 10.0.403 and runtime 10.0.14, so setup-dotnet fails to install 10.0.x. Go back to 10.0.x once that release downloads. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/dotnet.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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: