Include consumer and producer test

Farfetch · Oct 11, 2023 · a14f601 · a14f601
1 parent 23c3745
commit a14f601
Show file tree

Hide file tree

Showing 4 changed files with 141 additions and 11 deletions.
diff --git a/src/KafkaFlow.IntegrationTests/KafkaFlow.IntegrationTests.csproj b/src/KafkaFlow.IntegrationTests/KafkaFlow.IntegrationTests.csproj
@@ -25,6 +25,7 @@
           <PrivateAssets>all</PrivateAssets>
           <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
         </PackageReference>
+		<PackageReference Include="OpenTelemetry.Exporter.InMemory" Version="1.6.0" />
     </ItemGroup>
 
     <ItemGroup Condition="$([MSBuild]::IsOsPlatform('OSX'))">
@@ -37,6 +38,7 @@
         <ProjectReference Include="..\KafkaFlow.Compressor.Gzip\KafkaFlow.Compressor.Gzip.csproj" />
         <ProjectReference Include="..\KafkaFlow.Compressor\KafkaFlow.Compressor.csproj" />
         <ProjectReference Include="..\KafkaFlow.Microsoft.DependencyInjection\KafkaFlow.Microsoft.DependencyInjection.csproj" />
+        <ProjectReference Include="..\KafkaFlow.OpenTelemetry\KafkaFlow.OpenTelemetry.csproj" />
         <ProjectReference Include="..\KafkaFlow.Serializer.JsonCore\KafkaFlow.Serializer.JsonCore.csproj" />
         <ProjectReference Include="..\KafkaFlow.Serializer.ProtobufNet\KafkaFlow.Serializer.ProtobufNet.csproj" />
         <ProjectReference Include="..\KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro\KafkaFlow.Serializer.SchemaRegistry.ConfluentAvro.csproj" />

diff --git a/src/KafkaFlow.IntegrationTests/OpenTelemetryTests.cs b/src/KafkaFlow.IntegrationTests/OpenTelemetryTests.cs
@@ -0,0 +1,139 @@
+namespace KafkaFlow.IntegrationTests
+{
+    using System;
+    using System.IO;
+    using System.Threading;
+    using Confluent.SchemaRegistry;
+    using Confluent.SchemaRegistry.Serdes;
+    using global::Microsoft.Extensions.Configuration;
+    using global::Microsoft.Extensions.Hosting;
+    using KafkaFlow.Configuration;
+    using KafkaFlow.IntegrationTests.Core.Handlers;
+    using KafkaFlow.IntegrationTests.Core.Producers;
+    using KafkaFlow.Serializer.SchemaRegistry;
+    using KafkaFlow.TypedHandler;
+    using AutoOffsetReset = KafkaFlow.AutoOffsetReset;
+    using Microsoft.VisualStudio.TestTools.UnitTesting;
+    using System.Threading.Tasks;
+    using KafkaFlow.IntegrationTests.Core;
+    using AutoFixture;
+    using System.Collections.Generic;
+    using System.Diagnostics;
+    using global::OpenTelemetry;
+    using global::OpenTelemetry.Trace;
+    using Microsoft.Extensions.DependencyInjection;
+    using KafkaFlow.Compressor;
+    using KafkaFlow.Compressor.Gzip;
+    using KafkaFlow.IntegrationTests.Core.Middlewares;
+
+    [TestClass]
+    public class OpenTelemetryTests
+    {
+        private readonly Fixture fixture = new();
+
+        private List<Activity> exportedItems;
+
+        private IServiceProvider provider;
+
+        [TestInitialize]
+        public void Setup()
+        {
+            this.exportedItems = new List<Activity>();
+            //this.BuildTracerProvider();
+            this.provider = this.GetServiceProvider();
+            MessageStorage.Clear();
+        }
+
+        [TestMethod]
+        public async Task AddOpenTelemetry_ProducingAndConsumingOneMessage_TraceAndSpansAreCreatedCorrectly()
+        {
+            // Arrange
+            using var tracerProvider = Sdk.CreateTracerProviderBuilder()
+            .AddSource("KafkaFlow")
+            .AddInMemoryExporter(this.exportedItems)
+            .Build();
+
+            var producer = this.provider.GetRequiredService<IMessageProducer<GzipProducer>>();
+            var message = this.fixture.Create<byte[]>();
+
+            // Act
+            await producer.ProduceAsync(null, message);
+
+            // Assert
+            await Task.Delay(8000).ConfigureAwait(false);
+
+            Assert.IsNotNull(this.exportedItems);
+            var producerSpan = this.exportedItems.Find(x => x.Kind == ActivityKind.Producer);
+            var consumerSpan = this.exportedItems.Find(x => x.Kind == ActivityKind.Consumer);
+
+            Assert.IsNull(producerSpan.ParentId);
+            Assert.AreEqual(producerSpan.TraceId, consumerSpan.TraceId);
+            Assert.AreEqual(consumerSpan.ParentSpanId, producerSpan.SpanId);
+        }
+
+        private IServiceProvider GetServiceProvider()
+        {
+            string topicName = "Otel";
+
+            var builder = Host
+                .CreateDefaultBuilder()
+                .ConfigureAppConfiguration(
+                    (_, config) =>
+                    {
+                        config
+                            .SetBasePath(Directory.GetCurrentDirectory())
+                            .AddJsonFile(
+                                "conf/appsettings.json",
+                                false,
+                                true)
+                            .AddEnvironmentVariables();
+                    })
+                .ConfigureServices((context, services) =>
+                    services.AddKafka(
+                        kafka => kafka
+                            .UseLogHandler<TraceLogHandler>()
+                            .AddCluster(
+                                cluster => cluster
+                                    .WithBrokers(context.Configuration.GetValue<string>("Kafka:Brokers").Split(';'))
+                                    .CreateTopicIfNotExists(topicName, 1, 1)
+                                    .AddProducer<GzipProducer>(
+                                        producer => producer
+                                            .DefaultTopic(topicName)
+                                            .AddMiddlewares(
+                                                middlewares => middlewares
+                                                    .AddCompressor<GzipMessageCompressor>()))
+                                    .AddConsumer(
+                                        consumer => consumer
+                                            .Topic(topicName)
+                                            .WithGroupId(topicName)
+                                            .WithBufferSize(100)
+                                            .WithWorkersCount(10)
+                                            .WithAutoOffsetReset(AutoOffsetReset.Latest)
+                                            .AddMiddlewares(
+                                                middlewares => middlewares
+                                                    .AddCompressor<GzipMessageCompressor>()
+                                                    .Add<GzipMiddleware>())))
+                            .AddOpenTelemetryInstrumentation()))
+                .UseDefaultServiceProvider(
+                    (_, options) =>
+                    {
+                        options.ValidateScopes = true;
+                        options.ValidateOnBuild = true;
+                    });
+
+            var host = builder.Build();
+            var bus = host.Services.CreateKafkaBus();
+            bus.StartAsync().GetAwaiter().GetResult();
+
+            // Wait partition assignment
+            Thread.Sleep(10000);
+
+            return host.Services;
+        }
+
+        private void BuildTracerProvider()
+        {
+
+        }
+    }
+}
diff --git a/src/KafkaFlow.OpenTelemetry/OpenTelemetryConsumerObserver.cs b/src/KafkaFlow.OpenTelemetry/OpenTelemetryConsumerObserver.cs
@@ -36,10 +36,6 @@ public static Task OnConsumeStarted(IMessageContext context)
                 {
                     SetConsumerTags(context, activity);
                 }
-
-                Console.WriteLine(activity.TraceId);
-                Console.WriteLine(activity.SpanId);
-                Console.WriteLine(activity.ParentSpanId);
             }
             catch
             {
@@ -51,9 +47,6 @@ public static Task OnConsumeStarted(IMessageContext context)
         public static Task OnConsumeCompleted(IMessageContext context)
         {
             var activity = context.Items[KafkaFlowActivitySourceHelper.ActivityString] as Activity;
-            Console.WriteLine(activity.TraceId);
-            Console.WriteLine(activity.SpanId);
-            Console.WriteLine(activity.ParentSpanId);
 
             activity?.Dispose();
 

diff --git a/src/KafkaFlow.OpenTelemetry/OpenTelemetryProducerObserver.cs b/src/KafkaFlow.OpenTelemetry/OpenTelemetryProducerObserver.cs
@@ -50,10 +50,6 @@ public static Task OnProducerStarted(IMessageContext context)
                 {
                     SetProducerTags(context, activity);
                 }
-
-                Console.WriteLine(activity.TraceId);
-                Console.WriteLine(activity.SpanId);
-                Console.WriteLine(activity.ParentSpanId);
             }
             catch
             {