Sidekiq Worker Killer

sidekiq_worker_killer.rb

# https://github.com/gitlabhq/gitlabhq/blob/master/lib/gitlab/sidekiq_middleware/memory_killer.rb

class SidekiqWorkerKiller
  include Sidekiq::Util # for #hostname

  # Maximum RSS in KB for (default 0, meaning the WorkerKiller is disabled)
  MAX_RSS = (ENV['WORKER_KILLER_MAX_RSS'] || 0).to_s.to_i
  # Wait in seconds for running jobs to finish during graceful shutdown (default 60 minutes)
  SHUTDOWN_WAIT = (ENV['WORKER_KILLER_SHUTDOWN_WAIT'] || 60 * 60).to_s.to_i
  # Wait in seconds for process to exit after SIGTERM (default 60 seconds)
  KILL_WAIT = (ENV['WORKER_KILLER_KILL_WAIT'] || 60).to_s.to_i
  # Per the Sidekiq API, ProcessSet is updated every 5 seconds
  # https://github.com/mperham/sidekiq/wiki/API
  PROCESS_SET_REFRESH = 15


  # Create a mutex used to ensure there will be only one thread waiting to shut Sidekiq down
  MUTEX = Mutex.new

  def call(_worker, _job, _queue)
    yield
    current_rss = get_rss

    return unless MAX_RSS > 0 && current_rss > MAX_RSS

    Thread.new do
      # Return if another thread is already waiting to shut Sidekiq down
      return unless MUTEX.try_lock

      Sidekiq.logger.info "current RSS #{current_rss} exceeds maximum RSS #{MAX_RSS}"
      Sidekiq.logger.info "sending TSTP to #{hostname} PID #{pid}"
      kill('TSTP')

      process_set = Sidekiq::ProcessSet.new
      wait_start = Time.now
      sleep PROCESS_SET_REFRESH
      # Process must be accessed through ProcessSet to get updated information
      until process_set.find{|process| process['hostname'] == hostname}&.[]('busy')&.zero?
        if (Time.now - wait_start) > SHUTDOWN_WAIT
          Sidekiq.logger.warn "timeout for #{hostname} PID #{pid} (#{SHUTDOWN_WAIT}s)"
          break
        else
          sleep PROCESS_SET_REFRESH
        end
      end
      Sidekiq.logger.info "sending SIGTERM to #{hostname} PID #{pid}"
      kill('SIGTERM')
      sleep KILL_WAIT
      Sidekiq.logger.info "sending SIGKILL to #{hostname} PID #{pid}"
      kill('SIGKILL')
    end
  end

  private

  def get_rss
    `ps -o rss= -p #{pid}`.to_i
  end

  def pid
    ::Process.pid
  end

  def kill(signal)
    ::Process.kill(signal, pid)
  end

end

This is a modified version of GitLab's MemoryKiller that will stop the current Sidekiq process from taking on new jobs and attempt to wait until all jobs are completed before killing the process.

These environment variables should be set:

• WORKER_KILLER_MAX_RSS (in KB, the maximum RSS used before initiating a kill)
• WORKER_KILLER_SHUTDOWN_WAIT (in seconds, the maximum time to wait for jobs to finish)