refactor(platform): PAYMENTS-11567 Split JobPayload into ActiveJobPayload and VanillaResquePayload

Author: Willem Homan

lib/bigcommerce/prometheus.rb

@@ -45,6 +45,8 @@
 require_relative 'prometheus/integrations/railtie' if defined?(Rails)
 require_relative 'prometheus/integrations/puma'
 require_relative 'prometheus/integrations/resque'
+require_relative 'prometheus/integrations/resque/active_job_payload'
+require_relative 'prometheus/integrations/resque/vanilla_resque_payload'
 require_relative 'prometheus/integrations/resque/job_payload'
 require_relative 'prometheus/integrations/resque/job_metrics'
 

lib/bigcommerce/prometheus/integrations/resque/active_job_payload.rb

@@ -0,0 +1,69 @@
+# 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 'time'
+
+module Bigcommerce
+  module Prometheus
+    module Integrations
+      class Resque
+        ##
+        # Payload fields for an ActiveJob-shaped Resque job, read from the
+        # inner hash at `args[0]`. ActiveJob's JobWrapper stamps the three
+        # fields the per-job metrics consume:
+        #
+        #   * job_class    — the user's actual job class name; used as the
+        #                    metric label.
+        #   * enqueued_at  — ISO 8601 string; queue-latency anchor when
+        #                    scheduled_at is absent.
+        #   * scheduled_at — ISO 8601 string; preferred over enqueued_at
+        #                    when present (e.g. retries-with-backoff, so the
+        #                    intentional wait isn't counted as latency).
+        #
+        # Eagerly parses in #initialize and exposes plain attr_readers —
+        # does not hold a reference to the inner hash after construction.
+        #
+        class ActiveJobPayload
+          # @return [String] the user's actual job class name
+          attr_reader :job_class
+
+          # @return [Time, nil] the queue-latency anchor; nil when both
+          #   timestamps are absent or unparseable
+          attr_reader :anchor_time
+
+          # @param [Hash] inner the ActiveJob-shaped hash at `args[0]`;
+          #   JobPayload.for guarantees a truthy 'job_class'
+          def initialize(inner)
+            @job_class   = inner['job_class']
+            @anchor_time = parse_time(inner['scheduled_at']) || parse_time(inner['enqueued_at'])
+          end
+
+          private
+
+          def parse_time(value)
+            return value if value.is_a?(Time)
+            return nil if value.nil? || value.to_s.empty?
+
+            Time.iso8601(value.to_s)
+          rescue ArgumentError
+            nil
+          end
+        end
+      end
+    end
+  end
+end

lib/bigcommerce/prometheus/integrations/resque/job_metrics.rb

@@ -77,7 +77,7 @@ def start(client:)
             # retries-with-backoff don't show the intentional wait as latency),
             # falling back to enqueued_at.
             #
-            # @param [JobPayload] payload
+            # @param [ActiveJobPayload, VanillaResquePayload] payload
             #
             def record_queue_latency(payload)
               anchor = payload.anchor_time
@@ -105,7 +105,7 @@ def record_queue_latency(payload)
             # from the `Resque::Worker#perform_with_fork` prepend, so it measures
             # the full child lifetime (fork + reconnect + perform + exit).
             #
-            # @param [JobPayload] payload
+            # @param [ActiveJobPayload, VanillaResquePayload] payload
             # @param [Float] duration in seconds
             #
             def record_perform_duration(payload, duration)
@@ -134,13 +134,14 @@ def install_hooks
           ##
           # Prepended onto Resque::Worker to capture both queue latency
           # (before super) and perform duration (after super) for every
-          # job that goes through perform_with_fork. JobPayload is built
-          # once per job and shared between the two recordings.
+          # job that goes through perform_with_fork. The payload object is
+          # built once per job via JobPayload.for and shared between the
+          # two recordings.
           #
           module WorkerInstrumentation
             def perform_with_fork(job, &block)
               started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
-              payload    = JobPayload.new(job)
+              payload    = JobPayload.for(job)
               JobMetrics.record_queue_latency(payload)
               super
             ensure

lib/bigcommerce/prometheus/integrations/resque/job_payload.rb

@@ -15,79 +15,48 @@
 # 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 'time'
-
 module Bigcommerce
   module Prometheus
     module Integrations
       class Resque
         ##
-        # Extracts the fields per-job metrics need from a Resque::Job's
-        # payload. Eagerly parses in #initialize and exposes plain
-        # attr_readers — does not hold a reference to the Resque::Job after
-        # construction.
+        # Classifies a Resque::Job's payload and builds the matching
+        # shape-specific payload object for per-job metrics.
         #
-        # See JobMetrics's class-level docs for the ActiveJob-shaped payload
-        # contract this class consumes.
+        # A payload is ActiveJob-shaped when `args[0]` is a Hash carrying a
+        # truthy 'job_class' — the shape
+        # ActiveJob::QueueAdapters::ResqueAdapter::JobWrapper produces
+        # natively, and which service-local wrappers can produce
+        # deliberately. Everything else — vanilla Resque jobs with primitive
+        # args, nil or non-Hash payloads, `args` not being an Array — is
+        # treated as vanilla.
         #
-        class JobPayload
-          # @return [String] the user's actual job class name. For
-          #   ActiveJob-shaped payloads this is `args[0]['job_class']`.
-          #   Falls back to the top-level Resque `class` field for vanilla
-          #   Resque payloads. `'unknown'` if neither is available, or if
-          #   the payload is malformed (nil, non-Hash, etc.).
-          attr_reader :job_class
-
-          # @return [Time, nil] the queue-latency anchor. Prefers
-          #   `scheduled_at` over `enqueued_at` (so retries-with-backoff
-          #   don't count the intentional wait). nil for non-ActiveJob-shaped
-          #   payloads (where `args[0]` isn't a Hash), for payloads where
-          #   both timestamps are absent or unparseable, and for malformed
-          #   payloads.
-          attr_reader :anchor_time
-
-          # @param [Resque::Job] resque_job
-          def initialize(resque_job)
-            payload = resque_job.payload || {}
-            inner   = extract_activejob_inner(payload)
-
-            @job_class   = extract_job_class(payload, inner)
-            @anchor_time = extract_anchor_time(inner)
-          end
-
-          private
-
-          # Returns the inner ActiveJob-shaped payload (`args[0]` if it's a
-          # Hash), or nil for any payload shape that isn't ActiveJob-wrapped
-          # (vanilla Resque jobs with primitive args, non-Hash payloads,
-          # `args` not being an Array, etc.).
-          def extract_activejob_inner(payload)
-            return nil unless payload.is_a?(Hash)
-
-            args  = payload['args']
-            first = args.is_a?(Array) ? args.first : nil
-            first.is_a?(Hash) ? first : nil
-          end
-
-          def extract_job_class(payload, inner)
-            inner_class = inner.is_a?(Hash) ? inner['job_class'] : nil
-            outer_class = payload.is_a?(Hash) ? payload['class'] : nil
-            inner_class || outer_class || 'unknown'
-          end
+        # Both payload classes expose the same interface: #job_class
+        # (String) and #anchor_time (Time or nil).
+        #
+        module JobPayload
+          class << self
+            ##
+            # @param [Resque::Job] resque_job
+            # @return [ActiveJobPayload, VanillaResquePayload]
+            #
+            def for(resque_job)
+              payload = resque_job.payload
+              payload = {} unless payload.is_a?(Hash)
 
-          def extract_anchor_time(inner)
-            return nil unless inner.is_a?(Hash)
+              inner = activejob_inner(payload)
+              inner ? ActiveJobPayload.new(inner) : VanillaResquePayload.new(payload)
+            end
 
-            parse_time(inner['scheduled_at']) || parse_time(inner['enqueued_at'])
-          end
+            private
 
-          def parse_time(value)
-            return value if value.is_a?(Time)
-            return nil if value.nil? || value.to_s.empty?
+            def activejob_inner(payload)
+              args  = payload['args']
+              first = args.is_a?(Array) ? args.first : nil
+              return nil unless first.is_a?(Hash) && first['job_class']
 
-            Time.iso8601(value.to_s)
-          rescue ArgumentError
-            nil
+              first
+            end
           end
         end
       end

lib/bigcommerce/prometheus/integrations/resque/vanilla_resque_payload.rb

@@ -0,0 +1,49 @@
+# 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
+        ##
+        # Payload fields for a vanilla Resque job (`Resque.enqueue(MyJob,
+        # ...)`) — the top-level 'class' field is the real job class and the
+        # args are raw positional values. Vanilla payloads carry no enqueue
+        # timestamps, so there is never a queue-latency anchor: queue_latency
+        # no-ops for these jobs by construction. perform_duration is
+        # unaffected.
+        #
+        class VanillaResquePayload
+          # @return [String] the top-level Resque payload class, or
+          #   'unknown' for malformed payloads
+          attr_reader :job_class
+
+          # @param [Hash] payload the raw Resque payload (normalized to a
+          #   Hash by JobPayload.for)
+          def initialize(payload)
+            @job_class = payload['class'] || 'unknown'
+          end
+
+          # @return [nil] vanilla Resque payloads have no enqueue timestamps
+          def anchor_time
+            nil
+          end
+        end
+      end
+    end
+  end
+end

spec/bigcommerce/prometheus/integrations/resque/active_job_payload_spec.rb

@@ -0,0 +1,87 @@
+# 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::ActiveJobPayload do
+  # Mirrors the inner hash ActiveJob's JobWrapper stamps into args[0].
+  # JobPayload.for guarantees a truthy job_class before construction.
+  def inner_payload(job_class: 'MyJob', enqueued_at: nil, scheduled_at: nil)
+    {
+      'job_class' => job_class,
+      'arguments' => [],
+      'enqueued_at' => enqueued_at,
+      'scheduled_at' => scheduled_at
+    }.compact
+  end
+
+  describe '#job_class' do
+    it 'returns the inner job_class' do
+      expect(described_class.new(inner_payload(job_class: 'BigPay::SomePublishJob')).job_class)
+        .to eq('BigPay::SomePublishJob')
+    end
+  end
+
+  describe '#anchor_time' do
+    it 'prefers scheduled_at when both are present (excludes the intentional backoff)' do
+      enqueued = (Time.now - 30).iso8601(6)
+      scheduled = (Time.now - 1).iso8601(6)
+
+      anchor = described_class.new(inner_payload(enqueued_at: enqueued, scheduled_at: scheduled)).anchor_time
+
+      expect(anchor).to be_within(0.5).of(Time.now - 1)
+    end
+
+    it 'falls back to enqueued_at when scheduled_at is absent' do
+      enqueued = (Time.now - 3).iso8601(6)
+
+      anchor = described_class.new(inner_payload(enqueued_at: enqueued)).anchor_time
+
+      expect(anchor).to be_within(0.5).of(Time.now - 3)
+    end
+
+    it 'returns nil when neither timestamp is present' do
+      expect(described_class.new(inner_payload).anchor_time).to be_nil
+    end
+
+    it 'handles a Time instance directly in the payload' do
+      time = Time.now - 2
+
+      expect(described_class.new(inner_payload(enqueued_at: time)).anchor_time).to eq(time)
+    end
+
+    it 'returns nil for an unparseable string (no exception)' do
+      expect do
+        @anchor = described_class.new(inner_payload(enqueued_at: 'not a real time')).anchor_time
+      end.not_to raise_error
+      expect(@anchor).to be_nil
+    end
+
+    it 'returns nil for an empty string' do
+      expect(described_class.new(inner_payload(enqueued_at: '')).anchor_time).to be_nil
+    end
+  end
+
+  describe 'field independence under partial failure' do
+    it 'still returns job_class when the enqueued_at timestamp is unparseable' do
+      result = described_class.new(inner_payload(job_class: 'BigPay::SomeJob', enqueued_at: 'not a real time'))
+
+      expect(result.job_class).to eq('BigPay::SomeJob')
+      expect(result.anchor_time).to be_nil
+    end
+  end
+end

spec/bigcommerce/prometheus/integrations/resque/job_metrics_spec.rb

@@ -50,12 +50,12 @@ def resque_job_double(payload)
     double('Resque::Job', payload: payload)
   end
 
-  # Build a real JobPayload from a payload hash. Both record_* methods take
-  # a JobPayload instance (the WorkerInstrumentation prepend constructs it
-  # once per job and shares it between the two recordings); JobPayload's own
-  # parsing edge cases are covered in job_payload_spec.rb.
+  # Build a real payload object from a payload hash. Both record_* methods
+  # take the payload object the WorkerInstrumentation prepend builds once per
+  # job via JobPayload.for and shares between the two recordings;
+  # classification and parsing edge cases are covered in the payload specs.
   def payload_for(payload_hash)
-    Bigcommerce::Prometheus::Integrations::Resque::JobPayload.new(resque_job_double(payload_hash))
+    Bigcommerce::Prometheus::Integrations::Resque::JobPayload.for(resque_job_double(payload_hash))
   end
 
   def active_job_payload(job_class: 'MyJob', enqueued_at: nil, scheduled_at: nil)