diff --git a/lib/commanded/opentelemetry.ex b/lib/commanded/opentelemetry.ex index 81931fa9..0f764904 100644 --- a/lib/commanded/opentelemetry.ex +++ b/lib/commanded/opentelemetry.ex @@ -37,6 +37,7 @@ defmodule Commanded.OpenTelemetry do """ alias Commanded.OpenTelemetry.Aggregate + alias Commanded.OpenTelemetry.Application, as: OTelApplication alias Commanded.OpenTelemetry.EventHandler @typedoc """ @@ -57,6 +58,12 @@ defmodule Commanded.OpenTelemetry do default: [], doc: "Aggregate tracing configuration. Use `:disabled` to disable." ], + application: [ + type: {:in, [:disabled, []]}, + default: [], + doc: + "Application dispatch tracing configuration. Use `:disabled` to disable." + ], event_handler: [ type: {:or, @@ -92,6 +99,9 @@ defmodule Commanded.OpenTelemetry do # Disable aggregate tracing Commanded.OpenTelemetry.setup(aggregate: :disabled) + # Disable application dispatch tracing + Commanded.OpenTelemetry.setup(application: :disabled) + # Disable event handler tracing Commanded.OpenTelemetry.setup(event_handler: :disabled) @@ -108,6 +118,11 @@ defmodule Commanded.OpenTelemetry do _config -> Aggregate.setup() end + case opts[:application] do + :disabled -> :ok + _config -> OTelApplication.setup() + end + case opts[:event_handler] do :disabled -> :ok config -> EventHandler.setup(config) diff --git a/lib/commanded/opentelemetry/application.ex b/lib/commanded/opentelemetry/application.ex new file mode 100644 index 00000000..b90facdc --- /dev/null +++ b/lib/commanded/opentelemetry/application.ex @@ -0,0 +1,125 @@ +defmodule Commanded.OpenTelemetry.Application do + @moduledoc false + + alias Commanded.OpenTelemetry.CommandedAttributes + alias Commanded.OpenTelemetry.Helpers + alias OpenTelemetry.SemConv.ErrorAttributes + alias OpenTelemetry.SemConv.Incubating.CodeAttributes + alias OpenTelemetry.SemConv.Incubating.MessagingAttributes + alias OpenTelemetry.Span + + @tracer_id __MODULE__ + + def setup do + :ok = + :telemetry.attach_many( + {__MODULE__, :dispatch}, + [ + [:commanded, :application, :dispatch, :start], + [:commanded, :application, :dispatch, :stop], + [:commanded, :application, :dispatch, :exception] + ], + &__MODULE__.handle_telemetry_event/4, + %{} + ) + end + + def handle_telemetry_event( + [:commanded, :application, :dispatch, :start], + _measurements, + meta, + _config + ) do + context = meta.execution_context + + # Attach trace context from command metadata if present. + # The TraceContextPropagator middleware injects W3C traceparent/tracestate headers + # into metadata, allowing explicit context propagation when needed. + Helpers.attach_ctx(context.metadata) + + handler_module_name = Helpers.module_name(context.handler) + + attributes = [ + # OTel Messaging SemConv + {MessagingAttributes.messaging_system(), "commanded"}, + {MessagingAttributes.messaging_operation_type(), :receive}, + {MessagingAttributes.messaging_operation_name(), "dispatch"}, + {MessagingAttributes.messaging_destination_name(), handler_module_name}, + {MessagingAttributes.messaging_message_id(), context.causation_id}, + {MessagingAttributes.messaging_message_conversation_id(), context.correlation_id}, + # OTel Code SemConv + {CodeAttributes.code_function(), to_string(context.function)}, + {CodeAttributes.code_namespace(), handler_module_name}, + # Commanded-specific + {CommandedAttributes.commanded_handler_kind(), "command_handler"}, + {CommandedAttributes.commanded_application(), meta.application}, + {CommandedAttributes.commanded_command(), Helpers.struct_name(context.command)}, + {CommandedAttributes.commanded_correlation_id(), context.correlation_id}, + {CommandedAttributes.commanded_causation_id(), context.causation_id} + ] + + # OTel semconv: span name = "{operation.name} {destination.name}" + span_name = "dispatch #{handler_module_name}" + + OpentelemetryTelemetry.start_telemetry_span( + @tracer_id, + span_name, + meta, + %{ + kind: :consumer, + attributes: attributes + } + ) + end + + def handle_telemetry_event( + [:commanded, :application, :dispatch, :stop], + _measurements, + meta, + _config + ) do + ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) + + if error = meta[:error] do + Span.set_attribute( + ctx, + ErrorAttributes.error_type(), + Helpers.to_error_type(error, @tracer_id) + ) + + Span.set_status(ctx, OpenTelemetry.status(:error, Helpers.format_error(error))) + end + + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) + end + + def handle_telemetry_event( + [:commanded, :application, :dispatch, :exception], + _measurements, + %{kind: kind, reason: reason, stacktrace: stacktrace} = meta, + _config + ) do + ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) + + # TODO: follow up with OTEL team to add exception.kind SemConv attribute + Span.set_attribute(ctx, :"erlang.exception.kind", kind) + + # Normalize all errors to Elixir exceptions + exception = Exception.normalize(kind, reason, stacktrace) + + Span.set_attribute( + ctx, + ErrorAttributes.error_type(), + Helpers.to_error_type(exception, @tracer_id) + ) + + Span.record_exception(ctx, exception, stacktrace) + + Span.set_status( + ctx, + OpenTelemetry.status(:error, Exception.format_banner(kind, reason, stacktrace)) + ) + + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) + end +end diff --git a/test/opentelemetry/aggregate_test.exs b/test/opentelemetry/aggregate_test.exs index 997e9134..4775a62c 100644 --- a/test/opentelemetry/aggregate_test.exs +++ b/test/opentelemetry/aggregate_test.exs @@ -15,7 +15,10 @@ defmodule Commanded.OpenTelemetry.AggregateTest do setup do start_supervised!(DefaultApp) + + detach_handlers() Aggregate.setup() + :ok end diff --git a/test/opentelemetry/application_test.exs b/test/opentelemetry/application_test.exs new file mode 100644 index 00000000..6959f908 --- /dev/null +++ b/test/opentelemetry/application_test.exs @@ -0,0 +1,294 @@ +defmodule Commanded.OpenTelemetry.ApplicationTest do + use Commanded.OpenTelemetryCase, async: false + + alias Commanded.DefaultApp + alias Commanded.Middleware.Commands.IncrementCount + alias Commanded.OpenTelemetry.Application, as: OTelApplication + alias Commanded.TestSupport.Factory + alias Commanded.UUID + + setup do + start_supervised!(DefaultApp) + + detach_handlers() + OTelApplication.setup() + + :ok + end + + defmodule TestRouter do + use Commanded.Commands.Router + + alias Commanded.Middleware.Commands.CommandHandler + alias Commanded.Middleware.Commands.CounterAggregateRoot + + dispatch IncrementCount, + to: CommandHandler, + aggregate: CounterAggregateRoot, + identity: :aggregate_uuid + end + + describe "setup/0" do + test "attaches telemetry handlers for application dispatch events" do + detach_handlers() + + OTelApplication.setup() + + for event <- [ + [:commanded, :application, :dispatch, :start], + [:commanded, :application, :dispatch, :stop], + [:commanded, :application, :dispatch, :exception] + ] do + handlers = :telemetry.list_handlers(event) + + assert Enum.any?( + handlers, + &match?(%{id: {OTelApplication, :dispatch}}, &1) + ), + "Expected handler for event #{inspect(event)}" + end + end + + test "calling setup twice raises MatchError (fail fast)" do + detach_handlers() + + :ok = OTelApplication.setup() + + handlers = :telemetry.list_handlers([:commanded, :application, :dispatch, :start]) + assert length(handlers) == 1 + + assert_raise MatchError, fn -> + OTelApplication.setup() + end + end + end + + describe "application dispatch" do + test "creates span on successful dispatch" do + aggregate_uuid = UUID.uuid4() + causation_id = UUID.uuid4() + correlation_id = UUID.uuid4() + + command = %IncrementCount{aggregate_uuid: aggregate_uuid} + + assert :ok = + TestRouter.dispatch(command, + application: DefaultApp, + command_uuid: causation_id, + correlation_id: correlation_id + ) + + assert_receive {:span, + span( + name: "dispatch Commanded.Middleware.Commands.CommandHandler", + kind: :consumer, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "dispatch", + "messaging.destination.name": "Commanded.Middleware.Commands.CommandHandler", + "messaging.message.id": causation_id, + "messaging.message.conversation_id": correlation_id, + "code.function": "handle", + "code.namespace": "Commanded.Middleware.Commands.CommandHandler", + "commanded.handler.kind": "command_handler", + "commanded.application": Commanded.DefaultApp, + "commanded.command": "Commanded.Middleware.Commands.IncrementCount", + "commanded.correlation_id": correlation_id, + "commanded.causation_id": causation_id + } + end + end + + describe "error handling" do + setup do + detach_handlers() + OTelApplication.setup() + :ok + end + + test "sets error status when dispatch returns error in stop" do + 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, :error, error_message}, + attributes: span_attrs + )}, + 1000 + + assert error_message == ":validation_failed" + + assert :otel_attributes.map(span_attrs) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "dispatch", + "messaging.destination.name": "Commanded.TestSupport.TestDomain.Account", + "messaging.message.id": causation_id, + "messaging.message.conversation_id": correlation_id, + "code.function": "execute", + "code.namespace": "Commanded.TestSupport.TestDomain.Account", + "commanded.handler.kind": "command_handler", + "commanded.application": MockApp, + "commanded.command": "Commanded.TestSupport.TestDomain.OpenAccount", + "commanded.correlation_id": correlation_id, + "commanded.causation_id": causation_id, + "error.type": "validation_failed" + } + end + + test "handles ArgumentError exception" do + {_event_name, _measurements, meta} = + Factory.build_telemetry_event(:application_dispatch_exception, + reason: %ArgumentError{message: "invalid command argument"} + ) + + context = meta.execution_context + + :telemetry.execute([:commanded, :application, :dispatch, :start], %{}, meta) + + :telemetry.execute( + [:commanded, :application, :dispatch, :exception], + %{duration: 100}, + meta + ) + + assert_receive {:span, + span( + name: "dispatch Commanded.TestSupport.TestDomain.Account", + status: {:status, :error, error_msg}, + attributes: span_attrs + )}, + 1000 + + assert error_msg == "** (ArgumentError) invalid command argument" + + assert :otel_attributes.map(span_attrs) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "dispatch", + "messaging.destination.name": "Commanded.TestSupport.TestDomain.Account", + "messaging.message.id": context.causation_id, + "messaging.message.conversation_id": context.correlation_id, + "code.function": to_string(context.function), + "code.namespace": "Commanded.TestSupport.TestDomain.Account", + "commanded.handler.kind": "command_handler", + "commanded.application": meta.application, + "commanded.command": "Commanded.TestSupport.TestDomain.OpenAccount", + "commanded.correlation_id": context.correlation_id, + "commanded.causation_id": context.causation_id, + "erlang.exception.kind": :error, + "error.type": "Elixir.ArgumentError" + } + end + + test "handles RuntimeError exception" do + {_event_name, _measurements, meta} = + Factory.build_telemetry_event(:application_dispatch_exception, + reason: %RuntimeError{message: "dispatch execution failed"} + ) + + context = meta.execution_context + + :telemetry.execute([:commanded, :application, :dispatch, :start], %{}, meta) + + :telemetry.execute( + [:commanded, :application, :dispatch, :exception], + %{duration: 100}, + meta + ) + + assert_receive {:span, + span( + name: "dispatch Commanded.TestSupport.TestDomain.Account", + status: {:status, :error, error_msg}, + attributes: span_attrs + )}, + 1000 + + assert error_msg == "** (RuntimeError) dispatch execution failed" + + assert :otel_attributes.map(span_attrs) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "dispatch", + "messaging.destination.name": "Commanded.TestSupport.TestDomain.Account", + "messaging.message.id": context.causation_id, + "messaging.message.conversation_id": context.correlation_id, + "code.function": to_string(context.function), + "code.namespace": "Commanded.TestSupport.TestDomain.Account", + "commanded.handler.kind": "command_handler", + "commanded.application": meta.application, + "commanded.command": "Commanded.TestSupport.TestDomain.OpenAccount", + "commanded.correlation_id": context.correlation_id, + "commanded.causation_id": context.causation_id, + "erlang.exception.kind": :error, + "error.type": "Elixir.RuntimeError" + } + end + + test "records exception event with proper attributes" do + {_event_name, _measurements, meta} = + Factory.build_telemetry_event(:application_dispatch_exception, + reason: %RuntimeError{message: "failed"} + ) + + :telemetry.execute([:commanded, :application, :dispatch, :start], %{}, meta) + + :telemetry.execute( + [:commanded, :application, :dispatch, :exception], + %{duration: 100}, + meta + ) + + assert_receive {:span, + span( + name: "dispatch Commanded.TestSupport.TestDomain.Account", + events: events + )}, + 1000 + + events_list = :otel_events.list(events) + exception_event = Enum.find(events_list, fn event -> elem(event, 2) == :exception end) + {:event, _timestamp, :exception, attrs_tuple} = exception_event + {:attributes, _, _, _, attrs_map} = attrs_tuple + + assert attrs_map[:"exception.type"] == "Elixir.RuntimeError" + assert attrs_map[:"exception.message"] == "failed" + end + end + + defp detach_handlers do + for event <- [ + [:commanded, :application, :dispatch, :start], + [:commanded, :application, :dispatch, :stop], + [:commanded, :application, :dispatch, :exception] + ] do + for handler <- :telemetry.list_handlers(event) do + :telemetry.detach(handler.id) + end + end + end +end diff --git a/test/support/factory.ex b/test/support/factory.ex index 84cf0acc..c8d89d94 100644 --- a/test/support/factory.ex +++ b/test/support/factory.ex @@ -1,4 +1,5 @@ defmodule Commanded.TestSupport.Factory do + alias Commanded.Aggregates.ExecutionContext alias Commanded.EventStore.RecordedEvent alias Commanded.TestSupport.TestDomain alias Commanded.UUID @@ -482,6 +483,59 @@ defmodule Commanded.TestSupport.Factory do ) end + def build_application_dispatch_metadata(opts \\ []) do + causation_id = Keyword.get(opts, :causation_id, UUID.uuid4()) + correlation_id = Keyword.get(opts, :correlation_id, UUID.uuid4()) + + command = + Keyword.get_lazy(opts, :command, fn -> + build_open_account() + end) + + execution_context = + Keyword.get_lazy(opts, :execution_context, fn -> + %ExecutionContext{ + command: command, + causation_id: causation_id, + correlation_id: correlation_id, + handler: Keyword.get(opts, :handler, TestDomain.Account), + function: Keyword.get(opts, :function, :execute), + metadata: Keyword.get(opts, :metadata, %{}) + } + end) + + defaults = [ + application: Keyword.get(opts, :application, MockApp), + execution_context: execution_context + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + execution_context: Keyword.fetch!(opts, :execution_context), + error: nil + } + end + + def build_application_dispatch_stop_metadata(opts \\ []) do + base = build_application_dispatch_metadata(opts) + + Map.merge(base, %{ + error: Keyword.get(opts, :error, nil) + }) + end + + def build_application_dispatch_exception_metadata(opts \\ []) do + base = build_application_dispatch_metadata(opts) + + Map.merge(base, %{ + kind: Keyword.get(opts, :kind, :error), + reason: Keyword.get(opts, :reason, %RuntimeError{message: "Test error"}), + stacktrace: Keyword.get(opts, :stacktrace, []) + }) + end + def build_aggregate_execute_metadata(opts \\ []) do aggregate_uuid = Keyword.get(opts, :aggregate_uuid, UUID.uuid4()) @@ -608,6 +662,30 @@ defmodule Commanded.TestSupport.Factory do {event_name, measurements, metadata} end + def build_telemetry_event(:application_dispatch_start, opts) do + measurements = build_telemetry_start_measurements() + metadata = build_application_dispatch_metadata(opts) + event_name = [:commanded, :application, :dispatch, :start] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:application_dispatch_stop, opts) do + measurements = build_telemetry_stop_measurements() + metadata = build_application_dispatch_stop_metadata(opts) + event_name = [:commanded, :application, :dispatch, :stop] + + {event_name, measurements, metadata} + end + + def build_telemetry_event(:application_dispatch_exception, opts) do + measurements = build_telemetry_exception_measurements() + metadata = build_application_dispatch_exception_metadata(opts) + event_name = [:commanded, :application, :dispatch, :exception] + + {event_name, measurements, metadata} + end + def build_account_scenario(opts \\ []) do account_id = Keyword.get(opts, :account_id, UUID.uuid4()) owner = Keyword.get(opts, :owner, "Test User") diff --git a/test/support/opentelemetry_case.ex b/test/support/opentelemetry_case.ex index 1aa97435..f8e3fa9d 100644 --- a/test/support/opentelemetry_case.ex +++ b/test/support/opentelemetry_case.ex @@ -44,7 +44,10 @@ defmodule Commanded.OpenTelemetryCase do [:commanded, :event, :batch, :exception], [:commanded, :aggregate, :execute, :start], [:commanded, :aggregate, :execute, :stop], - [:commanded, :aggregate, :execute, :exception] + [:commanded, :aggregate, :execute, :exception], + [:commanded, :application, :dispatch, :start], + [:commanded, :application, :dispatch, :stop], + [:commanded, :application, :dispatch, :exception] ] for event <- commanded_events,