Filterプラグインをv0.14対応させる #Ruby

アクセスログの解析などで使うために https://github.com/daichirata/fluent-plugin-uri-parser というものを書いているんですが、それをv0.14対応の為に久しぶりに色々いじったのでその時のメモ。

fluent-plugin-uri-parser

このプラグインは、ログに含まれるURIとQuery Stringをparseするためのfilterプラグインです。uri_parserとquery_string_parserが含まれていて、私が使うときはだいたいこの２つを組み合わせて使ってます。以下はREADMEからの抜粋ですが、なんとなくどういう物かがわかると思います。

<filter>
  @type uri_parser
  key_name uri
  inject_key_prefix parsed

  out_key_scheme scheme
  out_key_host host
  out_key_port port
  out_key_path path
  out_key_query query
  out_key_fragment fragment
</filter>
# input string of data: {"uri": "http://example.com/path?foo=bar#t=1"}
# output data: {"parsed.scheme":"http","parsed.host":"example.com","parsed.port":80,"parsed.path":"/path","parsed.query":"foo=bar","parsed.ragment":"t=1"}

<filter>
  @type query_string_parser
  key_name parsed.query
  hash_value_field parsed.query
</filter>
# input string of data: {"parsed.scheme":"http","parsed.host":"example.com","parsed.port":80,"parsed.path":"/path","parsed.query":"foo=bar","parsed.ragment":"t=1"}
# output data: {"parsed.scheme":"http","parsed.host":"example.com","parsed.port":80,"parsed.path":"/path","parsed.query":{"foo":"bar"},"parsed.ragment":"t=1"}

v0.14

そもそもv0.14に対応するというのは一体どういうことなのかという感じですが、大体以下のページを見るとわかります。

Filterプラグインは以下の書き換えを行えば大丈夫なようです。

継承クラスを Fluent::Plugin::Filter に変更
#filter_stream を　#filter へ置き換え
- 別に v0.14 の為というわけではないのですが、v0.14で最適化が入ってる事と、プラグインには #filter か #filter_with_time を実装して欲しいという雰囲気が本体側のコードからも出てたので移行しました
テスト用のドライバの変更

継承クラスの変更

Github diff

# class Fluent::QueryStringParserFilter < Fluent::Filter

module Fluent
  module Plugin
    class QueryStringParserFilter < Filter

プラグインのクラス自体も Fluent::Plugin 以下に定義したほうが良さそうだったのでそのように変更しています。

#filter_stream の置き換え

Github diff

#filter を実装する場合、自分で新しいEventStreamを作る必要は無くレコードが１つずつ渡ってくるようになるので、それに合わせてEventStream周りの操作のコードも消しています。返り値に関しても修正が必要で、レコードを消したい場合はfalseyな値を、それ以外はfilter済みのrecordを返します。

特定の値をtimeに置き換えるなど、timeに関しても操作したい場合は #filter_with_time を実装するようにします。

テストコード修正

Github diff

どちらかと言うと、ここに一番時間がかかりました。

まず、Test Driverを以下のように Fluent::Test::Driver::Filter を使うように書き換えます。

# Fluent::Test::FilterTestDriver.new(Fluent::QueryStringParserFilter, tag).configure(conf)

Fluent::Test::Driver::Filter.new(Fluent::Plugin::URIParserFilter).configure(conf)

新しいDriverは引数にtagを受け取らなるので、 create_driver などのhelper methodを定義している場合は全体的にそこを修正します。

次に、driver#run の部分を修正します。

# d1 = create_driver(config, "test.no.change")
# d1.run do
#   d1.filter({ "query" => "foo=bar&hoge=fuga" }, @time)
# end

d1 = create_driver(config)
d1.run(default_tag: @tag) do
  d1.feed(@time, { "query" => "foo=bar&hoge=fuga" })
end

#filterはもう無いので#feedに置き換えます。 #feed の使い方はこんな感じです。

# d.run do                                                     
#   d.feed('tag', time, {record})                              
#   d.feed('tag', [ [time, {record}], [time, {record}], ... ]) 
#   d.feed('tag', es)                                          
# end                                                          
# d.run(default_tag: 'tag') do                                 
#   d.feed({record})                                           
#   d.feed(time, {record})                                     
#   d.feed([ [time, {record}], [time, {record}], ... ])        
#   d.feed(es)                                                 
# end

filter後のレコードは #filtered_records で受け取れるので、その値をテストするようにアサーションを修正します。

リリース

リリースまでに、以下の手順が推奨されてます。

masterに未リリースがある場合はリリースしておく
masterからfluentd-v0.12ブランチを作成する
fluentdへの依存関係を>= 0.14.0に変更
- v1までplugin apiを変更しないので~>ではなく=>が良いとの事
コードとテストをコミット
CIをRuby 2.1以上のみに変更
バージョンを上げてv0.14対応版をリリース

今回も上記の手順に沿って対応を進めていきました。

おわり

正直、v0.14がどれだけ使われているのかは分からないのでv0.12を切り捨てちゃって良いのか迷うところはあるんですが、もうv0.14が出て暫く経つしv0.14かなり便利になってる所が多いのでみんなもどんどん使っていったらいいんじゃないかな。