You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/11/22 12:30:38 UTC
[GitHub] [airflow] stijndehaes opened a new pull request, #27834: Make sure we can get out of a faulty scheduler state
stijndehaes opened a new pull request, #27834:
URL: https://github.com/apache/airflow/pull/27834
This PR fixed the case where we have a faulty state in the database. The state that is fixed is that both the unmapped task instance and mapped task instances exist at the same time.
So we have instances with map_index [-1, 0, 1].
The -1 task instances should be removed in this case.
<!--
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: https://github.com/apache/airflow/discussions/26959
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] stijndehaes commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
stijndehaes commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031541359
##########
airflow/models/abstractoperator.py:
##########
@@ -494,18 +493,15 @@ def expand_mapped_task(self, run_id: str, *, session: Session) -> tuple[Sequence
total_length,
)
unmapped_ti.state = TaskInstanceState.SKIPPED
- indexes_to_map = ()
else:
- # Otherwise convert this into the first mapped index, and create
- # TaskInstance for other indexes.
- unmapped_ti.map_index = 0
- self.log.debug("Updated in place to become %s", unmapped_ti)
- all_expanded_tis.append(unmapped_ti)
- indexes_to_map = range(1, total_length)
- state = unmapped_ti.state
- elif not total_length:
+ # Delete the original task so we can create new ones
+ self.log.debug("Deleting the original task instance: %s", unmapped_ti)
+ state = unmapped_ti.state
+ session.delete(unmapped_ti)
Review Comment:
I update the code to have this change, please let me know if there is anything else you would like to see different
--
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] uranusjr commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1034231937
##########
airflow/models/dagrun.py:
##########
@@ -780,8 +780,7 @@ def _expand_mapped_task_if_needed(ti: TI) -> Iterable[TI] | None:
except NotMapped: # Not a mapped task, nothing needed.
return None
if expanded_tis:
- assert expanded_tis[0] is ti
- return expanded_tis[1:]
+ return expanded_tis
Review Comment:
If you prefer to keep the new behaviour (return all instead of only new), you need to modify the outer logic (naming, docstrings, and comments) to match.
--
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] stijndehaes commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
stijndehaes commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031520320
##########
airflow/models/abstractoperator.py:
##########
@@ -494,18 +493,15 @@ def expand_mapped_task(self, run_id: str, *, session: Session) -> tuple[Sequence
total_length,
)
unmapped_ti.state = TaskInstanceState.SKIPPED
- indexes_to_map = ()
else:
- # Otherwise convert this into the first mapped index, and create
- # TaskInstance for other indexes.
- unmapped_ti.map_index = 0
- self.log.debug("Updated in place to become %s", unmapped_ti)
- all_expanded_tis.append(unmapped_ti)
- indexes_to_map = range(1, total_length)
- state = unmapped_ti.state
- elif not total_length:
+ # Delete the original task so we can create new ones
+ self.log.debug("Deleting the original task instance: %s", unmapped_ti)
+ state = unmapped_ti.state
+ session.delete(unmapped_ti)
Review Comment:
Ok I'll try to make it work in a new commit.
So if I understood it correctly, what is preferred is:
- I keep the old logic that changes the `-1` map_index TI into the `0` map_index TI
- After that I detect if a `0` map_index ti already exists if so I deleted the original `-1` TI
--
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] tanelk commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
tanelk commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031324504
##########
airflow/models/abstractoperator.py:
##########
@@ -494,18 +493,15 @@ def expand_mapped_task(self, run_id: str, *, session: Session) -> tuple[Sequence
total_length,
)
unmapped_ti.state = TaskInstanceState.SKIPPED
- indexes_to_map = ()
else:
- # Otherwise convert this into the first mapped index, and create
- # TaskInstance for other indexes.
- unmapped_ti.map_index = 0
- self.log.debug("Updated in place to become %s", unmapped_ti)
- all_expanded_tis.append(unmapped_ti)
- indexes_to_map = range(1, total_length)
- state = unmapped_ti.state
- elif not total_length:
+ # Delete the original task so we can create new ones
+ self.log.debug("Deleting the original task instance: %s", unmapped_ti)
+ state = unmapped_ti.state
+ session.delete(unmapped_ti)
Review Comment:
in dagrun creation we explicitly check for duplicates - could do the same here.
https://github.com/apache/airflow/blob/01a819a42daa7990c30ab9776208b3dcb9f3a28b/airflow/jobs/scheduler_job.py#L1046-L1053
--
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 merged pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
potiuk merged PR #27834:
URL: https://github.com/apache/airflow/pull/27834
--
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] uranusjr commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031172476
##########
airflow/models/abstractoperator.py:
##########
@@ -494,18 +493,15 @@ def expand_mapped_task(self, run_id: str, *, session: Session) -> tuple[Sequence
total_length,
)
unmapped_ti.state = TaskInstanceState.SKIPPED
- indexes_to_map = ()
else:
- # Otherwise convert this into the first mapped index, and create
- # TaskInstance for other indexes.
- unmapped_ti.map_index = 0
- self.log.debug("Updated in place to become %s", unmapped_ti)
- all_expanded_tis.append(unmapped_ti)
- indexes_to_map = range(1, total_length)
- state = unmapped_ti.state
- elif not total_length:
+ # Delete the original task so we can create new ones
+ self.log.debug("Deleting the original task instance: %s", unmapped_ti)
+ state = unmapped_ti.state
+ session.delete(unmapped_ti)
Review Comment:
The fact that the original task instance is _converted_ instead of deleted was explicitly mentioned in the original proposal, so it would be best if we could keep the behaviour (I think there are some subtle differences but don’t recall exactly what). Would it be possible?
--
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] stijndehaes commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
stijndehaes commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1033304372
##########
airflow/models/dagrun.py:
##########
@@ -780,8 +780,7 @@ def _expand_mapped_task_if_needed(ti: TI) -> Iterable[TI] | None:
except NotMapped: # Not a mapped task, nothing needed.
return None
if expanded_tis:
- assert expanded_tis[0] is ti
- return expanded_tis[1:]
+ return expanded_tis
Review Comment:
I explicitely return all here
Since there are 2 use cases for the `-1` instance:
- it might be deleted
- it might have a changed index
I now return all so I do not have to care about what happened on line 805.
If there was an expansion, I just add all so they will looped over.
If we only return new ones I would have to detect that on line 805 somehow.
I hope this makes the current way it works more clear.
If you want I could only return new ti objects, but I would need to figure out a way to see if the ti in the loop has been deleted yes or no
--
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] uranusjr commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031173795
##########
airflow/models/abstractoperator.py:
##########
@@ -494,18 +493,15 @@ def expand_mapped_task(self, run_id: str, *, session: Session) -> tuple[Sequence
total_length,
)
unmapped_ti.state = TaskInstanceState.SKIPPED
- indexes_to_map = ()
else:
- # Otherwise convert this into the first mapped index, and create
- # TaskInstance for other indexes.
- unmapped_ti.map_index = 0
- self.log.debug("Updated in place to become %s", unmapped_ti)
- all_expanded_tis.append(unmapped_ti)
- indexes_to_map = range(1, total_length)
- state = unmapped_ti.state
- elif not total_length:
+ # Delete the original task so we can create new ones
+ self.log.debug("Deleting the original task instance: %s", unmapped_ti)
+ state = unmapped_ti.state
+ session.delete(unmapped_ti)
Review Comment:
If the scheduler already reached a faulty state (e.g. `[-1, 0, 1]` in your case) we can of course delete the ti to rescure it, but otherwise it’d be best to not delete things.
--
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] stijndehaes commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
stijndehaes commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1035781517
##########
airflow/models/dagrun.py:
##########
@@ -780,8 +780,7 @@ def _expand_mapped_task_if_needed(ti: TI) -> Iterable[TI] | None:
except NotMapped: # Not a mapped task, nothing needed.
return None
if expanded_tis:
- assert expanded_tis[0] is ti
- return expanded_tis[1:]
+ return expanded_tis
Review Comment:
I changed the docstring of `_expand_mapped_task_if_needed`.
I think all the other comments and the naming is still fine.
If you think something needs to be changed just let me know :)
--
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] stijndehaes commented on pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
stijndehaes commented on PR #27834:
URL: https://github.com/apache/airflow/pull/27834#issuecomment-1323672679
I created this PR to fix the wrong state but I still don't know how this faulty state was achieved.
However we have noticed it at least 3 times in the last 4 days. And always we manually removed the `-1` record to fix 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
[GitHub] [airflow] uranusjr commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031924565
##########
airflow/models/abstractoperator.py:
##########
@@ -494,18 +495,30 @@ def expand_mapped_task(self, run_id: str, *, session: Session) -> tuple[Sequence
total_length,
)
unmapped_ti.state = TaskInstanceState.SKIPPED
- indexes_to_map = ()
else:
- # Otherwise convert this into the first mapped index, and create
- # TaskInstance for other indexes.
- unmapped_ti.map_index = 0
- self.log.debug("Updated in place to become %s", unmapped_ti)
- all_expanded_tis.append(unmapped_ti)
- indexes_to_map = range(1, total_length)
- state = unmapped_ti.state
- elif not total_length:
+ zero_index_ti_exists = session.query(
+ exists().where(
+ TaskInstance.dag_id == self.dag_id,
+ TaskInstance.task_id == self.task_id,
+ TaskInstance.run_id == run_id,
+ TaskInstance.map_index == 0,
+ )
+ ).scalar()
Review Comment:
IIRC `EXISTS` has some compatibility issues across databases (don’t remember what exactly), so we generally use `query(count())...scalar() > 0` instead.
--
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] uranusjr commented on a diff in pull request #27834: Make sure we can get out of a faulty scheduler state
Posted by GitBox <gi...@apache.org>.
uranusjr commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031924843
##########
airflow/models/dagrun.py:
##########
@@ -780,8 +780,7 @@ def _expand_mapped_task_if_needed(ti: TI) -> Iterable[TI] | None:
except NotMapped: # Not a mapped task, nothing needed.
return None
if expanded_tis:
- assert expanded_tis[0] is ti
- return expanded_tis[1:]
+ return expanded_tis
Review Comment:
Since this function only returns _new_ ti objects, should we do something like this?
```python
if expanded_tis[0] is ti:
return expanded_tis[1:]
return expanded_tis
```
--
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