Akka.Streams: end-to-end OpenTelemetry trace context propagation

[Aaronontheweb]

Problem

AsyncLocal<Activity> does not flow across the stream actor's dispatcher boundary, so any trace context alive on a producer thread is lost by the time downstream stage handlers run on the interpreter dispatcher. Anything the user's code creates inside a SelectAsync lambda — SqlClient.ExecuteAsync, HttpClient.SendAsync, or any other instrumented call — becomes an orphaned root span with a random TraceId, disconnected from the request that triggered it.

A secondary, related problem: when a fan-in stage like BatchWeighted or GroupedWithin merges N independent input elements into a single output, any trace continuity information from the other N-1 contributing elements is lost. The downstream span ends up attached to whichever input happened to trigger the flush, and every other contributing request becomes an orphaned root.

Approach

Same shape as #7995 (LogEvent.ActivityContext capture for log-trace correlation across mailbox boundaries):

1. Capture on the producer side, before the boundary hop, while Activity.Current is still alive.
2. Carry in a framework-owned field on the in-flight element / queued event.
3. Restore on the consumer side before invoking the user's stage handler.

For fan-in, the canonical OpenTelemetry answer is span links: pick one primary parent (first-wins) and attach the remaining N-1 inputs as ActivityLinks on the flushed stage span, so trace viewers can jump between contributing traces.

Commits

1. Akka.Streams: per-element ActivityContext carry through GraphInterpreter

• New framework-owned ActivitySource("Akka.Streams"). Users register via .AddSource("Akka.Streams") on their OTel TracerProvider (same pattern as Microsoft.EntityFrameworkCore).
• Connection.SlotContext parallels Connection.Slot and carries the ActivityContext of the element in flight on that connection.
• GraphStageLogic.Push captures Activity.Current?.Context into SlotContext after writing the slot; Grab clears it alongside.
• GraphInterpreter.ProcessPush starts an akka.stream.stage {StageName} Activity with the slot context as parent before invoking OnPush. Disposes in finally. Mirrors the existing previousInterpreter try/finally restore pattern.

2. Akka.Streams: generalize trace ingress capture via shared callback primitive

• GraphStageLogicWithCallbackWrapper.NotInitialized.Args carries the producer's ActivityContext alongside each queued arg, captured at InvokeCallbacks time. InitCallback drains queued args inside a restore scope so the callback runs with the right ambient Activity.Current even if it was queued before PreStart.
• ConcurrentAsyncCallback<T>.InvokeWithPromise captures Activity.Current?.Context and wraps the handler with an akka.stream.ingress {StageName} Activity on the interpreter thread.

ConcurrentAsyncCallback is the shared primitive used by Source.Queue, custom StageActor-based sources, and any source going through GraphStageLogicWithCallbackWrapper, so one fix lights up all of them without per-stage code.

3. Akka.Streams: fan-in stage span linking with ActivityLinks

• Connection.SlotLinks (ActivityContext[]) carries additional parent contexts alongside the primary SlotContext, plus PendingPushPrimaryContext / PendingPushLinks as a per-push override mechanism.
• GraphStageLogic exposes two internal APIs for fan-in stages:
  • CurrentInletTraceContext(inlet) — read the upstream element's trace context before Grab clears it
  • SetFanInTraceContext(outlet, primary, links) — stage the next Push as a fan-in flush with a specific primary parent + link set
• GraphStageLogic.Push consumes the pending override when present, falling back to Activity.Current capture otherwise (preserves existing behavior for non-fan-in stages).
• GraphInterpreter.ProcessPush starts the downstream stage Activity with both the primary context as parent and the accumulated SlotLinks as ActivityLinks, tagging the span with stream.fan_in.links for visibility.
• Batch.Logic (covers both Batch and BatchWeighted) accumulates each inbound element's trace context in OnPush and emits the aggregate on Flush. Pending-element state carries its own context across the flush/reseed boundary so that the element triggering a boundary flush also contributes to the next aggregate's links correctly.

4. Akka.Streams: fan-in/fan-out stage trace propagation + spec suite

Extends fan-in linking to the other built-in stages and adds fan-out + regression coverage:

• GroupedWeightedWithin.Logic — captures each inbound element's context alongside the buffer, plus the pending-element context for boundary overflow. On EmitGroup emits the first collected context as the primary parent and the rest as ActivityLinks.
• Merge.Logic — captures the inlet's SlotContext on both the fast path (outlet immediately available) and the slow path (element enqueued for later DequeueAndDispatch), staging it as the downstream Push's primary parent so trace continuity survives the OnPull boundary where Activity.Current would otherwise be null.
• MergePreferred.Logic — same, for both the preferred and secondary Emit paths.
• Concat.Logic — same, for every secondary inlet's OnPush.

Merge/MergePreferred/Concat are 1-to-1 pass-throughs, so they never actually attach multiple ActivityLinks — the fan-in API is used only to override the primary parent from Activity.Current (which on OnPull is null) to the captured upstream inlet context. Batch/BatchWeighted and GroupedWeightedWithin are true fan-ins that attach N-1 links.

Fan-out stages (Broadcast, Balance) require no changes. Each downstream branch's ProcessPush already captures Activity.Current (which is the fan-out stage's own span, itself parented to the upstream slot context), so every branch inherits the producer's trace id via the existing per-element carry.

Tests

src/core/Akka.Streams.Tests/Implementation/ — 14 scenarios across 4 focused spec files plus a shared helper:

StreamsDiagnosticsSpec (4 tests) — linear chain + basic ingress

1. Sync Source.Queue → Select → Sink.Seq — stage spans emitted with correct parent chain.
2. Source.Queue → SelectAsync → Sink.Ignore with a user span created inside the async lambda — verifies async-boundary continuity.
3. Two offers from two distinct traced scopes through the same Source.Queue — multi-producer interleaving lands each element's downstream spans in its own trace, never mixed.
4. Element offered without Activity.Current set — zero stream spans emitted.

StreamsFanInSpec (5 tests) — true fan-in and pass-through fan-ins

• BatchWeighted with 3 concurrent traced producers — asserts primary parent = first input, 2 ActivityLinks to the other two inputs.
• BatchWeighted with 7 traced inputs — asserts link count is exactly N - 1.
• GroupedWithin count-driven flush — asserts primary + 3 links on the emitted group.
• Merge — two independent queues each offer one traced element; both elements reach the downstream Select stage with their originating producer's trace id intact.
• Concat — sequential drain of two sources; each element carries its own source's trace forward.

StreamsFanOutSpec (2 tests) — Broadcast/Balance validation-only

• Broadcast(2) — one traced offer fans out to two Select stages, both inherit the producer's trace id.
• Balance(2) — two traced offers distributed round-robin, both branches run under the producer's trace.

StreamsRegressionSpec (3 tests) — cardinality guards

• Background Source.Tick → Select → SelectAsync → Sink.Ignore with no traced producer — asserts zero "Akka.Streams" spans emitted. Guards against unbounded span accumulation on long-lived background streams.
• Mixed traced + untraced via Merge — traced offer produces spans, tick ticks stay invisible, every emitted span belongs to the producer's trace.
• GraphDsl-composed sub-graph with internal Select → Select chain — end-to-end trace id propagation through the composition boundary.

StreamsActivityCollector (helper)

Shared internal helper extracted so each spec file can subscribe to the "Akka.Streams" ActivitySource the same way without duplicating 15 lines of ActivityListener boilerplate.

Verification

• 1,815 / 1,815 Akka.Streams.Tests (net10.0) passing, 34 pre-existing skips. Zero behavioral regressions from the new primitives.
• 14 / 14 new trace specs passing.
• End-to-end trace continuity visually confirmed in a real OTLP → dashboard pipeline against a live Source.Queue → Select → BatchWeighted → SelectAsync → HttpClient pipeline with concurrent producers: the flushed batch's downstream stage span carries the expected Links: N > 0, and each producer's individual trace backlinks into the shared batch span.

Performance

• Zero overhead when the Akka.Streams source has no listener (StartActivity returns null) or when Activity.Current is null at the producer (background streams, internal sources like Source.Tick).
• One closure allocation per ConcurrentAsyncCallback.Invoke only when a producer trace context exists.
• One stage Activity allocation per stage transition per element when tracing is active.
• Fan-in stages allocate one List<ActivityContext> per logic instance (lifetime-of-stage, not per-element) plus one ActivityLink[] per flush when there are 2+ contributing traces.

Happy to add BenchmarkDotNet coverage if reviewers want to quantify any of those.

Scope

Covered in this PR:

• GetAsyncCallback-based sources: Source.Queue, custom StageActor-based sources, anything going through GraphStageLogicWithCallbackWrapper
• Fan-in: Batch, BatchWeighted, GroupedWeightedWithin, Merge, MergePreferred, Concat
• Fan-out: Broadcast, Balance (validation-only, no stage-specific changes)

Not covered (follow-up work):

• Source.ActorRef / Source.ActorPublisher — reactive-streams OnNext path, would need a separate ingress capture site
• MergePrioritized, Zip family (ZipWith, ZipLatest, ZipN, Interleave), OrElse — additional fan-in shapes not yet wired
• Attribute enrichment from Attributes.Name — stage spans currently use type-reflected names (e.g. Select) rather than user-supplied .Named("...") labels
• Dedicated akka.stream.batch.flush span type — current implementation inlines fan-in links onto the existing downstream stage span, which is simpler but less explicit in trace viewers

Status

Draft. Opened to invite design feedback before any merge consideration.

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[Aaronontheweb]

This is essentially our answer to petabridge/phobos-issues#42, and the results are pretty solid with our latest Phobos prototype that is built on this branch.

Before this PR, a SqlClient.ExecuteAsync or HttpClient.SendAsync inside a SelectAsync body becomes an orphaned root span with a random TraceId — AsyncLocal<Activity> doesn't flow across the stream actor's dispatcher boundary, so the producer-side context is just gone by the time the handler runs. And when multiple concurrent requests get merged by BatchWeighted, everyone except whichever request happened to trigger the flush loses their trace identity entirely.

After this PR, both of those work. Every stage in the graph emits a span parented to the right place, and fan-in merges attach ActivityLink references from the flushed batch back to each contributing request's trace — so any viewer that speaks OTLP can jump across sibling traces that share downstream work.

Here's the pipeline I've been exercising, side-by-side with what Jaeger renders:

var orderQueue = Source.Queue<Order>(100, OverflowStrategy.DropNew) .Select(Normalize) .BatchWeighted( max: 5L, costFunction: _ => 1L, seed: o => new List<Order> { o }, aggregate: (acc, o) => { acc.Add(o); return acc; }) .SelectAsync(1, async batch => { using var resp = await httpClient .PostAsync("anything?batch=1", Serialize(batch)); return batch; }) .ToMaterialized( Sink.Ignore<List<Order>>(), Keep.Left) .Run(_materializer); Twelve producers fire at the queue concurrently; BatchWeighted merges five of them into one outbound HttpClient POST. Opting the new spans into your OpenTelemetry pipeline is one line: tracing.AddSource("Akka.Streams"); Same shape as AddSource("Microsoft.EntityFrameworkCore"). Zero cost if nobody's listening.

The stuff I want to call out here: The entire span tree on the left is one trace. Actor OnReceive → QueueSource ingress → Select → Batch → SelectAsync → outbound HttpClient POST. Every stage becomes its own span, parented correctly. The tags row on the right shows otel.library.name = Akka.Streams and stream.fan_in.links = 4. That second tag is the framework advertising from the emit side — "this span has four cross-trace links on it." The References (5) section lists one CHILD_OF → the local Batch span (normal parent/child inside this trace), plus four FOLLOWS_FROM entries → < span in another trace >. Jaeger renders OpenTelemetry ActivityLink as FOLLOWS_FROM. That's five concurrent requests merged into one outbound HTTP call, every one of them keeping its own trace identity and backlinking to the shared flush.

A few more views of the same batch for reference:

Full span tree without the detail panel

SelectAsync detail panel with References collapsed

One of the contributing requests on its own trace

A request whose element got batched into the flush above. Its own trace stops at QueueSource — the Select / Batch / SelectAsync / IgnoreSink downstream work lives on the merged-batch trace, reachable via the FOLLOWS_FROM link. Click it in a viewer that supports cross-trace navigation and you land on the batched span.

[Aaronontheweb]

One thing I should flag since the screenshot above shows an actor in the root position of the trace: none of what this PR adds actually requires an actor at the top, and none of it requires Phobos. The framework change just reads Activity.Current?.Context at whatever thread is calling OfferAsync (or any GetAsyncCallback-based ingress path) and carries it through. It doesn't care where that Activity came from.

Concretely, all of these Just Work once you .AddSource("Akka.Streams") on your OpenTelemetry TracerProvider:

• ASP.NET Core controller → direct OfferAsync. OpenTelemetry.Instrumentation.AspNetCore sets Activity.Current for the duration of the request handler, so a controller that materializes (or holds a reference to) a Source.Queue and offers into it directly gets the full chain — incoming HTTP span → ingress → stage spans → whatever the SelectAsync body does. No actor, no Phobos.
• BackgroundService doing await httpClient.GetAsync(...) then await queue.OfferAsync(result). The HttpClient instrumentation sets Activity.Current to the outbound call's span for the duration of the continuation, so the stream work lands underneath the HTTP client's trace.
• User code that wraps work in its own ActivitySource.StartActivity(...) scope. Any span the user owns becomes the parent of whatever the stream does from that scope. Completely untethered from anything Akka.NET-specific.

The reason the sample above happens to show an actor at the top is just that we were stress-testing the actor-to-stream path specifically — that's the case where AsyncLocal<Activity> breaks across the mailbox boundary and you need some mechanism to re-establish the context on the receiving side. Phobos is how we do that in our prototype, but a handwritten Tell-capturing wrapper or a pattern where the stream is materialized outside the actor would work identically — this PR doesn't care.

TL;DR: this is a framework-level capability. Opt in with one line of OTel registration, get end-to-end trace continuity through any graph shape, from any producer.

[Aaronontheweb]

Known limitation: trace context does not cross a StreamRef hop.

StreamRefs serialize elements across the network via StreamRefsProtocol.proto, which doesn't have a field for ActivityContext. So when an element travels from Node A's SourceRef/SinkRef to Node B, the context that was alive on Node A during Push is dropped at serialization time. On Node B, the interpreter thread runs with Activity.Current = null, the new context-capture sees null, and downstream stages on Node B emit no spans.

Not a regression — before this PR there was no context propagation at all. After this PR, local-node traces work end-to-end; cross-node traces that flow through a StreamRef still break the chain at the wire boundary.

The fix is pretty small and follows the same pattern as #7995 (which added LogEvent.ActivityContext so log events can carry trace context across the mailbox boundary): add an optional ActivityContext field to the OnNext payload in StreamRefsProtocol.proto, capture on the sending side in SourceRefImpl, restore on the receiving side before the local Push call. Wire-compat safe because the proto field is optional, and entirely additive — no existing StreamRef behavior changes.

Out of scope for this PR. Opening a tracking issue to come back to it after this lands.

[Aaronontheweb]

Known limitation: trace context does not cross a StreamRef hop.

StreamRefs serialize elements across the network via StreamRefsProtocol.proto, which doesn't have a field for ActivityContext. So when an element travels from Node A's SourceRef/SinkRef to Node B, the context that was alive on Node A during Push is dropped at serialization time. On Node B, the interpreter thread runs with Activity.Current = null, the new context-capture sees null, and downstream stages on Node B emit no spans.

Not a regression — before this PR there was no context propagation at all. After this PR, local-node traces work end-to-end; cross-node traces that flow through a StreamRef still break the chain at the wire boundary.

The fix is pretty small and follows the same pattern as #7995 (which added LogEvent.ActivityContext so log events can carry trace context across the mailbox boundary): add an optional ActivityContext field to the OnNext payload in StreamRefsProtocol.proto, capture on the sending side in SourceRefImpl, restore on the receiving side before the local Push call. Wire-compat safe because the proto field is optional, and entirely additive — no existing StreamRef behavior changes.

Out of scope for this PR. Opening a tracking issue to come back to it after this lands.

LLM generated comment but it's correct - not going to bother getting this to work with StreamRefs because IMHO StreamRef serialization sucks today and will improve a lot once we add code-generated serialization in v1.6.

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).