From 45faeab8866e25ad60e6022a31324f82c745da60 Mon Sep 17 00:00:00 2001 From: Pavel Borisov Date: Fri, 14 Aug 2026 00:28:23 +0200 Subject: [PATCH] fix(subscriptions): link to the restored trace context, don't parent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The consume span was made a child of the append that produced the event, using the tracing context restored from the event's metadata. That context is durable data on the event, so every redelivery re-joined the same trace, with no live root to end it and no sampling decision left to take. A hot resubscribe loop grew one trace to 251MB and OOM-killed a shared Tempo twice. The consume span now roots its own trace and links to the restored context. Tags are unchanged. Parent-child is kept where the causality is in-process: an ambient Eventuous activity, or the activity stored for the async path. The SignalR client had the same bug and is fixed alongside. Rooting takes two steps, because .NET has no parentless argument and binds the parent from Activity.Current at Start() — which on the async path happens far down the pipe. Activity.Current is suppressed at creation, and the span is pinned to a zero parent span id so a late Start cannot re-parent it. The trace id is the sampler's own where it read one (only ratio-based samplers do) and generated otherwise, since an unstarted activity reports all zeros. Three defects found alongside, each wrong on its own: - ToActivityContext returned a zeroed context instead of null, so both callers' null checks never fired. - AddTracingMeta compared a 32-char trace id to a 16-char constant, so its empty-id guard never fired and all-zero contexts were written to events. - AddEventuousTracing called SetSampler, replacing the application's sampler for the whole provider; its drop branch matched no activity that exists, so it only ever returned RecordAndSample. Co-Authored-By: Claude Opus 5 (1M context) --- .../MetadataExtensions.cs | 6 +- .../Eventuous.Diagnostics/Tags/TracingMeta.cs | 26 +- .../Diagnostics/SubscriptionActivity.cs | 81 +++++- .../SubscriptionActivityTests.cs | 252 ++++++++++++++++++ .../Eventuous.Tests/TracingMetadataTests.cs | 72 +++++ .../TracerProviderBuilderExtensions.cs | 14 +- .../ProducerTracesTests.cs | 40 ++- .../Eventuous.SignalR.Client.csproj | 3 + .../TypedStreamSubscription.cs | 35 ++- .../ConsumeTracingTests.cs | 74 +++++ 10 files changed, 551 insertions(+), 52 deletions(-) create mode 100644 src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionActivityTests.cs create mode 100644 src/Core/test/Eventuous.Tests/TracingMetadataTests.cs create mode 100644 src/SignalR/test/Eventuous.Tests.SignalR/ConsumeTracingTests.cs diff --git a/src/Core/src/Eventuous.Diagnostics/MetadataExtensions.cs b/src/Core/src/Eventuous.Diagnostics/MetadataExtensions.cs index ed473967e..4eecef9f2 100644 --- a/src/Core/src/Eventuous.Diagnostics/MetadataExtensions.cs +++ b/src/Core/src/Eventuous.Diagnostics/MetadataExtensions.cs @@ -21,8 +21,8 @@ public Metadata AddActivityTags(Activity? activity) { [MethodImpl(MethodImplOptions.AggressiveInlining)] Metadata AddTracingMeta(TracingMeta tracingMeta) - => metadata.ContainsKey(TraceId) || tracingMeta.TraceId == EmptyId - ? metadata // don't override existing tracing data + => metadata.ContainsKey(TraceId) || !tracingMeta.IsValid() + ? metadata // don't override existing tracing data, and don't persist an unusable one : metadata .AddNotNull(TraceId, tracingMeta.TraceId) .AddNotNull(SpanId, tracingMeta.SpanId); @@ -31,6 +31,4 @@ Metadata AddTracingMeta(TracingMeta tracingMeta) public TracingMeta GetTracingMeta() => new(metadata.GetString(TraceId), metadata.GetString(SpanId)); } - - const string EmptyId = "0000000000000000"; } diff --git a/src/Core/src/Eventuous.Diagnostics/Tags/TracingMeta.cs b/src/Core/src/Eventuous.Diagnostics/Tags/TracingMeta.cs index 79860d3b1..e34b43f3d 100644 --- a/src/Core/src/Eventuous.Diagnostics/Tags/TracingMeta.cs +++ b/src/Core/src/Eventuous.Diagnostics/Tags/TracingMeta.cs @@ -4,20 +4,34 @@ namespace Eventuous.Diagnostics; public record TracingMeta(string? TraceId, string? SpanId) { - bool IsValid() => TraceId != null && SpanId != null; + /// + /// Whether the metadata carries a tracing context that can be used: both ids present and neither all-zero. + /// An all-zero id is what an activity reports when it was never given a real one, and persisting or + /// restoring that is worse than having nothing, so both ends of the round trip check it here. + /// + internal bool IsValid() => TraceId is not (null or EmptyTraceId) && SpanId is not (null or EmptySpanId); + /// + /// Restores an activity context from the persisted tracing metadata, or null when the metadata carries + /// no usable context. Callers check for null to decide whether there is anything to correlate with, so an + /// absent or all-zero context must not come back as a zeroed . + /// public ActivityContext? ToActivityContext(bool isRemote) { try { - return IsValid() ? - new ActivityContext( + return IsValid() + ? new ActivityContext( ActivityTraceId.CreateFromString(TraceId), ActivitySpanId.CreateFromString(SpanId), ActivityTraceFlags.Recorded, isRemote: isRemote - ) : default; + ) + : null; } catch (Exception) { - return default; + return null; } } -} \ No newline at end of file + + const string EmptyTraceId = "00000000000000000000000000000000"; + const string EmptySpanId = "0000000000000000"; +} diff --git a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionActivity.cs b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionActivity.cs index 53dde9ca9..e1262cac7 100644 --- a/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionActivity.cs +++ b/src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionActivity.cs @@ -15,8 +15,12 @@ static class SubscriptionActivity { IMessageConsumeContext context, IEnumerable>? tags = null ) { - context.ParentContext = GetParentContext(context); - var activity = Create(name, activityKind, context.ParentContext, tags); + var (parentContext, link) = GetParentOrLink(context); + context.ParentContext = parentContext; + + var activity = link == null + ? Create(name, activityKind, parentContext, tags) + : CreateLinkedRoot(name, activityKind, link.Value, tags); return activity?.SetContextTags(context); } @@ -42,25 +46,82 @@ static class SubscriptionActivity { .SetOrCopyParentTag(TelemetryTags.Messaging.CorrelationId, context.Metadata?.GetCorrelationId()); } - static ActivityContext? GetParentContext(IBaseConsumeContext context) { + /// + /// Decides how the consume span relates to what came before it: a parent, when the causality is real + /// in-process causality inside a single consume, or a link, when the only thing available is the tracing + /// context persisted in the message metadata. + /// + static (ActivityContext? ParentContext, ActivityLink? Link) GetParentOrLink(IBaseConsumeContext context) { // The current activity is only trusted as a parent when Eventuous created it: that's the // message's own pipeline activity (handler, then filters nested under it). A foreign ambient // activity — a test framework's per-test span, a client library's delivery span — must not - // override the remote context propagated in the message metadata, or the consumer span gets - // detached from the producer trace. + // override the remote context propagated in the message metadata. if (Activity.Current?.Source.Name.StartsWith(EventuousDiagnostics.InstrumentationName, StringComparison.Ordinal) == true) { - return Activity.Current.Context; + return (Activity.Current.Context, null); } if (context.Items.TryGetItem(ContextItemKeys.Activity, out var parentActivity)) { - return parentActivity?.Context; + return (parentActivity?.Context, null); + } + + // The trace context restored from message metadata is durable data on the message, so it outlives the + // process that produced it. Parenting to it would put every redelivery of that message — a replay, a + // checkpoint reset, a hot resubscribe loop — into the same trace, forever: there is no live root to end + // it and no sampling decision left to take, so the trace grows without any bound a collector can impose. + // A link records the same causality without joining the trace, which is also what the OpenTelemetry + // messaging conventions prescribe for an asynchronous "process" operation. + if (context.Metadata?.GetTracingMeta().ToActivityContext(true) is { } remoteContext) { + return (null, new ActivityLink(remoteContext)); } - var tracingData = context.Metadata?.GetTracingMeta(); + return (Activity.Current?.Context, null); + } - if (tracingData?.ToActivityContext(true) is { } remoteContext) return remoteContext; + static Activity? CreateLinkedRoot( + string name, + ActivityKind activityKind, + ActivityLink link, + IEnumerable>? tags + ) { + // .NET offers no "explicitly parentless" argument: a default parent context makes Activity.Current the + // parent. Suppressing it keeps an unrelated ambient span out of both the parentage and the sampling + // decision, which has to be taken as a root. Assigning Activity.Current copies the execution context, + // so it is only touched when there is actually something to suppress: this runs per consumed message. + var ambient = Activity.Current; - return Activity.Current?.Context; + if (ambient != null) Activity.Current = null; + + try { + var activity = EventuousDiagnostics.ActivitySource.CreateActivity( + name, + activityKind, + parentContext: default, + tags, + links: [link], + idFormat: ActivityIdFormat.W3C + ); + + if (activity == null) return null; + + // .NET only materialises the trace id at creation if the sampler actually read it — a ratio-based + // sampler does, an always-on one does not — and an unstarted activity reports an all-zero id + // otherwise. Reuse the sampler's id when there is one, so the decision belongs to the span it was + // taken for, and generate one when there isn't: pinning zeroes below would make the id permanent. + var traceId = activity.TraceId; + + // Activity binds its parent from Activity.Current when it is started, and the async handling path + // starts this activity further down the pipe, where something else may be ambient. Pinning the trace + // id with a zero parent span id makes the span a root that a late Start cannot re-parent. The trace + // flags are carried over, otherwise the span would silently stop being recorded. + return activity.SetParentId( + traceId == default ? ActivityTraceId.CreateRandom() : traceId, + default, + activity.ActivityTraceFlags + ); + } + finally { + if (ambient != null) Activity.Current = ambient; + } } public static Activity? Create( diff --git a/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionActivityTests.cs b/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionActivityTests.cs new file mode 100644 index 000000000..a0927d264 --- /dev/null +++ b/src/Core/test/Eventuous.Tests.Subscriptions/SubscriptionActivityTests.cs @@ -0,0 +1,252 @@ +using System.Diagnostics; +using Eventuous.Diagnostics; +using Eventuous.Subscriptions.Context; +using Eventuous.Subscriptions.Diagnostics; +using Eventuous.TestHelpers; + +namespace Eventuous.Tests.Subscriptions; + +[NotInParallel] +public class SubscriptionActivityTests : IDisposable { + const string ActivityName = "subscription.test/TestEvent"; + const string ForeignSource = "some.other.instrumentation"; + const string ProducerTraceId = "0af7651916cd43dd8448eb211c80319c"; + const string ProducerSpanId = "b7ad6b7169203331"; + + readonly ActivityListener _listener; + readonly ActivitySource _foreignSource = new(ForeignSource); + + public SubscriptionActivityTests() { + // This sampler deliberately does not read options.TraceId. .NET only materialises the trace id of a new + // root when a sampler asks for it, so a sampler that reads it hides the case where an unstarted activity + // still reports an all-zero id — which is the realistic default and the harder half of the contract. + _listener = new() { + ShouldListenTo = source => source.Name == EventuousDiagnostics.InstrumentationName || source.Name == ForeignSource, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded + }; + ActivitySource.AddActivityListener(_listener); + Activity.Current = null; + } + + static MessageConsumeContext CreateContextWithTracingMeta() + => CreateContext( + new Metadata() + .With(DiagnosticTags.TraceId, ProducerTraceId) + .With(DiagnosticTags.SpanId, ProducerSpanId) + .With(MetaTags.CorrelationId, "correlation-1") + ); + + static MessageConsumeContext CreateContext(Metadata? metadata) { + return new( + Guid.NewGuid().ToString(), + "TestEvent", + "application/json", + "test-stream", + 0, + 0, + 0, + 0, + DateTime.UtcNow, + new object(), + metadata, + "test-subscription", + CancellationToken.None + ); + } + + [Test] + public async Task ShouldLinkToRestoredContextInsteadOfParentingToIt() { + var context = CreateContextWithTracingMeta(); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, context); + + await Assert.That(activity).IsNotNull(); + await Assert.That(activity!.Parent).IsNull(); + await Assert.That(activity.ParentSpanId.ToHexString()).IsEqualTo("0000000000000000"); + await Assert.That(activity.TraceId.ToHexString()).IsNotEqualTo(ProducerTraceId); + + // Rooting the span means giving it a trace id of its own. An unstarted activity reports an all-zero id + // unless one was materialised for it, and pinning that zero would make it permanent. + await Assert.That(activity.TraceId.ToHexString()).IsNotEqualTo(RecordedTrace.DefaultTraceId); + + var links = activity.Links.ToArray(); + await Assert.That(links.Length).IsEqualTo(1); + await Assert.That(links[0].Context.TraceId.ToHexString()).IsEqualTo(ProducerTraceId); + await Assert.That(links[0].Context.SpanId.ToHexString()).IsEqualTo(ProducerSpanId); + + // Rooting the span must not drop the sampler's decision, or it would silently stop being exported. + await Assert.That(activity.Recorded).IsTrue(); + await Assert.That(activity.IsAllDataRequested).IsTrue(); + } + + [Test] + public async Task ShouldKeepCorrelationTagsWhenLinking() { + var context = CreateContextWithTracingMeta(); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, context); + + // TagObjects, not Tags: the latter only exposes string-valued tags, and the stream is a StreamName struct. + var tags = activity!.TagObjects.ToDictionary(x => x.Key, x => x.Value?.ToString()); + + await Assert.That(tags[TelemetryTags.Messaging.MessageId]).IsEqualTo(context.MessageId); + await Assert.That(tags[TelemetryTags.Messaging.CorrelationId]).IsEqualTo("correlation-1"); + await Assert.That(tags[TelemetryTags.Eventuous.Stream]).IsEqualTo(context.Stream.ToString()); + await Assert.That(tags[TelemetryTags.Eventuous.Subscription]).IsEqualTo(context.SubscriptionId); + } + + [Test] + public async Task ShouldProduceDistinctTracesOnRedelivery() { + using var first = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, CreateContextWithTracingMeta()); + using var second = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, CreateContextWithTracingMeta()); + + await Assert.That(first!.TraceId.ToHexString()).IsNotEqualTo(second!.TraceId.ToHexString()); + await Assert.That(first.TraceId.ToHexString()).IsNotEqualTo(ProducerTraceId); + await Assert.That(second.TraceId.ToHexString()).IsNotEqualTo(ProducerTraceId); + } + + [Test] + public async Task ShouldKeepTheTraceIdTheSamplerDecidedOn() { + // A ratio-based sampler decides by reading the trace id it is offered, and .NET then materialises that id + // on the activity. Rooting the span has to keep it, or the span carries an id the sampler never evaluated. + _listener.Dispose(); + + var sampledTraceIds = new List(); + + using var samplingListener = new ActivityListener { + ShouldListenTo = source => source.Name == EventuousDiagnostics.InstrumentationName, + Sample = (ref ActivityCreationOptions options) => { + sampledTraceIds.Add(options.TraceId.ToHexString()); + + return ActivitySamplingResult.AllDataAndRecorded; + } + }; + + ActivitySource.AddActivityListener(samplingListener); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, CreateContextWithTracingMeta()); + + await Assert.That(sampledTraceIds).Contains(activity!.TraceId.ToHexString()); + } + + [Test] + public async Task ShouldParentToTheActivityStoredForTheAsyncPath() { + // On the async handling path the subscription creates the message activity, stores it in the context items + // and lets a filter start it further down the pipe. That stored activity is the message's own span, so it + // stays a parent: the link belongs on it, and everything nested under it is plain in-process causality. + using var stored = EventuousDiagnostics.ActivitySource.StartActivity("stored"); + Activity.Current = null; + + var context = CreateContextWithTracingMeta(); + context.Items.AddItem(ContextItemKeys.Activity, stored!); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, context); + + await Assert.That(activity!.TraceId.ToHexString()).IsEqualTo(stored!.TraceId.ToHexString()); + await Assert.That(activity.ParentSpanId.ToHexString()).IsEqualTo(stored.SpanId.ToHexString()); + await Assert.That(activity.Links.ToArray().Length).IsEqualTo(0); + } + + [Test] + public async Task ShouldParentToAmbientEventuousActivity() { + using var ambient = EventuousDiagnostics.ActivitySource.StartActivity("ambient"); + var context = CreateContextWithTracingMeta(); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, context); + + await Assert.That(activity!.TraceId.ToHexString()).IsEqualTo(ambient!.TraceId.ToHexString()); + await Assert.That(activity.ParentSpanId.ToHexString()).IsEqualTo(ambient.SpanId.ToHexString()); + await Assert.That(activity.Links.ToArray().Length).IsEqualTo(0); + } + + [Test] + public async Task ShouldNotParentToForeignAmbientActivity() { + using var foreign = _foreignSource.StartActivity("foreign"); + var context = CreateContextWithTracingMeta(); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, context); + + await Assert.That(foreign).IsNotNull(); + await Assert.That(activity!.Parent).IsNull(); + await Assert.That(activity.TraceId.ToHexString()).IsNotEqualTo(foreign!.TraceId.ToHexString()); + await Assert.That(activity.TraceId.ToHexString()).IsNotEqualTo(ProducerTraceId); + await Assert.That(activity.Links.ToArray().Length).IsEqualTo(1); + } + + [Test] + public async Task ShouldNotLinkWhenMetadataCarriesNoTracingContext() { + // Without a restored context there is nothing to link to, so the pre-existing fallback to the ambient + // activity must stay untouched — an all-zero link would be worse than no link. + using var ambient = _foreignSource.StartActivity("foreign"); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, CreateContext(new Metadata())); + + await Assert.That(activity!.Links.ToArray().Length).IsEqualTo(0); + await Assert.That(activity.TraceId.ToHexString()).IsEqualTo(ambient!.TraceId.ToHexString()); + await Assert.That(activity.ParentSpanId.ToHexString()).IsEqualTo(ambient.SpanId.ToHexString()); + } + + [Test] + public async Task ShouldNotLinkWhenTracingContextIsAllZeroes() { + using var ambient = _foreignSource.StartActivity("foreign"); + + var metadata = new Metadata() + .With(DiagnosticTags.TraceId, RecordedTrace.DefaultTraceId) + .With(DiagnosticTags.SpanId, RecordedTrace.DefaultSpanId); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, CreateContext(metadata)); + + await Assert.That(activity!.Links.ToArray().Length).IsEqualTo(0); + await Assert.That(activity.TraceId.ToHexString()).IsEqualTo(ambient!.TraceId.ToHexString()); + } + + [Test] + public async Task ShouldNotLinkWhenThereIsNoMetadataAtAll() { + using var ambient = _foreignSource.StartActivity("foreign"); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, CreateContext(null)); + + await Assert.That(activity!.Links.ToArray().Length).IsEqualTo(0); + await Assert.That(activity.TraceId.ToHexString()).IsEqualTo(ambient!.TraceId.ToHexString()); + } + + [Test] + public async Task ShouldStayRootWhenStartedLaterUnderAmbientActivity() { + // The async handling path creates the activity in the subscription handler and starts it further down + // the pipe, where another activity may be ambient. A late Start must not re-parent the consume span. + var context = CreateContextWithTracingMeta(); + + using var activity = SubscriptionActivity.Create(ActivityName, ActivityKind.Internal, context); + + using var ambient = EventuousDiagnostics.ActivitySource.StartActivity("started-in-between"); + activity!.Start(); + + await Assert.That(activity.Parent).IsNull(); + await Assert.That(activity.ParentSpanId.ToHexString()).IsEqualTo("0000000000000000"); + await Assert.That(activity.TraceId.ToHexString()).IsNotEqualTo(ambient!.TraceId.ToHexString()); + await Assert.That(activity.TraceId.ToHexString()).IsNotEqualTo(ProducerTraceId); + } + + [Test] + public async Task ShouldCarryTheNewTraceIntoWhatTheHandlerAppends() { + // A reactor appends while consuming, and both TracedEventWriter and ProducerActivity create their span + // with a default parent context, which makes the ambient consume span its parent. The append has to land + // in the consume's own trace, so the producer's old trace stops propagating down the event chain. + using var consume = SubscriptionActivity.Start(ActivityName, ActivityKind.Internal, CreateContextWithTracingMeta()); + + using var append = EventuousDiagnostics.ActivitySource.CreateActivity("append", ActivityKind.Client, parentContext: default); + append!.Start(); + + await Assert.That(append.TraceId.ToHexString()).IsEqualTo(consume!.TraceId.ToHexString()); + await Assert.That(append.TraceId.ToHexString()).IsNotEqualTo(ProducerTraceId); + await Assert.That(append.ParentSpanId.ToHexString()).IsEqualTo(consume.SpanId.ToHexString()); + + // What gets stamped onto the appended event is the new trace, not the one the consumed event carried. + var (traceId, _) = new Metadata().AddActivityTags(append).GetTracingMeta(); + await Assert.That(traceId).IsEqualTo(append.TraceId.ToHexString()); + } + + public void Dispose() { + _listener.Dispose(); + _foreignSource.Dispose(); + } +} diff --git a/src/Core/test/Eventuous.Tests/TracingMetadataTests.cs b/src/Core/test/Eventuous.Tests/TracingMetadataTests.cs new file mode 100644 index 000000000..91094fde5 --- /dev/null +++ b/src/Core/test/Eventuous.Tests/TracingMetadataTests.cs @@ -0,0 +1,72 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +using System.Diagnostics; +using Eventuous.Diagnostics; +using Eventuous.TestHelpers; + +namespace Eventuous.Tests; + +[NotInParallel] +public class TracingMetadataTests : IDisposable { + const string SourceName = "tracing.metadata.tests"; + + readonly ActivitySource _source = new(SourceName); + readonly ActivityListener _listener; + + public TracingMetadataTests() { + // The sampler does not read options.TraceId on purpose: .NET only materialises the trace id of a new root + // when a sampler asks for it, so this is what an activity looks like before it is started. + _listener = new() { + ShouldListenTo = source => source.Name == SourceName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded + }; + ActivitySource.AddActivityListener(_listener); + Activity.Current = null; + } + + [Test] + public async Task ShouldPersistTracingMetaFromStartedActivity() { + using var activity = _source.StartActivity("append"); + + var (traceId, spanId) = new Metadata().AddActivityTags(activity).GetTracingMeta(); + + await Assert.That(traceId).IsEqualTo(activity!.TraceId.ToHexString()); + await Assert.That(spanId).IsEqualTo(activity.SpanId.ToHexString()); + } + + [Test] + public async Task ShouldNotPersistTracingMetaWithoutRealIds() { + // An unstarted activity reports all-zero ids. Writing those to an event is worse than writing nothing: + // it produces a tracing context that looks present but resolves to no trace at all. + using var activity = _source.CreateActivity("append", ActivityKind.Client); + + await Assert.That(activity!.TraceId.ToHexString()).IsEqualTo(RecordedTrace.DefaultTraceId); + + var metadata = new Metadata().AddActivityTags(activity); + var (traceId, spanId) = metadata.GetTracingMeta(); + + await Assert.That(traceId).IsNull(); + await Assert.That(spanId).IsNull(); + } + + [Test] + public async Task ShouldNotOverrideExistingTracingMeta() { + using var activity = _source.StartActivity("append"); + + var metadata = new Metadata() + .With(DiagnosticTags.TraceId, "0af7651916cd43dd8448eb211c80319c") + .With(DiagnosticTags.SpanId, "b7ad6b7169203331") + .AddActivityTags(activity); + + var (traceId, spanId) = metadata.GetTracingMeta(); + + await Assert.That(traceId).IsEqualTo("0af7651916cd43dd8448eb211c80319c"); + await Assert.That(spanId).IsEqualTo("b7ad6b7169203331"); + } + + public void Dispose() { + _listener.Dispose(); + _source.Dispose(); + } +} diff --git a/src/Diagnostics/src/Eventuous.Diagnostics.OpenTelemetry/TracerProviderBuilderExtensions.cs b/src/Diagnostics/src/Eventuous.Diagnostics.OpenTelemetry/TracerProviderBuilderExtensions.cs index 4afe7f822..701866837 100644 --- a/src/Diagnostics/src/Eventuous.Diagnostics.OpenTelemetry/TracerProviderBuilderExtensions.cs +++ b/src/Diagnostics/src/Eventuous.Diagnostics.OpenTelemetry/TracerProviderBuilderExtensions.cs @@ -1,7 +1,6 @@ // Copyright (C) Eventuous HQ OÜ. All rights reserved // Licensed under the Apache License, Version 2.0. -using System.Diagnostics; using OpenTelemetry.Trace; namespace Eventuous.Diagnostics.OpenTelemetry; @@ -9,7 +8,8 @@ namespace Eventuous.Diagnostics.OpenTelemetry; [PublicAPI] public static class TracerProviderBuilderExtensions { /// - /// Adds an Eventuous activity source to OpenTelemetry trace collection + /// Adds an Eventuous activity source to OpenTelemetry trace collection. Sampling is left to the application: + /// this only registers the source, so whatever sampler is configured on the provider stays in effect. /// /// instance /// @@ -18,14 +18,6 @@ public static TracerProviderBuilder AddEventuousTracing(this TracerProviderBuild // After adding the activity source to OpenTelemetry, we don't need a fake listener. EventuousDiagnostics.RemoveDummyListener(); - return Ensure.NotNull(builder).AddSource(EventuousDiagnostics.InstrumentationName).SetSampler(new PollingSampler()); - } - - class PollingSampler : Sampler { - public override SamplingResult ShouldSample(in SamplingParameters samplingParameters) { - return samplingParameters.ParentContext is { TraceFlags: ActivityTraceFlags.None } && samplingParameters is { Kind: ActivityKind.Client, Name: "eventuous" } - ? new SamplingResult(SamplingDecision.Drop) - : new(SamplingDecision.RecordAndSample); - } + return Ensure.NotNull(builder).AddSource(EventuousDiagnostics.InstrumentationName); } } diff --git a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/ProducerTracesTests.cs b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/ProducerTracesTests.cs index 1a654dcf9..e82bea8d1 100644 --- a/src/KurrentDB/test/Eventuous.Tests.KurrentDB/ProducerTracesTests.cs +++ b/src/KurrentDB/test/Eventuous.Tests.KurrentDB/ProducerTracesTests.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using System.Diagnostics; using Eventuous.Diagnostics; using Eventuous.Producers; @@ -5,11 +6,13 @@ using Eventuous.Tests.KurrentDB.Subscriptions.Fixtures; using Eventuous.Tests.Subscriptions.Base; using Shouldly; +using Constants = Eventuous.Diagnostics.Tracing.Constants; namespace Eventuous.Tests.KurrentDB; public class TracesTests : LegacySubscriptionFixture { - readonly ActivityListener _listener; + readonly ActivityListener _listener; + readonly ConcurrentBag _startedActivities = []; static TracesTests() => TypeMap.Instance.AddType(TestEvent.TypeName); @@ -17,13 +20,12 @@ public TracesTests() : base(new()) { _listener = new() { ShouldListenTo = _ => true, // ReSharper disable once RedundantLambdaParameterType - Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData, - ActivityStarted = activity => Log.LogTrace( - "Started {Activity} with {Id}, parent {ParentId}", - activity.DisplayName, - activity.Id, - activity.ParentId - ), + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllData, + ActivityStarted = activity => { + _startedActivities.Add(activity); + + Log.LogTrace("Started {Activity} with {Id}, parent {ParentId}", activity.DisplayName, activity.Id, activity.ParentId); + }, ActivityStopped = activity => Log.LogTrace("Stopped {Activity}", activity.DisplayName) }; @@ -32,7 +34,7 @@ public TracesTests() : base(new()) { [Test] [Category("Diagnostics")] - public async Task ShouldPropagateRemoteContext(CancellationToken cancellationToken) { + public async Task ShouldLinkToRemoteContextWithoutJoiningItsTrace(CancellationToken cancellationToken) { var testEvent = TestEvent.Create(); await Producer.Produce(Stream, testEvent, new(), cancellationToken: cancellationToken); @@ -59,8 +61,24 @@ public async Task ShouldPropagateRemoteContext(CancellationToken cancellationTok recordedTrace.IsDefaultTraceId.ShouldBeFalse(); recordedTrace.IsDefaultSpanId.ShouldBeFalse(); - recordedTrace.TraceId!.Value.ToString().ShouldBe(traceId); - recordedTrace.ParentSpanId!.Value.ToString().ShouldBe(spanId); + + // The consume span roots its own trace and links back to the producer, so that redelivering a stored + // event can never keep growing the trace the producer started. + recordedTrace.TraceId!.Value.ToString().ShouldNotBe(traceId); + + // The link sits on the subscription activity, which restored the context. It isn't reachable from the + // handler's ambient activity on the async consume path, so it's asserted on the recorded activity instead. + // The listener is process-wide, so the activity is matched on this subscription's own id to keep any + // test running in parallel out of the result. + var subscriptionActivity = _startedActivities + .Single(x => x.OperationName.StartsWith($"{Constants.Components.Subscription}.{Subscription.SubscriptionId}/", StringComparison.Ordinal)); + + subscriptionActivity.TraceId.ToString().ShouldNotBe(traceId); + subscriptionActivity.Parent.ShouldBeNull(); + + var link = subscriptionActivity.Links.ShouldHaveSingleItem(); + link.Context.TraceId.ToString().ShouldBe(traceId); + link.Context.SpanId.ToString().ShouldBe(spanId); } [After(Test)] diff --git a/src/SignalR/src/Eventuous.SignalR.Client/Eventuous.SignalR.Client.csproj b/src/SignalR/src/Eventuous.SignalR.Client/Eventuous.SignalR.Client.csproj index 018077b29..91ca7a079 100644 --- a/src/SignalR/src/Eventuous.SignalR.Client/Eventuous.SignalR.Client.csproj +++ b/src/SignalR/src/Eventuous.SignalR.Client/Eventuous.SignalR.Client.csproj @@ -7,6 +7,9 @@ + + + diff --git a/src/SignalR/src/Eventuous.SignalR.Client/TypedStreamSubscription.cs b/src/SignalR/src/Eventuous.SignalR.Client/TypedStreamSubscription.cs index 8852f177f..73f68a92a 100644 --- a/src/SignalR/src/Eventuous.SignalR.Client/TypedStreamSubscription.cs +++ b/src/SignalR/src/Eventuous.SignalR.Client/TypedStreamSubscription.cs @@ -117,23 +117,38 @@ async Task ConsumeLoop(ChannelReader reader, CancellationTo } } - static Activity? StartTraceActivity(string jsonMetadata) { + internal static Activity? StartTraceActivity(string jsonMetadata) { try { var metaDict = JsonSerializer.Deserialize>(jsonMetadata); if (metaDict == null) return null; - var metadata = new Metadata(metaDict); - var tracingMeta = metadata.GetTracingMeta(); - var parentContext = tracingMeta.ToActivityContext(isRemote: true); + var metadata = new Metadata(metaDict); + var tracingMeta = metadata.GetTracingMeta(); + var producerContext = tracingMeta.ToActivityContext(isRemote: true); - if (parentContext == null) return null; + if (producerContext == null) return null; - return EventuousDiagnostics.ActivitySource.StartActivity( - "signalr.consume", - ActivityKind.Consumer, - parentContext.Value - ); + // The restored context is durable data on the event, so parenting to it would put every redelivery + // into the trace the producer started, without any bound. Linking records the same causality while + // the consume span roots its own trace. Suppressing the ambient activity keeps .NET from adopting it + // as the parent, since a default parent context means "use Activity.Current". + var ambient = Activity.Current; + + if (ambient != null) Activity.Current = null; + + try { + return EventuousDiagnostics.ActivitySource.StartActivity( + "signalr.consume", + ActivityKind.Consumer, + parentContext: default, + tags: null, + links: [new ActivityLink(producerContext.Value)] + ); + } + finally { + if (ambient != null) Activity.Current = ambient; + } } catch (Exception) { // Tracing is the best effort; malformed metadata must not break event consumption return null; diff --git a/src/SignalR/test/Eventuous.Tests.SignalR/ConsumeTracingTests.cs b/src/SignalR/test/Eventuous.Tests.SignalR/ConsumeTracingTests.cs new file mode 100644 index 000000000..b6563024a --- /dev/null +++ b/src/SignalR/test/Eventuous.Tests.SignalR/ConsumeTracingTests.cs @@ -0,0 +1,74 @@ +// Copyright (C) Eventuous HQ OÜ. All rights reserved +// Licensed under the Apache License, Version 2.0. + +extern alias SignalRClient; +using System.Diagnostics; +using System.Text.Json; +using Eventuous.Diagnostics; +using SignalRClient::Eventuous.SignalR.Client; + +namespace Eventuous.Tests.SignalR; + +[NotInParallel] +public class ConsumeTracingTests : IDisposable { + const string ProducerTraceId = "0af7651916cd43dd8448eb211c80319c"; + const string ProducerSpanId = "b7ad6b7169203331"; + + readonly ActivityListener _listener; + + public ConsumeTracingTests() { + _listener = new() { + ShouldListenTo = source => source.Name == EventuousDiagnostics.InstrumentationName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded + }; + ActivitySource.AddActivityListener(_listener); + Activity.Current = null; + } + + static string MetadataWithTracingContext() + => JsonSerializer.Serialize( + new Dictionary { + [DiagnosticTags.TraceId] = ProducerTraceId, + [DiagnosticTags.SpanId] = ProducerSpanId + } + ); + + [Test] + public async Task ShouldLinkToRestoredContextInsteadOfParentingToIt() { + using var activity = TypedStreamSubscription.StartTraceActivity(MetadataWithTracingContext()); + + await Assert.That(activity).IsNotNull(); + await Assert.That(activity!.Parent).IsNull(); + await Assert.That(activity.ParentSpanId.ToHexString()).IsEqualTo("0000000000000000"); + await Assert.That(activity.TraceId.ToHexString()).IsNotEqualTo(ProducerTraceId); + + var links = activity.Links.ToArray(); + await Assert.That(links.Length).IsEqualTo(1); + await Assert.That(links[0].Context.TraceId.ToHexString()).IsEqualTo(ProducerTraceId); + await Assert.That(links[0].Context.SpanId.ToHexString()).IsEqualTo(ProducerSpanId); + } + + [Test] + public async Task ShouldProduceDistinctTracesOnRedelivery() { + using var first = TypedStreamSubscription.StartTraceActivity(MetadataWithTracingContext()); + using var second = TypedStreamSubscription.StartTraceActivity(MetadataWithTracingContext()); + + await Assert.That(first!.TraceId.ToHexString()).IsNotEqualTo(second!.TraceId.ToHexString()); + } + + [Test] + public async Task ShouldReturnNullWithoutTracingContext() { + using var activity = TypedStreamSubscription.StartTraceActivity("""{"some":"meta"}"""); + + await Assert.That(activity).IsNull(); + } + + [Test] + public async Task ShouldReturnNullOnMalformedMetadata() { + using var activity = TypedStreamSubscription.StartTraceActivity("not json"); + + await Assert.That(activity).IsNull(); + } + + public void Dispose() => _listener.Dispose(); +}