Merge pull request #33 from babylist/middleware-use-job-payload-instead-of-worker-instance

Use job payload instead of worker instance in middleware

Author: aflanagan

lib/sidekiq/cronitor.rb

@@ -13,72 +13,63 @@
 
 module Sidekiq::Cronitor
   class ServerMiddleware
-    def call(worker, message, queue)
-      ping(worker: worker, state: 'run')
+    # @param [Object] worker the instance of the job that was queued
+    # @param [Hash] job_payload the full job payload
+    #   * @see https://github.com/sidekiq/sidekiq/wiki/Job-Format
+    # @param [String] queue the name of the queue the job was pulled from
+    # @yield the next middleware in the chain or worker `perform` method
+    def call(worker, job_payload, queue)
+      ping(job_payload: job_payload, state: 'run')
 
       result = yield
     rescue => e
-      ping(worker: worker, state: 'fail', message: e.to_s)
+      ping(job_payload: job_payload, state: 'fail', message: e.to_s)
 
       raise e
     else
-      ping(worker: worker, state: 'complete')
+      ping(job_payload: job_payload, state: 'complete')
       result # to be consistent with client middleware, return results of yield
     end
 
     private
 
-    def cronitor(worker)
-      Cronitor::Monitor.new(job_key(worker))
+    def cronitor(job_payload)
+      Cronitor::Monitor.new(job_key(job_payload))
     end
 
-    def cronitor_disabled?(worker)
-      if worker.class.sidekiq_options.has_key?("cronitor_enabled")
-        !worker.class.sidekiq_options.fetch("cronitor_enabled", Cronitor.auto_discover_sidekiq)
+    def cronitor_disabled?(job_payload)
+      if job_payload.has_key?("cronitor_enabled")
+        !job_payload.fetch("cronitor_enabled", Cronitor.auto_discover_sidekiq)
       else
-        worker.class.sidekiq_options.fetch("cronitor_disabled", options(worker).fetch(:disabled, !Cronitor.auto_discover_sidekiq))
+        job_payload.fetch("cronitor_disabled", options(job_payload).fetch("disabled", !Cronitor.auto_discover_sidekiq))
       end
     end
 
-    def job_key(worker)
-      periodic_job_key(worker) || worker.class.sidekiq_options.fetch('cronitor_key', nil) ||
-        options(worker).fetch(:key, worker.class.name)
+    def job_key(job_payload)
+      job_payload['cronitor_key'] || options(job_payload)['key'] || job_payload['class']
     end
 
-    def periodic_job_key(worker)
-      return unless defined?(Sidekiq::Periodic)
-
-      periodic_job = Sidekiq::Periodic::LoopSet.new.find do |lop|
-        lop.history.find { |j| j[0] == worker.jid }
-      end
-
-      periodic_job.present? && periodic_job.options.fetch('cronitor_key', nil)
-    end
-
-    def options(worker)
+    def options(job_payload)
       # eventually we will delete this method of passing options
       # ultimately we want all cronitor options to be top level keys
-      opts = worker.class.sidekiq_options.fetch("cronitor", {})
-      # symbolize_keys is a rails helper, so only use it if it's defined
-      opts = opts.symbolize_keys if opts.respond_to?(:symbolize_keys)
-      opts
+      job_payload.fetch("cronitor", {})
     end
 
-    def ping(worker:, state:, message: nil)
-      return unless should_ping?(worker)
+    def ping(job_payload:, state:, message: nil)
+      return unless should_ping?(job_payload)
 
-      Sidekiq.logger.debug("[cronitor] ping: worker=#{job_key(worker)} state=#{state} message=#{message}")
+      Sidekiq.logger.debug("[cronitor] ping: worker=#{job_key(job_payload)} state=#{state} message=#{message}")
 
-      cronitor(worker).ping(state: state, message: message)
+      cronitor(job_payload).ping(state: state, message: message)
     rescue Cronitor::Error => e
-      Sidekiq.logger.error("[cronitor] error during ping: worker=#{job_key(worker)} error=#{e.message}")
+      Sidekiq.logger.error("[cronitor] error during ping: worker=#{job_key(job_payload)} error=#{e.message}")
     rescue => e
-      Sidekiq.logger.error("[cronitor] unexpected error: worker=#{job_key(worker)} error=#{e.message}")
+      Sidekiq.logger.error("[cronitor] unexpected error: worker=#{job_key(job_payload)} error=#{e.message}")
       Sidekiq.logger.error(e.backtrace.first)
     end
 
-    def should_ping?(worker)
-      !cronitor(worker).api_key.nil? && !cronitor_disabled?(worker)
+    def should_ping?(job_payload)
+      !cronitor(job_payload).api_key.nil? && !cronitor_disabled?(job_payload)
     end
   end
 end

lib/sidekiq/cronitor/periodic_jobs.rb

@@ -6,7 +6,7 @@ def self.sync_schedule!
       monitors_payload = []
       loops = Sidekiq::Periodic::LoopSet.new
       loops.each do |lop|
-        if lop.options.has_key?('cronitor_enabled') || lop.klass.constantize.sidekiq_options.has_key?('cronitor_enabled')
+        if parsed_options(lop).has_key?('cronitor_enabled') || Object.const_get(lop.klass).sidekiq_options.has_key?('cronitor_enabled')
           next unless fetch_option(lop, 'cronitor_enabled', Cronitor.auto_discover_sidekiq)
         else
           next if fetch_option(lop, 'cronitor_disabled', !Cronitor.auto_discover_sidekiq)
@@ -29,9 +29,17 @@ def self.sync_schedule!
       Sidekiq.logger.error("[cronitor] error during #{name}.#{__method__}: #{e}")
     end
 
+    def self.parsed_options(lop)
+      if lop.options.is_a?(String)
+        JSON.parse(lop.options)
+      else
+        lop.options
+      end
+    end
+
     def self.fetch_option(lop, key, default = nil)
-      lop.options.fetch(key, default) ||
-        lop.klass.constantize.sidekiq_options.fetch(key, default)
+      parsed_options(lop).fetch(key, default) ||
+        Object.const_get(lop.klass).sidekiq_options.fetch(key, default)
     end
   end
 end

spec/sidekiq/cronitor/periodic_jobs_spec.rb

@@ -2,12 +2,70 @@
 
 require 'sidekiq/cronitor/periodic_jobs'
 
+class DummyWorker
+  def self.sidekiq_options=(options)
+    @sidekiq_options ||= {}
+    @sidekiq_options.merge!(options)
+  end
+  def self.sidekiq_options
+    @sidekiq_options ||= {}
+  end
+  def self.reset_options
+    @sidekiq_options = {}
+  end
+end
+
 RSpec.describe Sidekiq::Cronitor::PeriodicJobs do
+  let(:cronitor_key) { "dummy-worker" }
+  let(:loops) { [job] }
+
   before do
-    allow(Sidekiq::Periodic::LoopSet).to receive(:new).and_return([])
+    class_double("Sidekiq::Periodic::LoopSet").as_stubbed_const
+    allow(Sidekiq::Periodic::LoopSet).to receive(:new).and_return(loops)
   end
 
   describe '.sync_schedule!' do
-    xit { expect { described_class.sync_schedule! }.not_to raise_error }
+    context "when no loops" do
+      let(:loops) { [] }
+      it { expect { described_class.sync_schedule! }.not_to raise_error }
+    end
+
+    context "when job options are stringified JSON" do
+      let(:job) {
+        instance_double(
+          "Sidekiq::Periodic::Loop",
+          klass: DummyWorker.name,
+          schedule: "* * * * *",
+          tz_name: "Etc/UTC",
+          options: {
+            "cronitor_key" => cronitor_key,
+            "cronitor_group" => "dummy-team"
+          }.to_json
+        )
+      }
+      it "updates monitors" do
+        expect(Cronitor::Monitor).to receive(:put).with(monitors: [hash_including(key: cronitor_key)])
+        described_class.sync_schedule!
+      end
+    end
+
+    context "when job options are a hash" do
+      let(:job) {
+        instance_double(
+          "Sidekiq::Periodic::Loop",
+          klass: DummyWorker.name,
+          schedule: "* * * * *",
+          tz_name: "Etc/UTC",
+          options: {
+            "cronitor_key" => cronitor_key,
+            "cronitor_group" => "dummy-team"
+          }
+        )
+      }
+      it "updates monitors" do
+        expect(Cronitor::Monitor).to receive(:put).with(monitors: [hash_including(key: cronitor_key)])
+        described_class.sync_schedule!
+      end
+    end
   end
 end