Repository: commanded/commanded
Pull request #423 (2020-12-29 22:34:55)

Telemetry [:commanded, :application, :dispatch]
Adds telemetry to the dispatcher, which will add start and stop events
to all dispatches.

- [x] Documentation
- [x] Tests

Contributes to https://github.com/commanded/commanded/issues/387


Patch:
diff --git a/lib/application.ex b/lib/application.ex
index fa357d4..61f01bc 100644
--- a/lib/application.ex
+++ b/lib/application.ex
@@ -1,4 +1,27 @@
 defmodule Commanded.Application do
+  use TelemetryRegistry
+
+  telemetry_event(%{
+    event: [:commanded, :application, :dispatch, :start],
+    description: "Emitted when an application starts dispatching a command",
+    measurements: "%{system_time: integer()}",
+    metadata: """
+    %{application: Commanded.Application.t(),
+      execution_context: Commanded.Aggregates.ExecutionContext.t()}
+    """
+  })
+
+  telemetry_event(%{
+    event: [:commanded, :application, :dispatch, :stop],
+    description: "Emitted when an application stops dispatching a command",
+    measurements: "%{duration: non_neg_integer()}",
+    metadata: """
+    %{application: Commanded.Application.t(),
+      execution_context: Commanded.Aggregates.ExecutionContext.t(),
+      error: nil | any()}
+    """
+  })
+
   @moduledoc """
   Defines a Commanded application.
 
@@ -131,6 +154,11 @@ defmodule Commanded.Application do
 
   See the `Commanded.Commands.Router` module for more details about the
   supported options.
+
+  ## Telemetry
+
+  #{telemetry_docs()}
+
   """
 
   @type t :: module
diff --git a/lib/commanded/commands/dispatcher.ex b/lib/commanded/commands/dispatcher.ex
index 0b67c2a..34bba86 100644
--- a/lib/commanded/commands/dispatcher.ex
+++ b/lib/commanded/commands/dispatcher.ex
@@ -6,6 +6,7 @@ defmodule Commanded.Commands.Dispatcher do
   alias Commanded.Aggregates.Aggregate
   alias Commanded.Aggregates.ExecutionContext
   alias Commanded.Middleware.Pipeline
+  alias Commanded.Telemetry
 
   defmodule Payload do
     @moduledoc false
@@ -32,10 +33,7 @@ defmodule Commanded.Commands.Dispatcher do
     ]
   end
 
-  @doc """
-  Dispatch the given command to the handler module for the aggregate as
-  identified.
-  """
+  # Dispatch the given command to the handler module for the aggregate as identified.
   @spec dispatch(payload :: struct) ::
           :ok
           | {:ok, aggregate_state :: struct}
@@ -44,10 +42,12 @@ defmodule Commanded.Commands.Dispatcher do
           | {:ok, Commanded.Commands.ExecutionResult.t()}
           | {:error, error :: term}
   def dispatch(%Payload{} = payload) do
-    pipeline =
-      payload
-      |> to_pipeline()
-      |> before_dispatch(payload)
+    pipeline = to_pipeline(payload)
+    telemetry_metadata = telemetry_metadata(pipeline, payload)
+
+    start_time = telemetry_start(telemetry_metadata)
+
+    pipeline = before_dispatch(pipeline, payload)
 
     # Stop command execution if pipeline has been halted
     unless Pipeline.halted?(pipeline) do
@@ -55,10 +55,12 @@ defmodule Commanded.Commands.Dispatcher do
 
       pipeline
       |> execute(payload, context)
+      |> telemetry_stop(start_time, telemetry_metadata)
       |> Pipeline.response()
     else
       pipeline
       |> after_failure(payload)
+      |> telemetry_stop(start_time, telemetry_metadata)
       |> Pipeline.response()
     end
   end
@@ -195,4 +197,32 @@ defmodule Commanded.Commands.Dispatcher do
 
     Pipeline.chain(pipeline, :after_failure, middleware)
   end
+
+  defp telemetry_start(telemetry_metadata) do
+    Telemetry.start([:commanded, :application, :dispatch], telemetry_metadata)
+  end
+
+  defp telemetry_stop(%Pipeline{assigns: assigns} = pipeline, start_time, telemetry_metadata) do
+    event_prefix = [:commanded, :application, :dispatch]
+
+    case assigns do
+      %{error: error} ->
+        Telemetry.stop(event_prefix, start_time, Map.put(telemetry_metadata, :error, error))
+
+      _ ->
+        Telemetry.stop(event_prefix, start_time, telemetry_metadata)
+    end
+
+    pipeline
+  end
+
+  defp telemetry_metadata(%Pipeline{} = pipeline, %Payload{} = payload) do
+    context = to_execution_context(pipeline, payload)
+
+    %{
+      application: payload.application,
+      error: nil,
+      execution_context: context
+    }
+  end
 end


Test patch:
diff --git a/test/application/telemetry_test.exs b/test/application/telemetry_test.exs
new file mode 100644
index 0000000..1348545
--- /dev/null
+++ b/test/application/telemetry_test.exs
@@ -0,0 +1,83 @@
+defmodule Commanded.Application.TelemetryTest do
+  use ExUnit.Case
+
+  alias Commanded.DefaultApp
+  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
+
+  test "emit `[:commanded, :application, :dispatch, :start | :stop]` event" do
+    command = %IncrementCount{aggregate_uuid: UUID.uuid4()}
+
+    assert :ok = TestRouter.dispatch(command, application: DefaultApp)
+
+    assert_receive {[:commanded, :application, :dispatch, :start], 1, _meas, _meta}
+    assert_receive {[:commanded, :aggregate, :execute, :start], 2, _meas, _meta}
+    assert_receive {[:commanded, :aggregate, :execute, :stop], 3, _meas, _meta}
+    assert_receive {[:commanded, :application, :dispatch, :stop], 4, _meas, meta}
+
+    assert %{application: DefaultApp, error: nil, execution_context: %{command: ^command}} = meta
+  end
+
+  test "emit `[:commanded, :application, :dispatch, :start | :stop]` event on error" do
+    command = %RaiseError{aggregate_uuid: UUID.uuid4()}
+    error = %RuntimeError{message: "failed"}
+
+    assert {:error, ^error} = TestRouter.dispatch(command, application: DefaultApp)
+
+    assert_receive {[:commanded, :application, :dispatch, :start], 1, _meas, _meta}
+    assert_receive {[:commanded, :aggregate, :execute, :start], 2, _meas, _meta}
+    assert_receive {[:commanded, :aggregate, :execute, :exception], 3, _meas, _meta}
+    assert_receive {[:commanded, :application, :dispatch, :stop], 4, _meas, meta}
+
+    assert %{application: DefaultApp, error: ^error, execution_context: %{command: ^command}} =
+             meta
+  end
+
+  defp attach_telemetry do
+    agent = start_supervised!({Agent, fn -> 1 end})
+
+    :telemetry.attach_many(
+      "test-handler",
+      [
+        [:commanded, :aggregate, :execute, :start],
+        [:commanded, :aggregate, :execute, :stop],
+        [:commanded, :aggregate, :execute, :exception],
+        [:commanded, :application, :dispatch, :start],
+        [:commanded, :application, :dispatch, :stop]
+      ],
+      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("test-handler")
+    end)
+  end
+end