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/01 18:00:22 UTC
[GitHub] [airflow] matthewblock opened a new issue, #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
matthewblock opened a new issue, #27449:
URL: https://github.com/apache/airflow/issues/27449
### Apache Airflow version
2.4.2
### What happened
We have dynamic tasks that use `partial`/`expand`. These tasks are within task groups and have upstream tasks.
Occasionally these dynamic tasks are marked as state `upstream_failed` even though none of their upstream tasks and `upstream_failed` or `failed`. This occurs before their upstream tasks complete.
All of the dynamic tasks have the default trigger rule `all_success`.
Note that this occurs sporadically - some dynamic tasks _do_ complete successfully, and it appears to be random per DAG run.
<img width="1220" alt="dag_deid" src="https://user-images.githubusercontent.com/10802053/199304550-30a2d731-9e05-44bf-a472-27b1e4addd0d.png">
### What you think should happen instead
The dynamic tasks should successfully execute.
The tasks immediately upstream of these `upstream_failed` tasks have this near the end of their logs:
```
[2022-11-01, 09:15:31 PDT] {local_task_job.py:273} INFO - 0 downstream tasks scheduled from follow-on schedule check
```
### How to reproduce
Difficult to reproduce, working on a test DAG.
* Run a DAG that has dynamic tasks with trigger rule `all_success`
* Note that occasionally dynamic tasks get marked as `upstream_failed`
### Operating System
debian buster
### Versions of Apache Airflow Providers
_No response_
### Deployment
Docker-Compose
### Deployment details
_No response_
### Anything else
Some dynamic tasks successfully complete and some get marked as `upstream_failed`. It appears to be random per DAG run.
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1303252721
Oh and the removal is not a bug (at least not in the parital_subset function itself): because `include_direct_upstream=True` that function only goes one level up from `last_task`.
--
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] matthewblock commented on issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
matthewblock commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1301198284
A few other details I discovered while debugging:
- The "first" (most upstream) `upstream_failed` tasks have no `start_date` or `end_date` but the tasks immediately downstream of them that are also `upstream_failed` _do_ have `start_date` and `end_date`
- I noticed in another task that ran successfully, that is in parallel to the `upstream_failed` tasks, log this:
```
[2022-11-01, 09:15:25 PDT] {mappedoperator.py:622} INFO - Cannot expand <Mapped(MyCustomOperator): my_custom_task_1> for run manual__2022-11-01T16:06:39.483587+00:00; missing upstream values: ['my_xcom_value']
[2022-11-01, 09:15:25 PDT] {mappedoperator.py:622} INFO - Cannot expand <Mapped(MyCustomOperator): my_custom_task_2> for run manual__2022-11-01T16:06:39.483587+00:00; missing upstream values: ['my_xcom_value']
```
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1299753370
Can you share the dag code? You can redact some values. Without sharing your code it might be difficult to understand what's going on
--
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] ashb commented on issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1303620967
Such a change shouldn't be made without measuring the performance before and after. Yes removing it would make things simpler, but I think a tiny bit of complexity is worth it for the speed boost. (I can't remember the figures off the top of my head, but I know we measured it before adding this, and it did make a difference to the "inter-task lag")
The place where this change really helps is when you have a large number (500-1000+) of DAGs, as this mini scheduler only locks the _current_ dag, leaving the scheduler on to do other DAGs.
--
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] ashb commented on issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ashb commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1303250538
Okay, so the problem here is, as Ephraim identified, that the mini scheduler operates on a subset of the DAG, not the complete dag.
So when `first_task` finishes we get a partial DAG containing the following tasks: `first_task, last_task, middle_task`. Not crucially that we _dont_ have `second_task`.
The reason we use a partial subset rather than the whole dag was a performance optimization. Essentially: this task just finished, lets look at the downstream task and see if any of them can be scheduled. But in order to check if those can be scheduled, we need the upstream of _those_ tasks (which is how we get to include `middle_task`).
So I think we have two options here:
1) Remove `include_direct_upstream` so that the partial dag includes _all_ upstreams. Con to this: mini scheduler does more work (but likely not all that much more)
2) Change the expansion so that we don't fail mapped tasks if we can't expand when `self.dag.partial is True`?
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
potiuk commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1303554730
Wild idea.
This was a loose thought I have for quite some time while reviewing some of issues connected to mini-scheduler. I think the mini-scheduler "approach" was a good idea - triggering direct downstream task when task is finished as soon as possible is generally cool. However it introduces complexity (similar to above) because it is IMHO a tea that is "almost, but not quite, entirely unlike tea" to quote Douglas Adams. I looks like scheduler scheduling tasks, it behaves like that, but there are some small quirks (like above) that make it susceptible to subtle bugs.
Also I think the original idea that the mini-scheduler will be run in a more "distributed" fashion is a bit hampered by the fact that bulk of what the mini-scheduler does is done via the DB anyway and it synchronizes on DagRun lock and DB operation so pretty much all the "distribution" benefits are all but gone. Havint multiple schedulers already provides a way to "distribute" scheduling and distributing it even more does not change much. Yes, it uses the fact that the DAG is already loaded in memory and some of the DB objects are effectively cached by SQLAlchemy but I think with 2.4 and the "micro-pipelines" approach where our users have much better ways to split their DAGs into smaller, independent DAGs, this is becomes far less of an issue.
I thought that we could implement "mini-scheduling" slightly differently. Rather than actually do the scheduling, we could add DagRun of just completed task to a some kind of table where we could keep "priority" DagRuns to schedule by a "real" scheduler. Then it would be pretty immediately picked up by (one of) the schedulers at the next scheduling loop and scheduled. The latency there would be slightly bigger, but not that much IMHO and we would loose the "DAG structure in-memory", but I think it would make it a bit more robust - especially when we start introducing new concepts.
It's not entirely hashed out - there are probably some edge cases where for example we have a lot of those "priority" dagruns to process (what should we do with the non-priority ones for example).
I am mentioning it here because I think such an approach would pretty much automatically solve the above problem.
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1302266697
A
> @eladkal - I corrected my test case with `op_kwargs`.
>
> Do you have `AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION` set to `True` or `False`? The default value `True` causes this error, but setting it to `False` prevents the mini scheduler from forcing an expansion too early, and the DAG succeeds.
I used the default for `AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION` which is `True`.
What database are you using? Did you set any other environment variable other than the above? I can't reproduce it running your updated dag
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1302284419
> A
>
> > @eladkal - I corrected my test case with `op_kwargs`.
> > Do you have `AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION` set to `True` or `False`? The default value `True` causes this error, but setting it to `False` prevents the mini scheduler from forcing an expansion too early, and the DAG succeeds.
>
> I used the default for `AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION` which is `True`. What database are you using? Did you set any other environment variable other than the above? I can't reproduce it running your updated dag
>
> However, I see that in my case, the second task starts before the first task and they finish together
Reproduced with LocalExecutor!
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1303195096
> Something doesn’t click. Why does the non-expanded mapped task not have a proper `upstream_list`? If it’s not yet expanded, the task should be in the DAG, and has proper dependency information set.
It was removed here https://github.com/apache/airflow/blob/3aadc44a13d0d100778792691a0341818723c51c/airflow/models/dag.py#L2240-L2244
when `partial_subset` was called https://github.com/apache/airflow/blob/3aadc44a13d0d100778792691a0341818723c51c/airflow/jobs/local_task_job.py#L253-L258
The removal seems like a bug? maybe @ashb can chip in
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
potiuk commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1303646767
> Such a change shouldn't be made without measuring the performance before and after. Yes removing it would make things simpler, but I think a tiny bit of complexity is worth it for the speed boost. (I can't remember the figures off the top of my head, but I know we measured it before adding this, and it did make a difference to the "inter-task lag")
Absolutely - I think there might be some effect I have not thought about. Just wanted to bounce off of you the idea and thought I had. I do realise there are performance implications if we change it, just wanted to raise it as an option so that other might also hink about it and see as possibility :)
--
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 closed issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ephraimbuddy closed issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
URL: https://github.com/apache/airflow/issues/27449
--
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] romsla commented on issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
romsla commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1302276944
Having the same issue with airflow 2.4.1.
I got this bug on kubernetes installation with 8 schedulers on pgsql, and will try to reproduce the case myself later today.
The first task has to finish faster than the second one, this seems to be the prerequisite.
In my case the first task is using a cached function call, so first runs are ok. And sometimes the runs along the line are ok too.
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1303255803
Ok. Thanks Ash, I think option 2 is better:
> Change the expansion so that we don't fail mapped tasks if we can't expand when self.dag.partial is True?
Since it will limit the change to mapped task. Let me try 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] ephraimbuddy commented on issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1301706656
I was not able to reproduce with your test case. You also have an issue on the test case:
```python
last_task = PythonOperator(
task_id="last_task",
python_callable=nop,
)
```
should have `op_args` argument since `nop` takes an argument. so:
It should be this:
```python
last_task = PythonOperator(
task_id="last_task",
python_callable=nop,
op_args = [1]
)
```
for example
--
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] matthewblock commented on issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
matthewblock commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1302233394
@eladkal - I corrected my test case with `op_kwargs`.
Do you have `AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION` set to `True` or `False`? The default value `True` causes this error, but setting it to `False` prevents the mini scheduler from forcing an expansion too early, and the DAG succeeds.
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1302721277
Something doesn’t click. Why does the non-expanded mapped task not have a proper `upstream_list`? If it’s not yet expanded, the task should be in the DAG, and has proper dependency information set.
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1302463474
It looks like non-expanded mapped task instances don't have `upstream_list ` i.e `ti.task.upstream_list` is `[]`. I got the test case to work with this:
```diff
diff --git a/airflow/ti_deps/deps/trigger_rule_dep.py b/airflow/ti_deps/deps/trigger_rule_dep.py
index 39722c6df3..f50043c7b6 100644
--- a/airflow/ti_deps/deps/trigger_rule_dep.py
+++ b/airflow/ti_deps/deps/trigger_rule_dep.py
@@ -67,6 +67,9 @@ class TriggerRuleDep(BaseTIDep):
def _get_dep_statuses(self, ti, session, dep_context: DepContext):
# Checking that all upstream dependencies have succeeded
if not ti.task.upstream_list:
+ if ti.task.is_mapped and ti.map_index == -1:
+ yield self._failing_status(reason="Task is a mapped task and has not expanded.")
+ return
yield self._passing_status(reason="The task instance did not have any upstream tasks.")
return
```
cc: @uranusjr, would like to hear what you think about this
--
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] matthewblock commented on issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
matthewblock commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1301283384
> Can you share the dag code? You can redact some values. Without sharing your code it might be difficult to understand what's going on
Unfortunately it's proprietary code from my employer, hence the de-ID. I am attempting to create a new test DAG that I can share to reproduce the error.
--
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] matthewblock commented on issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
matthewblock commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1301463418
> > Can you share the dag code? You can redact some values. Without sharing your code it might be difficult to understand what's going on
>
> Unfortunately it's proprietary code from my employer, hence the de-ID. I am attempting to create a new test DAG that I can share to reproduce the error.
I was able to successfully create a test case. I also updated the issue language reflecting more research.
At this point I can't imagine why this behavior is desirable but it's likely I'm missing something.
--
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 issue #27449: Dynamic tasks marked as `upstream_failed` when none of their upstream tasks are `failed` or `upstream_failed`
Posted by GitBox <gi...@apache.org>.
uranusjr commented on issue #27449:
URL: https://github.com/apache/airflow/issues/27449#issuecomment-1299763219
Without looking at anything, one UPSTREAM_FAILED case specific to a mapped task is when the upstream pushes a non-mappable value (an int, for example). This feels likely for your case.
--
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