refactor(otel): centralize tag keys, align with semconv, and unify ITraceable

src/Core/EcommerceDDD.Core.Infrastructure/GlobalUsings.cs

@@ -1,5 +1,6 @@
 global using Asp.Versioning;
 global using Confluent.Kafka;
+global using Confluent.Kafka.Admin;
 global using Duende.IdentityModel;
 global using Duende.IdentityModel.Client;
 global using EcommerceDDD.Core.CQRS.CommandHandling;
@@ -42,17 +43,17 @@
 global using Microsoft.OpenApi;
 global using Newtonsoft.Json;
 global using Newtonsoft.Json.Linq;
+global using Npgsql;
 global using OpenTelemetry.Logs;
 global using OpenTelemetry.Metrics;
 global using OpenTelemetry.Resources;
 global using OpenTelemetry.Trace;
 global using Polly;
 global using Swashbuckle.AspNetCore.SwaggerGen;
 global using System.Diagnostics;
-global using System.Linq.Expressions;
 global using System.IdentityModel.Tokens.Jwt;
+global using System.Linq.Expressions;
 global using System.Net.Mime;
 global using System.Reflection;
 global using System.Security.Claims;
 global using System.Text;
-global using Npgsql;

src/Core/EcommerceDDD.Core.Infrastructure/Kafka/Consumer/KafkaConsumer.cs

@@ -1,6 +1,4 @@
-using Confluent.Kafka.Admin;
-
-namespace EcommerceDDD.Core.Infrastructure.Kafka.Consumer;
+namespace EcommerceDDD.Core.Infrastructure.Kafka.Consumer;
 
 public class KafkaConsumer : IEventConsumer
 {
@@ -260,13 +258,13 @@ private async Task ConsumeNextMessage(IConsumer<string, string> consumer,
 				parentContext: default,
 				tags: null,
 				links: links);
-			activity?.SetTag("messaging.system", "kafka");
-			activity?.SetTag("messaging.destination", result.Topic);
-			activity?.SetTag("messaging.operation", "receive");
-			activity?.SetTag("messaging.kafka.consumer_group", _config.Group);
-			activity?.SetTag("messaging.kafka.partition", result.Partition.Value);
-			activity?.SetTag("messaging.kafka.message_offset", result.Offset.Value);
-			activity?.SetTag("messaging.message_id", result.Message.Key);
+			activity?.SetTag(MessagingAttributes.AttributeMessagingSystem, MessagingAttributes.MessagingSystemValues.Kafka);
+			activity?.SetTag(MessagingAttributes.AttributeMessagingDestinationName, result.Topic);
+			activity?.SetTag(MessagingAttributes.AttributeMessagingOperationType, MessagingAttributes.MessagingOperationTypeValues.Receive);
+			activity?.SetTag(MessagingAttributes.AttributeMessagingConsumerGroupName, _config.Group);
+			activity?.SetTag(MessagingAttributes.AttributeMessagingDestinationPartitionId, result.Partition.Value);
+			activity?.SetTag(MessagingAttributes.AttributeMessagingKafkaOffset, result.Offset.Value);
+			activity?.SetTag(MessagingAttributes.AttributeMessagingKafkaMessageKey, result.Message.Key);
 
 			// Deserialize the domain event from the raw outbox payload
 			var messageBytes = Encoding.UTF8.GetBytes(result.Message.Value);
@@ -312,7 +310,7 @@ await _eventPublisher
 	private async Task EnsureTopicsExistAsync()
 	{
 		if (_config is null)
-			throw new InvalidOperationException(nameof(Config));
+			throw new InvalidOperationException("Kafka consumer config is not initialized.");
 
 		try
 		{

src/Core/EcommerceDDD.Core.Infrastructure/Marten/MartenRepository.cs

@@ -78,10 +78,10 @@ public void AppendToOutbox(INotification @event)
             $"outbox.publish {eventName}",
             ActivityKind.Producer);
 
-        _pendingOutboxActivity?.SetTag("messaging.system", "kafka");
-        _pendingOutboxActivity?.SetTag("messaging.destination.name", "outbox");
-        _pendingOutboxActivity?.SetTag("messaging.operation", "publish");
-        _pendingOutboxActivity?.SetTag("event.type", eventName);
+        _pendingOutboxActivity?.SetTag(MessagingAttributes.AttributeMessagingSystem, MessagingAttributes.MessagingSystemValues.Kafka);
+        _pendingOutboxActivity?.SetTag(MessagingAttributes.AttributeMessagingDestinationName, TelemetryTags.DestinationValues.Outbox);
+        _pendingOutboxActivity?.SetTag(MessagingAttributes.AttributeMessagingOperationType, MessagingAttributes.MessagingOperationTypeValues.Send);
+        _pendingOutboxActivity?.SetTag(TelemetryTags.EventType, eventName);
 
         var integrationEvent = IntegrationEvent.FromNotification(@event!);
 

src/Core/EcommerceDDD.Core.Infrastructure/OpenTelemetry/ITraceable.cs

@@ -2,5 +2,5 @@ namespace EcommerceDDD.Core.Infrastructure.OpenTelemetry;
 
 public interface ITraceable
 {
-    IEnumerable<KeyValuePair<string, string>> GetSpanTags();
+    IEnumerable<KeyValuePair<string, object>> GetSpanTags();
 }

src/Core/EcommerceDDD.Core.Infrastructure/OpenTelemetry/MessagingAttributes.cs

@@ -0,0 +1,49 @@
+// Local subset of OpenTelemetry messaging semantic conventions.
+// Todo: replace constant with the official `OpenTelemetry.SemanticConventions` package is released.
+// Source: https://github.com/open-telemetry/opentelemetry-dotnet-contrib/blob/main/src/OpenTelemetry.SemanticConventions/Attributes/MessagingAttributes.cs
+
+namespace EcommerceDDD.Core.Infrastructure.OpenTelemetry;
+
+public static class MessagingAttributes
+{
+	/// <summary>An identifier for the messaging system being used. See <c>MessagingSystemValues</c>.</summary>
+	public const string AttributeMessagingSystem = "messaging.system";
+
+	/// <summary>The message destination name (queue, topic, or broker-specific entity).</summary>
+	public const string AttributeMessagingDestinationName = "messaging.destination.name";
+
+	/// <summary>The partition identifier within <c>messaging.destination.name</c>.</summary>
+	public const string AttributeMessagingDestinationPartitionId = "messaging.destination.partition.id";
+
+	/// <summary>The type of the messaging operation. See <c>MessagingOperationTypeValues</c>.</summary>
+	public const string AttributeMessagingOperationType = "messaging.operation.type";
+
+	/// <summary>The name of the consumer group with which a consumer is associated.</summary>
+	public const string AttributeMessagingConsumerGroupName = "messaging.consumer.group.name";
+
+	/// <summary>The offset of a record in the corresponding Kafka partition.</summary>
+	public const string AttributeMessagingKafkaOffset = "messaging.kafka.offset";
+
+	/// <summary>The Kafka message key (partition routing key; not a unique message id).</summary>
+	public const string AttributeMessagingKafkaMessageKey = "messaging.kafka.message.key";
+
+	/// <summary>A value used by the messaging system as an identifier for the message.</summary>
+	public const string AttributeMessagingMessageId = "messaging.message.id";
+
+	public static class MessagingSystemValues
+	{
+		public const string Kafka = "kafka";
+	}
+
+	public static class MessagingOperationTypeValues
+	{
+		/// <summary>One or more messages are provided for sending to an intermediary.</summary>
+		public const string Send = "send";
+
+		/// <summary>One or more messages are requested by a consumer (pull-based).</summary>
+		public const string Receive = "receive";
+
+		/// <summary>One or more messages are processed by a consumer.</summary>
+		public const string Process = "process";
+	}
+}

src/Core/EcommerceDDD.Core.Infrastructure/OpenTelemetry/TelemetryTags.cs

@@ -0,0 +1,17 @@
+namespace EcommerceDDD.Core.Infrastructure.OpenTelemetry;
+
+/// <summary>
+/// Project-specific telemetry tag keys and literal values that are not part of
+/// the OpenTelemetry semantic conventions spec. Keys defined by the spec live
+/// in <see cref="MessagingAttributes"/> (and future domain attribute files).
+/// </summary>
+public static class TelemetryTags
+{
+	public const string OrderId   = "order.id";
+	public const string EventType = "event.type";
+
+	public static class DestinationValues
+	{
+		public const string Outbox = "outbox";
+	}
+}

src/Services/EcommerceDDD.OrderProcessing/Application/Orders/CancelingOrder/CancelOrder.cs

@@ -15,8 +15,8 @@ public static CancelOrder Create(
 		return new CancelOrder(orderId, CancellationReason);
 	}
 
-	public IEnumerable<KeyValuePair<string, string>> GetSpanTags() =>
-		[new("order.id", OrderId.Value.ToString())];
+	public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+		[new(TelemetryTags.OrderId, OrderId.Value)];
 
 	private CancelOrder(
 		OrderId orderId,

src/Services/EcommerceDDD.OrderProcessing/Application/Orders/ConfirmingDelivery/ConfirmDelivery.cs

@@ -12,8 +12,8 @@ public static ConfirmDelivery Create(OrderId orderId)
 		return new ConfirmDelivery(orderId);
 	}
 
-	public IEnumerable<KeyValuePair<string, string>> GetSpanTags() =>
-		[new("order.id", OrderId.Value.ToString())];
+	public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+		[new(TelemetryTags.OrderId, OrderId.Value)];
 
 	private ConfirmDelivery(OrderId orderId)
 	{

src/Services/EcommerceDDD.OrderProcessing/Application/Orders/PlacingOrder/PlaceOrderHandler.cs

@@ -43,7 +43,7 @@ public async Task<Result> HandleAsync(PlaceOrder command, CancellationToken canc
 			orderItems);
 
 		var order = Order.Place(orderData);
-		Activity.Current?.SetTag("order.id", order.Id.Value.ToString());
+		Activity.Current?.SetTag(TelemetryTags.OrderId, order.Id.Value);
 
 		var orderPlacedEvent = order.GetUncommittedEvents()
 			.OfType<OrderPlaced>().FirstOrDefault();

src/Services/EcommerceDDD.OrderProcessing/Application/Orders/ProcessingOrder/ProcessOrder.cs

@@ -21,8 +21,8 @@ public static ProcessOrder Create(
 		return new ProcessOrder(customerId, orderId, quoteId);
 	}
 
-	public IEnumerable<KeyValuePair<string, string>> GetSpanTags() =>
-		[new("order.id", OrderId.Value.ToString())];
+	public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+		[new(TelemetryTags.OrderId, OrderId.Value)];
 
 	private ProcessOrder(
 		CustomerId customerId,

src/Services/EcommerceDDD.OrderProcessing/Application/Payments/RecordingPayment/RecordPayment.cs

@@ -21,8 +21,8 @@ public static RecordPayment Create(
 		return new RecordPayment(orderId, paymentId, totalPaid);
 	}
 
-	public IEnumerable<KeyValuePair<string, string>> GetSpanTags() =>
-		[new("order.id", OrderId.Value.ToString())];
+	public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+		[new(TelemetryTags.OrderId, OrderId.Value)];
 
 	private RecordPayment(
 		OrderId orderId,

src/Services/EcommerceDDD.OrderProcessing/Application/Payments/RequestingPayment/RequestPayment.cs

@@ -25,8 +25,8 @@ public static RequestPayment Create(
         return new RequestPayment(customerId, orderId, totalPrice, currency);
     }
 
-    public IEnumerable<KeyValuePair<string, string>> GetSpanTags() =>
-        [new("order.id", OrderId.Value.ToString())];
+    public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+        [new(TelemetryTags.OrderId, OrderId.Value)];
 
     private RequestPayment(
         CustomerId customerId, 

src/Services/EcommerceDDD.OrderProcessing/Application/Shipments/RecordingShipment/RecordShipment.cs

@@ -17,8 +17,8 @@ public static RecordShipment Create(
 		return new RecordShipment(orderId, shipmentId);
 	}
 
-	public IEnumerable<KeyValuePair<string, string>> GetSpanTags() =>
-		[new("order.id", OrderId.Value.ToString())];
+	public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+		[new(TelemetryTags.OrderId, OrderId.Value)];
 
 	private RecordShipment(
 		OrderId orderId,

src/Services/EcommerceDDD.OrderProcessing/Application/Shipments/RequestingShipment/RequestShipment.cs

@@ -12,8 +12,8 @@ public static RequestShipment Create(OrderId orderId)
         return new RequestShipment(orderId);
     }
 
-    public IEnumerable<KeyValuePair<string, string>> GetSpanTags() =>
-        [new("order.id", OrderId.Value.ToString())];
+    public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+        [new(TelemetryTags.OrderId, OrderId.Value)];
 
     private RequestShipment(OrderId orderId) => OrderId = orderId;
 }

src/Services/EcommerceDDD.PaymentProcessing.Tests/Application/ProcessPaymentHandlerTests.cs

@@ -28,7 +28,7 @@ public async Task ProcessPayment_WithCommand_ShouldCompletePayment()
 			.Returns(Task.FromResult(true));
 
 		// When
-		var processPayment = ProcessPayment.Create(payment.Id);
+		var processPayment = ProcessPayment.Create(payment.Id, orderId);
 		var processPaymentHandler = new ProcessPaymentHandler(
 			_productInventoryHandler, _customerCreditChecker, paymentWriteRepository);
 		await processPaymentHandler.HandleAsync(processPayment, CancellationToken.None);
@@ -68,7 +68,7 @@ public async Task ProcessPayment_WithoutEnoughCredit_CancelPayment()
 			.CheckProductsInStockAsync(Arg.Any<IReadOnlyList<ProductItem>>(), CancellationToken.None)
 			.Returns(Task.FromResult(true));
 
-		var processPayment = ProcessPayment.Create(payment.Id);
+		var processPayment = ProcessPayment.Create(payment.Id, orderId);
 		var processPaymentHandler = new ProcessPaymentHandler(
 			_productInventoryHandler, _customerCreditChecker, paymentWriteRepository);
 

src/Services/EcommerceDDD.PaymentProcessing/Application/ProcessingPayment/ProcessPayment.cs

@@ -1,19 +1,26 @@
-namespace EcommerceDDD.PaymentProcessing.Application.ProcessingPayment;
+namespace EcommerceDDD.PaymentProcessing.Application.ProcessingPayment;
 
-public record class ProcessPayment : ICommand
+public record class ProcessPayment : ICommand, ITraceable
 {
 	public PaymentId PaymentId { get; private set; }
+	public OrderId OrderId { get; private set; }
 
-	public static ProcessPayment Create(PaymentId paymentId)
+	public static ProcessPayment Create(PaymentId paymentId, OrderId orderId)
 	{
 		if (paymentId is null)
 			throw new ArgumentNullException(nameof(paymentId));
+		if (orderId is null)
+			throw new ArgumentNullException(nameof(orderId));
 
-		return new ProcessPayment(paymentId);
+		return new ProcessPayment(paymentId, orderId);
 	}
 
-	private ProcessPayment(PaymentId paymentId)
+	public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+		[new(TelemetryTags.OrderId, OrderId.Value)];
+
+	private ProcessPayment(PaymentId paymentId, OrderId orderId)
 	{
 		PaymentId = paymentId;
+		OrderId = orderId;
 	}
-}
+}

src/Services/EcommerceDDD.PaymentProcessing/Application/ProcessingPayment/ProcessPaymentHandler.cs

@@ -18,7 +18,6 @@ public async Task<Result> HandleAsync(ProcessPayment command, CancellationToken
 		if (payment is null)
 			return Result.Fail($"Payment {command.PaymentId} was not found.");
 
-		Activity.Current?.SetTag("order.id", payment.OrderId.Value.ToString());
 		try
 		{
 			if (!await _creditChecker

src/Services/EcommerceDDD.PaymentProcessing/Application/RequestingPayment/RequestPayment.cs

@@ -29,8 +29,8 @@ public static RequestPayment Create(
 		return new RequestPayment(customerId, orderId, totalPrice, currency, productItems);
 	}
 
-	public IEnumerable<KeyValuePair<string, string>> GetSpanTags() =>
-		[new("order.id", OrderId.Value.ToString())];
+	public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+		[new(TelemetryTags.OrderId, OrderId.Value)];
 
 	private RequestPayment(
 		CustomerId customerId,

src/Services/EcommerceDDD.PaymentProcessing/Application/RequestingPayment/RequestPaymentHandler.cs

@@ -22,6 +22,6 @@ await _paymentWriteRepository
 			.AppendEventsAsync(payment, cancellationToken);
 
         return await _commandBus.SendAsync(
-            ProcessPayment.Create(payment.Id), cancellationToken);
+            ProcessPayment.Create(payment.Id, command.OrderId), cancellationToken);
     }
 }

src/Services/EcommerceDDD.ShipmentProcessing.Tests/Application/ShipPackageHandlerTests.cs

@@ -21,7 +21,7 @@ public async Task ShipPackage_WithCommand_ShouldShipPackage()
         var shipment = shipmentWriteRepository.AggregateStream.First().Aggregate;
         Assert.NotNull(shipment);
 
-        var shipPackage = ProcessShipment.Create(shipment.Id);
+        var shipPackage = ProcessShipment.Create(shipment.Id, orderId);
         var shipPackageHandler = new ProcessShipmentHandler(shipmentWriteRepository);
 
         // When

src/Services/EcommerceDDD.ShipmentProcessing/Application/ProcessingShipment/ProcessShipment.cs

@@ -1,19 +1,26 @@
-namespace EcommerceDDD.ShipmentProcessing.Application.ProcessingShipment;
+namespace EcommerceDDD.ShipmentProcessing.Application.ProcessingShipment;
 
-public record class ProcessShipment : ICommand
+public record class ProcessShipment : ICommand, ITraceable
 {
 	public ShipmentId ShipmentId { get; private set; }
+	public OrderId OrderId { get; private set; }
 
-	public static ProcessShipment Create(ShipmentId shipmentId)
+	public static ProcessShipment Create(ShipmentId shipmentId, OrderId orderId)
 	{
 		if (shipmentId is null)
 			throw new ArgumentNullException(nameof(shipmentId));
+		if (orderId is null)
+			throw new ArgumentNullException(nameof(orderId));
 
-		return new ProcessShipment(shipmentId);
+		return new ProcessShipment(shipmentId, orderId);
 	}
 
-	private ProcessShipment(ShipmentId shipmentId)
+	public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+		[new(TelemetryTags.OrderId, OrderId.Value)];
+
+	private ProcessShipment(ShipmentId shipmentId, OrderId orderId)
 	{
 		ShipmentId = shipmentId;
+		OrderId = orderId;
 	}
-}
+}

src/Services/EcommerceDDD.ShipmentProcessing/Application/ProcessingShipment/ProcessShipmentHandler.cs

@@ -16,7 +16,6 @@ public async Task<Result> HandleAsync(ProcessShipment command, CancellationToken
 		if (shipment is null)
 			return Result.Fail($"The shipment {command.ShipmentId} was not found.");
 
-		Activity.Current?.SetTag("order.id", shipment.OrderId.Value.ToString());
 		try
 		{
 			shipment.Complete();

src/Services/EcommerceDDD.ShipmentProcessing/Application/RequestingShipment/RequestShipment.cs

@@ -17,8 +17,8 @@ public static RequestShipment Create(
 		return new RequestShipment(orderId, productItems);
 	}
 
-	public IEnumerable<KeyValuePair<string, string>> GetSpanTags() =>
-		[new("order.id", OrderId.Value.ToString())];
+	public IEnumerable<KeyValuePair<string, object>> GetSpanTags() =>
+		[new(TelemetryTags.OrderId, OrderId.Value)];
 
 	private RequestShipment(
 		OrderId orderId,

src/Services/EcommerceDDD.ShipmentProcessing/Application/RequestingShipment/RequestShipmentHandler.cs

@@ -19,6 +19,6 @@ await _shipmentWriteRepository
 			.AppendEventsAsync(shipment, cancellationToken);
 
         return await _commandBus
-			.SendAsync(ProcessShipment.Create(shipment.Id), cancellationToken);
+			.SendAsync(ProcessShipment.Create(shipment.Id, command.OrderId), cancellationToken);
     }
 }