SubDag Operator broken when used with pools - infinite retry of tasks

[syvineckruyk]

Dear Airflow Maintainers,

Before I tell you about my issue, let me describe my environment:

Environment

Centos 6
CeleryExecutor (redis)
Two workers, 1 scheduler, 1 webserver
MySql

• Version of Airflow 1.6.2.10, 1.7.0rc1, HEAD
• Example code to reproduce the bug (as a code snippet in markdown)

from datetime import datetime, timedelta
from airflow.models import DAG
from airflow.operators import BashOperator, SubDagOperator



dag_name = 'test_18'
test_pool = 'test_pool'
email_to = 'your@email.com'

now = datetime.now()
start_time = datetime(now.year, now.month, now.day, 12, 0, 0)
start_time = start_time + timedelta(days=-1)

default_args = {
    'owner': 'Test',
    'depends_on_past': True,
    'start_date': start_time,
    'email': [email_to],
    'email_on_failure': True,
    'email_on_retry': True,
    'wait_for_downstream': False,
}

# Create the dag object
dag = DAG(dag_name,
          default_args=default_args,
          schedule_interval='0 * * * *')


def get_subdag(dag, sd_id, pool=None):
    subdag = DAG(
        dag_id='{parent_dag}.{sd_id}'.format(
            parent_dag=dag.dag_id,
            sd_id=sd_id),
        params=dag.params,
        default_args=dag.default_args,
        template_searchpath=dag.template_searchpath,
        user_defined_macros=dag.user_defined_macros,
    )

    t1 = BashOperator(
        task_id='test_task',
        bash_command='echo "hello" && sleep 10',
        dag=subdag,
        pool=pool
    )

    t2 = BashOperator(
        task_id='test_task_2_with_exit_1',
        bash_command='echo "hello" && sleep 10 && exit 1',
        dag=subdag,
        pool=pool
    )

    t2.set_upstream(t1)

    sdo = SubDagOperator(
        task_id=sd_id,
        subdag=subdag,
        retries=2,
        retry_delay=timedelta(seconds=30),
        dag=dag,
        depends_on_past=True,
    )

    return sdo

sd1 = get_subdag(dag, sd_id='test_subdag')
sd2 = get_subdag(dag, sd_id='test_subdag_with_pool', pool=test_pool)

• Screen shots of your DAG's graph and tree views:

1. sub-task of non pooled subdag fails
   
2. subdag enters retry state
   
3. after all retries exhausted non pooled subdag fails
   
4. sub-task of pooled subdag fails
   
5. pooled subdag remains in running state, sub-task of pooled subdag enters queued state
   
6. sub-task of pooled subdag enters running state
   
7. sub-task failures of pooled subdags generate email alerts for all failures
   
8. logs (from UI) also show observed behavior

Id  Dttm    Dag Id  Task Id Event   Execution Date  Owner   Extra
1910    03-27T15:37:36  test_18.test_subdag test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1911    03-27T15:37:46  test_18.test_subdag test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1916    03-27T15:39:18  test_18.test_subdag test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1918    03-27T15:39:28  test_18.test_subdag test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1924    03-27T15:41:18  test_18.test_subdag test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1930    03-27T15:41:28  test_18.test_subdag test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1912    03-27T15:37:54  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1914    03-27T15:38:05  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1917    03-27T15:39:23  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1920    03-27T15:39:33  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1921    03-27T15:40:54  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1922    03-27T15:41:04  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1925    03-27T15:41:26  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1926    03-27T15:41:36  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1927    03-27T15:41:52  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1928    03-27T15:42:02  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1929    03-27T15:42:18  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1931    03-27T15:42:28  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1933    03-27T15:43:52  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1934    03-27T15:44:02  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1935    03-27T15:44:30  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1936    03-27T15:44:40  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1937    03-27T15:44:56  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1938    03-27T15:45:06  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1939    03-27T15:46:35  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1940    03-27T15:46:45  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1941    03-27T15:47:42  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1942    03-27T15:47:52  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
1943    03-27T15:49:20  test_18.test_subdag_with_pool   test_task_2_with_exit_1 running 03-26T12:00:00  Test    
1944    03-27T15:49:30  test_18.test_subdag_with_pool   test_task_2_with_exit_1 failed  03-26T12:00:00  Test    
...

• Operating System: (Windows Version or $ uname -a)
  Linux **** 2.6.32-431.29.2.el6.x86_64 Improving the search functionality in the graph view #1 SMP Tue Sep 9 21:36:05 UTC 2014 x86_64 x86_64 x86_64 GNU/Linux
• Python Version: $ python --version
  Python 2.7.10

Now that you know a little about me, let me tell you about the issue I am having:

Description of Issue

When subdag operators are used with pools, sub-task failures are not "bubbled" up to the subdag operator. The contained failed task will retry indefinitely even when retries is set to 0.

When a subtask failure occurs the subdag operator remains in the running state, the sub-task quickly enters a queued state, and shortly there after re-enters a running state.

• What did you expect to happen?
• sub-task fails
• subdag operator enters retry state
• subdag operator enters running state when retry_delay has been met
• sub-tasks get queued
• sub-tasks enter running state
• What happened instead?
• sub-task fails
• sub-task enters queued state
• sub-task runs
• rinse, repeat
• Here is how you can reproduce this issue on your machine:

  Reproduction Steps

• create a pool
• set the email and pool in included DAG code
• submit to your dags folder
• observe non pooled subdag handle failure as specified
• observe pooled subdag enter infinite loop

This behavior has been observed on multiple versions. I did the testing for submitting this issue on 1.7.0rc1.

I also tested on HEAD 2016-03-26. The issue seems to get worst. The subdag which is not pooled (which works as expected on 1.7.0rc1) enters the retry state as it should .. but then never re-enters a running state. Hanging indefinitely.

[jlowin]

@syvineckruyk Thanks so much for taking the time to fill out such a complete report. I believed I've identified the issue, but would you mind verifying it? The fix is in my branch, so you can either clone it or you can install it directly like this:

pip install git+https://github.com/jlowin/airflow@subdag_pool_issue_1225

This branch includes the fixes from #1220, so we'd have to merge that first, but I think it's just about ready to go. I don't think #1220 directly impacts your issue, but it touches the same code so I'd like to avoid a messy conflict. Please let me know if you have any issues getting it running -- if this works for you I'll put through the PR.

If you're curious, here's what I believe is going on:

• When you run a task in a pool, it doesn't actually run right away; first it becomes queued. This means that running a task in a pool actually requires executing the run command at least twice: first to place it in a queue, and second to move it from the queue to the executor.
• When you run a SchedulerJob, it constantly checks for queued tasks and schedules them appropriately, so the "run it twice" problem is taken care of.
• The catch is that SubDagOperators don't submit their subdags to the scheduler. Instead, they kick them off with a BackfillJob. And BackfillJob had a little logic hole that would never run queued tasks the critical second time. Instead, queued tasks would just sit in the Backfill loop forever and the backfill job would never end.

[syvineckruyk]

@jlowin Awesome ... I'll give it a go ... will update later today

[syvineckruyk]

@jlowin still getting some weird (but different) behavior ... the subdag with pool did what it should the first time but then actually succeeded which should be impossible as the task returns an exit code of 1. I need a bit more time for a better analysis.

[jlowin]

Strange! These subdags might be more trouble than they're worth :) @syvineckruyk instead of watching your DAG through the scheduler and airflow UI, could you just try kicking off a backfill and see if it completes or hangs (and if it properly sets the task states -- you can check those in the UI)? I want to try to minimize any other variables that could be affecting things.

airflow backfill test_18 -s 2016-03-25 -e 2016-03-25

(note with your settings this will kick off more than one run -- you might want to set your interval to 1 day for simplicity)

It's a more controllable way to see what's happening than the Scheduler and in your case should give the same result because the subdag is ultimately just running that exact code.

Sorry for the iterative debugging process -- if you post any modifications you make to your code I'll run it as well.

[jlowin]

If I run that backfill command (with a fresh resetdb and a new test_pool with 10 slots), I get failures as expected when I check the UI after it finishes:

That's Python 2.7.11 using the branch I linked above. Next I'll run it in the scheduler, and let's see if we can figure out what's going on.

I modified your code slightly to make it run faster (literally, shortened the sleep and retry delays)

from datetime import datetime, timedelta
from airflow.models import DAG
from airflow.operators import BashOperator, SubDagOperator



dag_name = 'test_18'
test_pool = 'test_pool'

start_time = datetime(2016, 3, 25)

default_args = {
    'owner': 'Test',
    'depends_on_past': True,
    'start_date': start_time,
    # 'email': [email_to],
    # 'email_on_failure': True,
    # 'email_on_retry': True,
    # 'wait_for_downstream': False,
}

# Create the dag object
dag = DAG(dag_name,
          default_args=default_args,
          schedule_interval=timedelta(days=1))


def get_subdag(dag, sd_id, pool=None):
    subdag = DAG(
        dag_id='{parent_dag}.{sd_id}'.format(
            parent_dag=dag.dag_id,
            sd_id=sd_id),
        params=dag.params,
        default_args=dag.default_args,
        template_searchpath=dag.template_searchpath,
        user_defined_macros=dag.user_defined_macros,
    )

    t1 = BashOperator(
        task_id='test_task',
        bash_command='echo "hello" && sleep 1',
        dag=subdag,
        pool=pool
    )

    t2 = BashOperator(
        task_id='test_task_2_with_exit_1',
        bash_command='echo "hello" && sleep 1 && exit 1',
        dag=subdag,
        pool=pool
    )

    t2.set_upstream(t1)

    sdo = SubDagOperator(
        task_id=sd_id,
        subdag=subdag,
        retries=1,
        retry_delay=timedelta(seconds=3),
        dag=dag,
        depends_on_past=True,
    )

    return sdo

sd1 = get_subdag(dag, sd_id='test_subdag')
sd2 = get_subdag(dag, sd_id='test_subdag_with_pool', pool=test_pool)

[jlowin]

Running with the scheduler (and full airflow resetdb), I also get the expected failures (and I think nothing else runs because depends_on_past=True):

Please let me know if you think something else is going on -- I definitely want to get to the bottom of this.

[syvineckruyk]

@jlowin just got through a clean run ... (after resetdb) ... I want to perform a couple more tests/runs to make sure the behavior is consistent as I cannot explain my previous runs that had issues on your branch.

[syvineckruyk]

@jlowin still some issues happening. So my current testing round is based on issuing backfill from the cli .. but i suspect the behavior is the same from the scheduler.

I ran
airflow backfill test_25 -s 2016-03-25 -e 2016-03-25
an it worked as expected ... got the same result as you.
Then I ran
airflow backfill test_25 -s 2016-03-26 -e 2016-03-26
and got a weird condition which should not be possible

The tasks in retry (stuck there btw)... Have retries set to 0 ... so this should not be possible. Also as you can see the subdag operators that are parent to the sub-tasks are in success ... so i think we both saw clean runs due to reset db .. but something is still off if there are previous dag runs / task instances.

Thanks for your help with all of this and let me know if you need anything from me.

[jlowin]

@syvineckruyk is test_25 the code I posted?

[syvineckruyk]

@jlowin yeah .. im bumping up the version for clean logs.

[jlowin]

@syvineckruyk unfortunately I can't reproduce that weird case in your screenshot. I'm putting the exact
code I'm running at the bottom of this message. Thanks for your patience trying to solve this. If I had a gold medal to hand out for the "issue that identifies the most problems at once", you'd get it!

I run:

airflow resetdb -y
airflow backfill test_25 -s 2016-03-25 -e 2016-03-25
airflow backfill test_25 -s 2016-03-26 -e 2016-03-26

The 3/26 backfill simply doesn't run because its dependencies haven't been met. That's the correct behavior, since depends_on_past=True and the previous day failed.

(Something weird happens that I do need to fix but I don't think it's what you're seeing. When a task doesn't run because of dependencies, the command exits without failing. Therefore, the executor thinks it succeeded. But the task doesn't report a success, and so the executor concludes that something went wrong, puts up the "The airflow run command failed at reporting an error message" and, after three loops, fails the task. So you'll see the 3/26 task show up as failed with 0 try_number... that's wrong.)

I've updated my branch to pull the latest fixes from master -- do you mind checking it out again?

In my dags folder:

from datetime import datetime, timedelta
from airflow.models import DAG, Pool
from airflow.operators import BashOperator, SubDagOperator

import airflow
session = airflow.settings.Session()

pool = (
    session.query(Pool)
    .filter(Pool.pool == 'test_pool')
    .first())
if not pool:
    session.add(Pool(pool='test_pool', slots=10))
    session.commit()


dag_name = 'test_25'
test_pool = 'test_pool'

start_time = datetime(2016, 3, 25)

default_args = {
    'owner': 'Test',
    'depends_on_past': True,
    'start_date': start_time,
    # 'email': [email_to],
    # 'email_on_failure': True,
    # 'email_on_retry': True,
    # 'wait_for_downstream': False,
}

# Create the dag object
dag = DAG(dag_name,
          default_args=default_args,
          schedule_interval=timedelta(days=1))


def get_subdag(dag, sd_id, pool=None):
    subdag = DAG(
        dag_id='{parent_dag}.{sd_id}'.format(
            parent_dag=dag.dag_id,
            sd_id=sd_id),
        params=dag.params,
        default_args=dag.default_args,
        template_searchpath=dag.template_searchpath,
        user_defined_macros=dag.user_defined_macros,
    )

    t1 = BashOperator(
        task_id='test_task',
        bash_command='echo "hello" && sleep 0.2',
        dag=subdag,
        pool=pool
    )

    t2 = BashOperator(
        task_id='test_task_2_with_exit_1',
        bash_command='echo "hello" && sleep 0.2 && exit 1',
        dag=subdag,
        pool=pool
    )

    t2.set_upstream(t1)

    sdo = SubDagOperator(
        task_id=sd_id,
        subdag=subdag,
        retries=0,
        retry_delay=timedelta(seconds=0.2),
        dag=dag,
        depends_on_past=True,
    )

    return sdo

# sd1 = get_subdag(dag, sd_id='test_subdag')
sd2 = get_subdag(dag, sd_id='test_subdag_with_pool', pool=test_pool)