[jira] [Created] (AIRFLOW-2229) Scheduler cannot retry abrupt task failures within factory-generated DAGs

["James Meickle (JIRA)" <ji...@apache.org>]

James Meickle created AIRFLOW-2229:
--------------------------------------

             Summary: Scheduler cannot retry abrupt task failures within factory-generated DAGs
                 Key: AIRFLOW-2229
                 URL: https://issues.apache.org/jira/browse/AIRFLOW-2229
             Project: Apache Airflow
          Issue Type: Bug
          Components: scheduler
    Affects Versions: 1.9.0
            Reporter: James Meickle


We had an issue where one of our tasks failed without the worker updating state (unclear why, but let's assume it was an OOM), resulting in this series of error messages:

{{Mar 20 14:27:05 airflow-core-i-0fc1f995414837b8b.stg.int.dynoquant.com airflow_scheduler-stdout.log: [2018-03-20 14:27:04,993] \{{models.py:1595}} ERROR - Executor reports task instance %s finished (%s) although the task says its %s. Was the task killed externally?}}
{{Mar 20 14:27:05 airflow-core-i-0fc1f995414837b8b.stg.int.dynoquant.com airflow_scheduler-stdout.log: NoneType}}
{{Mar 20 14:27:05 airflow-core-i-0fc1f995414837b8b.stg.int.dynoquant.com airflow_scheduler-stdout.log: [2018-03-20 14:27:04,994] \{{jobs.py:1435}} ERROR - Cannot load the dag bag to handle failure for <TaskInstance: nightly_dataload.dummy_operator 2018-03-19 00:00:00 [queued]>. Setting task to FAILED without callbacks or retries. Do you have enough resources?}}

Mysterious failures are not unexpected, because we are in the cloud, after all. The concern is the last line: ignoring callbacks and retries, implying that it's a lack of resources. However, the machine was totally underutilized at the time.

I dug into this code a bit more and as far as I can tell this error is happening in this code path: [https://github.com/apache/incubator-airflow/blob/1.9.0/airflow/jobs.py#L1427]

{{self.log.error(msg)}}
{{try:}}
{{    simple_dag = simple_dag_bag.get_dag(dag_id)}}
{{    dagbag = models.DagBag(simple_dag.full_filepath)}}
{{    dag = dagbag.get_dag(dag_id)}}
{{    ti.task = dag.get_task(task_id)}}
{{    ti.handle_failure(msg)}}
{{except Exception:}}
{{    self.log.error("Cannot load the dag bag to handle failure for %s"}}
{{    ". Setting task to FAILED without callbacks or "}}
{{    "retries. Do you have enough resources?", ti)}}
{{    ti.state = State.FAILED}}
{{    session.merge(ti)}}
{{    session.commit()}}{{}}

I am not very familiar with this code, nor do I have time to attach a debugger at the moment, but I think what is happening here is:
 * I have a factory Python file, which imports and instantiates DAG code from other files.
 * The scheduler loads the DAGs from the factory file on the filesystem. It gets a fileloc (as represented in the DB) not of the factory file, but of the file it loaded code from.
 * The scheduler makes a simple DAGBag from the instantiated DAGs.
 * This line of code uses the simple DAG, which references the original DAG object's fileloc, to create a new DAGBag object.
 * This DAGBag looks for the original DAG in the fileloc, which is the file containing that DAG's _code_, but is not actually importable by Airflow.
 * An exception is raised trying to load the DAG from the DAGBag, which found nothing.
 * Handling of the task failure never occurs.
 * The over-broad Exception code swallows all of the above occurring.
 * There's just a generic error message that is not helpful to a system operator.

If this is the case, at minimum, the try/except should be rewritten to be more graceful and to have a better error message. But I question whether this level of DAGBag abstraction/indirection isn't making this failure case worse than it needs to be; under normal conditions the scheduler is definitely able to find the relevant factory-generated DAGs and execute tasks within them as expected, even with the fileloc set to the code path and not the import path.

 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)