Revert "FIX: Heartbeat check per sidekiq process (#7873)"
This reverts commit 340855da55
.
This commit is contained in:
parent
aecadcb267
commit
e805d44965
|
@ -2,8 +2,19 @@
|
||||||
|
|
||||||
module Jobs
|
module Jobs
|
||||||
class RunHeartbeat < Jobs::Base
|
class RunHeartbeat < Jobs::Base
|
||||||
|
|
||||||
|
sidekiq_options queue: 'critical'
|
||||||
|
|
||||||
|
def self.heartbeat_key
|
||||||
|
'heartbeat_last_run'
|
||||||
|
end
|
||||||
|
|
||||||
def execute(args)
|
def execute(args)
|
||||||
Demon::Sidekiq.trigger_heartbeat(args[:queue_name])
|
$redis.set(self.class.heartbeat_key, Time.new.to_i.to_s)
|
||||||
|
end
|
||||||
|
|
||||||
|
def self.last_heartbeat
|
||||||
|
$redis.get(heartbeat_key).to_i
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -7,9 +7,7 @@ module Jobs
|
||||||
every 3.minute
|
every 3.minute
|
||||||
|
|
||||||
def execute(args)
|
def execute(args)
|
||||||
Demon::Sidekiq::QUEUE_IDS.each do |identifier|
|
Jobs.enqueue(:run_heartbeat, {})
|
||||||
Jobs.enqueue(:run_heartbeat, queue_name: identifier, queue: identifier)
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -144,32 +144,25 @@ before_fork do |server, worker|
|
||||||
@sidekiq_next_heartbeat_check ||= Time.new.to_i + @sidekiq_heartbeat_interval
|
@sidekiq_next_heartbeat_check ||= Time.new.to_i + @sidekiq_heartbeat_interval
|
||||||
|
|
||||||
if @sidekiq_next_heartbeat_check < Time.new.to_i
|
if @sidekiq_next_heartbeat_check < Time.new.to_i
|
||||||
@sidekiq_next_heartbeat_check = Time.new.to_i + @sidekiq_heartbeat_interval
|
|
||||||
restarted = false
|
last_heartbeat = Jobs::RunHeartbeat.last_heartbeat
|
||||||
|
restart = false
|
||||||
|
|
||||||
if out_of_memory?
|
if out_of_memory?
|
||||||
Rails.logger.warn("Sidekiq is consuming too much memory (using: %0.2fM) for '%s', restarting" % [(max_rss.to_f / 1.megabyte), ENV["DISCOURSE_HOSTNAME"]])
|
Rails.logger.warn("Sidekiq is consuming too much memory (using: %0.2fM) for '%s', restarting" % [(max_rss.to_f / 1.megabyte), ENV["DISCOURSE_HOSTNAME"]])
|
||||||
|
restart = true
|
||||||
|
end
|
||||||
|
|
||||||
|
if last_heartbeat < Time.new.to_i - @sidekiq_heartbeat_interval
|
||||||
|
STDERR.puts "Sidekiq heartbeat test failed, restarting"
|
||||||
|
Rails.logger.warn "Sidekiq heartbeat test failed, restarting"
|
||||||
|
|
||||||
|
restart = true
|
||||||
|
end
|
||||||
|
@sidekiq_next_heartbeat_check = Time.new.to_i + @sidekiq_heartbeat_interval
|
||||||
|
|
||||||
|
if restart
|
||||||
Demon::Sidekiq.restart
|
Demon::Sidekiq.restart
|
||||||
restarted = true
|
|
||||||
end
|
|
||||||
|
|
||||||
if !restarted
|
|
||||||
Demon::Sidekiq::QUEUE_IDS.each do |identifier|
|
|
||||||
last_heartbeat = Demon::Sidekiq.get_queue_last_heartbeat(identifier)
|
|
||||||
|
|
||||||
if last_heartbeat < Time.new.to_i - @sidekiq_heartbeat_interval
|
|
||||||
if demon = Demon::Sidekiq.demons.values.find { |d| d.identifier == identifier }
|
|
||||||
STDERR.puts "Sidekiq heartbeat test for worker #{demon.pid} failed, restarting"
|
|
||||||
Rails.logger.warn "Sidekiq heartbeat test for worker #{demon.pid} failed, restarting"
|
|
||||||
demon.stop
|
|
||||||
demon.start
|
|
||||||
restarted = true
|
|
||||||
end
|
|
||||||
end
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
if restarted
|
|
||||||
sleep 10
|
sleep 10
|
||||||
force_kill_rogue_sidekiq
|
force_kill_rogue_sidekiq
|
||||||
end
|
end
|
||||||
|
|
|
@ -11,7 +11,6 @@ class Demon::Base
|
||||||
|
|
||||||
def self.start(count = 1, verbose: false)
|
def self.start(count = 1, verbose: false)
|
||||||
@demons ||= {}
|
@demons ||= {}
|
||||||
before_start(count)
|
|
||||||
count.times do |i|
|
count.times do |i|
|
||||||
(@demons["#{prefix}_#{i}"] ||= new(i, verbose: verbose)).start
|
(@demons["#{prefix}_#{i}"] ||= new(i, verbose: verbose)).start
|
||||||
end
|
end
|
||||||
|
@ -38,10 +37,7 @@ class Demon::Base
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
def self.before_start(count)
|
attr_reader :pid, :parent_pid, :started, :index
|
||||||
end
|
|
||||||
|
|
||||||
attr_reader :pid, :parent_pid, :started, :index, :identifier
|
|
||||||
attr_accessor :stop_timeout
|
attr_accessor :stop_timeout
|
||||||
|
|
||||||
def initialize(index, rails_root: nil, parent_pid: nil, verbose: false)
|
def initialize(index, rails_root: nil, parent_pid: nil, verbose: false)
|
||||||
|
|
|
@ -3,38 +3,6 @@
|
||||||
require "demon/base"
|
require "demon/base"
|
||||||
|
|
||||||
class Demon::Sidekiq < 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
|
def self.prefix
|
||||||
"sidekiq"
|
"sidekiq"
|
||||||
|
@ -44,11 +12,6 @@ class Demon::Sidekiq < Demon::Base
|
||||||
blk ? (@blk = blk) : @blk
|
blk ? (@blk = blk) : @blk
|
||||||
end
|
end
|
||||||
|
|
||||||
def run
|
|
||||||
@identifier = QUEUE_IDS[@index]
|
|
||||||
super
|
|
||||||
end
|
|
||||||
|
|
||||||
private
|
private
|
||||||
|
|
||||||
def suppress_stdout
|
def suppress_stdout
|
||||||
|
@ -73,7 +36,7 @@ class Demon::Sidekiq < Demon::Base
|
||||||
|
|
||||||
options = ["-c", GlobalSetting.sidekiq_workers.to_s]
|
options = ["-c", GlobalSetting.sidekiq_workers.to_s]
|
||||||
|
|
||||||
[['critical', 8], [@identifier, 8], ['default', 4], ['low', 2], ['ultra_low', 1]].each do |queue_name, weight|
|
[['critical', 8], ['default', 4], ['low', 2], ['ultra_low', 1]].each do |queue_name, weight|
|
||||||
custom_queue_hostname = ENV["UNICORN_SIDEKIQ_#{queue_name.upcase}_QUEUE_HOSTNAME"]
|
custom_queue_hostname = ENV["UNICORN_SIDEKIQ_#{queue_name.upcase}_QUEUE_HOSTNAME"]
|
||||||
|
|
||||||
if !custom_queue_hostname || custom_queue_hostname.split(',').include?(`hostname`.strip)
|
if !custom_queue_hostname || custom_queue_hostname.split(',').include?(`hostname`.strip)
|
||||||
|
|
|
@ -2,7 +2,6 @@
|
||||||
|
|
||||||
require 'rails_helper'
|
require 'rails_helper'
|
||||||
require_dependency 'jobs/base'
|
require_dependency 'jobs/base'
|
||||||
require_dependency 'demon/sidekiq'
|
|
||||||
|
|
||||||
describe Jobs::Heartbeat do
|
describe Jobs::Heartbeat do
|
||||||
after do
|
after do
|
||||||
|
@ -11,14 +10,12 @@ describe Jobs::Heartbeat do
|
||||||
|
|
||||||
it "still enqueues heartbeats in readonly mode" do
|
it "still enqueues heartbeats in readonly mode" do
|
||||||
freeze_time 1.week.from_now
|
freeze_time 1.week.from_now
|
||||||
Demon::Sidekiq.clear_heartbeat_queues!
|
|
||||||
Jobs.run_immediately!
|
|
||||||
|
|
||||||
Discourse.enable_readonly_mode
|
Discourse.enable_readonly_mode
|
||||||
|
|
||||||
queue = SecureRandom.hex
|
Sidekiq::Testing.inline! do
|
||||||
Demon::Sidekiq::QUEUE_IDS << queue
|
Jobs::Heartbeat.new.perform(nil)
|
||||||
Jobs::Heartbeat.new.perform(nil)
|
expect(Jobs::RunHeartbeat.last_heartbeat).to eq(Time.new.to_i)
|
||||||
expect(Demon::Sidekiq.get_queue_last_heartbeat(queue)).to eq(Time.new.to_i)
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
Loading…
Reference in New Issue