Replace statsd-client with statsd-instrument (#5278)

Author: johha

Gemfile

@@ -29,7 +29,7 @@ gem 'sequel', '~> 5.106'
 gem 'sequel_pg', require: 'sequel'
 gem 'sinatra', '~> 4.2'
 gem 'sinatra-contrib'
-gem 'statsd-ruby', '~> 1.5.0'
+gem 'statsd-instrument', '~> 3.11'
 gem 'talentbox-delayed_job_sequel', '~> 4.4.0'
 gem 'uri', '~> 1.1'
 gem 'vmstat', '~> 2.3'

Gemfile.lock

@@ -431,7 +431,7 @@ GEM
     spring (4.7.0)
     spring-commands-rspec (1.0.4)
       spring (>= 0.9.1)
-    statsd-ruby (1.5.0)
+    statsd-instrument (3.11.1)
     stringio (3.2.0)
     talentbox-delayed_job_sequel (4.4.0)
       delayed_job (~> 4.1)
@@ -532,7 +532,7 @@ DEPENDENCIES
   solargraph
   spring
   spring-commands-rspec
-  statsd-ruby (~> 1.5.0)
+  statsd-instrument (~> 3.11)
   talentbox-delayed_job_sequel (~> 4.4.0)
   timecop
   uri (~> 1.1)

app/jobs/diego/sync.rb

@@ -1,6 +1,6 @@
 require 'cloud_controller/diego/processes_sync'
 require 'cloud_controller/diego/tasks_sync'
-require 'statsd'
+require 'statsd/instrument'
 
 module VCAP::CloudController
   module Jobs
@@ -25,7 +25,7 @@ def perform
             ##       so feed in the entire value!
             elapsed_ms = ((finish - start) * 1000).round
 
-            @statsd.timing('cc.diego_sync.duration', elapsed_ms)
+            @statsd.measure('cc.diego_sync.duration', elapsed_ms)
           end
         end
 

lib/cloud_controller/dependency_locator.rb

@@ -24,6 +24,7 @@
 require 'cloud_controller/packager/local_bits_packer'
 require 'credhub/client'
 require 'cloud_controller/metrics/prometheus_updater'
+require 'statsd/instrument'
 require 'cloud_controller/execution_context'
 
 module CloudController
@@ -367,8 +368,7 @@ def statsd_client
       if @dependencies[:statsd_client]
         @dependencies[:statsd_client]
       elsif config.get(:enable_statsd_metrics) == true || config.get(:enable_statsd_metrics).nil?
-        Statsd.logger = Steno.logger('statsd.client')
-        register(:statsd_client, Statsd.new(config.get(:statsd_host), config.get(:statsd_port)))
+        register(:statsd_client, StatsD::Instrument::Client.new(sink: statsd_sink))
       else
         register(:statsd_client, NullStatsdClient.new)
       end
@@ -378,6 +378,14 @@ def statsd_client
 
     private
 
+    # Memoized per address so one long-lived UDP socket serves the process; statsd-instrument
+    # pins its socket in a thread-local store and never reclaims it, so a fresh sink per call would leak one.
+    def statsd_sink
+      addr = "#{config.get(:statsd_host)}:#{config.get(:statsd_port)}"
+      @statsd_sinks ||= {}
+      @statsd_sinks[addr] ||= StatsD::Instrument::Sink.for_addr(addr)
+    end
+
     def build_stager_client
       build_bbs_stager_client
     end
@@ -464,19 +472,15 @@ def create_paginated_collection_renderer(opts={})
   end
 
   class NullStatsdClient
-    def timing(_key, _value)
-      # Null implementation
-    end
-
-    def increment(_key)
+    def gauge(_stat, _value, *)
       # Null implementation
     end
 
-    def gauge(_stat, _value, _sample_rate=1)
+    def increment(_stat, _value=1, *)
       # Null implementation
     end
 
-    def batch
+    def measure(_stat, _value=nil, *)
       # Null implementation
     end
   end

lib/cloud_controller/deployment_updater/scheduler.rb

@@ -57,7 +57,7 @@ def update(update_frequency:, statsd_client:)
           ##       milliseconds, then do know that the lack of precision here is not desired
           ##       so feed in the entire value!
           update_duration_ms = update_duration * 1000
-          statsd_client.timing('cc.deployments.update.duration', update_duration_ms)
+          statsd_client.measure('cc.deployments.update.duration', update_duration_ms)
 
           logger.info("Update loop took #{update_duration}s")
 

lib/cloud_controller/metrics/request_metrics.rb

@@ -1,4 +1,4 @@
-require 'statsd'
+require 'statsd/instrument'
 
 module VCAP::CloudController
   module Metrics

lib/cloud_controller/metrics/statsd_updater.rb

@@ -1,4 +1,4 @@
-require 'statsd'
+require 'statsd/instrument'
 
 module VCAP::CloudController::Metrics
   class StatsdUpdater
@@ -16,64 +16,50 @@ def update_user_count(user_count)
     end
 
     def update_job_queue_length(pending_job_count_by_queue, total)
-      @statsd.batch do |batch|
-        pending_job_count_by_queue.each do |key, value|
-          batch.gauge("cc.job_queue_length.#{key}", value)
-        end
-        batch.gauge('cc.job_queue_length.total', total)
+      pending_job_count_by_queue.each do |key, value|
+        @statsd.gauge("cc.job_queue_length.#{key}", value)
       end
+      @statsd.gauge('cc.job_queue_length.total', total)
     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)
+      pending_job_load_by_queue.each do |key, value|
+        @statsd.gauge("cc.job_queue_load.#{key}", value)
       end
+      @statsd.gauge('cc.job_queue_load.total', total)
     end
 
     def update_thread_info_thin(thread_info)
-      @statsd.batch do |batch|
-        batch.gauge('cc.thread_info.thread_count', thread_info[:thread_count])
-        batch.gauge('cc.thread_info.event_machine.connection_count', thread_info[:event_machine][:connection_count])
-        batch.gauge('cc.thread_info.event_machine.threadqueue.size', thread_info[:event_machine][:threadqueue][:size])
-        batch.gauge('cc.thread_info.event_machine.threadqueue.num_waiting', thread_info[:event_machine][:threadqueue][:num_waiting])
-        batch.gauge('cc.thread_info.event_machine.resultqueue.size', thread_info[:event_machine][:resultqueue][:size])
-        batch.gauge('cc.thread_info.event_machine.resultqueue.num_waiting', thread_info[:event_machine][:resultqueue][:num_waiting])
-      end
+      @statsd.gauge('cc.thread_info.thread_count', thread_info[:thread_count])
+      @statsd.gauge('cc.thread_info.event_machine.connection_count', thread_info[:event_machine][:connection_count])
+      @statsd.gauge('cc.thread_info.event_machine.threadqueue.size', thread_info[:event_machine][:threadqueue][:size])
+      @statsd.gauge('cc.thread_info.event_machine.threadqueue.num_waiting', thread_info[:event_machine][:threadqueue][:num_waiting])
+      @statsd.gauge('cc.thread_info.event_machine.resultqueue.size', thread_info[:event_machine][:resultqueue][:size])
+      @statsd.gauge('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, total)
-      @statsd.batch do |batch|
-        failed_jobs_by_queue.each do |key, value|
-          batch.gauge("cc.failed_job_count.#{key}", value)
-        end
-        batch.gauge('cc.failed_job_count.total', total)
+      failed_jobs_by_queue.each do |key, value|
+        @statsd.gauge("cc.failed_job_count.#{key}", value)
       end
+      @statsd.gauge('cc.failed_job_count.total', total)
     end
 
     def update_vitals(vitals)
-      @statsd.batch do |batch|
-        vitals.each do |key, val|
-          batch.gauge("cc.vitals.#{key}", val)
-        end
+      vitals.each do |key, val|
+        @statsd.gauge("cc.vitals.#{key}", val)
       end
     end
 
     def update_log_counts(counts)
-      @statsd.batch do |batch|
-        counts.each do |key, val|
-          batch.gauge("cc.log_count.#{key}", val)
-        end
+      counts.each do |key, val|
+        @statsd.gauge("cc.log_count.#{key}", val)
       end
     end
 
     def update_task_stats(total_running_tasks, total_memory_in_mb)
-      @statsd.batch do |batch|
-        batch.gauge('cc.tasks_running.count', total_running_tasks)
-        batch.gauge('cc.tasks_running.memory_in_mb', total_memory_in_mb)
-      end
+      @statsd.gauge('cc.tasks_running.count', total_running_tasks)
+      @statsd.gauge('cc.tasks_running.memory_in_mb', total_memory_in_mb)
     end
 
     def update_synced_invalid_lrps(lrp_count)
@@ -86,12 +72,12 @@ def start_staging_request_received
 
     def report_staging_success_metrics(duration_ns)
       @statsd.increment('cc.staging.succeeded')
-      @statsd.timing('cc.staging.succeeded_duration', nanoseconds_to_milliseconds(duration_ns))
+      @statsd.measure('cc.staging.succeeded_duration', nanoseconds_to_milliseconds(duration_ns))
     end
 
     def report_staging_failure_metrics(duration_ns)
       @statsd.increment('cc.staging.failed')
-      @statsd.timing('cc.staging.failed_duration', nanoseconds_to_milliseconds(duration_ns))
+      @statsd.measure('cc.staging.failed_duration', nanoseconds_to_milliseconds(duration_ns))
     end
 
     def start_request
@@ -103,11 +89,9 @@ def complete_request(status)
       http_status_code = "#{status.to_s[0]}XX"
       http_status_metric = "cc.http_status.#{http_status_code}"
       @statsd.gauge('cc.requests.outstanding.gauge', store.decrement_request_outstanding_gauge)
-      @statsd.batch do |batch|
-        batch.decrement 'cc.requests.outstanding'
-        batch.increment 'cc.requests.completed'
-        batch.increment http_status_metric
-      end
+      @statsd.increment('cc.requests.outstanding', -1)
+      @statsd.increment('cc.requests.completed')
+      @statsd.increment(http_status_metric)
     end
 
     private

spec/unit/controllers/base/base_controller_spec.rb

@@ -9,7 +9,7 @@ module VCAP::CloudController
     let(:config) { double(Config, get: nil) }
     let(:dependencies) do
       {
-        statsd_client: double(Statsd)
+        statsd_client: double(StatsD::Instrument::Client)
       }
     end
 

spec/unit/controllers/base/model_controller_spec.rb

@@ -8,7 +8,7 @@ module VCAP::CloudController
       {
         object_renderer: nil,
         collection_renderer: nil,
-        statsd_client: double(Statsd)
+        statsd_client: double(StatsD::Instrument::Client)
       }
     end
     let(:config) { double(Config, get: nil) }

spec/unit/controllers/internal/log_access_controller_spec.rb

@@ -20,7 +20,7 @@ module VCAP::CloudController
         {},
         nil,
         {
-          statsd_client: double(Statsd)
+          statsd_client: double(StatsD::Instrument::Client)
         }
       )
     end