MySQLのクエロリグをtailして良い感じにfluentdに突っ込むプラグイン

in_mysql_query.rb

#<source>
#  type mysql_query
#  log /var/log/mysqld/query.log
#</source>
# TODO:複数行クエリ対応
# TODO:余計なクエリの除外せねば
# TODO:タイムスタンプをログから取得するように
# TODO:ログイン後にuseされると接続DBがわからない
# TODO:タグ名を変数にしないと
# TODO:td-agent停止時にtailプロセスをkillできてない

class MysqlQuery < Fluent::Input
  Fluent::Plugin.register_input('mysql_query', self)

  def configure(conf)
    super
    @log = conf['log']
    @command = "tail -f #{@log}"
  end

  def start
    super
    @io = IO.popen(@command, "r")
    @pid = @io.pid
    @thread = Thread.new(&method(:run))
  end

  def shutdown
    super
    if @thread.join(60)
      return
    end
    Process.kill(:KILL, @pid)
    @thread.join
  end

  def run
    @io.each_line(&method(:each_line))
  end

private
  def each_line(line)
    line.chomp!

    if /(\d+) Connect\s(\w+)@(\S+) on (\w+)/ =~ line
      id, user, src, db = $1, $2, $3, $4
      @conns[id] = [ user, src, db ]
    elsif /(\d+) Query\s+(.*)$/ =~ line
      id, query = $1, $2
      return unless @conns[id]
      user, src, db = @conns[id]

      tag = "mysql.query"
      time = Time.now.to_i
      record = {
        "user"  => user,
        "src"   => src,
        "db"    => db,
        "query" => query
      }
      Fluent::Engine.emit(tag, time, record)
    elsif /(\d+) Quit/ =~ line
      id = $1
      @conns.delete(id)
    end
  end

end