Rake task to transfer Sidekiq jobs from one redis instance to another

sidekiq_transfer.rake

# This task should be run inside an environment that is already configured to connect to the redis
# instance that we're transfering AWAY FROM.
#
# The task should be handed the URL of the redis instance that we're MOVING TO.
#
# To run it and pass in the destination Redis you'd do something like this:
# rake sidekiq:transfer[redis://...]
#
# As jobs are added to the destination Redis, they're deleted from the source Redis. This
# allows the task to be restarted cleanly if it fails in the middle due to a network error
# or something like that.
#
namespace :sidekiq do
  desc 'Transfer sidekiq jobs from one instance to another'
  task :transfer, [:destination] => :environment  do |task, args|
    
    def transfer_queue queue, destination_client
      puts "transfering queue : #{queue.name}"
      queue.each do |job|
        destination_client.push 'class' => job.klass, 'args' => job.args
        job.delete
      end
    end
    
    def transfer_set set, destination_client
      set.each do |job|
        destination_client.push 'class' => job.klass, 'args' => job.args, 'at' => job.at.to_i
        scheduled_job.delete
      end
    end

    destination_pool = ConnectionPool.new { Redis.new(url: args[:destination]) }
    destination_client = Sidekiq::Client.new(destination_pool)

    source_queues = Sidekiq::Queue.all
    source_queues.each do |queue|
      transfer_queue queue, destination_client
    end

    puts "transfering scheduled jobs"
    ss = Sidekiq::ScheduledSet.new
    transfer_set ss, destination_client

    puts "transfering retried jobs"
    rs = Sidekiq::RetrySet.new
    transfer_set rs, destination_client
  end
end

Hey, thank you very much for the script!

Three little things though:

• A mistake, scheduled_job on line 28 should be job
• To restore jobs in the same queue, add , 'queue' => queue.name on line 20, or they'll all go to default
  • Result is destination_client.push 'class' => job.klass, 'args' => job.args, 'queue' => queue.name
• Linting: ss (line 41 and 42) and rs (45 and 46) should be renamed scheduled_set and retry_set to be communicative variable names

Thanks from me too!

Another comment: Sidekiq::Queue.all not show all the queues. It only shows queue which have processed at least a message, but you can pause a new queue and move jobs there and these jobs will be out of the scope of this script unless you hardcode that queue name and append it to the queue names list.

Thanks a lot 🙌
This is my procedural version I've used today, which includes checks on queue size to validate the migration:

# Connect & check old redis queue
Sidekiq.configure_client { |config| config.redis = { url: ENV['THE_OLD_REDIS_URL'] } }
Sidekiq::ScheduledSet.new.size

# Init new redis client
dst_pool = ConnectionPool.new { Redis.new(url: ENV['THE_NEW_REDIS_URL']) }
dst_client = Sidekiq::Client.new(dst_pool)

# load old jobs
old_jobs = []
Sidekiq::ScheduledSet.new.each do |job|
  old_jobs << job
end

# push old jobs to new redis
old_jobs.each do |job|
  puts "Adding #{job.klass} with #{job.args} at #{job.at.to_i}"
  dst_client.push 'class' => job.klass, 'args' => job.args, 'at' => job.at.to_i
end

# check if new jobs were added
Sidekiq.configure_client { |config| config.redis = { url: ENV['THE_NEW_REDIS_URL'] } }
Sidekiq::ScheduledSet.new.size

# delete old job if needed
# Sidekiq.configure_client { |config| config.redis = { url: ENV['THE_OLD_REDIS_URL'] } }
# old_jobs.each do |job|
#   job.delete
# end