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.

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,
 

lib/cloud_controller/db.rb

@@ -72,7 +72,7 @@ def self.add_connection_expiration_extension(db, opts)
 
     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)
+      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/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,17 +55,36 @@ 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'
     end
 
+    def registry
+      @registry
+    end
+
     def update_gauge_metric(metric, value, labels: {})
       @registry.get(metric).set(value, labels:)
     end

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_endpoint if @publish_metrics
     BackgroundJobEnvironment.new(config).setup_environment(readiness_port)
 
     logger = Steno.logger('cc-worker')
@@ -92,4 +98,34 @@ def readiness_port
   def is_first_generic_worker_on_machine?
     RakeConfig.context != :api && ENV['INDEX']&.to_i == 1
   end
+
+  def setup_metrics_endpoint
+    prometheus_dir = File.join(RakeConfig.config.get(:directories, :tmpdir), 'prometheus')
+    Prometheus::Client.config.data_store = Prometheus::Client::DataStores::DirectFileStore.new(dir: prometheus_dir)
+    return unless is_first_generic_worker_on_machine?
+
+    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 '0.0.0.0', 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,12 @@ 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'
+
+    CloudController::DelayedWorker.new(queues: queues,
+                                       name: args.name,
+                                       num_threads: args.num_threads,
+                                       thread_grace_period_seconds: args.thread_grace_period_seconds,
+                                       publish_metrics: true).start_working
   end
 end