この記事は食べログアドベントカレンダー2021の21日目の記事です🎅🏻🎄
はじめに
こんにちは。技術部マイクロサービス化チームの @SinceK13 です。
マイクロサービス化チームは、「巨大なモノリシックサービスにおける開発の辛さを解消し、少人数のチームが自律的に意思決定しながら開発するためのシステム基盤を作る」というミッションのもと活動しています。
本エントリではマイクロサービス化チームで取り組んだ、食べログの内製 Pub/Sub メッセージング基盤を Apache Kafka（以下 Kafka）を利用した Pub/Sub メッセージング基盤にリプレイスした話について紹介します。
また、12/14 に同チームの @weakboson が「食べログのレストラン検索を支える Debezium と Apache Kafka」というタイトルで、本エントリとは異なる Kafka の活用事例を紹介していますのでご興味ある方はぜひご覧ください^^
Pub/Sub メッセージングとは
メッセージ（データ）の送信者が、特定の受信者を想定せずにメッセージ（データ）を送るようにプログラムされたものになります。
送信者をパブリッシャ（Publisher、出版者）、受信者をサブスクライバ（Subscriber、購読者）と呼ぶので「Pub/Sub メッセージングモデル（出版-購読型モデル）」と呼ばれます。
仲介者は、メッセージを仲介するので「メッセージブローカ」と呼ばれます。
Pub/Sub メッセージングとして代表的なトピックベースのシステムでは、メッセージは「トピック」という単位で分類され、パブリッシャはトピックを指定してメッセージを送信し、サブスクライバは関心のあるトピックを受信します。
Pub/Sub メッセージングの特徴
主に以下の3つの特徴があります。
非同期 ・・・パブリッシャはメッセージを送信したらサブスクライバの応答を待たずに次の処理を始めることができます。
疎結合 ・・・パブリッシャもサブスクライバもメッセージブローカと通信できればいいので互いの存在を知る必要がありません。
スケーラブル ・・・パブリッシャとサブスクライバは1対1に限らず、それぞれが増えても互いに意識する必要がありません。
このような特徴からマイクロサービス間で非同期処理を行いたいときに有用です。
以前のPub/Sub メッセージング基盤の構成
以前の Pub/Sub メッセージング基盤は2017年に開発され、下記のような構成となっていました。
メッセージの発行は、パブリッシャのアプリケーションから「メッセージAPI」というアプリケーションを介して行われます。
メッセージキューに登録されたメッセージは、内製 Pub/Sub によってサブスクライブされているアプリケーションのジョブキューに登録され、サブスクライバのアプリケーションの Sidekiq Worker がジョブキューからメッセージを取り出して処理を行います。
また、以前の Pub/Sub メッセージング基盤の機能の1つとして、メッセージを追跡できる仕組みがありました。
Pub/Sub メッセージングではパブリッシャもサブスクライバも互いの存在を知る必要がないと前述しましたが、業務としては問い合わせがあったときにメッセージがどうなったのか調べなくてはならないことがあるためこのような仕組みが必要でした。
具体的には、発行した（発行しようとした）メッセージには全システムにおいてユニークな「相関 ID（correlation id）」が振られており、パブリッシャによるメッセージ発行、メッセージ API によるメッセージの中継、サブスクライバでの処理成功、そして時には処理失敗といったそれぞれのポイントでタイムスタンプやデータを相関 ID と一緒にログ出力します。（トレースログと呼んでいます。）
これにより、相関 ID で関連付けてメッセージの流れを追跡できます。
トレースログは下記の項目を JSON で出力します。
| フィールド名 | 説明 | 例 |
|---|---|---|
| cid | 相関ID（メッセージごとにユニーク） | "04b85967-2ab1-4d3a-8285-ed230c698429" |
| topic | トピック | "hoge_topic" |
| event | メッセージ発行、中継、処理など | "occurred"（メッセージ発行） |
| body | メッセージのデータ | {"hoge_id":12345} |
| processing | 処理に要した時間 | 0.009870573 |
| occurred_at | メッセージが発行された時刻 | "2021-12-21T08:06:34+09:00" |
| processed_at | メッセージがホストで処理された時刻 | "2021-12-21T08:06:36.939+09:00" |
| host | メッセージを処理したホスト | "hoge-server1" |
課題
以前の Pub/Sub メッセージング基盤には以下のような課題がありました。
- 内製 Pub/Sub が不安定でときどきメッセージ消失を起こしてしまう
- メッセージは正常に処理されても即消えてしまい、トレースログは数日間しか保持されない
- メッセージの追跡調査や処理に失敗した時のリトライはサーバに入って CLI で行う必要があり大変
昨年のアドベントカレンダーの @tkyowa の「食べログの大規模なレガシーシステムを段階的に改善していく取り組み」の中でも述べられている通り、この Pub/Sub メッセージング基盤で動作している機能のビジネス重要度が年々高まっており、一刻も早い改善が必要でした。
新しい Pub/Sub メッセージング基盤の構成
上記の課題を解決すべく、新しい Pub/Sub メッセージング基盤はこのような構成をとりました。
処理の流れとしては、まずパブリッシャ（Producer）の Kafka Producer から、指定した Kafka のトピックへメッセージが発行されます。
次にサブスクライバの Kafka Consumer がトピックからメッセージを取り出して Redis に書き込みます。
最後に Sidekiq Worker が Redis からメッセージを取り出して処理を実行します。
技術選定
メッセージ消失を防ぐために
まず、「内製 Pub/Sub が不安定でときどきメッセージ消失を起こしてしまう」という課題を解決するために、Pub/Sub メッセージング基盤のコア部分であるメッセージブローカには、Kafka を採用しました。
Kafka には以下のような特徴があります。
- 大量のメッセージを高速に処理できる
- Broker 複数台構成が前提でメッセージを複数台に同期して冗長化する
- 耐障害性が高い（1台でも稼働していればメッセージを消失しない設定が可能）
- メッセージをディスクに書き込む
- データの永続化が可能
- Consumer Group ごとに独立してメッセージを読み出すことができる
- ファンアウトと同等のことができる
これらの特徴から、高速なうえにメッセージ消失が起きにくいメッセージブローカと言えます。
「1台でも稼働していればメッセージを消失しない設定が可能」と記載しましたが、Broker 3台構成の場合を例として説明します。
Broker と Producer のポイントとなる設定は下記の通りです。
Broker 設定
| 項目 | 設定値 | 項目説明 | 備考 |
|---|---|---|---|
| Broker 数 | 3 | - | 最小限冗長性の定番 |
| default.replication.factor | 3 | Replica の複製数。 Leader Replica と Follower Replica の合計数 | O'Reilly Kafka でも最小 Replica 数として3台が推奨されている |
| min.insync.replicas | 2 | 正常終了と見なすために最低限同期が必要な Replica 数 | この Replica 数同期を行う必要があるので、1台の故障に対しては問題なく動作を続けることができる。2台故障した場合はメッセージ受付を停止する |
Producer 設定
| 項目 | 設定値 | 項目説明 | 備考 |
|---|---|---|---|
| acks | all | Producer が送信成功と見なすタイミング | all にすることで、すべての insync.replicas のメッセージ受信完了を待ってから、確認応答またはエラーを返すようになる。複数の Broker がメッセージを保持するので、いずれかがクラッシュしてもメッセージは確実に残る |
このように設定すると、まず3台とも稼働してる時は3台にデータを同期します。
1台故障しても2台分データを同期できるので継続して利用できます。
これは2台にデータが冗長化されていれば、更に1台故障しても同期したデータは失われないためです。
2台故障した場合はメッセージ受付を停止してデータ保護を優先します。
よって、1台でも稼働していればメッセージを消失しないようになります。
また「メッセージは正常に処理されても即消えてしまう」という課題は、Kafka のオフセット（offset）という既読管理の概念によって解決されます。
Kafka はキューのように処理済みメッセージを削除するのではなく、処理済みの位置をオフセットという位置情報で管理します。
処理済みメッセージは指定した期間または容量だけ保存されるので、過去のメッセージを調査したり、オフセットを巻き戻して再処理することもできます。
メッセージの追跡調査や処理に失敗した時のリトライを容易にするために
次に「メッセージの追跡調査や処理に失敗した時のリトライはサーバに入って CLI で行う必要があり大変」という課題を解決するために、GUI で操作できるようにする解決策を考えました。
食べログでは、Web サーバや AP サーバのログを Google Cloud Platform（以下 GCP）の Cloud Logging に転送しており、ログの調査は GCP のログエクスプローラ上で行うことが多いです。
よって、Pub/Sub メッセージング基盤のトレースログも Cloud Logging に転送することで、他のログと同じ方法で調査を行えるようにしました。
また、BigQuery にトレースログをエクスポートすることで、より詳細な調査が可能になりました。
また、Ruby 用の Kafka ライブラリとしては Karafka を採用しました。
Karafka は特に Consumer 機能を手厚くサポートしている gem です。デプロイに絡む SIGNAL ハンドリングや Graceful シャットダウン、CLI といった機能が提供されます。
さらに Karafka には、Consumer のメッセージ処理を Sidekiq に移譲できる Karafka Sidekiq Backend という gem が用意されています。
以前の Pub/Sub メッセージング基盤でも Sidekiq を利用していたため、Karafka Sidekiq Backend を利用することでサブスクライバ側のコードはほぼそのまま流用できました。
さらに、Sidekiq の GUI から進捗確認、強制停止、失敗したメッセージ処理のリトライができる Sidekiq WebUI も利用でき、より調査を容易にすることができました。
Kafka Consumer には管理画面がないので、こちらも Karafka Sidekiq Backend を利用するメリットの1つです。
また、Karafka Sidekiq Backend を使うと Sidekiq がメッセージ処理を行うので、処理に数十秒〜数分かかるような重いメッセージを扱いやすくなるということも採用理由の1つです。
Kafka Consumer で直接メッセージ処理を行うこともできますが、 メッセージの処理時間が Consumer の死活確認間隔より長くなるとメッセージ処理に失敗したとみなされてしまうため、処理時間の長いメッセージを扱う場合は Karafka Sidekiq Backend を利用すると良いと思われます。
結果
メッセージが消失しなくなった
前述したようにメッセージブローカとして Kafka を採用したことで、「内製 Pub/Sub が不安定でときどきメッセージ消失を起こしてしまう」課題を解決することができました。
7月から新 Pub/Sub メッセージング基盤での本番稼働が開始しておりますが、これまで一度もメッセージ消失は起きずに運用を継続できています。
数週間、数ヶ月前の調査ができるようになった
以前の Pub/Sub メッセージング基盤では、「メッセージは正常に処理されても即消えてしまい、トレースログは数日間しか保持されない」という課題がありましたが、Kafka はメッセージをディスクに書き込むので永続化されるようになりました。
また、トレースログを GCP に転送して長期間保持するようにしたことにより、数週間、数ヶ月前の調査ができるようになりました。
GUI から簡単に調査、失敗したメッセージ処理のリトライができるようになった
以前の Pub/Sub メッセージング基盤では、「メッセージの調査や処理に失敗した時のリトライはサーバに入って CLI で行う必要があり大変」という課題がありましたが、トレースログを Cloud Logging に転送することにより、GUI から簡単に調査ができるようになりました。
さらに BigQuery にトレースログをエクスポートすることで、より詳細な SQL による調査を行うことが可能になりました。
また、トレースログの調査が他のログと同じ方法で行えるようになったことも成果の1つです。
そして、Sidekiq WebUI でメッセージの進捗確認、強制停止、失敗したメッセージ処理のリトライを簡単に行えるようになりました。
特に失敗したメッセージ処理のリトライについては、以前の Pub/Sub メッセージング基盤の場合、対象のサーバに入ってリトライコマンドを実行する必要がありましたが、Sidekiq WebUI ではリトライしたいメッセージをチェックして再実行ボタンをポチッとするだけでリトライ可能になり、大幅に運用を改善することができました。
さいごに
マイクロサービス化チームでは共に改善を進めてゆく仲間を募集中です。
マイクロサービス化チームは戦略立案、導入技術の選定といったシステム改善の初期フェーズから意思決定に参加できる、大きな裁量とやりがいのあるポジションです。「ユーザーと飲食店をつなぐ」という食べログのコンセプトに共感された方はぜひご応募お願いします。
まずはカジュアル面談で情報交換をしてみたいという方も大歓迎です。その場合はご応募いただくときに、フリーテキスト記入欄に「カジュアル面談希望」とご記載ください。
明日は @sadashi の「食べログアプリでの技術的負債との向き合い方」です。お楽しみに!