feat: Add OpenTelemetry instrumentation for aggregate populate

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()

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

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

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

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]