feat: upgrade exception events for OpenTelemetry

src/KafkaFlow.IntegrationTests/OpenTelemetryTests.cs

@@ -16,6 +16,7 @@
     using KafkaFlow.IntegrationTests.Core.Handlers;
     using KafkaFlow.IntegrationTests.Core.Middlewares;
     using KafkaFlow.IntegrationTests.Core.Producers;
+    using KafkaFlow.OpenTelemetry;
     using Microsoft.Extensions.Configuration;
     using Microsoft.Extensions.DependencyInjection;
     using Microsoft.Extensions.Hosting;
@@ -45,7 +46,7 @@ public async Task AddOpenTelemetry_ProducingAndConsumingOneMessage_TraceAndSpans
             MessageStorage.Clear();
 
             using var tracerProvider = Sdk.CreateTracerProviderBuilder()
-            .AddSource("KafkaFlow.OpenTelemetry")
+            .AddSource(KafkaFlowInstrumentation.ActivitySourceName)
             .AddInMemoryExporter(this.exportedItems)
             .Build();
 
@@ -78,7 +79,7 @@ public async Task AddOpenTelemetry_ProducingAndConsumingOneMessage_BaggageIsProp
             var baggageValue2 = "TestBaggageValue2";
 
             using var tracerProvider = Sdk.CreateTracerProviderBuilder()
-            .AddSource("KafkaFlow.OpenTelemetry")
+            .AddSource(KafkaFlowInstrumentation.ActivitySourceName)
             .AddSource(kafkaFlowTestString)
             .AddInMemoryExporter(this.exportedItems)
             .Build();

src/KafkaFlow.OpenTelemetry/ActivitySourceAccessor.cs

@@ -2,40 +2,22 @@
 
 namespace KafkaFlow.OpenTelemetry
 {
-    using System;
-    using System.Collections.Generic;
     using System.Diagnostics;
-    using System.Reflection;
     using Conventions = SemanticConventions::OpenTelemetry.Trace.TraceSemanticConventions;
 
     internal static class ActivitySourceAccessor
     {
         internal const string ActivityString = "otel_activity";
-        internal const string ExceptionEventKey = "exception";
         internal const string MessagingSystemId = "kafka";
         internal const string AttributeMessagingOperation = "messaging.operation";
         internal const string AttributeMessagingKafkaMessageKey = "messaging.kafka.message.key";
         internal const string AttributeMessagingKafkaMessageOffset = "messaging.kafka.message.offset";
-        internal static readonly AssemblyName AssemblyName = typeof(ActivitySourceAccessor).Assembly.GetName();
-        internal static readonly string ActivitySourceName = AssemblyName.Name;
-        internal static readonly string Version = Assembly.GetExecutingAssembly().GetName().Version.ToString();
-        internal static readonly ActivitySource ActivitySource = new(ActivitySourceName, Version);
 
-        public static void SetGenericTags(Activity activity)
-        {
-            activity?.SetTag(Conventions.AttributeMessagingSystem, MessagingSystemId);
-        }
+        internal static readonly ActivitySource ActivitySource = new(KafkaFlowInstrumentation.ActivitySourceName, KafkaFlowInstrumentation.Version);
 
-        public static ActivityEvent CreateExceptionEvent(Exception exception)
+        internal static void SetGenericTags(Activity activity)
         {
-            var activityTagCollection = new ActivityTagsCollection(
-                new[]
-                {
-                    new KeyValuePair<string, object>(Conventions.AttributeExceptionMessage, exception.Message),
-                    new KeyValuePair<string, object>(Conventions.AttributeExceptionStacktrace, exception.StackTrace),
-                });
-
-            return new ActivityEvent(ExceptionEventKey, DateTimeOffset.UtcNow, activityTagCollection);
+            activity?.SetTag(Conventions.AttributeMessagingSystem, MessagingSystemId);
         }
     }
 }

src/KafkaFlow.OpenTelemetry/KafkaFlowInstrumentation.cs

@@ -0,0 +1,19 @@
+namespace KafkaFlow.OpenTelemetry
+{
+    using System.Reflection;
+
+    /// <summary>
+    /// KafkaFlow OTEL instrumentation properties
+    /// </summary>
+    public static class KafkaFlowInstrumentation
+    {
+        internal static readonly AssemblyName AssemblyName = typeof(KafkaFlowInstrumentation).Assembly.GetName();
+        internal static readonly string Version = AssemblyName.Version.ToString();
+
+        /// <summary>
+        /// ActivitySource name to be used when adding
+        /// KafkaFlow as source to an OTEL listener
+        /// </summary>
+        public static readonly string ActivitySourceName = AssemblyName.Name;
+    }
+}

src/KafkaFlow.OpenTelemetry/OpenTelemetryConsumerEventsHandler.cs

@@ -7,6 +7,7 @@
     using System.Threading.Tasks;
     using global::OpenTelemetry;
     using global::OpenTelemetry.Context.Propagation;
+    using global::OpenTelemetry.Trace;
 
     internal static class OpenTelemetryConsumerEventsHandler
     {
@@ -67,9 +68,8 @@ public static Task OnConsumeError(IMessageContext context, Exception ex)
         {
             if (context.Items.TryGetValue(ActivitySourceAccessor.ActivityString, out var value) && value is Activity activity)
             {
-                var exceptionEvent = ActivitySourceAccessor.CreateExceptionEvent(ex);
-
-                activity?.AddEvent(exceptionEvent);
+                activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
+                activity?.RecordException(ex);
 
                 activity?.Dispose();
             }

src/KafkaFlow.OpenTelemetry/OpenTelemetryProducerEventsHandler.cs

@@ -7,6 +7,7 @@
     using System.Threading.Tasks;
     using global::OpenTelemetry;
     using global::OpenTelemetry.Context.Propagation;
+    using global::OpenTelemetry.Trace;
 
     internal static class OpenTelemetryProducerEventsHandler
     {
@@ -77,9 +78,8 @@ public static Task OnProducerError(IMessageContext context, Exception ex)
         {
             if (context.Items.TryGetValue(ActivitySourceAccessor.ActivityString, out var value) && value is Activity activity)
             {
-                var exceptionEvent = ActivitySourceAccessor.CreateExceptionEvent(ex);
-
-                activity?.AddEvent(exceptionEvent);
+                activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
+                activity?.RecordException(ex);
 
                 activity?.Dispose();
             }