Remove EventMachine from Puma (#4661)

* preparation for Thin removal
* replace EventMachine with Concurrent::TimerTask
* remove "thread_info_event_machine" metrics

Author: jochenehret

lib/cloud_controller/metrics/periodic_updater.rb

@@ -16,23 +16,21 @@ def initialize(start_time, log_counter, logger, statsd_updater, prometheus_updat
 
     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 } }
-      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 } }
+      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
     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
@@ -110,13 +108,6 @@ 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)
 
@@ -173,24 +164,5 @@ 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,15 +41,6 @@ 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 },
@@ -88,7 +79,6 @@ 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
 
@@ -136,15 +126,6 @@ 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,14 +59,8 @@ 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' })
-        Thread.new do
-          EM.run { periodic_updater.setup_updates }
-        end
-      end
-      events.after_stopped do
-        EM.stop
+        periodic_updater.setup_updates
       end
-
       @puma_launcher = Puma::Launcher.new(puma_config, log_writer:, events:)
     end
 

spec/request/internal/metrics_spec.rb

@@ -119,19 +119,6 @@
     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,27 +6,10 @@ 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)
@@ -73,7 +56,6 @@ 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)
@@ -83,14 +65,11 @@ 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
@@ -113,11 +92,6 @@ 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
@@ -138,15 +112,16 @@ module VCAP::CloudController::Metrics
         periodic_updater.setup_updates
       end
 
-      context 'when EventMachine periodic_timer tasks are run' do
+      context 'when Concurrent::TimerTasks are run' do
         before do
           @periodic_timers = []
 
-          allow(EventMachine).to receive(:add_periodic_timer) do |interval, &block|
+          allow(Concurrent::TimerTask).to receive(:new) do |opts, &block|
             @periodic_timers << {
-              interval:,
-              block:
+              interval: opts[:execution_interval],
+              block: block
             }
+            double('TimerTask', execute: nil, shutdown: nil, kill: nil, running?: false)
           end
 
           periodic_updater.setup_updates
@@ -176,44 +151,36 @@ module VCAP::CloudController::Metrics
           @periodic_timers[2][:block].call
         end
 
-        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_thread_info).once
-          expect(@periodic_timers[3][:interval]).to eq(30)
-
-          @periodic_timers[3][:block].call
-        end
-
         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_failed_job_count).once
-          expect(@periodic_timers[4][:interval]).to eq(30)
+          expect(@periodic_timers[3][:interval]).to eq(30)
 
-          @periodic_timers[4][:block].call
+          @periodic_timers[3][:block].call
         end
 
         it 'updates the vitals' do
           expect(periodic_updater).to receive(:catch_error).once.and_call_original
           expect(periodic_updater).to receive(:update_vitals).once
-          expect(@periodic_timers[5][:interval]).to eq(30)
+          expect(@periodic_timers[4][:interval]).to eq(30)
 
-          @periodic_timers[5][:block].call
+          @periodic_timers[4][:block].call
         end
 
         it 'updates the log counts' do
           expect(periodic_updater).to receive(:catch_error).once.and_call_original
           expect(periodic_updater).to receive(:update_log_counts).once
-          expect(@periodic_timers[6][:interval]).to eq(30)
+          expect(@periodic_timers[5][:interval]).to eq(30)
 
-          @periodic_timers[6][:block].call
+          @periodic_timers[5][: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)
+          expect(@periodic_timers[6][:interval]).to eq(30)
 
-          @periodic_timers[7][:block].call
+          @periodic_timers[6][:block].call
         end
       end
     end
@@ -538,75 +505,6 @@ 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)
@@ -736,7 +634,6 @@ 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