Add ConfluentKafka instrumentation

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -30,6 +30,7 @@ body:
         - OpenTelemetry.Instrumentation.AWS
         - OpenTelemetry.Instrumentation.AWSLambda
         - OpenTelemetry.Instrumentation.Cassandra
+        - OpenTelemetry.Instrumentation.ConfluentKafka        
         - OpenTelemetry.Instrumentation.ElasticsearchClient
         - OpenTelemetry.Instrumentation.EntityFrameworkCore
         - OpenTelemetry.Instrumentation.EventCounters

.github/ISSUE_TEMPLATE/feature_request.yml

@@ -30,6 +30,7 @@ body:
         - OpenTelemetry.Instrumentation.AWS
         - OpenTelemetry.Instrumentation.AWSLambda
         - OpenTelemetry.Instrumentation.Cassandra
+        - OpenTelemetry.Instrumentation.ConfluentKafka
         - OpenTelemetry.Instrumentation.ElasticsearchClient
         - OpenTelemetry.Instrumentation.EntityFrameworkCore
         - OpenTelemetry.Instrumentation.EventCounters

.github/codecov.yml

@@ -63,6 +63,11 @@ flags:
     paths:
       - src/OpenTelemetry.Instrumentation.AspNetCore
 
+  unittests-Instrumentation.ConfluentKafka:
+    carryforward: true
+    paths:
+      - src/OpenTelemetry.Instrumentation.ConfluentKafka
+
   unittests-Instrumentation.EventCounters:
     carryforward: true
     paths:

.github/component_owners.yml

@@ -35,6 +35,8 @@ components:
     - ppittle
   src/OpenTelemetry.Instrumentation.Cassandra/:
     - xsoheilalizadeh
+  src/OpenTelemetry.Instrumentation.ConfluentKafka/:
+    - g7ed6e    
   src/OpenTelemetry.Instrumentation.ElasticsearchClient/:
     - ejsmith
   src/OpenTelemetry.Instrumentation.EventCounters/:
@@ -130,6 +132,8 @@ components:
     - ppittle
   test/OpenTelemetry.Instrumentation.Cassandra.Tests/:
     - xsoheilalizadeh
+  test/OpenTelemetry.Instrumentation.ConfluentKafka.Tests/:
+    - g7ed6e    
   test/OpenTelemetry.Instrumentation.ElasticsearchClient.Tests/:
     - ejsmith
   test/OpenTelemetry.Instrumentation.EventCounters.Tests/:

.github/workflows/ci.yml

@@ -36,6 +36,7 @@ jobs:
           instrumentation-aws: ['*/OpenTelemetry.Instrumentation.AWS/**', '*/OpenTelemetry.Instrumentation.AWS.Tests/**', '!**/*.md']
           instrumentation-aws-lambda: ['*/OpenTelemetry.Instrumentation.AWSLambda/**', '*/OpenTelemetry.Instrumentation.AWSLambda.Tests/**', '!**/*.md']
           instrumentation-cassandra: ['*/OpenTelemetry.Instrumentation.Cassandra*/**', '!**/*.md']
+          instrumentation-confluentkafka: ['*/OpenTelemetry.Instrumentation.ConfluentKafka*/**', 'examples/kafka/**', '!**/*.md']
           instrumentation-elasticsearchclient: ['*/OpenTelemetry.Instrumentation.ElasticsearchClient*/**', '!**/*.md']
           instrumentation-entityframeworkcore: ['*/OpenTelemetry.Instrumentation.EntityFrameworkCore*/**', '!**/*.md']
           instrumentation-eventcounters: ['*/OpenTelemetry.Instrumentation.EventCounters*/**', 'examples/event-counters/**', '!**/*.md']
@@ -222,6 +223,27 @@ jobs:
       project-name: Component[OpenTelemetry.Instrumentation.Cassandra]
       code-cov-name: Instrumentation.Cassandra
 
+  build-test-instrumentation-confluentkafka:
+    needs: detect-changes
+    if: |
+      contains(needs.detect-changes.outputs.changes, 'instrumentation-confluentkafka')
+      || contains(needs.detect-changes.outputs.changes, 'build')
+      || contains(needs.detect-changes.outputs.changes, 'shared')
+    uses: ./.github/workflows/Component.BuildTest.yml
+    with:
+      project-name: Component[OpenTelemetry.Instrumentation.ConfluentKafka]
+      code-cov-name: Instrumentation.ConfluentKafka
+
+  build-test-instrumentation-confluentkafka-integration:
+    needs: detect-changes
+    if: |
+      contains(needs.detect-changes.outputs.changes, 'instrumentation-confluentkafka')
+      || contains(needs.detect-changes.outputs.changes, 'build')
+      || contains(needs.detect-changes.outputs.changes, 'shared')
+    uses: ./.github/workflows/integration.yml
+    with:
+      job: kafka-integration-test
+
   build-test-instrumentation-elasticsearchclient:
     needs: detect-changes
     if: |
@@ -375,6 +397,8 @@ jobs:
       || contains(needs.detect-changes.outputs.changes, 'build')
       || contains(needs.detect-changes.outputs.changes, 'shared')
     uses: ./.github/workflows/integration.yml
+    with:
+      job: redis-integration-test
 
   build-test-instrumentation-wcf:
     needs: detect-changes
@@ -518,6 +542,7 @@ jobs:
       || contains(needs.detect-changes.outputs.changes, 'instrumentation-aspnetcore')
       || contains(needs.detect-changes.outputs.changes, 'instrumentation-aws')
       || contains(needs.detect-changes.outputs.changes, 'instrumentation-awslambda')
+      || contains(needs.detect-changes.outputs.changes, 'instrumentation-confluentkafka')
       || contains(needs.detect-changes.outputs.changes, 'instrumentation-eventcounters')
       || contains(needs.detect-changes.outputs.changes, 'instrumentation-grpcnetclient')
       || contains(needs.detect-changes.outputs.changes, 'instrumentation-http')
@@ -556,6 +581,7 @@ jobs:
       build-test-instrumentation-aws,
       build-test-instrumentation-awslambda,
       build-test-instrumentation-cassandra,
+      build-test-instrumentation-confluentkafka,
       build-test-instrumentation-elasticsearchclient,
       build-test-instrumentation-entityframeworkcore,
       build-test-instrumentation-eventcounters,

.github/workflows/integration.yml

@@ -2,9 +2,14 @@ name: Integration Build OpenTelemetry.Instrumentation.StackExchangeRedis
 
 on:
   workflow_call:
+    inputs:
+      job:
+        required: true
+        type: string
 
 jobs:
   redis-integration-test:
+    if: inputs.job == 'all' || inputs.job == 'redis-integration-test'
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
@@ -13,5 +18,18 @@ jobs:
     steps:
       - uses: actions/checkout@v4
 
-      - name: Run redis docker-compose.integration
+      - name: Run redis docker-compose
         run: docker-compose --file=test/OpenTelemetry.Instrumentation.StackExchangeRedis.Tests/docker-compose.yml --file=build/docker-compose.${{ matrix.version }}.yml --project-directory=. up --exit-code-from=tests --build
+
+  kafka-integration-test:
+    if: inputs.job == 'all' || inputs.job == 'kafka-integration-test'
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        version: [net6.0, net7.0, net8.0]
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Run kafka docker-compose
+        run: docker-compose --file=test/OpenTelemetry.Instrumentation.ConfluentKafka.Tests/docker-compose.yml --file=build/docker-compose.${{ matrix.version }}.yml --project-directory=. up --exit-code-from=tests --build

.github/workflows/prepare-release.yml

@@ -20,6 +20,7 @@ on:
         - OpenTelemetry.Instrumentation.AWS
         - OpenTelemetry.Instrumentation.AWSLambda
         - OpenTelemetry.Instrumentation.Cassandra
+        - OpenTelemetry.Instrumentation.ConfluentKafka
         - OpenTelemetry.Instrumentation.ElasticsearchClient
         - OpenTelemetry.Instrumentation.EntityFrameworkCore
         - OpenTelemetry.Instrumentation.EventCounters

build/Common.props

@@ -43,6 +43,7 @@
     <OpenTelemetryCoreLatestVersion>[1.9.0,2.0)</OpenTelemetryCoreLatestVersion>
     <OpenTelemetryCoreLatestPrereleaseVersion>[1.9.0-rc.1]</OpenTelemetryCoreLatestPrereleaseVersion>
     <StackExchangeRedisPkgVer>[2.1.58,3.0)</StackExchangeRedisPkgVer>
+    <ConfluentKafkaPkgVer>[2.3.0,3.0)</ConfluentKafkaPkgVer>
     <CassandraCSharpDriverPkgVer>[3.16.0,4.0)</CassandraCSharpDriverPkgVer>
     <StyleCopAnalyzersPkgVer>[1.2.0-beta.507,2.0)</StyleCopAnalyzersPkgVer>
     <SystemNetHttp>[4.3.4,)</SystemNetHttp>

build/Projects/OpenTelemetry.Instrumentation.ConfluentKafka.proj

@@ -0,0 +1,34 @@
+<Project>
+
+  <PropertyGroup>
+    <RepoRoot>$([System.IO.Directory]::GetParent($(MSBuildThisFileDirectory)).Parent.Parent.FullName)</RepoRoot>
+    <MinVerTagPrefix>Instrumentation.ConfluentKafka-</MinVerTagPrefix>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <SolutionProjects Include="$(RepoRoot)\src\OpenTelemetry.Instrumentation.ConfluentKafka\OpenTelemetry.Instrumentation.ConfluentKafka.csproj" />
+    <SolutionProjects Include="$(RepoRoot)\test\OpenTelemetry.Instrumentation.ConfluentKafka.Tests\OpenTelemetry.Instrumentation.ConfluentKafka.Tests.csproj" />
+    <SolutionProjects Include="$(RepoRoot)\examples\kafka\**\*.csproj" />
+
+    <PackProjects Include="$(RepoRoot)\src\OpenTelemetry.Instrumentation.ConfluentKafka\OpenTelemetry.Instrumentation.ConfluentKafka.csproj" />
+
+    <TestProjects Include="$(RepoRoot)\test\OpenTelemetry.Instrumentation.ConfluentKafka.Tests\OpenTelemetry.Instrumentation.ConfluentKafka.Tests.csproj" />
+  </ItemGroup>
+
+  <Target Name="Build">
+    <MSBuild Projects="@(SolutionProjects)" Targets="Build" ContinueOnError="ErrorAndStop" />
+  </Target>
+
+  <Target Name="Restore">
+    <MSBuild Projects="@(SolutionProjects)" Targets="Restore" ContinueOnError="ErrorAndStop" />
+  </Target>
+
+  <Target Name="Pack">
+    <MSBuild Projects="@(PackProjects)" Targets="Pack" ContinueOnError="ErrorAndStop" />
+  </Target>
+
+  <Target Name="VSTest">
+    <MSBuild Projects="@(TestProjects)" Targets="VSTest" ContinueOnError="ErrorAndStop" />
+  </Target>
+
+</Project>

examples/kafka/Constants.cs

@@ -0,0 +1,9 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+namespace Examples.ConfluentKafka;
+
+public static class Constants
+{
+    public static readonly string Topic = $"test-topic-{Guid.NewGuid()}";
+}

examples/kafka/Examples.ConfluentKafka.csproj

@@ -0,0 +1,17 @@
+<Project Sdk="Microsoft.NET.Sdk.Worker">
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net8.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+  <ItemGroup>
+    <ProjectReference Include="$(RepoRoot)\src\OpenTelemetry.Instrumentation.ConfluentKafka\OpenTelemetry.Instrumentation.ConfluentKafka.csproj" />
+  </ItemGroup>
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" />
+    <PackageReference Include="OpenTelemetry.Extensions.Hosting" Version="$(OpenTelemetryCoreLatestVersion)" />
+    <PackageReference Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="$(OpenTelemetryCoreLatestVersion)" />
+    <PackageReference Include="OpenTelemetry.Exporter.Console" Version="$(OpenTelemetryCoreLatestVersion)" />
+  </ItemGroup>
+</Project>

examples/kafka/ProduceConsumeHostedService.cs

@@ -0,0 +1,53 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+using Confluent.Kafka;
+using OpenTelemetry.Instrumentation.ConfluentKafka;
+
+namespace Examples.ConfluentKafka;
+
+public class ProduceConsumeHostedService(
+    InstrumentedProducerBuilder<string, string> instrumentedProducerBuilder,
+    InstrumentedConsumerBuilder<string, string> instrumentedConsumerBuilder)
+    : BackgroundService
+{
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        IProducer<string, string> producer = instrumentedProducerBuilder.Build();
+        IConsumer<string, string> consumer = instrumentedConsumerBuilder.Build();
+
+        for (int j = 0; j < 100; j++)
+        {
+            await producer.ProduceAsync(
+                Constants.Topic,
+                new Message<string, string> { Key = "any_key", Value = $"any_value_{j}" },
+                stoppingToken);
+        }
+
+        for (int j = 0; j < 100; j++)
+        {
+            producer.Produce(
+                Constants.Topic,
+                new Message<string, string> { Key = "any_key", Value = $"any_value_{j}" });
+        }
+
+        producer.Flush(stoppingToken);
+
+        consumer.Subscribe(Constants.Topic);
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            ConsumeResult<string, string> consumeResult = consumer.Consume(stoppingToken);
+            if (consumeResult == null)
+            {
+                continue;
+            }
+
+            if (consumeResult.IsPartitionEOF)
+            {
+                break;
+            }
+
+            Console.WriteLine($"Consumer {consumer.Name} received message: {consumeResult.Message.Value}");
+        }
+    }
+}

examples/kafka/Program.cs

@@ -0,0 +1,50 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+using Confluent.Kafka;
+using Examples.ConfluentKafka;
+using OpenTelemetry.Instrumentation.ConfluentKafka;
+using OpenTelemetry.Metrics;
+using OpenTelemetry.Trace;
+
+var builder = Host.CreateApplicationBuilder(args);
+
+const string bootstrapServers = "localhost:9092";
+
+builder.Services.AddSingleton(_ =>
+{
+    ProducerConfig producerConfig = new() { BootstrapServers = bootstrapServers };
+    return new InstrumentedProducerBuilder<string, string>(producerConfig);
+});
+builder.Services.AddSingleton(_ =>
+{
+    ConsumerConfig consumerConfigA = new()
+    {
+        BootstrapServers = bootstrapServers,
+        GroupId = "group-a",
+        AutoOffsetReset = AutoOffsetReset.Earliest,
+        EnablePartitionEof = true,
+    };
+    return new InstrumentedConsumerBuilder<string, string>(consumerConfigA);
+});
+
+builder.Services.AddOpenTelemetry()
+    .WithTracing(tracing =>
+    {
+        tracing.AddConsoleExporter()
+            .AddOtlpExporter()
+            .AddKafkaProducerInstrumentation<string, string>()
+            .AddKafkaConsumerInstrumentation<string, string>();
+    })
+    .WithMetrics(metering =>
+    {
+        metering.AddConsoleExporter()
+            .AddOtlpExporter()
+            .AddKafkaProducerInstrumentation<string, string>()
+            .AddKafkaConsumerInstrumentation<string, string>();
+    });
+
+builder.Services.AddHostedService<ProduceConsumeHostedService>();
+
+var app = builder.Build();
+await app.RunAsync();

examples/kafka/README.md

@@ -0,0 +1,13 @@
+# Run Examples.ConfluentKafka
+
+Start the Confluent Kafka stack:
+
+```cmd
+docker run -d --name kafka -p 9092:9092 confluentinc/confluent-local
+```
+
+Start the Aspire Dashboard:
+
+```cmd
+docker run --rm -it -p 18888:18888 -p 4317:18889 -d --name aspire-dashboard mcr.microsoft.com/dotnet/nightly/aspire-dashboard:8.0.0
+```

opentelemetry-dotnet-contrib.sln

@@ -287,9 +287,13 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Projects", "Projects", "{04
 	ProjectSection(SolutionItems) = preProject
 		build\Projects\Component.proj = build\Projects\Component.proj
 		build\Projects\OpenTelemetry.Contrib.Shared.Tests.proj = build\Projects\OpenTelemetry.Contrib.Shared.Tests.proj
+		build\Projects\OpenTelemetry.Exporter.InfluxDB.proj = build\Projects\OpenTelemetry.Exporter.InfluxDB.proj
+		build\Projects\OpenTelemetry.Extensions.Enrichment.proj = build\Projects\OpenTelemetry.Extensions.Enrichment.proj
 		build\Projects\OpenTelemetry.Instrumentation.AspNet.proj = build\Projects\OpenTelemetry.Instrumentation.AspNet.proj
 		build\Projects\OpenTelemetry.Instrumentation.AspNetCore.proj = build\Projects\OpenTelemetry.Instrumentation.AspNetCore.proj
+		build\Projects\OpenTelemetry.Instrumentation.ConfluentKafka.proj = build\Projects\OpenTelemetry.Instrumentation.ConfluentKafka.proj
 		build\Projects\OpenTelemetry.Instrumentation.EventCounters.proj = build\Projects\OpenTelemetry.Instrumentation.EventCounters.proj
+		build\Projects\OpenTelemetry.Instrumentation.GrpcCore.proj = build\Projects\OpenTelemetry.Instrumentation.GrpcCore.proj
 		build\Projects\OpenTelemetry.Instrumentation.Owin.proj = build\Projects\OpenTelemetry.Instrumentation.Owin.proj
 		build\Projects\OpenTelemetry.Instrumentation.Process.proj = build\Projects\OpenTelemetry.Instrumentation.Process.proj
 		build\Projects\OpenTelemetry.Instrumentation.Runtime.proj = build\Projects\OpenTelemetry.Instrumentation.Runtime.proj
@@ -377,6 +381,14 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "ISSUE_TEMPLATE", "ISSUE_TEM
 		.github\ISSUE_TEMPLATE\feature_request.yml = .github\ISSUE_TEMPLATE\feature_request.yml
 	EndProjectSection
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "OpenTelemetry.Instrumentation.ConfluentKafka", "src\OpenTelemetry.Instrumentation.ConfluentKafka\OpenTelemetry.Instrumentation.ConfluentKafka.csproj", "{96341E23-990E-4144-A7E3-9EF0DAFF3232}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "OpenTelemetry.Instrumentation.ConfluentKafka.Tests", "test\OpenTelemetry.Instrumentation.ConfluentKafka.Tests\OpenTelemetry.Instrumentation.ConfluentKafka.Tests.csproj", "{BE40900A-2859-471D-8802-21DFD73DDAA7}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "kafka", "kafka", "{3A464E7A-42F3-44B0-B8D7-80521A7704A6}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Examples.ConfluentKafka", "examples\kafka\Examples.ConfluentKafka.csproj", "{9B994669-E839-4C42-A0F1-DF9DD058C1DC}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -779,6 +791,18 @@ Global
 		{B13394D6-D3D7-453E-B91A-24C199F41C5E}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{B13394D6-D3D7-453E-B91A-24C199F41C5E}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{B13394D6-D3D7-453E-B91A-24C199F41C5E}.Release|Any CPU.Build.0 = Release|Any CPU
+		{96341E23-990E-4144-A7E3-9EF0DAFF3232}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{96341E23-990E-4144-A7E3-9EF0DAFF3232}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{96341E23-990E-4144-A7E3-9EF0DAFF3232}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{96341E23-990E-4144-A7E3-9EF0DAFF3232}.Release|Any CPU.Build.0 = Release|Any CPU
+		{BE40900A-2859-471D-8802-21DFD73DDAA7}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{BE40900A-2859-471D-8802-21DFD73DDAA7}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{BE40900A-2859-471D-8802-21DFD73DDAA7}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{BE40900A-2859-471D-8802-21DFD73DDAA7}.Release|Any CPU.Build.0 = Release|Any CPU
+		{9B994669-E839-4C42-A0F1-DF9DD058C1DC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{9B994669-E839-4C42-A0F1-DF9DD058C1DC}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{9B994669-E839-4C42-A0F1-DF9DD058C1DC}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{9B994669-E839-4C42-A0F1-DF9DD058C1DC}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -899,6 +923,10 @@ Global
 		{70CA77D4-5D7F-4D70-A6B5-8AAC07A8EA3C} = {2097345F-4DD3-477D-BC54-A922F9B2B402}
 		{45D29DAA-0DB9-4808-B879-1AECC37EF366} = {824BD1DE-3FA8-4FE0-823A-FD365EAC78AF}
 		{40373C78-0513-4067-A96B-96A851369761} = {1A06E14B-DD2F-4536-9D2E-F708C0C43555}
+		{96341E23-990E-4144-A7E3-9EF0DAFF3232} = {22DF5DC0-1290-4E83-A9D8-6BB7DE3B3E63}
+		{BE40900A-2859-471D-8802-21DFD73DDAA7} = {2097345F-4DD3-477D-BC54-A922F9B2B402}
+		{3A464E7A-42F3-44B0-B8D7-80521A7704A6} = {B75EE478-97F7-4E9F-9A5A-DB3D0988EDEA}
+		{9B994669-E839-4C42-A0F1-DF9DD058C1DC} = {3A464E7A-42F3-44B0-B8D7-80521A7704A6}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {B0816796-CDB3-47D7-8C3C-946434DE3B66}