📑

高負荷環境でFluentdを安定運用するための3つの観点

2021/06/30に公開

Kubernetes

Fluentd

tech

本記事について

Fluentdは機能としてはシンプルですが、
高負荷環境で安定的に運用するためにはある程度の知識が求められます。

そこで、本記事ではそれなりにログ流量の高い環境下で私が考慮した観点をまとめました。

本記事では、KubernetesでFluentdの信頼性を担保するための3つの観点に加え、

「高負荷時の安定運用」に焦点を当て、

負荷分散
適切なモニタリング
トラブルシューティングとチューニング

の3つの観点について整理しています。

前提となるアーキテクチャ

アーキテクチャとしては実際に私が構築した図の構成を前提とします。

アーキテクチャの特徴

1つのKubernetesクラスタに、FluentdがForwarderとAggregatorという2つのロールで存在しています。

Forwarder
- DaemonSetでデプロイされる
- 各コンテナの出力ログをあつめ、Aggregatorに送信することだけが唯一の責務
- Aggregatorの名前解決はServiceリソースで行う
Aggregator
- StatefulSetでデプロイされ、バッファファイルやposfileの永続化のためにPersistentVolumeがアタッチされる
- ForwarderからTCPでログを受け取る
- filterを用いた加工処理、最終的なデータストアへのログ送信を担う

1. 負荷分散

負荷分散として考慮することは、

Aggregatorの水平スケール手段の確立
複数のAggregatorに均等にログが送られること

です。
AggregatorはStatefulSetなので、HPAを使った自動スケールや手動操作によって簡単に水平スケールすることができます。
しかし、均等にログを送るためには踏み込んだ考慮が必要です。
Aggregatorの前段にL4ロードバランサーやKafkaなどのメッセージングキューを置くことでより精密な分散ができますが、
今回はあくまでもDNSによる名前解決の構成で考えます。

※近い将来にKafkaの導入を検討しています

DNSキャッシュについて考慮する

FluentdはデフォルトでDNSで名前解決したIPを永続的にキャッシュします。
よって、このままだと仮にAggregatorがスケールしたとしても新しく作成されたPodにルーティングされてくれません。
そこでDNSキャッシュの時間を短く or 無効にすることで水平スケールに対応できるようにします。

※その分DNS側に負荷がかかるので許容できない場合はLBやメッセージキューを検討することになります

expire_dns_cache

Client-side DNS round robin

DNSラウンドロビンでリクエストが偏る場合は、クライアントサイドの分散設定を考慮するのが有効です。
forwarder pluginには、dns_round_robin設定があり、
これを有効にすると、DNSから返却されたIPリストからランダムで一つピックアップして使用する、という挙動になります。

2. モニタリングで備える

Fluentdで負荷に備えるためには、モニタリングで見るべきポイントを明確にしておく必要があります。
そこで大前提として意識しなければならないのは、「ログの経路」と、各経路での「単位時間辺りのログが流れる量」です。

ログの経路とログの流量の大原則

ログはForwarder、Aggregatorのそれぞれのポイントで以下のような経路をたどっていきます。

In Forwarder: 監視対象のログファイルから読み取り、加工処理(filter, match)をしてバッファに書き出し
In Forwarder: バッファに溜まったログを、チャンク単位でFlush Queueにキューイング
In Forwarder: Flush QueueからAggregatorへ送信
In Aggregator: Chunk単位でログを受信し、加工処理(filter, match)をしてバッファに書き出し
In Aggregator: バッファに溜まったログを、チャンク単位でFlush Queueにキューイング
In Aggregator: Flush Queueから各送信先へ送信

つまり大原則として、Forwarder、Aggregator両サイドにおいて、

単位時間あたりのログからバッファに書き込まれる量 < 単位時間あたりのバッファからFlush Queueuにキューイングされる量
単位時間あたりのFlush Queueにキューイングされる量 < 単位時間あたりのFlush Queueから宛先に送られる量

である状態をなるべくの間維持しておかなくてはなりません。
そうでなければ、常に一定量のバッファがたまった状態になり一定時間遅れで集計され続けるか、最悪の場合バッファが溢れて容量限界を迎えてしまいます。

以上の点を踏まえて監視を行っていきます。

監視すべきメトリクス

まず他のコンテナと同様に、

CPU使用率
メモリ使用率
プロセスの死活監視
ファイルディスクリプタの使用率(※取得可能な場合)

などはモニタリングします。

それに加えてFluentdは、monitoring-prometheusに記載されている通り、Prometheus形式のメトリクスをexportすることができ、

記載されているmetricsのように、output時に発生したエラー数や、Slow Flushの数、Retryの数などもエラー検知に役に立ちます。

また、負荷計測の観点では下記のメトリクスをチェックしておくことで適切なログ流量設定の参考にすることができます。

ログの流量に関する重要メトリクス

メトリクス名	説明	目的
fluentd_output_status_buffer_total_bytes	buffer stage + buffer queueを合計したログのデータ量	全体的なバッファのたまる速度、はける速度をモニタリング
fluentd_output_status_buffer_stage_byte_size	buffer stage領域に溜まっているログのデータ量	バッファにたまる速度、Flush Queueにキューイングされる速度をモニタリング
fluentd_output_status_queue_byte_size	buffer queue領域に溜まっているログのデータ量	Flush Queueにキューイングされる速度、実際にFlushされる速度をモニタリング
buffer_available_buffer_space_ratios	bufferの有効な残りスペースの割合	バッファが足りそうかをモニタリング

私の推進するプロジェクトでは、上記メトリクスを使ってダッシュボードを作り常に可視化しています。

3. トラブルシューティングに対応する

最後のセクションではトラブルシューティングに役立つコマンド、ノウハウと、おまけとして私が実際に遭遇した障害ケースを紹介します。

前提1: FluentdのThreadモデル

Fluentdの過負荷をトラブルシュートするにあたって、Threadモデルを理解しておく必要があります。

Fluentdで抑えておくべきThreadの種類は以下の3つです。

event_thread
- input pluginが作るevent_loop用のスレッド
- リクエストを受けて加工し、バッファに書き込むまでが仕事
enqueue_thread
- flushできるChunkをflush queueにenqueueするスレッド
flush_thread
- キューイングされたChunkを実際の宛先に送信するスレッド

前提2: トラブルシューティングの流れ

基本的にFluentdが過負荷になったり、ログが送られなくなった場合、私は下記のような流れでトラブルシュートしています。

top -H コマンドで過負荷になっているFluentdのスレッドを特定する
netstat, lsofを使い、接続は確立しているかを確認する
tcpdumpを使い、データは実際に送られているかなどを確認する
straceを使ってstuckしているsyscallを確認する
SIGCONT signalを飛ばしてsigdumpを吐き出し、どこでstuckしているのかを確認する

Fluentdはsigdumpが組み込まれており、SIGCONT signalを飛ばすことで特定の場所にこのようなスレッドダンプのファイルを吐き出してくれます。
つまり、stuckしている場合はこのdumpファイルを吐き出すことで実際にRubyコードのどの部分でstuckしているのかを可視化することができます。

Case1: 1つのChunkサイズが大きすぎてAggregatorが過負荷になった

事象としては下記の状態になっていました。

一定時間ごとにAggregatorのCPU使用率が100%近くで張り付く
Forwarderへtimeout時間までにackを返せずにログ送信が失敗するエラーがたくさん出ている
ずっとCPU使用率が張り付いているわけはなく、定期的にバラバラのaggregatorに負荷が集中して収まるのを繰り返している

よって、負荷が高まっているコンテナ内で

top -H

を実行したところ、前述した1つのevent_threadだけが高負荷なことが分かりました。
つまり、受け取るところ or 加工するところ or バッファに書き込むところのどこかがボトルネックになっていることになります。

また、

netstat

を実行したところForwarderと接続確立していたのは1つだけだったため、リクエストの数が高負荷になっているわけでもなさそうでした。
またバッファに書き込むところが詰まっているならIO待ちの割合が増えそうですがそんなこともなくCPUも高負荷なので、消去法で加工するところが重いと仮説を立てました。

そこでForwarderから送るChunkあたりのサイズを小さくしたところ、無事に解決しました。

参考: Buffering Parameters

Case2: Flush Threadが常時CPU使用率100%近くの過負荷になった

事象としては、

Aggregatorの送信先のElasticsearchクラスタに全くログが送られなくなった
AggregatorコンテナのCPU使用率が常時100%近くで張り付いている

ということが発生しました。
topコマンドの結果、flush threadが張り付いていることが分かりました。
また、netstatしてもESへの接続がなく、tcpdumpで確認しても一切データが送られていない状態であることも分かりました。
そこでflush threadに対してstraceすると、getrandomというsyscallを無限に呼び続けていることが確認できました。
つまり確実にどこかの処理でstuckしているので、SIGCONT signalを投げてスレッドダンプをとったところ、下記のようなダンプが取れました。

Thread #<Thread:0x0000562c82ebbf90@flush_thread_2 /srv/fluentd/lib/ruby/gems/2.7.0/gems/fluentd-1.12.3/lib/fluent/plugin_helper/thread.rb:70 run> status=run priority=0
      /srv/fluentd/lib/ruby/2.7.0/securerandom.rb:123:in `urandom'
      /srv/fluentd/lib/ruby/2.7.0/securerandom.rb:123:in `gen_random_urandom'
      /srv/fluentd/lib/ruby/2.7.0/securerandom.rb:73:in `bytes'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:606:in `random_number'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:606:in `random'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:625:in `block in allocate_request_id'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:622:in `synchronize'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:622:in `allocate_request_id'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:842:in `sender'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:527:in `block in fetch_resource'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:1124:in `block (3 levels) in resolv'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:1122:in `each'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:1122:in `block (2 levels) in resolv'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:1121:in `each'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:1121:in `block in resolv'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:1119:in `each'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:1119:in `resolv'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:521:in `fetch_resource'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:507:in `each_resource'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:402:in `each_address'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:116:in `block in each_address'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:115:in `each'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:115:in `each_address'
      /srv/fluentd/lib/ruby/2.7.0/resolv.rb:58:in `each_address'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/socket.rb:110:in `connect'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/socket.rb:49:in `initialize'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/connection.rb:463:in `new'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/connection.rb:463:in `socket'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/connection.rb:118:in `request_call'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/middlewares/mock.rb:57:in `request_call'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/middlewares/instrumentor.rb:34:in `request_call'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/middlewares/idempotent.rb:19:in `request_call'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/middlewares/base.rb:22:in `request_call'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/middlewares/base.rb:22:in `request_call'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/excon-0.81.0/lib/excon/connection.rb:273:in `request'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/faraday-excon-1.1.0/lib/faraday/adapter/excon.rb:31:in `block in call'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/faraday-1.4.1/lib/faraday/adapter.rb:55:in `connection'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/faraday-excon-1.1.0/lib/faraday/adapter/excon.rb:31:in `call'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/faraday-1.4.1/lib/faraday/rack_builder.rb:154:in `build_response'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/faraday-1.4.1/lib/faraday/connection.rb:492:in `run_request'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/elasticsearch-transport-7.12.0/lib/elasticsearch/transport/transport/http/faraday.rb:48:in `block in perform_request'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/elasticsearch-transport-7.12.0/lib/elasticsearch/transport/transport/base.rb:288:in `perform_request'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/elasticsearch-transport-7.12.0/lib/elasticsearch/transport/transport/http/faraday.rb:37:in `perform_request'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/elasticsearch-transport-7.12.0/lib/elasticsearch/transport/client.rb:191:in `perform_request'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/elasticsearch-api-7.12.0/lib/elasticsearch/api/actions/bulk.rb:69:in `bulk'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/fluent-plugin-elasticsearch-5.0.3/lib/fluent/plugin/out_elasticsearch.rb:1062:in `send_bulk'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/fluent-plugin-elasticsearch-5.0.3/lib/fluent/plugin/out_elasticsearch.rb:875:in `block in write'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/fluent-plugin-elasticsearch-5.0.3/lib/fluent/plugin/out_elasticsearch.rb:874:in `each'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/fluent-plugin-elasticsearch-5.0.3/lib/fluent/plugin/out_elasticsearch.rb:874:in `write'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/fluentd-1.12.3/lib/fluent/plugin/output.rb:1138:in `try_flush'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/fluentd-1.12.3/lib/fluent/plugin/output.rb:1450:in `flush_thread_run'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/fluentd-1.12.3/lib/fluent/plugin/output.rb:462:in `block (2 levels) in start'
      /srv/fluentd/lib/ruby/gems/2.7.0/gems/fluentd-1.12.3/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create'

見る限りfluentd側に問題がありそうなので、issueを掘ってみると全く同じ事象が報告されていたため、
一旦事象が再現しないバージョンに変えてデプロイすることで解決しました。

まとめ

Fluentdを安定運用するために考慮した観点と実際のトラブルケースなどについて紹介してきました。
今回はクライアント側の設定で負荷分散したりと、シンプルさを優先していますが、
将来的にはForwarderとAggregatorの間にKafkaを導入することで、より効率的かつ安全な負荷&障害対策ができることを期待しています。