feat(telemetry): propagate caller trace context to aggregate load spans

guides/explanations/fork-differences.md

@@ -218,11 +218,11 @@ end
 With a dedicated protocol, the API response format and the event store stream ID format are properly separated and can evolve independently.
 
 ### **OpenTelemetry Integration**
-PRs: [#37](https://github.com/straw-hat-team/commanded/pull/37), [#41](https://github.com/straw-hat-team/commanded/pull/41), [#45](https://github.com/straw-hat-team/commanded/pull/45), [#46](https://github.com/straw-hat-team/commanded/pull/46), [#47](https://github.com/straw-hat-team/commanded/pull/47), [#58](https://github.com/straw-hat-team/commanded/pull/58), [#60](https://github.com/straw-hat-team/commanded/pull/60), [#61](https://github.com/straw-hat-team/commanded/pull/61)
+PRs: [#37](https://github.com/straw-hat-team/commanded/pull/37), [#41](https://github.com/straw-hat-team/commanded/pull/41), [#45](https://github.com/straw-hat-team/commanded/pull/45), [#46](https://github.com/straw-hat-team/commanded/pull/46), [#47](https://github.com/straw-hat-team/commanded/pull/47), [#58](https://github.com/straw-hat-team/commanded/pull/58), [#60](https://github.com/straw-hat-team/commanded/pull/60), [#61](https://github.com/straw-hat-team/commanded/pull/61), [#90](https://github.com/straw-hat-team/commanded/pull/90)
 
 **Changes:**
 - Added `Commanded.OpenTelemetry` module for distributed tracing
-- Creates spans for event handlers (PR #41), EventStore operations (PR #37), aggregate execution (PR #45), application dispatch (PR #46), aggregate load (PR #58), aggregate populate (PR #47), aggregate snapshots (PR #60), and wrong_expected_version span event (PR #61)
+- Creates spans for event handlers (PR #41), EventStore operations (PR #37), aggregate execution (PR #45), application dispatch (PR #46), aggregate load (PR #58), aggregate populate (PR #47), aggregate snapshots (PR #60), wrong_expected_version span event (PR #61), and aggregate load trace context propagation (PR #90)
 - Added `opentelemetry_api`, `opentelemetry_telemetry`, and `opentelemetry_semantic_conventions` as required dependencies
 
 **Usage:**
@@ -273,6 +273,12 @@ end
 - OTel execute span records `commanded.aggregate.wrong_expected_version` event when count > 0
 - Enables alerting on optimistic concurrency conflicts via telemetry
 
+### **Aggregate Load Trace Context Propagation**
+[PR #90](https://github.com/straw-hat-team/commanded/pull/90)
+
+**Changes:**
+- Propagate the dispatch caller's W3C trace context (`traceparent`/`tracestate`) through aggregate startup so that `load` and `populate` spans become children of the dispatch trace instead of orphaned root spans
+
 ### **Event Handler Processing Latency Telemetry**
 [PR #66](https://github.com/straw-hat-team/commanded/pull/66)
 

lib/commanded/aggregates/aggregate.ex

@@ -221,14 +221,16 @@ defmodule Commanded.Aggregates.Aggregate do
     snapshotting = Keyword.get(config, :snapshotting, %{})
     snapshot_options = Map.get(snapshotting, aggregate_module, [])
 
+    metadata = Keyword.get(aggregate_opts, :metadata, %{})
+
     state = %Aggregate{
       application: application,
       aggregate_module: aggregate_module,
       aggregate_uuid: aggregate_uuid,
       snapshotting: Snapshotting.new(application, aggregate_uuid, snapshot_options)
     }
 
-    GenServer.start_link(__MODULE__, state, start_opts)
+    GenServer.start_link(__MODULE__, {state, metadata}, start_opts)
   end
 
   @doc false
@@ -336,16 +338,14 @@ defmodule Commanded.Aggregates.Aggregate do
 
   @doc false
   @impl GenServer
-  def init(%Aggregate{} = state) do
-    # Initial aggregate state is populated by loading its state snapshot and/or
-    # events from the event store.
-    {:ok, state, {:continue, :populate_aggregate_state}}
+  def init({%Aggregate{} = state, metadata}) do
+    {:ok, state, {:continue, {:populate_aggregate_state, metadata}}}
   end
 
   @doc false
   @impl GenServer
-  def handle_continue(:populate_aggregate_state, %Aggregate{} = state) do
-    state = AggregateStateBuilder.populate(state)
+  def handle_continue({:populate_aggregate_state, metadata}, %Aggregate{} = state) do
+    %Aggregate{} = state = AggregateStateBuilder.populate(state, metadata)
 
     # Subscribe to aggregate's events to catch any events appended to its stream
     # by another process, such as directly appended to the event store.

lib/commanded/aggregates/aggregate_state_builder.ex

@@ -72,7 +72,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
   Otherwise start with the aggregate struct and stream all existing events for
   the aggregate from the event store to rebuild its state from those events.
   """
-  def populate(%Aggregate{} = state) do
+  def populate(%Aggregate{} = state, metadata \\ %{}) do
     %Aggregate{aggregate_module: aggregate_module, snapshotting: snapshotting} = state
 
     {aggregate, snapshot_used, snapshot_source_version} =
@@ -98,7 +98,8 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
 
     rebuild_from_events(aggregate,
       snapshot_used: snapshot_used,
-      snapshot_source_version: snapshot_source_version
+      snapshot_source_version: snapshot_source_version,
+      metadata: metadata
     )
   end
 
@@ -113,9 +114,10 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
   def rebuild_from_events(%Aggregate{} = state, opts \\ []) do
     snapshot_used = Keyword.get(opts, :snapshot_used, false)
     snapshot_source_version = Keyword.get(opts, :snapshot_source_version)
+    metadata = Keyword.get(opts, :metadata, %{})
 
     load_prefix = [:commanded, :aggregate, :load]
-    meta = telemetry_metadata(state)
+    meta = telemetry_metadata(state, metadata)
     load_start = Telemetry.start(load_prefix, meta)
 
     %Aggregate{
@@ -190,7 +192,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
     })
   end
 
-  defp telemetry_metadata(%Aggregate{} = state) do
+  defp telemetry_metadata(%Aggregate{} = state, metadata \\ %{}) do
     %Aggregate{
       application: application,
       aggregate_module: aggregate_module,
@@ -204,7 +206,8 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do
       aggregate_module: aggregate_module,
       aggregate_uuid: aggregate_uuid,
       aggregate_state: aggregate_state,
-      aggregate_version: aggregate_version
+      aggregate_version: aggregate_version,
+      metadata: metadata
     }
   end
 end

lib/commanded/aggregates/supervisor.ex

@@ -24,7 +24,9 @@ defmodule Commanded.Aggregates.Supervisor do
   Returns `{:ok, aggregate_uuid}` when a process is successfully started, or is
   already running.
   """
-  def open_aggregate(application, aggregate_module, aggregate_uuid)
+  def open_aggregate(application, aggregate_module, aggregate_uuid, metadata \\ %{})
+
+  def open_aggregate(application, aggregate_module, aggregate_uuid, metadata)
       when is_atom(application) and is_atom(aggregate_module) and is_binary(aggregate_uuid) do
     Logger.debug(fn ->
       "Locating aggregate process for `#{inspect(aggregate_module)}` with UUID " <>
@@ -37,7 +39,8 @@ defmodule Commanded.Aggregates.Supervisor do
     args = [
       application: application,
       aggregate_module: aggregate_module,
-      aggregate_uuid: aggregate_uuid
+      aggregate_uuid: aggregate_uuid,
+      metadata: metadata
     ]
 
     case Registration.start_child(application, aggregate_name, supervisor_name, {Aggregate, args}) do
@@ -55,7 +58,7 @@ defmodule Commanded.Aggregates.Supervisor do
     end
   end
 
-  def open_aggregate(_application, _aggregate_module, aggregate_uuid),
+  def open_aggregate(_application, _aggregate_module, aggregate_uuid, _metadata),
     do: {:error, {:unsupported_aggregate_identity_type, aggregate_uuid}}
 
   def init(args) do

lib/commanded/commands/dispatcher.ex

@@ -116,7 +116,8 @@ defmodule Commanded.Commands.Dispatcher do
       Commanded.Aggregates.Supervisor.open_aggregate(
         application,
         aggregate_module,
-        aggregate_uuid
+        aggregate_uuid,
+        context.metadata
       )
 
     task_dispatcher_name = Module.concat([application, Commanded.Commands.TaskDispatcher])

lib/commanded/opentelemetry/aggregate_populate.ex

@@ -30,6 +30,11 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do
         meta,
         _config
       ) do
+    case Helpers.extract_propagated_ctx(meta[:metadata]) do
+      {_links, :undefined} -> :ok
+      {_links, ctx} -> :otel_ctx.attach(ctx)
+    end
+
     aggregate_module_name = Helpers.module_name(meta.aggregate_module)
 
     attributes = [

test/opentelemetry/aggregate_populate_test.exs

@@ -11,6 +11,8 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do
   alias Commanded.TestSupport.Factory
   alias Commanded.UUID
 
+  require OpenTelemetry.Tracer, as: Tracer
+
   setup do
     detach_populate_handlers()
     AggregatePopulate.setup()
@@ -259,6 +261,133 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do
     end
   end
 
+  describe "trace context propagation" do
+    setup do
+      detach_populate_handlers()
+      AggregatePopulate.setup()
+      :ok
+    end
+
+    test "load span becomes child of caller when traceparent is in metadata" do
+      {parent_trace_id, parent_span_id, traceparent} =
+        Tracer.with_span "parent.command.dispatch" do
+          ctx = Tracer.current_span_ctx()
+
+          {
+            :otel_span.trace_id(ctx),
+            :otel_span.span_id(ctx),
+            encode_traceparent(ctx)
+          }
+        end
+
+      meta =
+        Factory.build_aggregate_populate_metadata(
+          metadata: %{"traceparent" => traceparent}
+        )
+
+      :telemetry.execute([:commanded, :aggregate, :load, :start], %{}, meta)
+
+      stop_meta =
+        Factory.build_aggregate_load_stop_metadata(meta,
+          snapshot_used: false,
+          aggregate_version: 0
+        )
+
+      :telemetry.execute([:commanded, :aggregate, :load, :stop], %{count: 0}, stop_meta)
+
+      assert_receive {:span,
+                      span(
+                        name: "load MockAggregate",
+                        trace_id: child_trace_id,
+                        parent_span_id: received_parent_span_id
+                      )},
+                     1000
+
+      assert child_trace_id == parent_trace_id
+      assert received_parent_span_id == parent_span_id
+    end
+
+    test "load span is independent when no traceparent in metadata" do
+      meta = Factory.build_aggregate_populate_metadata(metadata: %{})
+
+      :telemetry.execute([:commanded, :aggregate, :load, :start], %{}, meta)
+
+      stop_meta =
+        Factory.build_aggregate_load_stop_metadata(meta,
+          snapshot_used: false,
+          aggregate_version: 0
+        )
+
+      :telemetry.execute([:commanded, :aggregate, :load, :stop], %{count: 0}, stop_meta)
+
+      assert_receive {:span,
+                      span(
+                        name: "load MockAggregate",
+                        parent_span_id: :undefined
+                      )},
+                     1000
+    end
+
+    test "load span is independent when traceparent is invalid" do
+      meta =
+        Factory.build_aggregate_populate_metadata(
+          metadata: %{"traceparent" => "invalid-format"}
+        )
+
+      :telemetry.execute([:commanded, :aggregate, :load, :start], %{}, meta)
+
+      stop_meta =
+        Factory.build_aggregate_load_stop_metadata(meta,
+          snapshot_used: false,
+          aggregate_version: 0
+        )
+
+      :telemetry.execute([:commanded, :aggregate, :load, :stop], %{count: 0}, stop_meta)
+
+      assert_receive {:span,
+                      span(
+                        name: "load MockAggregate",
+                        parent_span_id: :undefined
+                      )},
+                     1000
+    end
+
+    test "load span is independent when metadata key is nil" do
+      meta =
+        Factory.build_aggregate_populate_metadata()
+        |> Map.put(:metadata, nil)
+
+      :telemetry.execute([:commanded, :aggregate, :load, :start], %{}, meta)
+
+      stop_meta =
+        Factory.build_aggregate_load_stop_metadata(meta,
+          snapshot_used: false,
+          aggregate_version: 0
+        )
+
+      :telemetry.execute([:commanded, :aggregate, :load, :stop], %{count: 0}, stop_meta)
+
+      assert_receive {:span,
+                      span(
+                        name: "load MockAggregate",
+                        parent_span_id: :undefined
+                      )},
+                     1000
+    end
+  end
+
+  defp encode_traceparent(span_ctx) do
+    trace_id = :otel_span.trace_id(span_ctx)
+    span_id = :otel_span.span_id(span_ctx)
+    trace_flags = span_ctx(span_ctx, :trace_flags)
+
+    hex_trace_id = :io_lib.format("~32.16.0b", [trace_id]) |> IO.iodata_to_binary()
+    hex_span_id = :io_lib.format("~16.16.0b", [span_id]) |> IO.iodata_to_binary()
+    hex_flags = :io_lib.format("~2.16.0b", [trace_flags]) |> IO.iodata_to_binary()
+
+    "00-#{hex_trace_id}-#{hex_span_id}-#{hex_flags}"
+  end
+
   defp detach_populate_handlers do
     for event <- [
           [:commanded, :aggregate, :load, :start],

test/support/factory.ex

@@ -643,7 +643,8 @@ defmodule Commanded.TestSupport.Factory do
       aggregate_module: Keyword.fetch!(opts, :aggregate_module),
       aggregate_uuid: Keyword.fetch!(opts, :aggregate_uuid),
       aggregate_state: Keyword.fetch!(opts, :aggregate_state),
-      aggregate_version: Keyword.fetch!(opts, :aggregate_version)
+      aggregate_version: Keyword.fetch!(opts, :aggregate_version),
+      metadata: Keyword.get(opts, :metadata, %{})
     }
   end
 

test/support/opentelemetry_case.ex

@@ -29,10 +29,11 @@ defmodule Commanded.OpenTelemetryCase do
     :application.set_env(:opentelemetry, :tracer, :otel_tracer_default)
 
     :application.set_env(:opentelemetry, :processors, [
-      {:otel_batch_processor, %{scheduled_delay_ms: 1, exporter: {:otel_exporter_pid, self()}}}
+      {:otel_batch_processor, %{scheduled_delay_ms: 1}}
     ])
 
     :application.start(:opentelemetry)
+    :otel_batch_processor.set_exporter(:otel_exporter_pid, self())
 
     on_exit(fn ->
       commanded_events = [