diff --git a/guides/howtos/setting-up-opentelemetry-tracing.md b/guides/howtos/setting-up-opentelemetry-tracing.md index 846aca1a..f03b3e17 100644 --- a/guides/howtos/setting-up-opentelemetry-tracing.md +++ b/guides/howtos/setting-up-opentelemetry-tracing.md @@ -61,3 +61,37 @@ Note: `setup/1` should only be called once during application startup. Commanded.OpenTelemetry.setup(event_handler: :disabled) ``` +## Override Returned Error Status + +You can override the span status used for returned `:stop` errors on application +dispatch and aggregate execution spans: + +```elixir +Commanded.OpenTelemetry.setup( + application: [ + error_status: fn + _event_name, _measurements, %{error: :validation_failed}, _config -> :unset + _event_name, _measurements, _meta, _config -> :error + end + ], + aggregate: [ + error_status: fn + _event_name, _measurements, %{error: :validation_failed}, _config -> :unset + _event_name, _measurements, _meta, _config -> :error + end + ] +) +``` + +The callback receives the telemetry event name first, then the stop measurements, +then the full telemetry metadata map, and finally the instrumentation config. It +may return: + +- `:unset`, `:ok`, or `:error` +- `nil` to leave the status unset + +When the callback returns `:error`, Commanded derives the status description from +the returned error using `Commanded.OpenTelemetry.Helpers.format_error/1`. + +Exceptions are not routed through this callback; they continue to record +exception events and set span status to error. diff --git a/lib/commanded/opentelemetry.ex b/lib/commanded/opentelemetry.ex index c4073dc3..ba66abb2 100644 --- a/lib/commanded/opentelemetry.ex +++ b/lib/commanded/opentelemetry.ex @@ -55,9 +55,36 @@ defmodule Commanded.OpenTelemetry do """ @type span_relationship :: :link | :child | :none + @typedoc """ + Callback used to decide the OpenTelemetry span status for returned `:stop` errors. + + The callback is only invoked when Commanded emits stop metadata containing `:error`. + Exception telemetry continues to use OpenTelemetry exception semantics directly. + """ + @type error_status_callback :: + (event_name :: [atom()], + measurements :: map(), + metadata :: map(), + config :: keyword() -> + OpenTelemetry.status_code() | nil) + + @error_status_options [ + error_status: [ + type: {:fun, 4}, + type_doc: "`t:error_status_callback/0`", + doc: + "Override the span status for returned `:stop` errors. Return `nil` to leave the status unset." + ] + ] + @nimble_schema NimbleOptions.new!( aggregate: [ - type: {:in, [:disabled, []]}, + type: + {:or, + [ + {:in, [:disabled]}, + keyword_list: @error_status_options + ]}, default: [], doc: "Aggregate tracing configuration. Use `:disabled` to disable." ], @@ -72,7 +99,12 @@ defmodule Commanded.OpenTelemetry do doc: "Aggregate snapshot tracing configuration. Use `:disabled` to disable." ], application: [ - type: {:in, [:disabled, []]}, + type: + {:or, + [ + {:in, [:disabled]}, + keyword_list: @error_status_options + ]}, default: [], doc: "Application dispatch tracing configuration. Use `:disabled` to disable." @@ -123,6 +155,16 @@ defmodule Commanded.OpenTelemetry do # Disable event handler tracing Commanded.OpenTelemetry.setup(event_handler: :disabled) + # Leave returned domain errors unset for dispatch spans + Commanded.OpenTelemetry.setup( + application: [ + error_status: fn + _event_name, _measurements, %{error: :validation_failed}, _config -> :unset + _event_name, _measurements, _meta, _config -> :error + end + ] + ) + # Disable aggregate populate tracing Commanded.OpenTelemetry.setup(aggregate_populate: :disabled) @@ -142,7 +184,7 @@ defmodule Commanded.OpenTelemetry do case opts[:aggregate] do :disabled -> :ok - _config -> Aggregate.setup() + config -> Aggregate.setup(config) end case opts[:aggregate_populate] do @@ -157,7 +199,7 @@ defmodule Commanded.OpenTelemetry do case opts[:application] do :disabled -> :ok - _config -> OTelApplication.setup() + config -> OTelApplication.setup(config) end case opts[:event_handler] do diff --git a/lib/commanded/opentelemetry/aggregate.ex b/lib/commanded/opentelemetry/aggregate.ex index a5530909..078a0b0f 100644 --- a/lib/commanded/opentelemetry/aggregate.ex +++ b/lib/commanded/opentelemetry/aggregate.ex @@ -10,7 +10,7 @@ defmodule Commanded.OpenTelemetry.Aggregate do @tracer_id __MODULE__ - def setup do + def setup(config \\ []) do :ok = :telemetry.attach_many( {__MODULE__, :execute}, @@ -20,7 +20,7 @@ defmodule Commanded.OpenTelemetry.Aggregate do [:commanded, :aggregate, :execute, :exception] ], &__MODULE__.handle_telemetry_event/4, - %{} + config ) end @@ -79,10 +79,11 @@ defmodule Commanded.OpenTelemetry.Aggregate do def handle_telemetry_event( [:commanded, :aggregate, :execute, :stop], - _measurements, + measurements, meta, - _config + config ) do + event_name = [:commanded, :aggregate, :execute, :stop] ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) events = Map.get(meta, :events, []) @@ -95,7 +96,7 @@ defmodule Commanded.OpenTelemetry.Aggregate do Helpers.to_error_type(error, @tracer_id) ) - Span.set_status(ctx, OpenTelemetry.status(:error, Helpers.format_error(error))) + Helpers.set_error_status(ctx, error, event_name, measurements, meta, config, @tracer_id) end wrong_expected_version_count = Map.get(meta, :wrong_expected_version_count, 0) diff --git a/lib/commanded/opentelemetry/application.ex b/lib/commanded/opentelemetry/application.ex index ff41a3a6..e26baff8 100644 --- a/lib/commanded/opentelemetry/application.ex +++ b/lib/commanded/opentelemetry/application.ex @@ -10,7 +10,7 @@ defmodule Commanded.OpenTelemetry.Application do @tracer_id __MODULE__ - def setup do + def setup(config \\ []) do :ok = :telemetry.attach_many( {__MODULE__, :dispatch}, @@ -20,7 +20,7 @@ defmodule Commanded.OpenTelemetry.Application do [:commanded, :application, :dispatch, :exception] ], &__MODULE__.handle_telemetry_event/4, - %{} + config ) end @@ -69,10 +69,11 @@ defmodule Commanded.OpenTelemetry.Application do def handle_telemetry_event( [:commanded, :application, :dispatch, :stop], - _measurements, + measurements, meta, - _config + config ) do + event_name = [:commanded, :application, :dispatch, :stop] ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) if error = meta[:error] do @@ -82,7 +83,7 @@ defmodule Commanded.OpenTelemetry.Application do Helpers.to_error_type(error, @tracer_id) ) - Span.set_status(ctx, OpenTelemetry.status(:error, Helpers.format_error(error))) + Helpers.set_error_status(ctx, error, event_name, measurements, meta, config, @tracer_id) end OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) diff --git a/lib/commanded/opentelemetry/helpers.ex b/lib/commanded/opentelemetry/helpers.ex index b73244da..70fc14f0 100644 --- a/lib/commanded/opentelemetry/helpers.ex +++ b/lib/commanded/opentelemetry/helpers.ex @@ -1,6 +1,8 @@ defmodule Commanded.OpenTelemetry.Helpers do @moduledoc false + alias OpenTelemetry.Span + def extract_propagated_ctx(nil), do: {[], :undefined} def extract_propagated_ctx(metadata) when is_map(metadata) do @@ -65,6 +67,41 @@ defmodule Commanded.OpenTelemetry.Helpers do def format_error(error) when is_binary(error), do: error def format_error(error), do: inspect(error) + def set_error_status(ctx, error, event_name, measurements, meta, config, tracer_id) do + status_code = + case Keyword.get(config, :error_status) do + nil -> :error + fun when is_function(fun, 4) -> fun.(event_name, measurements, meta, config) + end + + apply_error_status(ctx, status_code, error, tracer_id) + end + + defp apply_error_status(_ctx, nil, _error, _tracer_id), do: :ok + + defp apply_error_status(ctx, :error, error, _tracer_id) do + Span.set_status(ctx, OpenTelemetry.status(:error, format_error(error))) + end + + defp apply_error_status(ctx, code, _error, _tracer_id) when code in [:unset, :ok] do + Span.set_status(ctx, OpenTelemetry.status(code)) + end + + defp apply_error_status(ctx, status_code, error, tracer_id) do + :telemetry.execute( + [:commanded, :opentelemetry, :warning], + %{count: 1}, + %{ + message: "Unknown error status encountered, falling back to error status", + error: error, + error_status: status_code, + tracer_id: tracer_id + } + ) + + Span.set_status(ctx, OpenTelemetry.status(:error, format_error(error))) + end + def module_name(nil), do: nil def module_name(module) when is_atom(module), do: inspect(module) def module_name(_), do: nil diff --git a/test/opentelemetry/aggregate_test.exs b/test/opentelemetry/aggregate_test.exs index 66f7a4f5..790f7dfb 100644 --- a/test/opentelemetry/aggregate_test.exs +++ b/test/opentelemetry/aggregate_test.exs @@ -324,6 +324,64 @@ defmodule Commanded.OpenTelemetry.AggregateTest do } end + test "allows returned aggregate error status to stay unset" do + detach_handlers() + + OTelAggregate.setup( + error_status: fn _event_name, _measurements, _meta, _config -> :unset end + ) + + aggregate_uuid = UUID.uuid4() + causation_id = UUID.uuid4() + correlation_id = UUID.uuid4() + + meta = + Factory.build_aggregate_execute_metadata( + aggregate_uuid: aggregate_uuid, + causation_id: causation_id, + correlation_id: correlation_id + ) + + :telemetry.execute([:commanded, :aggregate, :execute, :start], %{}, meta) + + stop_meta = Map.put(meta, :error, :validation_failed) + :telemetry.execute([:commanded, :aggregate, :execute, :stop], %{duration: 1000}, stop_meta) + + assert_receive {:span, + span( + name: "execute Commanded.TestSupport.TestDomain.Account", + status: status, + attributes: span_attrs + )}, + 1000 + + assert status == {:status, :unset, ""} + assert :otel_attributes.map(span_attrs)[:"error.type"] == "validation_failed" + end + + test "uses the formatted aggregate error message when callback returns error" do + detach_handlers() + + OTelAggregate.setup( + error_status: fn _event_name, _measurements, _meta, _config -> :error end + ) + + meta = Factory.build_aggregate_execute_metadata(aggregate_uuid: UUID.uuid4()) + + :telemetry.execute([:commanded, :aggregate, :execute, :start], %{}, meta) + + :telemetry.execute( + [:commanded, :aggregate, :execute, :stop], + %{duration: 1000}, + Map.put(meta, :error, :validation_failed) + ) + + assert_receive {:span, span(status: {:status, :error, error_message})}, + 1000 + + assert error_message == ":validation_failed" + end + test "handles ArgumentError exception" do # Commanded's rescue blocks always emit kind: :error {_event_name, _measurements, meta} = diff --git a/test/opentelemetry/application_test.exs b/test/opentelemetry/application_test.exs index 8194c662..8864b115 100644 --- a/test/opentelemetry/application_test.exs +++ b/test/opentelemetry/application_test.exs @@ -234,6 +234,105 @@ defmodule Commanded.OpenTelemetry.ApplicationTest do } end + test "allows returned error status to stay unset" do + detach_handlers() + test_pid = self() + + OTelApplication.setup( + error_status: fn event_name, measurements, meta, config -> + send(test_pid, {:error_status_callback, event_name, measurements, meta, config}) + :unset + end + ) + + causation_id = UUID.uuid4() + correlation_id = UUID.uuid4() + + meta = + Factory.build_application_dispatch_metadata( + causation_id: causation_id, + correlation_id: correlation_id + ) + + :telemetry.execute([:commanded, :application, :dispatch, :start], %{}, meta) + + stop_meta = Map.put(meta, :error, :validation_failed) + + :telemetry.execute( + [:commanded, :application, :dispatch, :stop], + %{duration: 1000}, + stop_meta + ) + + assert_receive {:span, + span( + name: "dispatch Commanded.TestSupport.TestDomain.Account", + status: status, + attributes: span_attrs + )}, + 1000 + + assert_receive {:error_status_callback, callback_event_name, callback_measurements, + callback_meta, callback_config}, + 1000 + + assert callback_event_name == [:commanded, :application, :dispatch, :stop] + assert callback_measurements.duration == 1000 + assert callback_meta.error == :validation_failed + assert is_function(Keyword.fetch!(callback_config, :error_status), 4) + assert status == {:status, :unset, ""} + assert :otel_attributes.map(span_attrs)[:"error.type"] == "validation_failed" + end + + test "uses the formatted error message when callback returns error" do + detach_handlers() + + OTelApplication.setup( + error_status: fn _event_name, _measurements, _meta, _config -> :error end + ) + + meta = Factory.build_application_dispatch_metadata() + + :telemetry.execute([:commanded, :application, :dispatch, :start], %{}, meta) + + :telemetry.execute( + [:commanded, :application, :dispatch, :stop], + %{duration: 1000}, + Map.put(meta, :error, :validation_failed) + ) + + assert_receive {:span, span(status: {:status, :error, error_message})}, + 1000 + + assert error_message == ":validation_failed" + end + + test "keeps exception status handling unchanged when error_status is configured" do + detach_handlers() + + OTelApplication.setup( + error_status: fn _event_name, _measurements, _meta, _config -> :unset end + ) + + {_event_name, _measurements, meta} = + Factory.build_telemetry_event(:application_dispatch_exception, + reason: %ArgumentError{message: "invalid command argument"} + ) + + :telemetry.execute([:commanded, :application, :dispatch, :start], %{}, meta) + + :telemetry.execute( + [:commanded, :application, :dispatch, :exception], + %{duration: 100}, + meta + ) + + assert_receive {:span, span(status: {:status, :error, error_msg})}, + 1000 + + assert error_msg == "** (ArgumentError) invalid command argument" + end + test "handles ArgumentError exception" do {_event_name, _measurements, meta} = Factory.build_telemetry_event(:application_dispatch_exception,