Contents

# frozen_string_literal: true

module PrometheusExporter::Instrumentation
  class SidekiqQueue
    def self.start(client: nil, frequency: 30, all_queues: false)
      client ||= PrometheusExporter::Client.default
      sidekiq_queue_collector = new(all_queues: all_queues)

      Thread.new do
        loop do
          begin
            client.send_json(sidekiq_queue_collector.collect)
          rescue StandardError => e
            client.logger.error("Prometheus Exporter Failed To Collect Sidekiq Queue metrics #{e}")
          ensure
            sleep frequency
          end
        end
      end
    end

    def initialize(all_queues: false)
      @all_queues = all_queues
      @pid = ::Process.pid
      @hostname = Socket.gethostname
    end

    def collect
      {
        type: 'sidekiq_queue',
        queues: collect_queue_stats
      }
    end

    def collect_queue_stats
      sidekiq_queues = ::Sidekiq::Queue.all

      unless @all_queues
        queues = collect_current_process_queues
        sidekiq_queues.select! { |sidekiq_queue| queues.include?(sidekiq_queue.name) }
      end

      sidekiq_queues.map do |queue|
        {
          backlog: queue.size,
          latency_seconds: queue.latency.to_i,
          labels: { queue: queue.name }
        }
      end.compact
    end

    private

    def collect_current_process_queues
      ps = ::Sidekiq::ProcessSet.new

      process = ps.find do |sp|
        sp['hostname'] == @hostname && sp['pid'] == @pid
      end

      process.nil? ? [] : process['queues']
    end
  end
end

Version data entries

3 entries across 3 versions & 1 rubygems

Version	Path
prometheus_exporter-2.0.0	lib/prometheus_exporter/instrumentation/sidekiq_queue.rb
prometheus_exporter-1.0.1	lib/prometheus_exporter/instrumentation/sidekiq_queue.rb
prometheus_exporter-1.0.0	lib/prometheus_exporter/instrumentation/sidekiq_queue.rb