Removes the next_dagrun_create_after reset by avkirilishin · Pull Request #21214 · apache/airflow

avkirilishin · 2022-01-30T00:56:31Z

closes: #21083
closes: #19901

The next_dagrun_create_after reset causes a problem (details in the issue).
But now we check _should_update_dag_next_dagruns before calculate_dagrun_date_fields call. So we can remove the reset.

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

ashb · 2022-01-31T21:05:24Z

This needs more thought/a deeper review than I can give right now.

I think by removing that then we might have undone the fix where we introduced this change in the first place

avkirilishin · 2022-02-02T19:18:04Z

I think by removing that then we might have undone the fix where we introduced this change in the first place

I checked this. Let's reconstruct the sequence of events.

Initially next_dagrun_create_after resets only in calculate_dagrun_date_fields and only with other "next*" attributes (next_dagrun, next_dagrun_data_interval). It didn't control max_active_runs.
Fix catchup by limiting queued dagrun creation using max_active_runs #18897: unconditional _update_dag_next_dagruns call introduced, now next_dagrun_create_after used to control max_active_runs
Fix missing dagruns when catchup=True #19528: _should_update_dag_next_dagruns was added, so we don't call _update_dag_next_dagruns without check and we don't need next_dagrun_create_after = NULL anymore

ephraimbuddy

We need to have a test that creates a dagrun and while the dagrun is still running, the scheduler loops again and should not create more dagrun

ephraimbuddy

This works fine in that it doesn't create more runs when max_active_runs is reached but it didn't resolve the linked issue.
Can you work on fixing the linked issue too as I like that next_dagrun_create_after is no longer nullified

avkirilishin · 2022-02-04T19:15:53Z

but it didn't resolve the linked issue

This fix resolves the linked issue. The new test test_runs_are_created_after_max_active_runs_was_reached verifies this.

I have to explain how it works:

The value of max_active_runs has been reached, and the value of next_dagrun_create_after has been reset:

airflow/airflow/jobs/scheduler_job.py

Line 975 in ab762a5

dag_model.next_dagrun_create_after = None

Due to filtering next_dagrun_create_after, DagModel.dags_needing_dagruns (referred to in SchedulerJob._create_dagruns_for_dags) cannot select Dags to create new runs. Comparison predicates with NULL values always return UNKNOWN (==FALSE in that case).

airflow/airflow/models/dag.py

Lines 2866 to 2876 in ab762a5

    
           query = ( 
        
               session.query(cls) 
        
               .filter( 
        
                   cls.is_paused == expression.false(), 
        
                   cls.is_active == expression.true(), 
        
                   cls.has_import_errors == expression.false(), 
        
                   cls.next_dagrun_create_after <= func.now(), 
        
               ) 
        
               .order_by(cls.next_dagrun_create_after) 
        
               .limit(cls.NUM_DAGS_PER_DAGRUN_QUERY) 
        
           )

Dag sticks and runs are no longer being created

If AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL is low, the scheduler processes dag's file and calls Dag.bulk_write_to_db, which calls calculate_dagrun_date_fields and calculates the new next_dagrun_create_after value

airflow/airflow/models/dag.py

Lines 2441 to 2444 in ab762a5

    
           if num_active_runs.get(dag.dag_id, 0) >= orm_dag.max_active_runs: 
        
               orm_dag.next_dagrun_create_after = None 
        
           else: 
        
               orm_dag.calculate_dagrun_date_fields(dag, data_interval)

But if AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL is high, nobody calculates next_dagrun_create_after
This merge request corrects situation by removing the next_dagrun_create_after nullify

next_dagrun_create_after has been reset:

min_file_process_interval is 180. After 3 minutes (16:21:36 - 16:24:37), the scheduler processes dag's file and calculates the new next_dagrun_create_after value:

There is a gap in launches:

Pay attention to the same last_parsed_time and Queued At in the last two screenshots.

avkirilishin · 2022-02-04T19:16:23Z

I did load testing today, and it turned out that now the scheduler consumes a lot more CPU, since it constantly selects all the necessary DAGs in the DagModel.dags_needing_dagruns function. Maybe I can add a dict with the time of the last check? But it won't work for AIRFLOW__SCHEDULER__NUM_RUNS = 1

ephraimbuddy · 2022-02-06T09:36:37Z

@avkirilishin , Can you test with AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL=86400

avkirilishin · 2022-02-06T10:41:09Z

Can you test with AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL=86400

@ephraimbuddy

ephraimbuddy · 2022-02-06T22:14:58Z

Hi @avkirilishin, I debugged this issue and it turned out that the problem is not that the next_dagrun_create_after is being nullified. The problem is that this line

airflow/airflow/jobs/scheduler_job.py

Line 1060 in 1d170f8

active_runs = dag.get_num_active_runs(only_running=False, session=session)

is returning a wrong figure. You can verify this.

I have not been able to pin out why it's returning a wrong figure but If you change this line of code:

airflow/airflow/jobs/scheduler_job.py

Line 1088 in 1d170f8

if self._should_update_dag_next_dagruns(dag, dag_model, active_runs):

to

if self._should_update_dag_next_dagruns(dag, dag_model, active_runs-1):

The problem will be gone.

avkirilishin · 2022-02-07T08:33:44Z

I have not been able to pin out why it's returning a wrong figure but If you change this line of code:
The problem will be gone.

@ephraimbuddy I tested your suggestion. The problem has not gone away. DAG sticks after the first launch.

ephraimbuddy · 2022-02-07T08:34:53Z

I tested your suggestion. The problem has not gone away. DAG sticks after the first launch.

Sorry, I have updated my comment, I gave the wrong permalink. You should change line 1088

airflow/airflow/jobs/scheduler_job.py

Line 1088 in 1d170f8

if self._should_update_dag_next_dagruns(dag, dag_model, active_runs):

avkirilishin · 2022-02-07T09:39:27Z

I tested your suggestion. The problem has not gone away. DAG sticks after the first launch.

Sorry, I have updated my comment, I gave the wrong permalink. You should change line 1088

airflow/airflow/jobs/scheduler_job.py

Line 1088 in 1d170f8

if self._should_update_dag_next_dagruns(dag, dag_model, active_runs):

This is a good change, and I think we need to apply it. But only it won't work if one sets the DagRun state manually (Success or Failed).

ephraimbuddy · 2022-02-07T11:44:27Z

This is a good change, and I think we need to apply it. But only it won't work if one sets the DagRun state manually (Success or Failed).

It worked for me. Can you tests too?

avkirilishin · 2022-02-07T12:13:44Z

This is a good change, and I think we need to apply it. But only it won't work if one sets the DagRun state manually (Success or Failed).

It worked for me. Can you tests too?

With main plus "active_runs-1" or with this branch plus "active_runs-1"?

Main plus "active_runs-1" - the manually changed DagRun state leads to a hang for me (with AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL=86400).
This branch plus "active_runs-1" - the manually changed DagRun state works fine.

ephraimbuddy · 2022-02-07T12:41:52Z

With main plus "active_runs-1" or with this branch plus "active_runs-1"?

Main plus "active_runs-1" - the manually changed DagRun state leads to a hang for me (with AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL=86400). This branch plus "active_runs-1" - the manually changed DagRun state works fine.

It still works for me in main, not sure why. I will suggest we implement active_runs-1 and look at fixing it when the state is manually changed.
We can also create a boolean column on DagModel and instead of nullifying next_dagrun_create_after, we nullify the column. That way, we don't need to calculate the column again. We would just set the new column when max_active_runs is reached and unset it when dagrun finishes running or are marked manually as failed.
That together with active_runs-1 at lines 1088 and 1062 will be a better fix

avkirilishin · 2022-02-07T15:08:32Z

But we can (and we do it) calculate the current running DagRun count at any time. Why do we need a new column?

Maybe this way:

implement active_runs-1
remove nullifying next_dagrun_create_after
and a little later I will redo the load testing, maybe my previous conclusions were wrong

?

ephraimbuddy · 2022-02-07T15:41:45Z

But we can (and we do it) calculate the current running DagRun count at any time. Why do we need a new column?

Maybe this way:

implement active_runs-1

remove nullifying next_dagrun_create_after

and a little later I will redo the load testing, maybe my previous conclusions were wrong

?

I want to avoid https://github.com/apache/airflow/pull/21214/files#diff-62c8e300ee91e0d59f81e0ea5d30834f04db71ae74f2e155a10b51056b00b59bR2874. That query is run a lot, the conditional check and providing a list of IDs to look into won't be efficient in large deployments

avkirilishin · 2022-02-07T16:12:45Z

Is there anything else I can do? And whether to close this merge request or not?

ephraimbuddy · 2022-02-07T17:32:56Z

Is there anything else I can do? And whether to close this merge request or not?

If you don't mind you can apply the active_runs-1 change and leave out the addition of new column for now until this issue #16078 is resolved.

ephraimbuddy · 2022-02-07T17:35:01Z

Is there anything else I can do? And whether to close this merge request or not?

If you don't mind you can apply the active_runs-1 change and leave out the addition of new column for now until this issue #16078 is resolved.

You can do it in a new PR if you want

ephraimbuddy · 2022-02-08T07:20:31Z

I have added the PR: #21413

avkirilishin · 2022-02-08T07:44:57Z

Can we add test_runs_are_created_after_max_active_runs_was_reached to #21413 from this merge request?

airflow/tests/jobs/test_scheduler_job.py

Lines 2927 to 2961 in 039ba52

    
               def test_runs_are_created_after_max_active_runs_was_reached(self, dag_maker, session): 
        
                   """ 
        
                   Test that when creating runs once max_active_runs is reached the runs does not stick 
        
                   """ 
        
                   with conf_vars({('scheduler', 'min_active_runs_check_interval'): '0'}): 
        
                       self.scheduler_job = SchedulerJob(subdir=os.devnull) 
        
                       self.scheduler_job.executor = MockExecutor(do_update=True) 
        
                       self.scheduler_job.processor_agent = mock.MagicMock(spec=DagFileProcessorAgent) 
        
                       with dag_maker(max_active_runs=1, session=session) as dag: 
        
                           # Need to use something that doesn't immediately get marked as success by the scheduler 
        
                           BashOperator(task_id='task', bash_command='true') 
        
                       dag_run = dag_maker.create_dagrun( 
        
                           state=State.RUNNING, 
        
                           session=session, 
        
                       ) 
        
                       # Reach max_active_runs 
        
                       for _ in range(3): 
        
                           self.scheduler_job._do_scheduling(session) 
        
                       # Complete dagrun 
        
                       # Add dag_run back in to the session (_do_scheduling does an expunge_all) 
        
                       dag_run = session.merge(dag_run) 
        
                       session.refresh(dag_run) 
        
                       dag_run.get_task_instance(task_id='task', session=session).state = State.SUCCESS 
        
                       # create new run 
        
                       for _ in range(3): 
        
                           self.scheduler_job._do_scheduling(session) 
        
                       # Assert that new runs has created 
        
                       dag_runs = DagRun.find(dag_id=dag.dag_id, session=session) 
        
                       assert len(dag_runs) == 2

I will remove min_active_runs_check_interval from it.

ephraimbuddy · 2022-02-08T08:00:57Z

Done

Removing the reset next_dagrun_create_after

bb9ed07

avkirilishin requested review from XD-DENG, ashb and kaxil as code owners January 30, 2022 00:56

Merge branch 'apache:main' into main

81091c0

ashb requested a review from ephraimbuddy February 3, 2022 10:37

ephraimbuddy requested changes Feb 3, 2022

View reviewed changes

Comment thread tests/jobs/test_scheduler_job.py Outdated

Improve tests

913ea9e

avkirilishin requested a review from ephraimbuddy February 3, 2022 22:19

ephraimbuddy requested changes Feb 4, 2022

View reviewed changes

avkirilishin requested a review from ephraimbuddy February 4, 2022 19:16

avkirilishin added 2 commits February 5, 2022 16:35

Reduce CPU usage

0953943

Files were modified by black

039ba52

avkirilishin force-pushed the main branch from 12913af to 039ba52 Compare February 5, 2022 15:18

holyachon mentioned this pull request Feb 7, 2022

No scheduling when max_active_runs is 1 #19901

Closed

2 tasks

avkirilishin closed this Feb 8, 2022

Uh oh!

Conversation

avkirilishin commented Jan 30, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ashb commented Jan 31, 2022

Uh oh!

avkirilishin commented Feb 2, 2022

Uh oh!

ephraimbuddy left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ephraimbuddy left a comment

Choose a reason for hiding this comment

Uh oh!

avkirilishin commented Feb 4, 2022

Uh oh!

avkirilishin commented Feb 4, 2022

Uh oh!

ephraimbuddy commented Feb 6, 2022

Uh oh!

avkirilishin commented Feb 6, 2022

Uh oh!

ephraimbuddy commented Feb 6, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

avkirilishin commented Feb 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ephraimbuddy commented Feb 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

avkirilishin commented Feb 7, 2022

Uh oh!

ephraimbuddy commented Feb 7, 2022

Uh oh!

avkirilishin commented Feb 7, 2022

Uh oh!

ephraimbuddy commented Feb 7, 2022

Uh oh!

avkirilishin commented Feb 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ephraimbuddy commented Feb 7, 2022

Uh oh!

avkirilishin commented Feb 7, 2022

Uh oh!

ephraimbuddy commented Feb 7, 2022

Uh oh!

ephraimbuddy commented Feb 7, 2022

Uh oh!

ephraimbuddy commented Feb 8, 2022

Uh oh!

avkirilishin commented Feb 8, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ephraimbuddy commented Feb 8, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

avkirilishin commented Jan 30, 2022 •

edited

Loading

ephraimbuddy commented Feb 6, 2022 •

edited

Loading

avkirilishin commented Feb 7, 2022 •

edited

Loading

ephraimbuddy commented Feb 7, 2022 •

edited

Loading

avkirilishin commented Feb 7, 2022 •

edited

Loading

avkirilishin commented Feb 8, 2022 •

edited

Loading