From 8796a3f45beab66bc61c77b92c1614839721b929 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 20 Apr 2026 19:50:07 -0400 Subject: [PATCH] fix(telemetry): add db and connection attributes to event store spans Signed-off-by: Yordis Prieto --- lib/commanded/opentelemetry/event_store.ex | 41 +++++++++----- lib/commanded/opentelemetry/helpers.ex | 18 ++++++ test/opentelemetry/event_store_test.exs | 64 +++++++++++++--------- 3 files changed, 84 insertions(+), 39 deletions(-) diff --git a/lib/commanded/opentelemetry/event_store.ex b/lib/commanded/opentelemetry/event_store.ex index 43414d2e..f4192d1c 100644 --- a/lib/commanded/opentelemetry/event_store.ex +++ b/lib/commanded/opentelemetry/event_store.ex @@ -6,6 +6,7 @@ defmodule Commanded.OpenTelemetry.EventStore do alias Commanded.OpenTelemetry.Helpers alias OpenTelemetry.SemConv.ErrorAttributes alias OpenTelemetry.SemConv.Incubating.CodeAttributes + alias OpenTelemetry.SemConv.Incubating.DBAttributes alias OpenTelemetry.SemConv.Incubating.MessagingAttributes alias OpenTelemetry.Span @@ -50,8 +51,11 @@ defmodule Commanded.OpenTelemetry.EventStore do ) do operation_type = operation_type_for(action) action_name = to_string(action) - destination_name = event_store_destination_name(meta) + {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) source_uuid = extract_source_uuid(meta) + connection_config = lookup_connection_config(adapter, event_store_name) attributes = [ @@ -60,6 +64,7 @@ defmodule Commanded.OpenTelemetry.EventStore do {CodeAttributes.code_function(), action_name}, {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]) @@ -67,6 +72,7 @@ defmodule Commanded.OpenTelemetry.EventStore do |> maybe_add_subscription_name(meta[:subscription_name]) |> maybe_add_source_uuid(source_uuid) |> maybe_add_start_from(meta[:start_from]) + |> Helpers.maybe_add_connection_attributes(connection_config, peer_service: destination_name) span_name = case destination_name do @@ -79,7 +85,7 @@ defmodule Commanded.OpenTelemetry.EventStore do span_name, meta, %{ - kind: :internal, + kind: :client, attributes: attributes } ) @@ -192,21 +198,28 @@ defmodule Commanded.OpenTelemetry.EventStore do defp extract_source_uuid(%{snapshot: %{source_uuid: uuid}}) when is_binary(uuid), do: uuid defp extract_source_uuid(_), do: nil - defp event_store_destination_name(meta) do - meta[:application] - |> lookup_event_store_name() - |> to_destination_name() + defp maybe_add_db_system(attrs, adapter) do + case db_system_for(adapter) do + nil -> attrs + system -> [{DBAttributes.db_system(), system} | attrs] + end end - defp lookup_event_store_name(application) do - application - |> fetch_event_store_adapter_meta() - |> event_store_name() + defp db_system_for(Commanded.EventStore.Adapters.EventStore), do: :postgresql + defp db_system_for(Commanded.EventStore.Adapters.InMemory), do: :in_memory + defp db_system_for(_), do: nil + + defp lookup_connection_config(Commanded.EventStore.Adapters.EventStore, event_store_name) + when is_atom(event_store_name) and not is_nil(event_store_name) do + EventStore.Config.lookup(event_store_name) + rescue + _ -> [] end - defp fetch_event_store_adapter_meta(application) do - {_adapter, adapter_meta} = CommandedApplication.event_store_adapter(application) - adapter_meta + defp lookup_connection_config(_adapter, _event_store_name), do: [] + + defp fetch_event_store_adapter(application) do + {_adapter, _adapter_meta} = CommandedApplication.event_store_adapter(application) rescue error -> :telemetry.execute( @@ -221,7 +234,7 @@ defmodule Commanded.OpenTelemetry.EventStore do } ) - nil + {nil, nil} end defp event_store_name(adapter_meta) when is_map(adapter_meta), diff --git a/lib/commanded/opentelemetry/helpers.ex b/lib/commanded/opentelemetry/helpers.ex index 70fc14f0..9deb7abf 100644 --- a/lib/commanded/opentelemetry/helpers.ex +++ b/lib/commanded/opentelemetry/helpers.ex @@ -1,6 +1,9 @@ defmodule Commanded.OpenTelemetry.Helpers do @moduledoc false + alias OpenTelemetry.SemConv.Incubating.DBAttributes + alias OpenTelemetry.SemConv.Incubating.PeerAttributes + alias OpenTelemetry.SemConv.ServerAttributes alias OpenTelemetry.Span def extract_propagated_ctx(nil), do: {[], :undefined} @@ -108,4 +111,19 @@ 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 + 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]) + end + + def maybe_add_connection_attributes(attrs, _config, _opts), do: attrs + + def maybe_add_attr(attrs, _key, nil), do: attrs + def maybe_add_attr(attrs, key, value), do: [{key, value} | attrs] end diff --git a/test/opentelemetry/event_store_test.exs b/test/opentelemetry/event_store_test.exs index 1d98abbe..fa30413b 100644 --- a/test/opentelemetry/event_store_test.exs +++ b/test/opentelemetry/event_store_test.exs @@ -84,7 +84,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert :ok = EventStore.append_to_stream(DefaultApp, stream_uuid, 0, build_events(1)) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("append_to_stream #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -95,7 +95,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "code.function": "append_to_stream", "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, - "commanded.expected_version": 0 + "commanded.expected_version": 0, + "db.system": :in_memory } end @@ -109,7 +110,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert [_event] = EventStore.stream_forward(DefaultApp, stream_uuid, 0) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("stream_forward #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -119,7 +120,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "messaging.destination.name": destination_name, "code.function": "stream_forward", "commanded.application": DefaultApp, - "commanded.stream.uuid": stream_uuid + "commanded.stream.uuid": stream_uuid, + "db.system": :in_memory } end @@ -133,7 +135,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert_receive {:subscribed, ^subscription}, 1000 - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("subscribe_to #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -146,7 +148,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.application": DefaultApp, "commanded.stream.uuid": :all, "commanded.subscription.name": subscription_name, - "commanded.start_from": "origin" + "commanded.start_from": "origin", + "db.system": :in_memory } end @@ -168,7 +171,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert :ok = EventStore.ack_event(DefaultApp, subscription, event) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("ack_event #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -177,7 +180,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "messaging.operation.name": "ack_event", "messaging.destination.name": destination_name, "code.function": "ack_event", - "commanded.application": DefaultApp + "commanded.application": DefaultApp, + "db.system": :in_memory } end @@ -189,7 +193,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert :ok = EventStore.record_snapshot(DefaultApp, snapshot) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("record_snapshot #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -199,7 +203,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "messaging.destination.name": destination_name, "code.function": "record_snapshot", "commanded.application": DefaultApp, - "commanded.source.uuid": source_uuid + "commanded.source.uuid": source_uuid, + "db.system": :in_memory } end @@ -215,7 +220,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert {:ok, %SnapshotData{source_uuid: ^source_uuid}} = EventStore.read_snapshot(DefaultApp, source_uuid) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("read_snapshot #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -225,7 +230,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "messaging.destination.name": destination_name, "code.function": "read_snapshot", "commanded.application": DefaultApp, - "commanded.source.uuid": source_uuid + "commanded.source.uuid": source_uuid, + "db.system": :in_memory } end @@ -240,7 +246,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert :ok = EventStore.delete_snapshot(DefaultApp, source_uuid) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("delete_snapshot #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -249,7 +255,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "messaging.destination.name": destination_name, "code.function": "delete_snapshot", "commanded.application": DefaultApp, - "commanded.source.uuid": source_uuid + "commanded.source.uuid": source_uuid, + "db.system": :in_memory } end @@ -260,7 +267,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert :ok = EventStore.subscribe(DefaultApp, stream_uuid) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("subscribe #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -270,7 +277,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "messaging.destination.name": destination_name, "code.function": "subscribe", "commanded.application": DefaultApp, - "commanded.stream.uuid": stream_uuid + "commanded.stream.uuid": stream_uuid, + "db.system": :in_memory } end @@ -287,7 +295,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert :ok = EventStore.unsubscribe(DefaultApp, subscription) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("unsubscribe #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -295,7 +303,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "messaging.operation.name": "unsubscribe", "messaging.destination.name": destination_name, "code.function": "unsubscribe", - "commanded.application": DefaultApp + "commanded.application": DefaultApp, + "db.system": :in_memory } end @@ -315,7 +324,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert :ok = EventStore.delete_subscription(DefaultApp, :all, subscription_name) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("delete_subscription #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -323,7 +332,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "messaging.operation.name": "delete_subscription", "messaging.destination.name": destination_name, "code.function": "delete_subscription", - "commanded.application": DefaultApp + "commanded.application": DefaultApp, + "db.system": :in_memory } end @@ -334,7 +344,7 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do assert {:error, :stream_not_found} = EventStore.stream_forward(DefaultApp, stream_uuid) - assert span(kind: :internal, attributes: attributes) = + assert span(kind: :client, attributes: attributes) = assert_receive_span_named("stream_forward #{destination_name}") assert :otel_attributes.map(attributes) == %{ @@ -344,7 +354,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "messaging.destination.name": destination_name, "code.function": "stream_forward", "commanded.application": DefaultApp, - "commanded.stream.uuid": stream_uuid + "commanded.stream.uuid": stream_uuid, + "db.system": :in_memory } end end @@ -474,7 +485,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.stream.uuid": stream_uuid, "commanded.expected_version": 0, "erlang.exception.kind": :error, - "error.type": "Elixir.FunctionClauseError" + "error.type": "Elixir.FunctionClauseError", + "db.system": :in_memory } assert_handler_attached(:append_to_stream) @@ -526,7 +538,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.application": DefaultApp, "commanded.stream.uuid": stream_uuid, "commanded.expected_version": 0, - "error.type": "stream_not_found" + "error.type": "stream_not_found", + "db.system": :in_memory } end @@ -574,7 +587,8 @@ defmodule Commanded.OpenTelemetry.EventStoreTest do "commanded.stream.uuid": stream_uuid, "commanded.expected_version": 0, "erlang.exception.kind": :error, - "error.type": "Elixir.RuntimeError" + "error.type": "Elixir.RuntimeError", + "db.system": :in_memory } assert_exception_event(events, "Elixir.RuntimeError", "failed")