Fluentdのtailプラグインでpathで指定したファイル名を後続処理で利用する

Fluentdのtailプラグインを使っている際に、ファイル名に日付等の情報を埋め込んで後続処理で使う方法を調べました。

※WindowsのFluentd使っています

例えば、「C:\data\csv\data.csv.20190913」ような感じで、日付をファイル名に埋め込みます。このファイルの 20190913 の部分を後続で利用したいです。

やってみる

ファイルの中身は簡単なCSVです。

AAA,BBB,CCC
DDD,EEE,FFF

以下のような設定ファイルで実行するとtagに設定してあるtest.*の*部分がファイル名に置き換わります。

<source>
  @type tail
  path C:/data/csv/data.csv.*
  pos_file C:/var/run/pos/data.pos
  refresh_interval 10

  read_from_head true

  <parse>
    @type csv
    keys col1,col2,col3
  </parse>

  tag test.*
</source>

<match test.**>
  @type stdout
</match>

以下が実行結果です。実際のフォルダパスが . で置き換わった形で出力されています。

C:\data\csv\data.csv.20190913　⇨　C:.data.csv.data.csv.20190913

2019-09-20 00:19:27.278545000 +0900 test.C:.data.csv.data.csv.20190913: {"col1":"AAA","col2":"BBB","col3":"CCC"}
2019-09-20 00:19:27.278545000 +0900 test.C:.data.csv.data.csv.20190913: {"col1":"DDD","col2":"EEE","col3":"FFF"}

応用

ここまで来ると、record_transformerのプラグインとか使えば何でもできるようになります。

以下、ファイル名の日付部分を date フィールドとして追加している例です。

<source>
  @type tail
  path C:/data/csv/data.csv.*
  pos_file C:/var/run/pos/data.pos
  refresh_interval 10

  read_from_head true

  <parse>
    @type csv
    keys col1,col2,col3
  </parse>

  tag test.*
</source>

<filter test.**>
  @type record_transformer
  enable_ruby
  <record>
    date ${tag.split(".")[-1]}
  </record>
</filter>

<match test.**>
  @type stdout
</match>

C:\data\csv\data.csv.20190919 というファイルを読み込ませた実行結果です。 date フィールドにファイル名の日付部分を入れ込むことができました。

2019-09-20 00:32:11.229816000 +0900 test.C:.data.csv.data.csv.20190919: {"col1":"AAA","col2":"BBB","col3":"CCC","date":"20190919"}
2019-09-20 00:32:11.229816000 +0900 test.C:.data.csv.data.csv.20190919: {"col1":"DDD","col2":"EEE","col3":"FFF","date":"20190919"}