exec_filterを使えば指定したコマンドでフィルタリング(っていうか中継?)したり出来る模様

/etc/td-agent/td-agent.conf

<source>
    type tail
    format apache
    path /var/log/nginx/access.log
    pos_file /var/log/td-agent/nginx.pos
    tag accesslog.filter
</source>

<match accesslog.filter>
    type exec_filter
    command /var/lib/fluentd/accesslog_filter.pl
    in_format json
    out_format msgpack
    tag accesslog.filter.output
    flush_interval 1s
</match>

<match accesslog.filter.output>
    type file
    path /var/log/td-agent/accesslog
</match>

/var/log/nginx/access.logをtailした結果をexec_filterで/var/lib/fluentd/accesslog_filter.plをかましてから/var/log/td-agent/accesslogへ持っていく

/var/lib/fluentd/accesslog_filter.pl

#!/usr/bin/env perl

use strict;
use warnings;
use Data::MessagePack;
use JSON::XS qw/decode_json/;
use Try::Tiny;

$| = 1;

my $mp = Data::MessagePack->new;

sub parse_json {
    my ($json_string, $cb) = @_;

    try {
        my $json = decode_json($json_string);
        my $host = $json->{host};
        my $path = $json->{path};

        if ($host and $path) {
            $cb->(+{ host => $host, path => $path });
        }
    } catch {
        warn;
    };
}

while (my $data = <STDIN>) {
    parse_json($data, sub {
        my $json = shift;

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

/var/log/nginx/access.logをtailして、キャッチされた時点で標準入力にin_formatで指定した形式で入ってくる(んだと思われる)。なので、それをdecode_jsonでデコードしてそこからホストとパスだけを取って、MessagePack形式で出力する

結果

2012-11-17T18:34:01+09:00   accesslog.filter.output {"host":"127.0.0.1","path":"/"}
2012-11-17T18:34:01+09:00   accesslog.filter.output {"host":"127.0.0.1","path":"/favicon.ico"}

んな形式でaccesslog.filter.output先で出力される。っていう感じで中間的にプログラムを挟んでごにょごにょ出来る模様

公式ドキュメント: http://docs.fluentd.org/articles/out_exec_filter