feat: Include OpenTelemetry tracing

samples/KafkaFlow.Sample/KafkaFlow.Sample.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
     <PropertyGroup>
         <OutputType>Exe</OutputType>
@@ -22,6 +22,7 @@
       <ProjectReference Include="..\..\src\KafkaFlow.Compressor\KafkaFlow.Compressor.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.LogHandler.Console\KafkaFlow.LogHandler.Console.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Microsoft.DependencyInjection\KafkaFlow.Microsoft.DependencyInjection.csproj" />
+      <ProjectReference Include="..\..\src\KafkaFlow.OpenTelemetry\KafkaFlow.OpenTelemetry.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Serializer.JsonCore\KafkaFlow.Serializer.JsonCore.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Serializer.ProtobufNet\KafkaFlow.Serializer.ProtobufNet.csproj" />
       <ProjectReference Include="..\..\src\KafkaFlow.Serializer\KafkaFlow.Serializer.csproj" />
@@ -31,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,11 +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,6 +43,7 @@
                         )
                 )
         )
+        .AddOpenTelemetryInstrumentation()
 );
 
 var provider = services.BuildServiceProvider();
@@ -52,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 @@
         /// <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.Abstractions/IConsumerContext.cs

@@ -1,6 +1,7 @@
 namespace KafkaFlow
 {
     using System;
+    using System.Collections.Generic;
     using System.Threading;
 
     /// <summary>
@@ -73,5 +74,8 @@ public interface IConsumerContext
         /// Resume Kafka's message fetch
         /// </summary>
         void Resume();
+
+        // TODO: Need to discuss
+        IReadOnlyCollection<string> Brokers { get; }
     }
 }

src/KafkaFlow.Admin.Dashboard/KafkaFlow.Admin.Dashboard.csproj

@@ -39,5 +39,5 @@
     <ProjectReference Include="..\KafkaFlow.Admin\KafkaFlow.Admin.csproj" />
     <ProjectReference Include="..\KafkaFlow\KafkaFlow.csproj" />
   </ItemGroup>
-
+  
 </Project>

src/KafkaFlow.OpenTelemetry/ExtensionMethods.cs

@@ -0,0 +1,31 @@
+namespace KafkaFlow.Configuration
+{
+    using KafkaFlow.OpenTelemetry.Trace;
+
+    /// <summary>
+    /// Adds OpenTelemetry instrumentation
+    /// </summary>
+    public static class ExtensionMethods
+    {
+        /// <summary>
+        /// Adds OpenTelemetry instrumentation
+        /// </summary>
+        /// <param name="builder">The Kafka configuration builder</param>
+        /// <returns></returns>
+        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/KafkaFlow.OpenTelemetry.csproj

@@ -0,0 +1,19 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netstandard2.0</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="OpenTelemetry.Api" Version="1.6.0" />
+    <PackageReference Include="OpenTelemetry.Extensions.Propagators" Version="1.6.0" />
+    <PackageReference Include="OpenTelemetry.SemanticConventions" Version="1.0.0-rc9.9">
+      <Aliases>SemanticConventions</Aliases>
+    </PackageReference>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\KafkaFlow.Abstractions\KafkaFlow.Abstractions.csproj" />
+  </ItemGroup>
+
+</Project>

src/KafkaFlow.OpenTelemetry/Trace/KafkaFlowActivitySourceHelper.cs

@@ -0,0 +1,33 @@
+extern alias SemanticConventions;
+
+namespace KafkaFlow.OpenTelemetry.Trace
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Diagnostics;
+    using System.Reflection;
+    using Conventions = SemanticConventions::OpenTelemetry.Trace.TraceSemanticConventions;
+
+    public static class KafkaFlowActivitySourceHelper
+    {
+        internal static readonly IEnumerable<KeyValuePair<string, object>> CreationTags = new[]
+        {
+            new KeyValuePair<string, object>(Conventions.AttributePeerService, KafkaString),
+            new KeyValuePair<string, object>(Conventions.AttributeMessagingSystem, KafkaString),
+        };
+
+        internal static readonly string KafkaFlowString = "KafkaFlow";
+        internal static readonly string KafkaString = "kafka";
+
+        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);
+            // TODO: Broker information below. Set values after
+            activity?.SetTag("peer.service", KafkaString);
+        }
+    }
+}