ラベル sidekiq の投稿を表示しています。すべての投稿を表示

2023年7月2日日曜日

ruby sidekiq-scheduler を試す

概要
環境
インストール
サンプルコード (ワーカー)
スケジュール設定 (yaml ファイル)
スケジューラ起動
動作確認
ちょっと応用: スケジューラの登録を動的に行う
最後に
参考サイト

概要

Sidekiq のキューイングを cron のように定期的に実行できるのが sidekiq-scheduler です
今回はインストールから簡単な動作確認までやってみました

環境

macOS 13.4
redis 7.0.11
Ruby 3.2.2
sidekiq-scheculer 5.0.3

インストール

vim Gemfile

gem 'sidekiq-scheduler'

bundle config path vendor
bundle install

サンプルコード (ワーカー)

エンキューはスケジューラが実行するため必要なのはワーカーになります
ワーカーは普通の sidekiq のように perform を実装するだけです

vim worker.rb

# frozen_string_literal: true

require 'sidekiq-scheduler'

# ワーカークラス
class HelloWorld
  include Sidekiq::Worker

  def perform
    puts 'Hello world'
  end
end

スケジュール設定 (yaml ファイル)

この設定ファイルを元にスケジューラがエンキューします
sidekiq-scheduler の cron 設定は秒単位まで指定できます

vim config/sidekiq.yml

:scheduler:
  :schedule:
    hello_world:
      cron: '0 * * * * *'  # 毎分0秒に実行
      class: HelloWorld

スケジューラ起動

あとはワーカーとともにスケジューラを起動するだけです

bundle exec sidekiq -r ./worker.rb

動作確認

ワーカーのログに毎分 0 秒のときに以下のようなログが出力されれば OK です

2023-06-12T01:28:09.357Z pid=87946 tid=1x4e INFO: Running in ruby 3.2.2 (2023-03-30 revision e51014f9c0) [arm64-darwin22]
2023-06-12T01:28:09.357Z pid=87946 tid=1x4e INFO: See LICENSE and the LGPL-3.0 for licensing details.
2023-06-12T01:28:09.357Z pid=87946 tid=1x4e INFO: Upgrade to Sidekiq Pro for more features and support: https://sidekiq.org
2023-06-12T01:28:09.357Z pid=87946 tid=1x4e INFO: Sidekiq 7.1.1 connecting to Redis with options {:size=>10, :pool_name=>"internal", :url=>nil}
2023-06-12T01:28:09.359Z pid=87946 tid=1x4e INFO: Sidekiq 7.1.1 connecting to Redis with options {:size=>5, :pool_name=>"default", :url=>nil}
2023-06-12T01:28:09.360Z pid=87946 tid=1x4e INFO: Loading Schedule
2023-06-12T01:28:09.360Z pid=87946 tid=1x4e INFO: Scheduling hello_world {"cron"=>"0 * * * * *", "class"=>"HelloWorld", "queue"=>"default"}
2023-06-12T01:28:09.527Z pid=87946 tid=1x4e INFO: Schedules Loaded
2023-06-12T01:28:09.527Z pid=87946 tid=1x4e INFO: Starting processing, hit Ctrl-C to stop
2023-06-12T01:29:00.036Z pid=87946 tid=1wwa INFO: queueing HelloWorld (hello_world)
2023-06-12T01:29:00.037Z pid=87946 tid=1wwu class=HelloWorld jid=a2522a3f2f3de9a7c706f1f9 INFO: start
Hello world
2023-06-12T01:29:00.038Z pid=87946 tid=1wwu class=HelloWorld jid=a2522a3f2f3de9a7c706f1f9 elapsed=0.001 INFO: done

ちょっと応用: スケジューラの登録を動的に行う

yaml ファイルでスケジュールを管理するのは大変なので ruby から動的に登録することも可能です

vim job_create.rb

# frozen_string_literal: true

require 'sidekiq-scheduler'

Sidekiq.set_schedule('hello', { 'every' => ['1m'], 'class' => 'HelloWorld' })

これを起動することでスケジューラを登録できます

bundle exec ruby job_create.rb

redis を確認するとスケジューラが登録されています

redis-cli keys '*'
1) "schedules_changed"
2) "schedules"

redis-cli hgetall 'schedules'
1) "hello"
2) "{\"every\":[\"1m\"],\"class\":\"HelloWorker\"}"

また設定ファイルでは dynamic というオプションを true にします

vim config/sidekiq.yml

:scheduler:
  :dynamic: true

あとはワーカーを起動して毎分実行されることを確認しましょう

bundle exec sidekiq -r ./worker.rb

ワーカーがスケジューラを認識しない場合はサイド job_create.rb を実行してみてください
同一スケジューラ名の場合は新規で追加はせずスケジューラが更新されるだけになります
またスケジューラの変更はワーカーは動的に行ってくれます

最後に

とりあえず動かすところまでやってみました
スケジュールの設定が yaml ファイルなのが特徴的なところかなと思います
また秒単位まで指定できるので細かい時間指定も可能かなと思います

参考サイト

https://github.com/sidekiq-scheduler/sidekiq-scheduler

2021年5月14日金曜日

Sidekiq で DDD を使ってみた

概要
環境
前提
redis を使ってリポジトリを作成
- ドメインモデルの修正
Sidekiq ワーカーの作成
Sinatra アプリの修正
- Sidekiq の設定を rack 起動時に読み込ませる
動作確認
最後に

概要

前回 Sinatra に DDD を適用してみました今回は更に Sidekiq を追加して非同期処理に DDD を適用してみますなお以下は前回のアプリを修正しているので修正箇所とポイントのみ紹介しています

環境

macOS 11.3.1
Ruby 3.0.0
- sinatra 2.1.0
- sidekiq 6.2.1

前提

Sidekiq をドメインとして抽出してもいいのですがどちらかというとプレゼンテーションレイヤーに近い機能だと思うので今回はプレゼンテーションレイヤーとして定義します当然プレゼンテーションレイヤー以外でも定義可能なので今回の方法は参考程度に御覧ください

redis を使ってリポジトリを作成

Sidekiq の場合 Sinatra のプロセスとメモリ空間を共有することはできません前回はインメモリリポジトリでデータを共有できないので redis を使ったリポジトリを作成します

ユーザデータは key/value で key にユーザ名を value にはユーザ情報を文字列の JSON で登録します

vim redis/user/redis_user_repository.rb

require 'redis'
require 'json'

require './domain/model/user/user'
require './domain/model/user/user_id'
require './domain/model/user/user_name'
require './domain/model/user/user_mail'

class RedisUserRepository
  def initialize
    @client = Redis.new(
      :host => 'localhost',
      :port => 6379
    )
  end

  def find(name)
    user_value = @client.get(name.value)
    return nil if user_value.nil?
    user_hash = JSON.parse(user_value)
    id = UserId.new(user_hash["id"])
    name = UserName.new(user_hash["name"])
    mail = UserMail.new(user_hash["mail"])
    User.restore(name, mail, id)
  end

  def save(user)
    @client.set(user.name.value, user.to_h.to_json)
  end
end

リポジトリが増えてきたらインタフェース的なクラスを一つ作成してそれを継承して実装するほうが良いですまたリポジトリのインタフェースは domain/model/user/user_repository_interface.rb という感じでドメイン配下に作成します

ドメインモデルの修正

Redis リポジトリ用に to_h メソッドを追加しました

vim domain/model/user/user.rb

require 'securerandom'
require './domain/model/user/user_id'
require './domain/model/user/user_name'
require './domain/model/user/user_mail'

class User
  def initialize(name, mail, id=nil)
    raise Exception.new("A name must be in the UserName class") unless name.is_a?(UserName)
    raise Exception.new("A mail must be in the UserMail class") unless mail.is_a?(UserMail)
    @id = id || UserId.new(SecureRandom.uuid)
    @name = name
    @mail = mail
  end

  def self.restore(name, mail, id)
    raise Exception.new("A name must be in the UserName class") unless name.is_a?(UserName)
    raise Exception.new("A mail must be in the UserMail class") unless mail.is_a?(UserMail)
    raise Exception.new("An id must be in the UserId class") unless id.is_a?(UserId)
    @name = name
    @mail = mail
    @id = id
    return User.new(name, mail, id)
  end

  def change_name(name)
    raise Exception.new("A name must be in the UserName class") unless name.is_a?(UserName)
    @name = name
  end

  def to_h
    {
      "id" => @id.value,
      "name" => @name.value,
      "mail" => @mail.value,
    }
  end

  attr_reader :id, :name, :mail
end

Sidekiq ワーカーの作成

プレゼンテーションレイヤーにワーカーを作成します Sinatra のルーティングで受け取ったパラメータはプリミティブな型のまま Sidekiq に渡されるのでそれを元にコマンドを作成してアプリケーションサービスの機能を呼び出します

vim presentation/worker/user_worker.rb

require 'sidekiq'
require 'logger'

require './domain/service/user_service'

require './redis/user/redis_user_repository'

require './application/user/user_application_service'
require './application/user/user_create_command'

class UserWorker
  include Sidekiq::Worker

  def perform(name, mail)
    repository = RedisUserRepository.new
    user_service = UserService.new(repository)
    app_service = UserApplicationService.new(repository, user_service)
    create_command = UserCreateCommand.new(name, mail)
    user = app_service.create(create_command)
    logger.info(user.to_h)
  end
end

Sinatra アプリの修正

今回は非同期用の API を一つ追加しています

vim presentation/app.rb

require 'sinatra'
require 'sinatra/json'
require 'json'

require './domain/service/user_service'

require './in_memory/user/in_memory_user_repository'
require './redis/user/redis_user_repository'

require './application/user/user_application_service'
require './application/user/user_create_command'
require './application/user/user_get_command'

require './presentation/worker/user_worker'

class DDDWebApp < Sinatra::Base
  configure do
    # repository = InMemoryUserRepository.new
    repository = RedisUserRepository.new
    user_service = UserService.new(repository)
    app_service = UserApplicationService.new(repository, user_service)
    set :repository, repository
    set :user_service, user_service
    set :app_service, app_service
  end

  get '/user/:name' do
    get_command = UserGetCommand.new(params["name"])
    user = settings.app_service.get(get_command)
    json user.to_h
  end

  post '/user', :provides => 'json' do
    body = JSON.parse request.body.read
    create_command = UserCreateCommand.new(body["name"], body["mail"])
    user = settings.app_service.create(create_command)
    json user.to_h
  end

  post '/async_user', :provides => 'json' do
    body = JSON.parse request.body.read
    result = UserWorker.perform_async(body["name"], body["mail"])
    json result
  end
end

Sidekiq の設定を rack 起動時に読み込ませる

最後に Sidekiq の設定を行いますアプリ起動時に行うのが良いので config.ru を変更します

vim config/sidekiq_config.rb

require 'sidekiq'

class SidekiqConfig
  def self.set_client
    Sidekiq.configure_client do |config|
      config.redis = { url: "redis://#{ENV['REDIS_HOST']}:#{ENV['REDIS_PORT']}" }
    end
  end

  def self.set_server
    Sidekiq.configure_server do |config|
      config.redis = { url: "redis://#{ENV['REDIS_HOST']}:#{ENV['REDIS_PORT']}" }
    end
  end
end

vim presentation/config.ru

require './presentation/app'
require './config/sidekiq_config'

SidekiqConfig.set_client
SidekiqConfig.set_server

run DDDWebApp

動作確認

アプリとワーカーを起動します

bundle exec rackup presentation/config.ru
bundle exec sidekiq -r ./presentation/worker/user_worker.rb

あとは REST にアクセスします新たに作成した非同期 API を使ってユーザ登録が確認できることと元からあった同期 API を使って登録したユーザが取得できることを確認します

curl -XPOST localhost:9292/async_user -d '{"name":"user02","mail":"user01@mail.domain"}'
curl -XGET localhost:9292/user/user01

最後に

今回は Sidekiq のワーカーをプレゼンテーションレイヤーとして定義してみました流れとしてはプレゼンテーション (Sinatra) -> プレゼンテーション (Sidekiq) という流れになっていますプレゼンテーション -> アプリケーションサービスという流れが自然な感じもするのでアプリケーションサービスとして定義しても良いかなと思います

Sidekiq の仕様としてコンストラクタが持てないのとキューイングする際の perform の引数に ruby のオブジェクトは渡せないので Worker 側でアプリケーションサービスの作成を行っていますこれは Sidekiq を使う上ではどうしようもない仕様かなともいます本当はワーカークラスに repository, user_service, app_service を Sinatra 側から引数で渡して持ちたいのですができないので perform 上で生成しています書き方としてはよろしくない部分になっています (DI にしたほうがいい)

ワーカーの返り値が Sidekiq の jid になっています本来であればここで返された値を使って進捗を確認することになるので、進捗を確認できる機能も必要になります

2020年4月18日土曜日

概要

前回 Rails6 と PostgreSQL を使って簡単な Web アプリを作ってみました
今回は Rails6 と Sidekiq を組み合わせて非同期な Web アプリを作ってみます

Redis のインストール

brew install redis
brew services start redis

Sidekiq のインストール

rails new で作成したアプリ配下にある Gemfile を編集します

vim Gemfile

gem "sidekiq"

bundle install

Sidekiq の初期化

config/initializers/sidekiq.rb を作成してここに Redis のエンドポイントなど Sidekiq の初期化に必要なパラメータなどを設定します

vim config/initializers/sidekiq.rb

Sidekiq.configure_server do |config|
  config.redis = { url: 'redis://localhost:6379/0' }
end

Sidekiq.configure_client do |config|
  config.redis = { url: 'redis://localhost:6379/0' }
end

Worker の作成

app/workers 配下に作成します
デフォルトだとディレクトリが存在しないので作成します
ワーカーのファイル名は名前はワーカークラス名のスネークケースで作成します

mkdir app/workers
vim app/workers/wait_worker.rb

class WaitWorker
  include Sidekiq::Worker

  def perform(time, params)
    sleep time
    my_task = MyTask.new
    params = JSON.parse(params)
    my_task.title = params["title"]
    my_task.description = params["description"]
    my_task.public = params["public"]
    my_task.updated_date = Time.now
    if my_task.public
      my_task.posted_date = Time.now
    end
    my_task.save
  end
end

Sidekiq を経由すると params の Rails 側アプリとは少し異なる点に注意してください
後述していますがハッシュを JSON 文字列に変換してから Sidekiq に渡しています
やっていることは前回のタスクを登録する処理と同じです

エンキューする処理の実装

Rails アプリ側でエンキューするようにしましょう
コントローラを修正します

vim app/controllers/my_tasks_controller.rb

class MyTasksController < ApplicationController
  protect_from_forgery

  def list
    @my_tasks = MyTask.order(posted_date: "DESC").all
  end

  def show
    id = params[:id]
    @my_task = MyTask.find(id)
  end

  def create
    WaitWorker.perform_async(5, params.to_json)
    redirect_to '/my_tasks'
  end
end

1 つポイントがあり Sidekiq にハッシュ値を渡したい場合は一旦 JSON 文字列に変換する必要があります
そしてワーカー側で JSON 文字列からハッシュに再度変換します
今回は適当に遅延させたいので 5 秒ウェイトしてから DB に INSERT します

動作確認

まずはワーカーを起動します

bundle exec sidekiq

そして Rails アプリを起動しましょう

rails server

あとはタスクを作成してみましょう
5 秒ほど遅延があってからデータが挿入されるのが確認できると思います

最後に

Rails6 + Sidekiq を試してみました
Rails 上で使う場合はディレクトリ構成が決まっているのでその点だけ注意が必要かなと思います
Rails 上の場合は require 系が不要なのが記述が少なく済むので嬉しいのかなと思います

参考サイト

https://github.com/mperham/sidekiq/wiki/Getting-Started

2019年3月28日木曜日

rspec-test でブロックが渡される関数をスタブする方法

概要
環境
実行コード
スタブテストコード
応用
参考サイト

概要

例えば Sidekiq.redis { |c| c.get("hoge") } みたいな使い方をしているときに redis にはアクセスしないでテストしたい場合があると思います
そんな場合に rspec-mocks を使ってブロックをスタブする方法を紹介します

環境

macOS 10.14.3
Ruby 2.5.1p57

実行コード

例えば以下のような感じで Sidekiq.redis をコールしているとします

require 'sinatra/base'
require './worker.rb'

class Web < Sinatra::Base
  get '/:key' do
    Sidekiq.redis { |c|
      return c.get(params[:key])
    }
  end
end

これを rspec-mocks を使ってスタブするテストを作成します

スタブテストコード

and_yield を使うとブロック内の処理をスタブすることができます

ENV['RACK_ENV'] = 'test'

require './app.rb'
require 'rspec'
require 'rack/test'

describe "Web Tests" do
  include Rack::Test::Methods

  def app
    Web
  end

  it "could be fetch a value from redis" do
    redis = instance_double(Redis, :get => "fuga_test")
    allow(Sidekiq).to receive(:redis).and_yield(redis)
    get '/hoge'
    expect(last_response.body).to eq "fuga_test"
  end
end

これでテストすると Redis にはアクセスすることなくテストできます

応用

同じメソッド名で引数の key が異なる呼び出しをブロック内で複数しているとします
その場合は引数の key を正規表現を使って判断し引数に応じて別々のレスポンスを返却するように stub を定義します

require 'sinatra/base'
require './worker.rb'

class Web < Sinatra::Base
  get '/:key' do
    Sidekiq.redis { |c|
      if c.get("fuga") == "fuga_test"
        return c.get(params[:key])
      end
    }
  end
end

ENV['RACK_ENV'] = 'test'

require './app.rb'
require 'rspec'
require 'rack/test'

describe "Web Tests" do
  include Rack::Test::Methods

  def app
    Web
  end

  it "could be fetch a value from redis" do
    redis = instance_double(Redis)
    allow(redis).to receive(:get).with(/hoge/).and_return("hoge_test")
    allow(redis).to receive(:get).with(/fuga/).and_return("fuga_test")
    allow(Sidekiq).to receive(:redis).and_yield(redis)
    get '/hoge'
    expect(last_response.body).to eq "hoge_test"
  end
end

allow(redis).to receive(:get).with(/hoge/).and_return("hoge_test") を引数の key が異なる呼び出しごとに定義しています
これで c.get("fuga") の場合は fuga_test が返り c.get("hoge") の場合は hoge_test が stub から返るようになります

参考サイト

2019年3月21日木曜日

Sidekiq でワーカーにパスワードを渡す場合は暗号化しよう

概要
環境
Base64 にする
AES-256-CBC を使う
その他
最後に
参考サイト

概要

Sidekiq でジョブをエンキューするときは基本的に平文です
Enterprise Sidekiq であれば Encryption という機能があるのですがフリー版にはありません
そこで Redis にエンキューする際に暗号化する方法を考えます
エンキューする側は Sinatra を想定しています

環境

macOS 10.14.3
Ruby 2.5.1p57

Base64 にする

かなり単純ですが Base64 にするだけでも一応暗号化できます
エンキューする側で Base64 にエンコードし受け取るワーカー側でデコードします

vim app.rb

require 'sinatra/base'
require 'base64'
require 'openssl'
require './worker.rb'

class App < Sinatra::Base
  get '/start' do
    pass = Base64.encode64('password123')
    Worker.perform_async(pass)
  end
end

vim worker.rb

require 'sidekiq'
require 'base64'

class Worker
  include Sidekiq::Worker

  def perform(pass)
    puts pass
    puts Base64.decode64(pass)
  end
end

結果的には以下のようにワーカー側で取得できます

cGFzc3dvcmQxMjM=
password123

AES-256-CBC を使う

共通鍵暗号方式です
鍵になる文字列 (key) を使って AES-256-CBC で暗号化したバイト列を作成します
Sidekiq の場合バイト列はワーカーに渡せないので、バイト列を Base64 で文字列にします

vim app.rb

require 'sinatra/base'
require 'base64'
require 'openssl'
require './worker.rb'

class App < Sinatra::Base
  get '/start' do
    key = "test_key"
    pass = 'password123'
    encrypted = encrypt(pass, key)
    b64pass = Base64.encode64(encrypted)
    Worker.perform_async(b64pass)
  end

  def encrypt(pass, key)
    enc = OpenSSL::Cipher.new("aes-256-cbc")
    enc.encrypt
    enc.pkcs5_keyivgen(key)
    enc.update(pass) + enc.final
  end
end

vim worker.rb

require 'sidekiq'
require 'base64'

class Worker
  include Sidekiq::Worker

  def perform(b64pass)
    key = "test_key"
    puts b64pass
    pass = Base64.decode64(b64pass)
    decrypted = decrypt(pass, key)
    puts decrypted
  end

  def decrypt(pass, key)
    dec = OpenSSL::Cipher.new("aes-256-cbc")
    dec.decrypt
    dec.pkcs5_keyivgen(key)
    dec.update(pass) + dec.final
  end
end

ワーカー側で受け取れる情報は以下の通りです

ZpZtwrsrz6u5XXVHrQiBoA==
password123

こっちのほうがよりセキュアにはなると思います

その他

今回の場合は元の文字列を復号できることが条件になります
なので不可逆なアルゴリズムは使えません
また今回は共通鍵暗号方式になります
Sidekiq のようにワーカーとエンキューする側がシステム内で閉じている場合などは共通鍵暗号方式でいいかなと思います
逆に不特定多数の第三者に対してエンキューする場合などは公開鍵暗号方式のほうがおすすめです

最後に

Sidekiq で Redis に格納する際に暗号化する方法を検討しました
とりあえずパスワードなどが平文で Redis に格納されないだけでも良いかなと思います

今回はエンキューする側が Sinatra のような Web アプリを想定して行いました
もしかするとリクエストにパスワードなどが乗ってくる場合があると思います
そのような場合はすでに暗号化されている可能性があるのでエンキューする側は特に暗号化はせずエンキューしてワーカー側で復号するだけという手段も取れるかなと思います

その辺はシステムの仕様なので好きなように決めればいいかなと思います

参考サイト

https://www.kunihikokaneko.com/free/ruby/rubyopenssl.html

2019年2月13日水曜日

Sinatra + Sidekiq でレジューム可能なワーカーを考える

概要
環境
仕組みを検討する
Worker
API
考慮すべき点
最後に
参考サイト

概要

Sidekiq にはワーカーのプロセスを制御する API はありますがワーカー内で動作するジョブを制御する API はありません
なので例えば特定のジョブを停止させたい場合はワーカー自体を停止するしかありません
ただその場合はワーカーがリトライ用のジョブを再登録するための再度起動後に同じジョブを 1 から実行してしまいます

そうではなくジョブを停止して再実行した場合に停止した箇所から再実行できるようなワーカーを考えてみたいと思います

環境

macOS 10.14.2
Redis 4.0.9
Sidekiq 5.2.5

仕組みを検討する

Sidekiq::Process はワーカーのプロセスを制御する API です
これで stop! を掛けてしまうとワーカーが停止してしまいます
なのでこれは使わず特定のキーを Redis に登録することで「ジョブの停止」
「どこまで実行したかのステート」「実行中のジョブ」を管理したいと思います

Worker

まずはワーカーです
ポイントは 2 つです

すでに「どこまで実行したかのステート」がある場合はそのステートを使う
ジョブ実行中に停止用のキーが Redis にセットされたら「どこまで実行したかのステート」をセットしてジョブを終了する

になります
それらを考慮したコードは以下の通りです

require 'sidekiq'

class ResumeWorker
  include Sidekiq::Worker
  COUNT = 20
  KEY_PREFIX = 'resume'
  STOP = -1
  RUN = -3

  def perform(jid = nil)
    jid, counter = init(jid)
    while
      sleep 1
      counter += 1
      logger.info counter
      Sidekiq.redis { |c|
        v = c.get(jid).to_i
        if v == STOP
          c.set(jid, counter)
          return
        end
      }
      break if counter == COUNT
    end
    Sidekiq.redis { |c| c.del(jid) }
  rescue Interrupt
    logger.warn "stop worker by SIGTERM"
  end

  def init(jid)
    jid = self.jid if jid.nil?
    jid = "#{KEY_PREFIX}:#{jid}"
    counter = 0
    Sidekiq.redis { |c|
      v = c.get(jid)
      counter = v.to_i unless v.nil?
      c.set(jid, RUN)
      return jid, counter
    }
  end
end

登録したキーを識別できるように prefix を設けています
init 内で「どこまで実行したかのステート」をチェックし設定しています
今回ワーカーは単純なカウントをするだけのワーカーになっています
20 まで行ったら終了します
なのでステートはカウントした数字になります

ワーカーのメイン処理「perform」内でカウントするたびに Redis 内に停止用のキー (-1) が登録されたかをチェックします
もし登録されていたらどこまでカウントしたかを Redis に登録し return しています
カウントごとに Redis を read するので少しパフォーマンスが気になりますが今回は単純なカウントなので毎回チェックしています
やり方はいろいろありますが、ある一定のタイミングでチェックするような仕組みであれば別の方法でも良いと思います

API

次に API です
今回は API を叩くことでジョブの新規登録 (/start)、再開 (/start/:id)、停止 (/stop/:id) をできるようにします
フレームワークは Sinatra を使っています
ポイントは 2 つです

再開時に停止中のジョブがあるかないかチェックする
停止するため条件は「すでに実行中である」「停止の信号をすでに送信している場合は何もしない」「実行中の Sidekiq::Workers に該当のジョブ ID が存在している」

特に停止時は注意が必要です
コードは以下の通りです

require 'sinatra/base'
require 'sidekiq/api'
require 'logger'
require './resume_worker.rb'

class ResumeApp < Sinatra::Base
  logger = Logger.new(STDOUT)

  get '/start' do
    ResumeWorker.perform_async
  end

  get '/start/:id' do
    Sidekiq.redis { |c|
      v = c.get("#{ResumeWorker::KEY_PREFIX}:#{params['id']}")
      logger.info v
      return "no resumed id: #{params['id']}" if v.nil? || v.to_i ==  ResumeWorker::RUN
      ResumeWorker.perform_async(params['id'])
      return params['id']
    }
  end

  get '/stop/:id' do
    Sidekiq.redis { |c|
      v = c.get("#{ResumeWorker::KEY_PREFIX}:#{params['id']}")
      logger.info v
      return c.set("#{ResumeWorker::KEY_PREFIX}:#{params['id']}", ResumeWorker::STOP) if v.to_i == ResumeWorker::RUN
      return "alreay send stop signal #{params['id']}" if v.to_i != ResumeWorker::RUN
      workers = Sidekiq::Workers.new
      return "worker does no running #{params['id']}" unless workers.any? { |pid, tid, info| info['payload']['jid'] == params['id'] }
      c.set("#{ResumeWorker::KEY_PREFIX}:#{params['id']}", ResumeWorker::STOP)
    }
  end
end

ジョブを停止する信号を送信するとワーカー側でそれを検知してステートを保存します
その際にすでに信号は登録されているがワーカー自体はまだジョブを手放していない状態になる可能性があります
その状態で /stop をコールした際に保存したステートを上書きしないようにチェックする必要があります

新規でジョブが開始された場合とジョブを再開した場合の戻り値はジョブ ID を返します
ジョブを停止した場合は「OK」だけを返しています
それ以外のエラー系に関してはジョブ ID＋メッセージを返しています

考慮すべき点

今回の場合サンプルレベルなのでそこまでいろいろ考えていません
実際に運用する場合は更に以下を考慮したほうが良いと思います

/stop を全くジョブが動作していない状況でもできるようにする
レジューム中のジョブの一覧などを取得できるようにする
ワーカーが強制停止してしまった場合の対応

あたりは考えたほうが良いと思います
最後の強制停止 (SIGTERM) は rescue Interrupt で拾えるのは拾えるのですがこの中で Sidekiq.redis が使えません
なのでまた別の方法でステートを保存する必要が出てきます

最後に

Sinatra + Sidekiq でジョブをレジュームできるワーカーを簡単ですが実装してみました
やってみるとレジューム機能がかなり考えることが多く難しい機能だということがわかりました
機能的に便利なので余裕があれば実装してみると良いかなと思います

参考サイト

http://n8.hatenablog.com/entry/2015/05/01/100424

2018年4月14日土曜日

sidekiq/api を使ってキューやジョブの操作をしてみる

概要
環境
Redis に接続する
キューの一覧を取得する
キュー内に溜まったジョブの一覧を取得する
リトライ対象のジョブの一覧を取得する
リトライジョブを削除する
最後に
参考サイト

概要

Sidekiq はワーカーがエラーをキャッチすると自動でリトライ処理をしてくれます
故にエラーが解消されない限りはワーカーが永遠にジョブを処理し続けてしまいます
そんな場合にはキューからジョブを削除しなければなりません
直接 redis を操作しても問題ないですが Sidekiq にはキューやジョブを操作する API が備わっているので使ってみました

環境

macOS 10.13.2
Ruby 2.4.1p111
Redis 3.2.1
sidekiq 5.1.3

Redis に接続する

今回の作業はすべて irb (bundle console) で行います

デフォルトだと localhost:6379 のデータベースインデックス 0 に接続します
今回はインデックスが 1 のデータベースに接続するため切り替える必要があります

irb(main):001:0> require 'sidekiq/api'
=> true
irb(main):002:0> Sidekiq.redis { |con| con.select 1 }
=> "OK"

これでインデックス 1 のデータベースに切り替えできました

キューの一覧を取得する

特にしていしなければ default のみあると思います

irb(main):003:0> Sidekiq::Queue.all
=> [#<Sidekiq::Queue:0x007fc13b3a0238 @name="default", @rname="queue:default">]
irb(main):004:0> q = Sidekiq::Queue.new
=> #<Sidekiq::Queue:0x007fc13b371d20 @name="default", @rname="queue:default">

Sidekiq::Queue.new でキューオブジェクトを作成します
引数に文字列でキュー名を指定することができます
指定しない場合は default になります
今回はこの default キュー内のジョブを削除してみます

キュー内に溜まったジョブの一覧を取得する

each で回すだけです

irb(main):008:0> q.each { |job|
irb(main):009:1* p job
irb(main):010:1> }
=> nil

特に何も入っていないので何も表示されませんでした

リトライ対象のジョブの一覧を取得する

リトライ対象のジョブは別のクラスから取得することができます
Sidekiq::RetrySet を使います

irb(main):014:0> rs = Sidekiq::RetrySet.new
=> #<Sidekiq::RetrySet:0x007fc13b83fc08 @name="retry", @_size=1>
irb(main):015:0> rs.each { |job|
irb(main):016:1* p job['jid']
irb(main):017:1> p job['retry_count']
irb(main):018:1> }
"85d82e45decb7474d5fbfbaf"
6
=> nil

長いので jid と retry_count だけ表示しています
ここにジョブが入っていると default キューを見るワーカーがこのジョブを処理し続けます

リトライジョブを削除する

ジョブに対して delete メソッドをコールするだけです

irb(main):022:0> rs.each { |job|
irb(main):023:1* job.delete
irb(main):024:1> }
=> nil
irb(main):025:0> rs.each { |job|
irb(main):026:1* p job.jid
irb(main):027:1> }
=> nil
irb(main):028:0> rs.size
=> 0

もし特定のジョブだけを削除したい場合は一旦 jid のリストを取得して、指定の jid とマッチしたら delete するという感じの処理を追加すれば OK です
おそらくこの操作が一番運用で使うんじゃないかなと思います

最後に

sidekiq/api を使ってキューやジョブの操作をしてみました
前回紹介したダッシュボードの機能を使えば UI でもリトライジョブを削除することができると思います
1 つや 2 つであれば UI でもいいですが、何百何千という単位で処理したい場合はこれを使ってプログラムを組んだほうが良いかなと思います

参考サイト

2018年4月13日金曜日

Sidekiq でキューを指定してワーカーを起動する方法

概要
環境
ワーカークラスでキューを指定する
ワーカーを起動する時にキューを指定する
最後に
参考サイト

概要

Sidekiq はデフォルトで default という名前のキューを使用します
用途の異なる複数のワーカーを作成する場合にキューを指定しないとすべてのワーカーが default を見てしまい本来掴みたくないジョブを掴んでしまうことがあります
その場合はワーカーごとにキューをわけましょう

環境

macOS 10.13.2
Ruby 2.4.1p111
Redis 3.2.1
sidekiq 5.1.3
sinatra 2.0.1

ワーカークラスでキューを指定する

これはエンキューするクライアントが参照します
当初ここに記載しておけばワーカー側も自動で参照してくれると思ったのですが、これはクライアント用の設定みたいです

require 'sidekiq'

Sidekiq.configure_server do |config|
  config.redis = { 'db' => 1 }
end

class MyWorker
  include Sidekiq::Worker
  sidekiq_options queue: 'test_queue'

  def perform(complexity)
    case complexity
    when 'super_hard'
      sleep 20
      puts 'super_hard'
    when 'hard'
      sleep 10
      puts 'hard'
    else
     sleep 1
     puts 'easy'
    end
  end
end

sidekiq_options queue: 'test_queue' の設定を 1 行追加しているだけです
これで MyWorker.perform_async したときに test_queue にエンキューされます

ワーカーを起動する時にキューを指定する

ワーカーは起動するときのオプションで参照するキューを指定します

bundle exec sidekiq -q test_queue -r ./worker.rb

これで default ではなく test_queue を参照するようになります
試しに redis-cli を使って確認したところちゃんと指定のキューが作成されていることがわかると思います

$ redis-cli -n 1
127.0.0.1:6379[1]> SMEMBERS queues
1) "test_queue"

最後に

Sidekiq でキューを指定してワーカーを動作させる方法を紹介しました
ワーカーを起動する時にオプションが必要になるのがポイントでしょうか

もう一つ方法として namespace を使う方法もあるらしいですがこれはまた別の方法かなと思います
新たにキューを作成するわけではなく更に上位の概念で分割しようという機能だと思います
また、namespace を使う場合には redis-namespace という gem も必要になります

参考サイト

https://github.com/mperham/sidekiq/wiki/Advanced-Options

2018年4月12日木曜日

Sinatra + Sidekiq で簡単非同期 Web アプリ

概要
環境
ワーカー
アプリ
動作確認
最後に

概要

前回 Sidekiq に入門してみました
今回は Sinatra と組み合わせてみました

環境

macOS 10.13.2
Ruby 2.4.1p111
Redis 3.2.1
sidekiq 5.1.3
sinatra 2.0.1

ワーカー

基本的には前回と同じです

vim worker.rb

require 'sidekiq'

Sidekiq.configure_server do |config|
  config.redis = { 'db' => 1 }
end

class MyWorker
  include Sidekiq::Worker

  def perform(complexity)
    case complexity
    when 'super_hard'
      sleep 20
      puts 'super_hard'
    when 'hard'
      sleep 10
      puts 'hard'
    else
     sleep 1
     puts 'easy'
    end
  end
end

bundle exec sidekiq -r ./worker.rb -P ./tmp/sidekiq.pid

とりあえず起動しておきましょう

アプリ

普通の Sinatra アプリに enqueue する処理をいれるだけです

require 'sinatra/base'
require './worker.rb'

Sidekiq.configure_client do |config|
        config.redis = { 'db' => 1 }
end

class MyApp < Sinatra::Base
        get '/enqueue/:value' do
                MyWorker.perform_async(params['value'])
                "enqueue #{params['value']}"
        end
end

bundle exec rackup config.ru

動作確認

curl 'localhost:9292/enqueue/super_hard'
curl 'localhost:9292/enqueue/hard'
curl 'localhost:9292/enqueue/easy'

と実行してみましょう
すぐにレスポンスが返ってくると思います

そしてワーカーのログを確認するとちゃんと非同期処理が実行されていることが確認できると思います

easy
2018-04-04T09:20:43.050Z 5683 TID-ovxtvzpkn MyWorker JID-170aa25c66fcdafb78e7c049 INFO: done: 1.001 sec
hard
2018-04-04T09:20:49.878Z 5683 TID-ovxtw03kf MyWorker JID-22663fbd98445bde4119a4d1 INFO: done: 10.006 sec
super_hard
2018-04-04T09:20:57.103Z 5683 TID-ovxtvzpbn MyWorker JID-895bd0adfb849652c0860eae INFO: done: 20.005 sec

最後に

Sinatra + Sidekiq で超かんたんな非同期 Web アプリを作成してみました
処理も切り離せて管理もしやすいと思います

サービスにする場合はちゃんとキューの状態などを追える仕組みを考えないとダメだと思います

2018年4月11日水曜日

Sidekiq で管理画面を表示する方法

概要
環境
ダッシュボード用の Rack アプリの作成
動作確認
最後に
参考サイト

概要

Sidekiq には管理画面がデフォルトでついており現在実行中のジョブなどを視覚的に確認することができます
デフォルトだと使えないようなので有効にする方法を紹介します

環境

macOS 10.13.2
Ruby 2.4.1p111
Redis 3.2.1
sidekiq 5.1.3

ダッシュボード用の Rack アプリの作成

すでにモジュールが用意されているのでそれを実行するだけです

vim config.ru

require 'sidekiq'

Sidekiq.configure_client do |config|
  config.redis = { :size => 1 }
end

require 'sidekiq/web'
run Sidekiq::Web

動作確認

実行してみましょう
Rack で動作します

bundle exec rackup config.ru

Rack なのでデフォルトは 9292 ポートで動作します
ブラウザでアクセスすると以下のような UI が表示されると思います

ちゃんとローカライズもされていました
もちろんワーカーを動作させるとグラフが描画されるのを確認することができます

最後に

Sidekiq のダッシュボードアプリを作成してみました
簡単に作れるのでとりあえず立てておくと運用も楽になるかなと思います

Heroku にデプロイできるボタンもあったので Redis が外部から参照できるのであれば Heroku にデプロイしちゃってもいいかなと思います

参考サイト

2018年4月10日火曜日

Sidekiq 超入門

概要
環境
インストール
ワーカー作成
クライアント作成
動作確認
最後に
参考サイト

概要

Sidekiq はキュー&ワーカー方式で非同期処理を実現するためのツールです
バックエンドに Redis を使っています
今回は動作させるところまでやってみました

環境

macOS 10.13.2
Ruby 2.4.1p111
Redis 3.2.1
sidekiq 5.1.3

インストール

bundle init
vim Gemfile

gem "sidekiq"

bundle install --path vendor

Redis も必要になるのでインストールしていない場合は brew などを使ってインストールしてください

ワーカー作成

vim worker.rb

require 'sidekiq'

Sidekiq.configure_server do |config|
  config.redis = { 'db' => 1 }
end

class MyWorker
  include Sidekiq::Worker

  def perform(complexity)
    case complexity
    when 'super_hard'
      sleep 20
      puts 'super_hard'
    when 'hard'
      sleep 10
      puts 'hard'
    else
     sleep 1
     puts 'easy'
    end
  end
end

include Sidekiq::Worker し perform メソッドを実装することでワーカーを作成します
クライアント側でキューイングする際に perform_async を呼び出すことでこの perform メソッドが自動的に呼ばれます

今回は非同期で実行されているか確認できるように sleep を入れています

クライアント作成

vim test_client.rb

require './worker.rb'

Sidekiq.configure_client do |config|
  config.redis = { 'db' => 1 }
end

class MyClient
  def enqueue(complexity)
    MyWorker.perform_async(complexity)
  end
end

cli = MyClient.new
cli.enqueue('super_hard')
cli.enqueue('hard')
cli.enqueue('easy')

クライアント側は worker クラスが必要になります
MyWorker クラスでは include Sidekiq::Worker することでエンキュー用の perform_async というメソッドが使えるようになっているのでこれをコールします

今回はワーカー側もクライアント側も localhost:6379 のデフォルトの Redis を参照しているため特にホストやポートの指定はしていませんが、ホストやポートが異なる場合は url というオプションがあるのでそれを使って指定してください (参考)

動作確認

まずはワーカーを起動します
ワーカーは Redis を必要するので Redis も起動しておいてください

redis-server
mkdir ./tmp
bundle exec sidekiq -r ./worker.rb -P ./tmp/sidekiq.pid

sidekiq というコマンドがあるのでこれを使ってワーカースクリプトを指定します
sidekiq コマンドの詳しいオプションは bundle exec sidekiq --help で確認できます
-P オプションで PID ファイルを作成することができます
ワーカーを停止する際に必要になるので指定しておきます
ずらずらーとログが流れてくると思います
キックしている AA とエラーログが流れなければ OK です

次にクライアントを起動します

bundle exec ruby test_client.rb

すると以下のようにログが流れると思います

2018-04-04T06:17:54.025Z 4252 TID-oxfu7p3gg MyWorker JID-4473b56c73ed2dd8cf1f89a7 INFO: start
easy
2018-04-04T06:17:55.029Z 4252 TID-oxfu7p3gg MyWorker JID-4473b56c73ed2dd8cf1f89a7 INFO: done: 1.004 sec
hard
2018-04-04T06:18:04.027Z 4252 TID-oxfu7zlbk MyWorker JID-b6be42cf2f3e5f34c7f47468 INFO: done: 10.002 sec
super_hard
2018-04-04T06:18:14.030Z 4252 TID-oxfu7zlh4 MyWorker JID-4ea3461949baefd67e58172c INFO: done: 20.005 sec

enqueue した順番は super_hard -> hard -> easy ですが出力されるのは easy -> hard -> super_hard で表示されており非同期でそれぞれ処理されているのがわかります

ワーカーを停止する場合は sidekiqctl というコマンドが用意されているのでこれを使うと良いです

bundle exec sidekiqctl stop ./tmp/sidekiq.pid

PID ファイルを指定すれば停止できます
全部のワーカーが完了するかタイムアウトの 10 秒後の終了します
タイムアウトの時間は設定で変更できます

最後に

SIdekiq の基本的な使い方を紹介しました
Redis を導入する必要はありますが比較的に簡単に導入できるので非同期処理を実現するにはオススメです

Web アプリではよく Rails に組み込む方法が紹介されていますが Sinatra などの軽量フレームワークでも使うことはできます
そのあたりも検証できたら紹介したいと思います

参考サイト

登録: 投稿 (Atom)