如何从Vert.x Kafka生产者客户端获取Kafka指标并导出Prometheus数据

我刚接触Vert.x，对Kafka也相对陌生，需要获取kafka_producer_buffer_total_bytes等常用的Kafka客户端指标。

已完成的基础配置 #

目前已经完成Vert.x自带Prometheus指标的启用，可以正常查看vertx_http_server_request_bytes_max这类指标：

// 简化代码
import io.vertx.kafka.client.producer.KafkaProducer;
// ...
Vertx vertx = Vertx.vertx(new VertxOptions().setMetricsOptions(
    new MicrometerMetricsOptions()
        .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true))
        .setEnabled(true)));
// ...
// 配置HttpServer并添加指标路由
Router.router(vertx).route("/metrics").handler(PrometheusScrapingHandler.create());
// ...
// 创建io.vertx.kafka.client.producer.KafkaProducer实例
producer = KafkaProducer.create(vertx, config);

原生Apache KafkaProducer的指标绑定逻辑 #

在使用Vert.x之前，使用原生KafkaProducer时可以通过如下方式绑定KafkaClientMetrics实现指标采集：

import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
import org.apache.kafka.clients.producer.KafkaProducer;
// ...
// 生产者创建完成后执行
new KafkaClientMetrics(producer).bindTo(BackendRegistries.getDefaultNow());

最终实现方案 #

参考Jonatan关于创建SharedProducer的代码建议，最终通过如下方案实现了Vert.x KafkaProducer的Kafka客户端指标采集：

KafkaWriteStream<String, String> stream = KafkaWriteStream.create(vertx, config);

// 拆包获取org.apache.kafka.clients.producer.Producer实例
Producer<String, String> apacheProducer = stream.unwrap();

// 创建io.vertx.kafka.client.producer.KafkaProducer实例
KafkaProducer vertxProducer = (new KafkaProducerImpl(vertx, stream)).registerCloseHook();

// 配置指标采集
MeterRegistry registry = BackendRegistries.getDefaultNow();
new KafkaClientMetrics(apacheProducer).bindTo(registry);

目前已经通过调用vertxProducer.send()并监控Prometheus指标完成了功能验证。

内容的提问来源于stack exchange，提问作者Gavin