diff --git a/.circleci/config.yml b/.circleci/config.yml index fdda4ac..9ae0943 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -51,6 +51,22 @@ workflows: <<: *ruby_3_4_defaults name: ruby-3_4-rspec_unit db: false + # Forks real Resque children against a redis, so it is opt-in and excluded from the run above. The ruby executor + # already provides redis on localhost:6379, which is where the specs look by default. + # + # One ruby version is enough: what it exercises is fork and HTTP behaviour rather than anything version + # specific, and forking a few hundred children per version buys nothing. 3.4 matches the consuming service. + - ruby/rspec-unit: + <<: *ruby_3_4_defaults + name: ruby-3_4-rspec_fork_integration + db: false + code-climate: false + report-code-coverage: false + additional_args: "spec/integration" + pre-exec-hooks: + - run: + name: Enable the fork integration specs + command: echo 'export FORK_INTEGRATION=1' >> "$BASH_ENV" ruby_4_0: jobs: - ruby/bundle-audit: diff --git a/.rubocop.yml b/.rubocop.yml index ce31336..f9d59f8 100644 --- a/.rubocop.yml +++ b/.rubocop.yml @@ -20,6 +20,11 @@ Lint/AmbiguousOperator: Layout/LineLength: Enabled: false +# Client carries the delivery path: the queue, the lock, the timeouts and the reporting. Cohesive rather than long, +# and splitting it would mean handing a collaborator most of the client's state. +Metrics/ClassLength: + Max: 120 + Metrics/MethodLength: Max: 20 diff --git a/CHANGELOG.md b/CHANGELOG.md index 50dd450..8204e9f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ Changelog for the bc-prometheus-ruby gem. +## 0.8.4 + +- Reset the Prometheus client in forked Resque children via `Resque.after_fork`. A child previously inherited a copy of the parent's undrained outbound queue and had to re-send every message in it before reaching its own, which Resque's `exit!` cut short. Observations pushed from inside a job were dropped as a result. +- Deliver a forked Resque child's own queued metrics before the child exits, by wrapping `Resque::Worker#perform`. Pushing only queues, and `exit!` does not wait for the thread that would deliver it, so metrics pushed from inside a job were unreliable regardless of the above. Costs one request per job that pushed something and nothing for jobs that did not. Disable with `PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED=0`. +- Serialise delivery to the collector on its own mutex, so a flush cannot return while the background thread still has a message in flight. An empty queue is not an empty wire: the worker thread pops before it sends, and a child exiting in that window destroyed the request. Also removes a hang where both threads saw one queued message, both called `pop`, and the loser blocked forever. +- Bound the connect, response and write timeouts when delivering to the collector, configurable via `PROMETHEUS_CLIENT_OPEN_TIMEOUT`, `PROMETHEUS_CLIENT_READ_TIMEOUT` and `PROMETHEUS_CLIENT_WRITE_TIMEOUT`. `Net::HTTP` defaults all three to 60 seconds, which an unhealthy collector could previously impose on the caller. +- Bound a flush to `PROMETHEUS_CLIENT_FLUSH_TIMEOUT`, 20ms by default, covering the wait for the delivery lock as well as the requests. A forked child holds up real work while it delivers, so an unhealthy collector now costs it a known amount rather than however long the network takes to give up. Past the deadline the observations are abandoned, because availability of the work matters more than completeness of its metrics. +- Report abandoned observations, and push the warning out of the process before `exit!` destroys it. A line written to a buffered STDOUT in a Resque child never reaches the log, so the buffers are flushed after warning rather than left to an exit that runs no handlers. + ## 0.8.3 - Add opt-in per-Resque-job histograms `resque_job_queue_latency_seconds` and `resque_job_perform_duration_seconds`, labelled by `job_class`. diff --git a/Gemfile b/Gemfile index 69117a8..012edec 100644 --- a/Gemfile +++ b/Gemfile @@ -26,4 +26,13 @@ gem 'rubocop-performance', '>= 1.5' gem 'rubocop-rspec' gem 'simplecov', '>= 0.16' +# Resque is an optional integration, but its fork-per-job lifecycle is the one thing the client has to survive, so the +# integration needs real coverage rather than stubs. +# +# Resque depends on sinatra for its web UI with a loose `>= 0.9.2`. There is no Gemfile.lock in this repo, so CI +# resolves cold and the resolver is free to pick an old sinatra that caps `rack < 3`, which conflicts with the +# gemspec's `rack >= 3.0`. Pinning sinatra forward keeps the resolution rack-3 compatible. +gem 'resque', '>= 2.0' +gem 'sinatra', '>= 4.0' + gemspec diff --git a/README.md b/README.md index a95b20d..6abe648 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,60 @@ require 'bigcommerce/prometheus' Bigcommerce::Prometheus::Instrumentors::Resque.new(app: Rails.application).start ``` +### Metrics pushed from inside a job + +Resque runs each job in a forked child that ends with `exit!`, which runs no at_exit handlers and does not wait for +threads. Pushing a metric only queues it; delivery happens on a background thread that wakes every +`client_thread_sleep` seconds. A child that pushes and then returns is normally torn down before that thread runs, so +the observation is silently discarded. + +Two things are done automatically to make an in-child push arrive: + +- The child is given a clean client queue at fork time, via `Resque.after_fork`. Without this it would inherit a copy + of whatever the parent had not yet drained and have to re-send all of it before reaching its own message. +- The child delivers its own queue on the calling thread before the job returns, by wrapping + `Resque::Worker#perform`. Delivery is serialised against the background thread, so a request already in progress + finishes before the child exits rather than being destroyed with it. + +Cost is one request to the local collector per job that pushed something, and nothing at all for jobs that pushed +nothing. Disable with `PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED=0` if a service would rather have the throughput and can +accept the loss. + +A job is real work, and it should not wait on the metrics pipeline for long. Delivery is therefore bounded by +`PROMETHEUS_CLIENT_FLUSH_TIMEOUT`, 20ms by default, covering the wait for the delivery lock as well as the requests +themselves. An unhealthy collector costs a job that much and no more. Past the deadline the observations are abandoned +and a warning is logged saying how many, which is the only signal you will get, since the metric that would have +reported the outage is the one being lost. + +Note that this applies to metrics your application code pushes from inside a job. The per-job histograms below are +recorded in the parent and never pay this cost. + +### Measuring delivery and cost + +`bin/resque_fork_bench` forks real children against a real redis and a real listener, and reports how many of the +observations pushed inside a job arrived and how much slower the job got. It needs a redis and takes tens of seconds, so +it is a manual tool rather than part of `script/test`. + +```bash +redis-server --port 6399 --save '' --appendonly no --daemonize yes + +REDIS_URL=redis://127.0.0.1:6399/15 bin/resque_fork_bench --smoke-test +``` + +`--smoke-test` sweeps a representative spread of job shapes. A single shape can be measured directly instead, varying +how many metrics the job pushes, how much work separates them, and how much work follows the last one: + +```bash +bin/resque_fork_bench --pushes 2 --gap 0.05 # two pushes, 50ms apart +bin/resque_fork_bench --pushes 1 --trailing 0.01 # one push, then 10ms of work +bin/resque_fork_bench --smoke-test --no-child-flush +``` + +Read the total column rather than the overhead column. Overhead subtracts a control run that performed the same sleeps, +so it hides any part of a wait that overlapped the job's own work. `--help` lists the rest, including +`--thread-sleep` for the drain cadence and `--redis-url`. Exits non-zero if any row lost an observation or exceeded the +overhead budget the integration specs enforce. + ### Per-job metrics (opt-in) Set `PROMETHEUS_RESQUE_PER_JOB_METRICS_ENABLED=1` on Resque worker pods to enable two additional histograms recorded from the parent worker process. @@ -67,6 +121,10 @@ After requiring the main file, you can further configure with: | client_custom_labels | A hash of custom labels to send with each client request | `{}` | None | | client_max_queue_size | The max amount of metrics to send before flushing | `10000` | `ENV['PROMETHEUS_CLIENT_MAX_QUEUE_SIZE']` | | client_thread_sleep | How often to sleep the worker thread that manages the client buffer (seconds) | `0.5` | `ENV['PROMETHEUS_CLIENT_THREAD_SLEEP']` | +| client_open_timeout | Connect timeout when delivering to the collector (seconds) | `0.5` | `ENV['PROMETHEUS_CLIENT_OPEN_TIMEOUT']` | +| client_read_timeout | Response timeout when delivering to the collector (seconds) | `1.0` | `ENV['PROMETHEUS_CLIENT_READ_TIMEOUT']` | +| client_write_timeout | Send timeout when delivering to the collector (seconds) | `0.5` | `ENV['PROMETHEUS_CLIENT_WRITE_TIMEOUT']` | +| client_flush_timeout | Total a synchronous flush will spend before abandoning what is queued (seconds) | `0.02` | `ENV['PROMETHEUS_CLIENT_FLUSH_TIMEOUT']` | | puma_collection_frequency | How often to poll puma collection metrics (seconds) | `30` | `ENV['PROMETHEUS_PUMA_COLLECTION_FREQUENCY']` | | server_host | The host to run the exporter on | `"0.0.0.0"` | `ENV['PROMETHEUS_SERVER_HOST']` | | server_port | The port to run the exporter on | `9394` | `ENV['PROMETHEUS_SERVER_PORT']` | @@ -74,6 +132,7 @@ After requiring the main file, you can further configure with: | process_name | What the current process name is (used in logging) | `"unknown"` | `ENV['PROCESS']` | | railtie_disabled | Opt out flag for Railtie; use `Bigcommerce::Prometheus::Instrumentors::Web.new(app: Rails.application).start` in your app's code to start it up yourself | `0` | `ENV['PROMETHEUS_DISABLE_RAILTIE']` | | resque_per_job_metrics_enabled | Enable per-job queue-latency and perform-duration histograms (parent-side, no synchronous flush) | `0` | `ENV['PROMETHEUS_RESQUE_PER_JOB_METRICS_ENABLED']` | +| resque_child_flush_enabled | Deliver a forked child's own queued metrics before Resque exits it | `1` | `ENV['PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED']` | ## Custom Collectors diff --git a/bin/resque_fork_bench b/bin/resque_fork_bench new file mode 100755 index 0000000..09be5e6 --- /dev/null +++ b/bin/resque_fork_bench @@ -0,0 +1,472 @@ +#!/usr/bin/env ruby +# frozen_string_literal: true + +# Copyright (c) 2019-present, BigCommerce Pty. Ltd. All rights reserved +# +# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated +# documentation files (the "Software"), to deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit +# persons to whom the Software is furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the +# Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE +# WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +# COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +# +# Measures what a Resque fork-per-job worker actually delivers to the collector, and what it costs per job to deliver +# it. Forks real children against a real redis and a real listener. +# +# Two numbers come out of each run. How many of the observations pushed inside a job arrived, and how much slower the +# job got. They are worth measuring together because changes in this area tend to trade one against the other: an +# implementation can deliver everything by making every job wait, or keep every job fast by quietly dropping the +# observations, and only looking at both catches either. +# +# Read the total column, not just the overhead column. Overhead subtracts a control run that performed the same sleeps, +# so it hides any part of a wait that overlapped the job's own work. +# +# The spec at spec/integration/resque_fork_delivery_spec.rb asserts these same two properties on one fixed shape. This +# is the exploratory version of it, for checking a change by hand before CI does. +# +require 'bundler/setup' +require 'logger' +require 'optparse' +require 'resque' +require 'bigcommerce/prometheus' +require_relative '../spec/support/counting_exporter' + +## +# The job under measurement. Everything that varies travels in the Resque payload rather than through globals, so the +# forked child picks it up the way a real job picks up its arguments. +# +class ForkBenchProbeJob + METRIC_NAME = 'resque_fork_bench_probe' + @queue = :bc_prometheus_fork_bench + + def self.perform(payload) + gap = payload['gap'].to_f + trailing = payload['trailing'].to_f + + payload['pushes'].to_i.times do |index| + sleep(gap) if index.positive? && gap.positive? + push + end + + # Work after the last push, with nothing pushed after it. This is what an asynchronous delivery needs in order to + # win: a job that returns the moment it has pushed gives the delivery no time at all. + sleep(trailing) if trailing.positive? + end + + def self.push + ::Bigcommerce::Prometheus.client.send_json( + type: 'resque_fork_bench', + name: METRIC_NAME, + value: 1.0 + ) + end +end + +## +# Stands in for a collector that is not working, so the cost of an outage can be measured rather than argued about. +# +# `stalled` accepts the connection and then never answers, which is what a saturated exporter looks like from the +# client's side and is the expensive case. `down` leaves nothing listening, so connections are refused immediately and +# cost nothing. Presents the same interface as CountingExporter, and counts nothing, because nothing arrives. +# +class UnhealthyCollector + attr_reader :port + + def initialize(mode) + server = TCPServer.new('127.0.0.1', 0) + @port = server.addr[1] + @server = mode == 'down' ? server.close : server + @accepted = [] + end + + def start + @thread = Thread.new { loop { @accepted << @server.accept } } if @server + self + end + + def stop + @thread&.kill + @accepted.each(&:close) + @server&.close + end + + def count_for(_name) + 0 + end + + def stats + Hash.new(0) + end +end + +## +# One job shape to measure. +# +Shape = Struct.new(:pushes, :gap, :trailing, keyword_init: true) + +## +# What one shape measured. +# +Result = Struct.new(:shape, :jobs, :control_seconds, :metrics_seconds, :delivered, keyword_init: true) do + def expected + jobs * shape.pushes + end + + def control_ms + control_seconds / jobs * 1000 + end + + def total_ms + metrics_seconds / jobs * 1000 + end + + def overhead_ms + total_ms - control_ms + end + + def complete? + delivered == expected + end +end + +## +# Runs the shapes and prints the table. +# +class ForkBench + # The budget spec/integration/resque_fork_delivery_spec.rb enforces on CI. Reported here so a local run can be + # compared against the thing that will actually block a merge. + OVERHEAD_BUDGET_MS = 25.0 + + # Long enough for a straggling delivery from the last child to land before its row is counted, so that a late arrival + # is not attributed to the next row. + SETTLE_SECONDS = 0.25 + + QUEUE = :bc_prometheus_fork_bench + + # A representative spread rather than an exhaustive one. Rows 1 and 2 vary the work after the last push, which is what + # decides whether an asynchronous delivery ever gets to run. Rows 3 to 5 vary the work between two pushes, which is + # what decides whether the child ends up waiting on the client's drain cadence. Row 6 varies metrics per job, to show + # whether cost is per metric or per job. + SMOKE_SHAPES = [ + Shape.new(pushes: 1, gap: 0.0, trailing: 0.0), + Shape.new(pushes: 1, gap: 0.0, trailing: 0.005), + Shape.new(pushes: 2, gap: 0.0, trailing: 0.0), + Shape.new(pushes: 2, gap: 0.05, trailing: 0.0), + Shape.new(pushes: 2, gap: 0.30, trailing: 0.0), + Shape.new(pushes: 5, gap: 0.0, trailing: 0.0) + ].freeze + + ROW_FORMAT = '%6s %8s %9s | %10s %10s %10s | %s' + HEADINGS = ['pushes', 'gap', 'trailing', 'control', 'total', 'overhead', 'delivered'].freeze + + def initialize(options) + @options = options + end + + ## + # @return [Integer] process exit status + # + def call + return 1 unless redis_reachable? + + setup + print_preamble + results = measure_all + print_footer(results) + print_listener_diagnostics + results.all? { |result| result.complete? && result.overhead_ms < OVERHEAD_BUDGET_MS } ? 0 : 1 + ensure + @exporter&.stop + end + + private + + def shapes + return SMOKE_SHAPES if @options[:smoke_test] + + [Shape.new(pushes: @options[:pushes], gap: @options[:gap], trailing: @options[:trailing])] + end + + def jobs + @options[:jobs] || (@options[:smoke_test] ? 50 : 100) + end + + # --- setup ------------------------------------------------------------- + + def redis_reachable? + Redis.new(url: @options[:redis_url], timeout: 1.0).ping + true + rescue StandardError => e + warn "Cannot reach redis at #{@options[:redis_url]}: #{e.message}" + warn 'Start one with: redis-server --port 6399 --save \'\' --appendonly no --daemonize yes' + false + end + + def setup + Resque.redis = Redis.new(url: @options[:redis_url]) + Resque.logger = Logger.new(File::NULL) + + @exporter = if @options[:collector] == 'healthy' + CountingExporter.new.start + else + UnhealthyCollector.new(@options[:collector]).start + end + + configure_prometheus + ::Bigcommerce::Prometheus::Integrations::Resque.start(client: client) + end + + def configure_prometheus + exporter_port = @exporter.port + thread_sleep = @options[:thread_sleep] + log_device = @options[:verbose] ? $stderr : File::NULL + + ::Bigcommerce::Prometheus.configure do |config| + config.enabled = true + config.logger = Logger.new(log_device) + config.server_host = '127.0.0.1' + config.server_port = exporter_port + config.client_thread_sleep = thread_sleep + config.client_flush_timeout = @options[:flush_timeout] + config.resque_child_flush_enabled = @options[:child_flush] + end + end + + ## + # The client is a singleton that reads host, port and drain cadence once, when it is first built. Set them on the + # instance as well as in the configuration so that the run is not at the mercy of whether something else built it + # first. + # + def client + @client ||= ::Bigcommerce::Prometheus.client.tap do |instance| + instance.instance_variable_set(:@host, '127.0.0.1') + instance.instance_variable_set(:@port, @exporter.port) + instance.instance_variable_set(:@thread_sleep, @options[:thread_sleep]) + instance.instance_variable_set(:@flush_timeout, @options[:flush_timeout]) + instance.reset_after_fork! if instance.respond_to?(:reset_after_fork!) + end + end + + # --- measurement ------------------------------------------------------- + + def measure_all + shapes.map do |shape| + result = measure(shape) + puts row_for(result) + result + end + end + + ## + # Times the same shape twice, once with metrics off and once on, so that the machine's own speed cancels out of the + # overhead figure. The sleeps happen in both runs; only the pushes are conditional. + # + # @param [Shape] shape + # @return [Result] + # + def measure(shape) + delivered_before = @exporter.count_for(ForkBenchProbeJob::METRIC_NAME) + + ::Bigcommerce::Prometheus.enabled = false + control_seconds = time_run(shape) + + ::Bigcommerce::Prometheus.enabled = true + metrics_seconds = time_run(shape) + + sleep SETTLE_SECONDS + + Result.new( + shape: shape, + jobs: jobs, + control_seconds: control_seconds, + metrics_seconds: metrics_seconds, + delivered: @exporter.count_for(ForkBenchProbeJob::METRIC_NAME) - delivered_before + ) + end + + ## + # Drains the queue one job at a time so the run contains no timers of its own. Each call still forks, runs the + # after_fork hooks and exits the child exactly as a live worker does. + # + # @param [Shape] shape + # @return [Float] seconds elapsed + # + def time_run(shape) + payload = { 'pushes' => shape.pushes, 'gap' => shape.gap, 'trailing' => shape.trailing } + + Resque.redis.redis.del("queue:#{QUEUE}") + jobs.times { Resque::Job.create(QUEUE, ForkBenchProbeJob, payload) } + worker = Resque::Worker.new(QUEUE) + + started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + jobs.times { worker.work_one_job } + Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at + end + + # --- output ------------------------------------------------------------ + + def print_preamble + puts "bc-prometheus-ruby #{::Bigcommerce::Prometheus::VERSION} child flush: #{child_flush_state} " \ + "flush timeout: #{(@options[:flush_timeout] * 1000).round}ms " \ + "client_thread_sleep: #{@options[:thread_sleep]}s collector: #{@options[:collector]}" + puts "#{jobs} jobs per row, run twice: once with metrics disabled as a control, once with them enabled." + puts 'Times are per job. delivered counts the envelopes that reached the listener.' + puts + puts "#{smoke_legend}\n" if @options[:smoke_test] + puts format(ROW_FORMAT, *HEADINGS) + puts separator + end + + def smoke_legend + <<~LEGEND + Rows 1 and 2 vary the work after the last push, which is what an asynchronous delivery would need in order to win. + Rows 3 to 5 vary the work between two pushes, which is what decides whether the child waits on the drain cadence. + Row 6 varies metrics per job. Expect roughly #{estimated_seconds} seconds, longer if children have to wait. + LEGEND + end + + def estimated_seconds + per_job = shapes.sum { |shape| shape.gap * (shape.pushes - 1) + shape.trailing + 0.004 } + (per_job * jobs * 2).round + end + + def child_flush_state + unless defined?(::Bigcommerce::Prometheus::Integrations::Resque::ChildFlush) + return 'not available in this version' + end + + @options[:child_flush] ? 'on' : 'off' + end + + def row_for(result) + format( + ROW_FORMAT, + result.shape.pushes, + duration(result.shape.gap), + duration(result.shape.trailing), + milliseconds(result.control_ms), + milliseconds(result.total_ms), + milliseconds(result.overhead_ms), + "#{result.delivered}/#{result.expected}#{annotation(result)}" + ) + end + + def annotation(result) + notes = [] + notes << "#{result.expected - result.delivered} lost" unless result.complete? + notes << 'over budget' if result.overhead_ms >= OVERHEAD_BUDGET_MS + notes.empty? ? '' : " <- #{notes.join(', ')}" + end + + def print_footer(results) + puts separator + incomplete = results.reject(&:complete?) + if incomplete.empty? + puts 'Every row delivered every observation.' + else + puts "#{incomplete.size} of #{results.size} rows lost observations." + end + puts format( + 'Worst per-job overhead %.1f ms, against the %.1f ms budget the integration specs enforce.', + results.map(&:overhead_ms).max, + OVERHEAD_BUDGET_MS + ) + end + + ## + # Printed only when the listener saw something it could not account for. Distinguishes an observation the client never + # managed to send from one that arrived and was not counted, which is the difference between a delivery bug and a + # measurement bug. + # + def print_listener_diagnostics + stats = @exporter.stats + unaccounted = stats[:accepted] - stats[:send_metrics] + return if unaccounted.zero? && stats[:parse_error].zero? && stats[:serve_error].zero? + + puts + puts "Listener accepted #{stats[:accepted]} connections and read #{stats[:send_metrics]} complete requests." + if stats[:abandoned].positive? + puts "Opened and then closed with no request line: #{stats[:abandoned]}. That is what a caller looks like when " \ + 'it dies between connecting and writing, so those observations left the queue and never reached the wire.' + end + puts "Malformed or short bodies: #{stats[:parse_error]} unparseable, #{stats[:short_body]} truncated, " \ + "#{stats[:serve_error]} errored." + end + + def separator + '-' * format(ROW_FORMAT, *HEADINGS).length + end + + def duration(seconds) + seconds.positive? ? format('%g ms', seconds * 1000) : '-' + end + + def milliseconds(value) + format('%.1f ms', value) + end +end + +options = { + jobs: nil, + pushes: 1, + gap: 0.0, + trailing: 0.0, + thread_sleep: 0.5, + flush_timeout: Bigcommerce::Prometheus.client_flush_timeout, + child_flush: true, + collector: 'healthy', + redis_url: ENV.fetch('REDIS_URL', 'redis://127.0.0.1:6379/15'), + smoke_test: false, + verbose: false +} + +OptionParser.new do |parser| + parser.banner = 'Usage: bin/resque_fork_bench [options]' + parser.separator '' + parser.separator 'Job shape:' + parser.on('-j', '--jobs N', Integer, 'Jobs per row (default 100, or 50 with --smoke-test)') { |v| options[:jobs] = v } + parser.on('-p', '--pushes N', Integer, 'Metrics the job pushes (default 1)') { |v| options[:pushes] = v } + parser.on('-g', '--gap SECONDS', Float, 'Work between pushes (default 0)') { |v| options[:gap] = v } + parser.on('-t', '--trailing SECONDS', Float, 'Work after the last push (default 0)') { |v| options[:trailing] = v } + parser.separator '' + parser.separator 'Client:' + parser.on('-s', '--thread-sleep SECONDS', Float, 'client_thread_sleep, the drain cadence (default 0.5)') do |v| + options[:thread_sleep] = v + end + parser.on('-f', '--flush-timeout SECONDS', Float, 'What a child will spend delivering before giving up') do |v| + options[:flush_timeout] = v + end + parser.on('--[no-]child-flush', 'Deliver a child\'s own queue before it exits (default on)') do |v| + options[:child_flush] = v + end + parser.separator '' + parser.separator 'Run:' + parser.on('--smoke-test', 'Sweep a representative spread of shapes instead of one') { options[:smoke_test] = true } + parser.on('--collector MODE', %w[healthy stalled down], + 'healthy (default), stalled (accepts, never answers) or down (refused)') do |v| + options[:collector] = v + end + parser.on('--redis-url URL', 'Redis to run Resque against (default $REDIS_URL or db 15 on localhost)') do |v| + options[:redis_url] = v + end + parser.on('-v', '--verbose', 'Send the gem\'s own log to stderr') { options[:verbose] = true } + parser.on('-h', '--help', 'Show this message') do + puts parser + exit 0 + end + parser.separator '' + parser.separator 'Exits non-zero if any row lost an observation or exceeded the specs\' overhead budget.' + parser.separator '' + parser.separator 'Examples:' + parser.separator ' bin/resque_fork_bench --smoke-test' + parser.separator ' bin/resque_fork_bench --pushes 2 --gap 0.05 # the shape that cost 500ms per job' + parser.separator ' bin/resque_fork_bench --smoke-test --no-child-flush # what the same sweep looks like without it' +end.parse! + +exit ForkBench.new(options).call diff --git a/lib/bigcommerce/prometheus.rb b/lib/bigcommerce/prometheus.rb index e10d2d9..f8a782a 100644 --- a/lib/bigcommerce/prometheus.rb +++ b/lib/bigcommerce/prometheus.rb @@ -49,6 +49,7 @@ require_relative 'prometheus/integrations/resque/vanilla_resque_payload' require_relative 'prometheus/integrations/resque/job_payload' require_relative 'prometheus/integrations/resque/job_metrics' +require_relative 'prometheus/integrations/resque/child_flush' require_relative 'prometheus/servers/puma/server' require_relative 'prometheus/servers/puma/rack_app' diff --git a/lib/bigcommerce/prometheus/client.rb b/lib/bigcommerce/prometheus/client.rb index caa1abd..e8c32f4 100644 --- a/lib/bigcommerce/prometheus/client.rb +++ b/lib/bigcommerce/prometheus/client.rb @@ -24,6 +24,12 @@ class Client < ::PrometheusExporter::Client include Singleton include Loggable + # How often `flush!` retries the delivery lock while waiting for another thread to finish sending. + LOCK_POLL_SECONDS = 0.001 + + # Below this there is no point starting a request, so the remaining budget is spent reporting the drop instead. + MINIMUM_ATTEMPT_SECONDS = 0.001 + ## # @param [String] host # @param [Integer] port @@ -42,6 +48,11 @@ def initialize(host: nil, port: nil, max_queue_size: nil, thread_sleep: nil, cus ) PrometheusExporter::Client.default = self @process_name = process_name || ::Bigcommerce::Prometheus.process_name + @open_timeout = ::Bigcommerce::Prometheus.client_open_timeout + @read_timeout = ::Bigcommerce::Prometheus.client_read_timeout + @write_timeout = ::Bigcommerce::Prometheus.client_write_timeout + @flush_timeout = ::Bigcommerce::Prometheus.client_flush_timeout + @delivery_mutex = Mutex.new end ## @@ -80,17 +91,171 @@ def send(str) ## # Process the current queue and flush to the collector # + # Serialised so that only one thread is ever delivering. Two callers reach here, the background worker thread and + # `flush!` on the caller's own thread, and without the lock either could return while the other still had a + # message in flight. It also closes a hang: both threads can see a length of one, both call `pop`, and the loser + # blocks on an empty queue forever. + # def process_queue + @delivery_mutex.synchronize { drain } + end + + ## + # Deliver anything queued, on the calling thread, and return once it has been sent. + # + # `send_json` only queues; delivery happens on a background thread that wakes every `client_thread_sleep` + # seconds. A process that is about to exit does not get that far, so anything pushed shortly before exit is + # discarded with it. Callers that are about to exit should flush. + # + # Returns once everything is delivered, including anything the background thread had already taken off the queue + # and was part way through sending. An empty queue is not the same as an empty wire, so this deliberately does not + # short circuit on one: the message the worker thread popped a microsecond ago is exactly the one about to be lost. + # + # Bounded by `client_flush_timeout`, covering the wait for the delivery lock and the requests themselves. The + # caller is normally about to exit and is holding up real work while it waits, so an unhealthy collector must cost + # it a known amount rather than however long the network takes to give up. Past the deadline the observations are + # abandoned and reported, because availability of the work matters more than completeness of its metrics. + # + # A caller that pushed nothing pays one uncontended lock acquire, since a process that never pushed never started + # a worker thread. Never raises: metric delivery must not take down the work that produced the metric. + # + def flush! + deliver_before(monotonic_now + @flush_timeout) + rescue StandardError => e + report("Prometheus Exporter failed to flush: #{e}") + ensure + report_undelivered + end + + ## + # Discard the state a forked child inherited from its parent. + # + # The client is a singleton, so `fork` hands the child a copy of the parent's outbound queue while leaving the + # thread that would drain it behind. Anything still queued in the parent therefore has to be re-sent by the child, + # one request each, before the child reaches its own observation — and a Resque child is torn down by `exit!` long + # before that finishes. Discarding the copy is safe: the parent still holds the originals and sends them on its + # own schedule. + # + # Both mutexes are reset for a rarer case. If the fork lands while another thread holds one, the child inherits a + # locked mutex with no owner, and blocks forever the first time it needs it. For `@mutex` that is the first push, + # and for `@delivery_mutex` it is the first delivery. + # + def reset_after_fork! + @queue = Queue.new + @worker_thread = nil + @mutex = Mutex.new + @delivery_mutex = Mutex.new + @socket = nil + @socket_started = nil + @socket_pid = nil + end + + private + + ## + # Take the delivery lock, send what is queued, and give the lock back. Gives up rather than queueing behind a + # delivery that will not finish in time. + # + # @param [Float] deadline monotonic clock reading to stop by + # + def deliver_before(deadline) + return unless acquire_delivery_lock(deadline) + + begin + drain(deadline) + ensure + @delivery_mutex.unlock + end + end + + ## + # @param [Float] deadline + # @return [Boolean] whether the lock was taken + # + def acquire_delivery_lock(deadline) + until @delivery_mutex.try_lock + return false if monotonic_now >= deadline + + sleep LOCK_POLL_SECONDS + end + true + end + + ## + # Send queued messages one at a time. The caller owns the delivery lock. + # + # @param [Float|NilClass] deadline monotonic clock reading to stop by, or nil to keep going until the queue is + # empty. The background thread passes nil, since it holds nothing up by waiting. + # + def drain(deadline = nil) while @queue.length.to_i.positive? + timeout = deadline && (deadline - monotonic_now) + break if timeout && timeout < MINIMUM_ATTEMPT_SECONDS + begin - message = @queue.pop - Net::HTTP.post(uri_path('/send-metrics'), message) + post_message(@queue.pop, timeout: timeout) rescue StandardError => e - logger.warn "[bigcommerce-prometheus][#{@process_name}] Prometheus Exporter is dropping a message to #{uri_path('/send-metrics')}: #{e}" + report("dropping a message to #{uri_path('/send-metrics')}: #{e}") raise end end end + + ## + # Post a single message with bounded timeouts. + # + # `Net::HTTP` defaults every timeout to 60 seconds. That is survivable on the background thread and is not + # survivable inline in a job, where an unhealthy collector would stall every unit of work. The collector is + # normally on localhost, so the defaults here are short, and an inline flush overrides them with whatever is left + # of its budget. + # + # @param [String] message + # @param [Float|NilClass] timeout overrides all three phases when given + # + def post_message(message, timeout: nil) + uri = uri_path('/send-metrics') + http = ::Net::HTTP.new(uri.host, uri.port) + http.open_timeout = timeout || @open_timeout + http.read_timeout = timeout || @read_timeout + http.write_timeout = timeout || @write_timeout + http.start { |connection| connection.post(uri.path, message) } + end + + ## + # Say so when observations are abandoned, rather than letting them disappear silently. This is the only signal + # that a collector outage is costing metrics, since the metric that would have reported it is the one being lost. + # + def report_undelivered + undelivered = @queue.size + return if undelivered.zero? + + report( + "abandoned #{undelivered} metric(s) after #{(@flush_timeout * 1000).round}ms: " \ + "#{uri_path('/send-metrics')} did not accept them in time" + ) + end + + ## + # Warn, then push the line out of the process. + # + # Callers of `flush!` are usually about to be torn down by `exit!`, which runs no handlers and flushes no + # buffers. A warning sitting in a buffered STDOUT is destroyed along with the metric it was reporting, so the + # buffers are emptied here rather than left to an exit that will never come. + # + # @param [String] message + # + def report(message) + logger.warn "[bigcommerce-prometheus][#{@process_name}] #{message}" + $stdout.flush + $stderr.flush + rescue StandardError + nil + end + + # @return [Float] + def monotonic_now + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end end end end diff --git a/lib/bigcommerce/prometheus/configuration.rb b/lib/bigcommerce/prometheus/configuration.rb index f3366ee..bce95e8 100644 --- a/lib/bigcommerce/prometheus/configuration.rb +++ b/lib/bigcommerce/prometheus/configuration.rb @@ -29,6 +29,10 @@ module Configuration client_custom_labels: nil, client_max_queue_size: ENV.fetch('PROMETHEUS_CLIENT_MAX_QUEUE_SIZE', 10_000).to_i, client_thread_sleep: ENV.fetch('PROMETHEUS_CLIENT_THREAD_SLEEP', 0.5).to_f, + client_open_timeout: ENV.fetch('PROMETHEUS_CLIENT_OPEN_TIMEOUT', 0.5).to_f, + client_read_timeout: ENV.fetch('PROMETHEUS_CLIENT_READ_TIMEOUT', 1.0).to_f, + client_write_timeout: ENV.fetch('PROMETHEUS_CLIENT_WRITE_TIMEOUT', 0.5).to_f, + client_flush_timeout: ENV.fetch('PROMETHEUS_CLIENT_FLUSH_TIMEOUT', 0.02).to_f, # Integration configuration puma_collection_frequency: ENV.fetch('PROMETHEUS_PUMA_COLLECTION_FREQUENCY', 30).to_i, @@ -36,6 +40,7 @@ module Configuration resque_collection_frequency: ENV.fetch('PROMETHEUS_RESQUE_COLLECTION_FREQUENCY', 30).to_i, resque_process_label: ENV.fetch('PROMETHEUS_RESQUE_PROCESS_LABEL', 'resque').to_s, resque_per_job_metrics_enabled: ENV.fetch('PROMETHEUS_RESQUE_PER_JOB_METRICS_ENABLED', 0).to_i.positive?, + resque_child_flush_enabled: ENV.fetch('PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED', 1).to_i.positive?, # Server configuration not_found_body: ENV.fetch('PROMETHEUS_SERVER_NOT_FOUND_BODY', 'Not Found! The Prometheus Ruby Exporter only listens on /metrics and /send-metrics').to_s, diff --git a/lib/bigcommerce/prometheus/integrations/resque.rb b/lib/bigcommerce/prometheus/integrations/resque.rb index 53de6c7..3e317aa 100644 --- a/lib/bigcommerce/prometheus/integrations/resque.rb +++ b/lib/bigcommerce/prometheus/integrations/resque.rb @@ -26,18 +26,75 @@ class Resque # Start the resque integration # def self.start(client: nil) + resque_client = client || ::Bigcommerce::Prometheus.client + + # Installed first, and independently of the flag below. Together these are the safety net for every + # observation pushed from a forked child, so they must not depend on the collectors that follow starting + # successfully. Order matters between them: the reset is what keeps the flush down to a single request. + install_fork_reset(resque_client) + install_child_flush + ::PrometheusExporter::Instrumentation::Process.start( - client: client || ::Bigcommerce::Prometheus.client, + client: resque_client, type: ::Bigcommerce::Prometheus.resque_process_label ) ::Bigcommerce::Prometheus::Collectors::Resque.start( - client: client || ::Bigcommerce::Prometheus.client, + client: resque_client, frequency: ::Bigcommerce::Prometheus.resque_collection_frequency ) ::Bigcommerce::Prometheus::Integrations::Resque::JobMetrics.start( - client: client || ::Bigcommerce::Prometheus.client + client: resque_client + ) + end + + ## + # Hand each forked child a clean client instead of a copy of whatever the parent had not yet drained. + # + # Resque runs after_fork in the child, after reconnect and before the job body, which is the only point where + # the inherited copy can be dropped before anything tries to use it. Registered whatever the per-job metrics + # setting is: Collectors::Resque pushes from the parent every 30 seconds, so a child can inherit queued + # messages either way. + # + # Idempotent, because Resque appends after_fork hooks rather than replacing them. + # + # @param [PrometheusExporter::Client] client + # + def self.install_fork_reset(client) + return if @fork_reset_installed + + ::Resque.after_fork { |_job| client.reset_after_fork! if client.respond_to?(:reset_after_fork!) } + @fork_reset_installed = true + end + private_class_method :install_fork_reset + + ## + # Deliver a forked child's own observations before Resque's `exit!` discards them. + # + # Prepends rather than using a Resque hook because Resque has no in-child hook that runs after the job body. + # `Resque::Worker#perform` is that boundary. + # + # Idempotent, since a repeated prepend of an already-prepended module is a no-op but the guard keeps the + # intent explicit. + # + def self.install_child_flush + return if @child_flush_installed + + unless ::Bigcommerce::Prometheus.resque_child_flush_enabled + ::Bigcommerce::Prometheus.logger&.warn( + '[bigcommerce-prometheus] resque child metric flush is disabled; metrics pushed from inside a job will ' \ + 'not be delivered' + ) + return + end + + ::Resque::Worker.prepend(ChildFlush) + @child_flush_installed = true + ::Bigcommerce::Prometheus.logger&.info( + '[bigcommerce-prometheus] resque child metric flush installed; a job that pushes metrics delivers them ' \ + 'before the child exits' ) end + private_class_method :install_child_flush end end end diff --git a/lib/bigcommerce/prometheus/integrations/resque/child_flush.rb b/lib/bigcommerce/prometheus/integrations/resque/child_flush.rb new file mode 100644 index 0000000..e5579dc --- /dev/null +++ b/lib/bigcommerce/prometheus/integrations/resque/child_flush.rb @@ -0,0 +1,55 @@ +# frozen_string_literal: true + +# Copyright (c) 2019-present, BigCommerce Pty. Ltd. All rights reserved +# +# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated +# documentation files (the "Software"), to deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit +# persons to whom the Software is furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the +# Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE +# WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +# COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +# +module Bigcommerce + module Prometheus + module Integrations + class Resque + ## + # Deliver a forked child's observations before Resque tears it down. + # + # Resque runs each job in a child that ends with `exit!`, which runs no at_exit handlers and does not wait for + # threads. The client only queues on push and delivers on a background thread that wakes every + # `client_thread_sleep` seconds, so anything a job pushes is normally destroyed with the child. Draining on the + # calling thread before the job returns is the only way an in-child observation reliably arrives. + # + # This is deliberately not the approach taken for the `resque_job` histograms, which are recorded in the parent + # precisely to avoid paying anything per job. It exists for the observations application code pushes from + # inside its own jobs, which the gem cannot move anywhere else. + # + # Cost is one request to the local collector, and nothing at all when the job pushed no metrics. It depends on + # the child having been given a clean queue at fork time; without that the child would synchronously send the + # parent's backlog too, which is the latency problem this design exists to avoid. + # + # Disable with PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED=0. + # + module ChildFlush + ## + # Wraps `Resque::Worker#perform`, which is the in-child entry point when the worker forks per job. Guarded on + # `fork_per_job?` so a non-forking worker keeps the asynchronous path, since it is long-lived and its + # background thread drains on its own. + # + def perform(job, &block) + super + ensure + ::Bigcommerce::Prometheus.client.flush! if fork_per_job? + end + end + end + end + end +end diff --git a/lib/bigcommerce/prometheus/version.rb b/lib/bigcommerce/prometheus/version.rb index 6ae1b43..957c02e 100644 --- a/lib/bigcommerce/prometheus/version.rb +++ b/lib/bigcommerce/prometheus/version.rb @@ -17,6 +17,6 @@ # module Bigcommerce module Prometheus - VERSION = '0.8.3' + VERSION = '0.8.4' end end diff --git a/spec/bigcommerce/prometheus/client_spec.rb b/spec/bigcommerce/prometheus/client_spec.rb index 1af420b..59ba65e 100644 --- a/spec/bigcommerce/prometheus/client_spec.rb +++ b/spec/bigcommerce/prometheus/client_spec.rb @@ -55,4 +55,259 @@ end end end + + describe '#flush!' do + # Populated directly rather than through #send, which starts the background thread and would race the assertion. + let(:queue) { client.instance_variable_get(:@queue) } + + let(:delivery_mutex) { client.instance_variable_get(:@delivery_mutex) } + + before do + allow(Bigcommerce::Prometheus).to receive(:enabled).and_return(true) + @original_flush_timeout = client.instance_variable_get(:@flush_timeout) + client.reset_after_fork! + end + + after do + client.instance_variable_set(:@flush_timeout, @original_flush_timeout) + client.reset_after_fork! + end + + context 'when nothing is queued' do + it 'sends nothing, so a caller that never pushed pays nothing' do + allow(Net::HTTP).to receive(:new) + client.flush! + expect(Net::HTTP).not_to have_received(:new) + end + end + + context 'when the background thread is part way through a delivery' do + # The bug this exists to catch: `flush!` used to return as soon as the queue looked empty, and the queue looks + # empty the instant the worker thread pops the last message, well before that message reaches the wire. A child + # that exits at that point destroys the request. + # + # Rendezvous rather than sleeps, so the interleaving is fixed rather than hoped for. The only timing in the + # example is the join timeout, which asserts a negative and is therefore generous. + let(:entered_delivery) { Queue.new } + let(:release_delivery) { Queue.new } + + before do + # Generous, because this example is about waiting rather than about the deadline that bounds the wait. + client.instance_variable_set(:@flush_timeout, 5) + allow(client).to receive(:post_message) do + entered_delivery << true + release_delivery.pop + end + queue << 'in_flight_message' + end + + it 'does not return until that delivery has finished' do + Thread.new { client.process_queue } + entered_delivery.pop + + flusher = Thread.new { client.flush! } + expect(flusher.join(0.2)).to be_nil + + release_delivery << true + expect(flusher.join(2)).to eq flusher + end + end + + context 'when messages are queued' do + before do + allow(client).to receive(:post_message) + queue << 'queued_message' + end + + it 'delivers them on the calling thread' do + client.flush! + expect(client).to have_received(:post_message).with('queued_message', timeout: anything) + end + end + + context 'when the collector cannot be reached' do + before do + allow(client).to receive(:post_message).and_raise(StandardError, 'collector unreachable') + queue << 'queued_message' + end + + it 'does not raise into the caller, since a lost metric must not fail the work that produced it' do + expect { client.flush! }.not_to raise_error + end + end + + context 'when a delivery is in flight for longer than the flush timeout' do + let(:prometheus_logger) { instance_double(Logger, warn: nil) } + let(:lock_taken) { Queue.new } + + before do + allow(Bigcommerce::Prometheus).to receive(:logger).and_return(prometheus_logger) + client.instance_variable_set(:@flush_timeout, 0.02) + queue << 'stranded_message' + + @lock_holder = Thread.new do + delivery_mutex.lock + lock_taken << true + sleep + end + lock_taken.pop + end + + after do + @lock_holder.kill + @lock_holder.join + end + + it 'gives up, so an unhealthy collector cannot hold the caller up indefinitely' do + started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + client.flush! + expect(Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at).to be < 1 + end + + it 'says how many observations it abandoned, since nothing else will report them' do + client.flush! + expect(prometheus_logger).to have_received(:warn).with(/abandoned 1 metric/) + end + end + + context 'with a collector that does not answer' do + let(:http) do + instance_double(Net::HTTP, :open_timeout= => nil, :read_timeout= => nil, :write_timeout= => nil, start: nil) + end + + before do + allow(Net::HTTP).to receive(:new).and_return(http) + queue << 'queued_message' + end + + it 'bounds an inline flush on what is left of its budget, not on the background timeouts' do + client.instance_variable_set(:@flush_timeout, 0.02) + client.flush! + expect(http).to have_received(:read_timeout=).with(a_value_between(0, 0.02)) + end + + it 'bounds the background thread on the configured open timeout' do + client.process_queue + expect(http).to have_received(:open_timeout=).with(Bigcommerce::Prometheus.client_open_timeout) + end + + it 'bounds the background thread on the configured read timeout' do + client.process_queue + expect(http).to have_received(:read_timeout=).with(Bigcommerce::Prometheus.client_read_timeout) + end + + it 'bounds the write timeout, which Net::HTTP otherwise leaves at 60 seconds' do + client.process_queue + expect(http).to have_received(:write_timeout=).with(Bigcommerce::Prometheus.client_write_timeout) + end + end + end + + describe '#flush! against a collector that stops answering' do + # A real socket that accepts and then never replies, which is what a saturated exporter looks like from here. + # Timeouts run against the clock, so a stub cannot show that the bound holds. Without it this example takes as + # long as the read timeout, five seconds when it was written. + let(:stalled_collector) { TCPServer.new('127.0.0.1', 0) } + let(:prometheus_logger) { instance_double(Logger, warn: nil) } + + before do + allow(Bigcommerce::Prometheus).to receive_messages(enabled: true, logger: prometheus_logger) + @accepted = [] + @acceptor = Thread.new { loop { @accepted << stalled_collector.accept } } + + @original = %i[@host @port @flush_timeout].to_h { |name| [name, client.instance_variable_get(name)] } + client.instance_variable_set(:@host, '127.0.0.1') + client.instance_variable_set(:@port, stalled_collector.addr[1]) + client.instance_variable_set(:@flush_timeout, 0.02) + client.reset_after_fork! + client.instance_variable_get(:@queue) << 'queued_message' + end + + after do + @acceptor.kill + @accepted.each(&:close) + stalled_collector.close + @original.each { |name, value| client.instance_variable_set(name, value) } + client.reset_after_fork! + end + + it 'gives up on the flush timeout rather than on the much longer read timeout' do + started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + client.flush! + elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at + + expect(elapsed).to be < Bigcommerce::Prometheus.client_read_timeout + end + + it 'says the observation was dropped, so an outage is not silent' do + client.flush! + expect(prometheus_logger).to have_received(:warn).with(/dropping a message/) + end + end + + describe '#reset_after_fork!' do + before do + allow(Bigcommerce::Prometheus).to receive(:enabled).and_return(true) + end + + after { client.reset_after_fork! } + + it 'discards the messages a forked child inherited from its parent' do + client.send('inherited_message') + expect { client.reset_after_fork! }.to change { client.instance_variable_get(:@queue).size }.to 0 + end + + it 'replaces the queue rather than draining it, so the child never sends the parent messages' do + original = client.instance_variable_get(:@queue) + client.reset_after_fork! + expect(client.instance_variable_get(:@queue)).not_to be original + end + + it 'clears the inherited worker thread reference, since threads do not survive a fork' do + client.send('inherited_message') + client.reset_after_fork! + expect(client.instance_variable_get(:@worker_thread)).to be_nil + end + + it 'replaces a mutex that may have been held when the fork landed' do + original = client.instance_variable_get(:@mutex) + original.lock + client.reset_after_fork! + expect(client.instance_variable_get(:@mutex)).not_to be original + end + + it 'leaves the replacement mutex unlocked, so the first push in the child cannot deadlock' do + client.instance_variable_get(:@mutex).lock + client.reset_after_fork! + expect(client.instance_variable_get(:@mutex)).not_to be_locked + end + + it 'replaces the delivery mutex too, so the first delivery in the child cannot deadlock' do + client.instance_variable_get(:@delivery_mutex).lock + client.reset_after_fork! + expect(client.instance_variable_get(:@delivery_mutex)).not_to be_locked + end + + context 'with socket state inherited from the parent' do + before do + client.instance_variable_set(:@socket, :inherited_socket) + client.instance_variable_set(:@socket_started, Time.now.to_f) + client.instance_variable_set(:@socket_pid, Process.pid) + + client.reset_after_fork! + end + + it 'clears the socket' do + expect(client.instance_variable_get(:@socket)).to be_nil + end + + it 'clears the socket start time' do + expect(client.instance_variable_get(:@socket_started)).to be_nil + end + + it 'clears the socket pid' do + expect(client.instance_variable_get(:@socket_pid)).to be_nil + end + end + end end diff --git a/spec/integration/resque_fork_delivery_spec.rb b/spec/integration/resque_fork_delivery_spec.rb new file mode 100644 index 0000000..fae30b9 --- /dev/null +++ b/spec/integration/resque_fork_delivery_spec.rb @@ -0,0 +1,150 @@ +# frozen_string_literal: true + +# Copyright (c) 2019-present, BigCommerce Pty. Ltd. All rights reserved +# +# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated +# documentation files (the "Software"), to deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit +# persons to whom the Software is furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the +# Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE +# WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +# COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +# +require 'spec_helper' +require 'resque' + +## +# Pushes a metric as the final statement of the job, with nothing after it. +# +# This is the worst case and the shape that broke in production: once the push returns there is no further work, so +# Resque's `exit!` follows immediately and anything relying on the background delivery thread is lost. +# +class ForkDeliveryProbeJob + METRIC_NAME = 'fork_delivery_probe_counter' + @queue = :bc_prometheus_fork_delivery + + def self.perform(_payload) + Bigcommerce::Prometheus.client.send_json( + type: 'fork_delivery_probe', + name: METRIC_NAME, + value: 1.0 + ) + end +end + +## +# Black box: N jobs run, N observations arrive, and instrumentation does not blow out job time. +# +# Deliberately says nothing about queues, threads or forks. Both properties have been broken by past changes from +# opposite directions. PAYMENTS-11567 met completeness by adding 480ms per job. PAYMENTS-11727 kept job time flat by +# silently dropping the metrics. A change that satisfies one at the expense of the other should fail here. +# +# Forks real children and needs a redis, so it is excluded from the default run. To run it: +# +# FORK_INTEGRATION=1 REDIS_URL=redis://127.0.0.1:6379/15 bundle exec rspec spec/integration +# +describe 'metric delivery from Resque forked children', :fork_integration do + JOB_COUNT = 100 + + # Per-job overhead the instrumentation is allowed to add, measured against an identical run with metrics off so the + # machine's own speed cancels out. + # + # Measured cost is 0.90ms per job, against 2.5ms for the same jobs with metrics disabled, so the budget sits about + # 27x above what this should take and about 19x below the 480ms regression it exists to catch. Wide on both sides, + # because a latency assertion on CI hardware has to be. + OVERHEAD_BUDGET_SECONDS = 0.025 + + let(:exporter) { CountingExporter.new.start } + let(:queue) { ForkDeliveryProbeJob.instance_variable_get(:@queue) } + + before do + skip "redis unavailable at #{redis_url}" unless redis_available? + + Resque.redis = Redis.new(url: redis_url) + Resque.redis.redis.flushdb + Resque.logger = Logger.new(File::NULL) + + Bigcommerce::Prometheus.configure do |config| + config.enabled = true + config.logger = Logger.new(File::NULL) + config.server_host = '127.0.0.1' + config.server_port = exporter.port + # Far above the 20ms default. Completeness here is a claim about whether the child delivers at all, not about + # whether it wins a race against a deadline, and a loaded CI runner would otherwise make that flaky. + config.client_flush_timeout = 5.0 + end + + # The client is a singleton that captures host and port when it is first built, which earlier specs in the run may + # already have done. Point the existing instance at this run's exporter and drop anything they left queued, so the + # result does not depend on spec ordering. + client = Bigcommerce::Prometheus.client + client.instance_variable_set(:@host, '127.0.0.1') + client.instance_variable_set(:@port, exporter.port) + client.instance_variable_set(:@flush_timeout, 5.0) + client.reset_after_fork! + + Bigcommerce::Prometheus::Integrations::Resque.start(client: client) + end + + after { exporter.stop } + + describe 'completeness' do + it 'delivers one observation for every job that pushed one' do + run_jobs(JOB_COUNT) + + expect(exporter.count_for(ForkDeliveryProbeJob::METRIC_NAME)).to eq JOB_COUNT + end + end + + describe 'overhead' do + it 'adds less than the per-job budget over an identical run with metrics disabled' do + expect(measured_overhead_per_job).to be < OVERHEAD_BUDGET_SECONDS + end + end + + # --- helpers ----------------------------------------------------------- + + # Drains the queue synchronously, one job at a time, so the run has no timers or sleeps in it. Each call still forks, + # runs the after_fork hooks and exits the child exactly as a live worker does. + # + # @param [Integer] count + # @return [Float] seconds elapsed + def run_jobs(count) + count.times { Resque::Job.create(queue, ForkDeliveryProbeJob, 'n' => 1) } + worker = Resque::Worker.new(queue) + + # Monotonic clock rather than Benchmark: benchmark stopped being a default gem in Ruby 4.0, and requiring it here + # would break every job in the suite, not just this one. RSpec loads all spec files before it applies tag filters, + # so a top-level require in an excluded file still runs. + started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + count.times { worker.work_one_job } + Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at + end + + # @return [Float] seconds of instrumentation cost per job + def measured_overhead_per_job + Bigcommerce::Prometheus.enabled = false + without_metrics = run_jobs(JOB_COUNT) + + Bigcommerce::Prometheus.enabled = true + with_metrics = run_jobs(JOB_COUNT) + + (with_metrics - without_metrics) / JOB_COUNT + end + + def redis_url + ENV.fetch('REDIS_URL', 'redis://127.0.0.1:6379/15') + end + + def redis_available? + Redis.new(url: redis_url, timeout: 0.5).ping + true + rescue StandardError + false + end +end diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index 07e6061..3634cc4 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -26,4 +26,8 @@ mocks.allow_message_expectations_on_nil = true end config.color = true + + # Specs tagged :fork_integration fork real Resque children and need a redis, so they are opt-in rather than part of + # the default run. Enable with FORK_INTEGRATION=1. + config.filter_run_excluding(:fork_integration) unless ENV.fetch('FORK_INTEGRATION', nil) end diff --git a/spec/support/counting_exporter.rb b/spec/support/counting_exporter.rb new file mode 100644 index 0000000..258ed7f --- /dev/null +++ b/spec/support/counting_exporter.rb @@ -0,0 +1,128 @@ +# frozen_string_literal: true + +# Copyright (c) 2019-present, BigCommerce Pty. Ltd. All rights reserved +# +# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated +# documentation files (the "Software"), to deal in the Software without restriction, including without limitation the +# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit +# persons to whom the Software is furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the +# Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE +# WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +# COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR +# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +# +require 'json' +require 'socket' + +## +# Stands in for the exporter server and counts what arrives at /send-metrics. +# +# The fork-integration specs assert on how many observations actually crossed the process boundary, so they need a real +# listener rather than a stubbed client. Counting envelopes here keeps those assertions independent of type collector +# registration and of the exposition format. +# +class CountingExporter + # @return [Integer] the ephemeral port the exporter bound to + attr_reader :port + + def initialize + @server = TCPServer.new('127.0.0.1', 0) + @port = @server.addr[1] + @envelopes = [] + @stats = Hash.new(0) + @mutex = Mutex.new + end + + def start + @thread = Thread.new do + loop do + connection = @server.accept + Thread.new(connection) { |socket| serve(socket) } + end + end + self + end + + def stop + @thread&.kill + begin + @server.close + rescue StandardError + nil + end + end + + ## + # @param [String] name only count envelopes for this metric, ignoring anything the collectors also push + # @return [Integer] + # + def count_for(name) + @mutex.synchronize { @envelopes.count { |envelope| envelope['name'] == name } } + end + + ## + # What happened at the socket layer, which is what separates an observation that was never sent from one that was sent + # and not counted. Keys, all counts: + # + # - `accepted`: connections established + # - `send_metrics`: complete requests to /send-metrics + # - `abandoned`: connections established and then closed with no request line, the signature of a caller that died + # between connecting and writing + # - `short_body`: fewer bytes arrived than Content-Length promised + # - `parse_error`, `serve_error`: malformed body, and anything else raised while serving + # + # @return [Hash] + # + def stats + @mutex.synchronize { @stats.dup } + end + + private + + def serve(socket) + count(:accepted) + request_line = socket.gets + count(:abandoned) if request_line.nil? + body = read_body(socket) + if request_line.to_s.include?('/send-metrics') + count(:send_metrics) + record(body) + end + socket.write("HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nOK") + rescue StandardError + count(:serve_error) + nil + ensure + begin + socket.close + rescue StandardError + nil + end + end + + def read_body(socket) + content_length = 0 + while (line = socket.gets) && line != "\r\n" + content_length = line.split(':', 2).last.to_i if line.downcase.start_with?('content-length:') + end + return '' unless content_length.positive? + + socket.read(content_length).to_s.tap { |body| count(:short_body) if body.bytesize < content_length } + end + + def record(body) + envelope = JSON.parse(body) + @mutex.synchronize { @envelopes << envelope } + rescue JSON::ParserError + count(:parse_error) + nil + end + + def count(key) + @mutex.synchronize { @stats[key] += 1 } + end +end