fluentd out_exec_filter を使ってみた

はじめに

in_tailでパースされたwebのアクセスログに含まれるクエリストリングを、外部プログラムでさらにパースする例を紹介します。
例えば、クエリストリングに"uid"ってキーが定常的に含まれる場合など、得られた結果をMongoDBなどに突っ込んで、"uid"にインデックスを張っておくと解析システムの開発の助けになるんじゃないでしょうか。

まずは超ベーシックなin_tailの例

fluentdを使ってwebサーバのアクセスログをリモートのサーバに転送する場合、in_tailを使うのがベーシックですよね。
例えば、webサーバがnginxだった場合はこんな感じが超ベーシック。

nginxのログフォーマット設定

combined+レスポンスタイムな感じの一般的なもの。

log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                  '$status $body_bytes_sent "$http_referer" '
                  '"$http_user_agent" $request_time';

in_tailの設定

formatは、http://fluentd.org/doc/plugin.html#tailの例をちょっといじって、responseを追加してます。

<source>
  type tail
  format /^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)" (?<response>[^ ]*))?$/
  time_format %d/%b/%Y:%H:%M:%S %z
  path /var/log/nginx/access.log
  tag nginx.access
</source>

出力

こんな感じのJSONが吐かれますよね。

{
  "host":"192.168.1.4",
  "user":"-",
  "method":"GET",
  "path":"/foo/bar/index.psgi?source=ig&hl=ja&rlz=1G1GGLQ_JAJP314&q=td-agent&oq=td-agent&aq=f&aqi=&aql=&gs_l=igoogle.3...0.0.0.387.0.0.0.0.0.0.0.0..0.0...0.0.",
  "code":"200",
  "size":"0",
  "referer":"-",
  "agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.56 Safari/536.5",
  "response":"0.000"
}

もっと細かくパースしたい！

前述のJSONのpathを見てみてください。クエリストリングがくっついてきています。
webのアクセスログを解析する場合は、クエリストリングをパースしてゴニョゴニョしますよね。
解析のたびにパースするんだし、どーせなら全部fluentdにやってもらいたいところですよね。

in_tailでは難しい？

クエリストリングは、決まった順番で決まったものが来るとかではないので、in_tailですべてのケースにマッチングさせる正規表現を書くのは至難です。(というかやれるのかなぁ？)
それじゃあどうするか。外部のパーサに渡して、結果をまたfluentdに戻せばよさそうですね。
それが出来るのが、out_exec_filterです。

out_exec_filterのおおまかな挙動

参考: Fluentd meetup in japan

fluentdからのSTDINを外部プログラムが受けて、外部プログラムからのSTDOUTがfluentdに返っていきます。

0.10.20から

入出力にJSONとMessagePackを使えるようになりました。2012.06.18現在、公式ドキュメントには載ってないです。
それまでは入出力ともにTSVしか使えず、どのキーを外部プログラムに渡すかと、返ってきた値をどのキーにヒモづけるかを明示的に指定する必要があり、ちょっと使いづらかった感じです。

こんな感じで使ってみました

in_tailによってパースされたデータを、fluentdからparser.plにJSONで渡し、さらにクエリストリングのパースをして、MessagePackでfluentdに返します。
なぜ入力がJSONなのに、出力がMessagePackなのかは後述します。

送信側td-agent.conf

<source>
  type   tail
  path   /var/log/nginx/access.log
  tag    nginx.access
  format /^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")? (?<response>[^ ]*)?$/
  time_format %d/%b/%Y:%H:%M:%S %z
  pos_file /tmp/pos
</source>
<match nginx.access>
  type forward
  host reciever
  flush_interval 1s
</match>

受信側td-agent.conf

exec_filterは送信側でやるとパースの負荷を分散できてよいかも。でもparser.plの変更をデプロイすんのがめんどいので受信側でやってます。

<source>
  type forward
</source>

<match nginx.access>
  type       exec_filter
  command    /path/to/perl /path/to/parser.pl
  in_format  json
  out_format msgpack
  tag        nginx.parsed
  flush_interval 1s
</match>

<match nginx.parsed>
  type file
  path /var/log/nginx/parsed.log
</match>

parser.pl

#!/usr/bin/env perl

use strict;
use warnings;
use Data::Dumper;
use Mojo::Parameters;
use JSON::XS;
use Data::MessagePack;

$| = 1;
my $mp = Data::MessagePack->new();

while ( my $json = <STDIN> ) { 
  my $decode = eval { decode_json($json); };
  next if ($@);

  unless ( defined $decode->{path} ) { 
    print $mp->pack($decode);
    next;
  }

  if ( $decode->{path} =~ s/\?(.+)//g ) { 
    $decode->{query_strings} = Mojo::Parameters->new($1)->to_hash;
  }

  print $mp->pack($decode);
}

出力

こんな感じになります。

{
  "agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.56 Safari/536.5",
  "code":"200",
  "host":"192.168.1.4",
  "method":"GET",
  "path":"/foo/bar/index.psgi",
  "referer":"-",
  "response":"0.000",
  "size":"0",
  "user":"-",
  "query_strings":{
                     "source":"ig",
                     "aqi":"",
                     "aq":"f",
                     "oq":"td-agent",
                     "rlz":"1G1GGLQ_JAJP314",
                     "hl":"ja",
                     "gs_l":"igoogle.3...0.0.0.387.0.0.0.0.0.0.0.0..0.0...0.0.",
                     "q":"td-agent",
                     "aql":""
                   },
}