OTLP経由のKafkaクライアントメトリクス — パズルのもう1ピース

前回の記事では、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が解決しようとしている問題です。

KIP-714の背後にある考え方

KIP-714は、Kafkaクライアントからメトリクスを収集する、ブローカー主導の仕組みを追加します。

各アプリケーションチームがそれぞれ独自のメトリクスエクスポーターを設定する代わりに、Kafkaクラスターがクライアントメトリクスのサブスクリプションを定義できます。サブスクリプションは、どのクライアントメトリクスを収集すべきか、そしてクライアントがそれらをどのくらいの頻度でプッシュすべきかを指定します。

クライアントは、Kafkaプロトコルを使ってメトリクスをブローカーへプッシュします。ペイロード自体はOTLPなので、ブローカー側のプラグインはそれをOpenTelemetry Collector、時系列データベース、あるいはその他の可観測性バックエンドへ転送できます。

重要なのは制御の方向です。

クラスター運用者が、何を収集するかを決定します。

KIP-714をサポートするクライアントは、サブスクリプションを受け取り、要求されたメトリクスをプッシュできます。アプリケーションはJMXを公開する必要がありません。サイドカーエクスポーターも必要ありません。カスタムのPrometheusエンドポイントも必要ありません。

それでも、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つのメトリクスソースを扱うようになりました。

Kafka Metrics SPI
Yammer / Coda Haleブローカーメトリクス
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のレイテンシが監視バックエンドに依存すべきではありません。

クライアントテレメトリレシーバーには、独自の自己監視メトリクスもあります。

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をダウンロードします。

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へコピーします。

cp monedula-metrics-reporter-0.10.0.jar $KAFKA_HOME/libs/

その後、ブローカーを設定します。

metric.reporters=dev.monedula.metricsreporter.OtlpMetricReporter

otlp.metric.reporter.endpoint=http://otel-collector:4317
otlp.metric.reporter.transport=grpc

これにより、Kafka SPIメトリクス、Yammerブローカーメトリクス、そしてクライアントテレメトリに対してレポーターが有効になります。

クライアントテレメトリは、ブローカー上でデフォルトで有効です。

otlp.metric.reporter.client.telemetry.enabled=true

明示的に無効化することもできます。

otlp.metric.reporter.client.telemetry.enabled=false

無効化すると、ブローカーはテレメトリ機能をクライアントへ通知しなくなります。

Collectorの設定

最小限のOpenTelemetry Collector設定で、レポーターからのOTLPを受信し、Prometheus向けにメトリクスを公開できます。

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をスクレイプできます。

scrape_configs:
  - job_name: otel-collector
    static_configs:
      - targets: ["otel-collector:8889"]

重要な部分: クライアントメトリクスサブスクリプション

プラグインをインストールするだけでは不十分です。

これはKIP-714で見落としやすい部分です。

クライアントは、クラスター上にクライアントメトリクスサブスクリプションが設定されている場合にのみ、メトリクスをプッシュします。

例えば次のようにします。

kafka-client-metrics.sh \
  --bootstrap-server broker:9092 \
  --alter \
  --name all-clients \
  --metrics "*" \
  --interval 5000

これはall-clientsという名前のサブスクリプションを作成または更新し、利用可能なすべてのクライアントメトリクスを要求し、プッシュ間隔を5秒に設定します。

本番環境での利用では、すべてを収集するのは過剰かもしれません。選択したメトリクスのプレフィックスだけをサブスクライブすることもできます。

例えば、標準のKafkaクライアントメトリクスから始めることができます。

kafka-client-metrics.sh \
  --bootstrap-server broker:9092 \
  --alter \
  --name standard-client-metrics \
  --metrics "org.apache.kafka." \
  --interval 10000

正確なサブスクリプション戦略は、何を観測したいか、そしてどれだけのメトリクス量を許容できるかに依存します。

重要なのは、サブスクリプションがブローカー側に存在することです。これにより、Kafka運用者は、各アプリケーションチームに監視構成の変更を求めることなく、何を収集すべきかを決定できます。

クライアント側の設定はどうするのか?

KIP-714をサポートするクライアントでは、メトリクスのプッシュは通常デフォルトで有効になっています。

それでも、明示的に指定することができます。

enable.metrics.push=true

Javaクライアントの場合、これはKafkaクライアント設定の一部です。

confluent-kafka-pythonのようなlibrdkafkaベースのクライアントでも、同じ考え方が当てはまります。例えば次のようにします。

Producer({
    "bootstrap.servers": "broker:9092",
    "client.id": "orders-producer",
    "enable.metrics.push": True,
})

そしてコンシューマーの場合は次のようにします。

Consumer({
    "bootstrap.servers": "broker:9092",
    "group.id": "orders-consumer-group",
    "client.id": "orders-consumer",
    "enable.metrics.push": True,
})

client.idは意図的に設定する価値があります。

レポーターはデフォルトでこれを転送されるクライアントメトリクスに追加するため、ダッシュボードはメトリクスをアプリケーションごとに区別できます。意味のあるclient idがなければ、メトリクスは得られるものの、トラブルシューティング時に格段に使いづらくなります。

拡張オプション

デフォルトでは、転送されるクライアントメトリクスはブローカーのアイデンティティとclient idで拡張されます。

otlp.metric.reporter.client.telemetry.enrich.broker=true
otlp.metric.reporter.client.telemetry.enrich.client.id=true

これにより、クライアントメトリクスをクラスター、ブローカー、アプリケーションごとにグループ化できます。

オプションの拡張もあります。

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を含むクイックスタートスタックが含まれています。

クイックスタートを起動します。

cd quickstart
docker compose up -d

デモサービスはKIP-714クライアントメトリクスサブスクリプションを作成し、プロデューサーとコンシューマーのトラフィックを開始します。そのため、スタックが起動すると、ブローカーダッシュボードとクライアントダッシュボードの両方がデータを受信し始めるはずです。

Grafanaは次のアドレスで利用できます。

http://localhost:3000

Prometheusは次のアドレスで利用できます。

http://localhost:9090

クイックスタートは、次のようにしてサブスクリプションを作成します。

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つの部分に分かれています。

標準パネル
拡張パネル

標準パネルは、関連する共通メトリクスを公開するすべての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で公開されています