fluentdのexec_filter Output PluginでPythonプログラムを使う

exec_filter Output Plugin (out_ecec_filter)は要するにフィルタ処理を任意のプログラムを使ってかませられる。例えば次の処理に不要なレコードを落したり、エラーとしてどこかに保存したり。ここでPythonプログラムを使ってみる。

設定

次の通りにしてみた。

<match hoge>
  type       exec_filter
  command    /path/to/python -m myapp.fluent_stream_filter

  time_key   time
  in_format  json
  out_format msgpack

  buffer_type file
  buffer_path /path/to/buffer
  buffer_chunk_limit 8m
  buffer_queue_limit 64
  flush_interval 10s

  tag        fuga
</match>

commandでモジュールとして実行しているのは、importが上手く動作するようにするため。これならPYTHONPATH指定は不要。

ファイルバッファを使っているのは、コードを修正した時にfluendの再起動が必要なため。exec_filter Outputプラグインはexec Outputプラグインと違い、指定したスクリプトは起動後に標準入力の入力待ちとなる。

スクリプト

exec_filter Output Pluginの要求として、スクリプトは標準入力を受けとり、標準出力に結果を吐かなければいけない。

fluent_stream_filter.py

# coding=utf-8
"""
fluentdのout_exec_filterで呼ばれるコード

標準入力から1行ずつ処理を行ない、結果を標準出力に吐く
"""
import json
import logging
import logging.config
import sys
import traceback

import msgpack

logging.config.fileConfig('/path/to/logging.conf')
logger = logging.getLogger('fluent-exec')
logger.propagate = False

def main():
    stdin = sys.stdin
    output(convert(validate(parse(readline(stdin)))))

def readline(stdin):
    for line in stdin:
        yield line

def parse(lines):
    for line in lines:
        yield json.loads(line)

def validate(rows):
    """ エラーデータの除去 """
    for row in rows:
        try:
		     # なんらかの検査
        except Exception, e:
            logger.warn(e)
            logger.warn(row)
            continue
        yield row

def convert(rows):
    """ なんらかの変換処理 """
    for row in rows:
        # do something
        yield row

def output(rows):
	""" 標準出力に吐き出す """
    for row in rows:
        sys.stdout.write(msgpack.packb(row))

if __name__ == '__main__':
    logger.info('Start main')
    try:
        main()
    except Exception, e:
        logger.error(traceback.format_exc())
        raise e
    logger.info('End main')

logging.getLoggerには__name__を渡すのが常套だが、ここではスクリプト起動されるため内容は '__main__' である、なので文字列でロガーの名前を指定している。

テストコード

標準入力を扱うメソッドはパラメータでstdioを受けとるようにすると良いかも。テストで標準入力をそのまま使う方法が見つからなかったのでStringIOのインスタンスを渡して、テストデータが読めるようにした。

# coding=utf-8
import types
from StringIO import StringIO

from nose.tools import ok_, eq_

from myapp import fluent_stream_filter


class TestAll(object):
    def test_readline(self):
        """ readlineのテスト、sys.stdinの代りにStringIOを渡す """
        stdin = StringIO()
        stdin.write(open('/path/to/test_data.log').read())
        stdin.seek(0)

        stream = fluent_stream_filter.readline(stdin)
        ok_(isinstance(stream, types.GeneratorType))
        eq_(stream.next(), u'{1行目の内容}')

    def test_parse(self):
        """ parse のテスト """
        stream = fluent_stream_filter.parse(iter(['{...}', '{...}']))
        ok_(isinstance(stream, types.GeneratorType))
        eq_(stream.next(), {...})
        eq_(stream.next(), {...})

    # (略)

ログ出力での注意

標準出力に処理結果以外の、例えばログを吐き出してしまうと、当然MessagePackやJSONとしてパースできないのでfluentd側でエラーになる。rootロガーにConsoleHandlerがセットされていたりして意図せず標準出力にログを吐いてしまう事故を防ぐために

logger.propagate = False

しておくと安全