Productionでtoyama0919/fluent-plugin-exec_cronというプラグイン作って動かしています。

Fluentdのoutput pluginにはTimeslicedOutputとBufferedOutputとがあって、TimeslicedOutputで「ある程度の時刻指定」によるflushが可能です。

ただinput pluginについてはintervalを指定したflushはできますが、時刻指定は出来ません。
Inputの処理が実行される時刻は、再起動した時刻に依存します。

そこでCronによる時刻指定が可能なInput pluginを作ってみました。

似たような話だと、AWSのLambdaもCronによる時刻指定がサポートされました。((レポート) CMP407: Cron としての Lambda: AWS Lambda での呼び出しのスケジュール #reinvent ｜ Developers.IO)

使い方

<source>
  type exec_cron
  tag exec_cron.example
  command echo '{"a":"a"}'
  format json
  cron * * * * *
  #cron */5 * * * * 5分おき
  #cron 0 * * * * 1時間おき
</source>

<match exec_cron.example>
  type stdout
</match>

実行結果はこんな感じ

2015-05-27 13:07:00 +0900 exec_cron.example: {"a":"a"}
2015-05-27 13:08:00 +0900 exec_cron.example: {"a":"a"}
2015-05-27 13:09:00 +0900 exec_cron.example: {"a":"a"}
2015-05-27 13:10:00 +0900 exec_cron.example: {"a":"a"}
2015-05-27 13:11:00 +0900 exec_cron.example: {"a":"a"}
2015-05-27 13:12:00 +0900 exec_cron.example: {"a":"a"}

時刻がピッタリ分刻みで出力されています。

もちろん、再起動しても時間はずれません。

実装的には

in_execと処理は大体同じです。
- in_execには現在、次の処理が終わるまでshutdownを待ってしまう問題があるようです。
intervalによる待ち時間をcron syntaxから動的に計算しています。

自分で作りたい場合

siebertm/parse-cronを使うと簡単に出来ます。

in_exec_cron.rb

  def initialize
    ...
    require 'parse-cron'
    ...
  end

  def configure(conf)
    ...
    @cron_parser = CronParser.new(@cron)
    ...
  end

  def start
    @finished = false
    @thread = Thread.new(&method(:run_periodic))
  end

  def run_periodic
    until @finished
      begin
        secs = @cron_parser.next(Time.now) - Time.now
        sleep secs
        # 処理
      rescue
        log.error "exec failed to run or shutdown child process", :error => $!.to_s, :error_class => $!.class.to_s
        log.warn_backtrace $!.backtrace
      end
    end
  end

まとめ

ある程度時刻を決め打ちたい場合にいかがでしょうか
FluentdのInput Pluginに過ぎないので、Fluentdが動いていなければ当然処理は動きません^^
また、bufferが詰まってInputに戻ってこれない場合も同様です。

参考