Improve draining of delayed worker (#3999)

This change improves the draining behavior of the delayed workers:

  - Add shutdown_delayed_worker method to Drain class which configurable timeout
  - Add jobs:clear_pending_locks rake task to clear pending job locks in the delayed jobs table for the given worker.

Additionally it allows operators to set the number of delayed worker threads and the grace period after which threads will be killed via the jobs rake task.

Author: johha

lib/cloud_controller/config_schemas/base/api_schema.rb

@@ -341,8 +341,7 @@ class ApiSchema < VCAP::Config
               optional(:app_usage_events_cleanup) => { timeout_in_seconds: Integer },
               optional(:blobstore_delete) => { timeout_in_seconds: Integer },
               optional(:diego_sync) => { timeout_in_seconds: Integer },
-              optional(:priorities) => Hash,
-              optional(:number_of_worker_threads) => Integer
+              optional(:priorities) => Hash
             },
 
             # perm settings no longer have any effect but are preserved here

lib/cloud_controller/config_schemas/base/worker_schema.rb

@@ -166,8 +166,7 @@ class WorkerSchema < VCAP::Config
               optional(:app_usage_events_cleanup) => { timeout_in_seconds: Integer },
               optional(:blobstore_delete) => { timeout_in_seconds: Integer },
               optional(:diego_sync) => { timeout_in_seconds: Integer },
-              optional(:priorities) => Hash,
-              optional(:number_of_worker_threads) => Integer
+              optional(:priorities) => Hash
             },
 
             volume_services_enabled: bool,

lib/cloud_controller/drain.rb

@@ -38,6 +38,23 @@ def shutdown_cc(pid_path)
         log_shutdown_error(pid, process_name) unless wait_for_shutdown(pid, process_name, CCNG_FINAL_TIMEOUT)
       end
 
+      def shutdown_delayed_worker(pid_path, timeout=15)
+        pid = File.read(pid_path).to_i
+        process_name = File.basename(pid_path, '.pid')
+
+        # Initiate shutdown.
+        send_signal('TERM', pid, process_name)
+
+        # Wait some additional time for delayed worker to be terminated; otherwise write an error log message.
+        log_shutdown_error(pid, process_name) unless wait_for_shutdown(pid, process_name, timeout)
+
+        # force shutdown
+        return if terminated?(pid, process_name)
+
+        log_info("Forcefully shutting down process '#{process_name}' with pid '#{pid}'")
+        send_signal('KILL', pid, process_name)
+      end
+
       private
 
       def send_signal(signal, pid, process_name)

lib/delayed_job/delayed_worker.rb

@@ -9,6 +9,10 @@ def initialize(options)
       worker_name: options[:name],
       quiet: true
     }
+    return unless options[:num_threads] && options[:num_threads].to_i > 0
+
+    @queue_options[:num_threads] = options[:num_threads].to_i
+    @queue_options[:grace_period_seconds] = options[:thread_grace_period_seconds].to_i if options[:thread_grace_period_seconds] && options[:thread_grace_period_seconds].to_i > 0
   end
 
   def start_working
@@ -17,16 +21,36 @@ def start_working
 
     logger = Steno.logger('cc-worker')
     logger.info("Starting job with options #{@queue_options}")
+    setup_app_log_emitter(config, logger)
+
+    worker = get_initialized_delayed_worker(config, logger)
+    worker.start
+  end
 
+  def clear_locks!
+    config = RakeConfig.config
+    BackgroundJobEnvironment.new(config).setup_environment(readiness_port)
+
+    logger = Steno.logger('cc-worker-clear-locks')
+    logger.info("Clearing pending locks with options {#{@queue_options.map { |k, v| "#{k}: #{v.inspect}" }.join(', ')}}")
     setup_app_log_emitter(config, logger)
+
+    worker = get_initialized_delayed_worker(config, logger)
+    Delayed::Job.clear_locks!(worker.name)
+
+    # Clear locks for all threads when using a threaded worker
+    worker.names_with_threads.each { |name| Delayed::Job.clear_locks!(name) } if worker.respond_to?(:names_with_threads)
+  end
+
+  private
+
+  def get_initialized_delayed_worker(config, logger)
     Delayed::Worker.destroy_failed_jobs = false
     Delayed::Worker.max_attempts = 3
     Delayed::Worker.max_run_time = config.get(:jobs, :global, :timeout_in_seconds) + 1
     Delayed::Worker.logger = logger
 
-    num_worker_threads = config.get(:jobs, :number_of_worker_threads)
-    unless num_worker_threads.nil?
-      @queue_options[:num_threads] = num_worker_threads
+    unless @queue_options[:num_threads].nil?
       # Dynamically alias Delayed::Worker to ThreadedWorker to ensure plugins etc are working correctly
       Delayed.module_eval do
         remove_const(:Worker) if const_defined?(:Worker)
@@ -36,11 +60,9 @@ def start_working
 
     worker = Delayed::Worker.new(@queue_options)
     worker.name = @queue_options[:worker_name]
-    worker.start
+    worker
   end
 
-  private
-
   def setup_app_log_emitter(config, logger)
     VCAP::AppLogEmitter.fluent_emitter = fluent_emitter(config) if config.get(:fluent)
     if config.get(:loggregator) && config.get(

lib/delayed_job/threaded_worker.rb

@@ -28,11 +28,11 @@ def start
         raise SignalException.new('INT') if self.class.raise_signal_exceptions && self.class.raise_signal_exceptions != :term
       end
 
-      say "Starting threaded delayed worker with #{@num_threads} threads"
+      say "Starting threaded delayed worker with #{@num_threads} threads and grace period of #{@grace_period_seconds} seconds"
 
       @num_threads.times do |thread_index|
         thread = Thread.new do
-          Thread.current[:thread_name] = "thread:#{thread_index + 1}"
+          Thread.current[:thread_index] = thread_index
           threaded_start
         rescue Exception => e # rubocop:disable Lint/RescueException
           say "Unexpected error: #{e.message}\n#{e.backtrace.join("\n")}", 'error'
@@ -49,10 +49,13 @@ def start
       raise 'Unexpected error occurred in one of the worker threads' if @unexpected_error
     end
 
+    def names_with_threads
+      base_name = name
+      @num_threads.times.map { |thread_index| extended_name_with_thread_index(base_name, thread_index) }
+    end
+
     def name
-      base_name = super
-      thread_name = Thread.current[:thread_name]
-      thread_name.nil? ? base_name : "#{base_name} #{thread_name}"
+      extended_name_with_thread_index(super)
     end
 
     def stop
@@ -64,7 +67,7 @@ def stop
           Thread.new do
             t.join(@grace_period_seconds)
             if t.alive?
-              say "Killing thread '#{t[:thread_name]}'"
+              say "Killing thread '#{t[:thread_index]}'"
               t.kill
             end
           end
@@ -98,5 +101,18 @@ def threaded_start
         end
       end
     end
+
+    private
+
+    def extended_name_with_thread_index(base_name, thread_index=nil)
+      raise ArgumentError.new('base_name cannot be nil or empty') if base_name.nil? || base_name.empty?
+
+      thread_index = Thread.current[:thread_index] if thread_index.nil?
+      thread_name = "thread:#{thread_index + 1}" if thread_index.is_a? Integer
+
+      return base_name if thread_name.nil?
+
+      "#{base_name} #{thread_name}"
+    end
   end
 end

lib/tasks/jobs.rake

@@ -11,6 +11,18 @@ namespace :jobs do
     end
   end
 
+  desc 'Clear pending locks for the current delayed worker.'
+  task :clear_pending_locks, [:name] => :environment do |_t, args|
+    puts RUBY_DESCRIPTION
+    puts "Clearing pending locks for worker: #{args.name}"
+    args.with_defaults(name: ENV.fetch('HOSTNAME', nil))
+
+    RakeConfig.context = :worker
+
+    CloudController::DelayedWorker.new(queues: [],
+                                       name: args.name).clear_locks!
+  end
+
   desc 'Start a delayed_job worker that works on jobs that require access to local resources.'
 
   task :local, [:name] => :environment do |_t, args|
@@ -25,9 +37,11 @@ namespace :jobs do
   end
 
   desc 'Start a delayed_job worker.'
-  task :generic, [:name] => :environment do |_t, args|
+  task :generic, %i[name num_threads thread_grace_period_seconds] => :environment do |_t, args|
     puts RUBY_DESCRIPTION
     args.with_defaults(name: ENV.fetch('HOSTNAME', nil))
+    args.with_defaults(num_threads: nil)
+    args.with_defaults(thread_grace_period_seconds: nil)
 
     RakeConfig.context = :worker
     queues = [
@@ -50,7 +64,6 @@ namespace :jobs do
       'prune_excess_app_revisions'
     ]
 
-    CloudController::DelayedWorker.new(queues: queues,
-                                       name: args.name).start_working
+    CloudController::DelayedWorker.new(queues: queues, name: args.name, num_threads: args.num_threads, thread_grace_period_seconds: args.thread_grace_period_seconds).start_working
   end
 end

spec/unit/lib/cloud_controller/drain_spec.rb

@@ -14,8 +14,9 @@ def log_contents
     end
 
     let(:pid) { 23_456 }
+    let(:pid_name) { 'pidfile' }
     let(:pid_dir) { Dir.mktmpdir }
-    let(:pid_path) { File.join(pid_dir, 'pidfile') }
+    let(:pid_path) { File.join(pid_dir, pid_name) }
 
     before do
       File.write(pid_path, pid)
@@ -38,7 +39,7 @@ def log_contents
         drain.shutdown_nginx(pid_path)
 
         log_contents do |log|
-          expect(log).to match(/Sending signal '\w+' to process '\w+' with pid '\d+'/)
+          expect(log).to include("Sending signal 'QUIT' to process '#{pid_name}' with pid '#{pid}'")
         end
       end
 
@@ -50,8 +51,9 @@ def log_contents
 
         expect(drain).to have_received(:sleep).twice
         log_contents do |log|
-          expect(log).to match(/Waiting \d+s for process '\w+' with pid '\d+' to shutdown/)
-          expect(log).to match(/Process '\w+' with pid '\d+' is not running/)
+          expect(log).to include("Waiting 30s for process '#{pid_name}' with pid '#{pid}' to shutdown")
+          expect(log).to include("Waiting 29s for process '#{pid_name}' with pid '#{pid}' to shutdown")
+          expect(log).to include("Process '#{pid_name}' with pid '#{pid}' is not running")
         end
       end
 
@@ -84,7 +86,7 @@ def log_contents
 
         expect(drain).to have_received(:sleep).exactly(40).times
         log_contents do |log|
-          expect(log).to match(/Process '\w+' with pid '\d+' is still running - this indicates an error in the shutdown procedure!/)
+          expect(log).to include("Process '#{pid_name}' with pid '#{pid}' is still running - this indicates an error in the shutdown procedure!")
         end
       end
     end
@@ -96,7 +98,7 @@ def log_contents
         drain.shutdown_cc(pid_path)
 
         log_contents do |log|
-          expect(log).to match(/Sending signal '\w+' to process '\w+' with pid '\d+'/)
+          expect(log).to match("Sending signal 'TERM' to process '#{pid_name}' with pid '#{pid}'")
         end
       end
 
@@ -107,7 +109,53 @@ def log_contents
 
         expect(drain).to have_received(:sleep).exactly(20).times
         log_contents do |log|
-          expect(log).to match(/Process '\w+' with pid '\d+' is still running - this indicates an error in the shutdown procedure!/)
+          expect(log).to match("Process '#{pid_name}' with pid '#{pid}' is still running - this indicates an error in the shutdown procedure!")
+        end
+      end
+    end
+
+    describe '#shutdown_delayed_worker' do
+      it 'sends TERM to the delayed worker process specified in the pid file' do
+        expect(Process).to receive(:kill).with('TERM', pid)
+
+        drain.shutdown_delayed_worker(pid_path)
+
+        log_contents do |log|
+          expect(log).to include("Sending signal 'TERM' to process '#{pid_name}' with pid '#{pid}'")
+        end
+      end
+
+      it 'waits 15s after sending TERM' do
+        allow(Process).to receive(:getpgid).with(pid).and_return(1)
+
+        drain.shutdown_delayed_worker(pid_path)
+
+        expect(drain).to have_received(:sleep).exactly(15).times
+        log_contents do |log|
+          expect(log).to include("Process '#{pid_name}' with pid '#{pid}' is still running - this indicates an error in the shutdown procedure!")
+        end
+      end
+
+      it 'waits for the given timeout after sending TERM' do
+        allow(Process).to receive(:getpgid).with(pid).and_return(1)
+
+        drain.shutdown_delayed_worker(pid_path, 30)
+
+        expect(drain).to have_received(:sleep).exactly(30).times
+        log_contents do |log|
+          expect(log).to include("Process '#{pid_name}' with pid '#{pid}' is still running - this indicates an error in the shutdown procedure!")
+        end
+      end
+
+      it 'sends KILL to the delayed worker process if it is still running after 15s' do
+        allow(Process).to receive(:getpgid).with(pid).and_return(1)
+        allow(Process).to receive(:kill).with('TERM', pid)
+        expect(Process).to receive(:kill).with('KILL', pid)
+
+        drain.shutdown_delayed_worker(pid_path)
+
+        log_contents do |log|
+          expect(log).to include("Forcefully shutting down process '#{pid_name}' with pid '#{pid}'")
         end
       end
     end