Sidekiq : Set of task/script to manage sidekiq

.Manage Sidekiq.rb

# Contains :
#   * deploy_git_sync.rb       : Check if a job is running 
#   * identify_errno_EMFILE.sh : identify a Errno::EMFILE over sidekiq 

identify_errno_EMFILE.sh

Initial state : 15 sidekiq instances with 15 queues (RUBY 2.3.0)
# Number of file_descriptor open for sidekiq :
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | wc -l
Result : ~17621
# Number of file_descriptor open for ElasticSearch
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep 9200 | wc -l
Result : 6
# Number of file_descriptor open for Postgres
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep postgres | wc -l
Result : 167
# Redis
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep '6379' | wc -l
Result :  147
# Library
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep '.so' | wc -l
Result : 2178
#Other
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep -v '.so' | grep -v '9200'| grep -v 'postgres'  | grep -v '6379' | wc -l
Result : ~15125
# PDF
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep -v '.so' | grep -v '9200' | grep -v 'postgres'  | grep -v '6379' | grep -v '_ruby' | grep -v typhon | grep '.pdf' | wc -l
Result : ~0
# Remain
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep -v '.so' | grep -v '9200'| grep -v 'postgres'  | grep -v '6379' | grep -v '_ruby' | grep -v typhon | grep -v '.pdf' | grep -v 'pipe' |  wc -l
Result : ~11993

# The leaks come from Here
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep -v '.so' | grep -v '9200'| grep -v 'postgres'  | grep -v '6379' | grep -v '_ruby' | grep -v typhon | grep -v '.pdf' | grep -v 'pipe' |  wc -l
Result : ~11714

# A lot of line like the following :
COMMAND PID TID USER FD TYPE DEVICE SIZE/OFF NODE NAME
ruby 2374 xxxx 1001r REG 8,3 27610 74787636 /.../production/releases/20160128124439/tmp/uploads/1454028547-2374-0162-
3376/181df07c5cb8df835548733867ae08c02fe17c4e-big.jpg (deleted)
ruby 2374 xxxx 722u REG 8,3 98951 187434219 /tmp/open-uri20160129-2374-1mwhr8m (deleted)
ruby 2374 xxxx 721r REG 8,3 98951 74787338 /home/xxxx/projects/xxxx/production/releases/20160128124439/tmp/uploads/1454028495-2374-0143-
1515/7fa0dd37-0e4e-42e6-b764-4da2945183a1.jpg (deleted)

======================================================================================================

Initial state : 15 sidekiq instances with 10 queues (RUBY 2.2.3)
# Number of file_descriptor open for sidekiq :
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | wc -l
Result : ~2400
# Number of file_descriptor open for ElasticSearch
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep 9200 | wc -l
Result : 16
# Number of file_descriptor open for Postgres
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep postgres | wc -l
Result : 165
# Redis
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep '6379' | wc -l
Result : 133
# Library
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep '.so' | wc -l
Result : 1796
#Other
lsof -p $(ps aux | grep sidekiq | tr -s " " | cut -d " " -f 2 | tr '\012' ,) | grep -v '.so' | grep -v '9200'| grep -v 'postgres'  | grep -v '6379' | wc -l
Result : ~359

sidekiq_manager_helper.rb

module Sidekiq
  module ManagerHelper

    ##[SEARCH FOR EXISTING JOB IN A QUEUE]######################################
    def self.running_jobs?(queue_name)
      !!(Sidekiq::Workers.new.find { |_, _, work| work['queue'] == queue_name.to_s })
    end

    def self.enqueue_jobs?(queue_name)
      !!(Sidekiq::Queue.new(queue_name).count > 0)
    end

    [:retry, :scheduled].each do |job_set_name|
      define_singleton_method "#{job_set_name}_jobs?" do |queue_name|
        job_set = "Sidekiq::#{job_set_name.to_s.classify}Set".constantize.new
        job_set.find do |sheduled_job_hash|
          _extract_queue_name(sheduled_job_hash) == queue_name.to_s
        end
      end
    end
    ##[/SEARCH FOR EXISTING JOB IN A QUEUE]#####################################

    ##[SEARCH FOR SPECIFIC JOB IN A QUEUE]######################################
    def self.running_job(queue_name, aj_globalid)
      workers = Sidekiq::Workers.new
      !!(workers.find do |process_id, thread_id, work|
          if work['queue'] == queue_name.to_s && (wp = work['payload']) && (wap = wp['args']) && (wpafa = wap.first['arguments'])
            wpafa.first && wpafa.first['_aj_globalid'].to_s == aj_globalid.to_s
          end
        end)
    end
    def self.running_job?(queue_name, aj_globalid) !!running_job(queue_name, aj_globalid); end

    def self.enqueue_job(queue_name, aj_globalid)
      queue = Sidekiq::Queue.new(queue_name)
      queue.find { |queue_job_hash| _extract_model_id(queue_job_hash) == aj_globalid.to_s }
    end
    def self.enqueue_job?(queue_name, aj_globalid); !!enqueue_job(queue_name, aj_globalid); end

    [:retry, :scheduled].each do |job_set_name|
      define_singleton_method "#{job_set_name}_job" do |queue_name, aj_globalid|
        job_set = "Sidekiq::#{job_set_name.to_s.classify}Set".constantize.new
        job_set.find do |sheduled_job_hash|
          ( _extract_model_id(sheduled_job_hash)   == aj_globalid.to_s &&
            _extract_queue_name(sheduled_job_hash) == queue_name.to_s )
        end
      end
      define_singleton_method "#{job_set_name}_job?" do |queue_name, aj_globalid|
        !!send("#{job_set_name}_job", queue_name, aj_globalid)
      end
    end
    ##[/SEARCH FOR SPECIFIC JOB IN A QUEUE]#####################################

    private
    def self._extract_queue_name(job_hash); job_hash.item['queue']; end

    # Not usefull for now but can be usefull if a queue wrap many job
    #def self._extract_wrapper_class(job_hash); job_hash.item['wrapped']; end

    def self._extract_model_id(job_hash)
      arguments = (siaf = job_hash.item['args'].first) && (siaf.is_a?(Hash) && siaf['arguments'])
      argument  = arguments && arguments.first
      (argument && argument.is_a?(Hash) ? argument['_aj_globalid'] : argument).to_s
    end
  end
end

Profiling memory leaky Sidekiq applications : https://gist.github.com/krasnoukhov/a036bb7bc8c8416e9481