Kafka オブザーバビリティのベストプラクティス¶
概要¶
Kafka は LinkedIn が開発した分散型のパブリッシュ/サブスクライブモデルに基づくメッセージキューであり、リアルタイムデータ処理システムで、水平スケーリングが可能です。RabbitMQ、RocketMQ などのミドルウェアと同様に、以下のような特徴があります。
- 非同期処理
- サービス疎結合
- トラフィックバーストの平滑化
以下は非同期処理の例を示す図です。
アーキテクチャ¶
以下の図のように、Kafka アーキテクチャは複数の Producer、複数の Consumer、複数の Broker、および Zookeeper クラスターで構成されます。
- Zookeeper：Kafka クラスターは Zookeeper を介してクラスター構成を管理します。Leader の選出、Consumer Group の変更時にリバランスを実行します。
- Broker：メッセージミドルウェアの処理ノードです。1 つのノードが 1 つの Broker であり、Kafka クラスターは 1 つまたは複数の Broker で構成されます。メッセージは複数の Broker に分散配置できます。
- Producer：プロデューサー。Broker にメッセージを発行します。
- Consumer：コンシューマー。Broker からメッセージを読み取ります。
- Consumer Group：各 Consumer は特定の Consumer Group に属します。この Group には名前を指定でき、指定しない場合はデフォルトの Group に属します。1 つのメッセージは複数の Group に送信できますが、1 つの Group 内では 1 つの Consumer のみがそのメッセージを消費できます。
Kafka はメッセージを分類し、クラスターに送信される各メッセージには Topic を指定する必要があります。Topic はメッセージのカテゴリであり、論理的にはキューと見なされます。Producer が生成する各メッセージには Topic を指定する必要があり、Consumer は購読している Topic に基づいて対応する Broker からメッセージをプルします。各 Topic には 1 つまたは複数の Partition が含まれ、Partition はフォルダに対応し、そのフォルダには Partition のデータとインデックスファイルが格納されます。各 Partition 内は順序が保証されます。このように、Topic は 1 つまたは複数の Partition に分割され、各 Partition には複数のレプリカが異なる Broker に分散配置されます。1 つのパーティションの複数のレプリカは、1 つの Leader（マスター）と複数の Follower（スレーブ）の関係にあり、Leader が外部にサービスを提供します。ここでの「外部」とは、クライアントプログラムとのやり取りを指し、Follower は Leader をパッシブに同期するだけで、外部とのやり取りはできません。このマルチレプリカメカニズムにより、障害時の自動フェイルオーバーが実現され、クラスター内の一部の Broker がダウンしてもサービス可用性が確保され、耐障害性が向上します。
以下の図に示すように、Kafka クラスターには 4 つの Broker があり、特定の Topic に 3 つのパーティションがあり、レプリカ係数も 3 に設定されていると仮定すると、各パーティションには 1 つの Leader と 2 つの Follower レプリカが存在します。
パーティションレプリカは異なる Broker に配置され、Producer と Consumer は Leader レプリカとのみやり取りし、Follower レプリカはメッセージの同期のみを担当します。Leader レプリカに障害が発生すると、Follower レプリカから新たな Leader レプリカが選出され、外部にサービスを提供します。
次に、Kafka のマルチレプリカメカニズムにおける重要な用語をいくつか紹介します。
- AR（Assigned Replicas）：1 つのパーティション内のすべてのレプリカを総称して AR と呼びます。
- ISR（In-Sync Replicas）：Leader レプリカと、Leader と一定の同期状態を維持しているすべての Follower レプリカ（Leader 自身を含む）で構成されます。
- OSR（Out-of-Sync Replicas）：ISR とは逆に、Leader レプリカと一定の同期状態を維持していないすべての Follower レプリカで構成されます。
まず、Producer はメッセージを Leader レプリカに送信し、その後 Follower レプリカが Leader からメッセージをプルして同期します。同じ時点では、すべてのレプリカ内のメッセージは完全には一致しません。つまり、同期中は Follower は Leader に対してある程度の遅延が生じます。これにより、三者間の関係は AR = ISR + OSR となります。
Leader は ISR セット内のすべての Follower レプリカの遅延状態を維持・追跡します。Follower の遅延が大きすぎるか、または障害が発生した場合、Leader はその Follower を ISR セットから除外します。もちろん、OSR セット内の Follower の同期状態が Leader に追いついた場合、Leader はその Follower を OSR セットから ISR セットに移動します。一般的に、Leader に障害が発生した場合、ISR セット内の Follower のみが新しい Leader に選出される資格を持ち、OSR セット内の Follower にはその資格はありません（ただし、パラメータ設定を変更することで変更可能です）。
Kafka を監視するための主要指標¶
次に、Kafka 指標の詳細情報を紹介します。
UnderReplicatedPartitions¶
UnderReplicatedPartitions：同期されていない状態のパーティション数、つまり利用不可レプリカのパーティション数です。異常値は 0 以外です。正常に動作しているクラスターでは、同期レプリカ（ISR）の数はレプリカの総数と完全に等しくなる必要があります。この値が 0 以外の場合、Broker 上の Leader パーティションに、完全に同期されておらず ISR に追いついていないレプリカを持つパーティションが存在することを示します。考えられる問題は以下の通りです。
- 特定の Broker がダウンしている。
- レプリカが存在するディスクの障害またはディスクフルにより、レプリカがオフラインになっている。OfflineLogDirectoryCount 指標が 0 以外の値であるかどうかと併せて判断できます。
- パフォーマンスの問題により、レプリカが同期できていない。2 つのケースが考えられます。1 つ目は、Follower レプリカのプロセスが停止し、一定期間 Leader に同期リクエストをまったく送信していない場合（例：頻繁な Full GC）。2 つ目は、Follower レプリカのプロセスの同期が遅く、一定期間内に Leader レプリカに追いつけない場合（例：I/O オーバーヘッドが大きい）。
| メジャーメント | kafka_replica_manager | |
|---|---|---|
| 指標 | 説明 | データ型 |
| UnderReplicatedPartitions | 同期されていない状態の Partition の数 | int |
| UnderMinIsrPartitionCount | 最小 ISR を下回る Partition の数 | int |
OfflineLogDirectoryCount¶
OfflineLogDirectoryCount：オフラインのログディレクトリ数。異常値は 0 以外です。オフラインのログディレクトリが存在するかどうかを確認するために、この指標を監視する必要があります。
| メジャーメント | kafka_log | |
|---|---|---|
| 指標 | 説明 | データ型 |
| OfflineLogDirectoryCount | オフラインのログディレクトリ数 | int |
IsrShrinksPerSec / IsrExpandsPerSec¶
任意のパーティションの同期状態にあるレプリカ数（ISR）は、Broker ノードを拡張しているか、パーティションを削除している場合を除き、安定している必要があります。高可用性を維持するために、Kafka クラスターは最小 ISR 数を保証し、パーティションの Leader がダウンした場合に Follower が引き継げるようにする必要があります。レプリカが ISR プールから削除される理由としては、Follower のオフセットが Leader より大幅に遅れている場合（replica.lag.max.messages 設定項目の変更）、または Follower が一定期間 Leader と連絡を失った場合（replica.socket.timeout.ms 設定項目の変更）などがあります。理由に関わらず、IsrShrinksPerSec（ISR 縮小）が増加したにもかかわらず、それに伴う IsrExpandsPerSec（ISR 拡大）の増加がない場合は、注意を払い、手動で介入する必要があります。
| メジャーメント | kafka_replica_manager | |
|---|---|---|
| 指標 | 説明 | データ型 |
| IsrShrinksPerSec.Count | ISR 縮小数 | int |
| IsrShrinksPerSec.OneMinuteRate | ISR 縮小頻度 | float |
| IsrExpandsPerSec.Count | ISR 拡大数 | int |
| IsrExpandsPerSec.OneMinuteRate | ISR 拡大頻度 | float |
ActiveControllerCount¶
ActiveControllerCount：現在アクティブなコントローラーの数。異常値は 0。Kafka クラスターで最初に起動したノードは自動的に Controller になります。このようなノードは 1 つだけでなければなりません。通常、Controller が存在する Broker 上のこの指標は 1 で、他の Broker 上のこの値は 0 です。Controller の役割は、パーティションの Leader のリストを維持し、Leader が利用不可になった場合に Leader の変更を調整することです。Controller の変更が必要な場合、新しい Controller は Zookeeper によって Broker プールからランダムに選出されます。通常、この値が 1 を超えることはありませんが、この値が 0 で一定期間（<1）続く場合は、明確な警告を発する必要があります。したがって、この指標はアラートに使用できます。
| メジャーメント | kafka_controller | |
|---|---|---|
| 指標 | 説明 | データ型 |
| ActiveControllerCount.Value | Controller の生存数 | int |
OfflinePartitionsCount¶
OfflinePartitionsCount：アクティブな Leader がないパーティションの数。異常値は 0 以外。すべての読み取りおよび書き込み操作は Partition Leader 上でのみ行われるため、アクティブな Leader がないパーティションは完全に利用不可となり、そのパーティション上の Consumer と Producer は、Leader が利用可能になるまでブロックされます。この指標はアラートに使用できます。
| メジャーメント | kafka_controller | |
|---|---|---|
| 指標 | 説明 | データ型 |
| OfflinePartitionsCount.Value | オフラインの Partition の数 | int |
LeaderElectionRateAndTimeMs¶
Partition Leader がダウンすると、Leader の選出がトリガーされます。LeaderElectionRateAndTimeMs を使用して、Leader が 1 秒間に何回選出されるか、選出頻度を監視できます。
| メジャーメント | kafka_controller | |
|---|---|---|
| 指標 | 説明 | データ型 |
| LeaderElectionRateAndTimeMs.Count | Leader 選出回数 | int |
| LeaderElectionRateAndTimeMs.OneMinuteRate | Leader 選出比率 | float |
| LeaderElectionRateAndTimeMs.50thPercentile | Leader 選出比率 | float |
| LeaderElectionRateAndTimeMs.75thPercentile | Leader 選出比率 | float |
| LeaderElectionRateAndTimeMs.99thPercentile | Leader 選出比率 | float |
UncleanLeaderElectionsPerSec¶
Kafka Broker のパーティション Leader が利用不可になると、unclean な Leader 選出が発生し、そのパーティションの ISR セットから新しい Leader が選出されます。本質的に、unclean leader 選出は、可用性の一貫性を犠牲にします。同期に利用可能なレプリカがなく、同期されていないレプリカから Leader を選出するしかない場合、以前の Leader の未同期メッセージはすべて永久に失われます。UncleanLeaderElectionsPerSec.Count の異常値は 0 以外であり、この場合はデータ損失を意味するため、アラートを発する必要があります。
| メジャーメント | kafka_controller | |
|---|---|---|
| 指標 | 説明 | データ型 |
| UncleanLeaderElectionsPerSec.Count | Unclean Leader 選出回数 | int |
TotalTimeMs¶
TotalTimeMs メトリクスは、以下の 4 つの指標の合計です。
- queue：リクエストキューで待機するのに費やされた時間
- local：リーダー処理に費やされた時間
- remote：フォロワーの応答を待機するのに費やされた時間（
requests.required.acks=-1の場合のみ） - response：応答の送信に費やされた時間
TotalTimeMs はサーバーリクエストの所要時間を測定するために使用されます。通常、この指標は非常に小さな変動のみで安定しています。異常が検出された場合、不規則なデータ変動が発生します。この場合は、queue、local、remote、response の各値を確認し、遅延の原因がどのセグメントにあるかを特定する必要があります。
| メジャーメント | kafka_request | |
|---|---|---|
| 指標 | 説明 | データ型 |
| TotalTimeMs.Count | 総リクエスト所要時間 | int |
PurgatorySize¶
PurgatorySize：プロデュース（produce）およびフェッチ（fetch）リクエストが必要になるまで待機する一時的な格納領域です。purgatory のサイズに注意することで、レイテンシの根本原因を特定するのに役立ちます。例えば、purgatory キュー内のフェッチリクエストの数が増加した場合、コンシューマーのフェッチ時間の増加を容易に説明できます。
| メジャーメント | kafka_purgatory | |
|---|---|---|
| 指標 | 説明 | データ型 |
| Fetch.PurgatorySize | Fetch Purgatory のサイズ | int |
| Produce.PurgatorySize | Produce Purgatory のサイズ | int |
| Rebalance.PurgatorySize | Rebalance Purgatory のサイズ | int |
| topic.PurgatorySize | topic Purgatory のサイズ | int |
| ElectLeader.PurgatorySize | Leader 選出 Purgatory のサイズ | int |
| DeleteRecords.PurgatorySize | レコード削除 Purgatory のサイズ | int |
| DeleteRecords.NumDelayedOperations | 遅延削除レコード数 | int |
| Heartbeat.NumDelayedOperations | ハートビート監視 | int |
BytesInPerSec / BytesOutPerSec¶
BytesInPerSec / BytesOutPerSec：受信/送信バイト数。通常、ディスクスループット、ネットワークスループットがボトルネックになる可能性があります。データセンター間でメッセージを送信する場合、Topic の数が多い場合、またはレプリカが Leader に追いついている場合、ネットワークスループットが Kafka のパフォーマンスに影響を与える可能性があります。これらの指標を使用して、Broker 上のネットワークスループットを追跡し、ボトルネックがどこにあるかを判断します。
| メジャーメント | kafka_topics | |
|---|---|---|
| 指標 | 説明 | データ型 |
| BytesInPerSec.Count | 1 秒あたりの流入バイト数 | int |
| BytesInPerSec.OneMinuteRate | 1 秒あたりの流入レート | float |
| BytesOutPerSec.Count | 1 秒あたりの流出バイト数 | int |
| BytesOutPerSec.OneMinuteRate | 1 秒あたりの流出レート | float |
RequestsPerSec¶
RequestsPerSec：1 秒あたりのリクエスト数。この指標を監視することで、Producer と Consumer のリクエストレートをリアルタイムで把握し、Kafka が効率的に通信していることを確認できます。この指標が継続的に高止まりしている場合は、Producer または Consumer の数を増やしてスループットを向上させ、不必要なネットワークオーバーヘッドを削減することを検討できます。
| メジャーメント | kafka_topics | |
|---|---|---|
| 指標 | 説明 | データ型 |
| TotalFetchRequestsPerSec.Count | 1 秒あたりのプルリクエスト数 | int |
| TotalProduceRequestsPerSec.Count | Producer が 1 秒あたりに書き込むリクエスト数 | int |
| FailedFetchRequestsPerSec.Count | Topic の失敗した Fetch 数 | int |
| FailedProduceRequestsPerSec.Count | 送信リクエストの失敗レート | int |
その他のよく使われる指標¶
| メジャーメント | kafka_controller | |
|---|---|---|
| 指標 | 説明 | データ型 |
| GlobalTopicCount.Value | クラスターの総 Topic 数 | int |
| GlobalPartitionCount.Value | パーティション数 | int |
| TotalQueueSize.Value | キューの総数 | int |
| EventQueueSize.Value | イベントキュー数 | int |
| メジャーメント | kafka_request | |
|---|---|---|
| 指標 | 説明 | データ型 |
| RequestQueueTimeMs.Count | リクエストキュー時間 | int |
| ResponseSendTimeMs.Count | 応答送信時間 | int |
| MessageConversionsTimeMs.Count | メッセージ変換時間 | int |
| メジャーメント | kafka_topics | |
|---|---|---|
| 指標 | 説明 | データ型 |
| PartitionCount.Value | Partition の数 | int |
| LeaderCount.Value | Leader の数 | int |
| BytesRejectedPerSec.Count | Topic リクエストが拒否された数 | int |
シナリオビュー¶
Guance を使用して Kafka のオブザーバビリティを開始する前に、まず Guance アカウント を登録し、登録後に Guance のワークスペースにログインする必要があります。その後、<Kafka 統合ドキュメント> に従って Kafka のオブザーバビリティを実現してください。
