Repository: commanded/commanded
Pull request #418 (2020-12-07 23:36:34)

Telemetry [:commanded, :process_manager, :handle]
Adds telemetry events for Process Managers.

Contributes to https://github.com/commanded/commanded/issues/387


Patch:
diff --git a/lib/commanded/process_managers/process_manager.ex b/lib/commanded/process_managers/process_manager.ex
index 656e770..572747a 100644
--- a/lib/commanded/process_managers/process_manager.ex
+++ b/lib/commanded/process_managers/process_manager.ex
@@ -1,4 +1,50 @@
 defmodule Commanded.ProcessManagers.ProcessManager do
+  use TelemetryRegistry
+
+  telemetry_event(%{
+    event: [:commanded, :process_manager, :handle, :start],
+    description: "Emitted when a process manager starts handling an event",
+    measurements: "%{system_time: integer()}",
+    metadata: """
+    %{application: Commanded.Application.t(),
+      process_manager_name: String.t() | Inspect.t(),
+      process_manager_module: module(),
+      process_state: term(),
+      process_uuid: String.t()}
+    """
+  })
+
+  telemetry_event(%{
+    event: [:commanded, :process_manager, :handle, :stop],
+    description: "Emitted when a process manager stops handling an event",
+    measurements: "%{system_time: integer()}",
+    metadata: """
+    %{application: Commanded.Application.t(),
+      commands: [struct()],
+      error: nil | any(),
+      process_manager_name: String.t() | Inspect.t(),
+      process_manager_module: module(),
+      process_state: term(),
+      process_uuid: String.t()}
+    """
+  })
+
+  telemetry_event(%{
+    event: [:commanded, :process_manager, :handle, :exception],
+    description: "Emitted when a process manager raises an exception",
+    measurements: "%{system_time: integer()}",
+    metadata: """
+    %{application: Commanded.Application.t(),
+      process_manager_name: String.t() | Inspect.t(),
+      process_manager_module: module(),
+      process_state: term(),
+      process_uuid: String.t(),
+      kind: :throw | :error | :exit,
+      reason: any(),
+      stacktrace: list()}
+    """
+  })
+
   @moduledoc """
   Macro used to define a process manager.
 
@@ -219,6 +265,11 @@ defmodule Commanded.ProcessManagers.ProcessManager do
 
   The above example requires three named Commanded applications to have already
   been started.
+
+  ## Telemetry
+
+  #{telemetry_docs()}
+
   """
 
   alias Commanded.ProcessManagers.FailureContext
diff --git a/lib/commanded/process_managers/process_manager_instance.ex b/lib/commanded/process_managers/process_manager_instance.ex
index a73c081..f00c6ff 100644
--- a/lib/commanded/process_managers/process_manager_instance.ex
+++ b/lib/commanded/process_managers/process_manager_instance.ex
@@ -9,6 +9,7 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   alias Commanded.ProcessManagers.{ProcessRouter, FailureContext}
   alias Commanded.EventStore
   alias Commanded.EventStore.{RecordedEvent, SnapshotData}
+  alias Commanded.Telemetry
 
   defmodule State do
     @moduledoc false
@@ -137,9 +138,10 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   """
   @impl GenServer
   def handle_cast({:process_event, event}, %State{} = state) do
-    case event_already_seen?(event, state) do
-      true -> process_seen_event(event, state)
-      false -> process_unseen_event(event, state)
+    if event_already_seen?(event, state) do
+      process_seen_event(event, state)
+    else
+      process_unseen_event(event, state)
     end
   end
 
@@ -173,8 +175,13 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
   defp process_seen_event(%RecordedEvent{} = event, %State{} = state) do
     %State{idle_timeout: idle_timeout} = state
 
+    telemetry_metadata = telemetry_metadata(event, state)
+    start_time = telemetry_start(telemetry_metadata)
+
     :ok = ack_event(event, state)
 
+    telemetry_stop(start_time, telemetry_metadata, [])
+
     {:noreply, state, idle_timeout}
   end
 
@@ -184,45 +191,56 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
 
     %State{idle_timeout: idle_timeout} = state
 
+    telemetry_metadata = telemetry_metadata(event, state)
+    start_time = telemetry_start(telemetry_metadata)
+
     case handle_event(event, state) do
-      {:error, _error} = error ->
+      {:error, error} ->
         failure_context = build_failure_context(event, context, nil, state)
 
-        handle_event_error(error, event, failure_context, state)
+        telemetry_stop(start_time, telemetry_metadata, {:error, error})
+
+        handle_event_error({:error, error}, event, failure_context, state)
 
       {:error, error, stacktrace} ->
         failure_context = build_failure_context(event, context, stacktrace, state)
 
-        handle_event_error({:error, error}, event, failure_context, state)
+        telemetry_stop(start_time, telemetry_metadata, {:error, error, stacktrace})
 
-      {:stop, _error, _state} = reply ->
-        reply
+        handle_event_error({:error, error}, event, failure_context, state)
 
       commands ->
         # Copy event id, as causation id, and correlation id from handled event.
         opts = [causation_id: event_id, correlation_id: correlation_id, returning: false]
 
-        with :ok <- commands |> List.wrap() |> dispatch_commands(opts, state, event) do
-          case mutate_state(event, state) do
-            {:error, error, stacktrace} ->
-              failure_context = build_failure_context(event, context, stacktrace, state)
+        commands
+        |> List.wrap()
+        |> dispatch_commands(opts, state, event)
+        |> case do
+          :ok ->
+            telemetry_stop(start_time, telemetry_metadata, commands)
 
-              handle_event_error({:error, error}, event, failure_context, state)
+            case mutate_state(event, state) do
+              {:error, error, stacktrace} ->
+                failure_context = build_failure_context(event, context, stacktrace, state)
 
-            process_state ->
-              state = %State{
-                state
-                | process_state: process_state,
-                  last_seen_event: event_number
-              }
+                handle_event_error({:error, error}, event, failure_context, state)
 
-              :ok = persist_state(event_number, state)
-              :ok = ack_event(event, state)
+              process_state ->
+                state = %State{
+                  state
+                  | process_state: process_state,
+                    last_seen_event: event_number
+                }
+
+                :ok = persist_state(event_number, state)
+                :ok = ack_event(event, state)
+
+                {:noreply, state, idle_timeout}
+            end
 
-              {:noreply, state, idle_timeout}
-          end
-        else
           {:stop, reason} ->
+            telemetry_stop(start_time, telemetry_metadata, {:error, reason})
             {:stop, reason, state}
         end
     end
@@ -516,4 +534,53 @@ defmodule Commanded.ProcessManagers.ProcessManagerInstance do
 
     inspect(process_manager_name) <> "-" <> inspect(process_uuid)
   end
+
+  defp telemetry_start(telemetry_metadata) do
+    Telemetry.start([:commanded, :process_manager, :handle], telemetry_metadata)
+  end
+
+  defp telemetry_stop(start_time, telemetry_metadata, handle_result) do
+    event_prefix = [:commanded, :process_manager, :handle]
+
+    case handle_result do
+      {:error, error} ->
+        telemetry_metadata =
+          telemetry_metadata
+          |> Map.put(:error, error)
+          |> Map.put_new(:commands, [])
+
+        Telemetry.stop(event_prefix, start_time, telemetry_metadata)
+
+      {:error, error, stacktrace} ->
+        Telemetry.exception(
+          event_prefix,
+          start_time,
+          :error,
+          error,
+          stacktrace,
+          telemetry_metadata
+        )
+
+      commands ->
+        commands = List.wrap(commands)
+
+        telemetry_metadata =
+          telemetry_metadata |> Map.put(:commands, commands) |> Map.put(:error, nil)
+
+        Telemetry.stop(event_prefix, start_time, telemetry_metadata)
+    end
+  end
+
+  defp telemetry_metadata(%RecordedEvent{} = event, %State{} = state) do
+    state
+    |> Map.from_struct()
+    |> Map.take([
+      :application,
+      :process_manager_name,
+      :process_manager_module,
+      :process_state,
+      :process_uuid
+    ])
+    |> Map.put(:recorded_event, event)
+  end
 end


Test patch:
diff --git a/test/process_managers/process_manager_telemetry_test.exs b/test/process_managers/process_manager_telemetry_test.exs
new file mode 100644
index 0000000..9e2984c
--- /dev/null
+++ b/test/process_managers/process_manager_telemetry_test.exs
@@ -0,0 +1,316 @@
+defmodule Commanded.ProcessManagers.ProcessManagerTelemetryTest do
+  use ExUnit.Case
+
+  import Mox
+
+  alias Commanded.ProcessManagers.ProcessManagerInstance
+
+  setup :set_mox_global
+  setup :verify_on_exit!
+
+  defmodule Commands do
+    defmodule Ok do
+      defstruct [:message]
+    end
+  end
+
+  defmodule Event do
+    @derive Jason.Encoder
+    defstruct [:message, :type]
+  end
+
+  defmodule Agg do
+    defstruct []
+
+    def execute(_, _) do
+      []
+    end
+
+    def apply(_, _) do
+      %__MODULE__{}
+    end
+  end
+
+  defmodule Router do
+    use Commanded.Commands.Router
+
+    dispatch(Commands.Ok, to: Agg, identity: :message)
+  end
+
+  defmodule App do
+    alias Commanded.EventStore.Adapters.InMemory
+    alias Commanded.Serialization.JsonSerializer
+
+    use Commanded.Application,
+      otp_app: :app,
+      event_store: [
+        adapter: InMemory,
+        serializer: JsonSerializer
+      ],
+      pubsub: :local,
+      registry: :local
+
+    router Router
+  end
+
+  defmodule ExamplePM do
+    use Commanded.ProcessManagers.ProcessManager,
+      application: App,
+      name: __MODULE__
+
+    alias Commands.Ok
+
+    @derive Jason.Encoder
+    defstruct message: "init"
+
+    def handle(%ExamplePM{}, %Event{type: type, message: message}) do
+      case type do
+        "ok" -> %Ok{message: message}
+        "error" -> {:error, message}
+        "retry" -> {:error, :retry}
+        "raise" -> raise message
+      end
+    end
+
+    def apply(%ExamplePM{}, %Event{message: message}),
+      do: %ExamplePM{message: message}
+
+    def error({:error, :retry}, %Event{}, failure_context) do
+      if failure_context.context[:retried?] do
+        :skip
+      else
+        {:retry, %{retried?: true}}
+      end
+    end
+  end
+
+  alias Commands.Ok
+
+  setup do
+    start_supervised!(App)
+
+    attach_telemetry()
+
+    :ok
+  end
+
+  @handler "test-pm-handler"
+
+  describe "process manager telemetry" do
+    test "emit `[:commanded, :process_manager, :handle, :start]` event" do
+      uuid = UUID.uuid4()
+
+      {:ok, instance} = start_process_manager_instance(uuid)
+
+      event = to_recorded_event(%Event{message: "start", type: "ok"})
+
+      :ok = ProcessManagerInstance.process_event(instance, event)
+
+      assert_receive {[:commanded, :process_manager, :handle, :start], 1, measurements, metadata}
+
+      assert match?(%{system_time: _system_time}, measurements)
+
+      assert match?(
+               %{
+                 application: App,
+                 process_manager_module: ExamplePM,
+                 process_manager_name: "ExamplePM",
+                 process_state: %ExamplePM{message: "init"},
+                 process_uuid: ^uuid,
+                 recorded_event: ^event
+               },
+               metadata
+             )
+
+      assert_receive {[:commanded, :process_manager, :handle, :stop], 2, _measurements, _metadata}
+
+      refute_receive {[:commanded, :process_manager, :handle, :exception], _, _measurements,
+                      _metadata}
+    end
+
+    test "emit `[:commanded, :process_manager, :handle, :stop]` event" do
+      uuid = UUID.uuid4()
+
+      {:ok, instance} = start_process_manager_instance(uuid)
+
+      event = to_recorded_event(%Event{message: "start", type: "ok"})
+
+      :ok = ProcessManagerInstance.process_event(instance, event)
+
+      assert_receive {[:commanded, :process_manager, :handle, :start], 1, _measurements,
+                      _metadata}
+
+      assert_receive {[:commanded, :process_manager, :handle, :stop], 2, measurements, metadata}
+
+      assert match?(%{duration: _}, measurements)
+      assert is_integer(measurements.duration)
+
+      assert match?(
+               %{
+                 application: App,
+                 process_manager_module: ExamplePM,
+                 process_manager_name: "ExamplePM",
+                 process_state: %ExamplePM{message: "init"},
+                 process_uuid: ^uuid,
+                 recorded_event: ^event,
+                 commands: [%Ok{message: "start"}]
+               },
+               metadata
+             )
+
+      refute_receive {[:commanded, :process_manager, :handle, :exception], _num, _measurements,
+                      _metadata}
+    end
+
+    test "emit `[:commanded, :process_manager, :handle, :stop]` with error event" do
+      uuid = UUID.uuid4()
+
+      {:ok, instance} = start_process_manager_instance(uuid)
+
+      event = to_recorded_event(%Event{message: "stop", type: "error"})
+
+      :ok = ProcessManagerInstance.process_event(instance, event)
+
+      assert_receive {[:commanded, :process_manager, :handle, :start], 1, _measurements,
+                      _metadata}
+
+      assert_receive {[:commanded, :process_manager, :handle, :stop], 2, measurements, metadata}
+
+      assert match?(%{duration: _}, measurements)
+      assert is_integer(measurements.duration)
+
+      assert match?(
+               %{
+                 application: App,
+                 process_manager_module: ExamplePM,
+                 process_manager_name: "ExamplePM",
+                 process_state: %ExamplePM{message: "init"},
+                 process_uuid: ^uuid,
+                 recorded_event: ^event,
+                 error: "stop"
+               },
+               metadata
+             )
+
+      refute_receive {[:commanded, :process_manager, :handle, :exception], _num, _measurements,
+                      _metadata}
+    end
+
+    test "events are emitted with discrete start/stop on retries" do
+      uuid = UUID.uuid4()
+
+      {:ok, instance} = start_process_manager_instance(uuid)
+
+      event = to_recorded_event(%Event{message: "retry", type: "retry"})
+
+      :ok = ProcessManagerInstance.process_event(instance, event)
+
+      assert_receive {[:commanded, :process_manager, :handle, :start], 1, _measurements,
+                      _metadata}
+
+      assert_receive {[:commanded, :process_manager, :handle, :stop], 2, measurements, metadata}
+
+      assert match?(%{duration: _}, measurements)
+      assert is_integer(measurements.duration)
+
+      assert match?(
+               %{
+                 application: App,
+                 process_manager_module: ExamplePM,
+                 process_manager_name: "ExamplePM",
+                 process_state: %ExamplePM{message: "init"},
+                 process_uuid: ^uuid,
+                 recorded_event: ^event,
+                 error: :retry
+               },
+               metadata
+             )
+
+      refute_receive {[:commanded, :process_manager, :handle, :exception], _num, _measurements,
+                      _metadata}
+
+      assert_receive {[:commanded, :process_manager, :handle, :start], 3, _measurements,
+                      _metadata}
+
+      assert_receive {[:commanded, :process_manager, :handle, :stop], 4, _measurements, _metadata}
+    end
+
+    @tag capture_log: true
+    test "emit `[:commanded, :process_manager, :handle, :exception]` event" do
+      uuid = UUID.uuid4()
+
+      {:ok, instance} = start_process_manager_instance(uuid)
+
+      event = to_recorded_event(%Event{message: "exception", type: "raise"})
+
+      :ok = ProcessManagerInstance.process_event(instance, event)
+
+      assert_receive {[:commanded, :process_manager, :handle, :start], 1, _measurements,
+                      _metadata}
+
+      refute_receive {[:commanded, :process_manager, :handle, :stop], _num, _measurements,
+                      _metadata}
+
+      assert_receive {[:commanded, :process_manager, :handle, :exception], 2, measurements,
+                      metadata}
+
+      assert match?(%{duration: _}, measurements)
+      assert is_integer(measurements.duration)
+
+      assert match?(
+               %{
+                 application: App,
+                 process_manager_module: ExamplePM,
+                 process_manager_name: "ExamplePM",
+                 process_state: %ExamplePM{message: "init"},
+                 process_uuid: ^uuid,
+                 recorded_event: ^event,
+                 kind: :error,
+                 reason: %RuntimeError{message: "exception"},
+                 stacktrace: _
+               },
+               metadata
+             )
+    end
+  end
+
+  defp attach_telemetry do
+    agent = start_supervised!({Agent, fn -> 1 end})
+
+    :telemetry.attach_many(
+      @handler,
+      [
+        [:commanded, :process_manager, :handle, :start],
+        [:commanded, :process_manager, :handle, :stop],
+        [:commanded, :process_manager, :handle, :exception]
+      ],
+      fn event_name, measurements, metadata, reply_to ->
+        num = Agent.get_and_update(agent, fn num -> {num, num + 1} end)
+        send(reply_to, {event_name, num, measurements, metadata})
+      end,
+      self()
+    )
+
+    on_exit(fn ->
+      :telemetry.detach(@handler)
+    end)
+  end
+
+  defp start_process_manager_instance(transfer_uuid) do
+    start_supervised(
+      {ProcessManagerInstance,
+       application: App,
+       idle_timeout: :infinity,
+       process_manager_name: "ExamplePM",
+       process_manager_module: ExamplePM,
+       process_router: self(),
+       process_uuid: transfer_uuid}
+    )
+  end
+
+  defp to_recorded_event(event) do
+    alias Commanded.EventStore.RecordedEvent
+
+    %RecordedEvent{event_number: 1, stream_id: "stream-id", stream_version: 1, data: event}
+  end
+end