Fluentd で、ネストされた特定のハッシュの中身だけを転送する

概要

Fluentd で渡ってくるイベント（JSON）内の、
ネストされた特定のハッシュの中身だけを転送する方法。

例：
こういうイベント（データ）がある場合、
{"data": {"name": "hanzo", "skill": "Python"}, "memo": "kesu"}

data フィールド（項目）の中身（ハッシュ）だけを転送したいケース。
{"name": "hanzo", "skill": "Python"}

用語

Fluentd の詳細な説明は省きます。

• Fluentd
  • ログ収集で使うツール。
  • 収集方法を定義ファイルに記載する必要があり、
    大きく source（入力）=> filter（整形） => match（出力）に分かれる。
• イベント
  • Fluentd で処理されるデータ1行のこと。
• フィールド
  • イベント内の項目のこと。
  • 例：{"name": "hanzo", "skill": "Python"} => name や skill がフィールド
• ハッシュ
  • key, value で構成されたフィールド。
  • 例：{"data": {"name": "hanzo", "skill": "Python"}} => data フィールドの中身がハッシュ

考え方

特定のフィールドのハッシュを展開するだけなら、
parser filter plugin で実現できます。

ただし、この plugin を使うためには一つ制約があります。
公式ドキュメントによると、

The parser filter plugin "parses" string field in event records 
※訳：parser filter plugin は、イベントレコード内の文字列フィールドをパースする

つまり、展開する対象（パース対象）が 文字列 である必要があります。

通常、Fluentd で JSON 形式のデータを取得すると、
全て hash に変換されてしまうため、
そのまま parser filter plugin にかけても抽出されません。
　※ハッシュの部分をあえて文字列で入れればできますが。。

なので事前に、取りだしたいハッシュだけを、
JSON 文字列に変換する必要があります。

例：
こういうハッシュを
{"data": {"name": "hanzo", "skill": "Python"}}
こんな感じで文字列にする。
{"data": "{\"name\": \"hanzo\", \"skill\": \"Python\"}"}
　※展開したいフィールド（例なら data）のハッシュだけを文字列にする

filter の書き方

source, match の記載は省略する。

その1：ハッシュを展開する（他のフィールドは捨てる）

ハッシュを展開して転送する。

# 文字列変換
<filter sometag>
  @type record_transformer
  # JSON 文字列に変換したフィールドだけを残す
  renew_record true
  enable_ruby true
  <record>
    # ハッシュ（data の中身）を JSON 文字列に変換する
    obj ${record["data"].to_json}
  </record>
</filter>

# parser で JSON 文字列を展開する
<filter sometag>
  @type parser
  key_name obj
  <parse>
    @type json
  </parse>
</filter>

実行例

# input data:
{"data": {"name": "hanzo", "skill": "Python"}, "memo": "kesu"}
# output data:
{"name":"hanzo","skill":"Python"}

その2：ハッシュを展開し、かつ他のフィールドも残す

展開したハッシュと、その他のフィールドを合わせて転送する。

# 文字列変換
<filter sometag>
  @type record_transformer
  renew_record true
  enable_ruby true
  <record>
    # ハッシュ（data の中身）を JSON 文字列に変換する
    obj ${record["data"].to_json}
    # 他に必要なフィールドがあれば追加する
    memo ${record["memo"]}
  </record>
</filter>

# parser で JSON 文字列 と 必要なフィールド を残す
<filter sometag>
  @type parser
  key_name obj
  # フィールドを全部残す
  reserve_data true
  # 展開前のフィールドは消す
  remove_key_name_field true
  <parse>
    @type json
  </parse>
</filter>

実行例

# input data:
{"data": {"name": "hanzo", "skill": "Python"}, "memo": "nokosu", "hoge": "kesu"}
# output data:
{"memo":"nokosu","name":"hanzo","skill":"Python"}

その3：複数のハッシュを展開する

複数のハッシュを展開し、それらをマージして転送する。

# 文字列変換
<filter sometag>
  @type record_transformer
  renew_record true
  enable_ruby true
  <record>
    # 1. ハッシュ同士をマージする
    # 2. マージしたハッシュを JSON 文字列に変換する
    obj ${tmp = record["data"].merge(record["data2"]); tmp.to_json}
  </record>
</filter>

# parser で JSON 文字列を展開する（ここは1回目と同じ処理）
<filter sometag>
  @type parser
  key_name obj
  <parse>
    @type json
  </parse>
</filter>

実行例

# input data:
{"data": {"name": "hanzo", "skill": "Python"}, "data2": {"office": "Tokyo", "skill": "Python3"}}
# output data:
{"name":"hanzo","skill":"Python3","office":"Tokyo"}
  => skill は二つのハッシュに共通するため、マージした方で上書きされる

お試しコード

source => filter => match の全部載せ。
とりあえず挙動を試したい場合。

td-agent.conf

<source>
  @type tail
  path /var/tmp/aaa
  pos_file /var/log/td-agent/aaa.pos
  tag sometag
  <parse>
    @type json
  </parse>
</source>

# ハッシュを文字列に変換する
<filter sometag>
  @type record_transformer
  renew_record true
  enable_ruby true
  <record>
    obj ${record["data"].to_json}
  </record>
</filter>

# parser で JSON 文字列を展開する
<filter sometag>
  @type parser
  key_name obj
  <parse>
    @type json
  </parse>
</filter>

<match sometag>
  @type file
  path /var/tmp/bbb
</match>

実行例

# /var/tmp/aaa:  =>以下の内容でファイルを作る
{"data": {"name": "hanzo", "skill": "Python"}, "memo": "kesu"}
# /var/tmp/bbb:
{"name":"hanzo","skill":"Python"}

まとめ

今回の記事を書いた切っ掛けは、
既存アプリケーションが吐いている JSON データを、
Fluentd 経由でとあるログ分析基盤に入れようとしたのが始まりです。

ネストされた項目だけを入れたくて、色々検討しました。

• 入力元から直す！
  開発側に依頼してログを変えてもらう。
  　＝＞改修のインパクトがデカすぎる
• 出力先で何とかする！
  ログ分析基盤にはそのまま入れて、検索時にうまく調整する。
  　＝＞クエリが複雑になる or パースする処理に時間がかかる

検討した結果、こんな感じの板挟みにあったので、
最終的に中間地点の Fluentd（ログ収集）で色々調べてみたところ、
これといった記事が見つからず、
公式のドキュメントを凝視して、今回の方法にたどり着きました。

時間が無いときは、つい結果だけを探してしまいますが、
詰まったときは原点に戻り、公式ドキュメントに目を通すことも大事 だと感じました。

同じような悩みを抱えた方のお役に立てれば光栄です。