Class: Sidekiq::CloudWatchMetrics::Publisher

Inherits:

Object

• Object
• Sidekiq::CloudWatchMetrics::Publisher

Defined in:

lib/sidekiq/cloudwatchmetrics.rb

Constant Summary

INTERVAL =

seconds

60

Instance Method Summary

• #initialize(config: Sidekiq, client: Aws::CloudWatch::Client.new, namespace: "Sidekiq", process_metrics: true, additional_dimensions: {}) ⇒ Publisher constructor

  A new instance of Publisher.

• #publish ⇒ Object
• #quiet ⇒ Object
• #run ⇒ Object
• #running? ⇒ Boolean
• #start ⇒ Object
• #stop ⇒ Object

Constructor Details

#initialize(config: Sidekiq, client: Aws::CloudWatch::Client.new, namespace: "Sidekiq", process_metrics: true, additional_dimensions: {}) ⇒ Publisher

Returns a new instance of Publisher.

# File 'lib/sidekiq/cloudwatchmetrics.rb', line 48

def initialize(config: Sidekiq, client: Aws::CloudWatch::Client.new, namespace: "Sidekiq", process_metrics: true, additional_dimensions: {})
  # Sidekiq 6.5+ requires @config, which defaults to the top-level
  # `Sidekiq` module, but can be overridden when running multiple Sidekiqs.
  @config = config
  @client = client
  @namespace = namespace
  @process_metrics = process_metrics
  @additional_dimensions = additional_dimensions.map { |k, v| {name: k.to_s, value: v.to_s} }
end

Instance Method Details

#publish ⇒ Object

# File 'lib/sidekiq/cloudwatchmetrics.rb', line 87

def publish
  now = Time.now
  stats = Sidekiq::Stats.new
  processes = Sidekiq::ProcessSet.new.to_enum(:each).to_a
  queues = stats.queues

  metrics = [
    {
      metric_name: "ProcessedJobs",
      timestamp: now,
      value: stats.processed,
      unit: "Count",
    },
    {
      metric_name: "FailedJobs",
      timestamp: now,
      value: stats.failed,
      unit: "Count",
    },
    {
      metric_name: "EnqueuedJobs",
      timestamp: now,
      value: stats.enqueued,
      unit: "Count",
    },
    {
      metric_name: "ScheduledJobs",
      timestamp: now,
      value: stats.scheduled_size,
      unit: "Count",
    },
    {
      metric_name: "RetryJobs",
      timestamp: now,
      value: stats.retry_size,
      unit: "Count",
    },
    {
      metric_name: "DeadJobs",
      timestamp: now,
      value: stats.dead_size,
      unit: "Count",
    },
    {
      metric_name: "Workers",
      timestamp: now,
      value: stats.workers_size,
      unit: "Count",
    },
    {
      metric_name: "Processes",
      timestamp: now,
      value: stats.processes_size,
      unit: "Count",
    },
    {
      metric_name: "DefaultQueueLatency",
      timestamp: now,
      value: stats.default_queue_latency,
      unit: "Seconds",
    },
    {
      metric_name: "Capacity",
      timestamp: now,
      value: calculate_capacity(processes),
      unit: "Count",
    },
  ]

  utilization = calculate_utilization(processes) * 100.0

  unless utilization.nan?
    metrics << {
      metric_name: "Utilization",
      timestamp: now,
      value: utilization,
      unit: "Percent",
    }
  end

  processes.group_by do |process|
    process["tag"]
  end.each do |(tag, tag_processes)|
    next if tag.nil?

    tag_utilization = calculate_utilization(tag_processes) * 100.0

    unless tag_utilization.nan?
      metrics << {
        metric_name: "Utilization",
        dimensions: [{name: "Tag", value: tag}],
        timestamp: now,
        value: tag_utilization,
        unit: "Percent",
      }
    end
  end

  if @process_metrics
    processes.each do |process|
      process_utilization = process["busy"] / process["concurrency"].to_f * 100.0

      unless process_utilization.nan?
        process_dimensions = [{name: "Hostname", value: process["hostname"]}]

        if process["tag"]
          process_dimensions << {name: "Tag", value: process["tag"]}
        end

        metrics << {
          metric_name: "Utilization",
          dimensions: process_dimensions,
          timestamp: now,
          value: process_utilization,
          unit: "Percent",
        }
      end
    end
  end

  queues.each do |(queue_name, queue_size)|
    metrics << {
      metric_name: "QueueSize",
      dimensions: [{name: "QueueName", value: queue_name}],
      timestamp: now,
      value: queue_size,
      unit: "Count",
    }

    queue_latency = Sidekiq::Queue.new(queue_name).latency

    metrics << {
      metric_name: "QueueLatency",
      dimensions: [{name: "QueueName", value: queue_name}],
      timestamp: now,
      value: queue_latency,
      unit: "Seconds",
    }
  end

  unless @additional_dimensions.empty?
    metrics = metrics.each do |metric|
      metric[:dimensions] = (metric[:dimensions] || []) + @additional_dimensions
    end
  end

  # We can only put 20 metrics at a time
  metrics.each_slice(20) do |some_metrics|
    @client.put_metric_data(
      namespace: @namespace,
      metric_data: some_metrics,
    )
  end
end

#quiet ⇒ Object

# File 'lib/sidekiq/cloudwatchmetrics.rb', line 259

def quiet
  logger.debug { "Quieting Sidekiq CloudWatch Metrics Publisher" }
  @stop = true
end

#run ⇒ Object

# File 'lib/sidekiq/cloudwatchmetrics.rb', line 69

def run
  logger.info { "Started Sidekiq CloudWatch Metrics Publisher" }

  # Publish stats every INTERVAL seconds, sleeping as required between runs
  now = Time.now.to_f
  tick = now
  until @stop
    logger.debug { "Publishing Sidekiq CloudWatch Metrics" }
    publish

    now = Time.now.to_f
    tick = [tick + INTERVAL, now].max
    sleep(tick - now) if tick > now
  end

  logger.debug { "Stopped Sidekiq CloudWatch Metrics Publisher" }
end

#running? ⇒ Boolean

Returns:

• (Boolean)

# File 'lib/sidekiq/cloudwatchmetrics.rb', line 65

def running?
  !@thread.nil? && @thread.alive?
end

#start ⇒ Object

# File 'lib/sidekiq/cloudwatchmetrics.rb', line 58

def start
  logger.debug { "Starting Sidekiq CloudWatch Metrics Publisher" }

  @done = false
  @thread = safe_thread("cloudwatch metrics publisher", &method(:run))
end

#stop ⇒ Object

# File 'lib/sidekiq/cloudwatchmetrics.rb', line 264

def stop
  logger.debug { "Stopping Sidekiq CloudWatch Metrics Publisher" }
  @stop = true
  @thread.wakeup
  @thread.join
rescue ThreadError
  # Don't raise if thread is already dead.
  nil
end