From 715988346a27468902921f0107ab5840c65c9158 Mon Sep 17 00:00:00 2001 From: Aaron Forsander Date: Thu, 6 Aug 2026 16:46:55 -0500 Subject: [PATCH] Emit 'busy_threads' as 'workers.busy_threads' Puma reports the number of threads currently serving requests as 'busy_threads'. Expose it as a 'workers.busy_threads' metric. Puma only added this stat in 6.6, so it is included in DEFAULT_PUMA_TELEMETRY on 6.6 and newer and left out on older versions. Selecting it explicitly on an older puma raises a Telemetry::Error naming the required version, rather than silently publishing a metric that is always 0. The puma dependency itself is unchanged. Co-Authored-By: Claude Opus 5 (1M context) --- .rubocop.yml | 4 +++ CHANGELOG.md | 3 ++ README.md | 4 ++- lib/puma/plugin/telemetry/config.rb | 40 +++++++++++++++++++-- lib/puma/plugin/telemetry/data.rb | 9 +++++ spec/integration/plugin_spec.rb | 25 +++++++++---- spec/puma/plugin/telemetry/config_spec.rb | 39 ++++++++++++++++++++ spec/puma/plugin/telemetry/data_spec.rb | 44 +++++++++++++++++++++++ spec/puma/plugin/telemetry_spec.rb | 27 ++++++++++---- 9 files changed, 179 insertions(+), 16 deletions(-) create mode 100644 spec/puma/plugin/telemetry/data_spec.rb diff --git a/.rubocop.yml b/.rubocop.yml index 218fc5a..e4dd437 100644 --- a/.rubocop.yml +++ b/.rubocop.yml @@ -12,3 +12,7 @@ Metrics/BlockLength: Metrics/ClassLength: Exclude: - spec/**/* + +Metrics/ModuleLength: + Exclude: + - spec/**/* diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ec55ef..8300e9a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 95dcec4..b565f5a 100644 --- a/README.md +++ b/README.md @@ -111,7 +111,7 @@ 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 @@ -119,6 +119,8 @@ Puma::Plugin::Telemetry.configure do |config| 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. diff --git a/lib/puma/plugin/telemetry/config.rb b/lib/puma/plugin/telemetry/config.rb index 96df50a..084eeba 100644 --- a/lib/puma/plugin/telemetry/config.rb +++ b/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 diff --git a/lib/puma/plugin/telemetry/data.rb b/lib/puma/plugin/telemetry/data.rb index 437a279..91b62f7 100644 --- a/lib/puma/plugin/telemetry/data.rb +++ b/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 diff --git a/spec/integration/plugin_spec.rb b/spec/integration/plugin_spec.rb index 5cfcf3b..816b913 100644 --- a/spec/integration/plugin_spec.rb +++ b/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 diff --git a/spec/puma/plugin/telemetry/config_spec.rb b/spec/puma/plugin/telemetry/config_spec.rb index 98e764f..aa4c0f0 100644 --- a/spec/puma/plugin/telemetry/config_spec.rb +++ b/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 diff --git a/spec/puma/plugin/telemetry/data_spec.rb b/spec/puma/plugin/telemetry/data_spec.rb new file mode 100644 index 0000000..93a5737 --- /dev/null +++ b/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 diff --git a/spec/puma/plugin/telemetry_spec.rb b/spec/puma/plugin/telemetry_spec.rb index 5a3f899..51c7ac6 100644 --- a/spec/puma/plugin/telemetry_spec.rb +++ b/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