Add prometheus metrics to cc workers (#4205)

* Publish cc-worker metrics

Implemented a small puma webserver, which will be started in a separate thread with the first worker process.
Using the exporter middleware provided by the prometheus client, the metrics stored in the registry will be published under `/metrics`.
All processes use the same DirectFileStore registry, so that metrics can be published in a single webserver.

* Turn off publishing metrics by default and make configurable

* Use worker process index as pid for metrics store

* Fix code style and unit tests

* Make metrics endpoint only available on localhost

Author: svkrieger

lib/cloud_controller/config_schemas/base/worker_schema.rb

@@ -37,6 +37,10 @@ class WorkerSchema < VCAP::Config
 
             log_audit_events: bool,
 
+            directories: {
+              tmpdir: String
+            },
+
             stacks_file: String,
             newrelic_enabled: bool,
 
@@ -139,6 +143,9 @@ class WorkerSchema < VCAP::Config
               optional(:port) => Integer
             },
 
+            optional(:publish_metrics) => bool,
+            optional(:prometheus_port) => Integer,
+
             skip_cert_verify: bool,
 
             optional(:routing_api) => {

lib/cloud_controller/db.rb

@@ -71,8 +71,8 @@ def self.add_connection_expiration_extension(db, opts)
     end
 
     def self.add_connection_metrics_extension(db)
-      # only add the metrics for api processes. Otherwise e.g. rake db:migrate would also initialize metric updaters, which need additional config
-      return if Object.const_defined?(:RakeConfig)
+      # only add the metrics for api and cc-worker processes. Otherwise e.g. rake db:migrate would also initialize metric updaters, which need additional config
+      return if Object.const_defined?(:RakeConfig) && RakeConfig.context != :worker
 
       db.extension(:connection_metrics)
       # so that we gather connection metrics from the beginning

lib/cloud_controller/dependency_locator.rb

@@ -74,6 +74,10 @@ def prometheus_updater
       @dependencies[:prometheus_updater] || register(:prometheus_updater, VCAP::CloudController::Metrics::PrometheusUpdater.new)
     end
 
+    def cc_worker_prometheus_updater
+      @dependencies[:cc_worker_prometheus_updater] || register(:cc_worker_prometheus_updater, VCAP::CloudController::Metrics::PrometheusUpdater.new(cc_worker: true))
+    end
+
     def statsd_updater
       @dependencies[:statsd_updater] || register(:statsd_updater, VCAP::CloudController::Metrics::StatsdUpdater.new(statsd_client))
     end

lib/cloud_controller/metrics/custom_process_id.rb

@@ -0,0 +1,23 @@
+# Storing the metrics of several worker processes on cc-worker VMs in a DirectFileStore residing in a single directory
+# did not work because the different processes are isolated by bpm and several processes used the same pid within their container.
+# This pid is used for the filename and resulted in corrupted data because several processes were writing data to the same files.
+# When requiring this file, the process_id method of the MetricStore will be overridden to first check for `INDEX` in
+# env variables before returning the actual pid. The `INDEX` is provided for cc-worker processes.
+
+module CustomProcessId
+  def process_id
+    ENV.fetch('INDEX', Process.pid).to_i
+  end
+end
+
+module Prometheus
+  module Client
+    module DataStores
+      class DirectFileStore
+        class MetricStore
+          prepend CustomProcessId
+        end
+      end
+    end
+  end
+end

lib/cloud_controller/metrics/prometheus_updater.rb

@@ -36,15 +36,7 @@ def self.allow_pid_label
       { type: :gauge, name: :cc_running_tasks_total, docstring: 'Total running tasks', aggregation: :most_recent },
       { type: :gauge, name: :cc_running_tasks_memory_bytes, docstring: 'Total memory consumed by running tasks', aggregation: :most_recent },
       { type: :gauge, name: :cc_users_total, docstring: 'Number of users', aggregation: :most_recent },
-      { type: :gauge, name: :cc_deployments_in_progress_total, docstring: 'Number of in progress deployments', aggregation: :most_recent },
-      { type: :gauge, name: :cc_acquired_db_connections_total, labels: %i[process_type], docstring: 'Number of acquired DB connections' },
-      { type: :histogram, name: :cc_db_connection_hold_duration_seconds, docstring: 'The time threads were holding DB connections', buckets: CONNECTION_DURATION_BUCKETS },
-      # cc_connection_pool_timeouts_total must be a gauge metric, because otherwise we cannot match them with processes
-      { type: :gauge, name: :cc_db_connection_pool_timeouts_total, labels: %i[process_type],
-        docstring: 'Number of threads which failed to acquire a free DB connection from the pool within the timeout' },
-      { type: :gauge, name: :cc_open_db_connections_total, labels: %i[process_type], docstring: 'Number of open DB connections (acquired + available)' },
-      { type: :histogram, name: :cc_db_connection_wait_duration_seconds, docstring: 'The time threads were waiting for an available DB connection',
-        buckets: CONNECTION_DURATION_BUCKETS }
+      { type: :gauge, name: :cc_deployments_in_progress_total, docstring: 'Number of in progress deployments', aggregation: :most_recent }
     ].freeze
 
     THIN_METRICS = [
@@ -63,12 +55,27 @@ def self.allow_pid_label
       { type: :gauge, name: :cc_puma_worker_backlog, docstring: 'Puma worker: backlog', labels: %i[index pid], aggregation: :most_recent }
     ].freeze
 
-    def initialize(registry=Prometheus::Client.registry)
+    DB_CONNECTION_POOL_METRICS = [
+      { type: :gauge, name: :cc_acquired_db_connections_total, labels: %i[process_type], docstring: 'Number of acquired DB connections' },
+      { type: :histogram, name: :cc_db_connection_hold_duration_seconds, docstring: 'The time threads were holding DB connections', buckets: CONNECTION_DURATION_BUCKETS },
+      # cc_connection_pool_timeouts_total must be a gauge metric, because otherwise we cannot match them with processes
+      { type: :gauge, name: :cc_db_connection_pool_timeouts_total, labels: %i[process_type],
+        docstring: 'Number of threads which failed to acquire a free DB connection from the pool within the timeout' },
+      { type: :gauge, name: :cc_open_db_connections_total, labels: %i[process_type], docstring: 'Number of open DB connections (acquired + available)' },
+      { type: :histogram, name: :cc_db_connection_wait_duration_seconds, docstring: 'The time threads were waiting for an available DB connection',
+        buckets: CONNECTION_DURATION_BUCKETS }
+    ].freeze
+
+    def initialize(registry: Prometheus::Client.registry, cc_worker: false)
       self.class.allow_pid_label
 
       @registry = registry
 
       # Register all metrics, to initialize them for discoverability
+      DB_CONNECTION_POOL_METRICS.each { |metric| register(metric) }
+
+      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'

lib/delayed_job/delayed_worker.rb

@@ -1,4 +1,7 @@
 require 'delayed_job/threaded_worker'
+require 'rack'
+require 'puma'
+require 'prometheus/middleware/exporter'
 
 class CloudController::DelayedWorker
   def initialize(options)
@@ -9,6 +12,8 @@ def initialize(options)
       worker_name: options[:name],
       quiet: true
     }
+
+    @publish_metrics = options.fetch(:publish_metrics, false)
     return unless options[:num_threads] && options[:num_threads].to_i > 0
 
     @queue_options[:num_threads] = options[:num_threads].to_i
@@ -17,6 +22,7 @@ def initialize(options)
 
   def start_working
     config = RakeConfig.config
+    setup_metrics(config) if @publish_metrics
     BackgroundJobEnvironment.new(config).setup_environment(readiness_port)
 
     logger = Steno.logger('cc-worker')
@@ -92,4 +98,40 @@ def readiness_port
   def is_first_generic_worker_on_machine?
     RakeConfig.context != :api && ENV['INDEX']&.to_i == 1
   end
+
+  def setup_metrics(config)
+    prometheus_dir = File.join(config.get(:directories, :tmpdir), 'prometheus')
+    Prometheus::Client.config.data_store = Prometheus::Client::DataStores::DirectFileStore.new(dir: prometheus_dir)
+
+    setup_webserver(config, prometheus_dir) if is_first_generic_worker_on_machine?
+
+    # initialize metric with 0 for discoverability, because it likely won't get updated on healthy systems
+    CloudController::DependencyLocator.instance.cc_worker_prometheus_updater.update_gauge_metric(:cc_db_connection_pool_timeouts_total, 0, labels: { process_type: 'cc-worker' })
+  end
+
+  def setup_webserver(config, prometheus_dir)
+    FileUtils.mkdir_p(prometheus_dir)
+
+    # Resetting metrics on startup
+    Dir["#{prometheus_dir}/*.bin"].each do |file_path|
+      File.unlink(file_path)
+    end
+
+    metrics_app = Rack::Builder.new do
+      use Prometheus::Middleware::Exporter, path: '/metrics'
+
+      map '/' do
+        run lambda { |_env|
+          # Return 404 for any other request
+          ['404', { 'Content-Type' => 'text/plain' }, ['Not Found']]
+        }
+      end
+    end
+
+    Thread.new do
+      server = Puma::Server.new(metrics_app)
+      server.add_tcp_listener '127.0.0.1', config.get(:prometheus_port) || 9394
+      server.run
+    end
+  end
 end

lib/sequel/extensions/connection_metrics.rb

@@ -23,7 +23,11 @@ def self.extended(pool)
 
       pool.instance_exec do
         sync do
-          @prometheus_updater = CloudController::DependencyLocator.instance.prometheus_updater
+          @prometheus_updater = if process_type == 'cc-worker'
+                                  CloudController::DependencyLocator.instance.cc_worker_prometheus_updater
+                                else
+                                  CloudController::DependencyLocator.instance.prometheus_updater
+                                end
           @connection_info = {}
         end
       end

lib/tasks/jobs.rake

@@ -64,6 +64,15 @@ namespace :jobs do
       'prune_excess_app_revisions'
     ]
 
-    CloudController::DelayedWorker.new(queues: queues, name: args.name, num_threads: args.num_threads, thread_grace_period_seconds: args.thread_grace_period_seconds).start_working
+    ENV['PROCESS_TYPE'] = 'cc-worker'
+    require 'cloud_controller/metrics/custom_process_id'
+
+    publish_metrics = RakeConfig.config.get(:publish_metrics) || false
+
+    CloudController::DelayedWorker.new(queues: queues,
+                                       name: args.name,
+                                       num_threads: args.num_threads,
+                                       thread_grace_period_seconds: args.thread_grace_period_seconds,
+                                       publish_metrics: publish_metrics).start_working
   end
 end

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

@@ -3,7 +3,7 @@
 
 module VCAP::CloudController::Metrics
   RSpec.describe PrometheusUpdater do
-    let(:updater) { PrometheusUpdater.new(prom_client) }
+    let(:updater) { PrometheusUpdater.new(registry: prom_client) }
     let(:tmpdir) { Dir.mktmpdir }
     let(:prom_client) do
       Prometheus::Client.config.data_store = Prometheus::Client::DataStores::DirectFileStore.new(dir: tmpdir)

spec/unit/lib/delayed_job/delayed_worker_spec.rb

@@ -64,6 +64,26 @@
       worker_instance = CloudController::DelayedWorker.new(options)
       expect(worker_instance.instance_variable_get(:@queue_options)).not_to include(:grace_period_seconds)
     end
+
+    describe 'publish metrics' do
+      context 'when not set' do
+        it 'does not publish metrics' do
+          worker_instance = CloudController::DelayedWorker.new(options)
+          expect(worker_instance.instance_variable_get(:@publish_metrics)).to be(false)
+        end
+      end
+
+      context 'when set to true' do
+        before do
+          options[:publish_metrics] = true
+        end
+
+        it 'publishes metrics' do
+          worker_instance = CloudController::DelayedWorker.new(options)
+          expect(worker_instance.instance_variable_get(:@publish_metrics)).to be(true)
+        end
+      end
+    end
   end
 
   describe '#start_working' do
@@ -120,6 +140,47 @@
         cc_delayed_worker.start_working
       end
     end
+
+    describe 'publish metrics' do
+      before do
+        allow(Prometheus::Client::DataStores::DirectFileStore).to receive(:new)
+      end
+
+      context 'when set to false' do
+        before do
+          options[:publish_metrics] = false
+        end
+
+        it 'does not publish metrics' do
+          cc_delayed_worker.start_working
+          expect(Prometheus::Client::DataStores::DirectFileStore).not_to have_received(:new)
+        end
+      end
+
+      context 'when set to true' do
+        before do
+          options[:publish_metrics] = true
+        end
+
+        it 'publishes metrics' do
+          cc_delayed_worker.start_working
+          expect(Prometheus::Client::DataStores::DirectFileStore).to have_received(:new)
+        end
+
+        context 'when first worker on machine' do
+          before do
+            allow(cc_delayed_worker).to receive(:is_first_generic_worker_on_machine?).and_return(true)
+            allow(cc_delayed_worker).to receive(:readiness_port)
+            allow(cc_delayed_worker).to receive(:setup_webserver)
+          end
+
+          it 'sets up a webserver' do
+            cc_delayed_worker.start_working
+            expect(cc_delayed_worker).to have_received(:setup_webserver)
+          end
+        end
+      end
+    end
   end
 
   describe '#clear_locks!' do