Fix lifecycle_type_backfill job never running and piling up (#5286)

* Fix lifecycle_type_backfill job never running and piling up

The job's queue was missing from the generic worker's allowlist, so it
was enqueued but never executed.

Also make the scheduler skip enqueuing when an untouched job is already
queued, and keep only the newest untouched job, removing the rest.

* Derive generic worker queues from the scheduler

The generic worker's queue list was maintained by hand and had to be
kept in sync with the clock scheduler. A missing entry meant a job was
enqueued but never consumed. Build the list from the scheduler's job
definitions so new periodic jobs are wired up automatically.

Author: philippthun

lib/cloud_controller/clock/distributed_executor.rb

@@ -13,6 +13,8 @@ def execute_job(name:, interval:, fudge:, timeout:)
       ClockJob.db.transaction do
         job = ClockJob.find(name:).lock!
 
+        remove_duplicate_untouched_jobs(job.name)
+
         need_to_run_job = need_to_run_job?(job, interval, timeout, fudge)
 
         if need_to_run_job
@@ -67,9 +69,33 @@ def job_in_progress?(job)
       end
     end
 
+    # Keep only the newest untouched job on the queue and remove the rest.
+    def remove_duplicate_untouched_jobs(name)
+      return if name == 'diego_sync'
+
+      newest = untouched_jobs(name).order(Sequel.desc(:created_at), Sequel.desc(:id)).first
+      return if newest.nil?
+
+      deleted = untouched_jobs(name).exclude(id: newest.id).delete
+      @logger.info("Removed #{deleted} duplicate untouched job(s) from queue #{name}") if deleted > 0
+    end
+
+    def untouched_job_queued?(job)
+      return false if job.name == 'diego_sync'
+
+      queued = untouched_jobs(job.name).any?
+      @logger.info("An untouched job is already queued for #{job.name}, skipping enqueue") if queued
+      queued
+    end
+
+    def untouched_jobs(name)
+      Delayed::Job.where(queue: name, failed_at: nil, locked_at: nil)
+    end
+
     def need_to_run_job?(job, interval, timeout, fudge=0)
       return true if never_run?(job)
       return false unless interval_elapsed?(job, interval, fudge)
+      return false if untouched_job_queued?(job)
       return true unless job_in_progress?(job)
 
       if timeout.nil?

lib/cloud_controller/clock/scheduler.rb

@@ -30,6 +30,10 @@ class Scheduler
       { name: 'lifecycle_type_backfill', class: Jobs::Runtime::LifecycleTypeBackfill }
     ].freeze
 
+    def self.queue_names
+      (CLEANUPS + FREQUENTS).pluck(:name)
+    end
+
     def initialize(config)
       @clock = Clock.new
       @config = config

lib/tasks/jobs.rake

@@ -39,30 +39,15 @@ namespace :jobs do
 
   desc 'Start a delayed_job worker.'
   task :generic, %i[name num_threads thread_grace_period_seconds] => :environment do |_t, args|
+    require 'cloud_controller/clock/scheduler'
     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)
 
     queues = [
       VCAP::CloudController::Jobs::Queues.generic,
-      'app_usage_events',
-      'audit_events',
-      'failed_jobs',
-      'service_operations_initial_cleanup',
-      'service_operations_create_in_progress_cleanup',
-      'service_usage_events',
-      'completed_tasks',
-      'expired_blob_cleanup',
-      'expired_resource_cleanup',
-      'expired_orphaned_blob_cleanup',
-      'orphaned_blobs_cleanup',
-      'pollable_job_cleanup',
-      'pending_droplets',
-      'pending_builds',
-      'prune_completed_deployments',
-      'prune_completed_builds',
-      'prune_excess_app_revisions'
+      *VCAP::CloudController::Scheduler.queue_names
     ]
 
     require 'cloud_controller/metrics/custom_process_id'

spec/unit/lib/cloud_controller/clock/distributed_executor_spec.rb

@@ -232,6 +232,40 @@ module VCAP::CloudController
           end
         end
       end
+
+      context 'when an untouched job is already queued and the job timeout has expired' do
+        before do
+          DistributedExecutor.new.execute_job(name: job_name, interval: 1.minute, fudge: 1.second, timeout: 5.minutes) do
+            Delayed::Job.create!(queue: job_name, failed_at: nil, locked_at: nil)
+          end
+          Timecop.travel(Time.now.utc + 6.minutes)
+        end
+
+        it 'does not enqueue another job' do
+          executed = false
+          DistributedExecutor.new.execute_job name: job_name, interval: 1.minute, fudge: 1.second, timeout: 5.minutes do
+            executed = true
+          end
+          expect(executed).to be(false)
+        end
+      end
+
+      context 'when multiple untouched jobs are already queued and the job timeout has expired' do
+        let!(:newer_job) { Delayed::Job.create!(queue: job_name, failed_at: nil, locked_at: nil, created_at: Time.now.utc) }
+        let!(:older_job) { Delayed::Job.create!(queue: job_name, failed_at: nil, locked_at: nil, created_at: Time.now.utc - 1.hour) }
+        let!(:locked_job) { Delayed::Job.create!(queue: job_name, failed_at: nil, locked_at: Time.now) }
+        let!(:failed_job) { Delayed::Job.create!(queue: job_name, failed_at: Time.now, locked_at: nil) }
+
+        before do
+          DistributedExecutor.new.execute_job(name: job_name, interval: 1.minute, fudge: 1.second, timeout: 5.minutes) {}
+          Timecop.travel(Time.now.utc + 6.minutes)
+        end
+
+        it 'purges all but the newest untouched job, it keeps locked and failed jobs' do
+          DistributedExecutor.new.execute_job(name: job_name, interval: 1.minute, fudge: 1.second, timeout: 5.minutes) {}
+          expect(Delayed::Job.where(queue: job_name).all).to contain_exactly(newer_job, locked_job, failed_job)
+        end
+      end
     end
   end
 end

spec/unit/lib/cloud_controller/clock/scheduler_spec.rb

@@ -236,5 +236,12 @@ module VCAP::CloudController
         end
       end
     end
+
+    describe '.queue_names' do
+      it 'returns the queue name of every scheduled job' do
+        expected = (Scheduler::CLEANUPS + Scheduler::FREQUENTS).pluck(:name)
+        expect(Scheduler.queue_names).to match_array(expected)
+      end
+    end
   end
 end