Sidekiq::Schedulerで動的にタスクをぶち込む

お魚の中でもメバルやタチウオなどは夜行性のものとして有名で、特にタチウオなどは昼間沖の深めにいて夜には岸辺の浅いところにくるなど、かなり規則的なパターンで活動するのだそうです。

さて、rubyで野良アプリ書くときバッチ処理エンジンにはSidekiqを好んで使っています。
Sidekiqには日次処理などを簡単に行うため(*1)にSidekiq::Schedulerという拡張gemがあり(*2)、任意の定義済みワーカを例えばcronライクに繰り返し実行することができます。
基本的には各ワーカの実行条件やパラメータをYAMLの形で全て静的に記述し起動時に読み込むというスタイルなのですが、時には動的にワーカをスケジュールに突っ込みたいことがあります。

Sidekiq::Schedulerはもちろんそれもサポートしています。
ただmoove-it/sidekiq-schedulerの説明がいまいちだったので、備忘録も兼ねてまとめました(*3)。

初期化

まず、初期化の段階でSidekiq::Schedulerがタスクの動的な追加削除を受け付けられるよう設定しなければいけません。
Railsなどであればinitializerに次のようなコードを書くなり、

Sidekiq::Scheduler.dynamic = true

Sidekiqを直接走らせる場合には設定を書いたyamlを渡すなり、

% cat config/sidekiq.yml
 :dynamic: true
% sidekiq -C config/sidekiq.yml

バッチ側だけでなく、フロント側やコンソールなど動的にタスクを突っ込む元となる全ロールでこの設定がなされている必要があります。

タスクの動的な追加

Reloading the schedulesに説明があるとおりですが、Sidekiq.set_scheduleメソッドを使って追加ができます。
第一引数の名前は一意であれば何でも構いません。名前を変えることで、同じワーカーを別のパラメータでスケジュールに突っ込むことができます。
第二引数のハッシュに指定できる項目は静的にyamlに書く際に指定できるものと同じです。キーはシンボルでも文字列でも大丈夫です。
また、スケジュールを突っ込んだあとは、reload_schedule!メソッドによってSidekiq::Schedulerに認識させる必要があります(*4)。

> Sidekiq.set_schedule('DynamicWorker', {every: '10s', class: 'DynamicWorker'})
> Sidekiq::Scheduler.reload_schedule!    # Don't forget!
2016-02-21T15:20:24.545Z 89314 TID-oxsqo8m4g INFO: Reloading Schedule
2016-02-21T15:20:24.546Z 89314 TID-oxsqo8m4g INFO: Loading Schedule
2016-02-21T15:20:24.546Z 89314 TID-oxsqo8m4g INFO: Scheduling DynamicWorker
2016-02-21T15:20:24.548Z 89314 TID-oxsqo8m4g INFO: Schedules Loaded

Railsなどでコンソールを叩いている場合、reloadしたさいに再読み込みされたタスク群のなかに新たなタスク(上記の場合DynamicWorker)が入っていればOKです。

これで、あとは勝手にキューイングしてくれるようになります。

...
2016-02-21T15:11:37.750Z 75551 TID-ouhmu4v8o INFO: queueing DynamicWorker (DynamicWorker)
2016-02-21T15:11:48.049Z 75551 TID-ouhmu4v8o INFO: queueing DynamicWorker (DynamicWorker)
2016-02-21T15:11:58.346Z 75551 TID-ouhmu4v8o INFO: queueing DynamicWorker (DynamicWorker)
...

で、Sidekiqが後ろで走っていれば勝手に実行してくれます。

...
worker.1 | 2016-02-21T15:14:59.644Z 75671 TID-ouxhncu88 DynamicWorker JID-a86e06a4ca000df092e1424c INFO: start
worker.1 | 2016-02-21T15:15:04.648Z 75671 TID-ouxhncu88 DynamicWorker JID-a86e06a4ca000df092e1424c INFO: done: 5.004 sec
worker.1 | 2016-02-21T15:15:09.661Z 75671 TID-ouxhncu88 DynamicWorker JID-2a7c2d743b88769da69d0a0d INFO: start
worker.1 | 2016-02-21T15:15:14.666Z 75671 TID-ouxhncu88 DynamicWorker JID-2a7c2d743b88769da69d0a0d INFO: done: 5.004 sec
...
タスクの動的な削除

上述の方法で動的に追加したタスクにかぎらず、Sidekiq::Schedulerが認識しているタスクはその名前を指定して削除することができます。

> Sidekiq.remove_schedule('DynamicWorker')
> Sidekiq::Scheduler.reload_schedule!    # Don't forget!
2016-02-21T15:21:40.522Z 89314 TID-oxsqo8m4g INFO: Reloading Schedule
2016-02-21T15:21:40.522Z 89314 TID-oxsqo8m4g INFO: Loading Schedule
2016-02-21T15:21:40.522Z 89314 TID-oxsqo8m4g INFO: Schedules Loaded  # "DynamicWorker" no longer exists!!

remove_scheduleは指定された名前のタスクが存在してもしなくてもtrueを返すので値を使わないよう要注意。

個人的にはset_schedule!/remove_schedule!なるメソッドもあってreloadを明示的に呼ばなくてもいいようになってると嬉しい人も多いかもなーなどと思っていました。ぼくのことです。

*1: ActiveJobに乗り移る気にならないのはこれがシンプルにできないからだったりします

*2: ちなみにResqueの場合Resque::Schedulerがあり、似たテイストで使えて便利です。どちらも後ろにrufus-schedulerを使っています。

*3: Readmeを改善するPR送ればいいだけの話なんですが。

*4: Sidekiq.reload_schedule!というメソッドもありますがSidekiq::Scheduler.reload_schedule!を呼ぶ必要があります。落とし穴です。はまったのはぼくだけですかすみません。

コメントを残す

メールアドレスが公開されることはありません。