Revert "Remove EventMachine from Puma"

This reverts commit 1187814.

Author: jochenehret

lib/cloud_controller/metrics/periodic_updater.rb

@@ -16,21 +16,23 @@ def initialize(start_time, log_counter, logger, statsd_updater, prometheus_updat
 
     def setup_updates
       update!
-      Concurrent::TimerTask.new(execution_interval: 600) { catch_error { update_user_count } }.execute
-      Concurrent::TimerTask.new(execution_interval: 30) { catch_error { update_job_queue_length } }.execute
-      Concurrent::TimerTask.new(execution_interval: 30) { catch_error { update_job_queue_load } }.execute
-      Concurrent::TimerTask.new(execution_interval: 30) { catch_error { update_failed_job_count } }.execute
-      Concurrent::TimerTask.new(execution_interval: 30) { catch_error { update_vitals } }.execute
-      Concurrent::TimerTask.new(execution_interval: 30) { catch_error { update_log_counts } }.execute
-      Concurrent::TimerTask.new(execution_interval: 30) { catch_error { update_task_stats } }.execute
-      Concurrent::TimerTask.new(execution_interval: 30) { catch_error { update_deploying_count } }.execute
-      Concurrent::TimerTask.new(execution_interval: 30) { catch_error { update_webserver_stats } }.execute
+      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 } }
+      EM.add_periodic_timer(30)  { catch_error { update_log_counts } }
+      EM.add_periodic_timer(30)  { catch_error { update_task_stats } }
+      EM.add_periodic_timer(30)  { catch_error { update_deploying_count } }
+      EM.add_periodic_timer(30)  { catch_error { update_webserver_stats } }
     end
 
     def update!
       update_user_count
       update_job_queue_length
       update_job_queue_load
+      update_thread_info
       update_failed_job_count
       update_vitals
       update_log_counts
@@ -108,6 +110,13 @@ def update_job_queue_load
       @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'
+
+      local_thread_info = thread_info_thin
+      [@statsd_updater, @prometheus_updater].each { |u| u.update_thread_info_thin(local_thread_info) }
+    end
+
     def update_failed_job_count
       jobs_by_queue_with_count = Delayed::Job.where(Sequel.lit('failed_at IS NOT NULL')).group_and_count(:queue)
 
@@ -164,5 +173,24 @@ def update_webserver_stats
       end
       @prometheus_updater.update_webserver_stats_puma(worker_count, worker_stats)
     end
+
+    def thread_info_thin
+      threadqueue = EM.instance_variable_get(:@threadqueue) || []
+      resultqueue = EM.instance_variable_get(:@resultqueue) || []
+      {
+        thread_count: Thread.list.size,
+        event_machine: {
+          connection_count: EventMachine.connection_count,
+          threadqueue: {
+            size: threadqueue.size,
+            num_waiting: threadqueue.is_a?(Array) ? 0 : threadqueue.num_waiting
+          },
+          resultqueue: {
+            size: resultqueue.size,
+            num_waiting: resultqueue.is_a?(Array) ? 0 : resultqueue.num_waiting
+          }
+        }
+      }
+    end
   end
 end

lib/cloud_controller/metrics/prometheus_updater.rb

@@ -41,6 +41,15 @@ def self.allow_pid_label
       { type: :gauge, name: :cc_deployments_in_progress_total, docstring: 'Number of in progress deployments', aggregation: :most_recent }
     ].freeze
 
+    THIN_METRICS = [
+      { type: :gauge, name: :cc_thread_info_thread_count, docstring: 'Thread count' },
+      { type: :gauge, name: :cc_thread_info_event_machine_connection_count, docstring: 'EventMachine connection count' },
+      { type: :gauge, name: :cc_thread_info_event_machine_threadqueue_size, docstring: 'EventMachine thread queue size' },
+      { type: :gauge, name: :cc_thread_info_event_machine_threadqueue_num_waiting, docstring: 'EventMachine num waiting in thread' },
+      { type: :gauge, name: :cc_thread_info_event_machine_resultqueue_size, docstring: 'EventMachine queue size' },
+      { type: :gauge, name: :cc_thread_info_event_machine_resultqueue_num_waiting, docstring: 'EventMachine requests waiting in queue' }
+    ].freeze
+
     PUMA_METRICS = [
       { type: :gauge, name: :cc_puma_worker_count, docstring: 'Puma worker count', aggregation: :most_recent },
       { type: :gauge, name: :cc_puma_worker_started_at, docstring: 'Puma worker: started_at', labels: %i[index pid], aggregation: :most_recent },
@@ -79,6 +88,7 @@ def initialize(registry: Prometheus::Client.registry, cc_worker: false)
       return if cc_worker
 
       METRICS.each { |metric| register(metric) }
+      THIN_METRICS.each { |metric| register(metric) } if VCAP::CloudController::Config.config&.get(:webserver) == 'thin'
       PUMA_METRICS.each { |metric| register(metric) } if VCAP::CloudController::Config.config&.get(:webserver) == 'puma'
     end
 
@@ -126,6 +136,15 @@ def update_job_queue_load(update_job_queue_load)
       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])
+      update_gauge_metric(:cc_thread_info_event_machine_threadqueue_size, thread_info[:event_machine][:threadqueue][:size])
+      update_gauge_metric(:cc_thread_info_event_machine_threadqueue_num_waiting, thread_info[:event_machine][:threadqueue][:num_waiting])
+      update_gauge_metric(:cc_thread_info_event_machine_resultqueue_size, thread_info[:event_machine][:resultqueue][:size])
+      update_gauge_metric(:cc_thread_info_event_machine_resultqueue_num_waiting, thread_info[:event_machine][:resultqueue][:num_waiting])
+    end
+
     def update_failed_job_count(failed_jobs_by_queue)
       failed_jobs_by_queue.each do |key, value|
         update_gauge_metric(:cc_failed_jobs_total, value, labels: { queue: key.to_s.underscore })

lib/cloud_controller/runners/puma_runner.rb

@@ -59,8 +59,14 @@ def initialize(config, app, logger, periodic_updater, request_logs)
       events = Puma::Events.new
       events.after_booted do
         prometheus_updater.update_gauge_metric(:cc_db_connection_pool_timeouts_total, 0, labels: { process_type: 'main' })
-        periodic_updater.setup_updates
+        Thread.new do
+          EM.run { periodic_updater.setup_updates }
+        end
+      end
+      events.after_stopped do
+        EM.stop
       end
+
       @puma_launcher = Puma::Launcher.new(puma_config, log_writer:, events:)
     end
 

spec/request/internal/metrics_spec.rb

@@ -119,6 +119,19 @@
     end
   end
 
+  context 'cc_thread_info' do
+    it 'reports thread info' do
+      get '/internal/v4/metrics', nil
+
+      expect(last_response.body).to match(/cc_thread_info_thread_count [0-9][0-9]*\.\d+/)
+      expect(last_response.body).to match(/cc_thread_info_event_machine_connection_count [0-9][0-9]*\.\d+/)
+      expect(last_response.body).to match(/cc_thread_info_event_machine_threadqueue_size [0-9][0-9]*\.\d+/)
+      expect(last_response.body).to match(/cc_thread_info_event_machine_threadqueue_num_waiting [0-9][0-9]*\.\d+/)
+      expect(last_response.body).to match(/cc_thread_info_event_machine_resultqueue_size [0-9][0-9]*\.\d+/)
+      expect(last_response.body).to match(/cc_thread_info_event_machine_resultqueue_num_waiting [0-9][0-9]*\.\d+/)
+    end
+  end
+
   context 'cc_failed_job_count' do
     before do
       Delayed::Job.enqueue(VCAP::CloudController::Jobs::Runtime::EventsCleanup.new(1), { queue: 'cc_api_0', run_at: Time.now + 1.day })

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

@@ -6,10 +6,27 @@ module VCAP::CloudController::Metrics
     let(:periodic_updater) { PeriodicUpdater.new(start_time, log_counter, logger, statsd_updater, prometheus_updater) }
     let(:statsd_updater) { double(:statsd_updater) }
     let(:prometheus_updater) { double(:prometheus_updater) }
+    let(:threadqueue) { double(EventMachine::Queue, size: 20, num_waiting: 0) }
+    let(:resultqueue) { double(EventMachine::Queue, size: 0, num_waiting: 1) }
     let(:start_time) { Time.now.utc - 90 }
     let(:log_counter) { double(:log_counter, counts: {}) }
     let(:logger) { double(:logger) }
 
+    before do
+      allow(EventMachine).to receive(:connection_count).and_return(123)
+
+      allow(EventMachine).to receive(:instance_variable_get) do |instance_var|
+        case instance_var
+        when :@threadqueue
+          threadqueue
+        when :@resultqueue
+          resultqueue
+        else
+          raise "Unexpected call: #{instance_var}"
+        end
+      end
+    end
+
     describe 'task stats' do
       before do
         allow(statsd_updater).to receive(:update_task_stats)
@@ -56,6 +73,7 @@ module VCAP::CloudController::Metrics
         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)
         allow(statsd_updater).to receive(:update_log_counts)
@@ -65,11 +83,14 @@ 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)
         allow(prometheus_updater).to receive(:update_log_counts)
         allow(prometheus_updater).to receive(:update_task_stats)
         allow(prometheus_updater).to receive(:update_deploying_count)
+
+        allow(EventMachine).to receive(:add_periodic_timer)
       end
 
       it 'bumps the number of users and sets periodic timer' do
@@ -92,6 +113,11 @@ module VCAP::CloudController::Metrics
         periodic_updater.setup_updates
       end
 
+      it 'updates thread count and event machine queues' do
+        expect(periodic_updater).to receive(:update_thread_info).once
+        periodic_updater.setup_updates
+      end
+
       it 'updates the vitals' do
         expect(periodic_updater).to receive(:update_vitals).once
         periodic_updater.setup_updates
@@ -112,16 +138,15 @@ module VCAP::CloudController::Metrics
         periodic_updater.setup_updates
       end
 
-      context 'when Concurrent::TimerTasks are run' do
+      context 'when EventMachine periodic_timer tasks are run' do
         before do
           @periodic_timers = []
 
-          allow(Concurrent::TimerTask).to receive(:new) do |opts, &block|
+          allow(EventMachine).to receive(:add_periodic_timer) do |interval, &block|
             @periodic_timers << {
-              interval: opts[:execution_interval],
-              block: block
+              interval:,
+              block:
             }
-            double('TimerTask', execute: nil, shutdown: nil, kill: nil, running?: false)
           end
 
           periodic_updater.setup_updates
@@ -151,37 +176,45 @@ module VCAP::CloudController::Metrics
           @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
 
@@ -505,6 +538,75 @@ module VCAP::CloudController::Metrics
       end
     end
 
+    describe '#update_thread_info' do
+      before do
+        allow(statsd_updater).to receive(:update_thread_info_thin)
+        allow(prometheus_updater).to receive(:update_thread_info_thin)
+      end
+
+      it 'contains EventMachine data and send it to all updaters' do
+        expected_thread_info = {
+          thread_count: Thread.list.size,
+          event_machine: {
+            connection_count: 123,
+            threadqueue: {
+              size: 20,
+              num_waiting: 0
+            },
+            resultqueue: {
+              size: 0,
+              num_waiting: 1
+            }
+          }
+        }
+
+        periodic_updater.update_thread_info
+
+        expect(statsd_updater).to have_received(:update_thread_info_thin).with(expected_thread_info)
+        expect(prometheus_updater).to have_received(:update_thread_info_thin).with(expected_thread_info)
+      end
+
+      context 'when resultqueue and/or threadqueue is not a queue' do
+        let(:resultqueue) { [] }
+        let(:threadqueue) { nil }
+
+        it 'does not blow up' do
+          expected_thread_info = {
+            thread_count: Thread.list.size,
+            event_machine: {
+              connection_count: 123,
+              threadqueue: {
+                size: 0,
+                num_waiting: 0
+              },
+              resultqueue: {
+                size: 0,
+                num_waiting: 0
+              }
+            }
+          }
+
+          periodic_updater.update_thread_info
+
+          expect(statsd_updater).to have_received(:update_thread_info_thin).with(expected_thread_info)
+          expect(prometheus_updater).to have_received(:update_thread_info_thin).with(expected_thread_info)
+        end
+      end
+
+      context 'when Puma is configured as webserver' do
+        before do
+          TestConfig.override(webserver: 'puma')
+        end
+
+        it 'does not send EventMachine data to updaters' do
+          periodic_updater.update_thread_info
+
+          expect(statsd_updater).not_to have_received(:update_thread_info_thin)
+          expect(prometheus_updater).not_to have_received(:update_thread_info_thin)
+        end
+      end
+    end
+
     describe '#update_vitals' do
       before do
         allow(statsd_updater).to receive(:update_vitals)
@@ -634,6 +736,7 @@ module VCAP::CloudController::Metrics
         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
         expect(periodic_updater).to receive(:update_log_counts).once