PAYMENTS-11727 Deliver metrics recorded inside Resque jobs

.circleci/config.yml

@@ -55,6 +55,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:

CHANGELOG.md

@@ -1,5 +1,14 @@
 Changelog for the bc-prometheus-ruby gem.
 
+## 0.9.1
+
+- 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_FORK_EXIT_FLUSH_ENABLED=1`. Jobs that record nothing pay nothing either way.
+- Accept a callable for `resque_fork_exit_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 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. `Net::HTTP` caps connect, write and read separately rather than bounding a request as a whole, so per-request timeouts alone would let one slow message overrun the budget several times over. The flush therefore runs on a thread that is stopped once the budget is spent, which holds the wall clock to the configured value rather than a multiple of it.
+- 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.9.0
 
 - Add `Bigcommerce::Prometheus::Instrumentors::Protorabbit` so protorabbit (RabbitMQ protobuf consumer) processes run an embedded Prometheus exporter server, fixing dropped/refused metric pushes (`Errno::ECONNREFUSED` on `/send-metrics`) from those processes.

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

README.md

@@ -43,6 +43,96 @@ 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.
+
+> **Experimental.** This has not run in production at scale yet, and the timeout defaults may change once
+> it has. Start with a single low-traffic worker pool. Watch its logs for abandoned-metric warnings.
+
+```bash
+PROMETHEUS_RESQUE_FORK_EXIT_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_fork_exit_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_fork_exit_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_fork_exit_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_fork_exit_flush_enabled = lambda do
+  ENV.fetch('PROMETHEUS_RESQUE_FORK_EXIT_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.
+
 ### 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.
@@ -77,13 +167,15 @@ 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_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']` |
 | server_thread_pool_size | The number of threads used for the exporter server | `3` | `ENV['PROMETHEUS_SERVER_THREAD_POOL_SIZE']` |
 | 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_fork_exit_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_FORK_EXIT_FLUSH_ENABLED']` |
 
 ## Custom Collectors
 

lib/bigcommerce/prometheus.rb

@@ -29,6 +29,7 @@
 require_relative 'prometheus/loggable'
 require_relative 'prometheus/configuration'
 require_relative 'prometheus/server'
+require_relative 'prometheus/delivery'
 require_relative 'prometheus/client'
 
 require_relative 'prometheus/collectors/base'
@@ -50,6 +51,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/fork_exit_flush'
+require_relative 'prometheus/integrations/resque/fork_reset'
 
 require_relative 'prometheus/servers/puma/server'
 require_relative 'prometheus/servers/puma/rack_app'

lib/bigcommerce/prometheus/client.rb

@@ -20,6 +20,9 @@ module Prometheus
     ##
     # Client implementation for Prometheus
     #
+    # Queueing and registration are the superclass's. Everything about getting a queued message to the collector
+    # belongs to `Delivery`, which this hands its own queue to and rebuilds whenever that queue is replaced.
+    #
     class Client < ::PrometheusExporter::Client
       include Singleton
       include Loggable
@@ -42,6 +45,7 @@ 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
+        @delivery = build_delivery
       end
 
       ##
@@ -61,14 +65,6 @@ def close_socket_if_old!
         close_socket! if @socket && ((@socket_started.to_i + MAX_SOCKET_AGE) < Time.now.to_f)
       end
 
-      ##
-      # @param [String] path
-      # @return [Module<URI>]
-      #
-      def uri_path(path)
-        URI("http://#{@host}:#{@port}#{path}")
-      end
-
       ##
       # @param [String] str
       def send(str)
@@ -80,16 +76,59 @@ def send(str)
       ##
       # Process the current queue and flush to the collector
       #
+      # Overridden because the superclass streams onto a long-lived socket, which this gem stopped doing. Called by
+      # the background worker thread only.
+      #
       def process_queue
-        while @queue.length.to_i.positive?
-          begin
-            message = @queue.pop
-            Net::HTTP.post(uri_path('/send-metrics'), message)
-          rescue StandardError => e
-            logger.warn "[bigcommerce-prometheus][#{@process_name}] Prometheus Exporter is dropping a message to #{uri_path('/send-metrics')}: #{e}"
-            raise
-          end
-        end
+        @delivery.process_queue
+      end
+
+      ##
+      # Deliver what is queued before the caller stops being able to.
+      #
+      # @return [Symbol] one of :empty, :success, :timeout, :error
+      #
+      def flush!
+        @delivery.flush!
+      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
+      # Anything still queued in the parent therefore has to be re-sent by the child,
+      #
+      # The mutex is reset for a rarer case. If the fork happens while another thread holds the mutex, the child inherits a
+      # locked mutex and can never claim it.
+      #
+      # `Delivery` is rebuilt last, and for both reasons at once. It has to be given the new queue, and its own
+      # delivery lock may have been held by a thread that did not survive the fork.
+      #
+      def reset_after_fork!
+        @queue = Queue.new
+        @worker_thread = nil
+        @mutex = Mutex.new
+        @delivery = build_delivery
+      end
+
+      private
+
+      ##
+      # Whatever is queued now, and the settings to deliver it under.
+      #
+      # Built rather than assigned once, so `reset_after_fork!` can replace it wholesale. That is how a child stops
+      # sharing a queue, and a delivery lock, with the parent it forked from.
+      #
+      # @return [Bigcommerce::Prometheus::Delivery]
+      #
+      def build_delivery
+        Delivery.new(
+          queue: @queue,
+          host: @host,
+          port: @port,
+          flush_timeout: ::Bigcommerce::Prometheus.client_flush_timeout,
+          process_name: @process_name
+        )
       end
     end
   end

lib/bigcommerce/prometheus/configuration.rb

@@ -29,13 +29,19 @@ 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_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,
         puma_process_label: ENV.fetch('PROMETHEUS_PUMA_PROCESS_LABEL', 'web').to_s,
         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_fork_exit_flush_enabled: ENV.fetch('PROMETHEUS_RESQUE_FORK_EXIT_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,