Confluent.Kafka.Extensions.OpenTelemetry 0.4.0

Confluent.Kafka.Extensions.OpenTelemetry

Confluent.Kafka.Extensions.OpenTelemetry软件包启用收集Confluent.Kafka库的仪器数据。应使用Confluent.Kafka.Extensions.Diagnostics来配置Confluent.Kafka库的实际仪器。

安装

Install-Package Confluent.Kafka.Extensions.OpenTelemetry

用法

Confluent.Kafka配置

由于Confluent.Kafka没有公开任何仪器数据，需要额外的配置。完整文档可在Confluent.Kafka.Extensions.Diagnostics 文档中找到。还有一个示例，说明如何在实际应用中使用该软件包。

生产者

using Confluent.Kafka;
using Confluent.Kafka.Extensions.Diagnostics;


using var producer =
    new ProducerBuilder<Null, string>(new ProducerConfig(new ClientConfig { BootstrapServers = "localhost:9092" }))
        .SetKeySerializer(Serializers.Null)
        .SetValueSerializer(Serializers.Utf8)
        .BuildWithInstrumentation();

await producer.ProduceAsync("topic", new Message<Null, string> { Value = "Hello World!" });

消费者

using Confluent.Kafka;
using Confluent.Kafka.Extensions.Diagnostics;

using var consumer = new ConsumerBuilder<Ignore, string>(
        new ConsumerConfig(new ClientConfig { BootstrapServers = "localhost:9092" })
        {
            GroupId = "group", AutoOffsetReset = AutoOffsetReset.Earliest
        })
    .SetValueDeserializer(Deserializers.Utf8)
    .Build();

consumer.Subscribe("topic");

consumer.ConsumeWithInstrumentation((result) =>
{
    Console.WriteLine(result.Message.Value);
});

OpenTelemetry配置

using Confluent.Kafka.Extensions.OpenTelemetry;
using OpenTelemetry.Resources;
using OpenTelemetry.Trace;

builder.Services.AddOpenTelemetry().WithTracing(traceBuilder =>
{
    traceBuilder
        .AddInMemoryExporter()
        .AddHttpClientInstrumentation()
        .AddAspNetCoreInstrumentation()
        .AddConfluentKafkaInstrumentation();  // <-- Add Confluent.Kafka OpenTelemetry support
});