diff --git a/lib/commanded/opentelemetry/commanded_attributes.ex b/lib/commanded/opentelemetry/commanded_attributes.ex index c0c48aae..2104f6db 100644 --- a/lib/commanded/opentelemetry/commanded_attributes.ex +++ b/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 diff --git a/lib/commanded/opentelemetry/event_handler.ex b/lib/commanded/opentelemetry/event_handler.ex index bd8ee307..ef143690 100644 --- a/lib/commanded/opentelemetry/event_handler.ex +++ b/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 diff --git a/test/opentelemetry/event_handler_test.exs b/test/opentelemetry/event_handler_test.exs index b318a9ab..4338095a 100644 --- a/test/opentelemetry/event_handler_test.exs +++ b/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