Include AddOpenTelemetryInstrumentation in builder

samples/KafkaFlow.Sample/KafkaFlow.Sample.csproj

@@ -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" />

samples/KafkaFlow.Sample/Program.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Threading.Tasks;
 using KafkaFlow;
+using KafkaFlow.Configuration;
 using KafkaFlow.Producers;
 using KafkaFlow.Sample;
 using KafkaFlow.Serializer;
@@ -37,6 +38,7 @@
                                 .AddTypedHandlers(h => h.AddHandler<PrintConsoleHandler>())
                         )
                 )
+                .AddOpenTelemetryInstrumentation()
         )
 );
 

src/KafkaFlow.Abstractions/Configuration/IClusterConfigurationBuilder.cs

@@ -82,5 +82,10 @@ IClusterConfigurationBuilder CreateTopicIfNotExists(
             string topicName,
             int numberOfPartitions,
             short replicationFactor);
+
+        // Gets the cluster configuration
+        IClusterConfigurationBuilder AddInstrumentation<TConsumerInstrumentationMiddleware, TProducerInstrumentationMiddleware>()
+            where TConsumerInstrumentationMiddleware : class, IMessageMiddleware
+            where TProducerInstrumentationMiddleware : class, IMessageMiddleware;
     }
 }

src/KafkaFlow.Abstractions/Configuration/IInstrumentationBuilder.cs

@@ -0,0 +1,11 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace KafkaFlow.Configuration
+{
+    public interface IInstrumentationBuilder
+    {
+        void AddInstrumentation();
+    }
+}

src/KafkaFlow.OpenTelemetry/ExtensionMethods.cs

@@ -0,0 +1,18 @@
+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 IClusterConfigurationBuilder AddOpenTelemetryInstrumentation(this IClusterConfigurationBuilder builder) =>
+            builder.AddInstrumentation<TracerConsumerMiddleware, TracerProducerMiddleware>();
+    }
+}

src/KafkaFlow.OpenTelemetry/KafkaFlow.OpenTelemetry.csproj

@@ -6,7 +6,7 @@
 
   <ItemGroup>
     <PackageReference Include="OpenTelemetry.Api" Version="1.6.0" />
-    <PackageReference Include="OpenTelemetry.Extensions.Propagators" 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>

src/KafkaFlow.OpenTelemetry/Trace/KafkaFlowActivitySourceHelper.cs

@@ -8,7 +8,7 @@ namespace KafkaFlow.OpenTelemetry.Trace
     using System.Reflection;
     using Conventions = SemanticConventions::OpenTelemetry.Trace.TraceSemanticConventions;
 
-    internal class KafkaFlowActivitySourceHelper
+    internal static class KafkaFlowActivitySourceHelper
     {
         public static readonly ActivitySource ActivitySource = new ActivitySource(KafkaFlowString, Version.ToString());
 

src/KafkaFlow/Configuration/ClusterConfigurationBuilder.cs

@@ -15,6 +15,8 @@ internal class ClusterConfigurationBuilder : IClusterConfigurationBuilder
         private IEnumerable<string> brokers;
         private string name;
         private Func<SecurityInformation> securityInformationHandler;
+        private Type instrumentationConsumerMiddleware = typeof(IMessageMiddleware);
+        private Type instrumentationProducerMiddleware = typeof(IMessageMiddleware);
 
         public ClusterConfigurationBuilder(IDependencyConfigurator dependencyConfigurator)
         {
@@ -35,7 +37,15 @@ public ClusterConfiguration Build(KafkaConfiguration kafkaConfiguration)
                 this.topicsToCreateIfNotExist);
 
             configuration.AddProducers(this.producers.Select(x => x.Build(configuration)));
-            configuration.AddConsumers(this.consumers.Select(x => x.Build(configuration)));
+            configuration.AddConsumers(this.consumers.Select(x =>
+            {
+                if(this.instrumentationConsumerMiddleware != null)
+                {
+                    x.AddInstrumentation<this.instrumentationConsumerMiddleware>();
+                }
+
+                return x.Build(configuration);
+            }));
 
             return configuration;
         }
@@ -96,6 +106,16 @@ public IClusterConfigurationBuilder AddConsumer(Action<IConsumerConfigurationBui
             return this;
         }
 
+        public IClusterConfigurationBuilder AddInstrumentation<TConsumerInstrumentationMiddleware, TProducerInstrumentationMiddleware>()
+            where TConsumerInstrumentationMiddleware : class, IMessageMiddleware
+            where TProducerInstrumentationMiddleware : class, IMessageMiddleware
+        {
+
+            this.instrumentationConsumerMiddleware = typeof(TConsumerInstrumentationMiddleware);
+
+            return this;
+        }
+
         public IClusterConfigurationBuilder OnStopping(Action<IDependencyResolver> handler)
         {
             this.onStoppingHandler = handler;

src/KafkaFlow/Configuration/ConsumerConfigurationBuilder.cs

@@ -229,6 +229,12 @@ public IConsumerConfigurationBuilder WithCustomFactory(ConsumerCustomFactory cus
             return this;
         }
 
+        public void AddInstrumentation<T>()
+            where T : class, IMessageMiddleware
+        {
+            this.middlewareConfigurationBuilder.AddAtBeginning<T>();
+        }
+
         public IConsumerConfiguration Build(ClusterConfiguration clusterConfiguration)
         {
             var middlewareConfiguration = this.middlewareConfigurationBuilder.Build();