diff --git a/.circleci/config.yml b/.circleci/config.yml index fdda4ac..03ce351 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -4,23 +4,38 @@ orbs: ci: bigcommerce/internal@volatile ruby: bigcommerce/internal-ruby@volatile +# The orb builds its bundler cache key as `-{{ arch }}-{{ checksum }}` and defaults +# checksumfile to Gemfile.lock. This repo ships no lockfile, deliberately, so that key cannot be computed: +# `restore_cache` fails outright and every job installs its gems cold. The failure does not fail the job, so builds stay +# green and the cost goes unnoticed. +# +# The Gemfile is the stable file that does exist. It pins nothing, so the cache is only a warm vendor/bundle to start +# from and bundler still resolves fresh on every run, which is what a gem wants: CI is the only place a breaking +# upstream release gets caught. defaults: &defaults notify_failure: false + checksumfile: Gemfile +# cache_key varies per ruby because `{{ arch }}` is OS and CPU, not ruby version. A single bucket shared across 3.3, 3.4 +# and 4.0 would have each job save over the others, and native extensions built against one ruby ABI are not loadable +# by another. ruby_3_3_defaults: &ruby_3_3_defaults <<: *defaults + cache_key: gem-cache-ruby-3_3 e: name: ruby/ruby ruby-version: '3.3' ruby_3_4_defaults: &ruby_3_4_defaults <<: *defaults + cache_key: gem-cache-ruby-3_4 e: name: ruby/ruby ruby-version: '3.4' ruby_4_0_defaults: &ruby_4_0_defaults <<: *defaults + cache_key: gem-cache-ruby-4_0 e: name: ruby/ruby ruby-version: '4.0' @@ -51,6 +66,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..f2a4799 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: 130 + Metrics/MethodLength: Max: 20 diff --git a/CHANGELOG.md b/CHANGELOG.md index 50dd450..b3bfe87 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ Changelog for the bc-prometheus-ruby gem. +## 0.8.4 + +- Reset the Prometheus client in forked Resque children, by wrapping `Resque::Worker#perform`. 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. `Worker#perform` is what runs the `after_fork` hooks, so the reset lands ahead of every one of them: an application hook that records a metric and was registered before this integration started would otherwise have had its observation enqueued and then discarded. The reset only runs in a forked child, identified by a changed pid, so a non-forking worker keeps the queue it is still responsible for sending. +- Optionally 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 recorded inside a job were unreliable regardless of the above. Drains whichever client `Integrations::Resque.start` was given, so the queue delivered is the one the reset cleared. **Off by default**, because it costs one request per observation a job records, each queued message being sent separately, and upgrading the gem should not change how long anyone's jobs take. Enable with `PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED=1`. Jobs that record nothing pay nothing either way. +- Accept a callable for `resque_child_flush_enabled`, asked in the parent before every fork so the child inherits the answer and never evaluates anything itself. Lets the flush be driven by a feature flag, per process or per job, without a restart and without the gem depending on any flag service. A callable taking an argument receives the `Resque::Job`. Anything it raises is treated as "do not flush", since it runs as a `Resque.before_fork` hook where an escaping exception would stop the worker. The env var supplies the default and an assignment overrides it, as with every other setting, so a callable replaces the env var rather than layering on top of it. +- 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. `flush!` returns `:empty`, `:success`, `:timeout` or `:error`, and the report is driven by that rather than by the queue length alone: a flush that cannot take the delivery lock leaves an empty queue while the background thread is still sending, so counting the queue reported nothing lost in the one case where the process is about to destroy a request. + ## 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..03d8f77 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,119 @@ 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. + +**Always on:** the child is given a clean client queue at fork time, by wrapping `Resque::Worker#perform`. 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. This costs nothing and needs no configuration. `Worker#perform` is what runs the `after_fork` hooks, so +the reset happens before all of them, including any of your own that record metrics. + +**Opt in:** the child can also deliver 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. + +```bash +PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED=1 +``` + +Off by default, because it costs one request to the local collector for every observation a job records, and upgrading +this gem should not change how long anybody's jobs take. Jobs that record nothing pay nothing either way. Turn it on if +you record metrics from inside Resque jobs and would rather have them than the throughput. + +Each queued message is sent as its own request. That is how this gem has delivered metrics since it stopped using the +upstream chunked socket, so the flush does not add requests, it moves ones that were already being made onto the job's +critical path. A job that records one observation pays for one request; a job that records ten pays for ten. + +That default is a deliberate position rather than caution waiting to be undone. Turning it on for everyone would change +how long other people's jobs take, which is a breaking change and wants a version bump to match. + +### Turning it on and off at runtime + +`resque_child_flush_enabled` also accepts anything callable, which is asked in the **parent** before every fork. The +child inherits the answer through the fork, so a feature flag client never has to survive one: + +```ruby +Bigcommerce::Prometheus.configure do |config| + config.resque_child_flush_enabled = -> { MyFeatureFlags.enabled?('resque_child_metric_flush') } +end +``` + +A callable that accepts an argument is handed the `Resque::Job`, so the decision can vary per job as well as per +process. `Bigcommerce::Prometheus::Integrations::Resque::JobPayload.for(job).job_class` unwraps ActiveJob's payload if +you want the real class name rather than the wrapper's: + +```ruby +config.resque_child_flush_enabled = lambda do |job| + MyFeatureFlags.enabled?('resque_child_metric_flush', queue: job.queue) +end +``` + +The callable must not be relied on to succeed. Anything it raises is caught and treated as "do not flush", because it +runs as a `Resque.before_fork` hook where an escaping exception would stop the worker processing jobs. + +The env var supplies the default and an assignment overrides it, as with every other setting here, so a callable +replaces the env var rather than layering on top of it. If you want the env var to stay an override, say so in your own +callable: + +```ruby +config.resque_child_flush_enabled = lambda do + ENV.fetch('PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED', '0').to_i.positive? && + MyFeatureFlags.enabled?('resque_child_metric_flush') +end +``` + +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, which is the only signal you will get, since the metric that would have reported the outage is +the one being lost. + +That budget is for the whole flush rather than for each request, and each queued observation is a request of its own. +So a job recording one observation has the full 20ms for it, and a job recording ten shares the same 20ms between ten. +The more a job records, the likelier it is to lose the tail of what it recorded. Raise +`PROMETHEUS_CLIENT_FLUSH_TIMEOUT` if your jobs record several metrics each and you would rather have them than the +latency. + +`flush!` returns `:empty`, `:success`, `:timeout` or `:error` if you want to act on the result yourself. A timeout says +either that the deadline expired part way through sending, in which case the warning says how many observations were +abandoned, or that the delivery lock could not be taken at all. The second case leaves the queue empty, because the +background thread had already taken the message it was sending, so the warning names the in-flight request instead of a +count. + +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 +180,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 +191,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. Accepts a callable, asked in the parent before every fork | `0` | `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..7fc55ad --- /dev/null +++ b/bin/resque_fork_bench @@ -0,0 +1,474 @@ +#!/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, + # Defaults to whatever the gem would do, so PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED reaches the bench instead of being + # silently overridden by the option's own default. + child_flush: Bigcommerce::Prometheus.resque_child_flush_enabled, + 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 off, as the gem is)') 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..c8fe438 100644 --- a/lib/bigcommerce/prometheus.rb +++ b/lib/bigcommerce/prometheus.rb @@ -49,6 +49,8 @@ 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/integrations/resque/fork_reset' 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..c0e93a2 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,16 +91,214 @@ 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. + # + # @return [Symbol] one of :empty, :success, :timeout, :error + # + def flush! + outcome = attempt_flush + report_outcome(outcome) + outcome + 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 + + ## + # Deliver, and turn anything raised into an outcome. + # + # Separated from `flush!` so the outcome can be reported and then returned. An `ensure` there would not change + # the return value without an explicit `return`, which would swallow it. + # + # @return [Symbol] + # + def attempt_flush + deliver_before(monotonic_now + @flush_timeout) + rescue StandardError => e + report("Prometheus Exporter failed to flush: #{e}") + :error + end + + ## + # 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 + # @return [Symbol] + # + def deliver_before(deadline) + return :timeout 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) + sent = 0 while @queue.length.to_i.positive? + timeout = deadline && (deadline - monotonic_now) + return :timeout if timeout && timeout < MINIMUM_ATTEMPT_SECONDS + begin - message = @queue.pop - Net::HTTP.post(uri_path('/send-metrics'), message) + post_message(@queue.pop, timeout: timeout) + sent += 1 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 + sent.zero? ? :empty : :success + 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. + # + # Driven by the outcome rather than by the queue length alone. A timeout can leave the queue empty and still have + # lost something, so a count of zero is not the same as nothing lost. + # + # @param [Symbol] outcome + # + def report_outcome(outcome) + return if %i[success empty].include?(outcome) + + undelivered = @queue.size + return report_abandoned(undelivered) if undelivered.positive? + + # Nothing queued, and still not a success. The background thread had already taken the message off the queue + # and was sending it, which is why the lock could not be acquired. `@queue.size` cannot see that message, so + # counting the queue alone would report this as nothing lost, in the one case where something is. + report( + "gave up after #{flush_timeout_ms}ms waiting for an in-flight send to #{uri_path('/send-metrics')}; " \ + 'anything it was carrying is lost with this process' + ) + end + + ## + # @param [Integer] undelivered + # + def report_abandoned(undelivered) + report( + "abandoned #{undelivered} metric(s) after #{flush_timeout_ms}ms: " \ + "#{uri_path('/send-metrics')} did not accept them in time" + ) + end + + # @return [Integer] + def flush_timeout_ms + (@flush_timeout * 1000).round + 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 diff --git a/lib/bigcommerce/prometheus/configuration.rb b/lib/bigcommerce/prometheus/configuration.rb index f3366ee..822d0a0 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,11 @@ 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?, + # Off deliberately, not by oversight. Enabling it adds a synchronous request to every Resque job that records + # a metric, so changing this default changes how long other people's jobs take. That is a breaking change and + # wants a version bump to match, the way 0.4.0 handled moving the thread pool default from 20 to 3. + # Also accepts anything callable, resolved in the parent before every fork. + resque_child_flush_enabled: ENV.fetch('PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED', 0).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..93b6b50 100644 --- a/lib/bigcommerce/prometheus/integrations/resque.rb +++ b/lib/bigcommerce/prometheus/integrations/resque.rb @@ -26,18 +26,137 @@ class Resque # Start the resque integration # def self.start(client: nil) + resque_client = client || ::Bigcommerce::Prometheus.client + + # Installed ahead of the collectors, and the reset installed whatever the flag below says. 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. The reset is also what keeps the flush cheap: without it a child would + # synchronously re-send the parent's backlog before reaching its own message. + # + # Both wrap `Resque::Worker#perform`. Which one is prepended first does not matter, since the reset runs + # before `super` and the flush after it either way. + install_fork_reset(resque_client) + install_child_flush(resque_client) + ::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. + # + # Installed whatever the per-job flush setting is: Collectors::Resque pushes from the parent every 30 seconds, + # so a child can inherit queued messages either way. See `ForkReset` for why this wraps `Worker#perform` + # instead of registering an after_fork hook. + # + # The pid is recorded here because this runs in the worker parent, which makes it the reading every forked + # child differs from. + # + # Idempotent. `Module#prepend` ignores a module already in the ancestors, and a second call assigns the same + # two values. + # + # @param [PrometheusExporter::Client] client + # + def self.install_fork_reset(client) + ForkReset.client = client + ForkReset.installed_in_pid = Process.pid + ::Resque::Worker.prepend(ForkReset) + end + private_class_method :install_fork_reset + + ## + # Deliver a forked child's own observations before Resque's `exit!` discards them. + # + # Two hooks, because the decision and the delivery happen in different processes. `before_fork` runs in the + # parent, so that is where `resque_child_flush_enabled` is resolved, and the child inherits the answer through + # the fork itself. `Resque::Worker#perform` is the only in-child boundary that runs after the job body, and it + # is a method rather than a hook, hence the prepend. + # + # Resolving per fork rather than once here is what lets a caller pass a callable and change its mind at + # runtime without a restart. Nothing in the child ever evaluates it. + # + # Idempotent. Resque appends `before_fork` hooks rather than replacing them, so a second call would otherwise + # register a second hook. + # + # Takes the client rather than reaching for the singleton at flush time, so that the queue drained here is the + # one `install_fork_reset` cleared. A caller passing `client:` would otherwise get two different queues. + # + # @param [PrometheusExporter::Client] client + # + def self.install_child_flush(client) + return if @child_flush_installed + + unless ::Bigcommerce::Prometheus.resque_child_flush_enabled + # Info rather than warn: this is the default, and it is the behaviour every caller already had. A warning + # on every worker boot of every service would only teach people to ignore warnings. Said out loud anyway, + # because a metric that never arrives is otherwise indistinguishable from one that was never recorded. + ::Bigcommerce::Prometheus.logger&.info( + '[bigcommerce-prometheus] resque child metric flush is off, so metrics recorded inside a job are not ' \ + 'delivered; set PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED=1 to deliver them, at the cost of one request ' \ + 'per observation a job records' + ) + return + end + + ChildFlush.client = client + ::Resque::Worker.prepend(ChildFlush) + ::Resque.before_fork { |job| ChildFlush.enabled = resolve_child_flush(job) } + @child_flush_installed = true + log_child_flush_installed + end + private_class_method :install_child_flush + + ## + # Resolve whether this child should flush. Runs in the parent, before the fork. + # + # `resque_child_flush_enabled` is either a plain value or something callable. A callable is handed the + # `Resque::Job` when it accepts one, so a caller can decide per job as well as per process. + # `JobPayload.for(job).job_class` unwraps ActiveJob's payload if the caller wants the real class name. + # + # Never raises. This is a `before_fork` hook, so an exception here propagates into `perform_with_fork` and + # takes down job processing. A flaky feature flag must not be able to stop a worker, so a failure means no + # flush, which is the asynchronous behaviour callers had before this existed. + # + # @param [Resque::Job] job + # @return [Boolean] + # + def self.resolve_child_flush(job) + setting = ::Bigcommerce::Prometheus.resque_child_flush_enabled + return !!setting unless setting.respond_to?(:call) + + # Procs answer `arity` themselves. An object with a `#call` method does not, and asking `method(:call).arity` + # of a proc reports `Proc#call(*args)` as -1, which would hand a job to a callable that takes none. So ask + # the object first and fall back to its method. + arity = setting.respond_to?(:arity) ? setting.arity : setting.method(:call).arity + + !!(arity.zero? ? setting.call : setting.call(job)) + rescue StandardError => e + ::Bigcommerce::Prometheus.logger&.warn( + "[bigcommerce-prometheus] resque child metric flush check failed, not flushing this job: #{e}" + ) + false + end + private_class_method :resolve_child_flush + + def self.log_child_flush_installed + dynamic = ::Bigcommerce::Prometheus.resque_child_flush_enabled.respond_to?(:call) + resolution = dynamic ? 'resolved in the parent before every fork' : 'enabled for every job' + + ::Bigcommerce::Prometheus.logger&.info( + "[bigcommerce-prometheus] resque child metric flush installed, #{resolution}; a job that pushes metrics " \ + 'delivers them before the child exits' ) end + private_class_method :log_child_flush_installed 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..f01cb35 --- /dev/null +++ b/lib/bigcommerce/prometheus/integrations/resque/child_flush.rb @@ -0,0 +1,100 @@ +# 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 per observation still queued, and nothing at all when the job + # pushed no metrics. `Client#drain` posts each message separately, which is how this gem has delivered since it + # stopped using the upstream chunked socket; the flush does not add requests, it moves ones that already + # existed onto the job's critical path. + # + # 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. + # + # Off unless asked for. `resque_child_flush_enabled` defaults from PROMETHEUS_RESQUE_CHILD_FLUSH_ENABLED and, + # like every setting here, an assignment overrides that default. Assign something callable and it is asked + # before every fork instead of once at boot. See `Integrations::Resque.resolve_child_flush`. + # + module ChildFlush + class << self + ## + # Whether the child about to run should flush, decided by the parent in `Resque.before_fork` and inherited + # through the fork. Reading it here rather than baking the decision into the prepend is what lets a caller + # change its mind at runtime, and it keeps every evaluation in the long-lived parent where a feature flag + # client is safe to use. + # + # Defaults to false so that a child which somehow runs without the hook does nothing rather than something + # unasked for. In fork-per-job mode `before_fork` always runs first, so the default is not observed. + # + # @return [Boolean] + # + attr_accessor :enabled + + ## + # The client to drain. The same object `ForkReset` was handed, so what is delivered here is the queue the + # child was given at fork time. + # + # `Bigcommerce::Prometheus.client` is the singleton and is not necessarily that object. + # `Integrations::Resque.start` accepts a `client:`, and a caller passing one would otherwise have the reset + # and the flush working on two different queues. + # + # @return [PrometheusExporter::Client] + # + attr_accessor :client + + ## + # Deliver, if the client knows how. + # + # Asked rather than assumed, because honouring the caller's client means a plain + # `PrometheusExporter::Client` can reach here and it has no `flush!`. The NoMethodError would be raised + # from the `ensure` below, where it would replace whatever the job was already raising. + # + def flush + client.flush! if client.respond_to?(:flush!) + end + end + self.enabled = false + + ## + # 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 + ChildFlush.flush if fork_per_job? && ChildFlush.enabled + end + end + end + end + end +end diff --git a/lib/bigcommerce/prometheus/integrations/resque/fork_reset.rb b/lib/bigcommerce/prometheus/integrations/resque/fork_reset.rb new file mode 100644 index 0000000..d9751f5 --- /dev/null +++ b/lib/bigcommerce/prometheus/integrations/resque/fork_reset.rb @@ -0,0 +1,81 @@ +# 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 + ## + # Give a forked child a clean client, before anything in that child can use the one it inherited. + # + # 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. Everything the parent had not yet sent therefore has to be re-sent by the + # child, one request each, before it reaches its own observation, and Resque's `exit!` arrives long before that + # finishes. Discarding the copy is safe: the parent still holds the originals and sends them on its own + # schedule. + # + # Wraps `Resque::Worker#perform` rather than registering a `Resque.after_fork` hook. A hook works, but + # `Resque.after_fork` appends and `Resque::Worker#run_hook` runs hooks in registration order, so an application + # hook that records a metric and happened to be registered first would have its observation enqueued here and + # then discarded. `Worker#perform` is what runs those hooks, so wrapping it puts the reset ahead of all of them + # whatever order anyone registers in. + # + module ForkReset + class << self + ## + # The client to reset. Held here rather than captured in a closure so the flush half of this integration + # can be handed the same object. See `Integrations::Resque.start`. + # + # @return [PrometheusExporter::Client] + # + attr_accessor :client + + ## + # The process that installed this. Any other process reaching `#perform` is a forked child. + # + # @return [Integer] + # + attr_accessor :installed_in_pid + + ## + # Discard the inherited queue, if and only if this is a forked child. + # + # `fork_per_job?` is not enough on its own. `Worker#perform` also runs in the long-lived parent, both for a + # worker started with FORK_PER_JOB=false and through the deprecated `Worker#process`, and a reset there + # would throw away the queue the parent is still responsible for sending. A changed pid is the fact that + # actually distinguishes the two, and it is the same test the upstream client applies to its own socket. + # + def reset_if_forked + return if installed_in_pid.nil? || Process.pid == installed_in_pid + return unless client.respond_to?(:reset_after_fork!) + + client.reset_after_fork! + end + end + + ## + # @param [Resque::Job] job + # + def perform(job, &block) + ForkReset.reset_if_forked + super + 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..ff641c2 100644 --- a/spec/bigcommerce/prometheus/client_spec.rb +++ b/spec/bigcommerce/prometheus/client_spec.rb @@ -55,4 +55,312 @@ 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 + + # Distinguished from :success so a caller can tell "the job recorded nothing" from "the job recorded something + # and it arrived". Reached only after taking the delivery lock, never by short circuiting on an empty queue. + it 'reports :empty' do + allow(Net::HTTP).to receive(:new) + expect(client.flush!).to eq :empty + 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 + + it 'reports :success' do + expect(client.flush!).to eq :success + 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 + + it 'reports :error' do + expect(client.flush!).to eq :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 + + it 'reports :timeout' do + expect(client.flush!).to eq :timeout + end + end + + # The regression this exists to catch. `@queue.size` cannot see the message the background thread has already + # popped and is sending, so counting the queue alone reported "nothing abandoned" in the one case where something + # is: the process is about to `exit!` and destroy that request. + context 'when the lock cannot be taken and nothing is left on the queue' 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) + + @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 'reports :timeout rather than :empty, since the queue being empty is not the same as nothing being lost' do + expect(client.flush!).to eq :timeout + end + + it 'says something was lost, rather than staying silent on a count of zero' do + client.flush! + expect(prometheus_logger).to have_received(:warn).with(/in-flight send/) + 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/bigcommerce/prometheus/integrations/resque/child_flush_spec.rb b/spec/bigcommerce/prometheus/integrations/resque/child_flush_spec.rb new file mode 100644 index 0000000..6fc2244 --- /dev/null +++ b/spec/bigcommerce/prometheus/integrations/resque/child_flush_spec.rb @@ -0,0 +1,118 @@ +# 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' + +describe Bigcommerce::Prometheus::Integrations::Resque::ChildFlush do + let(:client) { instance_double(Bigcommerce::Prometheus::Client, flush!: nil) } + let(:singleton_client) { instance_double(Bigcommerce::Prometheus::Client, flush!: nil) } + + # Stands in for Resque::Worker, which is not loaded here. The prepend only needs `perform` to wrap and + # `fork_per_job?` to consult, mirroring how job_metrics_spec covers WorkerInstrumentation. + let(:worker_class) do + klass = Class.new do + attr_writer :fork_per_job, :raise_on_perform + + def perform(_job, &_block) + raise 'job blew up' if @raise_on_perform + + :performed + end + + def fork_per_job? + @fork_per_job.nil? ? true : @fork_per_job + end + end + klass.prepend(described_class) + klass + end + + let(:worker) { worker_class.new } + let(:job) { double('Resque::Job') } + + before do + @original_enabled = described_class.enabled + @original_client = described_class.client + described_class.client = client + end + + after do + described_class.enabled = @original_enabled + described_class.client = @original_client + end + + context 'when the parent enabled it for this fork' do + before { described_class.enabled = true } + + it 'delivers what the job recorded before the child exits' do + worker.perform(job) + expect(client).to have_received(:flush!) + end + + it 'still runs the job' do + expect(worker.perform(job)).to eq :performed + end + + it 'delivers even when the job raises, since whatever it recorded before failing is still worth having' do + worker.raise_on_perform = true + + expect { worker.perform(job) }.to raise_error('job blew up') + expect(client).to have_received(:flush!) + end + + # The queue drained has to be the one ForkReset cleared. Reaching for the singleton instead would deliver a + # different queue whenever a caller passed `client:` to `Integrations::Resque.start`, as the fork integration spec + # and the bench both do. + it 'delivers the configured client rather than the singleton' do + allow(Bigcommerce::Prometheus).to receive(:client).and_return(singleton_client) + + worker.perform(job) + + expect(client).to have_received(:flush!) + expect(singleton_client).not_to have_received(:flush!) + end + + # A caller may hand `Integrations::Resque.start` a plain upstream client, which has no `flush!`. Raising from the + # `ensure` that calls this would replace whatever the job was already raising. + it 'does nothing rather than raising when the client cannot flush' do + described_class.client = instance_double(PrometheusExporter::Client) + + expect { worker.perform(job) }.not_to raise_error + end + end + + context 'when the parent disabled it for this fork' do + before { described_class.enabled = false } + + it 'delivers nothing, so a caller that has turned it off pays nothing' do + worker.perform(job) + expect(client).not_to have_received(:flush!) + end + end + + context 'when the worker does not fork per job' do + before do + described_class.enabled = true + worker.fork_per_job = false + end + + it 'leaves delivery to the background thread, since the process is long-lived' do + worker.perform(job) + expect(client).not_to have_received(:flush!) + end + end +end diff --git a/spec/bigcommerce/prometheus/integrations/resque/fork_reset_spec.rb b/spec/bigcommerce/prometheus/integrations/resque/fork_reset_spec.rb new file mode 100644 index 0000000..cfcdcfa --- /dev/null +++ b/spec/bigcommerce/prometheus/integrations/resque/fork_reset_spec.rb @@ -0,0 +1,114 @@ +# 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. +# +describe Bigcommerce::Prometheus::Integrations::Resque::ForkReset do + let(:client) { instance_double(Bigcommerce::Prometheus::Client, reset_after_fork!: nil) } + let(:events) { [] } + + # Stands in for Resque::Worker, which is not loaded here. Its `perform` records the two things it does in the real + # one, in the order it does them: the after_fork hooks at worker.rb:345, then the job body at worker.rb:347. + let(:worker_class) do + recorder = events + klass = Class.new do + define_method(:perform) do |_job, &_block| + recorder << :after_fork_hooks + recorder << :job + :performed + end + end + klass.prepend(described_class) + klass + end + + let(:worker) { worker_class.new } + let(:job) { double('Resque::Job') } + + around do |example| + original_client = described_class.client + original_pid = described_class.installed_in_pid + example.run + described_class.client = original_client + described_class.installed_in_pid = original_pid + end + + before { described_class.client = client } + + context 'when running in a forked child' do + # Any pid but this process's own. Forking for real is what spec/integration/resque_fork_delivery_spec.rb is for; + # here the changed pid is the whole condition under test, so it is set directly. + before { described_class.installed_in_pid = Process.pid + 1 } + + it 'discards the queue the child inherited from its parent' do + worker.perform(job) + + expect(client).to have_received(:reset_after_fork!) + end + + # The reason this is a prepend rather than a Resque.after_fork hook. Hooks run in registration order, so one + # registered before this integration started would push an observation into the queue that is about to be thrown + # away. + it 'discards it before the after_fork hooks run, so nothing they record is thrown away' do + allow(client).to receive(:reset_after_fork!) { events << :reset } + + worker.perform(job) + + expect(events).to eq %i[reset after_fork_hooks job] + end + + it 'still runs the job' do + expect(worker.perform(job)).to eq :performed + end + end + + context 'when running in the process that installed it' do + before { described_class.installed_in_pid = Process.pid } + + # `Worker#perform` runs in the long-lived parent for a FORK_PER_JOB=false worker and via the deprecated + # `Worker#process`. Resetting there would throw away messages nobody else is going to send. + it 'leaves the queue alone, since the parent is still responsible for sending it' do + worker.perform(job) + + expect(client).not_to have_received(:reset_after_fork!) + end + + it 'still runs the job' do + expect(worker.perform(job)).to eq :performed + end + end + + context 'when no pid was recorded' do + before { described_class.installed_in_pid = nil } + + it 'leaves the queue alone, since there is nothing to say this is a child' do + worker.perform(job) + + expect(client).not_to have_received(:reset_after_fork!) + end + end + + context 'when the client cannot reset itself' do + let(:client) { instance_double(PrometheusExporter::Client) } + + before { described_class.installed_in_pid = Process.pid + 1 } + + # A caller may hand `Integrations::Resque.start` a plain upstream client, which has no `reset_after_fork!`. That + # costs the child the clean queue, but it must not cost it the job. + it 'does nothing rather than raising' do + expect { worker.perform(job) }.not_to raise_error + end + end +end diff --git a/spec/bigcommerce/prometheus/integrations/resque_spec.rb b/spec/bigcommerce/prometheus/integrations/resque_spec.rb new file mode 100644 index 0000000..a6db1ca --- /dev/null +++ b/spec/bigcommerce/prometheus/integrations/resque_spec.rb @@ -0,0 +1,108 @@ +# 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' + +describe Bigcommerce::Prometheus::Integrations::Resque do + # Resolves whether the child about to be forked should flush. Runs in the parent, which is the whole point: a caller + # can hand over a feature flag client without any of it reaching a forked child. + describe '.resolve_child_flush' do + subject(:resolved) { described_class.send(:resolve_child_flush, job) } + + let(:job) { double('Resque::Job', queue: 'scheduled_action') } + + around do |example| + original = Bigcommerce::Prometheus.resque_child_flush_enabled + example.run + Bigcommerce::Prometheus.resque_child_flush_enabled = original + end + + context 'when the setting is a plain value' do + it 'is true when enabled' do + Bigcommerce::Prometheus.resque_child_flush_enabled = true + expect(resolved).to be true + end + + it 'is false when disabled' do + Bigcommerce::Prometheus.resque_child_flush_enabled = false + expect(resolved).to be false + end + end + + context 'when the setting is callable' do + it 'asks it, so the answer can change between forks without a restart' do + answers = [true, false].each + Bigcommerce::Prometheus.resque_child_flush_enabled = -> { answers.next } + + expect(described_class.send(:resolve_child_flush, job)).to be true + expect(described_class.send(:resolve_child_flush, job)).to be false + end + + it 'coerces a truthy answer to a boolean' do + Bigcommerce::Prometheus.resque_child_flush_enabled = -> { 'yes' } + expect(resolved).to be true + end + + it 'passes the job when the callable takes one, so a caller can decide per job' do + Bigcommerce::Prometheus.resque_child_flush_enabled = ->(j) { j.queue == 'scheduled_action' } + expect(resolved).to be true + end + + it 'does not pass the job when the callable takes none' do + Bigcommerce::Prometheus.resque_child_flush_enabled = -> { true } + expect { resolved }.not_to raise_error + end + + # An object with a #call method is as ordinary a callable as a lambda, but it has no #arity, so asking for one + # directly raises and the rescue below turns that into a silent "never flush". + it 'accepts an object that responds to call rather than only procs' do + checker = Class.new do + def call(job) + job.queue == 'scheduled_action' + end + end.new + Bigcommerce::Prometheus.resque_child_flush_enabled = checker + + expect(resolved).to be true + end + end + + context 'when the callable raises' do + let(:logger) { instance_double(Logger, warn: nil) } + + before do + allow(Bigcommerce::Prometheus).to receive(:logger).and_return(logger) + Bigcommerce::Prometheus.resque_child_flush_enabled = -> { raise 'flag service unreachable' } + end + + # This runs as a before_fork hook, so anything escaping here propagates into perform_with_fork and stops the + # worker processing jobs. A flaky feature flag must never be able to do that. + it 'does not propagate, since a metrics decision must not stop a worker' do + expect { resolved }.not_to raise_error + end + + it 'falls back to not flushing, which is the behaviour callers had before this existed' do + expect(resolved).to be false + end + + it 'says why, so a silently disabled flush is diagnosable' do + resolved + expect(logger).to have_received(:warn).with(/child metric flush check failed/) + end + end + end +end diff --git a/spec/bigcommerce/prometheus/servers/puma/server_spec.rb b/spec/bigcommerce/prometheus/servers/puma/server_spec.rb index 9363e45..c26ef4c 100644 --- a/spec/bigcommerce/prometheus/servers/puma/server_spec.rb +++ b/spec/bigcommerce/prometheus/servers/puma/server_spec.rb @@ -1,11 +1,19 @@ require 'spec_helper' describe Bigcommerce::Prometheus::Servers::Puma::Server do - let(:server) { described_class.new(port: default_port) } - let(:default_port) { 9800 + rand(100) } + # Captured on assignment rather than named in the `after` hook below, because `let` is lazy and naming it there would + # build a server for an example that never wanted one. + let(:server) { @server = described_class.new(port: default_port) } + # Port 0 asks the kernel for a free one. These examples bind a real listener, so drawing from a range of a hundred + # meant two of them eventually picked the same port and the second failed with EADDRINUSE. Only the example that + # asserts on the configured port needs to name one. + let(:default_port) { 0 } before do Bigcommerce::Prometheus.reset end + # The constructor binds, so an example that builds a server holds a listener until something closes it. Nothing did, + # and they accumulated for the whole run. + after { @server&.binder&.close } context 'when the server is initialized' do it 'has a valid rack app' do diff --git a/spec/integration/resque_fork_delivery_spec.rb b/spec/integration/resque_fork_delivery_spec.rb new file mode 100644 index 0000000..3b8bcda --- /dev/null +++ b/spec/integration/resque_fork_delivery_spec.rb @@ -0,0 +1,153 @@ +# 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 + # Opt in explicitly. The flush is off by default so that bumping the gem cannot change anyone's job latency, and + # these examples are about what it does once a caller has asked for it. + config.resque_child_flush_enabled = true + # 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