--- title: "OTLP経由のKafkaクライアントメトリクス — パズルのもう1ピース" date: 2026-07-07T00:00:00.000Z author: "grzegorz" excerpt: "KIP-714により、KafkaクライアントはOTLP経由で自身のメトリクスをブローカーへプッシュできます。monedula-metrics-reporterは、ブローカーSPIメトリクスやYammerメトリクスに加えて、これらのメトリクスも転送するようになりました。" --- [前回の記事](/blog/kafka-metrics-opentelemetry-otlp-monedula-metrics-reporter/)では、Kafkaブローカーメトリクスと、それらの収集が本来あるべき姿よりも難しい理由について書きました。 要点は次のとおりでした。Kafkaにはメトリクスシステムが1つだけあるわけではない、ということです。 Kafkaには、`metric.reporters`で設定する新しいKafka Metrics SPIと、`kafka.metrics.reporters`で設定する古いYammer/Coda Haleレジストリがあります。重要なブローカーメトリクスの一部は一方のシステムに、別の一部はもう一方のシステムに存在します。そして現実の監視構成の大半は、今なおJMX、正規表現、Prometheusマッピング、そしてエクスポートされた名前と一致しているかどうか怪しいダッシュボードを経由しています。 これが`monedula-metrics-reporter`を作った理由でした。これは、OTLP経由でメトリクスを直接OpenTelemetry Collectorへエクスポートし、両方のKafkaメトリクスレジストリを1つの設定で扱うKafkaプラグインです。 しかし、まだ足りないパーツが1つありました。 ブローカーメトリクスはクラスターについて多くのことを教えてくれますが、クライアントアプリケーションの内部で何が起きているのかを完全には説明してくれません。 ブローカーは、Produceリクエストが遅いことを教えてくれます。Fetchリクエストが遅いことも教えてくれます。リクエストキュー、ネットワークプロセッサー、under-replicatedパーティション、失敗したリクエスト、その他多くのことについても教えてくれます。 しかし、本当に知りたいのは次のことだけではない場合があります。 > ブローカーで何が起きているのか? それはまた、次のことでもあります。 > プロデューサーやコンシューマーの内部で何が起きているのか? プロデューサーは効率的にバッチ処理しているか? スロットリングされていないか? コンシューマーは定期的にポーリングしているか? リバランスに時間を費やしていないか? あるアプリケーションが別のアプリケーションと異なる振る舞いをしていないか? 従来、これらの問いに答えるには、すべてのクライアントアプリケーションを個別に計装する必要がありました。Javaでは、それはたいてい再びJMXを意味しました。他の言語では、クライアントライブラリ、フレームワーク、そしてチームが用意していた可観測性の構成に依存していました。 そして大規模な組織では、これは非常によくある問題になります。Kafkaチームはクラスターを所有していますが、クライアントアプリケーションは多くの異なるチームによって所有されています。そのため何か問題が起きたとき、クラスター運用者は必要なクライアント側メトリクスを持っていないことが多く、それらを得るには調整、コード変更、再デプロイ、少なくとも再起動が必要になります。 これこそが[KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability)が解決しようとしている問題です。 ## KIP-714の背後にある考え方 KIP-714は、Kafkaクライアントからメトリクスを収集する、ブローカー主導の仕組みを追加します。 各アプリケーションチームがそれぞれ独自のメトリクスエクスポーターを設定する代わりに、Kafkaクラスターがクライアントメトリクスのサブスクリプションを定義できます。サブスクリプションは、どのクライアントメトリクスを収集すべきか、そしてクライアントがそれらをどのくらいの頻度でプッシュすべきかを指定します。 クライアントは、Kafkaプロトコルを使ってメトリクスをブローカーへプッシュします。ペイロード自体はOTLPなので、ブローカー側のプラグインはそれをOpenTelemetry Collector、時系列データベース、あるいはその他の可観測性バックエンドへ転送できます。 重要なのは制御の方向です。 クラスター運用者が、何を収集するかを決定します。 KIP-714をサポートするクライアントは、サブスクリプションを受け取り、要求されたメトリクスをプッシュできます。アプリケーションはJMXを公開する必要がありません。サイドカーエクスポーターも必要ありません。カスタムのPrometheusエンドポイントも必要ありません。 それでも、2つの条件があります。 1. ブローカーにクライアントテレメトリプラグインが設定されていること。 2. 少なくとも1つのクライアントメトリクスサブスクリプションが存在すること。 サブスクリプションがなければ、クライアントがプロトコルをサポートしていても、メトリクスをプッシュしません。 これは優れた設計です。クライアントメトリクスは偶然に収集されることはありません。運用者がクラスター側で明示的に収集を有効化しなければなりません。 ## 重要な点: すべてのクライアントが同じメトリクスを送るわけではない KIP-714はクライアントメトリクスを収集するための共通の仕組みを作りますが、すべてのKafkaクライアントが魔法のように同じメトリクスセットを公開するようになるわけではありません。 得られるメトリクスは、クライアントライブラリに依存します。 現時点ではJavaクライアントが最も豊富なメトリクスセットを送ります。これは驚くことではありません。Javaクライアントはリファレンス実装であり、非常に成熟した内部メトリクスレジストリを持っており、多くのKafkaメトリクスがまずそこで設計されたからです。 他のクライアントは、より小さなセットを公開するかもしれません。Python、Go、.NET、Nodeのラッパーで使われるクライアントなど、librdkafkaベースのクライアントもKIP-714に参加できますが、利用可能なメトリクスがJavaクライアントの送るものと必ずしも同一とは限りません。 したがって、心構えとしては次のようになります。 > KIP-714は、クライアントメトリクスがどのように要求され、転送されるかを標準化する。 > 実際のメトリクスセットは、依然としてクライアント実装に依存する。 これはダッシュボードを構築する際に重要です。 一部のパネルは共通メトリクスに基づくことができ、異なるクライアントライブラリ間で機能するはずです。他のパネルは、クライアントが実際により豊富なメトリクスセットを送る場合にのみ意味を持ちます。実際のところ、今日最も完全な可観測性のストーリーを提供するのはJavaクライアントです。 ## なぜこれをmonedula-metrics-reporterに追加するのか? 当初、`monedula-metrics-reporter`はブローカーメトリクスに関するものでした。 Kafka SPIレジストリを扱い、 Yammerレジストリを扱い、 両方をOTLP経由でエクスポートし、 リソース属性としてブローカーのアイデンティティを追加し、 エミットされるメトリクスに合致するGrafanaダッシュボードを同梱していました。 しかしKIP-714が存在する以上、ごく自然な次のステップがあります。 > プラグインがすでにブローカー内で動作し、すでにOTLPをコレクターへエクスポートしているのなら、クライアントがプッシュするメトリクスも扱うべきだ。 そうでなければ、Kafkaの監視は依然として分断されたままになります。 ブローカーSPIメトリクス用の経路、Yammerメトリクス用の別の経路、そしてKIP-714クライアントテレメトリ用のさらに別のプラグインやコンポーネント、というふうにです。 それでも機能はしますが、Kafkaメトリクススタックを簡素化するという趣旨そのものを台無しにしてしまいます。 そこで新しいバージョンの`monedula-metrics-reporter`は、3つのメトリクスソースを扱うようになりました。 1. Kafka Metrics SPI 2. Yammer / Coda Haleブローカーメトリクス 3. KIP-714でクライアントがプッシュするテレメトリ 目標はシンプルです。1つのプラグイン、1つのコレクターパイプライン、1組のダッシュボードです。 クライアントメトリクスは、ブローカーメトリクスと同じOpenTelemetryの経路を通じて転送されます。それらは`kafka_cluster_id`や`kafka_node_id`といった同じブローカーラベルを持つことができ、さらに`client_id`で拡張することもできるため、異なるアプリケーションからのメトリクスが1つの匿名のストリームに混ざり合ってしまうことはありません。 この最後の点は非常に重要です。 2つのプロデューサーが同じメトリクス名をプッシュした場合、どちらのプロデューサーがどちらなのかを知る必要があります。デフォルトでは、レポーターはclient idをラベルとして追加します。client instance id、principal、client addressといった、よりセンシティブあるいは高カーディナリティなラベルはオプトインです。 ## 仕組み ブローカー側の部分は、Kafkaの`ClientTelemetry`インターフェースを通じて実装されています。 KIP-714対応のクライアントがブローカーへテレメトリをプッシュすると、プラグインはOTLPペイロードを受け取ります。リクエストハンドラースレッドが行うのは最小限の作業だけです。ペイロードを、上限のあるインメモリキューにコピーするだけです。その後、デーモンスレッドがそのキューから読み取り、ペイロードをOpenTelemetry SDKのメトリクスデータへ変換し、ラベルで拡張し、専用のメトリクスエクスポーターを使ってエクスポートします。 これはレポーターの他の部分と同じ設計原則です。 > Kafkaが可観測性パイプラインによってブロックされることは決してあってはならない。 コレクターが利用できない、遅い、あるいは誤設定されている場合でも、Kafkaはトラフィックの処理を続けるべきです。テレメトリのバッチは破棄されて構いません。Kafkaのレイテンシが監視バックエンドに依存すべきではありません。 クライアントテレメトリレシーバーには、独自の自己監視メトリクスもあります。 ```text monedula_reporter_clienttelemetry_received_total monedula_reporter_clienttelemetry_forwarded_total monedula_reporter_clienttelemetry_dropped_total monedula_reporter_clienttelemetry_unsupported_metrics_dropped_total monedula_reporter_clienttelemetry_queue_depth ```` そのため、クライアントテレメトリの流れが止まった場合、当てずっぽうで推測する前に確認できるものがあります。 ## インストール このアーティファクトはMaven Centralに公開されているため、手動でビルドする必要はありません。 バージョン`0.10.0`の場合、Maven CentralからJARをダウンロードします。 ```bash curl -L \ -o monedula-metrics-reporter-0.10.0.jar \ https://repo1.maven.org/maven2/dev/monedula/monedula-metrics-reporter/0.10.0/monedula-metrics-reporter-0.10.0.jar ``` 次に、それをKafkaのclasspathへコピーします。 ```bash cp monedula-metrics-reporter-0.10.0.jar $KAFKA_HOME/libs/ ``` その後、ブローカーを設定します。 ```properties metric.reporters=dev.monedula.metricsreporter.OtlpMetricReporter otlp.metric.reporter.endpoint=http://otel-collector:4317 otlp.metric.reporter.transport=grpc ``` これにより、Kafka SPIメトリクス、Yammerブローカーメトリクス、そしてクライアントテレメトリに対してレポーターが有効になります。 クライアントテレメトリは、ブローカー上でデフォルトで有効です。 ```properties otlp.metric.reporter.client.telemetry.enabled=true ``` 明示的に無効化することもできます。 ```properties otlp.metric.reporter.client.telemetry.enabled=false ``` 無効化すると、ブローカーはテレメトリ機能をクライアントへ通知しなくなります。 ## Collectorの設定 最小限のOpenTelemetry Collector設定で、レポーターからのOTLPを受信し、Prometheus向けにメトリクスを公開できます。 ```yaml receivers: otlp: protocols: grpc: endpoint: "0.0.0.0:4317" http: endpoint: "0.0.0.0:4318" exporters: prometheus: endpoint: "0.0.0.0:8889" resource_to_telemetry_conversion: enabled: true service: pipelines: metrics: receivers: [otlp] exporters: [prometheus] ``` `resource_to_telemetry_conversion`の設定は、Prometheusを使用する際に重要です。これはOpenTelemetryのリソース属性をPrometheusのラベルへと昇格させます。 これがないと、`kafka_cluster_id`、`kafka_node_id`、`client_id`といったラベルが、期待した場所に現れないことがあります。 その後、PrometheusがCollectorをスクレイプできます。 ```yaml scrape_configs: - job_name: otel-collector static_configs: - targets: ["otel-collector:8889"] ``` ## 重要な部分: クライアントメトリクスサブスクリプション プラグインをインストールするだけでは不十分です。 これはKIP-714で見落としやすい部分です。 クライアントは、クラスター上にクライアントメトリクスサブスクリプションが設定されている場合にのみ、メトリクスをプッシュします。 例えば次のようにします。 ```bash kafka-client-metrics.sh \ --bootstrap-server broker:9092 \ --alter \ --name all-clients \ --metrics "*" \ --interval 5000 ``` これは`all-clients`という名前のサブスクリプションを作成または更新し、利用可能なすべてのクライアントメトリクスを要求し、プッシュ間隔を5秒に設定します。 本番環境での利用では、すべてを収集するのは過剰かもしれません。選択したメトリクスのプレフィックスだけをサブスクライブすることもできます。 例えば、標準のKafkaクライアントメトリクスから始めることができます。 ```bash kafka-client-metrics.sh \ --bootstrap-server broker:9092 \ --alter \ --name standard-client-metrics \ --metrics "org.apache.kafka." \ --interval 10000 ``` 正確なサブスクリプション戦略は、何を観測したいか、そしてどれだけのメトリクス量を許容できるかに依存します。 重要なのは、サブスクリプションがブローカー側に存在することです。これにより、Kafka運用者は、各アプリケーションチームに監視構成の変更を求めることなく、何を収集すべきかを決定できます。 ## クライアント側の設定はどうするのか? KIP-714をサポートするクライアントでは、メトリクスのプッシュは通常デフォルトで有効になっています。 それでも、明示的に指定することができます。 ```properties enable.metrics.push=true ``` Javaクライアントの場合、これはKafkaクライアント設定の一部です。 `confluent-kafka-python`のようなlibrdkafkaベースのクライアントでも、同じ考え方が当てはまります。例えば次のようにします。 ```python Producer({ "bootstrap.servers": "broker:9092", "client.id": "orders-producer", "enable.metrics.push": True, }) ``` そしてコンシューマーの場合は次のようにします。 ```python Consumer({ "bootstrap.servers": "broker:9092", "group.id": "orders-consumer-group", "client.id": "orders-consumer", "enable.metrics.push": True, }) ``` `client.id`は意図的に設定する価値があります。 レポーターはデフォルトでこれを転送されるクライアントメトリクスに追加するため、ダッシュボードはメトリクスをアプリケーションごとに区別できます。意味のあるclient idがなければ、メトリクスは得られるものの、トラブルシューティング時に格段に使いづらくなります。 ## 拡張オプション デフォルトでは、転送されるクライアントメトリクスはブローカーのアイデンティティとclient idで拡張されます。 ```properties otlp.metric.reporter.client.telemetry.enrich.broker=true otlp.metric.reporter.client.telemetry.enrich.client.id=true ``` これにより、クライアントメトリクスをクラスター、ブローカー、アプリケーションごとにグループ化できます。 オプションの拡張もあります。 ```properties otlp.metric.reporter.client.telemetry.enrich.client.instance.id=true otlp.metric.reporter.client.telemetry.enrich.client.identity=true ``` client instance idは、個々の実行中のクライアントインスタンスを区別したい場合に有用ですが、高カーディナリティです。 client identityはprincipalとclient addressを追加します。これは管理された環境では非常に有用ですが、センシティブな情報を含む可能性もあるため、デフォルトでは無効になっています。 要するに、まずはブローカーラベルとclient idから始めましょう。それ以外は、本当に必要になったときにのみ有効化してください。 ## クイックスタート リポジトリには、Kafka、OpenTelemetry Collector、Prometheus、Grafanaを含むクイックスタートスタックが含まれています。 クイックスタートを起動します。 ```bash cd quickstart docker compose up -d ``` デモサービスはKIP-714クライアントメトリクスサブスクリプションを作成し、プロデューサーとコンシューマーのトラフィックを開始します。そのため、スタックが起動すると、ブローカーダッシュボードとクライアントダッシュボードの両方がデータを受信し始めるはずです。 Grafanaは次のアドレスで利用できます。 ```text http://localhost:3000 ``` Prometheusは次のアドレスで利用できます。 ```text http://localhost:9090 ``` クイックスタートは、次のようにしてサブスクリプションを作成します。 ```bash kafka-client-metrics.sh \ --bootstrap-server kafka1:9092 \ --alter \ --name quickstart-clients \ --metrics "*" \ --interval 5000 ``` これは意図的にシンプルにしています。すべてを収集し、5秒ごとにプッシュし、ダッシュボードを素早く動き出させるためです。 実際の環境では、私は通常もっと慎重に始めます。まず標準メトリクスを収集し、その量を検証し、それから必要に応じてメトリクスを追加します。 ## Javaクライアントと他のクライアント KIP-714の興味深い点の1つは、共通の仕組みを作り出す一方で、すべてのクライアントライブラリが同じ深さのメトリクスを公開するわけではない、ということです。 Javaクライアントは最も豊富なメトリクスセットを送ります。内部レジストリから多くのメトリクスをプッシュでき、その中には、バッチ処理、バッファ使用量、リクエストのタイミング、トピックレベルの内訳、コンシューマーのlagとlead、その他のJavaクライアント固有のシグナルといった、より豊富なプロデューサーとコンシューマーの詳細が含まれます。 他のクライアントライブラリは、より小さなセットを送るかもしれません。 だからといって、KIP-714がJavaクライアントにのみ有用というわけではありません。サポートするすべてのクライアントに対して、共通のブローカー主導の収集経路を持てることは、依然として非常に有用です。ただしダッシュボードを見るときには、パネルが欠けているからといって、必ずしも何かが壊れていることを意味するわけではない、と覚えておくべきです。 そのメトリクスが、このクライアントライブラリに単に存在しないだけの場合もあります。 このため、新しいGrafanaダッシュボードは2つの部分に分かれています。 1. 標準パネル 2. 拡張パネル 標準パネルは、関連する共通メトリクスを公開するすべてのKIP-714準拠クライアントで機能するはずです。 拡張パネルは、主にJavaクライアントで有用です。なぜなら、Javaクライアントが現時点で最も豊富なメトリクスセットを提供するからです。 この分割は意図的なものです。非JavaクライアントがJava固有のメトリクスを公開しないというだけで、ダッシュボードが壊れているように見えるのは避けたいのです。境界を可視化するほうが望ましいのです。 ## 新しいダッシュボード 新しいバージョンでは、クライアントに焦点を当てた2つのダッシュボードが追加されます。 * Kafka Producers (client) * Kafka Consumers (client) プロデューサーダッシュボードは、レコードを送信するクライアント側の視点に焦点を当てます。リクエストレイテンシ、スロットリング、キューイング、バッチ処理、エラー、スループットです。 コンシューマーダッシュボードは、レコードを消費するクライアント側の視点に焦点を当てます。fetchレイテンシ、コミットレート、ポーリングの振る舞い、リバランス関連のシグナル、アサインメントの変化、そして利用可能な場合はコンシューマー側のlag/leadです。 どちらのダッシュボードにも、クラスターとclient idのフィルターが含まれます。メトリクスセットがサポートしている場合には、トピックまたはグループレベルのドリルダウンも含まれます。
プロデューサーダッシュボード - クライアント側。
プロデューサーダッシュボード - クライアント側
コンシューマーダッシュボード - クライアント側。
コンシューマーダッシュボード - クライアント側
この構成について私が本当に気に入っている点が1つあります。同じGrafanaフォルダーで、いまや両側のストーリーを見せられるようになったことです。 ブローカーダッシュボードは次の問いに答えます。 > クラスターは健全か? クライアントダッシュボードは次の問いに答えます。 > アプリケーションはクラスターをどのように体験しているか? そして、どちらも一貫したラベルを持つ同じOTLPパイプラインから来ているため、それらを相関させることが格段に容易になります。 ## まとめ KIP-714は、Kafkaの可観測性にとって非常に重要な一歩です。 これはクライアントメトリクスを、各アプリケーションチームが独立して公開しなければならないものから、Kafkaクラスターが中央で要求できるものへと変えます。 `monedula-metrics-reporter`はこのモデルをサポートするようになりました。 依然として両方のKafkaメトリクスレジストリからブローカーメトリクスをエクスポートします。しかし今では、KIP-714クライアントからクライアントがプッシュするOTLPテレメトリも受信し、それを同じOpenTelemetry Collectorへ転送します。 そのため、プラグインは今や3つのソースをカバーします。 * Kafka SPIメトリクス * Yammerブローカーメトリクス * KIP-714クライアントテレメトリ その結果、よりシンプルなKafka監視アーキテクチャが実現します。ブローカー上の1つのプラグイン、1つのOTLPパイプライン、そしてブローカー側とクライアント側の両方の振る舞いをカバーするダッシュボードです。 このコードはオープンソースで、[GitHubで公開されています](https://github.com/monedula-dev/monedula-metrics-reporter)