diff --git a/guides/explanations/fork-differences.md b/guides/explanations/fork-differences.md index 6f49abeb..51982fc8 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) +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) **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), and aggregate populate (PR #47) +- Creates spans for event handlers (PR #41), EventStore operations (PR #37), aggregate execution (PR #45), application dispatch (PR #46), aggregate load (PR #58), and aggregate populate (PR #47) - Added `opentelemetry_api`, `opentelemetry_telemetry`, and `opentelemetry_semantic_conventions` as required dependencies **Usage:** @@ -243,3 +243,17 @@ end - Visualize event handler execution in your tracing backend - Correlate event processing with command dispatch using span links - Configurable span relationships (`:link`, `:child`, `:none`) + +### **Aggregate Load Telemetry** + +[PR #58](https://github.com/straw-hat-team/commanded/pull/58) + +**Changes:** +- Added `[:commanded, :aggregate, :load]` telemetry for full event store load (stream_forward + consumption) +- Load spans fire for both new aggregates (`stream_not_found`, `count: 0`) and existing aggregates +- Populate spans remain unchanged: only fire when applying events to rebuild state +- Trace hierarchy: `commanded.aggregate.load` (parent) → `commanded.aggregate.populate` (child, when events exist) + +**Benefits:** +- Measure event store read latency as a whole, including the "not found" case +- Separate load (event store I/O) from populate (state rebuild) in traces diff --git a/lib/commanded/aggregates/aggregate_state_builder.ex b/lib/commanded/aggregates/aggregate_state_builder.ex index 21634ae3..5abc89ca 100644 --- a/lib/commanded/aggregates/aggregate_state_builder.ex +++ b/lib/commanded/aggregates/aggregate_state_builder.ex @@ -7,9 +7,34 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do alias Commanded.Snapshotting alias Commanded.Telemetry + telemetry_event(%{ + event: [:commanded, :aggregate, :load, :start], + description: + "Emitted when an aggregate begins loading from the event store (stream_forward + consumption)", + measurements: "%{system_time: integer()}", + metadata: """ + %{application: Commanded.Application.t(), + aggregate_uuid: String.t(), + aggregate_state: struct(), + aggregate_version: non_neg_integer()} + """ + }) + + telemetry_event(%{ + event: [:commanded, :aggregate, :load, :stop], + description: "Emitted when an aggregate completes loading from the event store", + measurements: "%{duration: non_neg_integer(), count: non_neg_integer()}", + metadata: """ + %{application: Commanded.Application.t(), + aggregate_uuid: String.t(), + aggregate_state: struct(), + aggregate_version: non_neg_integer()} + """ + }) + telemetry_event(%{ event: [:commanded, :aggregate, :populate, :start], - description: "Emitted when an aggregate begins loading from the event store", + description: "Emitted when an aggregate begins applying events to rebuild state", measurements: "%{system_time: integer()}", metadata: """ %{application: Commanded.Application.t(), @@ -21,7 +46,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do telemetry_event(%{ event: [:commanded, :aggregate, :populate, :stop], - description: "Emitted when an aggregate completes loading from the event store", + description: "Emitted when an aggregate completes applying events to rebuild state", measurements: "%{duration: non_neg_integer(), count: non_neg_integer()}", metadata: """ %{application: Commanded.Application.t(), @@ -65,25 +90,35 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do Load events from the event store, in batches, to rebuild the aggregate state """ def rebuild_from_events(%Aggregate{} = state) do + load_prefix = [:commanded, :aggregate, :load] + load_start = Telemetry.start(load_prefix, telemetry_metadata(state)) + %Aggregate{ application: application, aggregate_uuid: aggregate_uuid, aggregate_version: aggregate_version } = state - case EventStore.stream_forward( - application, - aggregate_uuid, - aggregate_version + 1, - @read_event_batch_size - ) do - {:error, :stream_not_found} -> - # aggregate does not exist, return initial state - state - - event_stream -> - rebuild_from_event_stream(event_stream, state) - end + result = + case EventStore.stream_forward( + application, + aggregate_uuid, + aggregate_version + 1, + @read_event_batch_size + ) do + {:error, :stream_not_found} -> + # aggregate does not exist, return initial state + Telemetry.stop(load_prefix, load_start, telemetry_metadata(state), %{count: 0}) + {state, 0} + + event_stream -> + {state, count} = rebuild_from_event_stream(event_stream, state) + Telemetry.stop(load_prefix, load_start, telemetry_metadata(state), %{count: count}) + {state, count} + end + + {state, _count} = result + state end # Rebuild aggregate state from a `Stream` of its events. @@ -107,7 +142,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do Telemetry.stop(telemetry_prefix, start_time, telemetry_metadata(state), %{count: count}) - state + {state, count} end defp telemetry_metadata(%Aggregate{} = state) do diff --git a/lib/commanded/opentelemetry/aggregate_populate.ex b/lib/commanded/opentelemetry/aggregate_populate.ex index 12710e1f..9ad5b43e 100644 --- a/lib/commanded/opentelemetry/aggregate_populate.ex +++ b/lib/commanded/opentelemetry/aggregate_populate.ex @@ -11,8 +11,10 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do def setup do :ok = :telemetry.attach_many( - {__MODULE__, :populate}, + {__MODULE__, :load_populate}, [ + [:commanded, :aggregate, :load, :start], + [:commanded, :aggregate, :load, :stop], [:commanded, :aggregate, :populate, :start], [:commanded, :aggregate, :populate, :stop] ], @@ -21,6 +23,53 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do ) end + def handle_telemetry_event( + [:commanded, :aggregate, :load, :start], + _measurements, + meta, + _config + ) do + attributes = [ + {MessagingAttributes.messaging_system(), "commanded"}, + {MessagingAttributes.messaging_operation_type(), :receive}, + {MessagingAttributes.messaging_operation_name(), "load"}, + {CodeAttributes.code_function(), "load"}, + {CommandedAttributes.commanded_application(), meta.application}, + {CommandedAttributes.commanded_aggregate_uuid(), meta.aggregate_uuid}, + {CommandedAttributes.commanded_aggregate_version(), meta.aggregate_version} + ] + + OpentelemetryTelemetry.start_telemetry_span( + @tracer_id, + "commanded.aggregate.load", + meta, + %{ + kind: :internal, + attributes: attributes + } + ) + end + + def handle_telemetry_event( + [:commanded, :aggregate, :load, :stop], + measurements, + meta, + _config + ) do + ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) + + event_count = Map.get(measurements, :count, 0) + Span.set_attribute(ctx, CommandedAttributes.commanded_event_count(), event_count) + + Span.set_attribute( + ctx, + CommandedAttributes.commanded_aggregate_version(), + meta.aggregate_version + ) + + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) + end + def handle_telemetry_event( [:commanded, :aggregate, :populate, :start], _measurements, diff --git a/test/aggregates/aggregate_telemetry_test.exs b/test/aggregates/aggregate_telemetry_test.exs index bead0dcb..f707444f 100644 --- a/test/aggregates/aggregate_telemetry_test.exs +++ b/test/aggregates/aggregate_telemetry_test.exs @@ -2,7 +2,7 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do use Commanded.MockEventStoreCase alias Commanded.Aggregates.{Aggregate, ExecutionContext} - alias Commanded.{DefaultApp, UUID} + alias Commanded.{DefaultApp, MockedApp, UUID} defmodule Commands do defmodule Ok do @@ -193,7 +193,30 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do refute_received {[:commanded, :aggregate, :execute, :stop], _measurements, _metadata} end - test "emit `[:commanded, :aggregate, :populate]` events", + test "emit `[:commanded, :aggregate, :load]` events for new aggregate (stream_not_found)", + %{aggregate_uuid: aggregate_uuid} do + # Setup already started the aggregate. For a new aggregate, stream_forward returns + # {:error, :stream_not_found}. Load telemetry fires with count: 0; populate does not fire. + assert_receive {[:commanded, :aggregate, :load, :start], _measurements, _metadata} + assert_receive {[:commanded, :aggregate, :load, :stop], measurements, metadata} + + assert match?(%{count: 0}, measurements) + + assert match?( + %{ + aggregate_state: %ExampleAggregate{}, + aggregate_uuid: ^aggregate_uuid, + aggregate_version: 0, + application: DefaultApp + }, + metadata + ) + + refute_received {[:commanded, :aggregate, :populate, :start], _, _} + refute_received {[:commanded, :aggregate, :populate, :stop], _, _} + end + + test "emit `[:commanded, :aggregate, :load]` and `[:commanded, :aggregate, :populate]` for existing aggregate (reload)", %{aggregate_uuid: aggregate_uuid, pid: pid} do context = %ExecutionContext{ command: %Ok{message: "ok"}, @@ -210,13 +233,16 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do Process.exit(pid, :normal) - # Do the reload, we should now have telemetry - start_aggregate(aggregate_uuid) + # Do the reload. Consume initial load (count: 0) from setup, then reload load + populate. + assert_receive {[:commanded, :aggregate, :load, :start], _, _} + assert_receive {[:commanded, :aggregate, :load, :stop], %{count: 0}, _} - assert_receive {[:commanded, :aggregate, :populate, :start], _measurements, _metadata} - assert_receive {[:commanded, :aggregate, :populate, :stop], measurements, metadata} + start_aggregate(aggregate_uuid) - assert match?(%{count: ^count}, measurements) + assert_receive {[:commanded, :aggregate, :load, :start], _, _} + assert_receive {[:commanded, :aggregate, :populate, :start], _, _} + assert_receive {[:commanded, :aggregate, :populate, :stop], %{count: ^count}, metadata} + assert_receive {[:commanded, :aggregate, :load, :stop], %{count: ^count}, _} assert match?( %{ @@ -278,6 +304,63 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do end end + @tag :unit + describe "load/populate telemetry (unit: MockedApp + expect)" do + setup do + attach_telemetry() + :ok + end + + test "load only when stream_forward returns stream_not_found" do + aggregate_uuid = UUID.uuid4() + + expect(MockEventStore, :subscribe, fn _meta, ^aggregate_uuid -> :ok end) + + expect(MockEventStore, :stream_forward, fn _meta, ^aggregate_uuid, _from, _batch_size -> + {:error, :stream_not_found} + end) + + assert {:ok, _pid} = start_aggregate(aggregate_uuid, application: MockedApp) + + assert_receive {[:commanded, :aggregate, :load, :start], _, _} + assert_receive {[:commanded, :aggregate, :load, :stop], %{count: 0}, _} + + refute_received {[:commanded, :aggregate, :populate, :start], _, _} + refute_received {[:commanded, :aggregate, :populate, :stop], _, _} + end + + test "load + populate when stream_forward returns events" do + aggregate_uuid = UUID.uuid4() + count = 2 + + expect(MockEventStore, :subscribe, fn _meta, ^aggregate_uuid -> :ok end) + + expect(MockEventStore, :stream_forward, fn _meta, ^aggregate_uuid, _from, _batch_size -> + for i <- 1..count do + %Commanded.EventStore.RecordedEvent{ + event_id: UUID.uuid4(), + event_number: i, + stream_id: aggregate_uuid, + stream_version: i, + correlation_id: nil, + causation_id: nil, + event_type: "Elixir.Commanded.Aggregates.AggregateTelemetryTest.Event", + data: %Event{message: "event#{i}"}, + metadata: nil, + created_at: DateTime.utc_now() + } + end + end) + + assert {:ok, _pid} = start_aggregate(aggregate_uuid, application: MockedApp) + + assert_receive {[:commanded, :aggregate, :load, :start], _, _} + assert_receive {[:commanded, :aggregate, :populate, :start], _, _} + assert_receive {[:commanded, :aggregate, :populate, :stop], %{count: ^count}, _} + assert_receive {[:commanded, :aggregate, :load, :stop], %{count: ^count}, _} + end + end + def start_aggregate(aggregate_uuid) do Aggregate.start_link([application: DefaultApp], aggregate_module: ExampleAggregate, @@ -305,6 +388,8 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do [:commanded, :aggregate, :execute, :stop], [:commanded, :aggregate, :execute, :exception], [:commanded, :aggregate, :execute, :wrong_expected_version], + [:commanded, :aggregate, :load, :start], + [:commanded, :aggregate, :load, :stop], [:commanded, :aggregate, :populate, :start], [:commanded, :aggregate, :populate, :stop] ], diff --git a/test/opentelemetry/aggregate_populate_test.exs b/test/opentelemetry/aggregate_populate_test.exs index 807b80f3..e2e6cdf6 100644 --- a/test/opentelemetry/aggregate_populate_test.exs +++ b/test/opentelemetry/aggregate_populate_test.exs @@ -19,21 +19,20 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do end describe "setup/0" do - test "attaches telemetry handlers for aggregate populate events" do + test "attaches telemetry handlers for aggregate load and populate events" do detach_populate_handlers() AggregatePopulate.setup() for event <- [ + [:commanded, :aggregate, :load, :start], + [:commanded, :aggregate, :load, :stop], [:commanded, :aggregate, :populate, :start], [:commanded, :aggregate, :populate, :stop] ] do handlers = :telemetry.list_handlers(event) - assert Enum.any?( - handlers, - &match?(%{id: {AggregatePopulate, :populate}}, &1) - ), + assert length(handlers) >= 1, "Expected handler for event #{inspect(event)}" end end @@ -43,7 +42,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do :ok = AggregatePopulate.setup() - handlers = :telemetry.list_handlers([:commanded, :aggregate, :populate, :start]) + handlers = :telemetry.list_handlers([:commanded, :aggregate, :load, :start]) assert length(handlers) == 1 assert_raise MatchError, fn -> @@ -95,6 +94,46 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do end end + describe "load span (event store latency including stream_not_found)" do + setup do + detach_populate_handlers() + AggregatePopulate.setup() + :ok + end + + test "emits load span for stream_not_found (count: 0)" do + aggregate_uuid = UUID.uuid4() + + meta = + Factory.build_aggregate_populate_metadata( + aggregate_uuid: aggregate_uuid, + aggregate_version: 0 + ) + + :telemetry.execute([:commanded, :aggregate, :load, :start], %{}, meta) + :telemetry.execute([:commanded, :aggregate, :load, :stop], %{count: 0}, meta) + + assert_receive {:span, + span( + name: "commanded.aggregate.load", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "load", + "code.function": "load", + "commanded.application": MockApp, + "commanded.aggregate.uuid": aggregate_uuid, + "commanded.aggregate.version": 0, + "commanded.event.count": 0 + } + end + end + describe "edge cases" do setup do detach_populate_handlers() @@ -171,6 +210,8 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do defp detach_populate_handlers do for event <- [ + [:commanded, :aggregate, :load, :start], + [:commanded, :aggregate, :load, :stop], [:commanded, :aggregate, :populate, :start], [:commanded, :aggregate, :populate, :stop] ] do diff --git a/test/support/opentelemetry_case.ex b/test/support/opentelemetry_case.ex index 172c267a..74ed0f9e 100644 --- a/test/support/opentelemetry_case.ex +++ b/test/support/opentelemetry_case.ex @@ -45,6 +45,8 @@ defmodule Commanded.OpenTelemetryCase do [:commanded, :aggregate, :execute, :start], [:commanded, :aggregate, :execute, :stop], [:commanded, :aggregate, :execute, :exception], + [:commanded, :aggregate, :load, :start], + [:commanded, :aggregate, :load, :stop], [:commanded, :aggregate, :populate, :start], [:commanded, :aggregate, :populate, :stop], [:commanded, :application, :dispatch, :start],