fix(telemetry): add peer.service and DB semantic attributes to EventStore spans

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,13 +64,15 @@ 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])
       |> maybe_add_expected_version(meta[:expected_version])
       |> 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),

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

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,15 +295,16 @@ 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) == %{
                "messaging.system": "commanded",
                "messaging.operation.name": "unsubscribe",
                "messaging.destination.name": destination_name,
                "code.function": "unsubscribe",
-               "commanded.application": DefaultApp
+               "commanded.application": DefaultApp,
+               "db.system": :in_memory
              }
     end
 
@@ -315,15 +324,16 @@ 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) == %{
                "messaging.system": "commanded",
                "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")