Fluent-plugin-tail-mysql

in_tail_mysql.rb

# coding: utf-8

class Fluent::InTailMysql < Fluent::Input
  Fluent::Plugin.register_input('tail_mysql', self)

  require 'mysql'

  config_param :host, :string
  config_param :user, :string
  config_param :password, :string
  config_param :database, :string
  config_param :table, :string
  config_param :column, :string
  config_param :column_type, :string

  config_param :tag, :string

  #config_param :counter_table, :string
  #TODO: in_tail 同様にファイルにカウンタを保存する？

  def start
    $log.info "following tail of #{@database}.#{@table} table #{@column} column"
    @counter = nil
    client = Mysql.connect(@host, @user, @password, @database)
    sql = "select * from #{@table} order by #{@column}"

    loop do 
      time = Fluent::Engine.now
      client.query(sql).each_hash do |record|
        Fluent::Engine.emit(@tag, time, record)
        @counter = record[@column]
      end

      #TODO: カウンタ保存

      if @column_type == 'string'
        sql = "select * from #{@table} where #{@column} > '#{@counter}' order by #{@column}"
      else
        sql = "select * from #{@table} where #{@column} > #{@counter} order by #{@column}"
      end
      sleep 1
    end
  end
end

使い方

※カウンタ保存はまだ実装してない

<source>
  type tail_mysql
  host localhost     #MySQL を稼働させているホスト名
  user test          #MySQL のユーザー名
  password testpass  #MySQL のパスワード
  database test      #対象データベース名
  table sample       #対象テーブル
  column update_at   #対象カラム
  column_type string #対象カラムが文字列型なら string それ以外なら適当に

  tag mysql.tail     #出力タグ
</source>