feat: upgrade exception tags in otel (#476)

Change how the exceptions are being recorded in otel
activities to be in accordance with the official documentation.

Make the KafkaFlow ActivitySource name public so the users of
manual instrumentation can easily refer to without having
to type it.

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