feat(crons): Add resque-scheduler monitoring

sentry-resque/Gemfile

@@ -11,6 +11,7 @@ gemspec
 gem "sentry-ruby", path: "../sentry-ruby"
 
 gem "resque-retry", "~> 1.8"
+gem "resque-scheduler"
 
 gem "sentry-rails", path: "../sentry-rails"
 gem "rails", "> 5.2.0"

sentry-resque/lib/sentry-resque.rb

@@ -6,6 +6,7 @@
 require "sentry/resque/configuration"
 require "sentry/resque/version"
 require "sentry/resque"
+require "sentry/resque/scheduler"
 
 module Sentry
   module Resque

sentry-resque/lib/sentry/resque/configuration.rb

@@ -15,8 +15,14 @@ class Configuration
       # retry if it fails.
       attr_accessor :report_after_job_retries
 
+      # resque-scheduler schedule names to skip when the :resque_scheduler patch is enabled.
+      # Strings match the name exactly; Regexps are matched against it.
+      # @return [Array<String, Regexp>]
+      attr_accessor :exclude_scheduled_jobs
+
       def initialize
         @report_after_job_retries = false
+        @exclude_scheduled_jobs = []
       end
     end
   end

sentry-resque/lib/sentry/resque/scheduler.rb

@@ -0,0 +1,133 @@
+# frozen_string_literal: true
+
+module Sentry
+  module Resque
+    # Sends Crons check-ins for resque-scheduler `cron` and `every` schedules.
+    # Enable with `config.enabled_patches << :resque_scheduler`.
+    module Scheduler
+      PAYLOAD_KEY = "sentry_monitor"
+      CURRENT_MONITOR = :sentry_resque_scheduler_monitor
+
+      module SchedulerPatch
+        def load_schedule_job(name, config)
+          ret = super
+          # Monitored here, so the :rufus_scheduler patch skips them.
+          job = scheduled_jobs && scheduled_jobs[name]
+          job.opts[:sentry_monitor] = false if job.respond_to?(:opts) && job.opts.is_a?(Hash)
+          ret
+        end
+
+        private
+
+        def enqueue_recurring(name, config)
+          Thread.current[CURRENT_MONITOR] = Sentry::Resque::Scheduler.monitor_for(name, scheduled_jobs && scheduled_jobs[name])
+          super
+        ensure
+          Thread.current[CURRENT_MONITOR] = nil
+        end
+      end
+
+      module Push
+        def push(queue, item)
+          monitor = Thread.current[CURRENT_MONITOR]
+          if monitor && item.is_a?(Hash)
+            # Only the scheduled job, not jobs pushed by its enqueue hooks.
+            Thread.current[CURRENT_MONITOR] = nil
+            item = item.merge(PAYLOAD_KEY => monitor)
+          end
+          super(queue, item)
+        end
+      end
+
+      module Job
+        def perform
+          data = payload[PAYLOAD_KEY] if payload.is_a?(Hash)
+          return super unless data.is_a?(Hash) && Sentry.initialized?
+
+          Sentry::Resque::Scheduler.check_in(data) { super }
+        end
+      end
+
+      class << self
+        def excluded?(name)
+          Array(Sentry.configuration.resque.exclude_scheduled_jobs).any? { |pattern| pattern === name.to_s }
+        rescue StandardError => e
+          Sentry.sdk_logger.error(Sentry::LOGGER_PROGNAME) { "Could not check exclude_scheduled_jobs for #{name}: #{e.class}: #{e.message}" }
+          false
+        end
+
+        # Runs in the scheduler, which is where rufus-scheduler evaluates the schedule.
+        def monitor_for(name, rufus_job)
+          return unless rufus_job && Sentry.initialized?
+          return if excluded?(name)
+
+          slug = name.to_s.downcase.gsub(/[^a-z0-9_-]+/, "-").gsub(/\A-+|-+\z/, "")
+          slug = slug[0, Sentry::Cron::MonitorCheckIns::MAX_SLUG_LENGTH].delete_suffix("-")
+          return if slug.empty?
+
+          config =
+            case rufus_job
+            when ::Rufus::Scheduler::CronJob then Sentry::Cron::FugitSchedule.from_cron(rufus_job.cron_line)
+            when ::Rufus::Scheduler::EveryJob then Sentry::Cron::FugitSchedule.from_seconds(rufus_job.frequency)
+            end
+          { "slug" => slug, "schedule" => config&.schedule&.to_h, "timezone" => config&.timezone }.compact
+        rescue StandardError => e
+          Sentry.sdk_logger.error(Sentry::LOGGER_PROGNAME) { "Not monitoring resque-scheduler job #{name}: #{e.class}: #{e.message}" }
+          nil
+        end
+
+        def check_in(data)
+          slug = data["slug"]
+          monitor_config = monitor_config_for(data)
+          check_in_id = Sentry.capture_check_in(slug, :in_progress, monitor_config: monitor_config)
+          start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+          begin
+            ret = yield
+          rescue Exception
+            duration = Process.clock_gettime(Process::CLOCK_MONOTONIC) - start
+            Sentry.capture_check_in(slug, :error, check_in_id: check_in_id, duration: duration, monitor_config: monitor_config)
+            raise
+          end
+
+          duration = Process.clock_gettime(Process::CLOCK_MONOTONIC) - start
+          Sentry.capture_check_in(slug, :ok, check_in_id: check_in_id, duration: duration, monitor_config: monitor_config)
+          ret
+        end
+
+        private
+
+        def monitor_config_for(data)
+          schedule = data["schedule"]
+          return unless schedule.is_a?(Hash)
+
+          cron_config = Sentry.configuration.cron
+          options = { checkin_margin: cron_config.default_checkin_margin, max_runtime: cron_config.default_max_runtime }
+          case schedule["type"].to_s
+          when "crontab"
+            Sentry::Cron::MonitorConfig.from_crontab(schedule["value"], timezone: data["timezone"], **options)
+          when "interval"
+            Sentry::Cron::MonitorConfig.from_interval(schedule["value"], schedule["unit"].to_sym, **options)
+          end
+        end
+      end
+    end
+  end
+end
+
+Sentry.register_patch(:resque_scheduler) do |config|
+  begin
+    require "resque/scheduler"
+  rescue LoadError
+  end
+
+  if defined?(::Resque::Scheduler) && defined?(::Rufus::Scheduler::RepeatJob)
+    unless ::Resque::Job < Sentry::Resque::Scheduler::Job
+      ::Resque::Job.prepend(Sentry::Resque::Scheduler::Job)
+      ::Resque.singleton_class.prepend(Sentry::Resque::Scheduler::Push)
+      ::Resque::Scheduler.singleton_class.prepend(Sentry::Resque::Scheduler::SchedulerPatch)
+    end
+  else
+    config.sdk_logger.warn(Sentry::LOGGER_PROGNAME) { "You tried to enable the resque-scheduler integration but the `resque-scheduler` gem was not detected." }
+  end
+end

sentry-resque/spec/sentry/resque/scheduler_spec.rb

@@ -0,0 +1,155 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+
+begin
+  require "resque/scheduler"
+rescue LoadError
+end
+
+if defined?(::Resque::Scheduler)
+  Resque::Scheduler.logger = ::Logger.new(nil)
+
+  class ScheduledReportJob
+    @queue = :default
+
+    def self.perform(*); end
+  end
+
+  class FailingScheduledJob
+    @queue = :default
+
+    def self.perform
+      raise "boom"
+    end
+  end
+
+  RSpec.describe Sentry::Resque::Scheduler do
+    let(:excluded_jobs) { [] }
+
+    before do
+      perform_basic_setup do |config|
+        config.enabled_patches << :resque_scheduler
+        config.resque.exclude_scheduled_jobs = excluded_jobs
+      end
+      allow(Resque::Scheduler).to receive(:am_master).and_return(true)
+      Resque::Scheduler.clear_schedule!
+    end
+
+    after do
+      Resque::Scheduler.clear_schedule!
+    end
+
+    let(:worker) { Resque::Worker.new(:default) }
+
+    def run_schedule(name, config)
+      Resque::Scheduler.load_schedule_job(name, config)
+      Resque::Scheduler.scheduled_jobs.fetch(name).callable.call
+      worker.work_one_job(worker.reserve)
+    end
+
+    def check_ins
+      sentry_events.select { |event| event.is_a?(Sentry::CheckInEvent) }
+    end
+
+    it "sends in_progress and ok check-ins with the schedule name and cron" do
+      run_schedule("nightly_report", "cron" => "0 9 * * 1-5 Europe/Vienna", "class" => "ScheduledReportJob")
+
+      in_progress, ok = check_ins
+      expect(check_ins.size).to eq(2)
+      expect(in_progress.monitor_slug).to eq("nightly_report")
+      expect(in_progress.status).to eq(:in_progress)
+      expect(in_progress.monitor_config.to_h).to eq(
+        schedule: { type: :crontab, value: "0 9 * * 1-5" },
+        timezone: "Europe/Vienna"
+      )
+      expect(ok.status).to eq(:ok)
+      expect(ok.check_in_id).to eq(in_progress.check_in_id)
+    end
+
+    it "sends an interval config for every schedules" do
+      run_schedule("Hourly Sync", "every" => ["2h", { "first_in" => "1h" }], "class" => "ScheduledReportJob")
+
+      expect(check_ins.map(&:monitor_slug)).to eq(["hourly-sync", "hourly-sync"])
+      expect(check_ins.first.monitor_config.to_h[:schedule]).to eq(type: :interval, value: 2, unit: :hour)
+    end
+
+    it "uses the scheduler's local zone for zoneless crons" do
+      original = ENV["TZ"]
+      ENV["TZ"] = "Asia/Tokyo"
+      run_schedule("tokyo_report", "cron" => "0 9 * * *", "class" => "ScheduledReportJob")
+
+      expect(check_ins.first.monitor_config.timezone).to eq("Asia/Tokyo")
+    ensure
+      ENV["TZ"] = original
+    end
+
+    it "sends an error check-in when the job fails" do
+      run_schedule("failing", "cron" => "*/15 * * * * UTC", "class" => "FailingScheduledJob")
+
+      expect(check_ins.map(&:status)).to eq([:in_progress, :error])
+      expect(check_ins.first.monitor_config.to_h[:schedule]).to eq(type: :crontab, value: "*/15 * * * *")
+    end
+
+    it "uses each schedule's name when schedules share a job class" do
+      run_schedule("report_eu", "cron" => "0 9 * * * Europe/Berlin", "class" => "ScheduledReportJob", "args" => ["eu"])
+      run_schedule("report_us", "cron" => "0 9 * * * America/New_York", "class" => "ScheduledReportJob", "args" => ["us"])
+
+      in_progress = check_ins.select { |check_in| check_in.status == :in_progress }
+      expect(in_progress.map { |check_in| [check_in.monitor_slug, check_in.monitor_config.timezone] }).to eq(
+        [["report_eu", "Europe/Berlin"], ["report_us", "America/New_York"]]
+      )
+    end
+
+    it "sends check-ins without a config for schedules Crons cannot express" do
+      run_schedule("every_ten_seconds", "every" => "10s", "class" => "ScheduledReportJob")
+
+      expect(check_ins.map(&:status)).to eq([:in_progress, :ok])
+      expect(check_ins.map(&:monitor_config)).to all(be_nil)
+    end
+
+    it "marks its rufus jobs so the rufus-scheduler patch skips them" do
+      Resque::Scheduler.load_schedule_job("nightly_report", "cron" => "0 9 * * *", "class" => "ScheduledReportJob")
+
+      expect(Resque::Scheduler.scheduled_jobs["nightly_report"].opts[:sentry_monitor]).to eq(false)
+    end
+
+    it "does not monitor jobs enqueued outside a schedule" do
+      Resque.enqueue(ScheduledReportJob)
+      worker.work_one_job(worker.reserve)
+
+      expect(check_ins).to be_empty
+    end
+
+    it "excludes nothing by default" do
+      expect(Sentry.configuration.resque.exclude_scheduled_jobs).to eq([])
+    end
+
+    context "with excluded jobs" do
+      let(:excluded_jobs) { ["clear_cache", /\Atmp_/] }
+
+      it "skips schedules whose name matches a string or regexp" do
+        run_schedule("clear_cache", "cron" => "0 9 * * *", "class" => "ScheduledReportJob")
+        run_schedule("tmp_cleanup", "cron" => "0 9 * * *", "class" => "ScheduledReportJob")
+
+        expect(check_ins).to be_empty
+      end
+
+      it "monitors schedules that do not match" do
+        run_schedule("clear_cache_daily", "cron" => "0 9 * * *", "class" => "ScheduledReportJob")
+
+        expect(check_ins.map(&:monitor_slug)).to eq(["clear_cache_daily", "clear_cache_daily"])
+      end
+    end
+
+    context "with an exclusion pattern that raises" do
+      let(:excluded_jobs) { [->(_name) { raise "boom" }] }
+
+      it "still enqueues and monitors the job" do
+        run_schedule("daily_report", "cron" => "0 9 * * *", "class" => "ScheduledReportJob")
+
+        expect(check_ins.map(&:monitor_slug)).to eq(["daily_report", "daily_report"])
+      end
+    end
+  end
+end