From aeb38e0e324e6e2d7f873fe21f1f6d1b79446e66 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 20 Apr 2026 16:27:51 -0400 Subject: [PATCH] feat(telemetry): propagate caller trace context to aggregate load spans Signed-off-by: Yordis Prieto --- guides/explanations/fork-differences.md | 10 +- lib/commanded/aggregates/aggregate.ex | 14 +- .../aggregates/aggregate_state_builder.ex | 13 +- lib/commanded/aggregates/supervisor.ex | 9 +- lib/commanded/commands/dispatcher.ex | 3 +- .../opentelemetry/aggregate_populate.ex | 5 + .../opentelemetry/aggregate_populate_test.exs | 129 ++++++++++++++++++ test/support/factory.ex | 3 +- test/support/opentelemetry_case.ex | 3 +- 9 files changed, 169 insertions(+), 20 deletions(-) diff --git a/guides/explanations/fork-differences.md b/guides/explanations/fork-differences.md index b2011318..a3fad976 100644 --- a/guides/explanations/fork-differences.md +++ b/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) diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index 55fcda6c..2001aac2 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/lib/commanded/aggregates/aggregate.ex @@ -221,6 +221,8 @@ 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, @@ -228,7 +230,7 @@ defmodule Commanded.Aggregates.Aggregate do 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. diff --git a/lib/commanded/aggregates/aggregate_state_builder.ex b/lib/commanded/aggregates/aggregate_state_builder.ex index a267abc3..26da059b 100644 --- a/lib/commanded/aggregates/aggregate_state_builder.ex +++ b/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 diff --git a/lib/commanded/aggregates/supervisor.ex b/lib/commanded/aggregates/supervisor.ex index 66431c48..2d51a5df 100644 --- a/lib/commanded/aggregates/supervisor.ex +++ b/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 diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex index efb5bc42..4169d0b2 100644 --- a/lib/commanded/commands/dispatcher.ex +++ b/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]) diff --git a/lib/commanded/opentelemetry/aggregate_populate.ex b/lib/commanded/opentelemetry/aggregate_populate.ex index 37fab779..72db4bd1 100644 --- a/lib/commanded/opentelemetry/aggregate_populate.ex +++ b/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 = [ diff --git a/test/opentelemetry/aggregate_populate_test.exs b/test/opentelemetry/aggregate_populate_test.exs index de9453f6..8d584e80 100644 --- a/test/opentelemetry/aggregate_populate_test.exs +++ b/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], diff --git a/test/support/factory.ex b/test/support/factory.ex index bd99246a..a610055f 100644 --- a/test/support/factory.ex +++ b/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 diff --git a/test/support/opentelemetry_case.ex b/test/support/opentelemetry_case.ex index a8d59d5e..54a62cc4 100644 --- a/test/support/opentelemetry_case.ex +++ b/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 = [