Refactor lib/ for idiomatic, terse Ruby

lib/yabeda/rack/queue/header_timestamp_parser.rb

@@ -4,46 +4,27 @@ module Yabeda
   module Rack
     module Queue
       class HeaderTimestampParser
-        MIN_EPOCH_SECONDS = Time.utc(2000, 1, 1).to_f
-        FUTURE_TOLERANCE_SECONDS = 30.0
-        NORMALIZATION_DIVISORS = [1_000_000.0, 1_000.0, 1.0].freeze
-        NUMBER_PATTERN = /[+-]?(?:\d+(?:\.\d+)?|\.\d+)/
-        T_EQUALS_PATTERN = /t\s*=\s*(#{NUMBER_PATTERN.source})/i
+        MIN_EPOCH = Time.utc(2000, 1, 1).to_f
+        FUTURE_TOLERANCE = 30.0
+        DIVISORS = [1_000_000.0, 1_000.0, 1.0].freeze
+        NUMBER_RE = /[+-]?(?:\d+(?:\.\d+)?|\.\d+)/
+        T_EQUALS_RE = /t\s*=\s*(#{NUMBER_RE.source})/i
 
         def parse(value, now:)
-          first_value = first_header_value(value)
-          return nil if first_value.empty?
+          first = value.to_s.split(",", 2).first.to_s.strip
+          return if first.empty?
 
-          token = extract_numeric_token(first_value)
-          return nil if token.nil?
-
-          normalize(Float(token), now)
+          token = first[T_EQUALS_RE, 1] || first[NUMBER_RE, 0]
+          normalize(Float(token), now) if token
         rescue ArgumentError, TypeError
-          nil
         end
 
         private
 
-        def first_header_value(value)
-          value.to_s.split(",", 2).first.to_s.strip
-        end
-
-        def extract_numeric_token(value)
-          value[T_EQUALS_PATTERN, 1] || value[NUMBER_PATTERN, 0]
-        end
-
-        def normalize(raw_timestamp, now)
-          max_allowed = now + FUTURE_TOLERANCE_SECONDS
-
-          NORMALIZATION_DIVISORS.each do |divisor|
-            candidate = raw_timestamp / divisor
-            next if candidate < MIN_EPOCH_SECONDS
-            next if candidate > max_allowed
-
-            return candidate
-          end
-
-          nil
+        def normalize(raw, now)
+          max = now + FUTURE_TOLERANCE
+          divisor = DIVISORS.find { |d| (raw / d).between?(MIN_EPOCH, max) }
+          raw / divisor if divisor
         end
       end
     end

lib/yabeda/rack/queue/metric.rb

@@ -2,26 +2,11 @@
 
 require "yabeda"
 
-module Yabeda
-  module Rack
-    module Queue
-      HISTOGRAM_BUCKETS = [
-        0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60
-      ].freeze
-
-      METRIC_NAME = :rack_queue_duration
-      METRIC_GROUP = :rack_queue
-      METRIC_UNIT = :seconds
-      METRIC_DESCRIPTION = "Time a request waited in the upstream queue before reaching the application"
-    end
-  end
-end
-
 Yabeda.configure do
-  group Yabeda::Rack::Queue::METRIC_GROUP do
-    histogram Yabeda::Rack::Queue::METRIC_NAME,
-      comment: Yabeda::Rack::Queue::METRIC_DESCRIPTION,
-      unit: Yabeda::Rack::Queue::METRIC_UNIT,
-      buckets: Yabeda::Rack::Queue::HISTOGRAM_BUCKETS
+  group :rack_queue do
+    histogram :rack_queue_duration,
+      comment: "Time a request waited in the upstream queue before reaching the application",
+      unit: :seconds,
+      buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60]
   end
 end

lib/yabeda/rack/queue/middleware.rb

@@ -4,80 +4,48 @@ module Yabeda
   module Rack
     module Queue
       class Middleware
-        HEADER_KEYS = %w[HTTP_X_REQUEST_START HTTP_X_QUEUE_START].freeze
-        REQUEST_BODY_WAIT_KEY = "puma.request_body_wait"
-
         class StderrLogger
-          def warn(message)
-            Kernel.warn(message)
-          end
+          def warn(message) = Kernel.warn(message)
         end
 
         class YabedaReporter
-          def observe(value)
-            Yabeda.rack_queue.rack_queue_duration.measure({}, value)
-          end
+          def observe(value) = Yabeda.rack_queue.rack_queue_duration.measure({}, value)
         end
 
-        def initialize(app, reporter: YabedaReporter.new, parser: HeaderTimestampParser.new, logger: nil, clock: nil)
+        def initialize(app, reporter: YabedaReporter.new, logger: nil, clock: nil)
           @app = app
           @reporter = reporter
-          @parser = parser
+          @parser = HeaderTimestampParser.new
           @logger = logger || StderrLogger.new
           @clock = clock || -> { Process.clock_gettime(Process::CLOCK_REALTIME) }
         end
 
         def call(env)
-          x_request_start = env[HEADER_KEYS[0]]
-          x_queue_start = env[HEADER_KEYS[1]]
-
-          if x_request_start || x_queue_start
-            now = @clock.call
-            request_start = request_start_timestamp(x_request_start, x_queue_start, now)
-            report_queue_time(env, now, request_start) if request_start
-          end
-
+          measure_queue_time(env) if env["HTTP_X_REQUEST_START"] || env["HTTP_X_QUEUE_START"]
           @app.call(env)
         end
 
         private
 
-        def request_start_timestamp(x_request_start, x_queue_start, now)
-          parsed = parse_header_timestamp(x_request_start, now)
-          return parsed if parsed
-
-          parse_header_timestamp(x_queue_start, now)
-        end
-
-        def parse_header_timestamp(value, now)
-          return nil if value.nil?
-
-          @parser.parse(value, now: now)
+        def measure_queue_time(env)
+          now = @clock.call
+          start = @parser.parse(env["HTTP_X_REQUEST_START"], now: now) ||
+            @parser.parse(env["HTTP_X_QUEUE_START"], now: now)
+          report_queue_time(env, now, start) if start
         end
 
         def report_queue_time(env, now, request_start)
           queue_time = now - request_start
-          if queue_time.negative?
-            @logger.warn("Negative rack queue duration (#{queue_time}) observed; dropping measurement")
-            return
-          end
-
-          body_wait = parse_request_body_wait(env[REQUEST_BODY_WAIT_KEY])
-          queue_time -= body_wait if body_wait
-          queue_time = 0.0 if queue_time.negative?
+          return @logger.warn("Negative rack queue duration (#{queue_time}); dropping") if queue_time.negative?
 
-          @reporter.observe(queue_time)
+          body_wait = parse_body_wait(env["puma.request_body_wait"])
+          @reporter.observe([queue_time - (body_wait || 0), 0.0].max)
         end
 
-        def parse_request_body_wait(value)
-          return nil if value.nil?
-
-          milliseconds = Float(value)
-          return nil if milliseconds.negative?
-
-          milliseconds / 1_000.0
+        def parse_body_wait(value)
+          ms = Float(value)
+          ms / 1_000.0 unless ms.negative?
         rescue ArgumentError, TypeError
-          nil
         end
       end
     end