Repository navigation
feat(crons): Add Solid Queue recurring task monitoring #3140
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
wedamija
wants to merge
5
commits into
master
Choose a base branch
from
danf/solid-queue-crons
base: master
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
+342
−4
Open
Changes from all commits
Commits
Show all changes
5 commits
Select commit
Hold shift + click to select a range
764bd5b
feat(crons): Add Solid Queue recurring task monitoring
wedamija 89777ce
feat(crons): Add an exclude list for Solid Queue recurring tasks
wedamija 8c3299f
test(crons): Use a sub-minute schedule for the no-config Solid Queue …
wedamija 8f83d6a
fix(rails): Don't let a bad Solid Queue exclusion list block enqueues
wedamija 2a7158c
ref(crons): Use the shared slug helper for Solid Queue
wedamija File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| # frozen_string_literal: true | ||
|
|
||
| module Sentry | ||
| module Rails | ||
| # Sends Crons check-ins for Solid Queue recurring tasks. | ||
| # Enable with `config.enabled_patches << :solid_queue_recurring`. | ||
| module SolidQueueRecurring | ||
| PAYLOAD_KEY = "_sentry_recurring_task" | ||
| CURRENT_TASK = :sentry_solid_queue_recurring_task | ||
|
|
||
| module RecurringTask | ||
| def enqueue(at:) | ||
| Thread.current[CURRENT_TASK] = self unless Sentry::Rails::SolidQueueRecurring.excluded?(self) | ||
| super | ||
| ensure | ||
| Thread.current[CURRENT_TASK] = nil | ||
| end | ||
| end | ||
|
|
||
| module Job | ||
| def serialize | ||
| @_sentry_recurring_task ||= Sentry::Rails::SolidQueueRecurring.monitor_for(Thread.current[CURRENT_TASK], self) | ||
| payload = super | ||
| payload[PAYLOAD_KEY] = @_sentry_recurring_task if @_sentry_recurring_task | ||
| payload | ||
| end | ||
|
|
||
| def deserialize(job_data) | ||
| super | ||
| @_sentry_recurring_task = job_data[PAYLOAD_KEY] | ||
| end | ||
| end | ||
|
|
||
| class << self | ||
| def excluded?(task) | ||
| return false unless Sentry.initialized? | ||
|
|
||
| Array(Sentry.configuration.rails.solid_queue_exclude_recurring_tasks).any? { |pattern| pattern === task.key.to_s } | ||
| rescue StandardError => e | ||
| Sentry.sdk_logger.error(Sentry::LOGGER_PROGNAME) { "Could not check solid_queue_exclude_recurring_tasks for #{task.key}: #{e.class}: #{e.message}" } | ||
| false | ||
| end | ||
|
|
||
| # Runs in the scheduler, which is where Solid Queue evaluates the schedule. | ||
| def monitor_for(task, job) | ||
| return unless task && Sentry.initialized? | ||
| job_class = task.class_name.present? ? task.class_name.safe_constantize : task.class.default_job_class | ||
| return unless job.instance_of?(job_class) | ||
| return if job_class < Sentry::Cron::MonitorCheckIns::Patch | ||
|
|
||
| slug = Sentry::Cron::Slug.from(task.key) | ||
| return unless slug | ||
|
|
||
| config = Sentry::Cron::FugitSchedule.from_cron(cron_for(task)) | ||
| { "slug" => slug, "crontab" => config&.schedule&.value, "timezone" => config&.timezone }.compact | ||
| rescue StandardError => e | ||
| Sentry.sdk_logger.error(Sentry::LOGGER_PROGNAME) { "Not monitoring Solid Queue recurring task #{task.key}: #{e.class}: #{e.message}" } | ||
| nil | ||
| end | ||
|
|
||
| def perform(job) | ||
| data = job.instance_variable_get(:@_sentry_recurring_task) | ||
| return yield unless data && Sentry.initialized? | ||
|
|
||
| slug = data["slug"] | ||
| if data["crontab"] | ||
| cron_config = Sentry.configuration.cron | ||
| monitor_config = Sentry::Cron::MonitorConfig.from_crontab( | ||
| data["crontab"], | ||
| timezone: data["timezone"], | ||
| checkin_margin: cron_config.default_checkin_margin, | ||
| max_runtime: cron_config.default_max_runtime | ||
| ) | ||
| end | ||
|
|
||
| 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 | ||
|
|
||
| def patch | ||
| return if @patched | ||
|
|
||
| @patched = true | ||
| ActiveSupport.on_load(:active_job) do | ||
| prepend Sentry::Rails::SolidQueueRecurring::Job | ||
| around_perform { |job, block| Sentry::Rails::SolidQueueRecurring.perform(job, &block) } | ||
| end | ||
|
|
||
| apply = proc do | ||
| if defined?(::SolidQueue::RecurringTask) && !(::SolidQueue::RecurringTask < Sentry::Rails::SolidQueueRecurring::RecurringTask) | ||
| ::SolidQueue::RecurringTask.prepend(Sentry::Rails::SolidQueueRecurring::RecurringTask) | ||
| end | ||
| end | ||
| ::Rails.application.reloader.to_prepare(&apply) | ||
| apply.call if ::Rails.application.initialized? | ||
| end | ||
|
|
||
| private | ||
|
|
||
| # Mirrors RecurringTask, which applies SolidQueue.time_zone to zoneless schedules. | ||
| def cron_for(task) | ||
| cron = ::Fugit.parse(task.schedule) | ||
| zone = ::SolidQueue.time_zone if ::SolidQueue.respond_to?(:time_zone) | ||
| return cron unless cron.is_a?(::Fugit::Cron) && cron.zone.nil? && zone.present? | ||
|
|
||
| ::Fugit.parse("#{cron.to_cron_s} #{zone}") | ||
| end | ||
| end | ||
| end | ||
| end | ||
| end | ||
|
|
||
| Sentry.register_patch(:solid_queue_recurring) do |config| | ||
| if defined?(::SolidQueue) && defined?(::Rails.application) && ::Rails.application | ||
| Sentry::Rails::SolidQueueRecurring.patch | ||
| else | ||
| config.sdk_logger.warn(Sentry::LOGGER_PROGNAME) { "You tried to enable the Solid Queue recurring task integration but the `solid_queue` gem was not detected." } | ||
| end | ||
| end | ||
204 changes: 204 additions & 0 deletions
204
sentry-rails/spec/active_job/solid_queue_recurring_spec.rb
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,204 @@ | ||
| # frozen_string_literal: true | ||
|
|
||
| require "spec_helper" | ||
|
|
||
| if RAILS_VERSION >= 7.1 && RUBY_VERSION >= "3.1" | ||
| require "solid_queue" | ||
|
|
||
| SOLID_QUEUE_RECURRING_ADAPTER = ::ActiveJob::QueueAdapters::SolidQueueAdapter.new | ||
|
|
||
| RSpec.describe Sentry::Rails::SolidQueueRecurring, type: :job do | ||
| include_context "active_job backend harness", adapter: :solid_queue | ||
|
|
||
| let(:excluded_tasks) { [] } | ||
| let(:configure_sentry) do | ||
| proc do |config| | ||
| config.enabled_patches << :solid_queue_recurring | ||
| config.rails.solid_queue_exclude_recurring_tasks = excluded_tasks | ||
| end | ||
| end | ||
|
|
||
| def queue_adapter_for_test | ||
| SOLID_QUEUE_RECURRING_ADAPTER | ||
| end | ||
|
|
||
| def boot_adapter(_adapter) | ||
| Sentry::Rails::Test::Application.load_queue_schema | ||
| end | ||
|
|
||
| def reset_adapter(_adapter) | ||
| [ | ||
| SolidQueue::ReadyExecution, | ||
| SolidQueue::ClaimedExecution, | ||
| SolidQueue::FailedExecution, | ||
| SolidQueue::ScheduledExecution, | ||
| SolidQueue::RecurringExecution, | ||
| SolidQueue::Process, | ||
| SolidQueue::Job | ||
| ].each(&:delete_all) | ||
| end | ||
|
|
||
| def drain | ||
| process = SolidQueue::Process.register(kind: "Worker", pid: ::Process.pid, name: "spec-#{SecureRandom.hex(4)}") | ||
| SolidQueue::ReadyExecution.claim("*", 100, process.id).each(&:perform) | ||
| end | ||
|
|
||
| def run_task(key, **options) | ||
| SolidQueue::RecurringTask.from_configuration(key, **options).enqueue(at: Time.now.change(usec: 0)) | ||
| drain | ||
| end | ||
|
|
||
| def check_ins | ||
| sentry_events.select { |event| event.is_a?(Sentry::CheckInEvent) } | ||
| end | ||
|
|
||
| let(:report_job) do | ||
| job_fixture("ReportJob") do | ||
| def perform(*); end | ||
| end | ||
| end | ||
|
|
||
| before { report_job } | ||
|
|
||
| it "sends in_progress and ok check-ins with the task key and schedule" do | ||
| run_task("nightly_report", class: "ReportJob", schedule: "0 9 * * 1-5 Europe/Vienna") | ||
|
|
||
| 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 error check-in when the job fails" do | ||
| job_fixture("FailingReportJob") do | ||
| def perform | ||
| raise "boom" | ||
| end | ||
| end | ||
|
|
||
| expect { run_task("failing_report", class: "FailingReportJob", schedule: "every hour") }.to raise_error("boom") | ||
| expect(check_ins.map(&:status)).to eq([:in_progress, :error]) | ||
| expect(check_ins.first.monitor_config.to_h[:schedule]).to eq(type: :crontab, value: "0 * * * *") | ||
| end | ||
|
|
||
| it "uses each task's key and schedule when tasks share a job class" do | ||
| run_task("report_eu", class: "ReportJob", args: ["eu"], schedule: "0 9 * * * Europe/Berlin") | ||
| run_task("report_us", class: "ReportJob", args: ["us"], schedule: "0 9 * * * America/New_York") | ||
|
|
||
| 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 "monitors command tasks" do | ||
| run_task("Cleanup Cache", command: "1 + 1", schedule: "*/15 * * * * UTC") | ||
|
|
||
| expect(check_ins.map(&:monitor_slug)).to eq(["cleanup-cache", "cleanup-cache"]) | ||
| expect(check_ins.first.monitor_config.to_h[:schedule]).to eq(type: :crontab, value: "*/15 * * * *") | ||
| end | ||
|
|
||
| it "sends check-ins without a config for schedules Crons cannot express" do | ||
| run_task("every_ten_seconds", class: "ReportJob", schedule: "every 10 seconds") | ||
|
|
||
| expect(check_ins.map(&:status)).to eq([:in_progress, :ok]) | ||
| expect(check_ins.map(&:monitor_config)).to all(be_nil) | ||
| end | ||
|
|
||
| it "applies SolidQueue.time_zone to schedules without a zone" do | ||
| skip "SolidQueue.time_zone is not supported" unless SolidQueue.respond_to?(:time_zone=) | ||
|
|
||
| original = SolidQueue.time_zone | ||
| SolidQueue.time_zone = "Tokyo" | ||
| run_task("tokyo_report", class: "ReportJob", schedule: "0 9 * * *") | ||
|
|
||
| expect(check_ins.first.monitor_config.timezone).to eq("Asia/Tokyo") | ||
| ensure | ||
| SolidQueue.time_zone = original if SolidQueue.respond_to?(:time_zone=) | ||
| end | ||
|
|
||
| it "leaves jobs that include Sentry::Cron::MonitorCheckIns alone" do | ||
| job_fixture("MonitoredJob") do | ||
| include Sentry::Cron::MonitorCheckIns | ||
| sentry_monitor_check_ins slug: "own-slug" | ||
|
|
||
| def perform; end | ||
| end | ||
|
|
||
| run_task("monitored", class: "MonitoredJob", schedule: "0 9 * * *") | ||
|
|
||
| expect(check_ins.map(&:monitor_slug)).to eq(["own-slug", "own-slug"]) | ||
| end | ||
|
|
||
| it "monitors jobs that include Sentry::Cron::MonitorCheckIns without configuring it" do | ||
| job_fixture("UnconfiguredJob") do | ||
| include Sentry::Cron::MonitorCheckIns | ||
|
|
||
| def perform; end | ||
| end | ||
|
|
||
| run_task("unconfigured", class: "::UnconfiguredJob", schedule: "0 9 * * *") | ||
|
|
||
| expect(check_ins.map(&:monitor_slug)).to eq(["unconfigured", "unconfigured"]) | ||
| end | ||
|
|
||
| it "excludes nothing by default" do | ||
| expect(Sentry.configuration.rails.solid_queue_exclude_recurring_tasks).to eq([]) | ||
| end | ||
|
|
||
| context "with excluded tasks" do | ||
| let(:excluded_tasks) { ["clear_cache", /\Atmp_/] } | ||
|
|
||
| it "skips tasks whose key matches a string" do | ||
| run_task("clear_cache", class: "ReportJob", schedule: "0 9 * * *") | ||
|
|
||
| expect(check_ins).to be_empty | ||
| end | ||
|
|
||
| it "skips tasks whose key matches a regexp" do | ||
| run_task("tmp_cleanup", class: "ReportJob", schedule: "0 9 * * *") | ||
|
|
||
| expect(check_ins).to be_empty | ||
| end | ||
|
|
||
| it "monitors tasks that do not match" do | ||
| run_task("clear_cache_daily", class: "ReportJob", schedule: "0 9 * * *") | ||
|
|
||
| expect(check_ins.map(&:monitor_slug)).to eq(["clear_cache_daily", "clear_cache_daily"]) | ||
| end | ||
| end | ||
|
|
||
| context "with no excluded tasks set" do | ||
| let(:excluded_tasks) { nil } | ||
|
|
||
| it "monitors the task" do | ||
| run_task("daily_report", class: "ReportJob", schedule: "0 9 * * *") | ||
|
|
||
| expect(check_ins.map(&:monitor_slug)).to eq(["daily_report", "daily_report"]) | ||
| end | ||
| end | ||
|
|
||
| context "with an exclusion pattern that raises" do | ||
| let(:excluded_tasks) { [->(_key) { raise "boom" }] } | ||
|
|
||
| it "still enqueues and monitors the task" do | ||
| run_task("daily_report", class: "ReportJob", schedule: "0 9 * * *") | ||
|
|
||
| expect(check_ins.map(&:monitor_slug)).to eq(["daily_report", "daily_report"]) | ||
| end | ||
| end | ||
|
|
||
| it "does not monitor jobs enqueued outside a recurring task" do | ||
| report_job.perform_later | ||
| drain | ||
|
|
||
| expect(check_ins).to be_empty | ||
| end | ||
| end | ||
| end |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
RecurringTask patch skipped without autoload
High Severity
The RecurringTask prepend is applied only when
defined?(::SolidQueue::RecurringTask)is already true. That check does not trigger Zeitwerk, and the class lives in the engine'sapp/models, so it is usually not loaded atto_preparetime unlesseager_loadis on. In the Solid Queue scheduler under the default development setup,enqueueis never wrapped, so slug and schedule never reach the job payload and no check-ins are sent.Reviewed by Cursor Bugbot for commit 2a7158c. Configure here.