mirror of
https://github.com/discourse/discourse.git
synced 2024-11-25 19:03:45 +08:00
c3497559be
This reverts commit e805d44965.
We now have mechanisms in place to ensure heartbeat will always
be scheduled even if the scheduler is overloaded per: 098f938b
99 lines
2.5 KiB
Ruby
99 lines
2.5 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
require "demon/base"
|
|
|
|
class Demon::Sidekiq < Demon::Base
|
|
RANDOM_HEX = SecureRandom.hex
|
|
QUEUE_IDS = []
|
|
|
|
def self.queues_last_heartbeat_hash_key
|
|
@@queues_last_heartbeat_hash_key ||= "#{RANDOM_HEX}_queues_last_heartbeat_hash"
|
|
end
|
|
|
|
def self.trigger_heartbeat(name)
|
|
$redis.hset(queues_last_heartbeat_hash_key, name, Time.new.to_i.to_s)
|
|
extend_expiry(queues_last_heartbeat_hash_key)
|
|
end
|
|
|
|
def self.get_queue_last_heartbeat(name)
|
|
extend_expiry(queues_last_heartbeat_hash_key)
|
|
$redis.hget(queues_last_heartbeat_hash_key, name).to_i
|
|
end
|
|
|
|
def self.clear_heartbeat_queues!
|
|
$redis.del(queues_last_heartbeat_hash_key)
|
|
end
|
|
|
|
def self.before_start(count)
|
|
# cleans up heartbeat queues from previous boot up
|
|
Sidekiq::Queue.all.each { |queue| queue.clear if queue.name[/^\h{32}$/] }
|
|
count.times do
|
|
QUEUE_IDS << SecureRandom.hex
|
|
end
|
|
end
|
|
|
|
def self.extend_expiry(key)
|
|
$redis.expire(key, 60 * 60)
|
|
end
|
|
|
|
def self.prefix
|
|
"sidekiq"
|
|
end
|
|
|
|
def self.after_fork(&blk)
|
|
blk ? (@blk = blk) : @blk
|
|
end
|
|
|
|
def run
|
|
@identifier = QUEUE_IDS[@index]
|
|
super
|
|
end
|
|
|
|
private
|
|
|
|
def suppress_stdout
|
|
false
|
|
end
|
|
|
|
def suppress_stderr
|
|
false
|
|
end
|
|
|
|
def after_fork
|
|
Demon::Sidekiq.after_fork&.call
|
|
|
|
puts "Loading Sidekiq in process id #{Process.pid}"
|
|
require 'sidekiq/cli'
|
|
# CLI will close the logger, if we have one set we can be in big
|
|
# trouble, if STDOUT is closed in our process all sort of weird
|
|
# will ensue, resetting the logger ensures it will reinit correctly
|
|
# parent process is in charge of the file anyway.
|
|
Sidekiq::Logging.logger = nil
|
|
cli = Sidekiq::CLI.instance
|
|
|
|
options = ["-c", GlobalSetting.sidekiq_workers.to_s]
|
|
|
|
[['critical', 8], [@identifier, 8], ['default', 4], ['low', 2], ['ultra_low', 1]].each do |queue_name, weight|
|
|
custom_queue_hostname = ENV["UNICORN_SIDEKIQ_#{queue_name.upcase}_QUEUE_HOSTNAME"]
|
|
|
|
if !custom_queue_hostname || custom_queue_hostname.split(',').include?(`hostname`.strip)
|
|
options << "-q"
|
|
options << "#{queue_name},#{weight}"
|
|
end
|
|
end
|
|
|
|
# Sidekiq not as high priority as web, in this environment it is forked so a web is very
|
|
# likely running
|
|
Discourse::Utils.execute_command('renice', '-n', '5', '-p', Process.pid.to_s)
|
|
|
|
cli.parse(options)
|
|
load Rails.root + "config/initializers/100-sidekiq.rb"
|
|
cli.run
|
|
rescue => e
|
|
STDERR.puts e.message
|
|
STDERR.puts e.backtrace.join("\n")
|
|
exit 1
|
|
end
|
|
|
|
end
|