Fluentd + MySQL でログ保存 & 検索システムを考えてみた話

食べログ DevOps チームの @e-ronny と申します。
この記事は Advent Calendar の15日目の投稿です。

概要

以下のようなシステムの設計を検討する機会があったので、その過程をまとめてみます。

• アプリケーションから特定のタイミングで任意の key - value の形式でログを記録する
• 型は特に考えず文字列としてのみ扱うのでよい
• ログを記録する際にはそれを特定するためにログの種別と ID を指定する
• 記録したログは種別と ID を指定すると O(1) で検索できる
• ログを記録してから検索できるようになるまでは数秒程度の遅延は許容される
• ログを記録する実体はアプリケーション本体の DB とは分ける
• 全文検索するわけでもないのでディスクがとっても無駄になるから
• 各環境(開発・本番等)で同様な構成で構築するが別環境のログが混入することがない

実現方法の検討

具体的な実現方法としては以下のように考えました。

全体的な構成

• ログ収集といえば fluentd かな
• 可用性を考慮した構成 のためには forwarder(AP) & aggregator 構成かな
• ストレージの実体を後々差し替えることも考慮して aggregator とストレージの実体(以下 storage と表記)は別サーバにしよう
• サーバ構築用 ansible playbook を用意しよう
• 環境ごとの差異は変数で吸収しよう

forwarder の構成

• ログの書き出し先パスは in_tail の対象パスに合わせればいいけど、そのままだとディスクを無尽蔵に消費するので logrotate を設定しておこう
• ただし logrotate のタイミングでのログを拾いそびれて欠損しないように注意が必要
• key - value 部分は JSON 形式にするのでログのフォーマットは LTSV にしよう
• ログのフォーマットも JSON にするとネストして可読性が下がるので
• ログの欠損を防ぐため At most once ではなく At least once で aggregator に送ろう (参照)

aggregator の構成

• 基本的には in_forward / out_forward で forwarder と aggregator の仲介をすればいいけど、 out_copy で out_file もしておくとログ調査の際に全 forwarder を見に行かなくてもここに集約されているので楽そう
  • ファイルのフォーマットは in_tail 元に合わせて LTSV にしよう
• 発生元の forwarder も記録しておくと障害対応等で役立ちそう
• storage には記録する必要はない

storage の構成

• ストレージの実体としては手っ取り早く使い慣れた MySQL にしよう
• 将来的には MongoDB や s3 に移行するかも
• key - value を何も考えずに JSON 型 として記録するため MySQL のバージョンは5.7以上を使おう
• ログの種別はテーブル名として、ID はそのテーブルの primary key として表現しよう
• ログの種別をカラムとして表現し、ID との複合 primary key にする手もあるけど種別やログ総量が増えた場合の性能劣化が怖いし innodb_file_per_table にしておけばいらなくなったログ種別のゴミ掃除が楽そう
• ログの種別ごとにテーブルにする代わりに key ごとにカラムを分ける手もあるけどテーブルごとにスキーマが変わって面倒なので json カラムとして単純に扱おう
  • 各テーブルのスキーマが全く同じになるので取り扱いが楽
• MySQL 用の output plugin は 組み込み ではないので プラグイン一覧 で見つけた out_mysql_bulk を使おう
  • 同 gem 内の out_mysql は deprecated
• 更新クエリはこのホスト内で out_mysql_bulk からのみ発生するので更新用ユーザの接続元は localhost だけに絞れそう

サーバ構成

上記の内容をまとめると、サーバ構成は以下のようになります。

ロール	input プラグイン	output プラグイン	fluentd 以外のミドルウェア等
fowarder(AP)	in_tail(with parser_ltsv)	out_forward	アプリケーション本体
aggregator	in_forward	out_copy out_file(with formatter_ltsv) out_forward
storage	in_forward	out_mysql_bulk	MySQL

各ロールのざっくりした説明

それぞれのロールにおいて、設定の一部を記載してその意図やはまりかけた点などを説明します。
なお、以下の説明は省略します。

• forwarder
• アプリケーション本体が in_tail 対象のログを出力するところ
• aggregator
• 集約したログの取り扱い(バックアップ・障害発生時の storage への再送等)
• storage
• 個別のテーブルを作成するところ
• サーバの冗長化
• 参照クエリを投げるところ(対象のテーブルに対して primary key で検索するだけ・負荷が気になるなら replica を立ててそちらに向けてもよい)
• その他
• 外部プラグインを導入する際の設定(こちら をご参照ください)
• buffer まわり
• 監視・モニタリング用設定(こんなの とか こんなの とかあります)

forwarder

in_tail 設定

roles/forwarder/templates/td-agent.conf.j2

<source>
  @type tail

  tag data.{{ storage_env }}
  path {{ in_tail_log_paths }}
  read_from_head true

  <parse>
    @type ltsv
  </parse>
</source>

• storage_env を環境ごとに差し替えることでタグを環境ごとに分け、別の環境から誤って aggregator にイベントを送っても受け付けないようにしています。
• 環境によってログの出力パスが異なるのでそこも変数にしており、環境によっては対象のファイルが複数になる(パスに * を含む)ので read_from_head を指定しています。

out_forward 設定

roles/forwarder/templates/td-agent.conf.j2

<match data.{{ storage_env }}>
  @type forward

  require_ack_response true

{% for host in groups.aggregator %}
  <server>
    host {{ hostvars[host]['inventory_hostname'] }}
  </server>
{% endfor %}
<match>

• 対象のタグを絞り込むことで対応する環境のイベントだけ拾うようにしています。
• aggregator のホストも環境ごとの inventory に合わせて差し替えられるようにしています。
• At least once にするために require_ack_response を指定しています。

logrotate 設定

roles/forwarder/templates/logrotate_forwarder.j2

{{ in_tail_log_paths }} {
  daily
  rotate 7
  missingok
  create
  compress
  delaycompress
}

• 対象のログファイルの指定は in_tail での指定と同じものですが、 logrotate の仕様上記載できるような簡単な記述にしています。(/path/to/in_tail/*/*.log といった感じ)
• in_tail の説明 にある通り、ローテート後のファイルを検出できるように create を指定しています。

aggregator

in_forward 設定

roles/aggregator/templates/td-agent.conf.j2

<source>
  @type forward

  source_hostname_key forwarder_host
</source>

• source_hostname_key を指定することで forwarder のホストを記録するようにしています。

当初はこのオプションに気付かず forwarder 側の filter_record_transformer でやろうとしてました・・・

roles/forwarder/templates/td-agent.conf_old.j2

<filter>
  @type record_transformer

  <record>
    forwarder_host "#{Socket.gethostname}"
  </record>
</filter>

out_forward 設定

roles/aggregator/templates/td-agent.conf.j2

<match data.{{ storage_env }}>
  @type copy

  <store>
    @type forward

    require_ack_response true

    <server>
      host {{ storage_host }}
    </server>
  </store>

  <store>
    @type file

    path /path/to/backup
    append true

    <format>
      @type ltsv
    </format>
  </store>

• forwarder の out_forward でも制限していますが、forward 先ホスト間違いがあっても out_forward しないようにこちらでもタグによって対応する環境のみに制限しています。
• out_copy でバックアップ用のファイルにも書き出しています。

storage

out_mysql_bulk 設定

roles/storage/templates/td-agent.conf.j2

<match data.{{ storage_env }}>
  @type mysql_bulk

  table                    data_${category}
  column_names             id, data
  json_key_names           data
  on_duplicate_key_update  true
  on_duplicate_update_keys data

  <buffer category>
    @type file
    ...
  </buffer>
</match>

• タグによる制限は forwarder/aggregator の out_forward と同様です。
• テーブル名はログ中の category の値から動的に決定します。

chunk の key を指定しないと placeholder として使用できない点にはまりました・・・(参照)

• テーブルのスキーマはログの種別によらず固定なので column_names はべた書きしています。
• ログは JSON 形式でそのままカラムに記録するため当該カラムを json_key_names に指定しています。
• At least once による多重レコードの発生への対策として on_duplicate_key_update と on_duplicate_update_keys を指定しています。(参照)

まとめ

(大枠さえわかってしまえば) td-agent.conf の記述だけで結構なことができる fluentd はさすがだなと思いました。
実装も面白そうなので機会があれば深く読んでみたいと思います。

明日は @tsukasa_oishi さんによる「RubyでCPUコアをフル活用してみた」です。