diff --git a/lib/commanded/opentelemetry/commanded_attributes.ex b/lib/commanded/opentelemetry/commanded_attributes.ex index eaf3b9f8..9fa5b9fd 100644 --- a/lib/commanded/opentelemetry/commanded_attributes.ex +++ b/lib/commanded/opentelemetry/commanded_attributes.ex @@ -15,6 +15,41 @@ defmodule Commanded.OpenTelemetry.CommandedAttributes do - Use dot notation for namespacing (e.g., `commanded.stream.version`). - To extend a SemConv namespace, use `commanded..*` (e.g., `commanded.messaging.*`). + ## Latency vs Lag + + In messaging systems (Kafka, Pulsar, NATS, RabbitMQ), **"lag"** universally refers + to **offset-based distance** — how many messages the consumer is behind the head of + the stream. It is a count, not a duration. + + This library follows that convention: + + - **"latency"** = a time-based duration (e.g., `commanded.handler.processing_latency` + measures elapsed milliseconds from event creation to handler processing). + - **"lag"** = reserved for offset-based distance (e.g., a future + `commanded.subscription.lag` would measure how many events a subscription is + behind the stream head). + + ## Event Handler Span Attributes + + | Attribute | Type | Description | + |---|---|---| + | `commanded.application` | string | The Commanded application module | + | `commanded.handler.name` | string | The handler module name | + | `commanded.handler.kind` | string | Type of handler (`event_handler`) | + | `commanded.stream.id` | string | The event's stream identifier | + | `commanded.stream.version` | integer | The event's stream version | + | `commanded.correlation_id` | string | Correlation ID for tracing causality | + | `commanded.causation_id` | string | Causation ID for tracing causality | + | `commanded.handler.processing_latency` | integer | Milliseconds from event creation to handler processing (delivery latency) | + + Batch event handler spans additionally include: + + | Attribute | Type | Description | + |---|---|---| + | `commanded.event.count` | integer | Number of events in the batch | + | `commanded.batch.first_event_id` | string | Event ID of the first event in the batch | + | `commanded.batch.last_event_id` | string | Event ID of the last event in the batch | + ## Example iex> Commanded.OpenTelemetry.CommandedAttributes.commanded_event() @@ -173,11 +208,20 @@ defmodule Commanded.OpenTelemetry.CommandedAttributes do 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). + Elapsed milliseconds from event creation to handler processing (delivery latency). + + Measures the age of the event at processing time: `now - event.created_at`. This + reflects how long the event waited in the store before the handler picked it up, + which is a signal of subscription health and backpressure. + For batch handlers, reflects the oldest event in the batch (worst-case latency). + + This is distinct from `messaging.process.duration` (how long the handler took to + execute) and from offset-based "lag" (how many events behind the stream head). See + the module-level "Latency vs Lag" section for naming rationale. """ - @spec commanded_handler_lag() :: :"commanded.handler.lag" - def commanded_handler_lag, do: :"commanded.handler.lag" + @spec commanded_handler_processing_latency() :: :"commanded.handler.processing_latency" + def commanded_handler_processing_latency, do: :"commanded.handler.processing_latency" @doc """ The process registry adapter module (e.g. Commanded.Registration.GlobalRegistry). diff --git a/lib/commanded/opentelemetry/event_handler.ex b/lib/commanded/opentelemetry/event_handler.ex index 2d9b2cba..d1ddef53 100644 --- a/lib/commanded/opentelemetry/event_handler.ex +++ b/lib/commanded/opentelemetry/event_handler.ex @@ -120,7 +120,7 @@ defmodule Commanded.OpenTelemetry.EventHandler 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) + put_handler_processing_latency(ctx, measurements) if error = meta[:error] do Span.set_attribute( @@ -214,7 +214,7 @@ defmodule Commanded.OpenTelemetry.EventHandler 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) + put_handler_processing_latency(ctx, measurements) if error = meta[:error] do Span.set_attribute( @@ -258,12 +258,12 @@ defmodule Commanded.OpenTelemetry.EventHandler do OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) end - defp put_handler_lag(ctx, %{processing_latency_ms: latency}) + defp put_handler_processing_latency(ctx, %{processing_latency_ms: latency}) when is_integer(latency) do - Span.set_attribute(ctx, CommandedAttributes.commanded_handler_lag(), latency) + Span.set_attribute(ctx, CommandedAttributes.commanded_handler_processing_latency(), latency) end - defp put_handler_lag(_ctx, _measurements), do: :ok + defp put_handler_processing_latency(_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) diff --git a/test/opentelemetry/event_handler_test.exs b/test/opentelemetry/event_handler_test.exs index 9c1b4f8e..b24452cb 100644 --- a/test/opentelemetry/event_handler_test.exs +++ b/test/opentelemetry/event_handler_test.exs @@ -118,7 +118,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do "commanded.stream.id": recorded_event.stream_id, "commanded.stream.version": 7, "commanded.handler.kind": "event_handler", - "commanded.handler.lag": 250 + "commanded.handler.processing_latency": 250 } end end @@ -171,7 +171,7 @@ defmodule Commanded.OpenTelemetry.EventHandlerTest do "commanded.handler.kind": "event_handler", "commanded.batch.first_event_id": first_event_id, "commanded.batch.last_event_id": last_event_id, - "commanded.handler.lag": 500 + "commanded.handler.processing_latency": 500 } end end