You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "michaelmicheal (via GitHub)" <gi...@apache.org> on 2023/02/09 18:47:21 UTC
[GitHub] [airflow] michaelmicheal opened a new pull request, #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
michaelmicheal opened a new pull request, #29446:
URL: https://github.com/apache/airflow/pull/29446
When we deactivate stale DAGs, we add a buffer to the last parsed time of a DAG when comparing it to the last parsed time of its DAG file to account for DAG files that take a long time to process.
```python
for dag in dags_parsed:
# The largest valid difference between a DagFileStat's last_finished_time and a DAG's
# last_parsed_time is _processor_timeout. Longer than that indicates that the DAG is
# no longer present in the file.
if (
dag.fileloc in last_parsed
and (dag.last_parsed_time + processor_timeout) < last_parsed[dag.fileloc]
):
cls.logger().info("DAG %s is missing and will be deactivated.", dag.dag_id)
to_deactivate.add(dag.dag_id)
```
The upper bound on this delay is indeed the processor timeout, but practically speaking, this is unnecessary when we configure a long timeout. This can result in a large delay in stale DAG deactivation. I think this buffer should be configurable.
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] michaelmicheal commented on a diff in pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "michaelmicheal (via GitHub)" <gi...@apache.org>.
michaelmicheal commented on code in PR #29446:
URL: https://github.com/apache/airflow/pull/29446#discussion_r1101929995
##########
airflow/config_templates/default_airflow.cfg:
##########
@@ -1045,8 +1045,16 @@ min_file_process_interval = 30
# referenced and should be marked as orphaned.
parsing_cleanup_interval = 60
+# How long (in seconds) to wait after we've reparsed a DAG file before deactivating stale
+# DAGs (DAGs which are no longer present in the expected files). The reason why we need
+# this threshold is to account for the time between when the file is parsed and when the
+# DAG is loaded. The absolute maximum that this could take is `dag_file_processor_timeout`,
+# but when you have a long timeout configured, it results in a significant delay in the
+# deactivation of stale dags.
+stale_dag_threshold = 30
+
# How often (in seconds) to scan the DAGs directory for new files. Default to 5 minutes.
-dag_dir_list_interval = 300
+dag_dir_list_interval = 10
Review Comment:
Sorry, accident
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a diff in pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #29446:
URL: https://github.com/apache/airflow/pull/29446#discussion_r1111634767
##########
airflow/dag_processing/manager.py:
##########
@@ -523,13 +525,17 @@ def deactivate_stale_dags(
query = query.filter(DagModel.processor_subdir == dag_directory)
dags_parsed = query.all()
+ processor_timeout_seconds: int = conf.getint("core", "dag_file_processor_timeout")
+
+
Review Comment:
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #29446:
URL: https://github.com/apache/airflow/pull/29446#issuecomment-1454851594
Rebased to account for workarounded issue with boto3. But seems the issue has been fixed @ephraimbuddy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Taragolis commented on pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on PR #29446:
URL: https://github.com/apache/airflow/pull/29446#issuecomment-1435719997
I would recommend to use `pre-commit` to fix most of the statics checks, see: [Static code checks](https://github.com/apache/airflow/blob/main/STATIC_CODE_CHECKS.rst#id4)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] michaelmicheal commented on pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "michaelmicheal (via GitHub)" <gi...@apache.org>.
michaelmicheal commented on PR #29446:
URL: https://github.com/apache/airflow/pull/29446#issuecomment-1428573471
> Can we have a test with different values for the threshold? This would gaurd against removing the config without a _fight_
Is there an example of something like this for other configurations?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Taragolis commented on a diff in pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #29446:
URL: https://github.com/apache/airflow/pull/29446#discussion_r1111061966
##########
airflow/config_templates/config.yml:
##########
@@ -2049,6 +2049,18 @@ scheduler:
type: integer
example: ~
default: "60"
+ stale_dag_threshold:
+ description: |
+ How long (in seconds) to wait after we've reparsed a DAG file before deactivating stale
Review Comment:
```suggestion
How long (in seconds) to wait after we have reparsed a DAG file before deactivating stale
```
This should fix, spellchecker doesn't like some words with `'`
```console
Incorrect Spelling: 'reparsed'
Line with Error: 'How long (in seconds) to wait after we've reparsed a DAG file before deactivating stale'
Line Number: 6812
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a diff in pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #29446:
URL: https://github.com/apache/airflow/pull/29446#discussion_r1101909468
##########
airflow/config_templates/default_airflow.cfg:
##########
@@ -1045,8 +1045,16 @@ min_file_process_interval = 30
# referenced and should be marked as orphaned.
parsing_cleanup_interval = 60
+# How long (in seconds) to wait after we've reparsed a DAG file before deactivating stale
+# DAGs (DAGs which are no longer present in the expected files). The reason why we need
+# this threshold is to account for the time between when the file is parsed and when the
+# DAG is loaded. The absolute maximum that this could take is `dag_file_processor_timeout`,
+# but when you have a long timeout configured, it results in a significant delay in the
+# deactivation of stale dags.
+stale_dag_threshold = 30
+
# How often (in seconds) to scan the DAGs directory for new files. Default to 5 minutes.
-dag_dir_list_interval = 300
+dag_dir_list_interval = 10
Review Comment:
Why this change?
##########
airflow/dag_processing/manager.py:
##########
@@ -433,6 +433,8 @@ def __init__(
self.last_deactivate_stale_dags_time = timezone.make_aware(datetime.fromtimestamp(0))
# How often to check for DAGs which are no longer in files
self.parsing_cleanup_interval = conf.getint("scheduler", "parsing_cleanup_interval")
+ # How long to wait for a DAG to be reparsed after it's file has been parsed before disabling
Review Comment:
```suggestion
# How long to wait for a DAG to be reparsed after its file has been parsed before disabling
```
##########
airflow/config_templates/config.yml:
##########
@@ -2049,6 +2049,18 @@ scheduler:
type: integer
example: ~
default: "60"
+ stale_dag_threshold:
+ description: |
+ How long (in seconds) to wait after we've reparsed a DAG file before deactivating stale
+ DAGs (DAGs which are no longer present in the expected files). The reason why we need
+ this threshold is to account for the time between when the file is parsed and when the
+ DAG is loaded. The absolute maximum that this could take is `dag_file_processor_timeout`,
+ but when you have a long timeout configured, it results in a significant delay in the
+ deactivation of stale dags.
+ version_added: 2.6.0
+ type: integer
+ example: ~
+ default: "30"
Review Comment:
Should we keep this the same default as `dag_file_processor_timeout`? That way, there will be no change in behaviour
##########
airflow/dag_processing/manager.py:
##########
@@ -523,13 +525,16 @@ def deactivate_stale_dags(
query = query.filter(DagModel.processor_subdir == dag_directory)
dags_parsed = query.all()
+
+
for dag in dags_parsed:
# The largest valid difference between a DagFileStat's last_finished_time and a DAG's
- # last_parsed_time is _processor_timeout. Longer than that indicates that the DAG is
- # no longer present in the file.
+ # last_parsed_time is thg processor_timeout. Longer than that indicates that the DAG is
Review Comment:
```suggestion
# last_parsed_time is the processor_timeout. Longer than that indicates that the DAG is
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on PR #29446:
URL: https://github.com/apache/airflow/pull/29446#issuecomment-1428653320
> > Can we have a test with different values for the threshold? This would gaurd against removing the config without a _fight_
>
> Is there an example of something like this for other configurations?
Yep. https://github.com/apache/airflow/blob/02a2efeae409bddcfedafe273fffc353595815cc/tests/jobs/test_scheduler_job.py#L527
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Taragolis commented on a diff in pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on code in PR #29446:
URL: https://github.com/apache/airflow/pull/29446#discussion_r1111062123
##########
airflow/config_templates/config.yml:
##########
@@ -2049,6 +2049,18 @@ scheduler:
type: integer
example: ~
default: "60"
+ stale_dag_threshold:
+ description: |
+ How long (in seconds) to wait after we've reparsed a DAG file before deactivating stale
Review Comment:
You could also check it in [Breeze](https://github.com/apache/airflow/blob/main/BREEZE.rst#building-the-documentation) locally
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] michaelmicheal commented on pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "michaelmicheal (via GitHub)" <gi...@apache.org>.
michaelmicheal commented on PR #29446:
URL: https://github.com/apache/airflow/pull/29446#issuecomment-1454919583
> Rebased to account for workarounded issue with boto3. But seems the issue has been fixed @ephraimbuddy
Thanks is this good to merge?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #29446:
URL: https://github.com/apache/airflow/pull/29446#issuecomment-1445121785
Needs fixing tests :(
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy merged pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy merged PR #29446:
URL: https://github.com/apache/airflow/pull/29446
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #29446: Scheduler, make stale DAG deactivation threshold configurable instead of using dag processing timeout
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #29446:
URL: https://github.com/apache/airflow/pull/29446#issuecomment-1454927375
> > Rebased to account for workarounded issue with boto3. But seems the issue has been fixed @ephraimbuddy
>
> Thanks is this good to merge?
Waiitng for @ephraimbuddy to remove his "change request" . I cannot merge it without it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org