-
Notifications
You must be signed in to change notification settings - Fork 305
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add Producer trace support Add Consumer trace
- Loading branch information
Showing
42 changed files
with
2,903 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
33 changes: 33 additions & 0 deletions
33
build/Projects/OpenTelemetry.Instrumentation.ConfluentKafka.proj
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| <Project> | ||
|
|
||
| <PropertyGroup> | ||
| <RepoRoot>$([System.IO.Directory]::GetParent($(MSBuildThisFileDirectory)).Parent.Parent.FullName)</RepoRoot> | ||
| </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\redis\**\*.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> |
15 changes: 15 additions & 0 deletions
15
examples/Examples.ConfluentKafka/Examples.ConfluentKafka.csproj
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
| <PropertyGroup> | ||
| <OutputType>Exe</OutputType> | ||
| <TargetFramework>net6.0</TargetFramework> | ||
| <ImplicitUsings>enable</ImplicitUsings> | ||
| <Nullable>enable</Nullable> | ||
| </PropertyGroup> | ||
| <ItemGroup> | ||
| <ProjectReference Include="..\..\src\OpenTelemetry.Instrumentation.ConfluentKafka\OpenTelemetry.Instrumentation.ConfluentKafka.csproj" /> | ||
| </ItemGroup> | ||
| <ItemGroup> | ||
| <PackageReference Include="OpenTelemetry.Exporter.Zipkin" Version="$(OpenTelemetryCoreLatestVersion)" /> | ||
| <PackageReference Include="OpenTelemetry.Exporter.Console" Version="1.3.1" /> | ||
| </ItemGroup> | ||
| </Project> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| using Confluent.Kafka; | ||
| using OpenTelemetry; | ||
| using OpenTelemetry.Instrumentation.ConfluentKafka; | ||
| using OpenTelemetry.Trace; | ||
|
|
||
| ProducerConfig producerConfig = new(); | ||
| producerConfig.BootstrapServers = "localhost:9092"; | ||
| InstrumentedProducerBuilder<string, string> producerBuilder = new(producerConfig); | ||
|
|
||
| ConsumerConfig consumerConfigA = new(); | ||
| consumerConfigA.BootstrapServers = "localhost:9092"; | ||
| consumerConfigA.GroupId = "group-a"; | ||
| consumerConfigA.AutoOffsetReset = AutoOffsetReset.Earliest; | ||
| consumerConfigA.EnablePartitionEof = true; | ||
| InstrumentedConsumerBuilder<string, string> consumerBuilderA = new(consumerConfigA); | ||
|
|
||
| ConsumerConfig consumerConfigB = new(); | ||
| consumerConfigB.BootstrapServers = "localhost:9092"; | ||
| consumerConfigB.GroupId = "group-b"; | ||
| consumerConfigB.AutoOffsetReset = AutoOffsetReset.Earliest; | ||
| consumerConfigB.EnablePartitionEof = true; | ||
| InstrumentedConsumerBuilder<string, string> consumerBuilderB = new(consumerConfigB); | ||
|
|
||
| string topic = $"test-topic-{Guid.NewGuid()}"; | ||
|
|
||
| // Configure exporter to export traces to Zipkin | ||
| using var tracerProvider = Sdk.CreateTracerProviderBuilder() | ||
| .AddConsoleExporter() | ||
| .AddZipkinExporter() | ||
| .AddKafkaProducerInstrumentation(producerBuilder) | ||
| .AddKafkaConsumerInstrumentation("consumer-a", consumerBuilderA) | ||
| .AddKafkaConsumerInstrumentation("consumer-b", consumerBuilderB) | ||
| .Build(); | ||
|
|
||
| // produce a message to the topic "test-topic" with key "any_key" and value "any_value". | ||
| IProducer<string, string> producer = producerBuilder.Build(); | ||
| await producer.ProduceAsync(topic, new Message<string, string> | ||
| { | ||
| Key = "any_key", | ||
| Value = "any_value", | ||
| }); | ||
|
|
||
| tracerProvider.ForceFlush(); | ||
|
|
||
| IConsumer<string, string> consumerA = consumerBuilderA.Build(); | ||
| consumerA.Subscribe(topic); | ||
| while (true) | ||
| { | ||
| ConsumeResult<string, string> consumeResult = consumerA.Consume(); | ||
| if (consumeResult == null) | ||
| { | ||
| continue; | ||
| } | ||
|
|
||
| if (consumeResult.IsPartitionEOF) | ||
| { | ||
| break; | ||
| } | ||
|
|
||
| Console.WriteLine($"Consumer A received message: {consumeResult.Message.Value}"); | ||
| } | ||
|
|
||
| consumerA.Close(); | ||
|
|
||
| tracerProvider.ForceFlush(); | ||
|
|
||
| IConsumer<string, string> consumerB = consumerBuilderB.Build(); | ||
| consumerB.Subscribe(topic); | ||
| while (true) | ||
| { | ||
| ConsumeResult<string, string> consumeResult = consumerB.Consume(); | ||
| if (consumeResult == null) | ||
| { | ||
| continue; | ||
| } | ||
|
|
||
| if (consumeResult.IsPartitionEOF) | ||
| { | ||
| break; | ||
| } | ||
|
|
||
| Console.WriteLine($"Consumer B received message: {consumeResult.Message.Value}"); | ||
| } | ||
|
|
||
| consumerB.Close(); | ||
|
|
||
| tracerProvider.ForceFlush(); | ||
|
|
||
| // Wait for the exporter to finish sending | ||
| await Task.Delay(TimeSpan.FromSeconds(5)); | ||
|
|
||
| Console.WriteLine("done!"); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 2 additions & 0 deletions
2
src/OpenTelemetry.Instrumentation.ConfluentKafka/.publicApi/net6.0/PublicAPI.Shipped.txt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,2 @@ | ||
| #nullable enable | ||
|
|
32 changes: 32 additions & 0 deletions
32
src/OpenTelemetry.Instrumentation.ConfluentKafka/.publicApi/net6.0/PublicAPI.Unshipped.txt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions | ||
| OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions.ConfluentKafkaInstrumentationOptions() -> void | ||
| OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue> | ||
| OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue>.InstrumentedConsumerBuilder(System.Collections.Generic.IEnumerable<System.Collections.Generic.KeyValuePair<string!, string!>>! config) -> void | ||
| OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue> | ||
| OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue>.InstrumentedProducerBuilder(System.Collections.Generic.IEnumerable<System.Collections.Generic.KeyValuePair<string!, string!>>! config) -> void | ||
| OpenTelemetry.Metrics.MeterProviderBuilderExtensions | ||
| OpenTelemetry.Trace.TracerProviderBuilderExtensions | ||
| override OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue>.Build() -> Confluent.Kafka.IConsumer<TKey, TValue>! | ||
| override OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue>.Build() -> Confluent.Kafka.IProducer<TKey, TValue>! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue>! producerBuilder) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue>! producerBuilder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>! configure) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder, string? name, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue>? producerBuilder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>? configure) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>! configure) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue>! producerBuilder) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue>! producerBuilder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>! configure) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder, string? name, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue>? producerBuilder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>? configure) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Metrics.MeterProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Metrics.MeterProviderBuilder! builder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>! configure) -> OpenTelemetry.Metrics.MeterProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue>! consumerBuilder) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue>! consumerBuilder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>! configure) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, string? name, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue>! consumerBuilder) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, string? name, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedConsumerBuilder<TKey, TValue>? consumerBuilder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>? configure) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaConsumerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>! configure) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue>! producerBuilder) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue>! producerBuilder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>! configure) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, string? name, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue>! producerBuilder) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, string? name, OpenTelemetry.Instrumentation.ConfluentKafka.InstrumentedProducerBuilder<TKey, TValue>? producerBuilder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>? configure) -> OpenTelemetry.Trace.TracerProviderBuilder! | ||
| static OpenTelemetry.Trace.TracerProviderBuilderExtensions.AddKafkaProducerInstrumentation<TKey, TValue>(this OpenTelemetry.Trace.TracerProviderBuilder! builder, System.Action<OpenTelemetry.Instrumentation.ConfluentKafka.ConfluentKafkaInstrumentationOptions!>! configure) -> OpenTelemetry.Trace.TracerProviderBuilder! |
1 change: 1 addition & 0 deletions
1
src/OpenTelemetry.Instrumentation.ConfluentKafka/.publicApi/net8.0/PublicAPI.Shipped.txt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| #nullable enable |
Oops, something went wrong.