2018」が開催されました。メッセージアプリだけでなく、さまざまなサービスの開発や新たな技術領域への投資を行っているLINEが目指すビジョンについて、エンジニアたちの技術的知見や挑戦を通して紹介します。Kafkaプラットフォームの舞台裏と、実際に経験した課題と解決策について紹介しました。河村勇人氏:みなさん、こんにちは。最初に自己紹介から始めたいと思います。LINEでシニアソフトウェアエンジニアとして働いています。2015年に新卒として入社して以来、LINEサーバーの開発を中心にHBaseの運用などに携わり、現在では全社的なKafkaのプラットフォームを提供しているチームのリーディングをしています。Summitという大規模なカンファレンスでのプレゼンテーションもしてきました。今日はどうぞよろしくお願いします。今日お話しするのは、我々がLINEで運用している大規模なKafkaのプラットフォームについてです。Clusterといった非常に大規模な分散ミドルウェアを運用しているということをご存知かもしれません。Kafkaもその1つです。ご存知ない方のためにKafkaというソフトウェアについて簡単に説明をしておくと、Kafkaはストリーミングデータを取り扱うためのミドルウェアと、その周辺エコシステムです。我々のようなサービスを提供している企業にとって、重要な特性をいくつか持っています。1つは高いスケーラビリティとアベイラビリティを持っているということ。1つはデータのパーシステンシーをサポートしているということです。これはKafkaに1度書き込まれたデータは、その時点で永続化されたものと考えることができるということです。Pub-Subモデルでのデータのディストリビューションをサポートしていることも重要な特性の1つです。Kafkaを理解するための3つのコンポーネントさて、Kafkaのアーキテクチャを理解するうえでは、3つのコンポーネントについて理解する必要があります。左手にあるのがProducer。これはKafkaに対してデータを書き込むクライアントです。これは任意数のノードで構成されるクラスタで、トピックというデータの管理のユニットを任意数ホスティングすることができます。Producerはいずれかのトピックを対象にしてデータを書き込みます。(スライドを指して)右側にあるのがConsumer。これはKafkaからデータを読み出すクライアントです。Consumerはデータを読み出したいトピックを指定して、そこからデータを読み出します。重要な特性は、1つのトピックについてConsumerが複数個異なる目的のために存在できるということです。1度トピックに書き込まれたデータを異なるConsumerが異なる処理の目的のために複数回読み出すことができるのが、Pub-Subと呼ばれるデータのディストリビューションモデルです。LINEでは非常に大規模なKafkaのクラスタを運用しています。これは特定のサービスやシステムのみを対象にしたものではなく、社内のあらゆるサービスやシステムから利用できる汎用的なプラットフォームです。最初はLINEサーバーの開発チームの中で、複雑化しすぎたアーキテクチャを簡素化するための中間レイヤーとして採用されましたが、そこでの成功例や高い運用実績を見たほかの多くのチームから「私たちも使いたい」というニーズを受けて、全社的なプラットフォームへと自然的に発展してきました。LINEにおけるKafkaの使い方LINEにおけるKafkaの使い方は、大きく2つあります。1つ目はシンプルに分散キューイングシステムとしての利用です。あるサービスでWebアプリケーションサーバーの中で生じた重い処理について、その中では処理をせずに、別プロセスで動いているバックグラウンドのタスクプロセッサーにリクエストするためのキューとして利用されています。1つの利用法としては、データハブとしての利用があります。これはあるサービスで生じたデータのアップデートを、それを利用するほかの複数のサービスに対して伝搬するためのハブとしての利用です。具体的な例について、1つ説明してみたいと思います。LINEではユーザーの友だち関係をデータとして持っています。あるユーザーAが別のユーザーBを友だちとして追加した際に、LINE本体の中ではそれに応じた処理を行います。それに加えて、ここで発生した関係性のアップデートを、イベントとしてKafkaのトピックに書き込みます。ここで書き込まれたイベントは、統計システムや、タイムラインなどのユーザグラフを利用して提供されるサービスから利用されています。みなさんがふだんLINEをご利用されている中で、見知らぬ友だちが突然追加されてスパムや広告を送られた、という不快な経験をされた方がいらっしゃるかもしれませんが、我々はそうした望まないユーザーのアクティビティを解析して、自動的にペナルティを与えるシステムを動かしていて、そういったシステムからも利用されています。単一のクラスタにデータを集中させる理由さて、ここで重要なのは単一の大きなクラスタが独立した複数のサービスやシステムから共通して利用されているということです。これには大きく2つ理由があります。1つめはデータハブとしてのコンセプトです。我々は意図して多くのデータを単一のクラスタに集中させようとしています。これによって、データを利用したいサービスが簡単にそのデータを見つけることができて、かつ統一的な手段によってそのデータへアクセスすることができます。これによって、アーキテクチャもシンプルに保たれます。1つの理由としては、オペレーションの効率です。我々がサポートするすべてのサービスやシステムに対して、Kafkaのクラスタを1つずつ用意していかなければならないとすれば、我々のオペレーションコストは、サポートするサービス、システムの数に比例して増えていってしまうことになります。それを避けるために、我々はこういったサービスやシステムから単一のクラスタを利用してもらって、我々の持てるエンジニアリングリソースをすべてそのクラスタに注ぎ込んで、それの持つリライアビリティやパフォーマンスを最大化するという戦略を取っています。今日においてはLINEでは非常に多くのサービスが我々のKafkaプラットフォームに依存して動いています。LIVEといったサービスやデータの解析システム、このほかにも数多くのサービスがKafkaをバックエンドとして構築されています。結果として、我々のクラスタは非常に大規模なトラフィックを取り扱うものになっています。デイリーでは2,500億件を超えるレコードの書き込みがあります。データの量に関しては、デイリーで210テラバイト。ピーク時では秒間4ギガバイトを超えるデータの書き込みがあります。これらは50を超える独立したサービスやシステムから利用されているものです。これは単一のクラスタとしては世界最大級のスケールです。clusterは、多くのサービスのバックエンドとして利用されています。そのため、それに求められるリライアビリティやパフォーマンスは非常に高いものです。いったリクワイアメントをマルチテナンシーのクラスタで満たすためには、いくつか達成しなければならない条件があります。1つ目の条件は、クラスタとそれ自身をabusiveのワークロードから守れるということです。我々のプラットフォームは社内のサービスやシステムに向けて提供されているものなので、例えば悪意を持ってクラスタを攻撃するといったケースについては考える必要はありませんが、依然として、例えばコンフィグレーションでミスをしたり、デプロイでミスをしたりということによって、我々の予期しないワークロードがクラスタにもたらされることはあり得ます。我々はどのリクエストがどのクライアントから来ているのかをきちんと把握できる必要があります。これは我々の予期しないリソースのアクティビティが、例えばクラスタで検知された際に、その原因となっているクライアントを速やかに特定して、原因を解決しなければならないからです。最後に一定レベルでのワークロードのアイソレーションがクライアント間で保たれている必要があります。これは、例えばあるクライアントがそれ自身のもたらしたワークロードによって遅いレスポンスタイムを経験しているとき、同じ理由によって同じクラスタに接続しているまったく無関係なクライアントが同様のレスポンスタイムの悪化を経験するということはあってはいけないということです。リクエスト数をコントロールする1つ目の条件、クラスタをabusiveなワークロードから守ることについて、我々は、Kafkaのクラスタを運用するうえで重要なのは、データの量よりもリクエストの数をコントロールすることだと経験的に知っています。Kafkaは大容量のデータを取り扱ううえで、非常によく設計されたソフトウェアです。Kafkaはそれ自身でキャッシュのレイヤーを持っていません。代わりにオペレーティングシステムによって提供されるページキャッシュにキャッシュの機能を全面的に依存することによって、大容量のデータがアプリケーションのメモリとOSのページキャッシュに重複してキャッシュされてしまうことを防ぐようにしています。Kafkaのクライアントにはバッチングという機能がネイティブでサポートされています。これは複数個のレコードを1つの大きなリクエストにまとめてクラスタに送るということができる機能です。これによってデータの量やレコードの数が増えたとしても、リクエストとしては少ない数で、リクエストごとに生じるオーバーヘッドを避けられるデザインになっています。我々としてはクライアントが送ってくるリクエストの数を制御する必要があるわけです。そのために、我々はKafkaのリクエストクォータという機能を使っています。これは、あるクライアントが使用可能なブローカーのリソース......厳密にはブローカーが持っているスレッド時間を制限できる機能です。これをすべてのクライアントについてデフォルトで設定することによって、単一のクライアントが使えるブローカーのリソースの量を制限しています。これによって、例えば単一のクライアントが暴走して、我々の予期しないようなリクエストをクラスタにもたらしたケースでも、クラスタ全体が停止してしまったりパフォーマンスが著しく悪化したりということがないようにしています。プロダクション環境で起きたトラブルとその解決法もう1つの条件は、クライアント間でワークロードのアイソレーションを保つことです。これについては、実際に我々のプロダクション環境で発生したトラブルと、それを我々がどのように解決していったかについて、少し掘り下げてお話ししたいと思います。きっかけは、我々がクラスタ上でパフォーマンスの悪化を発見したことでした。ProduceというKafkaのクライアントがクラスタにデータを書き込む際に利用するAPIがあります。APIの99パーセンタイルのレスポンスタイムが、ふだんより50倍から100倍悪化していることを発見しました。我々は調査を開始して、そこで大きく2つ発見がありました。1つ目は、当該時間帯にブローカーのマシン上で非常に大きな量のディスクリードが起きていたということです。1つの発見は、ネットワークスレッドというKafkaの中で、クライアントとのIOを担当しているスレッドがありますが、このスレッドのユーティライゼーションが非常に上がっていたということです。Kafkaのリクエストハンドリングさて、ここから我々は問題の調査を開始していくわけですが、その前にKafkaがどのようにリクエストをハンドリングするのかということについて、簡単に説明しておきたいと思います。Kafkaのリクエストハンドリングは、大きく2つのスレッドレイヤーで構成されています。1つ目のレイヤー、ネットワークスレッド。これはクライアントとのIOを担当しているスレッドです。クライアントソケットに到着したリクエストを読み出して、リクエストオブジェクトを作る。用意ができたレスポンスオブジェクトをクライアントソケットに書き出すということを担当しています。1つのレイヤー、リクエストハンドラースレッド。これはネットワークスレッドによって読み出されたリクエストの内容を処理して必要なレスポンスオブジェクトをネットワークスレッドに返すという仕事をしています。全体の流れとしては、次のようになります。クライアントソケットにリクエストが到着すると、ネットワークスレッドがそれを読み出してリクエストオブジェクトを作ります。リクエストオブジェクトはリクエストキューに書き込まれます。このリクエストキューはブローカーの中で1つだけ用意されているキューで、すべてのリクエストハンドラースレッドは、このキューをポーリングしてリクエストを取得しています。どれかのリクエストハンドラースレッドによってリクエストがキューから取得されると処理されます。APIのタイプに応じてはローカルディスクとのIOが行われて、レスポンスオブジェクトが作られます。レスポンスオブジェクトはレスポンスキューに書き込まれてネットワークスレッドに取得されるんですが、レスポンスキューは1つのネットワークスレッドごとに1つずつ用意されているモデルになっています。このレスポンスキューに書き込まれたレスポンスというのは、それを担当するネットワークスレッドによってのみ処理されるようになっています。ネットワークスレッドがレスポンスキューからレスポンスを取得すると、その内容をクライアントソケットに書き出して処理が完了。これが一連の流れになります。ネットワークスレッドというのはイベントループを処理しているスレッドになります。いわゆるイベントドリブン非同期IOというものをやっています。複数のソケットを多重化してIOの準備ができたソケットを逐次処理する、というループをずっと回しているのが、ネットワークスレッドです。基本的に、ネットワークスレッドの中で行われている処理は、IOによってブロックされることがありません。あるコネクションが新しくブローカーに作られると、そのコネクションはブローカーの中にある複数のネットワークスレッドのうち、いずれかにアサインをされて、以降はそのネットワークスレッドによってのみ処理されるようになります。原因の検証さて、我々が今調べているネットワークスレッドのユーティライゼーションが上がっていた現象について、どのようなケースで起きるかということを考えてみました。1つ目のケースとしては、本当に処理しなければならないリクエストの数が増えていて、忙しく仕事をしている状態があり得ると思います。我々はリクエストの数に変化がないことを確認していたので、これとは別の可能性を考えました。1つの可能性として考えられたのは、ネットワークスレッドがそれを処理する過程のイベントループの中のどこかの処理において、ブロックされてしまっているという可能性です。それによってスレッドのユーティライゼーションが上がっている可能性を考えました。それはどのようなケースで発生し得るかを理解するために、我々はKafkaのソースコードを読みました。その結果、それが起き得る可能性が1つ見えてきました。それについて理解するためには、KafkaにおけるAPIごとのレスポンスの処理の仕方の違いを理解する必要があります。FetchというAPIとその他のAPIとでは、処理の仕方が違うことがわかりました。ここで説明するのは、その他のAPIについてです。Fetch以外のAPIのレスポンスの処理においては、レスポンスキューからレスポンスオブジェクトがネットワークスレッドに取られた時点で、クライアントソケットに書き出すべきデータがすべてメモリ上にそろっています。ネットワークスレッドがやることとしては、そのそろったデータをひたすらクライアントソケットにコピーしていくというだけのオペレーションになります。APIの場合は事情が異なります。FetchというAPIはConsumerがトピックのデータをクラスタから読み出す際に利用するAPIです。ブローカーはそのレスポンスを返す際に、ローカルディスクにストアされているトピックのデータをクライアントソケットにコピーしてあげる必要があります。この処理のために、ブローカーはsendfileというシステムコールを使っています。このsendfileというシステムコールは、Linuxカーネルなどで提供されているシステムコールで、ローカルディスク上のデータをクライアントソケットにダイレクトにコピーすることを可能にするAPIです。それによって、ユーザースペースのメモリに対して1度メモリをコピーするというオーバーヘッドを避けることができるので、非常に効率的に動作することが知られています。Kafkaはデザイン上、この機能に強く依存しています。sendfileを処理する際には、もしその対象のデータがページキャッシュ上に存在していれば、Linuxカーネルのやることはここからクライアントソケットにコピーをするだけなんですが、もしページキャッシュにデータが存在していない場合は、ローカルのディスクからデータをローディングしてくる必要があります。SystemTapを用いてシステムコールの処理時間を調べるさて、もしこのディスクからのデータのロードが起きていたらどうなるかを考えてみたいと思います。対象のデータがページキャッシュに存在している場合、先ほど述べたように、それはシンプルなメモリコピーのオペレーションです。典型的には数十マイクロセカンドから数百マイクロセカンドという非常に短い時間で処理を終わらせることができると知られています。一方で、もし対象のデータをディスクから読み出してくる必要がある場合、これは非常に遅いオペレーションになります。典型的には数ミリセカンドから数十ミリセカンドといった長い時間を要することが知られています。この処理がイベントループの中で起きていたとしたら、それは大きな問題になります。このシナリオについてもう少し確証を得るために、我々はsendfileシステムコールの処理時間について調べてみることにしました。このような調査を行う際に我々がよく使うツールの1つとして、SystemTapというものがあります。SystemTapはダイナミックトレーシングツールと呼ばれるもので、数行のスクリプトを書くことによって、カーネルの中で起きている種々の出来事を観測することができるようになるというツールです。簡単な例として右手に持ってきたのは、あるマシン上で呼ばれているシステムコールの数をカウンティングするというだけのスクリプトです。cntというところで宣言された変数に対して、その下のブロックの中でシステムコールが呼ばれる度に、その名前のエントリーをインクリメントしています。最後のブロックでcnt変数をイテレートしながらシステムコールの名前とそれが呼ばれた回数をプリントしていく。それだけのスクリプトです。ご覧のように、システムコールの名前とそれが呼ばれた回数が見えているとわかると思います。このSystemTapは、我々のようなスケールのサービスを提供しているものにとって、非常に重要な特性を1つ持っています。非常に低いオーバーヘッドで動作するということです。プロダクション環境のマシン上であっても、オーバーヘッドを気にすることなく安心して実行することができます。同様のツールとしてはDTraceやeBPFというものも知られていると思います。結果から考えられることさて、このSystemTapを用いてブローカーによって発行されているsendfileシステムコールの処理時間について計測してみました。スクリプトの内容については省略しますが、得られたのはご覧のようなヒストグラムです。ご覧のように、ほとんどの処理が2マイクロセカンドから32マイクロセカンドといった非常に短い時間で処理を完了しているのに対して、ごく一部のみ、8ミリセカンド以上という非常に長い時間がかかっているということが見て取れると思います。これによって確証を得た我々は、問題の仮説を次のように立てました。ブローカーがFetchリクエストを処理します。その過程においてsendfileを呼ぶ中でディスクのリードをする必要があると、それによってネットワークスレッドのイベントループがブロックされます。すると、現在処理中のそのリクエストだけではなく、同じネットワークスレッドによって処理されなければならない後続のレスポンス、ほかの無関係なAPIのレスポンスがすべてブロックされるという仮説です。これによって、ディスクリードのために生じたレイテンシが、今処理されることを待っているすべてのレスポンスに降りかかっている状態です。APIのレスポンスタイムが悪化していることについても説明ができます。検討した解決策この問題を解決するために、我々はいろいろな方法を考えましたが、最終的にたどり着いた結論は、本来ブロックすべきではないイベントループがブロックしないように直そうということです。そのために、sendfileがネットワークスレッドの中で呼ばれるケースにおいては、対象のデータが必ずページキャッシュ上に載っているようにするという改善をブローカーに施すことにしました。アイデアとしては、現在ブロッキングな挙動が起きているネットワークスレッドの中でのディスクからのデータのローディングをリクエストハンドラースレッド側に移してしまおうということです。リクエストハンドラースレッドは、単一のキューに対してみんなでポーリングをしているようなモデルで動いているので、ここでのブロッキングはほかのスレッドに一切影響を与えませんし、ほかのスレッドは後続のリクエストについてその間処理を継続することができます。単一のスレッドにおけるブロッキングは問題になりません。そこでwarmupされたページキャッシュは、そのあとネットワークスレッドで改めて読み出されてクライアントに書き出されるわけですが、ここでのブロッキングは起きません。ページキャッシュをwarmupするという処理をする必要があるわけですが、これをどのように行うかということについては、少し考える必要があります。シンプルなやり方としては、対象のデータに対してreadを呼ぶというやり方があります。この方法には懸念がありました。対象のデータに対してシンプルにreadしてしまうと、まずそのデータはディスクからローディングされるんですが、そのあとユーザースペースのバッファーにコピーされてきてしまうことになります。そもそも、なぜKafkaがsendfileというシステムコールを使っているかを思い出してみると、それは大量のデータのメモリのコピーのオーバーヘッドを避けたいためです。このやり方では、本来Kafkaが持っている非常に効率的なパフォーマンスの特性を犠牲にしてしまう可能性がありました。ほかの方法について考える必要があったわけです。試行錯誤の末に我々がたどり着いたやり方は、少しトリッキーなやり方でした。対象のデータに対してsendfileを呼びます。その際の宛先を/dev/nullに設定します。実はLinuxカーネルの実装では、sendfileが/dev/nullを宛先として呼ばれた際には、メモリコピーを一切行わないということがわかりました。ディスクからページキャッシュへのデータのローディングは行われるんですが、そのあとのメモリコピーは行われないという、我々にとって非常に理想的な動作だったわけです。この動作は非常に興味深いものだったので、我々は少し深追いしてLinuxカーネルのコードを読んで、この根拠を探してみました。深くは説明しませんが、sendfileというAPIは、Linuxカーネルの内部ではspliceという処理にマップされます。このspliceというのはデバイスドライバごとに実装を持っているというものになっています。ここにお見せしているnullデバイスのデバイスドライバに実装されたsplice_write_nullという関数は、内部的にpipe_to_nullという関数を呼んでいるんですが、そのpipe_to_nullの中ではメモリコピーが一切行われていないということがわかると思います。これによって、我々は確証を得ることができました。レイテンシの短縮も期待さて、このページキャッシュのwarmupという処理をブローカーに実装する必要があります。やり方としては非常にシンプルなものです。これはprepareForReadというページキャッシュのwarmupを行っている部分のみを抜き出してきたものです。やっていることは/dev/nullに対してファイルをオープンして、それにFileChannelのtransferToというメソッドを呼んでいるだけになります。javaの標準ライブラリに含まれているFileChannelが持つtransferToというメソッドは、内部的に使用可能であればsendfileを使うというように実装されています。我々はJNIを用いるようなプラットフォーム固有の実装を入れることなく、ピュアなjavaのコードのみで実装を完了することができました。これを実際に我々のプロダクションシステムに適用してデプロイしてみました。結果として問題は完全に解決されました。APIのレスポンスタイムには一切悪化がないことが見てわかると思います。結果を振り返るさて、我々はこのような成果を出した際に、それをオープンソースのコミュニティに還元するということを非常によくやります。今回のケースについても例外ではなくて、すでにアップストリームにチケットが作られていて、現在議論中ではありますが、もしこれがマージされればKafkaのレイテンシが50倍から100倍縮むということが期待できます。それ以外にも数多くのパフォーマンス改善に関するパッチというのを過去には貢献してきていて。我々はこういった活動を通じて、全世界のKafkaユーザーに対しても貢献しています。詳細にご興味がおありの方は、これらのイシューIDを検索すればアップストリームのチケットを見ることができますので、ぜひご覧になってください。それでは、まとめたいと思います。今日は我々がLINEにおいて運用している大規模な全社向けKafkaプラットフォームについて、我々がその裏でどのようなエンジニアリングを行っているのかについてお話ししました。KafkaのQuotaという機能やSystemTapといったツールを利用しながら、システムを深く理解したパッチなどの適用を通じて種々の問題を解決したのち、我々のホスティングポリシーは非常にうまくいっていると思います。データハブとしてのコンセプトを保ちながら、サポートするシステムやサービスの数に対して、オペレーションコストを比例して増やしていくことなくプラットフォームを効率的に維持できています。我々はそこで発生したパフォーマンス改善などの成果をオープンソースのコミュニティにも還元していくことによって、世界中のKafkaユーザーに対しても貢献してきています。今日の話を通じて、我々が日々行っているエンジニアリングがどのようなものであるのか、みなさまに少しでもご理解いただけていれば幸いです。最後に、今後の展望について、大きく2つの軸からお話ししたいと思います。1つ目はKafkaプラットフォーム自体の進化、今後についてです。現在、我々はいろいろな課題を持っており、そのうちの大きなものの1つとして、クライアントの標準化と管理があります。我々のプラットフォームを使っている多くのクライアントには、さまざまなフレームワークやバラバラのKafkaのバージョンを使ってしまっているという現状があります。これによって、我々が例えばクライアントをサポートするという際に、そこにあるさまざまな差異によって非常に非効率であったり難しい点というが数多くあります。Frameworkの開発やクライアントSDKの開発を通じてこのクライアントを標準化していき、さらに我々が管理しやすくすることを目指しています。高いアベイラビリティを目指すことも重要です。実は我々のプラットフォームは、最近初めて非常に大きな障害を経験しました。そこで見えてきたさまざまな問題は、今後の我々にとって多くの課題を残すものになりました。継続して高いアベイラビリティのためのエンジニアリングを行っていく必要があります。1つの軸としてSREという新たなチームの創設が現在計画されています。これは、現状ミドルウェアやサービスごとにバラバラなチームで、それぞれの中で行われてしまっているリライアビリティエンジニアリングやパフォーマンスエンジニアリングといったものの中で、とくにオペレーティングシステムレベルでの知識やツール、また特定のソフトウェアに依らない知識やツールについて、なるべくチームをまたいで共有することによって、全社的なサービスの信頼性向上に貢献していくというような計画です。いったドメインに興味をお持ちの方とぜひお話ししたいと思っています。私の話はこれで以上になります。本日はどうもありがとうございました。DAYに関する記事をまとめています。DAYに関する新着記事が公開された際に、通知を受け取ることができます。