Repository: commanded/commanded
Pull request #438 (2021-04-13 17:14:37)

EventStore telemetry events
Adds telemetry events to the `Commanded.EventStore` adapter

- [x] ~~Determine API coverage that we want (I'm starting with the functions that the aggregate uses)~~
- [x] Tests
- [x] Pass relevant metadata
- [x] Documentation

I am going to avoid using the TelemetryRegistry for documenmting these events until it has better support for generating span documentation, otherwise, it will overtake this module. If you're ok with that @slashdotdash then I'm happy to convert to `telemetry_event` definitions

### API To cover
- [x] ack_event/3
- [ ] adapter/2
- [x] append_to_stream/4
- [x] delete_snapshot/2
- [x] delete_subscription/3
- [x] read_snapshot/2
- [x] record_snapshot/2
- [x] stream_forward/2
- [x] stream_forward/3
- [x] stream_forward/4
- [x] subscribe/2
- [x] subscribe_to/5
- [x] subscribe_to/6
- [x] unsubscribe/2

davydog187: Testing here https://github.com/SimpleBet/opentelemetry_commanded/pull/5

Patch:
diff --git a/lib/commanded/event_store.ex b/lib/commanded/event_store.ex
index 67b4a20..14e65fa 100644
--- a/lib/commanded/event_store.ex
+++ b/lib/commanded/event_store.ex
@@ -1,6 +1,27 @@
 defmodule Commanded.EventStore do
   @moduledoc """
   Use the event store configured for a Commanded application.
+
+
+  ### Telemetry Events
+  Adds telemetry events for the following functions. Events are emitted in the form
+
+  `[:commanded, :event_store, event]` with their spannable postfixes (`start`, `stop`, `exception`)
+
+  * ack_event/3
+  * adapter/2
+  * append_to_stream/4
+  * delete_snapshot/2
+  * delete_subscription/3
+  * read_snapshot/2
+  * record_snapshot/2
+  * stream_forward/2
+  * stream_forward/3
+  * stream_forward/4
+  * subscribe/2
+  * subscribe_to/5
+  * subscribe_to/6
+  * unsubscribe/2
   """
 
   alias Commanded.Application
@@ -13,14 +34,22 @@ defmodule Commanded.EventStore do
   Append one or more events to a stream atomically.
   """
   def append_to_stream(application, stream_uuid, expected_version, events) do
-    {adapter, adapter_meta} = Application.event_store_adapter(application)
+    meta = %{
+      application: application,
+      stream_uuid: stream_uuid,
+      expected_version: expected_version
+    }
+
+    span(:append_to_stream, meta, fn ->
+      {adapter, adapter_meta} = Application.event_store_adapter(application)
 
-    adapter.append_to_stream(
-      adapter_meta,
-      stream_uuid,
-      expected_version,
-      events
-    )
+      adapter.append_to_stream(
+        adapter_meta,
+        stream_uuid,
+        expected_version,
+        events
+      )
+    end)
   end
 
   @doc """
@@ -28,20 +57,29 @@ defmodule Commanded.EventStore do
   originally written.
   """
   def stream_forward(application, stream_uuid, start_version \\ 0, read_batch_size \\ 1_000) do
-    {adapter, adapter_meta} = Application.event_store_adapter(application)
-
-    case adapter.stream_forward(
-           adapter_meta,
-           stream_uuid,
-           start_version,
-           read_batch_size
-         ) do
-      {:error, _error} = error ->
-        error
-
-      stream ->
-        Upcast.upcast_event_stream(stream, additional_metadata: %{application: application})
-    end
+    meta = %{
+      application: application,
+      stream_uuid: stream_uuid,
+      start_version: start_version,
+      read_batch_size: read_batch_size
+    }
+
+    span(:stream_forward, meta, fn ->
+      {adapter, adapter_meta} = Application.event_store_adapter(application)
+
+      case adapter.stream_forward(
+             adapter_meta,
+             stream_uuid,
+             start_version,
+             read_batch_size
+           ) do
+        {:error, _error} = error ->
+          error
+
+        stream ->
+          Upcast.upcast_event_stream(stream, additional_metadata: %{application: application})
+      end
+    end)
   end
 
   @doc """
@@ -53,9 +91,11 @@ defmodule Commanded.EventStore do
   The subscriber does not need to acknowledge receipt of the events.
   """
   def subscribe(application, stream_uuid) do
-    {adapter, adapter_meta} = Application.event_store_adapter(application)
+    span(:subscribe, %{application: application, stream_uuid: stream_uuid}, fn ->
+      {adapter, adapter_meta} = Application.event_store_adapter(application)
 
-    adapter.subscribe(adapter_meta, stream_uuid)
+      adapter.subscribe(adapter_meta, stream_uuid)
+    end)
   end
 
   @doc """
@@ -100,26 +140,36 @@ defmodule Commanded.EventStore do
         start_from,
         options \\ []
       ) do
-    {adapter, adapter_meta} = Application.event_store_adapter(application)
-
-    if function_exported?(adapter, :subscribe_to, 6) do
-      adapter.subscribe_to(
-        adapter_meta,
-        stream_uuid,
-        subscription_name,
-        subscriber,
-        start_from,
-        options
-      )
-    else
-      adapter.subscribe_to(
-        adapter_meta,
-        stream_uuid,
-        subscription_name,
-        subscriber,
-        start_from
-      )
-    end
+    meta = %{
+      application: application,
+      stream_uuid: stream_uuid,
+      subscription_name: subscription_name,
+      subscriber: subscriber,
+      start_from: start_from
+    }
+
+    span(:subscribe_to, meta, fn ->
+      {adapter, adapter_meta} = Application.event_store_adapter(application)
+
+      if function_exported?(adapter, :subscribe_to, 6) do
+        adapter.subscribe_to(
+          adapter_meta,
+          stream_uuid,
+          subscription_name,
+          subscriber,
+          start_from,
+          options
+        )
+      else
+        adapter.subscribe_to(
+          adapter_meta,
+          stream_uuid,
+          subscription_name,
+          subscriber,
+          start_from
+        )
+      end
+    end)
   end
 
   @doc """
@@ -127,9 +177,13 @@ defmodule Commanded.EventStore do
   a subscription to an event stream.
   """
   def ack_event(application, subscription, event) do
-    {adapter, adapter_meta} = Application.event_store_adapter(application)
+    meta = %{application: application, subscription: subscription, event: event}
 
-    adapter.ack_event(adapter_meta, subscription, event)
+    span(:ack_event, meta, fn ->
+      {adapter, adapter_meta} = Application.event_store_adapter(application)
+
+      adapter.ack_event(adapter_meta, subscription, event)
+    end)
   end
 
   @doc """
@@ -143,9 +197,11 @@ defmodule Commanded.EventStore do
 
   """
   def unsubscribe(application, subscription) do
-    {adapter, adapter_meta} = Application.event_store_adapter(application)
+    span(:unsubscribe, %{application: application, subscription: subscription}, fn ->
+      {adapter, adapter_meta} = Application.event_store_adapter(application)
 
-    adapter.unsubscribe(adapter_meta, subscription)
+      adapter.unsubscribe(adapter_meta, subscription)
+    end)
   end
 
   @doc """
@@ -157,9 +213,13 @@ defmodule Commanded.EventStore do
 
   """
   def delete_subscription(application, subscribe_to, handler_name) do
-    {adapter, adapter_meta} = Application.event_store_adapter(application)
+    meta = %{application: application, subscribe_to: subscribe_to, handler_name: handler_name}
+
+    span(:delete_subscription, meta, fn ->
+      {adapter, adapter_meta} = Application.event_store_adapter(application)
 
-    adapter.delete_subscription(adapter_meta, subscribe_to, handler_name)
+      adapter.delete_subscription(adapter_meta, subscribe_to, handler_name)
+    end)
   end
 
   @doc """
@@ -168,7 +228,9 @@ defmodule Commanded.EventStore do
   def read_snapshot(application, source_uuid) do
     {adapter, adapter_meta} = Application.event_store_adapter(application)
 
-    adapter.read_snapshot(adapter_meta, source_uuid)
+    span(:read_snapshot, %{application: application, source_uuid: source_uuid}, fn ->
+      adapter.read_snapshot(adapter_meta, source_uuid)
+    end)
   end
 
   @doc """
@@ -177,7 +239,9 @@ defmodule Commanded.EventStore do
   def record_snapshot(application, snapshot) do
     {adapter, adapter_meta} = Application.event_store_adapter(application)
 
-    adapter.record_snapshot(adapter_meta, snapshot)
+    span(:record_snapshot, %{application: application, snapshot: snapshot}, fn ->
+      adapter.record_snapshot(adapter_meta, snapshot)
+    end)
   end
 
   @doc """
@@ -186,7 +250,9 @@ defmodule Commanded.EventStore do
   def delete_snapshot(application, source_uuid) do
     {adapter, adapter_meta} = Application.event_store_adapter(application)
 
-    adapter.delete_snapshot(adapter_meta, source_uuid)
+    span(:delete_snapshot, %{application: application, source_uuid: source_uuid}, fn ->
+      adapter.delete_snapshot(adapter_meta, source_uuid)
+    end)
   end
 
   @doc """
@@ -213,4 +279,11 @@ defmodule Commanded.EventStore do
 
     {adapter, config}
   end
+
+  # TODO convert to macro
+  defp span(event, meta, func) do
+    :telemetry.span([:commanded, :event_store, event], meta, fn ->
+      {func.(), meta}
+    end)
+  end
 end


Test patch:
diff --git a/test/event_store/telemetry_test.exs b/test/event_store/telemetry_test.exs
new file mode 100644
index 0000000..7d19f29
--- /dev/null
+++ b/test/event_store/telemetry_test.exs
@@ -0,0 +1,193 @@
+defmodule Commanded.EventStore.TelemetryTest do
+  use ExUnit.Case
+
+  alias Commanded.DefaultApp
+  alias Commanded.EventStore
+  alias Commanded.EventStore.EventData
+  alias Commanded.EventStore.RecordedEvent
+  alias Commanded.EventStore.SnapshotData
+  alias Commanded.Middleware.Commands.IncrementCount
+  alias Commanded.Middleware.Commands.RaiseError
+
+  setup do
+    start_supervised!(DefaultApp)
+    attach_telemetry()
+
+    :ok
+  end
+
+  defmodule TestRouter do
+    use Commanded.Commands.Router
+
+    alias Commanded.Middleware.Commands.CommandHandler
+    alias Commanded.Middleware.Commands.CounterAggregateRoot
+
+    dispatch IncrementCount,
+      to: CommandHandler,
+      aggregate: CounterAggregateRoot,
+      identity: :aggregate_uuid
+
+    dispatch RaiseError,
+      to: CommandHandler,
+      aggregate: CounterAggregateRoot,
+      identity: :aggregate_uuid
+  end
+
+  describe "snapshotting telemetry events" do
+    test "emit `[:commanded, :event_store, :record_snapshot, :start | :stop]` event" do
+      snapshot = %SnapshotData{}
+      assert :ok = EventStore.record_snapshot(DefaultApp, snapshot)
+
+      assert_receive {[:commanded, :event_store, :record_snapshot, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :record_snapshot, :stop], 2, _meas, meta}
+      assert %{application: DefaultApp, snapshot: ^snapshot} = meta
+    end
+
+    test "emit `[:commanded, :event_store, :read_snapshot, :start | :stop]` event" do
+      uuid = UUID.uuid4()
+      assert {:error, :snapshot_not_found} = EventStore.read_snapshot(DefaultApp, uuid)
+
+      assert_receive {[:commanded, :event_store, :read_snapshot, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :read_snapshot, :stop], 2, _meas, meta}
+      assert %{application: DefaultApp, source_uuid: ^uuid} = meta
+    end
+
+    test "emit `[:commanded, :event_store, :delete_snapshot, :start | :stop]` event" do
+      uuid = UUID.uuid4()
+      assert :ok = EventStore.delete_snapshot(DefaultApp, uuid)
+
+      assert_receive {[:commanded, :event_store, :delete_snapshot, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :delete_snapshot, :stop], 2, _meas, meta}
+      assert %{application: DefaultApp, source_uuid: ^uuid} = meta
+    end
+  end
+
+  describe "streaming telemetry events" do
+    test "emit `[:commanded, :event_store, :stream_forward, :start | :stop]` event" do
+      uuid = UUID.uuid4()
+      assert {:error, :stream_not_found} = EventStore.stream_forward(DefaultApp, uuid)
+
+      assert_receive {[:commanded, :event_store, :stream_forward, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :stream_forward, :stop], 2, _meas, meta}
+
+      assert %{
+               application: DefaultApp,
+               stream_uuid: ^uuid,
+               start_version: 0,
+               read_batch_size: 1_000
+             } = meta
+    end
+  end
+
+  describe "ack_event telemetry events" do
+    test "emit `[:commanded, :event_store, :ack_event, :start | :stop]` event" do
+      pid = self()
+      event = %RecordedEvent{}
+      assert :ok = EventStore.ack_event(DefaultApp, pid, event)
+
+      assert_receive {[:commanded, :event_store, :ack_event, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :ack_event, :stop], 2, _meas, meta}
+      assert %{application: DefaultApp, subscription: ^pid, event: ^event} = meta
+    end
+  end
+
+  describe "append_to_stream telemetry events" do
+    test "emit `[:commanded, :event_store, :append_to_stream, :start | :stop]` event" do
+      uuid = UUID.uuid4()
+      assert :ok = EventStore.append_to_stream(DefaultApp, uuid, 0, [%EventData{}])
+
+      assert_receive {[:commanded, :event_store, :append_to_stream, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :append_to_stream, :stop], 2, _meas, meta}
+      assert %{application: DefaultApp, expected_version: 0, stream_uuid: ^uuid} = meta
+    end
+  end
+
+  describe "subscription telemetry events" do
+    test "emit `[:commanded, :event_store, :subscribe, :start | :stop]` event" do
+      uuid = UUID.uuid4()
+      assert :ok = EventStore.subscribe(DefaultApp, uuid)
+
+      assert_receive {[:commanded, :event_store, :subscribe, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :subscribe, :stop], 2, _meas, meta}
+      assert %{application: DefaultApp, stream_uuid: ^uuid} = meta
+    end
+
+    test "emit `[:commanded, :event_store, :subscribe_to, :start | :stop]` event" do
+      subscriber = self()
+      assert {:ok, pid} = EventStore.subscribe_to(DefaultApp, :all, "Test", subscriber, :current)
+
+      assert_receive {:subscribed, ^pid}
+      assert_receive {[:commanded, :event_store, :subscribe_to, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :subscribe_to, :stop], 2, _meas, meta}
+
+      assert %{
+               application: DefaultApp,
+               stream_uuid: :all,
+               subscription_name: "Test",
+               subscriber: ^subscriber,
+               start_from: :current
+             } = meta
+    end
+
+    test "emit `[:commanded, :event_store, :unsubscribe, :start | :stop]` event" do
+      assert {:ok, pid} = EventStore.subscribe_to(DefaultApp, :all, "Test", self(), :current)
+
+      assert_receive {:subscribed, ^pid}
+
+      assert_receive {[:commanded, :event_store, :subscribe_to, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :subscribe_to, :stop], 2, _meas, _meta}
+
+      assert :ok = EventStore.unsubscribe(DefaultApp, pid)
+
+      assert_receive {[:commanded, :event_store, :unsubscribe, :start], 3, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :unsubscribe, :stop], 4, _meas, meta}
+      assert %{application: DefaultApp, subscription: ^pid} = meta
+    end
+
+    test "emit `[:commanded, :event_store, :delete_subscription, :start | :stop]` event" do
+      assert {:error, :subscription_not_found} =
+               EventStore.delete_subscription(DefaultApp, :all, "Test")
+
+      assert_receive {[:commanded, :event_store, :delete_subscription, :start], 1, _meas, _meta}
+      assert_receive {[:commanded, :event_store, :delete_subscription, :stop], 2, _meas, meta}
+      assert %{application: DefaultApp, subscribe_to: :all, handler_name: "Test"} = meta
+    end
+  end
+
+  defp attach_telemetry do
+    agent = start_supervised!({Agent, fn -> 1 end})
+    handler = :"#{__MODULE__}-handler"
+
+    events = [
+      :ack_event,
+      :append_to_stream,
+      :delete_snapshot,
+      :delete_subscription,
+      :record_snapshot,
+      :read_snapshot,
+      :stream_forward,
+      :subscribe,
+      :subscribe_to,
+      :unsubscribe
+    ]
+
+    :telemetry.attach_many(
+      handler,
+      Enum.flat_map(events, fn event ->
+        [
+          [:commanded, :event_store, event, :start],
+          [:commanded, :event_store, event, :stop]
+        ]
+      end),
+      fn event_name, measurements, metadata, reply_to ->
+        num = Agent.get_and_update(agent, fn n -> {n, n + 1} end)
+        send(reply_to, {event_name, num, measurements, metadata})
+      end,
+      self()
+    )
+
+    on_exit(fn ->
+      :telemetry.detach(handler)
+    end)
+  end
+end