ruby Sidekiq本地处理统计数据

Posted 2021-05-15

# Emit key statistics about locally running  Sidekiq processes to a stream.
#
require 'sidekiq/api'

class SidekiqProcessesPrinter
  attr_reader :format

  def initialize(opts={})
    @format = opts.fetch(:format){:plain}
  end

  def metadata
    {
      source: "sidekiq"
    }
  end

  def run
    @processes = nil

    processes.each do |p|
      puts formatter.new(metadata.merge(process_stats(p))).inspect
    end
  end

  def processes
    @processes ||= build_processes
  end

  protected

  def formatter
    case format.to_sym
    when :json
      JsonFormatter
    else
      SimpleFormatter
    end
  end

  def build_processes
    Sidekiq::ProcessSet.new
  end

  def process_stats(process)
     # "hostname"=>"Evans-2015-MacBook-Pro.local",
     # "started_at"=>1509126804.594962,
     # "pid"=>98873,
     # "tag"=>"timebomb-api",
     # "concurrency"=>10,
     # "queues"=>["default", "high"],
     # "labels"=>[],
     # "identity"=>"Evans-2015-MacBook-Pro.local:98873:b8586eb0b416",
     # "busy"=>0,
     # "beat"=>1509127225.597922,
     # "quiet"=>"false"

    {}.tap do |stats|
      stats[:pid] = "#{process['hostname']}:#{process['pid']}"
      stats[:workers] = process['concurrency']
      stats[:running] = process['busy']
      stats[:queues] = process['queues'].join(', ')
    end
  end

  class Formatter
    attr_accessor :hash
    # source
    # queue
    # size
    # queuing_latency
    def initialize(hash)
      @hash = hash
    end
  end

  class SimpleFormatter < Formatter
    def inspect
      string =  "#{hash[:pid]}\n"
      string += "  workers: #{hash[:workers]}\n"
      string += "  running: #{hash[:running]}\n"
      string += "  queues: #{hash[:queues]}\n"
    end
  end

  class JsonFormatter < Formatter
    def inspect
      hash.to_json
    end
  end
end

opts = { format: :json }
SidekiqProcessesPrinter.new(opts).run