Emit 'busy_threads' as 'workers.busy_threads'

.rubocop.yml

@@ -12,3 +12,7 @@ Metrics/BlockLength:
 Metrics/ClassLength:
   Exclude:
     - spec/**/*
+
+Metrics/ModuleLength:
+  Exclude:
+    - spec/**/*

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- New `workers.busy_threads` telemetry, reporting puma's `busy_threads` stat. Puma exposes this stat from 6.6 onwards, so it is only a default on those versions. Selecting it explicitly on an older puma raises with an explanation instead of silently reporting `0`
+
 ## [1.2.0]
 
 ### Added

README.md

@@ -111,14 +111,16 @@ Puma::Plugin::Telemetry.configure do |config|
   config.enabled = true
   config.initial_delay = 10
   config.frequency = 30
-  config.puma_telemetry = %w[workers.requests_count queue.backlog queue.capacity]
+  config.puma_telemetry = %w[workers.requests_count workers.busy_threads queue.backlog queue.capacity]
   config.socket_telemetry!
   config.socket_parser = :inspect
   config.add_target :io, io: StringIO.new, formatter: :json, transform: :passthrough
   config.add_target :dogstatsd, client: Datadog::Statsd.new(tags: { env: ENV["RAILS_ENV"] })
 end
 ```
 
+`workers.busy_threads` is backed by puma's `busy_threads` stat, which puma only reports from 6.6 onwards. It is part of the default telemetry on those versions and left out on older ones; selecting it explicitly on an older puma raises a `Puma::Plugin::Telemetry::Error`.
+
 ### Custom Targets
 
 Target is a simple object that implements `call` methods that accepts `telemetry` hash object. This means it can be super simple `proc` or some sophisticated class calling some external API.

lib/puma/plugin/telemetry/config.rb

@@ -17,6 +17,11 @@ class Config
           # Current number of threads spawned.
           'workers.spawned_threads',
 
+          # Current number of threads busy serving requests. Requires
+          # puma 6.6 or newer, it is dropped from the defaults below
+          # on older versions.
+          'workers.busy_threads',
+
           # Maximum number of threads that can run .
           'workers.max_threads',
 
@@ -32,6 +37,11 @@ class Config
           'queue.capacity'
         ].freeze
 
+        BUSY_THREADS_TELEMETRY = 'workers.busy_threads'
+
+        # Puma exposes the `busy_threads` stat from this version onwards.
+        BUSY_THREADS_MIN_PUMA_VERSION = Gem::Version.new('6.6')
+
         TARGETS = {
           dogstatsd: Telemetry::Targets::DatadogStatsdTarget,
           io: Telemetry::Targets::IOTarget,
@@ -59,7 +69,7 @@ class Config
         # Which metrics to publish from puma stats. You can select
         # a subset from default ones that interest you the most.
         # - default: DEFAULT_PUMA_TELEMETRY
-        attr_accessor :puma_telemetry
+        attr_reader :puma_telemetry
 
         # Whenever to publish socket telemetry.
         # - default: false
@@ -84,11 +94,26 @@ def initialize
           @initial_delay = 5
           @frequency = 5
           @targets = []
-          @puma_telemetry = DEFAULT_PUMA_TELEMETRY
+          @puma_telemetry = default_puma_telemetry
           @socket_telemetry = false
           @socket_parser = :unpack
         end
 
+        def puma_telemetry=(telemetry)
+          if telemetry.include?(BUSY_THREADS_TELEMETRY) && !self.class.busy_threads_supported?
+            raise Telemetry::Error,
+                  "#{BUSY_THREADS_TELEMETRY} requires puma >= #{BUSY_THREADS_MIN_PUMA_VERSION}, " \
+                  "but puma #{::Puma::Const::PUMA_VERSION} is installed. Upgrade puma, or drop " \
+                  "#{BUSY_THREADS_TELEMETRY} from `config.puma_telemetry`."
+          end
+
+          @puma_telemetry = telemetry
+        end
+
+        def self.busy_threads_supported?
+          Gem::Version.new(::Puma::Const::PUMA_VERSION) >= BUSY_THREADS_MIN_PUMA_VERSION
+        end
+
         def enabled?
           !!@enabled
         end
@@ -118,6 +143,17 @@ def add_target(name_or_target, **args)
 
           @targets.push(target.new(**args))
         end
+
+        private
+
+        # `workers.busy_threads` is only a default when the installed
+        # puma can actually report it. Selecting it explicitly on an
+        # older puma raises instead, see `#puma_telemetry=`.
+        def default_puma_telemetry
+          return DEFAULT_PUMA_TELEMETRY if self.class.busy_threads_supported?
+
+          DEFAULT_PUMA_TELEMETRY - [BUSY_THREADS_TELEMETRY]
+        end
       end
     end
   end

lib/puma/plugin/telemetry/data.rb

@@ -9,6 +9,7 @@ module CommonData
           'workers.booted' => :workers_booted,
           'workers.total' => :workers_total,
           'workers.spawned_threads' => :workers_spawned_threads,
+          'workers.busy_threads' => :workers_busy_threads,
           'workers.max_threads' => :workers_max_threads,
           'workers.requests_count' => :workers_requests_count,
           'queue.backlog' => :queue_backlog,
@@ -52,6 +53,10 @@ def workers_spawned_threads
           @stats.fetch(:running, 0)
         end
 
+        def workers_busy_threads
+          @stats.fetch(:busy_threads, 0)
+        end
+
         def queue_backlog
           @stats.fetch(:backlog, 0)
         end
@@ -79,6 +84,10 @@ def workers_spawned_threads
           sum_stat(:running)
         end
 
+        def workers_busy_threads
+          sum_stat(:busy_threads)
+        end
+
         def queue_backlog
           sum_stat(:backlog)
         end

spec/integration/plugin_spec.rb

@@ -45,18 +45,26 @@ class Plugin
           }
         end
 
+        # `workers.busy_threads` depends on what the server happens to be
+        # doing when telemetry is published, so only assert that it is there.
+        def expect_telemetry_line(line, target)
+          expect(line).to start_with "target=#{target} telemetry={"
+          expect(line).to match(/"workers\.busy_threads"\s*=>\s*\d+/)
+          expect(line).to include(*expected_telemetry.map { |metric, value| "#{metric.inspect}=>#{value.inspect}" })
+        end
+
         it 'runs telemetry' do
           expect(@server.lines).to include(/plugin=telemetry msg="enabled, setting up runner\.\.\."/)
         end
 
         it 'executes the first target' do
           true until (line = @server.next_line).include?('target=01')
-          expect(line).to start_with "target=01 telemetry=#{expected_telemetry.inspect}"
+          expect_telemetry_line(line, '01')
         end
 
         it 'executes the second target' do
           true until (line = @server.next_line).include?('target=02')
-          expect(line).to start_with "target=02 telemetry=#{expected_telemetry.inspect}"
+          expect_telemetry_line(line, '02')
         end
       end
 
@@ -89,9 +97,10 @@ class Plugin
         it "doesn't crash" do
           true until (line = @server.next_line).include?('DEBUG -- : Statsd')
 
-          lines = ([line.slice(/workers.*/)] + Array.new(6) { @server.next_line.strip })
+          lines = ([line.slice(/workers.*/)] + Array.new(7) { @server.next_line.strip })
 
-          expect(lines).to eq(expected_telemetry)
+          expect(lines).to include(*expected_telemetry)
+          expect(lines.grep(/\Aworkers\.busy_threads:\d+\|g\z/).size).to eq(1)
         end
       end
 
@@ -118,7 +127,10 @@ def next_line_including(pattern)
         it "doesn't crash" do
           matched_telemetry = {}
 
-          until matched_telemetry.size == expected_telemetry.size
+          # One more than `expected_telemetry`, for `puma.workers.busy_threads`,
+          # whose value depends on what the server is doing when telemetry is
+          # published and so is only checked for presence.
+          until matched_telemetry.size == expected_telemetry.size + 1
             break unless next_line_including('OpenTelemetry::SDK::Metrics::State::MetricData')
 
             name = @server.next_line&.slice(/name="(.*)"/, 1)
@@ -128,7 +140,8 @@ def next_line_including(pattern)
             matched_telemetry[name] = value.to_i
           end
 
-          expect(matched_telemetry).to eq(expected_telemetry)
+          expect(matched_telemetry).to include(expected_telemetry)
+          expect(matched_telemetry).to include('puma.workers.busy_threads')
         end
       end
 

spec/puma/plugin/telemetry/config_spec.rb

@@ -18,6 +18,45 @@ module Telemetry
           end
         end
 
+        describe '#puma_telemetry' do
+          context 'when puma reports busy threads' do
+            before { stub_const('::Puma::Const::PUMA_VERSION', '6.6.0') }
+
+            it 'includes workers.busy_threads by default' do
+              expect(config.puma_telemetry).to include('workers.busy_threads')
+            end
+
+            it 'accepts workers.busy_threads' do
+              config.puma_telemetry = ['workers.busy_threads']
+
+              expect(config.puma_telemetry).to eq ['workers.busy_threads']
+            end
+          end
+
+          context 'when puma is too old to report busy threads' do
+            before { stub_const('::Puma::Const::PUMA_VERSION', '6.5.0') }
+
+            it 'drops workers.busy_threads from the defaults' do
+              expect(config.puma_telemetry).not_to include('workers.busy_threads')
+            end
+
+            it 'keeps the remaining default telemetry' do
+              expect(config.puma_telemetry).to eq(described_class::DEFAULT_PUMA_TELEMETRY - ['workers.busy_threads'])
+            end
+
+            it 'raises when workers.busy_threads is selected explicitly' do
+              expect { config.puma_telemetry = ['workers.busy_threads'] }
+                .to raise_error(Telemetry::Error, /requires puma >= 6\.6, but puma 6\.5\.0 is installed/)
+            end
+
+            it 'allows selecting other telemetry' do
+              config.puma_telemetry = ['queue.backlog']
+
+              expect(config.puma_telemetry).to eq ['queue.backlog']
+            end
+          end
+        end
+
         describe '#socket_telemetry!' do
           context 'when TCP_INFO is available' do
             before do

spec/puma/plugin/telemetry/data_spec.rb

@@ -0,0 +1,44 @@
+# frozen_string_literal: true
+
+module Puma
+  class Plugin
+    module Telemetry
+      RSpec.describe WorkerData do
+        describe '#metrics' do
+          it 'emits busy threads from Puma stats' do
+            data = described_class.new(busy_threads: 3)
+
+            expect(data.metrics(['workers.busy_threads'])).to eq(
+              'workers.busy_threads' => 3
+            )
+          end
+
+          it 'falls back to 0 when puma does not expose busy threads' do
+            data = described_class.new({})
+
+            expect(data.metrics(['workers.busy_threads'])).to eq(
+              'workers.busy_threads' => 0
+            )
+          end
+        end
+      end
+
+      RSpec.describe ClusteredData do
+        describe '#metrics' do
+          it 'sums busy threads from worker statuses' do
+            data = described_class.new(
+              worker_status: [
+                { last_status: { busy_threads: 2 } },
+                { last_status: { busy_threads: 1 } }
+              ]
+            )
+
+            expect(data.metrics(['workers.busy_threads'])).to eq(
+              'workers.busy_threads' => 3
+            )
+          end
+        end
+      end
+    end
+  end
+end

spec/puma/plugin/telemetry_spec.rb

@@ -18,20 +18,33 @@ class Plugin
       end
 
       describe '.build' do
+        let(:puma_stats) do
+          {
+            booted_workers: 2,
+            max_threads: 4,
+            requests_count: 5,
+            running: 6,
+            busy_threads: 7,
+            backlog: 8,
+            pool_capacity: 9
+          }
+        end
+
         let(:default_telemetry) do
           {
-            'workers.booted' => 1,
+            'workers.booted' => 2,
             'workers.total' => 1,
-            'workers.max_threads' => 0,
-            'workers.requests_count' => 0,
-            'workers.spawned_threads' => 0,
-            'queue.backlog' => 0,
-            'queue.capacity' => 0
+            'workers.spawned_threads' => 6,
+            'workers.busy_threads' => 7,
+            'workers.max_threads' => 4,
+            'workers.requests_count' => 5,
+            'queue.backlog' => 8,
+            'queue.capacity' => 9
           }
         end
 
         it 'returns default telemetry hash' do
-          allow(::Puma).to receive(:stats_hash).and_return({})
+          allow(::Puma).to receive(:stats_hash).and_return(puma_stats)
           expect(described_class.build).to eq(default_telemetry)
         end
       end