2018年以降の記事はGitHub Pagesに移行しました

Fluentdの自作プラグインがロードできないのでソースの中身を追ってみる…

Ruby 備忘録 Fluentd

あらすじ

前回までに簡単なプラグインを作成する事はできた。
次はプラグインをGem化しようかなと思って色々いじってたら、実行時になんかうまくプラグインが読めない……。なんで？

参考サイト

今できてる事とできてない事

BundlerでFluentdをインストールし、

$ bundle exec fluentd -c /hoge/fluent.conf -p /hoge/plugin

を実行した時……

  • gem installで入れたfluentdのプラグインはちゃんとロードできる
    • Gemにしちゃえばロードできる
  • プラグインディレクトリ直下(-p /hoge/plugin)にササッと作ってみたプラグインもロードでき、動いた
    • 直下に放り込めばロードできる
/hoge/plugin
└in_hoge.rb              # これはロードできる
  • gem未満、ササッと作った以上のプラグインがロードできない
    • fluent-pluginっぽいディレクトリ構成にした程度のもの。まだローカルに置いてあるだけでgemには成り切れていない
/hoge/plugin
├in_hoge.rb              # これはロードできる
└fluent-plugin-hogehoge
　├lib
　│└fluent
　│　└plugin
　│　　└in_hogehoge.rb  # これがロードできない
　├fluent-plugin-hogehoge.gemspec
　└他

調査結果と(暫定)解決策

fluentdのプラグインロード順番

fluentdコマンドを実行すると、 なんやかんやあって プラグインをロードしにいく。ソースは$FLUENTD_HOME/lib/fluent/plugin.rb

  1. $FLUENTD_HOME/lib/fluent/plugin/直下のrbファイルをロードする
    • in_exec.rbとか
  2. gem_pluginをロードする
    • これは今回空だった
  3. /etc/fluent/plugin/直下のrbファイルをロードする
    • デフォルトのプラグイン置き場
    • デフォルトの設定は$FLUENTD_HOME/lib/fluent/env.rbに定義されている
  4. オプションで-p /hoge/pluginを指定していた場合、/hoge/plugin直下のrbファイルもロードする
    • また、-pオプションは複数指定できる
  5. fluent.confに定義されている type hogeが今までロードしたファイルにあるか確認。あった場合はそのままロードできる
  • なかった場合はもう少し色々な場所の探索にチャレンジする
    • $LOAD_PATHと、gem installしたディレクトリ直下のどこかにlib/fluent/plugin/_.rbが存在するか

解決策

  1. fluentd自体の$LOAD_PATHに追加
    • 本体に毎回修正かけるの？
  2. in_hogehoge.rbをロードしてくれる場所へコピー(デフォルトプラグイン(/etc/fluent/plugin)など)
    • コピペ？
  3. さっさと作ってgem installできるようにもっていく
    • うーん…
  4. オプション-pを複数指定する
    • 上の場合は -p /hoge/plugin -p /hoge/plugin/fluent-plugin-hogehoge/lib/fluent/plugin としてやればよい
      • fluentd本体でわざわざlib/fluent/plugin下を探すようにコーディングされてるのにこんな指定の仕方するものかな？　何か違う気がする……下みたいに未完成のプラグインが増えていくとオプションガンガン長くなっていく？
/hoge/plugin
├in_hoge.rb
├fluent-plugin-hogehoge
│└lib
│　└fluent
│　　└plugin
│　　　└in_hogehoge.rb
├fluent-plugin-hogehoge2
│└lib
│　└fluent
│　　└plugin
│　　　└in_hogehoge2.rb
└fluent-plugin-hogehoge3
　└lib
　　└fluent
　　　└plugin
　　　　└in_hogehoge3.rb

以下、調査ログ。

調査ログ

セットアップをすると指定したディレクトリにconfigファイルとpluginディレクトリができる。

$ fluentd -s fluentd-hoge
Installed fluentd-hoge/fluent.conf.
$ ls fluentd-hoge/
fluent.conf  plugin

pluginディレクトリにそのままプラグイン(rbファイル)を入れておくと、起動時に読み込まれる。

bundle gemコマンドを使ってプラグインを作っていく

$ cd plugin # プラグインを作るからとりあえずpluginディレクトリに移動する
$ bundle gem fluent-plugin-hogehoge
      create  fluent-plugin-hogehoge/Gemfile
      create  fluent-plugin-hogehoge/Rakefile
      create  fluent-plugin-hogehoge/LICENSE
      create  fluent-plugin-hogehoge/README.md
      create  fluent-plugin-hogehoge/.gitignore
      create  fluent-plugin-hogehoge/fluent-plugin-hogehoge.gemspec
      create  fluent-plugin-hogehoge/lib/fluent-plugin-hogehoge.rb
      create  fluent-plugin-hogehoge/lib/fluent-plugin-hogehoge/version.rb
Initializating git repo in /略/fluent-hoge/plugin/fluent-plugin-hogehoge

なんかいろいろ出来たがちょっとルールがあるようなので整形していく。公式サイトより

Installing custom plugins

To install a plugin, put a ruby script to /etc/fluent/plugin directory.

Or you can create gem package that includes lib/fluent/plugin/_.rb file.
TYPE is in for input plugins, out for output plugins and buf for buffer plugins.
It’s like lib/fluent/plugin/out_mail.rb.
The packaged gem can be distributed and installed using RubyGems.
See Searching plugins.

http://fluentd.org/doc/devel.html
  • プラグインは/etc/fluent/pluginにインストールするといいよ
    • また、-pオプションを指定すればfluent-hoge/plugin下も呼んでくれる
  • Gem packageを作る場合はパッケージの下にlib/fluent/plugin/_.rbって名前にしておいてね
    • 他の方が作ったプラグイン＠GitHubを見ていると確かにそうなっていた！
  • はinputプラグインならin、outputプラグインならoutといったように
    • このルールは後ほど効いてくる

ってわけでディレクトリはこんな構成に。

fluent-hoge/
├(略)
├fluent.conf
└plugin/　…　(1)
　└fluent-plugin-hogehoge
　　├Gemfile                      # この辺はほぼいじってない
　　├LICENSE                      # この辺はほぼいじってない
　　├README.md                    # この辺はほぼいじってない
　　├Rakefile                     # この辺はほぼいじってない
　　├fluent-plugin-inputs.gemspec # この辺はほぼいじってない
　　└lib/
　　　└fluent/
　　　　└plugin/
　　　　　└in_inputs.rb

プラグインも作成。ただ{"plugin"=>"yes"}を出力しまくるだけのソース。

class FluPluGem < Fluent::Input
  Fluent::Plugin.register_input("inputs", self)
  def initialize
    super
  end

  def configure(conf)
    super
  end

  def start
    puts "a starts"
    @thread = Thread.new(&method(:run))
  end

  def run
    loop do
      Fluent::Engine.emit("debugx.debug", Fluent::Engine.now, {"plugin" => "yes"})
      sleep(2)
    end
  end
end

fluent.confにプラグインを追加。

<source>
  type inputs
</source>

この状態で起動！

/usr/local/bin/fluentd -c /略/fluent-hoge/fluent.conf -p /略/fluent-hoge/plugin/
2012-09-07 19:57:48 +0900: starting fluentd-0.10.25
2012-09-07 19:57:48 +0900: reading config file path="/略/fluent-hoge/fluent.conf"
2012-09-07 19:57:48 +0900: adding source type="forward"
2012-09-07 19:57:48 +0900: adding source type="inputs"
2012-09-07 19:57:48 +0900: config error file="/略/fluent-hoge/fluent.conf" error="Unknown input plugin 'inputs'. Run 'gem search -rd fluent-plugin' to find plugins"
2012-09-07 19:57:48 +0900: process finished code=256
2012-09-07 19:57:48 +0900: process died within 1 second. exit.

失敗した…。inputsっていうプラグインなぞないって言われた。でも、プラグイン本体(in_inputs.rb)をplugin直下(ディレクトリ構成の(1)の部分)に持ってくると…

/usr/local/bin/fluentd -c /略/fluent-hoge/fluent.conf -p /略/fluent-hoge/plugin/
2012-09-07 20:02:28 +0900: starting fluentd-0.10.25
2012-09-07 20:02:28 +0900: reading config file path="/略/fluent-hoge/fluent.conf"
2012-09-07 20:02:28 +0900: adding source type="forward"
2012-09-07 20:02:28 +0900: adding source type="inputs"
2012-09-07 20:02:28 +0900: adding source type="thread"
2012-09-07 20:02:28 +0900: adding source type="tail"
2012-09-07 20:02:28 +0900: 'pos_file PATH' parameter is not set to a 'tail' source.
2012-09-07 20:02:28 +0900: this parameter is highly recommended to save the position to resume tailing.
2012-09-07 20:02:28 +0900: adding match pattern="stdd.std" type="file"
2012-09-07 20:02:28 +0900: adding match pattern="devid.devid" type="file"
2012-09-07 20:02:28 +0900: adding match pattern="jenkins.**" type="file"
2012-09-07 20:02:28 +0900: adding match pattern="debug.**" type="stdout"
2012-09-07 20:02:28 +0900: listening fluent socket on 0.0.0.0:24224

起動した…。

Or you can create gem package that includes lib/fluent/plugin/_.rb file.

この文の認識が間違ってるのかなー？

よくわからないので本丸を攻めてみる。pluginの処理を記述しているのはそのものplugin.rb、また、それを呼ぼうとしているはsupervisor.rb,engine.rbあたりっぽい。

ロードしているのはplugin内だろうと思い、いろいろ出力してみる。

  def load_plugins
    dir = File.join(File.dirname(__FILE__), "plugin")
puts "!load_plugins:#{dir}"
    load_plugin_dir(dir)
    load_gem_plugins
  end

  def load_plugin_dir(dir)
    dir = File.expand_path(dir)
puts "!load_plugin_dir:#{dir}"
    Dir.entries(dir).sort.each {|fname|
puts "!load_plugin_dir:#{fname}"
      if fname =~ /\.rb$/
        require File.join(dir, fname)
      end
    }
    nil
  end

  private
  def load_gem_plugins
    return unless defined? Gem
    plugins = Gem.find_files('fluent_plugin')
puts "!load_gem_plugins:#{plugins}"
    plugins.each {|plugin|
      begin
        load plugin
      rescue ::Exception => e
        msg = "#{plugin.inspect}: #{e.message} (#{e.class})"
        $log.warn "Error loading Fluent plugin #{msg}"
      end
    }
  end

  def try_load_plugin(name, type)
puts "!try_load_plugin:#{name}, #{type}"
    case name
    when 'input'
      path = "fluent/plugin/in_#{type}"
    when 'output'
      path = "fluent/plugin/out_#{type}"
    when 'buffer'
      path = "fluent/plugin/buf_#{type}"
    else
      return
    end

    # prefer LOAD_PATH than gems
    files = $LOAD_PATH.map {|lp|
puts "!try_load_plugin:#{lp}"
      lpath = File.join(lp, "#{path}.rb")
      File.exist?(lpath) ? lpath : nil
    }.compact
    unless files.empty?
      # prefer newer version
      require files.sort.last
      return
    end

    # search gems
    if defined?(::Gem::Specification) && ::Gem::Specification.respond_to?(:find_all)
      specs = Gem::Specification.find_all {|spec|
        spec.contains_requirable_file? path
      }

      # prefer newer version
      specs = specs.sort_by {|spec| spec.version }
      if spec = specs.last
        spec.require_paths.each {|lib|
          file = "#{spec.full_gem_path}/#{lib}/#{path}"
          require file
        }
      end

実行結果はこうなる。

$ /usr/local/bin/fluentd -c /略/fluent-hoge/fluent.conf -p /略/fluent-hoge/plugin/
2012-09-07 19:37:32 +0900: starting fluentd-0.10.25
2012-09-07 19:37:32 +0900: reading config file path="/略/fluent-hoge/fluent.conf"

→はじめはfluent本体のpluginを呼んでる。

!load_plugins:/usr/local/lib/ruby/gems/1.9.1/gems/fluentd-0.10.25/lib/fluent/plugin
!load_plugin_dir:/usr/local/lib/ruby/gems/1.9.1/gems/fluentd-0.10.25/lib/fluent/plugin
!load_plugin_dir:.
!load_plugin_dir:..
!load_plugin_dir:buf_file.rb
!load_plugin_dir:buf_memory.rb
!load_plugin_dir:buf_zfile.rb
!load_plugin_dir:in_exec.rb
!load_plugin_dir:in_forward.rb
!load_plugin_dir:in_gc_stat.rb
!load_plugin_dir:in_http.rb
!load_plugin_dir:in_object_space.rb
!load_plugin_dir:in_status.rb
!load_plugin_dir:in_stream.rb
!load_plugin_dir:in_syslog.rb
!load_plugin_dir:in_tail.rb
!load_plugin_dir:out_copy.rb
!load_plugin_dir:out_exec.rb
!load_plugin_dir:out_exec_filter.rb
!load_plugin_dir:out_file.rb
!load_plugin_dir:out_forward.rb
!load_plugin_dir:out_null.rb
!load_plugin_dir:out_roundrobin.rb
!load_plugin_dir:out_stdout.rb
!load_plugin_dir:out_stream.rb
!load_plugin_dir:out_test.rb

→gemはない。

!load_gem_plugins:[]

→ここから他のpluginディレクトリ。まずは/etc/fluent/plugin。

!load_plugin_dir:/etc/fluent/plugin
!load_plugin_dir:.
!load_plugin_dir:..

→そして-pオプションで指定したディレクトリ。

!load_plugin_dir:/略/fluent-hoge/plugin
!load_plugin_dir:.
!load_plugin_dir:..
!load_plugin_dir:fluent-plugin-hogehoge
!load_plugin_dir:simple_thread_plugin.rb

あれ？ディレクトリの上さらっと舐めるだけ？

ここからtypeがaddされている。みつかったものは特に何も表示されず次へ次へ……。

2012-09-07 20:16:24 +0900: adding source type="forward"
2012-09-07 20:16:24 +0900: adding source type="thread"
2012-09-07 20:16:24 +0900: adding source type="tail"
2012-09-07 20:16:24 +0900: 'pos_file PATH' parameter is not set to a 'tail' source.
2012-09-07 20:16:24 +0900: this parameter is highly recommended to save the position to resume tailing.

で、今回見つかってないtypeが出てくると、try_load_pluginが呼ばれる。

2012-09-07 20:16:24 +0900: adding source type="inputs"
!try_load_plugin:input, inputs
!try_load_plugin:/usr/local/lib/ruby/gems/1.9.1/gems/http_parser.rb-0.5.3/lib
!try_load_plugin:/usr/local/lib/ruby/gems/1.9.1/gems/msgpack-0.4.7/lib
!try_load_plugin:/usr/local/lib/ruby/gems/1.9.1/gems/yajl-ruby-1.1.0/lib
!try_load_plugin:/usr/local/lib/ruby/gems/1.9.1/gems/iobuffer-1.1.2/lib
!try_load_plugin:/usr/local/lib/ruby/gems/1.9.1/gems/cool.io-1.1.0/lib
!try_load_plugin:/usr/local/lib/ruby/gems/1.9.1/gems/http_parser.rb-0.5.3/lib
!try_load_plugin:/usr/local/lib/ruby/gems/1.9.1/gems/fluentd-0.10.25/lib
!try_load_plugin:/usr/local/lib/ruby/gems/1.9.1/gems/json-1.7.4/lib
!try_load_plugin:/usr/local/lib/ruby/site_ruby/1.9.1
!try_load_plugin:/usr/local/lib/ruby/site_ruby/1.9.1/i686-linux
!try_load_plugin:/usr/local/lib/ruby/site_ruby
!try_load_plugin:/usr/local/lib/ruby/vendor_ruby/1.9.1
!try_load_plugin:/usr/local/lib/ruby/vendor_ruby/1.9.1/i686-linux
!try_load_plugin:/usr/local/lib/ruby/vendor_ruby
!try_load_plugin:/usr/local/lib/ruby/1.9.1
!try_load_plugin:/usr/local/lib/ruby/1.9.1/i686-linux
!try_load_plugin:/usr/local/lib/ruby/gems/1.9.1/gems/fluentd-0.10.25/lib
2012-09-07 20:16:24 +0900: config error file="/略/fluent-hoge/fluent.conf" error="Unknown input plugin 'inputs'. Run 'gem search -rd fluent-plugin' to find plugins"
2012-09-07 20:16:24 +0900: process finished code=256
2012-09-07 20:16:24 +0900: process died within 1 second. exit.

まずプラグインの種類によって読み込まれるパスとファイル名が決まるみたい。

    case name
    when 'input'
      path = "fluent/plugin/in_#{type}"
    when 'output'
      path = "fluent/plugin/out_#{type}"
    when 'buffer'
      path = "fluent/plugin/buf_#{type}"
    else
      return
    end

Or you can create gem package that includes lib/fluent/plugin/_.rb file.

がここで効いてくるわけか。

そしたら$LOAD_PATHの中にfluent/plugin/in_inputs.rbがないか探しに行ってる。そしてそこにもなかった場合、インストールしたGemの中を探す。

    # prefer LOAD_PATH than gems
    files = $LOAD_PATH.map {|lp|
puts "!try_load_plugin:#{lp}"
      lpath = File.join(lp, "#{path}.rb")
      File.exist?(lpath) ? lpath : nil
    }.compact
    unless files.empty?
      # prefer newer version
      require files.sort.last
      return
    end

    # search gems
    if defined?(::Gem::Specification) && ::Gem::Specification.respond_to?(:find_all)
      specs = Gem::Specification.find_all {|spec|
        spec.contains_requirable_file? path
      }

      # prefer newer version
      specs = specs.sort_by {|spec| spec.version }
      if spec = specs.last
        spec.require_paths.each {|lib|
          file = "#{spec.full_gem_path}/#{lib}/#{path}"
          require file
        }
      end

っていう流れになっているらしい。

うーんGemで入れたら読めるんだけどなー。