Add OpenTelemetry end-to-end trace context propagation for Pekko Streams

[He-Pin]

Motivation

Pekko Streams has no distributed tracing support. Users running Pekko in microservices cannot trace how elements flow through stream topologies. This is increasingly expected in production environments for observability.

Proposed design (based on Akka.NET reference implementation)

1. Trace context in Connection slots: Extend Connection in GraphInterpreter with a trace context field. Capture io.opentelemetry.context.Context.current() on element push.

2. Stage-scoped spans: In processPush, start an OpenTelemetry span scoped to the stage. Span name = stage operation (e.g., MapAsync, Filter).

3. Fan-in links: For fan-in stages (Batch, Merge, Concat, GroupedWeightedWithin), record ActivityLinks from all input connections.

4. Cross-async-boundary propagation: Extend OnNext events with trace context. BatchingActorInputBoundary buffers contexts in parallel arrays and restores them on dequeue.

5. Zero-cost when no listener: When no OpenTelemetry listener is registered, all tracing code should be a no-op with negligible overhead.

JVM-specific considerations

• Use io.opentelemetry.api.trace.Tracer instead of .NET ActivitySource
• Use io.opentelemetry.context.Context and ThreadLocal propagation
• Consider integration with Pekko's existing ContextPropagation mechanism
• May need a new module/dependency for opentelemetry-api

Scope

This is a large, cross-cutting feature (~2000+ lines across 20+ files). It should be broken into multiple PRs:

1. Core trace context in GraphInterpreter Connection
2. Stage-scoped spans in processPush
3. Fan-in link support
4. Cross-async-boundary propagation
5. Source.Queue / Sink.Queue ingress/egress tracing
6. Configuration and opt-in mechanism

References

• OpenTelemetry Java SDK

[pjfanning]

https://opentelemetry.io/docs/languages/java/ has Pekko integration already based on AOP.

Can we discuss this before jumping in a major implementation?

• Instrument Pekko (by copying from akka) open-telemetry/opentelemetry-java-instrumentation#9527
• https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/pekko

There doesn't appear to be anything specifically for pekko-streams - just pekko-actor and pekko-http.

[He-Pin]

wow, that's nice.

[pjfanning]

I like the idea of supporting context propagation. I don't like the idea of adding jar dependencies like opentelemetry jars to pekko-streams.
If we could keep the opentelemetry dependencies in a separate pekko-streams-telemetry jar, that would be great.
Maybe it would be possible to write our own context and propagate that in the core pekko-streams code and find some way to integrate that with opentelemetry in an extension lib?
It might be worth looking at Kamon which adds context propagation capabilities to Pekko but not Pekko Streams as far as I know. Kamon has https://github.com/kamon-io/Kamon/tree/master/reporters/kamon-opentelemetry

Open discussion on this topic:
#1499

[pjfanning]

Copilot recommends an agnostic API for pekko-streams extending the existing pekko-streams ContextPropagation.

pekko/stream/src/main/scala/org/apache/pekko/stream/impl/ContextPropagation.scala

Line 39 in d75bf95

private[pekko] final class ContextPropagationImpl extends ContextPropagation {

The existing ContextPropagation looks like it relies on something like AOP to add-in a meaningful implementation.
We might want to consider reworking this so that we can allow for a ContextPropagation class to be be registered by adding a jar to the classpath that contains a compatible class - or some other non-AOP mechanism.

package org.apache.pekko.stream.impl

import org.apache.pekko.annotation.InternalStableApi

/**
 * INTERNAL API
 *
 * Manages context propagation for stream elements during execution.
 *
 * This trait supports two main use cases:
 * 1. Context suspension/resumption for elements flowing through stages
 * 2. Tracing lifecycle hooks for external observability integration (e.g., OpenTelemetry)
 *
 * The tracing methods provide extension points for external libraries to implement distributed tracing
 * without requiring Pekko to depend on any specific tracing framework.
 *
 * External integration libraries can:
 * - Implement this trait with real tracing logic
 * - Capture trace context from incoming elements via `captureTraceContext()`
 * - Emit trace spans on element push/pop via `onPushElement()` and `onPopElement()`
 * - Propagate context across async boundaries via `onAsyncBoundary()`
 *
 * When no tracing is configured, all methods remain no-ops with negligible overhead.
 */
@InternalStableApi trait ContextPropagation {

  /**
   * Suspend the current context, typically called when an element cannot be immediately processed downstream.
   */
  def suspendContext(): Unit

  /**
   * Resume a previously suspended context.
   */
  def resumeContext(): Unit

  /**
   * Get the current context object. Returns an opaque AnyRef that can be later restored.
   */
  def currentContext(): AnyRef

  /**
   * Resume a specific context that was previously captured via `currentContext()`.
   */
  def resumeContext(context: AnyRef): Unit

  /**
   * INTERNAL API
   *
   * Called when an element is about to be pushed through a stage output port.
   *
   * External tracing libraries can use this hook to:
   * - Start a span for element processing in this stage
   * - Record metrics about element flow
   * - Link to parent traces
   *
   * @param stageName The name/type of the stage (e.g., "map", "filter", "mapAsync")
   * @param portId The output port identifier through which the element is being pushed
   */
  def onPushElement(stageName: String, portId: Int): Unit = ()

  /**
   * INTERNAL API
   *
   * Called when an element is about to be popped from a stage input port.
   *
   * External tracing libraries can use this hook to:
   * - End/close spans created in `onPushElement`
   * - Record completion of element processing
   * - Capture any error states
   *
   * @param stageName The name/type of the stage
   * @param portId The input port identifier through which the element is being popped
   */
  def onPopElement(stageName: String, portId: Int): Unit = ()

  /**
   * INTERNAL API
   *
   * Called when an element crosses an async boundary (e.g., through an actor or thread pool).
   *
   * External tracing libraries can use this hook to:
   * - Capture the trace context before crossing the boundary
   * - Ensure proper context restoration after async operations
   * - Handle context propagation across threads or actor messages
   *
   * @param stageName The name/type of the stage performing the async operation
   */
  def onAsyncBoundary(stageName: String): Unit = ()

  /**
   * INTERNAL API
   *
   * Capture the current trace context for propagation across async boundaries.
   *
   * This is called before an element is sent to an async operation (e.g., via Future or actor message).
   * The returned context should be lightweight and serializable across threads.
   *
   * @return An opaque trace context object that can be later restored via `restoreTraceContext()`,
   *         or null if no tracing context is available
   */
  def captureTraceContext(): AnyRef = null

  /**
   * INTERNAL API
   *
   * Restore a trace context that was captured before an async boundary.
   *
   * This is called after an element returns from an async operation to restore the original
   * trace context and allow proper span linking.
   *
   * @param context The trace context captured via `captureTraceContext()`
   */
  def restoreTraceContext(context: AnyRef): Unit = ()
}

/**
 * INTERNAL API
 */
@InternalStableApi object ContextPropagation {

  /**
   * INTERNAL API
   */
  @InternalStableApi def apply(): ContextPropagation = new ContextPropagationImpl
}

private[pekko] final class ContextPropagationImpl extends ContextPropagation {
  def suspendContext(): Unit = ()
  def resumeContext(): Unit = ()
  def currentContext(): AnyRef = null
  def resumeContext(context: AnyRef): Unit = ()
}

This would be wired into pekko-streams existing lib.
We could then create an OpenTelemetry based implementation in a separate lib.

[pjfanning]

@He-Pin there is nothing wrong with us copying the approach used by akka.net. That code is Apache licensed. The only minor issue is that we need to add akka.net source headers to the Pekko files where we end up with akka.net inspired code and mention those files in our overall license file.