Add new metric job_queue_load for cc-worker (#3694)

* Add new metric job_queue_load for cc-worker

* Adjust code based on review

Author: WeiQuan0605

lib/cloud_controller/metrics/periodic_updater.rb

@@ -18,6 +18,7 @@ def setup_updates
       update!
       EM.add_periodic_timer(600) { catch_error { update_user_count } }
       EM.add_periodic_timer(30)  { catch_error { update_job_queue_length } }
+      EM.add_periodic_timer(30)  { catch_error { update_job_queue_load } }
       EM.add_periodic_timer(30)  { catch_error { update_thread_info } }
       EM.add_periodic_timer(30)  { catch_error { update_failed_job_count } }
       EM.add_periodic_timer(30)  { catch_error { update_vitals } }
@@ -30,6 +31,7 @@ def setup_updates
     def update!
       update_user_count
       update_job_queue_length
+      update_job_queue_load
       update_thread_info
       update_failed_job_count
       update_vitals
@@ -91,6 +93,23 @@ def update_job_queue_length
       @prometheus_updater.update_job_queue_length(pending_job_count_by_queue)
     end
 
+    def update_job_queue_load
+      jobs_by_queue_with_run_now = Delayed::Job.
+                                   where(Sequel.lit('run_at <= ?', Time.now)).
+                                   where(Sequel.lit('failed_at IS NULL')).group_and_count(:queue)
+
+      total = 0
+      pending_job_load_by_queue = jobs_by_queue_with_run_now.each_with_object({}) do |row, hash|
+        @known_job_queues[row[:queue].to_sym] = 0
+        total += row[:count]
+        hash[row[:queue].to_sym] = row[:count]
+      end
+
+      pending_job_load_by_queue.reverse_merge!(@known_job_queues)
+      @statsd_updater.update_job_queue_load(pending_job_load_by_queue, total)
+      @prometheus_updater.update_job_queue_load(pending_job_load_by_queue)
+    end
+
     def update_thread_info
       return unless VCAP::CloudController::Config.config.get(:webserver) == 'thin'
 

lib/cloud_controller/metrics/prometheus_updater.rb

@@ -19,6 +19,7 @@ def self.allow_pid_label
 
     METRICS = [
       { type: :gauge, name: :cc_job_queues_length_total, docstring: 'Job queues length of worker processes', labels: [:queue], aggregation: :most_recent },
+      { type: :gauge, name: :cc_job_queues_load_total, docstring: 'Number of background jobs ready to run now ', labels: [:queue], aggregation: :most_recent },
       { type: :gauge, name: :cc_failed_jobs_total, docstring: 'Number of failed jobs of worker processes', labels: [:queue], aggregation: :most_recent },
       { type: :counter, name: :cc_staging_requests_total, docstring: 'Number of staging requests' },
       { type: :histogram, name: :cc_staging_succeeded_duration_seconds, docstring: 'Durations of successful staging events', buckets: DURATION_BUCKETS },
@@ -102,6 +103,12 @@ def update_job_queue_length(pending_job_count_by_queue)
       end
     end
 
+    def update_job_queue_load(update_job_queue_load)
+      update_job_queue_load.each do |key, value|
+        update_gauge_metric(:cc_job_queues_load_total, value, labels: { queue: key.to_s.underscore })
+      end
+    end
+
     def update_thread_info_thin(thread_info)
       update_gauge_metric(:cc_thread_info_thread_count, thread_info[:thread_count])
       update_gauge_metric(:cc_thread_info_event_machine_connection_count, thread_info[:event_machine][:connection_count])

lib/cloud_controller/metrics/statsd_updater.rb

@@ -23,6 +23,15 @@ def update_job_queue_length(pending_job_count_by_queue, total)
       end
     end
 
+    def update_job_queue_load(pending_job_load_by_queue, total)
+      @statsd.batch do |batch|
+        pending_job_load_by_queue.each do |key, value|
+          batch.gauge("cc.job_queue_load.#{key}", value)
+        end
+        batch.gauge('cc.job_queue_load.total', total)
+      end
+    end
+
     def update_thread_info_thin(thread_info)
       @statsd.batch do |batch|
         batch.gauge('cc.thread_info.thread_count', thread_info[:thread_count])

spec/request/internal/metrics_spec.rb

@@ -83,6 +83,42 @@
     end
   end
 
+  context 'cc_job_queue_load' do
+    before do
+      Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), { queue: 'cc_api_0', run_at: Time.now })
+      Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), { queue: 'cc_generic', run_at: Time.now })
+    end
+
+    after do
+      Delayed::Job.dataset.delete
+    end
+
+    it 'includes job queue load metric labelled for each queue' do
+      get '/internal/v4/metrics', nil
+
+      expect(last_response.body).to match(/cc_job_queues_load_total{queue="cc_api_0"} 1\.0/)
+      expect(last_response.body).to match(/cc_job_queues_load_total{queue="cc_generic"} 1\.0/)
+    end
+  end
+
+  context 'cc_job_queue_load_not_ready_to_run_now' do
+    before do
+      Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), { queue: 'cc_api_0', run_at: Time.now + 1.minute })
+      Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), { queue: 'cc_generic', run_at: Time.now + 1.minute })
+    end
+
+    after do
+      Delayed::Job.dataset.delete
+    end
+
+    it 'includes job queue load metric labelled for each queue' do
+      get '/internal/v4/metrics', nil
+
+      expect(last_response.body).to match(/cc_job_queues_load_total{queue="cc_api_0"} 0\.0/)
+      expect(last_response.body).to match(/cc_job_queues_load_total{queue="cc_generic"} 0\.0/)
+    end
+  end
+
   context 'cc_thread_info' do
     it 'reports thread info' do
       get '/internal/v4/metrics', nil

spec/unit/lib/cloud_controller/metrics/periodic_updater_spec.rb

@@ -72,6 +72,7 @@ module VCAP::CloudController::Metrics
       before do
         allow(statsd_updater).to receive(:update_user_count)
         allow(statsd_updater).to receive(:update_job_queue_length)
+        allow(statsd_updater).to receive(:update_job_queue_load)
         allow(statsd_updater).to receive(:update_thread_info_thin)
         allow(statsd_updater).to receive(:update_failed_job_count)
         allow(statsd_updater).to receive(:update_vitals)
@@ -81,6 +82,7 @@ module VCAP::CloudController::Metrics
 
         allow(prometheus_updater).to receive(:update_user_count)
         allow(prometheus_updater).to receive(:update_job_queue_length)
+        allow(prometheus_updater).to receive(:update_job_queue_load)
         allow(prometheus_updater).to receive(:update_thread_info_thin)
         allow(prometheus_updater).to receive(:update_failed_job_count)
         allow(prometheus_updater).to receive(:update_vitals)
@@ -101,6 +103,11 @@ module VCAP::CloudController::Metrics
         periodic_updater.setup_updates
       end
 
+      it 'bumps the load of cc job queues and sets periodic timer' do
+        expect(periodic_updater).to receive(:update_job_queue_load).once
+        periodic_updater.setup_updates
+      end
+
       it 'bumps the length of cc failed job queues and sets periodic timer' do
         expect(periodic_updater).to receive(:update_failed_job_count).once
         periodic_updater.setup_updates
@@ -161,45 +168,53 @@ module VCAP::CloudController::Metrics
           @periodic_timers[1][:block].call
         end
 
-        it 'updates thread count and event machine queues' do
+        it 'bumps the load of cc job queues and sets periodic timer' do
           expect(periodic_updater).to receive(:catch_error).once.and_call_original
-          expect(periodic_updater).to receive(:update_thread_info).once
+          expect(periodic_updater).to receive(:update_job_queue_load).once
           expect(@periodic_timers[2][:interval]).to eq(30)
 
           @periodic_timers[2][:block].call
         end
 
-        it 'bumps the length of cc failed job queues and sets periodic timer' do
+        it 'updates thread count and event machine queues' do
           expect(periodic_updater).to receive(:catch_error).once.and_call_original
-          expect(periodic_updater).to receive(:update_failed_job_count).once
+          expect(periodic_updater).to receive(:update_thread_info).once
           expect(@periodic_timers[3][:interval]).to eq(30)
 
           @periodic_timers[3][:block].call
         end
 
-        it 'updates the vitals' do
+        it 'bumps the length of cc failed job queues and sets periodic timer' do
           expect(periodic_updater).to receive(:catch_error).once.and_call_original
-          expect(periodic_updater).to receive(:update_vitals).once
+          expect(periodic_updater).to receive(:update_failed_job_count).once
           expect(@periodic_timers[4][:interval]).to eq(30)
 
           @periodic_timers[4][:block].call
         end
 
-        it 'updates the log counts' do
+        it 'updates the vitals' do
           expect(periodic_updater).to receive(:catch_error).once.and_call_original
-          expect(periodic_updater).to receive(:update_log_counts).once
+          expect(periodic_updater).to receive(:update_vitals).once
           expect(@periodic_timers[5][:interval]).to eq(30)
 
           @periodic_timers[5][:block].call
         end
 
-        it 'updates the task stats' do
+        it 'updates the log counts' do
           expect(periodic_updater).to receive(:catch_error).once.and_call_original
-          expect(periodic_updater).to receive(:update_task_stats).once
+          expect(periodic_updater).to receive(:update_log_counts).once
           expect(@periodic_timers[6][:interval]).to eq(30)
 
           @periodic_timers[6][:block].call
         end
+
+        it 'updates the task stats' do
+          expect(periodic_updater).to receive(:catch_error).once.and_call_original
+          expect(periodic_updater).to receive(:update_task_stats).once
+          expect(@periodic_timers[7][:interval]).to eq(30)
+
+          @periodic_timers[7][:block].call
+        end
       end
     end
 
@@ -353,6 +368,82 @@ module VCAP::CloudController::Metrics
       end
     end
 
+    describe '#update_job_queue_load' do
+      before do
+        allow(statsd_updater).to receive(:update_job_queue_load)
+        allow(prometheus_updater).to receive(:update_job_queue_load)
+      end
+
+      context 'when there are pending jobs ready to run in the local queue' do
+        it 'emits the correct count' do
+          Delayed::Job.enqueue(
+            VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1),
+            queue: VCAP::CloudController::Jobs::Queues.local(VCAP::CloudController::Config.config),
+            run_at: Time.now
+          )
+          periodic_updater.update_job_queue_load
+
+          expected_pending_job_queue_load = {
+            'cc-api-0': 1
+          }
+          expected_total = 1
+
+          expect(statsd_updater).to have_received(:update_job_queue_load).with(expected_pending_job_queue_load, expected_total)
+          expect(prometheus_updater).to have_received(:update_job_queue_load).with(expected_pending_job_queue_load)
+        end
+      end
+
+      context 'when local queue does not have pending jobs' do
+        it 'emits the local queue load as 0 for discoverability' do
+          periodic_updater.update_job_queue_load
+
+          expected_pending_job_queue_load = {
+            'cc-api-0': 0
+          }
+          expected_total = 0
+
+          expect(statsd_updater).to have_received(:update_job_queue_load).with(expected_pending_job_queue_load, expected_total)
+          expect(prometheus_updater).to have_received(:update_job_queue_load).with(expected_pending_job_queue_load)
+        end
+      end
+
+      it 'includes the load of the delayed job queue and the total' do
+        Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), queue: 'cc_local', run_at: Time.now)
+        Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), queue: 'cc_local', run_at: Time.now)
+        Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), queue: 'cc_generic', run_at: Time.now)
+
+        periodic_updater.update_job_queue_load
+
+        expected_pending_job_queue_load = {
+          cc_local: 2,
+          cc_generic: 1,
+          'cc-api-0': 0
+        }
+        expected_total = 3
+
+        expect(statsd_updater).to have_received(:update_job_queue_load).with(expected_pending_job_queue_load, expected_total)
+        expect(prometheus_updater).to have_received(:update_job_queue_load).with(expected_pending_job_queue_load)
+      end
+
+      it 'does not contain failed jobs in job queue load metric' do
+        Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), queue: 'cc_local', failed_at: Time.now + 60, run_at: Time.now)
+        Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), queue: 'cc_local', run_at: Time.now)
+        Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), queue: 'cc_generic', failed_at: Time.now + 60, run_at: Time.now)
+        Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), queue: 'cc_generic', run_at: Time.now)
+
+        periodic_updater.update_job_queue_load
+
+        expected_pending_job_queue_load = {
+          cc_local: 1,
+          cc_generic: 1,
+          'cc-api-0': 0
+        }
+        expected_total = 2
+        expect(statsd_updater).to have_received(:update_job_queue_load).with(expected_pending_job_queue_load, expected_total)
+        expect(prometheus_updater).to have_received(:update_job_queue_load).with(expected_pending_job_queue_load)
+      end
+    end
+
     describe '#update_failed_job_count' do
       before do
         allow(statsd_updater).to receive(:update_failed_job_count)
@@ -644,6 +735,7 @@ module VCAP::CloudController::Metrics
       it 'calls all update methods' do
         expect(periodic_updater).to receive(:update_user_count).once
         expect(periodic_updater).to receive(:update_job_queue_length).once
+        expect(periodic_updater).to receive(:update_job_queue_load).once
         expect(periodic_updater).to receive(:update_thread_info).once
         expect(periodic_updater).to receive(:update_failed_job_count).once
         expect(periodic_updater).to receive(:update_vitals).once

spec/unit/lib/cloud_controller/metrics/prometheus_updater_spec.rb

@@ -78,6 +78,24 @@ module VCAP::CloudController::Metrics
       end
     end
 
+    describe '#update_job_queue_load' do
+      it 'records the load of the delayed job queues and total' do
+        expected_local_load   = 5
+        expected_generic_load = 6
+
+        pending_job_load_by_queue = {
+          cc_local: expected_local_load,
+          cc_generic: expected_generic_load
+        }
+
+        updater.update_job_queue_load(pending_job_load_by_queue)
+
+        metric = prom_client.get :cc_job_queues_load_total
+        expect(metric.get(labels: { queue: 'cc_local' })).to eq 5
+        expect(metric.get(labels: { queue: 'cc_generic' })).to eq 6
+      end
+    end
+
     describe '#update_failed_job_count' do
       it 'records the number of failed jobs in the delayed job queue and the total to statsd' do
         expected_local_length   = 5

spec/unit/lib/cloud_controller/metrics/statsd_updater_spec.rb

@@ -60,6 +60,32 @@ module VCAP::CloudController::Metrics
       end
     end
 
+    describe '#update_job_queue_load' do
+      let(:batch) { double(:batch) }
+
+      before do
+        allow(statsd_client).to receive(:batch).and_yield(batch)
+        allow(batch).to receive(:gauge)
+      end
+
+      it 'emits the load of the delayed job queues and total to statsd' do
+        expected_local_load   = 5
+        expected_generic_load = 6
+        total                   = expected_local_load + expected_generic_load
+
+        pending_job_load_by_queue = {
+          cc_local: expected_local_load,
+          cc_generic: expected_generic_load
+        }
+
+        updater.update_job_queue_load(pending_job_load_by_queue, total)
+
+        expect(batch).to have_received(:gauge).with('cc.job_queue_load.cc_local', expected_local_load)
+        expect(batch).to have_received(:gauge).with('cc.job_queue_load.cc_generic', expected_generic_load)
+        expect(batch).to have_received(:gauge).with('cc.job_queue_load.total', total)
+      end
+    end
+
     describe '#update_failed_job_count' do
       let(:batch) { double(:batch) }