From 295e911ad59229126d1a2c95736c018405277d5d Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sun, 5 Apr 2026 14:01:16 -0400 Subject: [PATCH] feat(telemetry): add aggregate_module to load/populate/snapshot telemetry metadata Enables OTel span names to follow the {operation} {destination} convention and surfaces messaging.destination.name + code.namespace attributes. Signed-off-by: Yordis Prieto --- lib/commanded/aggregates/aggregate.ex | 5 +++++ .../aggregates/aggregate_state_builder.ex | 6 ++++++ .../opentelemetry/aggregate_populate.ex | 13 +++++++++++-- .../opentelemetry/aggregate_snapshot.ex | 6 +++++- test/opentelemetry/aggregate_populate_test.exs | 18 +++++++++++++----- test/opentelemetry/aggregate_snapshot_test.exs | 8 +++++--- test/support/factory.ex | 4 ++++ 7 files changed, 49 insertions(+), 11 deletions(-) diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex index ad0ed416..ccfe9010 100644 --- a/lib/commanded/aggregates/aggregate.ex +++ b/lib/commanded/aggregates/aggregate.ex @@ -71,6 +71,7 @@ defmodule Commanded.Aggregates.Aggregate do measurements: "%{system_time: integer()}", metadata: """ %{application: Commanded.Application.t(), + aggregate_module: module(), aggregate_uuid: String.t(), aggregate_version: non_neg_integer(), snapshot_every: non_neg_integer() | nil, @@ -84,6 +85,7 @@ defmodule Commanded.Aggregates.Aggregate do measurements: "%{duration: non_neg_integer()}", metadata: """ %{application: Commanded.Application.t(), + aggregate_module: module(), aggregate_uuid: String.t(), aggregate_version: non_neg_integer(), snapshot_every: non_neg_integer() | nil, @@ -98,6 +100,7 @@ defmodule Commanded.Aggregates.Aggregate do measurements: "%{duration: non_neg_integer()}", metadata: """ %{application: Commanded.Application.t(), + aggregate_module: module(), aggregate_uuid: String.t(), aggregate_version: non_neg_integer(), snapshot_every: non_neg_integer() | nil, @@ -680,6 +683,7 @@ defmodule Commanded.Aggregates.Aggregate do defp do_take_snapshot(%Aggregate{} = state) do %Aggregate{ application: application, + aggregate_module: aggregate_module, aggregate_uuid: aggregate_uuid, aggregate_state: aggregate_state, aggregate_version: aggregate_version, @@ -691,6 +695,7 @@ defmodule Commanded.Aggregates.Aggregate do meta = %{ application: application, + aggregate_module: aggregate_module, aggregate_uuid: aggregate_uuid, aggregate_version: aggregate_version, snapshot_every: snapshot_every, diff --git a/lib/commanded/aggregates/aggregate_state_builder.ex b/lib/commanded/aggregates/aggregate_state_builder.ex index a9fd35f6..a267abc3 100644 --- a/lib/commanded/aggregates/aggregate_state_builder.ex +++ b/lib/commanded/aggregates/aggregate_state_builder.ex @@ -14,6 +14,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do measurements: "%{system_time: integer()}", metadata: """ %{application: Commanded.Application.t(), + aggregate_module: module(), aggregate_uuid: String.t(), aggregate_state: struct(), aggregate_version: non_neg_integer()} @@ -26,6 +27,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do measurements: "%{duration: non_neg_integer(), count: non_neg_integer()}", metadata: """ %{application: Commanded.Application.t(), + aggregate_module: module(), aggregate_uuid: String.t(), aggregate_state: struct(), aggregate_version: non_neg_integer(), @@ -40,6 +42,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do measurements: "%{system_time: integer()}", metadata: """ %{application: Commanded.Application.t(), + aggregate_module: module(), aggregate_uuid: String.t(), aggregate_state: struct(), aggregate_version: non_neg_integer()} @@ -52,6 +55,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do measurements: "%{duration: non_neg_integer(), count: non_neg_integer()}", metadata: """ %{application: Commanded.Application.t(), + aggregate_module: module(), aggregate_uuid: String.t(), aggregate_state: struct(), aggregate_version: non_neg_integer()} @@ -189,6 +193,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do defp telemetry_metadata(%Aggregate{} = state) do %Aggregate{ application: application, + aggregate_module: aggregate_module, aggregate_uuid: aggregate_uuid, aggregate_state: aggregate_state, aggregate_version: aggregate_version @@ -196,6 +201,7 @@ defmodule Commanded.Aggregates.AggregateStateBuilder do %{ application: application, + aggregate_module: aggregate_module, aggregate_uuid: aggregate_uuid, aggregate_state: aggregate_state, aggregate_version: aggregate_version diff --git a/lib/commanded/opentelemetry/aggregate_populate.ex b/lib/commanded/opentelemetry/aggregate_populate.ex index aea354d1..37fab779 100644 --- a/lib/commanded/opentelemetry/aggregate_populate.ex +++ b/lib/commanded/opentelemetry/aggregate_populate.ex @@ -2,6 +2,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do @moduledoc false alias Commanded.OpenTelemetry.CommandedAttributes + alias Commanded.OpenTelemetry.Helpers alias OpenTelemetry.SemConv.Incubating.CodeAttributes alias OpenTelemetry.SemConv.Incubating.MessagingAttributes alias OpenTelemetry.Span @@ -29,11 +30,15 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do meta, _config ) do + aggregate_module_name = Helpers.module_name(meta.aggregate_module) + attributes = [ {MessagingAttributes.messaging_system(), "commanded"}, {MessagingAttributes.messaging_operation_type(), :receive}, {MessagingAttributes.messaging_operation_name(), "load"}, + {MessagingAttributes.messaging_destination_name(), aggregate_module_name}, {CodeAttributes.code_function(), "load"}, + {CodeAttributes.code_namespace(), aggregate_module_name}, {CommandedAttributes.commanded_application(), meta.application}, {CommandedAttributes.commanded_aggregate_uuid(), meta.aggregate_uuid}, {CommandedAttributes.commanded_aggregate_version(), meta.aggregate_version} @@ -41,7 +46,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do OpentelemetryTelemetry.start_telemetry_span( @tracer_id, - "commanded.aggregate.load", + "load #{aggregate_module_name}", meta, %{ kind: :internal, @@ -90,13 +95,17 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do meta, _config ) do + aggregate_module_name = Helpers.module_name(meta.aggregate_module) + attributes = [ # OTel Messaging SemConv {MessagingAttributes.messaging_system(), "commanded"}, {MessagingAttributes.messaging_operation_type(), :receive}, {MessagingAttributes.messaging_operation_name(), "populate"}, + {MessagingAttributes.messaging_destination_name(), aggregate_module_name}, # OTel Code SemConv {CodeAttributes.code_function(), "populate"}, + {CodeAttributes.code_namespace(), aggregate_module_name}, # Commanded-specific {CommandedAttributes.commanded_application(), meta.application}, {CommandedAttributes.commanded_aggregate_uuid(), meta.aggregate_uuid}, @@ -105,7 +114,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulate do OpentelemetryTelemetry.start_telemetry_span( @tracer_id, - "commanded.aggregate.populate", + "populate #{aggregate_module_name}", meta, %{ kind: :internal, diff --git a/lib/commanded/opentelemetry/aggregate_snapshot.ex b/lib/commanded/opentelemetry/aggregate_snapshot.ex index 0dad5e4b..173da057 100644 --- a/lib/commanded/opentelemetry/aggregate_snapshot.ex +++ b/lib/commanded/opentelemetry/aggregate_snapshot.ex @@ -30,12 +30,16 @@ defmodule Commanded.OpenTelemetry.AggregateSnapshot do meta, _config ) do + aggregate_module_name = Helpers.module_name(meta.aggregate_module) + attributes = [ {MessagingAttributes.messaging_system(), "commanded"}, {MessagingAttributes.messaging_operation_type(), :publish}, {MessagingAttributes.messaging_operation_name(), "snapshot"}, + {MessagingAttributes.messaging_destination_name(), aggregate_module_name}, {CodeAttributes.code_function(), "snapshot"}, + {CodeAttributes.code_namespace(), aggregate_module_name}, {CommandedAttributes.commanded_application(), meta.application}, {CommandedAttributes.commanded_aggregate_uuid(), meta.aggregate_uuid}, {CommandedAttributes.commanded_aggregate_version(), meta.aggregate_version} @@ -45,7 +49,7 @@ defmodule Commanded.OpenTelemetry.AggregateSnapshot do OpentelemetryTelemetry.start_telemetry_span( @tracer_id, - "commanded.aggregate.snapshot", + "snapshot #{aggregate_module_name}", meta, %{ kind: :internal, diff --git a/test/opentelemetry/aggregate_populate_test.exs b/test/opentelemetry/aggregate_populate_test.exs index cda77349..de9453f6 100644 --- a/test/opentelemetry/aggregate_populate_test.exs +++ b/test/opentelemetry/aggregate_populate_test.exs @@ -73,7 +73,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do assert_receive {:span, span( - name: "commanded.aggregate.populate", + name: "populate MockAggregate", kind: :internal, attributes: attributes )}, @@ -85,7 +85,9 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do "messaging.system": "commanded", "messaging.operation.type": :receive, "messaging.operation.name": "populate", + "messaging.destination.name": "MockAggregate", "code.function": "populate", + "code.namespace": "MockAggregate", "commanded.application": MockApp, "commanded.aggregate.uuid": aggregate_uuid, "commanded.aggregate.version": 5, @@ -123,7 +125,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do assert_receive {:span, span( - name: "commanded.aggregate.load", + name: "load MockAggregate", kind: :internal, attributes: attributes )}, @@ -133,7 +135,9 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do "messaging.system": "commanded", "messaging.operation.type": :receive, "messaging.operation.name": "load", + "messaging.destination.name": "MockAggregate", "code.function": "load", + "code.namespace": "MockAggregate", "commanded.application": MockApp, "commanded.aggregate.uuid": aggregate_uuid, "commanded.aggregate.version": 0, @@ -164,7 +168,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do assert_receive {:span, span( - name: "commanded.aggregate.load", + name: "load MockAggregate", kind: :internal, attributes: attributes )}, @@ -198,7 +202,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do assert_receive {:span, span( - name: "commanded.aggregate.populate", + name: "populate MockAggregate", kind: :internal, attributes: attributes )}, @@ -208,7 +212,9 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do "messaging.system": "commanded", "messaging.operation.type": :receive, "messaging.operation.name": "populate", + "messaging.destination.name": "MockAggregate", "code.function": "populate", + "code.namespace": "MockAggregate", "commanded.application": MockApp, "commanded.aggregate.uuid": aggregate_uuid, "commanded.aggregate.version": 0, @@ -232,7 +238,7 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do assert_receive {:span, span( - name: "commanded.aggregate.populate", + name: "populate MockAggregate", kind: :internal, attributes: attributes )}, @@ -242,7 +248,9 @@ defmodule Commanded.OpenTelemetry.AggregatePopulateTest do "messaging.system": "commanded", "messaging.operation.type": :receive, "messaging.operation.name": "populate", + "messaging.destination.name": "MockAggregate", "code.function": "populate", + "code.namespace": "MockAggregate", "commanded.application": MockApp, "commanded.aggregate.uuid": aggregate_uuid, "commanded.aggregate.version": 10, diff --git a/test/opentelemetry/aggregate_snapshot_test.exs b/test/opentelemetry/aggregate_snapshot_test.exs index eeb410b9..1032b523 100644 --- a/test/opentelemetry/aggregate_snapshot_test.exs +++ b/test/opentelemetry/aggregate_snapshot_test.exs @@ -72,7 +72,7 @@ defmodule Commanded.OpenTelemetry.AggregateSnapshotTest do assert_receive {:span, span( - name: "commanded.aggregate.snapshot", + name: "snapshot MockAggregate", kind: :internal, attributes: attributes )}, @@ -82,7 +82,9 @@ defmodule Commanded.OpenTelemetry.AggregateSnapshotTest do "messaging.system": "commanded", "messaging.operation.type": :publish, "messaging.operation.name": "snapshot", + "messaging.destination.name": "MockAggregate", "code.function": "snapshot", + "code.namespace": "MockAggregate", "commanded.application": MockApp, "commanded.aggregate.uuid": aggregate_uuid, "commanded.aggregate.version": 10, @@ -108,7 +110,7 @@ defmodule Commanded.OpenTelemetry.AggregateSnapshotTest do assert_receive {:span, span( - name: "commanded.aggregate.snapshot", + name: "snapshot MockAggregate", kind: :internal, attributes: attributes )}, @@ -143,7 +145,7 @@ defmodule Commanded.OpenTelemetry.AggregateSnapshotTest do assert_receive {:span, span( - name: "commanded.aggregate.snapshot", + name: "snapshot MockAggregate", status: {:status, :error, _error_message}, attributes: span_attrs )}, diff --git a/test/support/factory.ex b/test/support/factory.ex index 749e9cf1..70c92507 100644 --- a/test/support/factory.ex +++ b/test/support/factory.ex @@ -601,6 +601,7 @@ defmodule Commanded.TestSupport.Factory do defaults = [ application: Keyword.get(opts, :application, MockApp), + aggregate_module: Keyword.get(opts, :aggregate_module, MockAggregate), aggregate_uuid: aggregate_uuid, aggregate_state: Keyword.get(opts, :aggregate_state, %{}), aggregate_version: Keyword.get(opts, :aggregate_version, 0) @@ -610,6 +611,7 @@ defmodule Commanded.TestSupport.Factory do %{ application: Keyword.fetch!(opts, :application), + aggregate_module: Keyword.fetch!(opts, :aggregate_module), aggregate_uuid: Keyword.fetch!(opts, :aggregate_uuid), aggregate_state: Keyword.fetch!(opts, :aggregate_state), aggregate_version: Keyword.fetch!(opts, :aggregate_version) @@ -630,6 +632,7 @@ defmodule Commanded.TestSupport.Factory do defaults = [ application: Keyword.get(opts, :application, MockApp), + aggregate_module: Keyword.get(opts, :aggregate_module, MockAggregate), aggregate_uuid: aggregate_uuid, aggregate_version: Keyword.get(opts, :aggregate_version, 0), snapshot_every: Keyword.get(opts, :snapshot_every), @@ -640,6 +643,7 @@ defmodule Commanded.TestSupport.Factory do %{ application: Keyword.fetch!(opts, :application), + aggregate_module: Keyword.fetch!(opts, :aggregate_module), aggregate_uuid: Keyword.fetch!(opts, :aggregate_uuid), aggregate_version: Keyword.fetch!(opts, :aggregate_version), snapshot_every: Keyword.get(opts, :snapshot_every),