Issue 1:
Environment

OS: Debian Wheezy (via docker-airflow)

Configuration (trimmed)

python 2.7.9, Airflow 1.6.1 from pypi, my project's requirements.txt (trimmed)

How to replicate and explanation

This DAG crashes my scheduler very quickly when manually (externally) triggered. I'm going to guess it will crash for everyone running Airflow 1.6.1 and up to master.

I find my DAGs start to be scheduled only after manually / externally triggering them. (Not sure why.) The manual trigger generates this DagRun:

<DagRun crash_dag @ 2015-12-17 16:28:34: test, externally triggered: True>

Then the subsequent scheduler heartbeats generates new scheduled DagRuns one at a time (one for each of the four previous minutes):

<DagRun crash_dag @ 2015-12-17 16:24:34: scheduled__2015-12-17T16:24:34, externally triggered: False>
<DagRun crash_dag @ 2015-12-17 16:25:34: scheduled__2015-12-17T16:25:34, externally triggered: False>
<DagRun crash_dag @ 2015-12-17 16:26:34: scheduled__2015-12-17T16:26:34, externally triggered: False>
<DagRun crash_dag @ 2015-12-17 16:27:34: scheduled__2015-12-17T16:27:34, externally triggered: False>

One minute passes, then it tries to schedule the next DagRun, but it crashes with this exception:

2015-12-17 16:29:35,097 - root - ERROR - (_mysql_exceptions.IntegrityError) (1062, "Duplicate entry 'crash_dag-2015-12-17 16:28:34' for key 'dag_id'") [SQL: u'INSERT INTO dag_run (dag_id, execution_date, state, run_id, external_trigger, conf) VALUES (%s, %s, %s, %s, %s, %s)'] [parameters: ('crash_dag', datetime.datetime(2015, 12, 17, 16, 28, 34), u'running', u'scheduled__2015-12-17T16:28:34', 0, None)]
Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/airflow/jobs.py", line 608, in _execute
self.schedule_dag(dag)
File "/usr/local/lib/python2.7/dist-packages/airflow/jobs.py", line 398, in schedule_dag
session.commit()
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py", line 801, in commit
self.transaction.commit()
<snip>

The reason is my Airflow database (tested on both MySQL and Postgres) has a unique key definition in the dag_run table on (dag_id, execution_date). Indeed, the DagRun attempting to be scheduled with ID scheduled__2015-12-17T16:28:34 is colliding with my DagRun with ID test.

After this collision and exception, the scheduler remains alive, and for each scheduler heartbeat, I get the following SQLA rollback message, preventing all future scheduling:

2015-12-17 16:29:35,115 - root - ERROR - This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (_mysql_exceptions.IntegrityError) (1062, "Duplicate entry 'crash_dag-2015-12-17 16:28:34' for key 'dag_id'") [SQL: u'INSERT INTO dag_run (dag_id, execution_date, state, run_id, external_trigger, conf) VALUES (%s, %s, %s, %s, %s, %s)'] [parameters: ('crash_dag', datetime.datetime(2015, 12, 17, 16, 28, 34), u'running', u'scheduled__2015-12-17T16:28:34', 0, None)]
Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/airflow/jobs.py", line 618, in _execute
self.import_errors(dagbag)
File "/usr/local/lib/python2.7/dist-packages/airflow/jobs.py", line 340, in import_errors
session.query(models.ImportError).delete()
File "/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/query.py", line 2883, in delete
delete_op.exec_()
<snip>

Looking into the Airflow codebase, I see that the DagRun model has a single unique index that is not the unique index causing this issue. There is, however, a migration that adds the unique DagRun index causing this issue.

How to fix

I'm going to try to work-around by manually deleting the unique constraint on the dag_run.(dag_id, execution_date) index. I'm guessing it will work okay for now.

I think the root cause is an inconsistency between the DagRun model and the list of migrations that needs to be resolved in such a way there isn't index collisions.

Issue 2:
Apache Airflow Version: 2.1.2

Environment:

OS: Linux

What Happens:

When the base_task_runner prepares to execute the task, it opens a temporary error file AND passes the error file name into the task. This temp file is created with default permissions based on the airflow worker, which is usually the airflow OS user.

When the task is then run as another user (via run_as_user='me' ), and the task fails, the task attempts to handle the failure by opening and writing to the error file within the subprocess, which is running as the subordinate user and not as the airflow user. This leads to a permission error on the attempt to open, and before the error email is sent out, which is particularly bad because when the task fails, no one is notified.

What you expect to happen:

Airflow should handle error files without creating its own permission errors.

Options to Correct:

We've temporarily gotten around this by inserting a chmod in line 88. This is a temporary error file that is delete when the task goes out of scope, and so, while ugly and inelegant, this is a simple way to avoid the identified problem and probably not a security risk.

87 self._error_file = NamedTemporaryFile(delete=True)
+ os.chmod(self._error_file.name, 0o666)
88 self._cfg_path = cfg_path

Alternatively, it looks like the there are few uses of the error file by the worker, and that the worker does not write to it. It is possible that the only use is in airflow.jobs.LocalTaskJob following task failure. An alternative solution would be to generate the temp error filename without actually creating the file, passing it to the subprocess as a parameter, and then opening and reading the file if it exists following task failure.

Reference Line of Code:

airflow/airflow/task/task_runner/base_task_runner.py

Line 87
in
e107891

self._error_file = NamedTemporaryFile(delete=True)