diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index f16057ad..33cd7334 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/lib/commanded/aggregates/aggregate.ex @@ -64,6 +64,49 @@ defmodule Commanded.Aggregates.Aggregate do """ }) + telemetry_event(%{ + event: [:commanded, :aggregate, :snapshot, :start], + description: "Emitted when an aggregate begins taking a snapshot", + measurements: "%{system_time: integer()}", + metadata: """ + %{application: Commanded.Application.t(), + aggregate_uuid: String.t(), + aggregate_version: non_neg_integer(), + snapshot_every: non_neg_integer() | nil, + snapshot_module_version: non_neg_integer()} + """ + }) + + telemetry_event(%{ + event: [:commanded, :aggregate, :snapshot, :stop], + description: "Emitted when an aggregate completes taking a snapshot", + measurements: "%{duration: non_neg_integer()}", + metadata: """ + %{application: Commanded.Application.t(), + aggregate_uuid: String.t(), + aggregate_version: non_neg_integer(), + snapshot_every: non_neg_integer() | nil, + snapshot_module_version: non_neg_integer(), + error: nil | any()} + """ + }) + + telemetry_event(%{ + event: [:commanded, :aggregate, :snapshot, :exception], + description: "Emitted when an aggregate raises during snapshot", + measurements: "%{duration: non_neg_integer()}", + metadata: """ + %{application: Commanded.Application.t(), + aggregate_uuid: String.t(), + aggregate_version: non_neg_integer(), + snapshot_every: non_neg_integer() | nil, + snapshot_module_version: non_neg_integer(), + kind: :throw | :error | :exit, + reason: any(), + stacktrace: list()} + """ + }) + @moduledoc """ Aggregate is a `GenServer` process used to provide access to an instance of an event sourced aggregate. @@ -628,22 +671,42 @@ defmodule Commanded.Aggregates.Aggregate do defp do_take_snapshot(%Aggregate{} = state) do %Aggregate{ + application: application, + aggregate_uuid: aggregate_uuid, aggregate_state: aggregate_state, aggregate_version: aggregate_version, - snapshotting: snapshotting + snapshotting: %Snapshotting{ + snapshot_every: snapshot_every, + snapshot_module_version: snapshot_module_version + } } = state - Logger.debug(describe(state) <> " recording snapshot") + meta = %{ + application: application, + aggregate_uuid: aggregate_uuid, + aggregate_version: aggregate_version, + snapshot_every: snapshot_every, + snapshot_module_version: snapshot_module_version + } + + :telemetry.span([:commanded, :aggregate, :snapshot], meta, fn -> + Logger.debug(describe(state) <> " recording snapshot") - case Snapshotting.take_snapshot(snapshotting, aggregate_version, aggregate_state) do - {:ok, snapshotting} -> - {:ok, %Aggregate{state | snapshotting: snapshotting}} + result = + case Snapshotting.take_snapshot(state.snapshotting, aggregate_version, aggregate_state) do + {:ok, snapshotting} -> + {:ok, %Aggregate{state | snapshotting: snapshotting}} - {:error, reason} = error -> - Logger.warning(describe(state) <> " snapshot failed due to: " <> inspect(reason)) + {:error, reason} = error -> + Logger.warning(describe(state) <> " snapshot failed due to: " <> inspect(reason)) + error + end - error - end + stop_meta = + Map.put(meta, :error, if(match?({:error, _}, result), do: elem(result, 1), else: nil)) + + {result, stop_meta} + end) end defp telemetry_wrong_expected_version(context, from, state) do diff --git a/lib/commanded/aggregates/aggregate_state_builder.ex b/lib/commanded/aggregates/aggregate_state_builder.ex index 5abc89ca..a9fd35f6 100644 --- a/lib/commanded/aggregates/aggregate_state_builder.ex +++ b/lib/commanded/aggregates/aggregate_state_builder.ex @@ -28,7 +28,9 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do %{application: Commanded.Application.t(), aggregate_uuid: String.t(), aggregate_state: struct(), - aggregate_version: non_neg_integer()} + aggregate_version: non_neg_integer(), + snapshot_used: boolean(), + snapshot_source_version: non_neg_integer() | nil} """ }) @@ -69,29 +71,48 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do def populate(%Aggregate{} = state) do %Aggregate{aggregate_module: aggregate_module, snapshotting: snapshotting} = state - aggregate = + {aggregate, snapshot_used, snapshot_source_version} = case Snapshotting.read_snapshot(snapshotting) do {:ok, %SnapshotData{source_version: source_version, data: data}} -> - %Aggregate{ + agg = %Aggregate{ state | aggregate_version: source_version, aggregate_state: data } + {agg, true, source_version} + {:error, _error} -> - # No snapshot present, or exists but for outdated state, so use initial empty state - %Aggregate{state | aggregate_version: 0, aggregate_state: struct(aggregate_module)} + agg = %Aggregate{ + state + | aggregate_version: 0, + aggregate_state: struct(aggregate_module) + } + + {agg, false, nil} end - rebuild_from_events(aggregate) + rebuild_from_events(aggregate, + snapshot_used: snapshot_used, + snapshot_source_version: snapshot_source_version + ) end @doc """ - Load events from the event store, in batches, to rebuild the aggregate state + Load events from the event store, in batches, to rebuild the aggregate state. + + ## Options + + * `:snapshot_used` - whether a snapshot was used as initial state (default: `false`) + * `:snapshot_source_version` - version of the snapshot, if used (default: `nil`) """ - def rebuild_from_events(%Aggregate{} = state) 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) + load_prefix = [:commanded, :aggregate, :load] - load_start = Telemetry.start(load_prefix, telemetry_metadata(state)) + meta = telemetry_metadata(state) + load_start = Telemetry.start(load_prefix, meta) %Aggregate{ application: application, @@ -107,13 +128,25 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do @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}) + Telemetry.stop( + load_prefix, + load_start, + load_stop_metadata(state, snapshot_used, snapshot_source_version), + %{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}) + + Telemetry.stop( + load_prefix, + load_start, + load_stop_metadata(state, snapshot_used, snapshot_source_version), + %{count: count} + ) + {state, count} end @@ -145,6 +178,14 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do {state, count} end + defp load_stop_metadata(aggregate, snapshot_used, snapshot_source_version) do + telemetry_metadata(aggregate) + |> Map.merge(%{ + snapshot_used: snapshot_used, + snapshot_source_version: snapshot_source_version + }) + end + defp telemetry_metadata(%Aggregate{} = state) do %Aggregate{ application: application, diff --git a/lib/commanded/opentelemetry.ex b/lib/commanded/opentelemetry.ex index cdc625b7..c4073dc3 100644 --- a/lib/commanded/opentelemetry.ex +++ b/lib/commanded/opentelemetry.ex @@ -38,6 +38,7 @@ defmodule Commanded.OpenTelemetry do alias Commanded.OpenTelemetry.Aggregate alias Commanded.OpenTelemetry.AggregatePopulate + alias Commanded.OpenTelemetry.AggregateSnapshot alias Commanded.OpenTelemetry.Application, as: OTelApplication alias Commanded.OpenTelemetry.EventHandler alias Commanded.OpenTelemetry.EventStore @@ -65,6 +66,11 @@ defmodule Commanded.OpenTelemetry do default: [], doc: "Aggregate populate tracing configuration. Use `:disabled` to disable." ], + aggregate_snapshot: [ + type: {:in, [:disabled, []]}, + default: [], + doc: "Aggregate snapshot tracing configuration. Use `:disabled` to disable." + ], application: [ type: {:in, [:disabled, []]}, default: [], @@ -120,6 +126,9 @@ defmodule Commanded.OpenTelemetry do # Disable aggregate populate tracing Commanded.OpenTelemetry.setup(aggregate_populate: :disabled) + # Disable aggregate snapshot tracing + Commanded.OpenTelemetry.setup(aggregate_snapshot: :disabled) + # Disable event store tracing Commanded.OpenTelemetry.setup(event_store: :disabled) @@ -141,6 +150,11 @@ defmodule Commanded.OpenTelemetry do _config -> AggregatePopulate.setup() end + case opts[:aggregate_snapshot] do + :disabled -> :ok + _config -> AggregateSnapshot.setup() + end + case opts[:application] do :disabled -> :ok _config -> OTelApplication.setup() diff --git a/lib/commanded/opentelemetry/aggregate_populate.ex b/lib/commanded/opentelemetry/aggregate_populate.ex index 9ad5b43e..aea354d1 100644 --- a/lib/commanded/opentelemetry/aggregate_populate.ex +++ b/lib/commanded/opentelemetry/aggregate_populate.ex @@ -67,6 +67,20 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do meta.aggregate_version ) + Span.set_attribute( + ctx, + CommandedAttributes.commanded_snapshot_used(), + meta[:snapshot_used] || false + ) + + if snapshot_source_version = meta[:snapshot_source_version] do + Span.set_attribute( + ctx, + CommandedAttributes.commanded_snapshot_source_version(), + snapshot_source_version + ) + end + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) end diff --git a/lib/commanded/opentelemetry/aggregate_snapshot.ex b/lib/commanded/opentelemetry/aggregate_snapshot.ex new file mode 100644 index 00000000..0dad5e4b --- /dev/null +++ b/lib/commanded/opentelemetry/aggregate_snapshot.ex @@ -0,0 +1,121 @@ +defmodule Commanded.OpenTelemetry.AggregateSnapshot do + @moduledoc false + + alias Commanded.OpenTelemetry.CommandedAttributes + alias Commanded.OpenTelemetry.Helpers + alias OpenTelemetry.SemConv.ErrorAttributes + alias OpenTelemetry.SemConv.Incubating.CodeAttributes + alias OpenTelemetry.SemConv.Incubating.MessagingAttributes + alias OpenTelemetry.Span + + @tracer_id __MODULE__ + + def setup do + :ok = + :telemetry.attach_many( + {__MODULE__, :snapshot}, + [ + [:commanded, :aggregate, :snapshot, :start], + [:commanded, :aggregate, :snapshot, :stop], + [:commanded, :aggregate, :snapshot, :exception] + ], + &__MODULE__.handle_telemetry_event/4, + %{} + ) + end + + def handle_telemetry_event( + [:commanded, :aggregate, :snapshot, :start], + _measurements, + meta, + _config + ) do + attributes = + [ + {MessagingAttributes.messaging_system(), "commanded"}, + {MessagingAttributes.messaging_operation_type(), :publish}, + {MessagingAttributes.messaging_operation_name(), "snapshot"}, + {CodeAttributes.code_function(), "snapshot"}, + {CommandedAttributes.commanded_application(), meta.application}, + {CommandedAttributes.commanded_aggregate_uuid(), meta.aggregate_uuid}, + {CommandedAttributes.commanded_aggregate_version(), meta.aggregate_version} + ] + |> maybe_add_snapshot_every(meta[:snapshot_every]) + |> maybe_add_snapshot_module_version(meta[:snapshot_module_version]) + + OpentelemetryTelemetry.start_telemetry_span( + @tracer_id, + "commanded.aggregate.snapshot", + meta, + %{ + kind: :internal, + attributes: attributes + } + ) + end + + def handle_telemetry_event( + [:commanded, :aggregate, :snapshot, :stop], + _measurements, + meta, + _config + ) do + ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) + + if error = meta[:error] do + Span.set_attribute( + ctx, + ErrorAttributes.error_type(), + Helpers.to_error_type(error, @tracer_id) + ) + + Span.set_status(ctx, OpenTelemetry.status(:error, Helpers.format_error(error))) + end + + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) + end + + def handle_telemetry_event( + [:commanded, :aggregate, :snapshot, :exception], + _measurements, + %{kind: kind, reason: reason, stacktrace: stacktrace} = meta, + _config + ) do + ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) + + Span.set_attribute(ctx, :"erlang.exception.kind", kind) + + exception = Exception.normalize(kind, reason, stacktrace) + + Span.set_attribute( + ctx, + ErrorAttributes.error_type(), + Helpers.to_error_type(exception, @tracer_id) + ) + + Span.record_exception(ctx, exception, stacktrace) + + Span.set_status( + ctx, + OpenTelemetry.status(:error, Exception.format_banner(kind, reason, stacktrace)) + ) + + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) + end + + defp maybe_add_snapshot_every(attrs, nil), do: attrs + + defp maybe_add_snapshot_every(attrs, snapshot_every) when is_integer(snapshot_every) do + [{CommandedAttributes.commanded_snapshot_every(), snapshot_every} | attrs] + end + + defp maybe_add_snapshot_every(attrs, _), do: attrs + + defp maybe_add_snapshot_module_version(attrs, nil), do: attrs + + defp maybe_add_snapshot_module_version(attrs, version) when is_integer(version) do + [{CommandedAttributes.commanded_snapshot_module_version(), version} | attrs] + end + + defp maybe_add_snapshot_module_version(attrs, _), do: attrs +end diff --git a/lib/commanded/opentelemetry/commanded_attributes.ex b/lib/commanded/opentelemetry/commanded_attributes.ex index 329f3f61..b841691b 100644 --- a/lib/commanded/opentelemetry/commanded_attributes.ex +++ b/lib/commanded/opentelemetry/commanded_attributes.ex @@ -141,4 +141,28 @@ defmodule Commanded.OpenTelemetry.CommandedAttributes do """ @spec commanded_batch_last_event_id() :: :"commanded.batch.last_event_id" def commanded_batch_last_event_id, do: :"commanded.batch.last_event_id" + + @doc """ + Whether a snapshot was used during aggregate populate. + """ + @spec commanded_snapshot_used() :: :"commanded.snapshot.used" + def commanded_snapshot_used, do: :"commanded.snapshot.used" + + @doc """ + The version the snapshot was taken at. + """ + @spec commanded_snapshot_source_version() :: :"commanded.snapshot.source_version" + def commanded_snapshot_source_version, do: :"commanded.snapshot.source_version" + + @doc """ + The configured snapshot_every value (events between snapshots). + """ + @spec commanded_snapshot_every() :: :"commanded.snapshot.every" + def commanded_snapshot_every, do: :"commanded.snapshot.every" + + @doc """ + The configured snapshot module version. + """ + @spec commanded_snapshot_module_version() :: :"commanded.snapshot.module_version" + def commanded_snapshot_module_version, do: :"commanded.snapshot.module_version" end diff --git a/test/aggregates/aggregate_telemetry_test.exs b/test/aggregates/aggregate_telemetry_test.exs index f707444f..8660883a 100644 --- a/test/aggregates/aggregate_telemetry_test.exs +++ b/test/aggregates/aggregate_telemetry_test.exs @@ -304,8 +304,9 @@ defmodule Commanded.Aggregates.AggregateTelemetryTest do end end - @tag :unit describe "load/populate telemetry (unit: MockedApp + expect)" do + @describetag :unit + setup do attach_telemetry() :ok diff --git a/test/aggregates/snapshotting_postgres_test.exs b/test/aggregates/snapshotting_postgres_test.exs new file mode 100644 index 00000000..2a9ecae4 --- /dev/null +++ b/test/aggregates/snapshotting_postgres_test.exs @@ -0,0 +1,84 @@ +defmodule Commanded.Aggregates.SnapshottingPostgresTest do + use ExUnit.Case + + @moduletag :eventstore_adapter + + alias Commanded.Aggregates.{ + Aggregate, + AppendItemsHandler, + ExampleAggregate, + ExecutionContext, + Supervisor + } + + alias Commanded.Aggregates.ExampleAggregate.Commands.AppendItems + alias Commanded.EventStore + alias Commanded.EventStore.SnapshotData + alias Commanded.UUID + + defmodule App do + use Commanded.Application, + otp_app: :commanded, + event_store: [ + adapter: Commanded.EventStore.Adapters.EventStore, + event_store: TestEventStore + ], + pubsub: :local, + registry: :local + end + + setup do + start_event_store() + start_supervised!({App, snapshotting: %{ExampleAggregate => [snapshot_every: 10]}}) + :ok + end + + test "snapshot data from PostgreSQL matches expected structure" do + aggregate_uuid = UUID.uuid4() + append_items(aggregate_uuid, 10) + + assert Aggregate.aggregate_version(App, ExampleAggregate, aggregate_uuid) == 10 + + assert {:ok, snapshot} = EventStore.read_snapshot(App, aggregate_uuid) + + assert %SnapshotData{ + source_uuid: ^aggregate_uuid, + source_version: 10, + source_type: "Elixir.Commanded.Aggregates.ExampleAggregate", + data: %ExampleAggregate{ + items: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], + last_index: 10 + }, + metadata: %{"snapshot_module_version" => 1} + } = snapshot + + assert is_struct(snapshot.data, ExampleAggregate) + assert snapshot.data.items == Enum.to_list(1..10) + assert snapshot.data.last_index == 10 + end + + defp append_items(aggregate_uuid, count) do + execution_context = %ExecutionContext{ + command: %AppendItems{count: count}, + handler: AppendItemsHandler, + function: :handle + } + + {:ok, ^aggregate_uuid} = + Supervisor.open_aggregate(App, ExampleAggregate, aggregate_uuid) + + {:ok, _count, _events, _aggregate_state} = + Aggregate.execute(App, ExampleAggregate, aggregate_uuid, execution_context) + end + + defp start_event_store do + alias Commanded.EventStore.Adapters.EventStore.Storage + + config = Storage.config() + + on_exit(fn -> + {:ok, conn} = Storage.connect(config) + Storage.reset!(conn, config) + end) + end +end diff --git a/test/opentelemetry/aggregate_populate_test.exs b/test/opentelemetry/aggregate_populate_test.exs index e2e6cdf6..cda77349 100644 --- a/test/opentelemetry/aggregate_populate_test.exs +++ b/test/opentelemetry/aggregate_populate_test.exs @@ -111,7 +111,15 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do ) :telemetry.execute([:commanded, :aggregate, :load, :start], %{}, meta) - :telemetry.execute([:commanded, :aggregate, :load, :stop], %{count: 0}, meta) + + stop_meta = + Factory.build_aggregate_load_stop_metadata(meta, + snapshot_used: false, + snapshot_source_version: nil, + aggregate_version: 0 + ) + + :telemetry.execute([:commanded, :aggregate, :load, :stop], %{count: 0}, stop_meta) assert_receive {:span, span( @@ -129,9 +137,44 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do "commanded.application": MockApp, "commanded.aggregate.uuid": aggregate_uuid, "commanded.aggregate.version": 0, - "commanded.event.count": 0 + "commanded.event.count": 0, + "commanded.snapshot.used": false } end + + test "emits load span with snapshot attributes when snapshot was used" do + aggregate_uuid = UUID.uuid4() + + meta = + Factory.build_aggregate_populate_metadata( + aggregate_uuid: aggregate_uuid, + aggregate_version: 5 + ) + + :telemetry.execute([:commanded, :aggregate, :load, :start], %{}, meta) + + stop_meta = + Factory.build_aggregate_load_stop_metadata(meta, + snapshot_used: true, + snapshot_source_version: 5, + aggregate_version: 5 + ) + + :telemetry.execute([:commanded, :aggregate, :load, :stop], %{count: 0}, stop_meta) + + assert_receive {:span, + span( + name: "commanded.aggregate.load", + kind: :internal, + attributes: attributes + )}, + 1000 + + attrs = :otel_attributes.map(attributes) + + assert attrs[:"commanded.snapshot.used"] == true + assert attrs[:"commanded.snapshot.source_version"] == 5 + end end describe "edge cases" do diff --git a/test/opentelemetry/aggregate_snapshot_test.exs b/test/opentelemetry/aggregate_snapshot_test.exs new file mode 100644 index 00000000..eeb410b9 --- /dev/null +++ b/test/opentelemetry/aggregate_snapshot_test.exs @@ -0,0 +1,167 @@ +defmodule Commanded.OpenTelemetry.AggregateSnapshotTest do + @moduledoc """ + Tests for AggregateSnapshot OpenTelemetry instrumentation. + """ + + use Commanded.OpenTelemetryCase, async: false + + alias Commanded.OpenTelemetry.AggregateSnapshot + alias Commanded.TestSupport.Factory + alias Commanded.UUID + + setup do + detach_snapshot_handlers() + AggregateSnapshot.setup() + + :ok + end + + describe "setup/0" do + test "attaches telemetry handlers for aggregate snapshot events" do + detach_snapshot_handlers() + + AggregateSnapshot.setup() + + for event <- [ + [:commanded, :aggregate, :snapshot, :start], + [:commanded, :aggregate, :snapshot, :stop], + [:commanded, :aggregate, :snapshot, :exception] + ] do + handlers = :telemetry.list_handlers(event) + + assert length(handlers) >= 1, + "Expected handler for event #{inspect(event)}" + end + end + + test "calling setup twice raises MatchError (fail fast)" do + detach_snapshot_handlers() + + :ok = AggregateSnapshot.setup() + + handlers = :telemetry.list_handlers([:commanded, :aggregate, :snapshot, :start]) + assert length(handlers) == 1 + + assert_raise MatchError, fn -> + AggregateSnapshot.setup() + end + end + end + + describe "snapshot spans" do + setup do + detach_snapshot_handlers() + AggregateSnapshot.setup() + :ok + end + + test "creates span with correct attributes" do + aggregate_uuid = UUID.uuid4() + + meta = + Factory.build_aggregate_snapshot_metadata( + aggregate_uuid: aggregate_uuid, + aggregate_version: 10, + snapshot_every: 5, + snapshot_module_version: 1 + ) + + :telemetry.span([:commanded, :aggregate, :snapshot], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "commanded.aggregate.snapshot", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :publish, + "messaging.operation.name": "snapshot", + "code.function": "snapshot", + "commanded.application": MockApp, + "commanded.aggregate.uuid": aggregate_uuid, + "commanded.aggregate.version": 10, + "commanded.snapshot.every": 5, + "commanded.snapshot.module_version": 1 + } + end + + test "creates span without optional snapshot attributes when nil" do + aggregate_uuid = UUID.uuid4() + + meta = + Factory.build_aggregate_snapshot_metadata( + aggregate_uuid: aggregate_uuid, + aggregate_version: 10, + snapshot_every: nil, + snapshot_module_version: nil + ) + + :telemetry.span([:commanded, :aggregate, :snapshot], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "commanded.aggregate.snapshot", + kind: :internal, + attributes: attributes + )}, + 1000 + + attrs = :otel_attributes.map(attributes) + + refute Map.has_key?(attrs, :"commanded.snapshot.every") + refute Map.has_key?(attrs, :"commanded.snapshot.module_version") + assert attrs[:"commanded.aggregate.uuid"] == aggregate_uuid + assert attrs[:"commanded.aggregate.version"] == 10 + end + + test "sets error status when stop includes error" do + aggregate_uuid = UUID.uuid4() + + meta = + Factory.build_aggregate_snapshot_metadata( + aggregate_uuid: aggregate_uuid, + aggregate_version: 10 + ) + + :telemetry.execute([:commanded, :aggregate, :snapshot, :start], %{}, meta) + + stop_meta = Map.put(meta, :error, :snapshotting_not_configured) + + :telemetry.execute( + [:commanded, :aggregate, :snapshot, :stop], + %{duration: 1000}, + stop_meta + ) + + assert_receive {:span, + span( + name: "commanded.aggregate.snapshot", + status: {:status, :error, _error_message}, + attributes: span_attrs + )}, + 1000 + + assert :otel_attributes.map(span_attrs)[:"error.type"] == "snapshotting_not_configured" + end + end + + defp detach_snapshot_handlers do + for event <- [ + [:commanded, :aggregate, :snapshot, :start], + [:commanded, :aggregate, :snapshot, :stop], + [:commanded, :aggregate, :snapshot, :exception] + ] do + for handler <- :telemetry.list_handlers(event) do + :telemetry.detach(handler.id) + end + end + end +end diff --git a/test/support/factory.ex b/test/support/factory.ex index e657d7f5..749e9cf1 100644 --- a/test/support/factory.ex +++ b/test/support/factory.ex @@ -616,6 +616,37 @@ defmodule Commanded.TestSupport.Factory do } end + def build_aggregate_load_stop_metadata(meta, opts \\ []) do + Map.merge(meta, %{ + snapshot_used: Keyword.get(opts, :snapshot_used, false), + snapshot_source_version: Keyword.get(opts, :snapshot_source_version), + aggregate_version: + Keyword.get(opts, :aggregate_version, Map.get(meta, :aggregate_version, 0)) + }) + end + + def build_aggregate_snapshot_metadata(opts \\ []) do + aggregate_uuid = Keyword.get(opts, :aggregate_uuid, UUID.uuid4()) + + defaults = [ + application: Keyword.get(opts, :application, MockApp), + aggregate_uuid: aggregate_uuid, + aggregate_version: Keyword.get(opts, :aggregate_version, 0), + snapshot_every: Keyword.get(opts, :snapshot_every), + snapshot_module_version: Keyword.get(opts, :snapshot_module_version) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + aggregate_uuid: Keyword.fetch!(opts, :aggregate_uuid), + aggregate_version: Keyword.fetch!(opts, :aggregate_version), + snapshot_every: Keyword.get(opts, :snapshot_every), + snapshot_module_version: Keyword.get(opts, :snapshot_module_version) + } + end + def build_telemetry_event(event_type, opts \\ []) def build_telemetry_event(:start, opts) do @@ -1131,11 +1162,13 @@ defmodule Commanded.TestSupport.Factory do snapshot = Keyword.get_lazy(opts, :snapshot, fn -> + account = build_account(account_id: source_uuid) + %Commanded.EventStore.SnapshotData{ source_uuid: source_uuid, source_version: 5, - source_type: "TestAggregate", - data: %{state: "test"}, + source_type: "Elixir.Commanded.TestSupport.TestDomain.Account", + data: account, metadata: %{}, created_at: DateTime.utc_now() } diff --git a/test/support/opentelemetry_case.ex b/test/support/opentelemetry_case.ex index 74ed0f9e..a8d59d5e 100644 --- a/test/support/opentelemetry_case.ex +++ b/test/support/opentelemetry_case.ex @@ -49,6 +49,9 @@ defmodule Commanded.OpenTelemetryCase do [:commanded, :aggregate, :load, :stop], [:commanded, :aggregate, :populate, :start], [:commanded, :aggregate, :populate, :stop], + [:commanded, :aggregate, :snapshot, :start], + [:commanded, :aggregate, :snapshot, :stop], + [:commanded, :aggregate, :snapshot, :exception], [:commanded, :application, :dispatch, :start], [:commanded, :application, :dispatch, :stop], [:commanded, :application, :dispatch, :exception]