fix: Improve the reliability of alerts & reports

[jfrag1]

SUMMARY

The goal of this PR is to make the alerts & reports scheduler job more reliable. See #25058 for a great description of one current reliability issue.

This PR includes two main changes to improve reliability

Adjusts the center of the window used to check cron schedules.

Currently, the following logic defines the window:

start_at = time_now - timedelta(seconds=1)
stop_at = time_now + timedelta(seconds=window_size) # window_size = 59

The consequence is that if more than a second had elapsed since the scheduler job was originally triggered, any alerts & reports supposed to be triggeredwould not be. This PR moves the center of the window 30 seconds towards the past, giving more leeway for delayed jobs

Reads current time based on when celery beat triggered the scheduler job, rather than having the scheduler job generate the current time.

As I was making the above change, I realized that if the celery queue was backed up, or there was any kind of delay between when celery beat placed the job into the queue and when it was executed, alerts & reports could be missed. It would be preferable if we could simply use the time at which celery beat triggered the job. I found that this was not currently supported natively by celery, but there is a workaround using the expires option. As noted on the celery repo, this is not a perfect solution, but I see it as much more reliable than the current approach.

For example, if a report is supposed to be sent at 12:00, but the celery queue is backed up such that the scheduler job triggered by beat isn't executed until 12:02, this change will make it so that now the report will be triggered, whereas currently it would not be.

Fixes #25058

BEFORE/AFTER SCREENSHOTS OR ANIMATED GIF

TESTING INSTRUCTIONS

ADDITIONAL INFORMATION

• Has associated issue:
• Required feature flags:
• Changes UI
• Includes DB Migration (follow approval process in SIP-59)
  • Migration is atomic, supports rollback & is backwards-compatible
  • Confirm DB migration upgrade and downgrade tested
  • Runtime estimates and downtime expectations provided
• Introduces new feature or API
• Removes existing feature or API

[john-bodley]

Why are we removing the tests which use the zone designator for the zero UTC offset?

[jfrag1]

They're not removed, I just changed the timestamp format to ISO (zero UTC offset represented by +00:00). The format used is inconsequential to the cron_schedule_window function being tested, since it's just passed a datetime object.

[john-bodley]

Is this used elsewhere? If not we should remove the freezegun library.

[jfrag1]

Yes, it's still used elsewhere

[villebro]

I love freezegun, hands off @john-bodley! 😄

[zephyring]

If using triggered_time which could be 2 mins ago, this stop_at(30 seconds) will be preventing the schedule from being executed right? see line 47.

I think we just need to base start_at 1 seconds before the triggered_at. Reasons being cron expression can support second level granularity. If cron job is every second and start at is 30 seconds ago, we will actually execute 30 tasks before the actual start at.

And base stop_at window_size after current_time, although I am not sure why we need that stop_at at all.

[jfrag1]

I don't think the stop_at will prevent anything we want to run from running. Say the job is triggered at 12:00, but run at 12:02, and the cron says the job should run at 12:00. start_at will be ~11:59:30, and stop_at will be ~12:00:30, so it will correctly find the 12:00 schedule since it's in the window.

Also I don't believe cron supports second level granularity, at least not that I've seen in Superset. IMO for Superset's use cases, getting the reliability right is much more important than supporting that level of granularity

[zephyring]

you're right. Cron is at minute level. and stop_at compares to triggered_at 😂

[zephyring]

LGTM

[betodealmeida]

The window approach seems fundamentally brittle to me. Why can't we set start_at to the last time the task was scheduled to run instead, and keep iterating until we hit the current time?

[jfrag1]

I believe this could lead to a situation where an alert/report is triggered more times that desired. For example, if a report runs at 12:00:00, and the celery queue is backed up from 11:55:00-12:00:01, all 5-6 queued scheduler jobs could run one after another, and each would trigger the report.

With the approach on this PR, the windows in this situation would be 11:54:30-11:55:30, 11:55:30-11:56:30, etc., so it should ensure we don't miss any scheduled alerts & reports or run any more times than desired.

Let me know if I'm missing something/there's something I'm not considering

[villebro]

As I was making the above change, I realized that if the celery queue was backed up, or there was any kind of delay between when celery beat placed the job into the queue and when it was executed, alerts & reports could be missed.

@jfrag1 I actually ran into this exact issue, and after investigating it further, I found out that unfortunately Celery only supports passing static variables from the scheduler to the worker 🙁 So I wasn't able to come up with a clean way of solving this. However, I think one alternative solution could be as follows:

1. the scheduler only triggers reports to be started at some interval. Let's say once per minute
2. this task is picked up by a worker, and it would acquire a distributed lock.
3. Instead of using the current time, it would check from the key value store when alerts were last executed, and then execute any reports that are in the interval between last execution time and now.
4. After this the last run would be updated, and the lock would be returned.

If another worker tries to start reports at the same time, they would silently go away, as the lock would already be taken. This would ensure the following:

1. No reports would be missed - if the queue would be clogged up, at some point a report scheduling task would get through, and then it would be able to clear the backlog of reports
2. duplicates would not happen, as the distributed lock would ensure that only one worker does report scheduling at a time

Thoughts @jfrag1 @zephyring ? I have code that I use internally for distributed locks on the Superset key_value store (we use it in another context, but it's been rock solid for the last 1 year or so), so I can collab on this if needed.

[jfrag1]

Thanks for the input @villebro! What you outlined does sound like the cleanest solution, but I'm not sure how to implement it myself, so if you're willing to collab that'd be great.

I found out that unfortunately Celery only supports passing static variables from the scheduler to the worker

I discovered the same while looking into this, but the expires solution in this PR is kind of a workaround way to have beat pass the time it placed the job into the queue to the worker. Of course it's not perfect, but do you feel it's "good enough" for this situation?

[villebro]

LGTM, thanks for this clever fix to a long standing problem!

[villebro]

not your fault, but having to do it like this just makes me cringe.. 😄 I'm hoping someone can add request.scheduled_at to Celery 😉

[villebro]

I love freezegun, hands off @john-bodley! 😄

[piyushdatazip]

Hi I've been checking this code, how is this solving the issue, it select datetime.utcnow() because datetime.fromisoformat(scheduler.request.expires) - app.config["CELERY_BEAT_SCHEDULER_EXPIRES"] = 0
Can you help me understand @jfrag1 ??

[jfrag1]

@piyushdatazip I'm not sure I fully understand what you're asking, but datetime.fromisoformat(scheduler.request.expires) is a datetime, and app.config["CELERY_BEAT_SCHEDULER_EXPIRES"] is a timedelta, so the result of this subtraction is a datetime, not 0.

[piyushdatazip]

Thanks @jfrag1 for replying, my bad I've understood it wrongly at first instance