Using depends_on_past in conjunction with trigger_rule='all_done'

[timeisapear]

It seems that depends_on_past will override a task's trigger_rule='all_done' if the task instance of the previous day has failed. Is that the case and expected behavior?

[mistercrunch]

depends_on_past mean "Don't move forward if the previous schedule has not succeeded, unless this current run is the the start_date". The logic that should be applied is : depends_on_past AND trigger_rule, is that not what you see?

[mistercrunch]

BTW when I designed trigger_rule I thought about a much more complex design where many trigger rules could be applied and potentially AND/OR and parenthesis could have been used to combine the rules. depends_on_past would have then become just another trigger rule that could have been applied as part of a more complex combination or rules.

But I ended up keeping it simple as we've been solving 90% of cases with just "all_done" and the current options probably bring us to 99.x% of our use cases..

[timeisapear]

The logic in using trigger_rule AND depends_on_past I was hoping for in our workflow would be: schedule today's task if the previous day's task instance is in a state defined in the current day's trigger_rule e.g. failed and not necessarily succeeded. That might not be desired for most people so I'm not surprised I don't see that behavior.

On a related note, we've been testing mostly with airflow backfill CLI and I've had trouble using merely trigger_rule='all_done' on a task downstream of multiple dependencies. If the tasks upstream of the trigger_rule='all_done' fail, they retry a few times and the relaxed trigger state task gets skipped. (note all my testing has been done off a pip install version of a few weeks ago.)

Our dag: even if all 90 anom tasks failed today and yesterday, we'd want dri to run. Also would prefer dri for today ran before yesterday's.

[mistercrunch]

I'm not sure if I understand "they retry a few times and the relaxed trigger state task gets skipped"

I just read the code to review the definition of all_done and right now it correspond to either: SUCCESS, FAILED, UPSTREAM_FAILED, SKIPPED. If all the task instances are in one of those states, only then the downstream task should trigger (well if depends_on_past and wait_for_downstream are met, independently from this). Let me know if you've observed otherwise or if you think the logic should be changed.

[timeisapear]

So I think I've narrowed down an unexpected behavior on my end.

When executing a backfill, if there is a task upstream in a failed state, the backfill correctly triggers the downstream tasks with the specified trigger_rule='all done'. However when the upstream task fails again after a retry (I've forced this for testing purposes) the following occurs:

2015-10-05 10:40:52,010 INFO - [backfill progress] waiting: 17 | succeeded: 245 | kicked_off: 68 | failed: 0 | skipped: 0
2015-10-05 10:40:52,061 INFO - Adding to queue: airflow run one_day anom14 2013-02-01T00:00:00   --local     -sd DAGS_FOLDER/one_day.py -s 2013-02-01T00:00:00
2015-10-05 10:40:57,010 INFO - LocalWorker running airflow run one_day anom14 2013-02-
2015-10-05 10:40:57,010 INFO - LocalWorker running airflow run one_day anom23 2013-02-01T00:00:00   --local     -sd DAGS_FOLDER/one_day.py -s 2013-02-01T00:00:00
2015-10-05 10:40:57,012 ERROR - Task instance ('one_day', 'opc14', datetime.datetime(2013, 2, 1, 0, 0)) failed
2015-10-05 10:40:57,014 INFO - [backfill progress] waiting: 0 | succeeded: 252 | kicked_off: 68 | failed: 1 | skipped: 9

The backfill throws an exception, immediately skips all the still-unscheduled tasks and quits with error:

Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 10, in <module>
    args.func(args)
  File "/usr/local/lib/python2.7/site-packages/airflow/bin/cli.py", line 66, in backfill
    ignore_dependencies=args.ignore_dependencies)
  File "/usr/local/lib/python2.7/site-packages/airflow/models.py", line 2195, in run
    job.run()
  File "/usr/local/lib/python2.7/site-packages/airflow/jobs.py", line 159, in run
    self._execute()
  File "/usr/local/lib/python2.7/site-packages/airflow/jobs.py", line 659, in _execute
    "Some tasks instances failed, here's the list:\n"+str(failed))
airflow.utils.AirflowException: Some tasks instances failed, here's the list:
[('one_day', 'opc14', datetime.datetime(2013, 2, 1, 0, 0))]

Thanks for all your thorough responses so far.

[timeisapear]

I found this code in airflow.jobs.py in https://github.com/airbnb/airflow/blob/master/airflow/jobs.py#L667-L681.

A downstream task with a depends_on_false='all_done' should not be skipped, correct? This occurs during a backfill.

if ti.state == State.FAILED:
    failed.append(key)
    logging.error("Task instance " + str(key) + " failed")
    del tasks_to_run[key]
    # Removing downstream tasks from the one that has failed
    for t in self.dag.get_task(task_id).get_flat_relatives(
            upstream=False):
        key = (ti.dag_id, t.task_id, execution_date)
        if key in tasks_to_run:
            wont_run.append(key)
            del tasks_to_run[key]

[sv3ndk]

Hi @timeisapear and @mistercrunch,

I believe @thoralf-gutierrez and myself have found an easy way to reproduce this issue as part as the unit tests.

We created a small "end-to-end dag tester" that rely on a BackFillJob to execute a DAG and then validates some post-conditions. We believe this is already super useful now and can grow into a great tool that continuously validates the robustness of Airflow! The corresponding PR together with some examples of sucessfull tests is here: #670.

We used this framework to test the trigger rules ALL_FAILED, ALL_DONE and ONE_FAILED (when executed in a BackFillJob) and noted that none of them trigger the execution of the corresponding task since the BackFillJob systematically removes it from the tasks_to_run list, in the code identified by @timeisapear above.

We're not submitting these 3 tests in a PR yet since they would break the build, though the code and the corresponding Travis build failure on our fork are visible below:

https://github.com/RealImpactAnalytics/airflow/tree/testing_trigger_rules_v3
https://travis-ci.org/RealImpactAnalytics/airflow/jobs/94600932

We suggest the following resolution path:

1. please review End to end DAG testing #670 and let us know if there's anything more we should do in order to have this validation framework included the master branch

2. we could then submit another PR with the 3 supplementary tests above + a fix that, as suggested by @timeisapear, would simply allow downstream tasks to stay in the tasks_to_run list if they can be triggered by failures.

What do you think ?

[r39132]

@svendx4f Please submit a PR!

[sv3ndk]

Thanks @r39132. There is an open PR regarding the end to end testing, which is the first necessary step in order to reproduce everything described above.

The PR is here: #748

It's regularly consuming energy on our side to rebase it and keep it current, so any support and lobbying to get it approved and merged is welcome :)

[r39132]

Thx. Will do.. let me share it with the committers and ask for thoughts and feedback. BTW, you can also do the same on the gitter channel if you feel your PRs are getting lost in the noise.

[bolkedebruin]

@r39132 or @svendx4f can you please convert this to a Jira issue and close this after warts