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();
+}