Repository navigation
CronTriggerTimetable lost one task occasionally #27399
Description
Activity
Mistakenly converted to discussion. Seems legit. This is likely only happening in 1 minute interval, so not really critical but woudl be nice to find out when it happens. - thanks @andi for detailed repot.
Could you please check @andi if it appears in the latest version of Airflow and maybe pinpoint some other cicumstances when it happens? How often? Do you see some usage spikes (memory/cpu etc. around that time? do you see any unsual task restarts etc. - we need to try to get some really reproducible case in order to be able to track tha one - otherwise it is a pure quess.
Can you try to see much more information @andi ?
I think the other issue is "first" schedule - and this one is "missing one in the middle" - it might be the same but it might be different.
Hey all, we had an issue in our configuration where our DAGs were reloading by the order of milliseconds (looking at last_parsed_time in the database) and we observed this issue happening a lot. The only DAG that failed to run in this circumstance were the ones that used CronTriggerTimetable. (We were using a test that should run every minute). I believe that if you set the DAG reload time very, very low, you will see this problem occur much more frequently.
When we fixed the reload issue in the configuration, we have noticed this problem much less frequently.
Steps to reproduce this (with other bug that reloads DAGs in order of milliseconds):
- Set environment variable TZ to UTC. (Or leave unset if UTC by default)
- Set environment variable AIRFLOW__CORE__DEFAULT_TIMEZONE=America/Chicago
- Create new DAG (this DAG must not currently be in the database) that uses CronTriggerTimetable, we used this:
with DAG( 'cron-trigger-test-repeat-new', description='cron-trigger-reproduce-bug', catchup=False, schedule=CronTriggerTimetable(f'* * * * *', timezone='America/Chicago'), default_args = { 'start_date': datetime(2022, 1, 1), 'depends_on_past': False, 'timezone': "America/Chicago" }, is_paused_upon_creation=False ) as echo: hello = BashOperator( task_id="echo", bash_command="echo Hello World!" ) echo- Wait for DAG to be picked up by the scheduler
- Observe that the last_parsed_time in the dag table in airflow database (we use postgres) is updating wayyyy faster than it needs to (multiple times per second).
- Observe that the DAG has a lot of missed runs every minute (we get less than 50% to trigger on schedule when we do this)
Can confirm when trying to schedule a DAG exactly one with
CronTriggerTimetableairflow [v2.4.2] in standalone mode (wheredag_dir_list_interval = 300seems to be ignored)import subprocess from airflow import DAG from airflow.decorators import dag, task from airflow.timetables.trigger import CronTriggerTimetable from datetime import datetime @task def mytask(): bash_command=f'/bin/echo run once' EXEC_OUT=(subprocess.check_output(bash_command, shell=True)).decode('utf-8') @dag( dag_id=f'run_once', start_date=datetime(2022,1,1), timetable=CronTriggerTimetable('03 18 16 11 *', timezone='Australia/Sydney'), catchup=False, ) def run_once_dag(): mytask() run_once_dag()In the minute before 18:03, the DAG was processed 22 times
The first processing at 18:03, the
next_dagrunwas set to the same time one year later, without executing the DAG[2022-11-16T18:02:50.448+1100] {processor.py:154} INFO - Started process (PID=11018) to work on /export/home/jwarburt/airflow-venv/dags/run-once.py [2022-11-16T18:02:50.506+1100] {logging_mixin.py:120} INFO - [2022-11-16T18:02:50.506+1100] {dag.py:3336} INFO - Setting next_dagrun for run_once to 2022-11-16T07:03:00+00:00, run_after=2022-11-16T07:03:00+00:00 [2022-11-16T18:02:50.609+1100] {processor.py:176} INFO - Processing /export/home/jwarburt/airflow-venv/dags/run-once.py took 0.164 seconds [2022-11-16T18:02:53.836+1100] {processor.py:154} INFO - Started process (PID=11026) to work on /export/home/jwarburt/airflow-venv/dags/run-once.py [2022-11-16T18:02:53.894+1100] {logging_mixin.py:120} INFO - [2022-11-16T18:02:53.894+1100] {dag.py:3336} INFO - Setting next_dagrun for run_once to 2022-11-16T07:03:00+00:00, run_after=2022-11-16T07:03:00+00:00 [2022-11-16T18:02:54.004+1100] {processor.py:176} INFO - Processing /export/home/jwarburt/airflow-venv/dags/run-once.py took 0.171 seconds [2022-11-16T18:02:59.105+1100] {processor.py:154} INFO - Started process (PID=11032) to work on /export/home/jwarburt/airflow-venv/dags/run-once.py [2022-11-16T18:02:59.163+1100] {logging_mixin.py:120} INFO - [2022-11-16T18:02:59.163+1100] {dag.py:3336} INFO - Setting next_dagrun for run_once to 2022-11-16T07:03:00+00:00, run_after=2022-11-16T07:03:00+00:00 [2022-11-16T18:02:59.272+1100] {processor.py:176} INFO - Processing /export/home/jwarburt/airflow-venv/dags/run-once.py took 0.171 seconds [2022-11-16T18:03:00.699+1100] {processor.py:154} INFO - Started process (PID=11034) to work on /export/home/jwarburt/airflow-venv/dags/run-once.py [2022-11-16T18:03:00.845+1100] {logging_mixin.py:120} INFO - [2022-11-16T18:03:00.845+1100] {dag.py:3336} INFO - Setting next_dagrun for run_once to 2023-11-16T07:03:00+00:00, run_after=2023-11-16T07:03:00+00:00 [2022-11-16T18:03:00.865+1100] {processor.py:176} INFO - Processing /export/home/jwarburt/airflow-venv/dags/run-once.py took 0.169 secondsIf I could slow down the processing interval, that might help?
- added this to the milestone
on Nov 16, 2022 @johnwarburton Have you had any luck with this? Seeing the same issue in 2.4.2 after switching from a plain cron expression (i.e.
schedule_interval=0 15 * * *) to using CronTriggerTimetable directly with a timezone (i.e.timetable=CronTriggerTimetable("0 15 * * *", timezone="America/Los_Angeles")). I noticed that all of the examples in this thread have a timezone attached.Some more context:
- Kubernetes Executor
- This isn't a case of a single DAG with a very frequently running cron expression like
* * * * *, but it does have a lot of DAGs that all run at the same time.
Hi @RNHTTR - no, I am waiting to see if there is a fix in the milestones
CronTriggerTimetablemust have a timezone attached:| timetable=CronTriggerTimetable('45 4 2 12 *'), | TypeError: __init__() missing 1 required keyword-only argument: 'timezone'30 remaining items
I think a reasonable logic would be to change the
catchup=Falselogic to cover one schedule before the current time instead, so in the above scenario, the timetable would make the next run cover 4am, and only skip the 4am run if the current time is pas 5am.Sounds good. It's natural to those who are used to Cron and Anacron.
I think we need to document the new behavior. The new behavior might be unnatural to those who are not familiar with Anacron. Say
- You have an hourly cron
- Last run on 3am
- You paused the cron DAG at 3:55 to skip the next run on 4am
- You unpaused it at 4:30
- The skipped run on 4am resumes (This might be unexpected)
Yeah. Or maybe we can addd a flag for this? Not sure what it should be called though.
I've been testing the following change to fix this issue seems to be working. Its similar to what has been suggested:
def next_dagrun_info( self, *, last_automated_data_interval: Optional[DataInterval], restriction: TimeRestriction, ) -> Optional[DagRunInfo]: if restriction.catchup: if last_automated_data_interval is None: if restriction.earliest is None: return None next_start_time = self._align_to_next(restriction.earliest) else: next_start_time = self._get_next(last_automated_data_interval.end) else: current_time = DateTime.utcnow() if restriction.earliest is not None and current_time < restriction.earliest: next_start_time = self._align_to_next(restriction.earliest) else: if last_automated_data_interval is None: # The DAG has never run before. next_start_time = self._align_to_next(current_time) elif last_automated_data_interval.end == self._align_to_prev(current_time): # The last run is the same as the previously scheduled run. Everything is working fine. next_start_time = self._align_to_next(current_time) else: # The last run is NOT the same as the previously scheduled run, we must have skipped it. next_start_time = self._align_to_prev(current_time) LOGGER.warn('DAG skipped previous schedule, rescheduling. last_automated_data_interval.end:"%s" next_start_time:"%s"', last_automated_data_interval.end, next_start_time) if restriction.latest is not None and restriction.latest < next_start_time: return None return DagRunInfo.exact(next_start_time)
Specifically this addition:
if last_automated_data_interval is None: # The DAG has never run before. next_start_time = self._align_to_next(current_time) elif last_automated_data_interval.end == self._align_to_prev(current_time): # The last run is the same as the previously scheduled run. Everything is working fine. next_start_time = self._align_to_next(current_time) else: # The last run is NOT the same as the previously scheduled run, we must have skipped it. next_start_time = self._align_to_prev(current_time) LOGGER.warn('DAG skipped previous schedule, rescheduling. last_automated_data_interval.end:"%s" next_start_time:"%s"', last_automated_data_interval.end, next_start_time)
I added the logging so I could monitor it and alert if it happens, just in case. Not sure if this could be done better though, I'm open to suggestions.
elif last_automated_data_interval.end == self._align_to_prev(current_time): # The last run is the same as the previously scheduled run. Everything is working fine. next_start_time = self._align_to_next(current_time)
This part has a bug when
current_timeis exactlylast_automated_data_interval.end,next_start_timewould be set tocurrent_timeagain (because the align functions will try to not move the time if it already lands on the cron). Very unlikely but still. I think it can be avoided withself._get_next(last_automated_data_interval.end)instead, to ensure the logic progresses one interval ahead.The
elseblock is essentially the same as the change discused above and looks good to me.Thanks @uranusjr for the feedback! I'll look into it, good points.
The bug occurs due to the fact that seconds and microseconds are not reset here:
start_time_candidates = [self._align_to_next(DateTime.utcnow())]and a conditionif self._get_prev(next_time) != current:in the method_align_to_nextchoosesnext_timeinsteadcurrent, when the current DateTime matches with the next_dagrun before the DagProcessor will process the DAG. Accordingly, the date switching occurs earlier than the DAG launching.The bug fix is only one line:
start_time_candidates = [self._align_to_next(DateTime.utcnow().replace(second=0, microsecond=0))]
I've already tested the fix - it works properly.Reacted by snippins@BohdanSemonov I tried to applied your fix and do the following test.
- current time is 10:00
- switch the cron expression to '2 10 * * *'
- wait for scheduler to update (done around 10:01)
- pause the dag
- wait until 10:03
- resume the dag
The dag start to run, maybe I misunderstand something but I thought the entire point of CronTriggerTimetable is to prevent this?
@snippins
The situation you describe is the Airflow feature whencatchup=True.
Try withcatchup=False.Hey @BohdanSemonov , if the fix works, could you please submit a PR? I think this issue is quite critical, and I also observe such behavior from CronTriggerTimetable
Reacted by NikolayThe situation you describe is the Airflow feature when
catchup=True.I have catchup=False already.
The situation you describe is the Airflow feature when
catchup=True.I have catchup=False already.
Sorry, maybe we didn't understand each other, but I observe such behavior when
catchup=True, otherwise I could not get the described DAG run.Hey @BohdanSemonov , if the fix works, could you please submit a PR? I think this issue is quite critical, and I also observe such behavior from CronTriggerTimetable
Done.
Discussed in #27398
Originally posted by AndrewTsao October 25, 2022
Apache Airflow version
Other Airflow 2 version (please specify below)
What happened
My Airflow version 2.4.0. I use CronTriggerTimetable in my DAG. I find it lost one task occasionally. I inspected schedule log,
It should be run at 2022-10-24T23:10:00.000+0800. BUT not run.
What you think should happen instead
I think it lost schedule infomation when reload dag file and reschedule dag.
How to reproduce
ag_dir_list_interval = 10.In scheduler log, We can find on that point, scheduler is reloading dag file.
Operating System
Centos 7.9
Versions of Apache Airflow Providers
Deployment
Virtualenv installation
Deployment details
No response
Anything else
No response
Are you willing to submit PR?
Code of Conduct