From 03c697d3d02aec85822a6872c3503898848e2210 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Mon, 19 Jan 2026 16:25:40 -0500 Subject: [PATCH] feat: Add OpenTelemetry instrumentation for aggregate populate Signed-off-by: Yordis Prieto --- lib/commanded/opentelemetry.ex | 14 ++ .../opentelemetry/aggregate_populate.ex | 73 +++++++ .../opentelemetry/aggregate_populate_test.exs | 182 ++++++++++++++++++ test/support/factory.ex | 20 ++ test/support/opentelemetry_case.ex | 2 + 5 files changed, 291 insertions(+) create mode 100644 lib/commanded/opentelemetry/aggregate_populate.ex create mode 100644 test/opentelemetry/aggregate_populate_test.exs diff --git a/lib/commanded/opentelemetry.ex b/lib/commanded/opentelemetry.ex index 0f764904..6b31257e 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.AggregatePopulate alias Commanded.OpenTelemetry.Application, as: OTelApplication alias Commanded.OpenTelemetry.EventHandler @@ -58,6 +59,11 @@ defmodule Commanded.OpenTelemetry do default: [], doc: "Aggregate tracing configuration. Use `:disabled` to disable." ], + aggregate_populate: [ + type: {:in, [:disabled, []]}, + default: [], + doc: "Aggregate populate tracing configuration. Use `:disabled` to disable." + ], application: [ type: {:in, [:disabled, []]}, default: [], @@ -105,6 +111,9 @@ defmodule Commanded.OpenTelemetry do # Disable event handler tracing Commanded.OpenTelemetry.setup(event_handler: :disabled) + # Disable aggregate populate tracing + Commanded.OpenTelemetry.setup(aggregate_populate: :disabled) + # Use parent-child relationships for event handlers Commanded.OpenTelemetry.setup(event_handler: [span_relationship: :child]) @@ -118,6 +127,11 @@ defmodule Commanded.OpenTelemetry do _config -> Aggregate.setup() end + case opts[:aggregate_populate] do + :disabled -> :ok + _config -> AggregatePopulate.setup() + end + case opts[:application] do :disabled -> :ok _config -> OTelApplication.setup() diff --git a/lib/commanded/opentelemetry/aggregate_populate.ex b/lib/commanded/opentelemetry/aggregate_populate.ex new file mode 100644 index 00000000..12710e1f --- /dev/null +++ b/lib/commanded/opentelemetry/aggregate_populate.ex @@ -0,0 +1,73 @@ +defmodule Commanded.OpenTelemetry.AggregatePopulate do + @moduledoc false + + alias Commanded.OpenTelemetry.CommandedAttributes + alias OpenTelemetry.SemConv.Incubating.CodeAttributes + alias OpenTelemetry.SemConv.Incubating.MessagingAttributes + alias OpenTelemetry.Span + + @tracer_id __MODULE__ + + def setup do + :ok = + :telemetry.attach_many( + {__MODULE__, :populate}, + [ + [:commanded, :aggregate, :populate, :start], + [:commanded, :aggregate, :populate, :stop] + ], + &__MODULE__.handle_telemetry_event/4, + %{} + ) + end + + def handle_telemetry_event( + [:commanded, :aggregate, :populate, :start], + _measurements, + meta, + _config + ) do + attributes = [ + # OTel Messaging SemConv + {MessagingAttributes.messaging_system(), "commanded"}, + {MessagingAttributes.messaging_operation_type(), :receive}, + {MessagingAttributes.messaging_operation_name(), "populate"}, + # OTel Code SemConv + {CodeAttributes.code_function(), "populate"}, + # Commanded-specific + {CommandedAttributes.commanded_application(), meta.application}, + {CommandedAttributes.commanded_aggregate_uuid(), meta.aggregate_uuid}, + {CommandedAttributes.commanded_aggregate_version(), meta.aggregate_version} + ] + + OpentelemetryTelemetry.start_telemetry_span( + @tracer_id, + "commanded.aggregate.populate", + meta, + %{ + kind: :internal, + attributes: attributes + } + ) + end + + def handle_telemetry_event( + [:commanded, :aggregate, :populate, :stop], + measurements, + meta, + _config + ) do + ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, meta) + + event_count = Map.get(measurements, :count, 0) + Span.set_attribute(ctx, CommandedAttributes.commanded_event_count(), event_count) + + Span.set_attribute( + ctx, + CommandedAttributes.commanded_aggregate_version(), + meta.aggregate_version + ) + + OpentelemetryTelemetry.end_telemetry_span(@tracer_id, meta) + end +end diff --git a/test/opentelemetry/aggregate_populate_test.exs b/test/opentelemetry/aggregate_populate_test.exs new file mode 100644 index 00000000..807b80f3 --- /dev/null +++ b/test/opentelemetry/aggregate_populate_test.exs @@ -0,0 +1,182 @@ +defmodule Commanded.OpenTelemetry.AggregatePopulateTest do + @moduledoc """ + Tests for AggregatePopulate OpenTelemetry instrumentation. + + Follows the same patterns as AggregateTest and EventHandlerTest. + """ + + use Commanded.OpenTelemetryCase, async: false + + alias Commanded.OpenTelemetry.AggregatePopulate + alias Commanded.TestSupport.Factory + alias Commanded.UUID + + setup do + detach_populate_handlers() + AggregatePopulate.setup() + + :ok + end + + describe "setup/0" do + test "attaches telemetry handlers for aggregate populate events" do + detach_populate_handlers() + + AggregatePopulate.setup() + + for event <- [ + [:commanded, :aggregate, :populate, :start], + [:commanded, :aggregate, :populate, :stop] + ] do + handlers = :telemetry.list_handlers(event) + + assert Enum.any?( + handlers, + &match?(%{id: {AggregatePopulate, :populate}}, &1) + ), + "Expected handler for event #{inspect(event)}" + end + end + + test "calling setup twice raises MatchError (fail fast)" do + detach_populate_handlers() + + :ok = AggregatePopulate.setup() + + handlers = :telemetry.list_handlers([:commanded, :aggregate, :populate, :start]) + assert length(handlers) == 1 + + assert_raise MatchError, fn -> + AggregatePopulate.setup() + end + end + end + + describe "attribute completeness" do + setup do + detach_populate_handlers() + AggregatePopulate.setup() + :ok + end + + test "includes ALL required span attributes" do + aggregate_uuid = UUID.uuid4() + + meta = + Factory.build_aggregate_populate_metadata( + aggregate_uuid: aggregate_uuid, + aggregate_version: 5 + ) + + :telemetry.span([:commanded, :aggregate, :populate], meta, fn -> + {:ok, meta} + end) + + assert_receive {:span, + span( + name: "commanded.aggregate.populate", + kind: :internal, + attributes: attributes + )}, + 1000 + + attrs = :otel_attributes.map(attributes) + + assert attrs == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "populate", + "code.function": "populate", + "commanded.application": MockApp, + "commanded.aggregate.uuid": aggregate_uuid, + "commanded.aggregate.version": 5, + "commanded.event.count": 0 + } + end + end + + describe "edge cases" do + setup do + detach_populate_handlers() + AggregatePopulate.setup() + :ok + end + + test "handles zero events populated" do + aggregate_uuid = UUID.uuid4() + + meta = + Factory.build_aggregate_populate_metadata( + aggregate_uuid: aggregate_uuid, + aggregate_version: 0 + ) + + :telemetry.execute([:commanded, :aggregate, :populate, :start], %{}, meta) + :telemetry.execute([:commanded, :aggregate, :populate, :stop], %{count: 0}, meta) + + assert_receive {:span, + span( + name: "commanded.aggregate.populate", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "populate", + "code.function": "populate", + "commanded.application": MockApp, + "commanded.aggregate.uuid": aggregate_uuid, + "commanded.aggregate.version": 0, + "commanded.event.count": 0 + } + end + + test "handles multiple events populated" do + aggregate_uuid = UUID.uuid4() + + meta = + Factory.build_aggregate_populate_metadata( + aggregate_uuid: aggregate_uuid, + aggregate_version: 0 + ) + + :telemetry.execute([:commanded, :aggregate, :populate, :start], %{}, meta) + + stop_meta = Map.put(meta, :aggregate_version, 10) + :telemetry.execute([:commanded, :aggregate, :populate, :stop], %{count: 10}, stop_meta) + + assert_receive {:span, + span( + name: "commanded.aggregate.populate", + kind: :internal, + attributes: attributes + )}, + 1000 + + assert :otel_attributes.map(attributes) == %{ + "messaging.system": "commanded", + "messaging.operation.type": :receive, + "messaging.operation.name": "populate", + "code.function": "populate", + "commanded.application": MockApp, + "commanded.aggregate.uuid": aggregate_uuid, + "commanded.aggregate.version": 10, + "commanded.event.count": 10 + } + end + end + + defp detach_populate_handlers do + for event <- [ + [:commanded, :aggregate, :populate, :start], + [:commanded, :aggregate, :populate, :stop] + ] 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 c8d89d94..fcef045e 100644 --- a/test/support/factory.ex +++ b/test/support/factory.ex @@ -596,6 +596,26 @@ defmodule Commanded.TestSupport.Factory do }) end + def build_aggregate_populate_metadata(opts \\ []) do + aggregate_uuid = Keyword.get(opts, :aggregate_uuid, UUID.uuid4()) + + defaults = [ + application: Keyword.get(opts, :application, MockApp), + aggregate_uuid: aggregate_uuid, + aggregate_state: Keyword.get(opts, :aggregate_state, %{}), + aggregate_version: Keyword.get(opts, :aggregate_version, 0) + ] + + opts = Keyword.merge(defaults, opts) + + %{ + application: Keyword.fetch!(opts, :application), + aggregate_uuid: Keyword.fetch!(opts, :aggregate_uuid), + aggregate_state: Keyword.fetch!(opts, :aggregate_state), + aggregate_version: Keyword.fetch!(opts, :aggregate_version) + } + end + def build_telemetry_event(event_type, opts \\ []) def build_telemetry_event(:start, opts) do diff --git a/test/support/opentelemetry_case.ex b/test/support/opentelemetry_case.ex index f8e3fa9d..172c267a 100644 --- a/test/support/opentelemetry_case.ex +++ b/test/support/opentelemetry_case.ex @@ -45,6 +45,8 @@ defmodule Commanded.OpenTelemetryCase do [:commanded, :aggregate, :execute, :start], [:commanded, :aggregate, :execute, :stop], [:commanded, :aggregate, :execute, :exception], + [:commanded, :aggregate, :populate, :start], + [:commanded, :aggregate, :populate, :stop], [:commanded, :application, :dispatch, :start], [:commanded, :application, :dispatch, :stop], [:commanded, :application, :dispatch, :exception]