Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 47 additions & 3 deletions lib/commanded/opentelemetry/commanded_attributes.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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.<namespace>.*` (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()
Expand Down Expand Up @@ -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"
Comment thread
yordis marked this conversation as resolved.

@doc """
The process registry adapter module (e.g. Commanded.Registration.GlobalRegistry).
Expand Down
10 changes: 5 additions & 5 deletions lib/commanded/opentelemetry/event_handler.ex
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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)
Expand Down
4 changes: 2 additions & 2 deletions test/opentelemetry/event_handler_test.exs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down