PoC: Include event-based instrumentation

samples/KafkaFlow.Sample/KafkaFlow.Sample.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
     <PropertyGroup>
         <OutputType>Exe</OutputType>
@@ -32,6 +32,8 @@
 
     <ItemGroup>
       <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="3.1.17" />
+      <PackageReference Include="OpenTelemetry" Version="1.6.0" />
+      <PackageReference Include="OpenTelemetry.Exporter.Console" Version="1.6.0" />
     </ItemGroup>
 
 </Project>

samples/KafkaFlow.Sample/Program.cs

@@ -1,12 +1,16 @@
 using System;
+using System.Diagnostics;
 using System.Threading.Tasks;
 using KafkaFlow;
 using KafkaFlow.Configuration;
+using KafkaFlow.OpenTelemetry.Trace;
 using KafkaFlow.Producers;
 using KafkaFlow.Sample;
 using KafkaFlow.Serializer;
 using KafkaFlow.TypedHandler;
 using Microsoft.Extensions.DependencyInjection;
+using OpenTelemetry;
+using OpenTelemetry.Trace;
 
 var services = new ServiceCollection();
 
@@ -38,8 +42,8 @@
                                 .AddTypedHandlers(h => h.AddHandler<PrintConsoleHandler>())
                         )
                 )
-                .AddOpenTelemetryInstrumentation()
         )
+        .AddOpenTelemetryInstrumentation()
 );
 
 var provider = services.BuildServiceProvider();
@@ -54,6 +58,11 @@
 
 Console.WriteLine("Type the number of messages to produce or 'exit' to quit:");
 
+using var tracerProvider = Sdk.CreateTracerProviderBuilder()
+    .AddSource("KafkaFlow")
+    .AddConsoleExporter()
+    .Build();
+
 while (true)
 {
     var input = Console.ReadLine();

src/KafkaFlow.Abstractions/Configuration/IKafkaConfigurationBuilder.cs

@@ -1,6 +1,7 @@
 namespace KafkaFlow.Configuration
 {
     using System;
+    using KafkaFlow.Events;
 
     /// <summary>
     /// A builder to configure KafkaFlow
@@ -21,5 +22,7 @@ public interface IKafkaConfigurationBuilder
         /// <returns></returns>
         IKafkaConfigurationBuilder UseLogHandler<TLogHandler>()
             where TLogHandler : ILogHandler;
+
+        IKafkaConfigurationBuilder SubscribeEvents(Action<IEventsListener> eventsListener);
     }
 }

src/KafkaFlow.Abstractions/Events/Args/ConsumeErrorEventArgs.cs

@@ -0,0 +1,14 @@
+using System;
+
+namespace KafkaFlow.Events.Args
+{
+    public class ConsumeErrorEventArgs
+    {
+        public ConsumeErrorEventArgs(Exception exception)
+        {
+            this.Exception = exception;
+        }
+
+        public Exception Exception { get; set; }
+    }
+}

src/KafkaFlow.Abstractions/Events/Args/ConsumeStartEventArgs.cs

@@ -0,0 +1,12 @@
+namespace KafkaFlow.Events.Args
+{
+    public class ConsumeStartEventArgs
+    {
+        internal ConsumeStartEventArgs(IMessageContext context)
+        {
+            this.MessageContext = context;
+        }
+
+        public IMessageContext MessageContext { get; set; }
+    }
+}

src/KafkaFlow.Abstractions/Events/Args/ProduceErrorEventArgs.cs

@@ -0,0 +1,14 @@
+using System;
+
+namespace KafkaFlow.Events.Args
+{
+    public class ProduceErrorEventArgs
+    {
+        public ProduceErrorEventArgs(Exception exception)
+        {
+            this.Exception = exception;
+        }
+
+        public Exception Exception { get; set; }
+    }
+}

src/KafkaFlow.Abstractions/Events/Args/ProduceStartEventArgs.cs

@@ -0,0 +1,12 @@
+namespace KafkaFlow.Events.Args
+{
+    public class ProduceStartEventArgs
+    {
+        internal ProduceStartEventArgs(IMessageContext context)
+        {
+            this.MessageContext = context;
+        }
+
+        public IMessageContext MessageContext { get; set; }
+    }
+}

src/KafkaFlow.Abstractions/Events/EventsManager.cs

@@ -0,0 +1,36 @@
+using System;
+using KafkaFlow.Events.Args;
+
+namespace KafkaFlow.Events
+{
+    public class EventsManager : IEventsListener, IEventsNotifier
+    {
+        public event EventHandler<ConsumeStartEventArgs> OnConsumeStart;
+
+        public event EventHandler<ProduceStartEventArgs> OnProduceStart;
+
+        public event EventHandler<ConsumeErrorEventArgs> OnConsumeError;
+
+        public event EventHandler<ProduceErrorEventArgs> OnProduceError;
+
+        public void NotifyOnConsumeError(Exception exception)
+        {
+            this.OnConsumeError?.Invoke(this, new ConsumeErrorEventArgs(exception));
+        }
+
+        public void NotifyOnConsumeStart(IMessageContext context)
+        {
+            this.OnConsumeStart?.Invoke(this, new ConsumeStartEventArgs(context));
+        }
+
+        public void NotifyOnProduceError(Exception exception)
+        {
+            this.OnProduceError?.Invoke(this, new ProduceErrorEventArgs(exception));
+        }
+
+        public void NotifyOnProduceStart(IMessageContext context)
+        {
+            this.OnProduceStart?.Invoke(this, new ProduceStartEventArgs(context));
+        }
+    }
+}

src/KafkaFlow.Abstractions/Events/IEventsListener.cs

@@ -0,0 +1,18 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using KafkaFlow.Events.Args;
+
+namespace KafkaFlow.Events
+{
+    public interface IEventsListener
+    {
+        event EventHandler<ConsumeStartEventArgs> OnConsumeStart;
+
+        event EventHandler<ProduceStartEventArgs> OnProduceStart;
+
+        event EventHandler<ConsumeErrorEventArgs> OnConsumeError;
+
+        event EventHandler<ProduceErrorEventArgs> OnProduceError;
+    }
+}

src/KafkaFlow.Abstractions/Events/IEventsNotifier.cs

@@ -0,0 +1,18 @@
+using System;
+
+namespace KafkaFlow.Events
+{
+    // Isto devia ser internal visto que a notificação de events não deveria ser publica, apenas a subscrição desses eventos
+
+    // Falta o OnConsumeEnd e OnProduceEnd
+    public interface IEventsNotifier
+    {
+        void NotifyOnConsumeError(Exception exception);
+
+        void NotifyOnConsumeStart(IMessageContext context);
+
+        void NotifyOnProduceError(Exception exception);
+
+        void NotifyOnProduceStart(IMessageContext context);
+    }
+}

src/KafkaFlow.Admin.Dashboard/Properties/launchSettings.json

@@ -0,0 +1,12 @@
+{
+  "profiles": {
+    "KafkaFlow.Admin.Dashboard": {
+      "commandName": "Project",
+      "launchBrowser": true,
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      },
+      "applicationUrl": "https://localhost:57596;http://localhost:57597"
+    }
+  }
+}

src/KafkaFlow.OpenTelemetry/ExtensionMethods.cs

@@ -12,7 +12,20 @@ public static class ExtensionMethods
         /// </summary>
         /// <param name="builder">The Kafka configuration builder</param>
         /// <returns></returns>
-        public static IClusterConfigurationBuilder AddOpenTelemetryInstrumentation(this IClusterConfigurationBuilder builder) =>
-            builder.AddInstrumentation<TracerConsumerMiddleware, TracerProducerMiddleware>();
+        public static IKafkaConfigurationBuilder AddOpenTelemetryInstrumentation(this IKafkaConfigurationBuilder builder)
+        {
+            var tracerConsumerMiddleware = new TracerConsumerMiddleware();
+            var tracerProducerMiddleware = new TracerProducerMiddleware();
+
+            builder.SubscribeEvents(events =>
+            {
+                events.OnConsumeError += (sender, args) => tracerConsumerMiddleware.UpdateActivityOnError(args.Exception);
+                events.OnConsumeStart += (sender, args) => tracerConsumerMiddleware.CreateActivityOnConsume(args.MessageContext);
+                events.OnProduceError += (sender, args) => tracerProducerMiddleware.UpdateActivityOnError(args.Exception);
+                events.OnProduceStart += (sender, args) => tracerProducerMiddleware.CreateActivityOnProduce(args.MessageContext);
+            });
+
+            return builder;
+        }
     }
 }

src/KafkaFlow.OpenTelemetry/Trace/KafkaFlowActivitySourceHelper.cs

@@ -8,10 +8,8 @@ namespace KafkaFlow.OpenTelemetry.Trace
     using System.Reflection;
     using Conventions = SemanticConventions::OpenTelemetry.Trace.TraceSemanticConventions;
 
-    internal static class KafkaFlowActivitySourceHelper
+    public static class KafkaFlowActivitySourceHelper
     {
-        public static readonly ActivitySource ActivitySource = new ActivitySource(KafkaFlowString, Version.ToString());
-
         internal static readonly IEnumerable<KeyValuePair<string, object>> CreationTags = new[]
         {
             new KeyValuePair<string, object>(Conventions.AttributePeerService, KafkaString),
@@ -23,6 +21,8 @@ internal static class KafkaFlowActivitySourceHelper
 
         private static readonly Version Version = Assembly.GetExecutingAssembly().GetName().Version;
 
+        public static readonly ActivitySource ActivitySource = new ActivitySource(KafkaFlowString, Version.ToString());
+
         public static void SetGenericTags(Activity activity)
         {
             activity?.SetTag("messaging.system", KafkaString);

src/KafkaFlow.OpenTelemetry/Trace/TracerConsumerMiddleware.cs

@@ -4,44 +4,53 @@
     using System.Collections.Generic;
     using System.Diagnostics;
     using System.Text;
-    using System.Threading.Tasks;
     using global::OpenTelemetry;
     using global::OpenTelemetry.Context.Propagation;
 
-    internal class TracerConsumerMiddleware : IConsumerInstrumentationMiddleware
+    internal class TracerConsumerMiddleware
     {
         private static readonly TextMapPropagator Propagator = Propagators.DefaultTextMapPropagator;
         private static readonly string ProcessString = "process";
 
-        public async Task Invoke(IMessageContext context, MiddlewareDelegate next)
+        public void CreateActivityOnConsume(IMessageContext context)
         {
-            var activityName = !string.IsNullOrEmpty(context?.ConsumerContext.Topic) ? $"{context.ConsumerContext.Topic} {ProcessString}" : ProcessString;
+            try
+            {
+                var activityName = !string.IsNullOrEmpty(context?.ConsumerContext.Topic) ? $"{context.ConsumerContext.Topic} {ProcessString}" : ProcessString;
 
-            // Extract the PropagationContext of the upstream parent from the message headers.
-            var parentContext = Propagator.Extract(new PropagationContext(default, Baggage.Current), context, this.ExtractTraceContextIntoBasicProperties);
-            Baggage.Current = parentContext.Baggage;
+                // Extract the PropagationContext of the upstream parent from the message headers.
+                var parentContext = Propagator.Extract(new PropagationContext(default, Baggage.Current), context, this.ExtractTraceContextIntoBasicProperties);
+                Baggage.Current = parentContext.Baggage;
 
-            // Start an activity with a name following the semantic convention of the OpenTelemetry messaging specification.
-            // The convention also defines a set of attributes (in .NET they are mapped as `tags`) to be populated in the activity.
-            // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/messaging.md
-            using var activity = KafkaFlowActivitySourceHelper.ActivitySource.StartActivity(activityName, ActivityKind.Consumer, parentContext.ActivityContext);
+                // Start an activity with a name following the semantic convention of the OpenTelemetry messaging specification.
+                // The convention also defines a set of attributes (in .NET they are mapped as `tags`) to be populated in the activity.
+                // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/messaging.md
+                using var activity = KafkaFlowActivitySourceHelper.ActivitySource.StartActivity(activityName, ActivityKind.Consumer, parentContext.ActivityContext);
 
-            try
-            {
                 KafkaFlowActivitySourceHelper.SetGenericTags(activity);
 
                 if (activity != null && activity.IsAllDataRequested)
                 {
                     this.SetConsumerTags(context, activity);
                 }
-
-                await next.Invoke(context);
             }
-            catch (Exception ex)
+            catch
             {
-                activity?.SetTag("exception.message", ex.Message);
+                // ignored
+            }
+        }
 
-                throw;
+        public void UpdateActivityOnError(Exception ex)
+        {
+            try
+            {
+                var activity = Activity.Current;
+
+                activity?.SetTag("exception.message", ex.Message);
+            }
+            catch
+            {
+                // ignored
             }
         }
 
@@ -63,7 +72,7 @@ private void SetConsumerTags(IMessageContext context, Activity activity)
             activity.SetTag("messaging.source.name", context.ConsumerContext.Topic);
             activity.SetTag("messaging.kafka.consumer.group", context.ConsumerContext.GroupId);
             activity.SetTag("messaging.kafka.message.key", context.Message.Key);
-            activity.SetTag("messaging.kafka.message.offset", context.ProducerContext.Offset);
+            activity.SetTag("messaging.kafka.message.offset", context.ConsumerContext.Offset);
             activity.SetTag("messaging.kafka.source.partition", context.ConsumerContext.Partition);
         }
     }