Replace statsd-client with statsd-instrument

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

spec/unit/controllers/runtime/legacy_api_base_spec.rb

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

spec/unit/controllers/services/service_keys_controller_spec.rb

@@ -52,7 +52,7 @@ def bind_url_regex(opts={})
         {
           object_renderer: object_renderer,
           collection_renderer: collection_renderer,
-          statsd_client: double(Statsd)
+          statsd_client: double(StatsD::Instrument::Client)
         }
       end
       let(:config) { double(Config, get: nil) }

spec/unit/jobs/diego/sync_spec.rb

@@ -3,7 +3,7 @@
 module VCAP::CloudController
   module Jobs::Diego
     RSpec.describe Sync, job_context: :clock do
-      let(:statsd_client) { instance_double(Statsd) }
+      let(:statsd_client) { instance_double(StatsD::Instrument::Client) }
       let(:processes_sync) { instance_double(Diego::ProcessesSync) }
       let(:tasks_sync) { instance_double(Diego::ProcessesSync) }
 
@@ -15,7 +15,7 @@ module Jobs::Diego
           allow(Diego::ProcessesSync).to receive(:new).and_return(processes_sync)
           allow(Diego::TasksSync).to receive(:new).and_return(tasks_sync)
 
-          allow(statsd_client).to receive(:timing)
+          allow(statsd_client).to receive(:measure)
           allow(processes_sync).to receive(:sync)
           allow(tasks_sync).to receive(:sync)
         end
@@ -38,7 +38,7 @@ module Jobs::Diego
           expect(processes_sync).to receive(:sync)
           expect(tasks_sync).to receive(:sync)
           expect(Time).to receive(:now).twice # Ensure that we get two time measurements. _Hopefully_ they get turned into an elapsed time and passed in where they need to be!
-          expect(statsd_client).to receive(:timing).with('cc.diego_sync.duration', kind_of(Numeric))
+          expect(statsd_client).to receive(:measure).with('cc.diego_sync.duration', kind_of(Numeric))
 
           job.perform
         end

spec/unit/lib/cloud_controller/clock/scheduler_spec.rb

@@ -179,7 +179,7 @@ module VCAP::CloudController
       end
 
       it 'schedules the frequent inline jobs' do
-        allow_any_instance_of(CloudController::DependencyLocator).to receive(:statsd_client).and_return(instance_double(Statsd))
+        allow_any_instance_of(CloudController::DependencyLocator).to receive(:statsd_client).and_return(instance_double(StatsD::Instrument::Client))
         allow(clock).to receive(:schedule_daily_job)
         allow(clock).to receive(:schedule_frequent_worker_job)
         expect(clock).to receive(:schedule_frequent_inline_job) do |args, &block|

spec/unit/lib/cloud_controller/dependency_locator_spec.rb

@@ -479,11 +479,13 @@
         statsd_port: port
       )
 
-      expected_client = double(Statsd)
+      sink = instance_double(StatsD::Instrument::Sink)
+      expected_client = instance_double(StatsD::Instrument::Client)
 
-      allow(Statsd).to receive(:new).with(host, port).
-        and_return(expected_client)
+      allow(StatsD::Instrument::Sink).to receive(:for_addr).with("#{host}:#{port}").and_return(sink)
+      allow(StatsD::Instrument::Client).to receive(:new).with(sink: sink).and_return(expected_client)
 
+      locator.instance_variable_set(:@statsd_sinks, nil)
       expect(locator.statsd_client).to eq(expected_client)
     end
 
@@ -510,11 +512,13 @@
         enable_statsd_metrics: nil
       )
 
-      expected_client = double(Statsd)
+      sink = instance_double(StatsD::Instrument::Sink)
+      expected_client = instance_double(StatsD::Instrument::Client)
 
-      allow(Statsd).to receive(:new).with(host, port).
-        and_return(expected_client)
+      allow(StatsD::Instrument::Sink).to receive(:for_addr).with("#{host}:#{port}").and_return(sink)
+      allow(StatsD::Instrument::Client).to receive(:new).with(sink: sink).and_return(expected_client)
 
+      locator.instance_variable_set(:@statsd_sinks, nil)
       expect(locator.statsd_client).to eq(expected_client)
     end
   end

spec/unit/lib/cloud_controller/deployment_updater/scheduler_spec.rb

@@ -27,7 +27,7 @@ module VCAP::CloudController
       let(:lock_runner) { instance_double(Locket::LockRunner, start: nil, lock_acquired?: nil) }
       let(:lock_worker) { instance_double(Locket::LockWorker) }
       let(:logger) { instance_double(Steno::Logger, info: nil, debug: nil, error: nil) }
-      let(:statsd_client) { instance_double(Statsd) }
+      let(:statsd_client) { instance_double(StatsD::Instrument::Client) }
 
       before do
         allow(Locket::LockRunner).to receive(:new).and_return(lock_runner)
@@ -38,8 +38,9 @@ module VCAP::CloudController
         allow(DeploymentUpdater::Scheduler).to receive(:sleep)
         allow(DeploymentUpdater::Dispatcher).to receive(:dispatch)
         allow(CloudController::DependencyLocator.instance).to receive_messages(statsd_client:)
-        allow(statsd_client).to receive(:timing)
-        allow(statsd_client).to receive(:batch) # mock vital metrics updates
+        allow(statsd_client).to receive(:measure)
+        allow(statsd_client).to receive(:gauge) # mock vital metrics updates
+        allow(statsd_client).to receive(:increment)
       end
 
       it 'correctly configures a LockRunner and uses it to initialize a LockWorker' do
@@ -130,7 +131,7 @@ module VCAP::CloudController
       describe 'metrics' do
         it 'records the deployment update duration' do
           expect(DeploymentUpdater::Dispatcher).to receive(:dispatch)
-          expect(statsd_client).to receive(:timing).with('cc.deployments.update.duration', kind_of(Numeric))
+          expect(statsd_client).to receive(:measure).with('cc.deployments.update.duration', kind_of(Numeric))
 
           DeploymentUpdater::Scheduler.start
         end