feat(telemetry): add and surface processing_latency_ms in event handler spans

lib/commanded/opentelemetry/commanded_attributes.ex

@@ -171,4 +171,11 @@ defmodule Commanded.OpenTelemetry.CommandedAttributes do
   """
   @spec commanded_wrong_expected_version_count() :: :"commanded.wrong_expected_version.count"
   def commanded_wrong_expected_version_count, do: :"commanded.wrong_expected_version.count"
+
+  @doc """
+  Elapsed milliseconds from event creation to handler completion (end-to-end processing latency).
+  For batch handlers, reflects the oldest event in the batch (worst-case latency).
+  """
+  @spec commanded_handler_lag() :: :"commanded.handler.lag"
+  def commanded_handler_lag, do: :"commanded.handler.lag"
 end

lib/commanded/opentelemetry/event_handler.ex

@@ -117,9 +117,11 @@ defmodule Commanded.OpenTelemetry.EventHandler do
     )
   end
 
-  def handle_telemetry_event([:commanded, :event, :handle, :stop], _measurements, meta, _config) do
+  def handle_telemetry_event([:commanded, :event, :handle, :stop], measurements, meta, _config) do
     ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta)
 
+    put_handler_lag(ctx, measurements)
+
     if error = meta[:error] do
       Span.set_attribute(
         ctx,
@@ -209,9 +211,11 @@ defmodule Commanded.OpenTelemetry.EventHandler do
     )
   end
 
-  def batch_telemetry_event([:commanded, :event, :batch, :stop], _measurements, meta, _config) do
+  def batch_telemetry_event([:commanded, :event, :batch, :stop], measurements, meta, _config) do
     ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta)
 
+    put_handler_lag(ctx, measurements)
+
     if error = meta[:error] do
       Span.set_attribute(
         ctx,
@@ -254,6 +258,13 @@ defmodule Commanded.OpenTelemetry.EventHandler do
     OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta)
   end
 
+  defp put_handler_lag(ctx, %{processing_latency_ms: latency})
+       when is_integer(latency) do
+    Span.set_attribute(ctx, CommandedAttributes.commanded_handler_lag(), latency)
+  end
+
+  defp put_handler_lag(_ctx, _measurements), do: :ok
+
   defp put_links(span_opts, []), do: span_opts
   defp put_links(span_opts, links), do: Map.put(span_opts, :links, links)
 end

test/opentelemetry/event_handler_test.exs

@@ -84,7 +84,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do
       recorded_event = meta.recorded_event
 
       :telemetry.span([:commanded, :event, :handle], meta, fn ->
-        {:ok, meta}
+        {:ok, %{processing_latency_ms: 250}, meta}
       end)
 
       assert_receive {:span,
@@ -117,7 +117,8 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do
                "commanded.handler.name": "MyApp.Projectors.AccountProjector",
                "commanded.stream.id": recorded_event.stream_id,
                "commanded.stream.version": 7,
-               "commanded.handler.kind": "event_handler"
+               "commanded.handler.kind": "event_handler",
+               "commanded.handler.lag": 250
              }
     end
   end
@@ -141,7 +142,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do
         )
 
       :telemetry.span([:commanded, :event, :batch], meta, fn ->
-        {:ok, meta}
+        {:ok, %{processing_latency_ms: 500}, meta}
       end)
 
       assert_receive {:span,
@@ -169,7 +170,8 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do
                "commanded.event.count": 100,
                "commanded.handler.kind": "event_handler",
                "commanded.batch.first_event_id": first_event_id,
-               "commanded.batch.last_event_id": last_event_id
+               "commanded.batch.last_event_id": last_event_id,
+               "commanded.handler.lag": 500
              }
     end
   end