From fdf1b3c04afe3adb2d9a5ff8018dd00f1e068a5e Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 20 Apr 2026 21:46:50 -0400 Subject: [PATCH] fix(telemetry): enrich EventStore spans with full OTel semantic attributes - Orphaned operations (ack_event, subscribe, subscribe_to, unsubscribe, delete_subscription) removed from EventStore instrumentation since they produce noise without meaningful tracing value - error.type now uses inspect() to drop the Elixir. prefix for cleaner APM display - Connection attributes (server.address, server.port, db.namespace) and peer.service set to the actual database name for proper service mapping - db.system derived from adapter type (postgresql for EventStore adapter, in_memory for InMemory) - event_count, start_version, read_batch_size added to span attributes - Shared attribute helpers extracted to Helpers module for reuse Signed-off-by: Yordis Prieto --- lib/commanded/event_store.ex | 3 +- .../opentelemetry/commanded_attributes.ex | 12 ++ lib/commanded/opentelemetry/event_store.ex | 77 +++----- lib/commanded/opentelemetry/helpers.ex | 62 +++++-- .../opentelemetry/aggregate_populate_test.exs | 8 +- .../opentelemetry/aggregate_snapshot_test.exs | 2 +- test/opentelemetry/aggregate_test.exs | 8 +- test/opentelemetry/application_test.exs | 8 +- test/opentelemetry/event_handler_test.exs | 12 +- test/opentelemetry/event_store_test.exs | 166 ++---------------- test/support/factory.ex | 6 +- 11 files changed, 122 insertions(+), 242 deletions(-) diff --git a/lib/commanded/event_store.ex b/lib/commanded/event_store.ex index 50b79d65..fd515a6c 100644 --- a/lib/commanded/event_store.ex +++ b/lib/commanded/event_store.ex @@ -36,7 +36,8 @@ defmodule Commanded.EventStore do meta = %{ application: application, stream_uuid: stream_uuid, - expected_version: expected_version + expected_version: expected_version, + event_count: length(events) } span(:append_to_stream, meta, fn -> diff --git a/lib/commanded/opentelemetry/commanded_attributes.ex b/lib/commanded/opentelemetry/commanded_attributes.ex index 69088c5f..eaf3b9f8 100644 --- a/lib/commanded/opentelemetry/commanded_attributes.ex +++ b/lib/commanded/opentelemetry/commanded_attributes.ex @@ -184,4 +184,16 @@ defmodule Commanded.OpenTelemetry.CommandedAttributes do """ @spec commanded_registry_adapter() :: :"commanded.registry.adapter" def commanded_registry_adapter, do: :"commanded.registry.adapter" + + @doc """ + The version number to start reading a stream from. + """ + @spec commanded_stream_start_version() :: :"commanded.stream.start_version" + def commanded_stream_start_version, do: :"commanded.stream.start_version" + + @doc """ + The batch size used when reading events from a stream. + """ + @spec commanded_stream_batch_size() :: :"commanded.stream.batch_size" + def commanded_stream_batch_size, do: :"commanded.stream.batch_size" end diff --git a/lib/commanded/opentelemetry/event_store.ex b/lib/commanded/opentelemetry/event_store.ex index f4192d1c..fddac7a3 100644 --- a/lib/commanded/opentelemetry/event_store.ex +++ b/lib/commanded/opentelemetry/event_store.ex @@ -13,16 +13,11 @@ defmodule Commanded.OpenTelemetry.EventStore do @tracer_id __MODULE__ @events ~w( - ack_event append_to_stream delete_snapshot - delete_subscription read_snapshot record_snapshot stream_forward - subscribe - subscribe_to - unsubscribe )a def setup do @@ -53,7 +48,7 @@ defmodule Commanded.OpenTelemetry.EventStore do action_name = to_string(action) {adapter, adapter_meta} = fetch_event_store_adapter(meta[:application]) event_store_name = event_store_name(adapter_meta) - destination_name = to_destination_name(event_store_name) + destination_name = Helpers.to_destination_name(event_store_name) source_uuid = extract_source_uuid(meta) connection_config = lookup_connection_config(adapter, event_store_name) @@ -65,14 +60,17 @@ defmodule Commanded.OpenTelemetry.EventStore do {CommandedAttributes.commanded_application(), meta[:application]} ] |> maybe_add_db_system(adapter) - |> maybe_add_operation_type(operation_type) - |> maybe_add_destination_name(destination_name) - |> maybe_add_stream_uuid(meta[:stream_uuid]) - |> maybe_add_expected_version(meta[:expected_version]) - |> maybe_add_subscription_name(meta[:subscription_name]) - |> maybe_add_source_uuid(source_uuid) + |> Helpers.maybe_add_operation_type(operation_type) + |> Helpers.maybe_add_destination_name(destination_name) + |> Helpers.maybe_add_stream_uuid(meta[:stream_uuid]) + |> Helpers.maybe_add_expected_version(meta[:expected_version]) + |> Helpers.maybe_add_event_count(meta[:event_count]) + |> Helpers.maybe_add_subscription_name(meta[:subscription_name]) + |> Helpers.maybe_add_source_uuid(source_uuid) |> maybe_add_start_from(meta[:start_from]) - |> Helpers.maybe_add_connection_attributes(connection_config, peer_service: destination_name) + |> maybe_add_start_version(meta[:start_version]) + |> maybe_add_read_batch_size(meta[:read_batch_size]) + |> Helpers.maybe_add_connection_attributes(connection_config) span_name = case destination_name do @@ -143,55 +141,25 @@ defmodule Commanded.OpenTelemetry.EventStore do defp operation_type_for(:append_to_stream), do: :publish defp operation_type_for(:stream_forward), do: :receive - defp operation_type_for(:subscribe), do: :receive - defp operation_type_for(:subscribe_to), do: :receive - defp operation_type_for(:ack_event), do: :settle defp operation_type_for(:delete_snapshot), do: nil - defp operation_type_for(:delete_subscription), do: nil defp operation_type_for(:read_snapshot), do: :receive defp operation_type_for(:record_snapshot), do: :publish - defp operation_type_for(:unsubscribe), do: nil defp operation_type_for(_), do: nil - defp maybe_add_operation_type(attrs, nil), do: attrs - - defp maybe_add_operation_type(attrs, type), - do: [{MessagingAttributes.messaging_operation_type(), type} | attrs] - - defp maybe_add_destination_name(attrs, nil), do: attrs - - defp maybe_add_destination_name(attrs, destination_name), - do: [{MessagingAttributes.messaging_destination_name(), destination_name} | attrs] - - defp maybe_add_stream_uuid(attrs, nil), do: attrs - - defp maybe_add_stream_uuid(attrs, stream_uuid), - do: [{CommandedAttributes.commanded_stream_uuid(), stream_uuid} | attrs] - - defp maybe_add_expected_version(attrs, nil), do: attrs - - defp maybe_add_expected_version(attrs, expected_version), - do: [{CommandedAttributes.commanded_expected_version(), expected_version} | attrs] - - defp maybe_add_subscription_name(attrs, nil), do: attrs + defp maybe_add_start_from(attrs, nil), do: attrs - defp maybe_add_subscription_name(attrs, name) do - [ - {MessagingAttributes.messaging_destination_subscription_name(), name}, - {CommandedAttributes.commanded_subscription_name(), name} - | attrs - ] - end + defp maybe_add_start_from(attrs, start_from), + do: [{CommandedAttributes.commanded_start_from(), to_start_from_attr(start_from)} | attrs] - defp maybe_add_source_uuid(attrs, nil), do: attrs + defp maybe_add_start_version(attrs, nil), do: attrs - defp maybe_add_source_uuid(attrs, source_uuid), - do: [{CommandedAttributes.commanded_source_uuid(), source_uuid} | attrs] + defp maybe_add_start_version(attrs, version), + do: [{CommandedAttributes.commanded_stream_start_version(), version} | attrs] - defp maybe_add_start_from(attrs, nil), do: attrs + defp maybe_add_read_batch_size(attrs, nil), do: attrs - defp maybe_add_start_from(attrs, start_from), - do: [{CommandedAttributes.commanded_start_from(), to_start_from_attr(start_from)} | attrs] + defp maybe_add_read_batch_size(attrs, size), + do: [{CommandedAttributes.commanded_stream_batch_size(), size} | attrs] # Extract source_uuid from metadata or nested snapshot struct (record_snapshot operation) defp extract_source_uuid(%{source_uuid: uuid}) when is_binary(uuid), do: uuid @@ -242,11 +210,6 @@ defmodule Commanded.OpenTelemetry.EventStore do defp event_store_name(_), do: nil - defp to_destination_name(nil), do: nil - defp to_destination_name(name) when is_binary(name), do: name - defp to_destination_name(name) when is_atom(name), do: inspect(name) - defp to_destination_name(_), do: nil - defp to_start_from_attr(nil), do: nil defp to_start_from_attr(atom) when is_atom(atom), do: to_string(atom) defp to_start_from_attr(other), do: other diff --git a/lib/commanded/opentelemetry/helpers.ex b/lib/commanded/opentelemetry/helpers.ex index 9deb7abf..92c9be43 100644 --- a/lib/commanded/opentelemetry/helpers.ex +++ b/lib/commanded/opentelemetry/helpers.ex @@ -1,7 +1,9 @@ defmodule Commanded.OpenTelemetry.Helpers do @moduledoc false + alias Commanded.OpenTelemetry.CommandedAttributes alias OpenTelemetry.SemConv.Incubating.DBAttributes + alias OpenTelemetry.SemConv.Incubating.MessagingAttributes alias OpenTelemetry.SemConv.Incubating.PeerAttributes alias OpenTelemetry.SemConv.ServerAttributes alias OpenTelemetry.Span @@ -45,12 +47,9 @@ defmodule Commanded.OpenTelemetry.Helpers do # Emits telemetry for unknown error types so users can detect unexpected # error formats in their monitoring and fix them. - def to_error_type(%{__struct__: module}, _tracer_id), do: to_string(module) + def to_error_type(error, _tracer_id) when is_struct(error), do: inspect(error.__struct__) - def to_error_type(%{__exception__: true} = exception, _tracer_id), - do: to_string(exception.__struct__) - - def to_error_type(error, _tracer_id) when is_atom(error), do: to_string(error) + def to_error_type(error, _tracer_id) when is_atom(error), do: inspect(error) def to_error_type(error, tracer_id) do :telemetry.execute( @@ -112,18 +111,61 @@ defmodule Commanded.OpenTelemetry.Helpers do def struct_name(%name{}), do: inspect(name) def struct_name(_), do: nil - def maybe_add_connection_attributes(attrs, config, opts \\ []) - - def maybe_add_connection_attributes(attrs, [_ | _] = config, opts) do + def maybe_add_connection_attributes(attrs, [_ | _] = config) do attrs |> maybe_add_attr(ServerAttributes.server_address(), config[:hostname]) |> maybe_add_attr(ServerAttributes.server_port(), config[:port]) |> maybe_add_attr(DBAttributes.db_namespace(), config[:database]) - |> maybe_add_attr(PeerAttributes.peer_service(), opts[:peer_service]) + |> maybe_add_attr(PeerAttributes.peer_service(), config[:database]) end - def maybe_add_connection_attributes(attrs, _config, _opts), do: attrs + def maybe_add_connection_attributes(attrs, _config), do: attrs def maybe_add_attr(attrs, _key, nil), do: attrs def maybe_add_attr(attrs, key, value), do: [{key, value} | attrs] + + def maybe_add_operation_type(attrs, nil), do: attrs + + def maybe_add_operation_type(attrs, type), + do: [{MessagingAttributes.messaging_operation_type(), type} | attrs] + + def maybe_add_destination_name(attrs, nil), do: attrs + + def maybe_add_destination_name(attrs, name), + do: [{MessagingAttributes.messaging_destination_name(), name} | attrs] + + def maybe_add_stream_uuid(attrs, nil), do: attrs + + def maybe_add_stream_uuid(attrs, uuid), + do: [{CommandedAttributes.commanded_stream_uuid(), uuid} | attrs] + + def maybe_add_expected_version(attrs, nil), do: attrs + + def maybe_add_expected_version(attrs, version), + do: [{CommandedAttributes.commanded_expected_version(), version} | attrs] + + def maybe_add_event_count(attrs, nil), do: attrs + + def maybe_add_event_count(attrs, count), + do: [{CommandedAttributes.commanded_event_count(), count} | attrs] + + def maybe_add_subscription_name(attrs, nil), do: attrs + + def maybe_add_subscription_name(attrs, name) do + [ + {MessagingAttributes.messaging_destination_subscription_name(), name}, + {CommandedAttributes.commanded_subscription_name(), name} + | attrs + ] + end + + def maybe_add_source_uuid(attrs, nil), do: attrs + + def maybe_add_source_uuid(attrs, uuid), + do: [{CommandedAttributes.commanded_source_uuid(), uuid} | attrs] + + def to_destination_name(nil), do: nil + def to_destination_name(name) when is_binary(name), do: name + def to_destination_name(name) when is_atom(name), do: inspect(name) + def to_destination_name(_), do: nil end diff --git a/test/opentelemetry/aggregate_populate_test.exs b/test/opentelemetry/aggregate_populate_test.exs index 8d584e80..4bf5d49e 100644 --- a/test/opentelemetry/aggregate_populate_test.exs +++ b/test/opentelemetry/aggregate_populate_test.exs @@ -281,9 +281,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do end meta = - Factory.build_aggregate_populate_metadata( - metadata: %{"traceparent" => traceparent} - ) + Factory.build_aggregate_populate_metadata(metadata: %{"traceparent" => traceparent}) :telemetry.execute([:commanded, :aggregate, :load, :start], %{}, meta) @@ -330,9 +328,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do test "load span is independent when traceparent is invalid" do meta = - Factory.build_aggregate_populate_metadata( - metadata: %{"traceparent" => "invalid-format"} - ) + Factory.build_aggregate_populate_metadata(metadata: %{"traceparent" => "invalid-format"}) :telemetry.execute([:commanded, :aggregate, :load, :start], %{}, meta) diff --git a/test/opentelemetry/aggregate_snapshot_test.exs b/test/opentelemetry/aggregate_snapshot_test.exs index 1032b523..89cf42ae 100644 --- a/test/opentelemetry/aggregate_snapshot_test.exs +++ b/test/opentelemetry/aggregate_snapshot_test.exs @@ -151,7 +151,7 @@ defmodule Commanded.OpenTelemetry.AggregateSnapshotTest do )}, 1000 - assert :otel_attributes.map(span_attrs)[:"error.type"] == "snapshotting_not_configured" + assert :otel_attributes.map(span_attrs)[:"error.type"] == ":snapshotting_not_configured" end end diff --git a/test/opentelemetry/aggregate_test.exs b/test/opentelemetry/aggregate_test.exs index 790f7dfb..54a68409 100644 --- a/test/opentelemetry/aggregate_test.exs +++ b/test/opentelemetry/aggregate_test.exs @@ -320,7 +320,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do "commanded.causation_id": causation_id, "commanded.registry.adapter": "Commanded.Registration.LocalRegistry", "commanded.event.count": 0, - "error.type": "validation_failed" + "error.type": ":validation_failed" } end @@ -356,7 +356,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do 1000 assert status == {:status, :unset, ""} - assert :otel_attributes.map(span_attrs)[:"error.type"] == "validation_failed" + assert :otel_attributes.map(span_attrs)[:"error.type"] == ":validation_failed" end test "uses the formatted aggregate error message when callback returns error" do @@ -424,7 +424,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do "commanded.causation_id": context.causation_id, "commanded.registry.adapter": "Commanded.Registration.LocalRegistry", "erlang.exception.kind": :error, - "error.type": "Elixir.ArgumentError" + "error.type": "ArgumentError" } end @@ -470,7 +470,7 @@ defmodule Commanded.OpenTelemetry.AggregateTest do "commanded.causation_id": context.causation_id, "commanded.registry.adapter": "Commanded.Registration.LocalRegistry", "erlang.exception.kind": :error, - "error.type": "Elixir.RuntimeError" + "error.type": "RuntimeError" } end end diff --git a/test/opentelemetry/application_test.exs b/test/opentelemetry/application_test.exs index 8864b115..8e73872b 100644 --- a/test/opentelemetry/application_test.exs +++ b/test/opentelemetry/application_test.exs @@ -230,7 +230,7 @@ defmodule Commanded.OpenTelemetry.ApplicationTest do "commanded.command": "Commanded.TestSupport.TestDomain.OpenAccount", "commanded.correlation_id": correlation_id, "commanded.causation_id": causation_id, - "error.type": "validation_failed" + "error.type": ":validation_failed" } end @@ -281,7 +281,7 @@ defmodule Commanded.OpenTelemetry.ApplicationTest do assert callback_meta.error == :validation_failed assert is_function(Keyword.fetch!(callback_config, :error_status), 4) assert status == {:status, :unset, ""} - assert :otel_attributes.map(span_attrs)[:"error.type"] == "validation_failed" + assert :otel_attributes.map(span_attrs)[:"error.type"] == ":validation_failed" end test "uses the formatted error message when callback returns error" do @@ -374,7 +374,7 @@ defmodule Commanded.OpenTelemetry.ApplicationTest do "commanded.correlation_id": context.correlation_id, "commanded.causation_id": context.causation_id, "erlang.exception.kind": :error, - "error.type": "Elixir.ArgumentError" + "error.type": "ArgumentError" } end @@ -419,7 +419,7 @@ defmodule Commanded.OpenTelemetry.ApplicationTest do "commanded.correlation_id": context.correlation_id, "commanded.causation_id": context.causation_id, "erlang.exception.kind": :error, - "error.type": "Elixir.RuntimeError" + "error.type": "RuntimeError" } end diff --git a/test/opentelemetry/event_handler_test.exs b/test/opentelemetry/event_handler_test.exs index c4b1f665..7100cc8e 100644 --- a/test/opentelemetry/event_handler_test.exs +++ b/test/opentelemetry/event_handler_test.exs @@ -224,7 +224,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do "commanded.stream.id": recorded_event.stream_id, "commanded.stream.version": recorded_event.stream_version, "commanded.handler.kind": "event_handler", - "error.type": "unique_constraint_violation" + "error.type": ":unique_constraint_violation" } end @@ -267,7 +267,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do "commanded.stream.version": recorded_event.stream_version, "commanded.handler.kind": "event_handler", "erlang.exception.kind": :error, - "error.type": "Elixir.KeyError" + "error.type": "KeyError" } events_list = :otel_events.list(events) @@ -338,7 +338,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do "commanded.stream.version": recorded_event.stream_version, "commanded.handler.kind": "event_handler", "erlang.exception.kind": :error, - "error.type": "Elixir.ArgumentError" + "error.type": "ArgumentError" } end @@ -386,7 +386,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do "commanded.stream.version": recorded_event.stream_version, "commanded.handler.kind": "event_handler", "erlang.exception.kind": :error, - "error.type": "Elixir.RuntimeError" + "error.type": "RuntimeError" } end end @@ -440,7 +440,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do "commanded.handler.kind": "event_handler", "commanded.batch.first_event_id": first_event_id, "commanded.batch.last_event_id": last_event_id, - "error.type": "transaction_rollback" + "error.type": ":transaction_rollback" } end @@ -484,7 +484,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do "commanded.batch.first_event_id": first_event_id, "commanded.batch.last_event_id": last_event_id, "erlang.exception.kind": :error, - "error.type": "Elixir.DBConnection.ConnectionError" + "error.type": "DBConnection.ConnectionError" } events_list = :otel_events.list(events) diff --git a/test/opentelemetry/event_store_test.exs b/test/opentelemetry/event_store_test.exs index fa30413b..07379d6a 100644 --- a/test/opentelemetry/event_store_test.exs +++ b/test/opentelemetry/event_store_test.exs @@ -18,16 +18,11 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do alias Commanded.UUID @events ~w( - ack_event append_to_stream delete_snapshot - delete_subscription read_snapshot record_snapshot stream_forward - subscribe - subscribe_to - unsubscribe )a setup do @@ -96,6 +91,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, "commanded.expected_version": 0, + "commanded.event.count": 1, "db.system": :in_memory } end @@ -121,66 +117,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "code.function": "stream_forward", "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, - "db.system": :in_memory - } - end - - test "subscribe_to uses the configured event store as destination", %{ - destination_name: destination_name - } do - subscription_name = unique_subscription_name() - - assert {:ok, subscription} = - EventStore.subscribe_to(DefaultApp, :all, subscription_name, self(), :origin) - - assert_receive {:subscribed, ^subscription}, 1000 - - assert span(kind: :client, attributes: attributes) = - assert_receive_span_named("subscribe_to #{destination_name}") - - assert :otel_attributes.map(attributes) == %{ - "messaging.system": "commanded", - "messaging.operation.type": :receive, - "messaging.operation.name": "subscribe_to", - "messaging.destination.name": destination_name, - "messaging.destination.subscription.name": subscription_name, - "code.function": "subscribe_to", - "commanded.application": DefaultApp, - "commanded.stream.uuid": :all, - "commanded.subscription.name": subscription_name, - "commanded.start_from": "origin", - "db.system": :in_memory - } - end - - test "ack_event uses the configured event store as destination", %{ - destination_name: destination_name - } do - subscription_name = unique_subscription_name() - stream_uuid = UUID.uuid4() - - assert {:ok, subscription} = - EventStore.subscribe_to(DefaultApp, :all, subscription_name, self(), :origin) - - assert_receive {:subscribed, ^subscription}, 1000 - _ = assert_receive_span_named("subscribe_to #{destination_name}") - - assert :ok = EventStore.append_to_stream(DefaultApp, stream_uuid, 0, build_events(1)) - assert_receive {:events, [event]}, 1000 - _ = assert_receive_span_named("append_to_stream #{destination_name}") - - assert :ok = EventStore.ack_event(DefaultApp, subscription, event) - - assert span(kind: :client, attributes: attributes) = - assert_receive_span_named("ack_event #{destination_name}") - - assert :otel_attributes.map(attributes) == %{ - "messaging.system": "commanded", - "messaging.operation.type": :settle, - "messaging.operation.name": "ack_event", - "messaging.destination.name": destination_name, - "code.function": "ack_event", - "commanded.application": DefaultApp, + "commanded.stream.start_version": 0, + "commanded.stream.batch_size": 1000, "db.system": :in_memory } end @@ -260,83 +198,6 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do } end - test "subscribe uses the configured event store as destination", %{ - destination_name: destination_name - } do - stream_uuid = UUID.uuid4() - - assert :ok = EventStore.subscribe(DefaultApp, stream_uuid) - - assert span(kind: :client, attributes: attributes) = - assert_receive_span_named("subscribe #{destination_name}") - - assert :otel_attributes.map(attributes) == %{ - "messaging.system": "commanded", - "messaging.operation.type": :receive, - "messaging.operation.name": "subscribe", - "messaging.destination.name": destination_name, - "code.function": "subscribe", - "commanded.application": DefaultApp, - "commanded.stream.uuid": stream_uuid, - "db.system": :in_memory - } - end - - test "unsubscribe uses the configured event store as destination", %{ - destination_name: destination_name - } do - subscription_name = unique_subscription_name() - - assert {:ok, subscription} = - EventStore.subscribe_to(DefaultApp, :all, subscription_name, self(), :origin) - - assert_receive {:subscribed, ^subscription}, 1000 - _ = assert_receive_span_named("subscribe_to #{destination_name}") - - assert :ok = EventStore.unsubscribe(DefaultApp, subscription) - - assert span(kind: :client, attributes: attributes) = - assert_receive_span_named("unsubscribe #{destination_name}") - - assert :otel_attributes.map(attributes) == %{ - "messaging.system": "commanded", - "messaging.operation.name": "unsubscribe", - "messaging.destination.name": destination_name, - "code.function": "unsubscribe", - "commanded.application": DefaultApp, - "db.system": :in_memory - } - end - - test "delete_subscription uses the configured event store as destination", %{ - destination_name: destination_name - } do - subscription_name = unique_subscription_name() - - assert {:ok, subscription} = - EventStore.subscribe_to(DefaultApp, :all, subscription_name, self(), :origin) - - assert_receive {:subscribed, ^subscription}, 1000 - _ = assert_receive_span_named("subscribe_to #{destination_name}") - - assert :ok = EventStore.unsubscribe(DefaultApp, subscription) - _ = assert_receive_span_named("unsubscribe #{destination_name}") - - assert :ok = EventStore.delete_subscription(DefaultApp, :all, subscription_name) - - assert span(kind: :client, attributes: attributes) = - assert_receive_span_named("delete_subscription #{destination_name}") - - assert :otel_attributes.map(attributes) == %{ - "messaging.system": "commanded", - "messaging.operation.name": "delete_subscription", - "messaging.destination.name": destination_name, - "code.function": "delete_subscription", - "commanded.application": DefaultApp, - "db.system": :in_memory - } - end - test "missing streams still keep the real destination on the span", %{ destination_name: destination_name } do @@ -355,6 +216,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "code.function": "stream_forward", "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, + "commanded.stream.start_version": 0, + "commanded.stream.batch_size": 1000, "db.system": :in_memory } end @@ -447,8 +310,9 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, "commanded.expected_version": 0, + "commanded.event.count": 1, "erlang.exception.kind": :error, - "error.type": "Elixir.MatchError" + "error.type": "MatchError" } assert_exception_event(events, "Elixir.MatchError") @@ -484,8 +348,9 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, "commanded.expected_version": 0, + "commanded.event.count": 1, "erlang.exception.kind": :error, - "error.type": "Elixir.FunctionClauseError", + "error.type": "FunctionClauseError", "db.system": :in_memory } @@ -538,7 +403,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, "commanded.expected_version": 0, - "error.type": "stream_not_found", + "commanded.event.count": 1, + "error.type": ":stream_not_found", "db.system": :in_memory } end @@ -586,8 +452,9 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, "commanded.expected_version": 0, + "commanded.event.count": 1, "erlang.exception.kind": :error, - "error.type": "Elixir.RuntimeError", + "error.type": "RuntimeError", "db.system": :in_memory } @@ -621,8 +488,9 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, "commanded.expected_version": 0, + "commanded.event.count": 1, "erlang.exception.kind": :error, - "error.type": "Elixir.RuntimeError" + "error.type": "RuntimeError" } assert_exception_event(events, "Elixir.RuntimeError") @@ -651,10 +519,6 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do defp build_events(count), do: AdapterTestData.build_opened_events(count) - defp unique_subscription_name do - "subscription-#{System.unique_integer([:positive])}" - end - defp assert_receive_span_named(name, timeout \\ 1000) do deadline = System.monotonic_time(:millisecond) + timeout diff --git a/test/support/factory.ex b/test/support/factory.ex index a610055f..d97ee448 100644 --- a/test/support/factory.ex +++ b/test/support/factory.ex @@ -1125,7 +1125,8 @@ defmodule Commanded.TestSupport.Factory do defaults = [ application: Keyword.get(opts, :application, MockApp), stream_uuid: Keyword.get(opts, :stream_uuid, UUID.uuid4()), - expected_version: Keyword.get(opts, :expected_version, 0) + expected_version: Keyword.get(opts, :expected_version, 0), + event_count: Keyword.get(opts, :event_count, 1) ] opts = Keyword.merge(defaults, opts) @@ -1133,7 +1134,8 @@ defmodule Commanded.TestSupport.Factory do %{ application: Keyword.fetch!(opts, :application), stream_uuid: Keyword.fetch!(opts, :stream_uuid), - expected_version: Keyword.fetch!(opts, :expected_version) + expected_version: Keyword.fetch!(opts, :expected_version), + event_count: Keyword.fetch!(opts, :event_count) } end