<h2 dir="auto">Environment</h2>
<ul dir="auto">
<li>OS: Debian Wheezy (via <a href="https://github.com/puckel/docker-airflow">docker-airflow</a>)</li>
<li><a href="https://gist.github.com/boxysean/5b5d73f4fc3541d63c4b#file-airflow-cfg">Configuration</a> (trimmed)</li>
<li>python 2.7.9, Airflow 1.6.1 from pypi, <a href="https://gist.github.com/boxysean/5b5d73f4fc3541d63c4b#file-requirements-txt">my project's requirements.txt</a> (trimmed)</li>
</ul>
<h2 dir="auto">How to replicate and explanation</h2>
<p dir="auto"><a href="https://gist.github.com/boxysean/5b5d73f4fc3541d63c4b#file-dag_that_crashes_scheduler-py">This DAG</a> 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.</p>
<p dir="auto">I find my DAGs start to be scheduled only after manually / externally triggering them. (Not sure why.) The manual trigger generates this DagRun:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="&lt;DagRun crash_dag @ 2015-12-17 16:28:34: test, externally triggered: True&gt;"><pre class="notranslate"><code class="notranslate">&lt;DagRun crash_dag @ 2015-12-17 16:28:34: test, externally triggered: True&gt;
</code></pre></div>
<p dir="auto">Then the subsequent scheduler heartbeats generates new scheduled DagRuns one at a time (one for each of the four previous minutes):</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="&lt;DagRun crash_dag @ 2015-12-17 16:24:34: scheduled__2015-12-17T16:24:34, externally triggered: False&gt;
&lt;DagRun crash_dag @ 2015-12-17 16:25:34: scheduled__2015-12-17T16:25:34, externally triggered: False&gt;
&lt;DagRun crash_dag @ 2015-12-17 16:26:34: scheduled__2015-12-17T16:26:34, externally triggered: False&gt;
&lt;DagRun crash_dag @ 2015-12-17 16:27:34: scheduled__2015-12-17T16:27:34, externally triggered: False&gt;"><pre class="notranslate"><code class="notranslate">&lt;DagRun crash_dag @ 2015-12-17 16:24:34: scheduled__2015-12-17T16:24:34, externally triggered: False&gt;
&lt;DagRun crash_dag @ 2015-12-17 16:25:34: scheduled__2015-12-17T16:25:34, externally triggered: False&gt;
&lt;DagRun crash_dag @ 2015-12-17 16:26:34: scheduled__2015-12-17T16:26:34, externally triggered: False&gt;
&lt;DagRun crash_dag @ 2015-12-17 16:27:34: scheduled__2015-12-17T16:27:34, externally triggered: False&gt;
</code></pre></div>
<p dir="auto">One minute passes, then it tries to schedule the next DagRun, but it crashes with this exception:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="2015-12-17 16:29:35,097 - root - ERROR - (_mysql_exceptions.IntegrityError) (1062, &quot;Duplicate entry 'crash_dag-2015-12-17 16:28:34' for key 'dag_id'&quot;) [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 &quot;/usr/local/lib/python2.7/dist-packages/airflow/jobs.py&quot;, line 608, in _execute
     self.schedule_dag(dag)
   File &quot;/usr/local/lib/python2.7/dist-packages/airflow/jobs.py&quot;, line 398, in schedule_dag
     session.commit()
   File &quot;/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/session.py&quot;, line 801, in commit
     self.transaction.commit()
&lt;snip&gt;"><pre class="notranslate"><code class="notranslate">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()
&lt;snip&gt;
</code></pre></div>
<p dir="auto">The reason is my Airflow database (tested on both MySQL and Postgres) has a unique key definition in the <code class="notranslate">dag_run</code> table on <code class="notranslate">(dag_id, execution_date)</code>. Indeed, the DagRun attempting to be scheduled with ID <code class="notranslate">scheduled__2015-12-17T16:28:34</code> is colliding with my DagRun with ID <code class="notranslate">test</code>.</p>
<p dir="auto">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:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="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, &quot;Duplicate entry 'crash_dag-2015-12-17 16:28:34' for key 'dag_id'&quot;) [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 &quot;/usr/local/lib/python2.7/dist-packages/airflow/jobs.py&quot;, line 618, in _execute
     self.import_errors(dagbag)
   File &quot;/usr/local/lib/python2.7/dist-packages/airflow/jobs.py&quot;, line 340, in import_errors
     session.query(models.ImportError).delete()
   File &quot;/usr/local/lib/python2.7/dist-packages/sqlalchemy/orm/query.py&quot;, line 2883, in delete
     delete_op.exec_()
&lt;snip&gt;"><pre class="notranslate"><code class="notranslate">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_()
&lt;snip&gt;
</code></pre></div>
<p dir="auto">Looking into the Airflow codebase, I see that <a href="https://github.com/airbnb/airflow/blob/master/airflow/models.py#L2721-L2723">the DagRun model has a single unique index</a> that is <em>not</em> the unique index causing this issue. There is, however, <a href="https://github.com/airbnb/airflow/blob/master/airflow/migrations/versions/1b38cef5b76e_add_dagrun.py#L29">a migration that adds the unique DagRun index</a> causing this issue.</p>
<h2 dir="auto">How to fix</h2>
<p dir="auto">I'm going to try to work-around by manually deleting the unique constraint on the <code class="notranslate">dag_run.(dag_id, execution_date)</code> index. I'm guessing it will work okay for now.</p>
<p dir="auto">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.</p>
<p dir="auto"><strong>Apache Airflow Version</strong>: 2.1.2</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li>OS: Linux</li>
</ul>
<p dir="auto"><strong>What Happens</strong>:</p>
<p dir="auto">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 <strong>airflow</strong> OS user.</p>
<p dir="auto">When the task is then run as another user (via <code class="notranslate">run_as_user='me'</code> ), <em>and the task fails</em>, 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 <strong>when the task fails, no one is notified</strong>.</p>
<p dir="auto"><strong>What you expect to happen</strong>:</p>
<p dir="auto">Airflow should handle error files without creating its own permission errors.</p>
<p dir="auto"><strong>Options to Correct</strong>:</p>
<ol dir="auto">
<li>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.</li>
</ol>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="  87 self._error_file = NamedTemporaryFile(delete=True)
  +  os.chmod(self._error_file.name, 0o666)
  88 self._cfg_path = cfg_path"><pre class="notranslate"><code class="notranslate">  87 self._error_file = NamedTemporaryFile(delete=True)
  +  os.chmod(self._error_file.name, 0o666)
  88 self._cfg_path = cfg_path
</code></pre></div>
<ol start="2" dir="auto">
<li>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.</li>
</ol>
<p dir="auto"><strong>Reference Line of Code:</strong></p>
<p dir="auto"></p><div class="Box Box--condensed my-2">
  <div class="Box-header f6">
    <p class="mb-0 text-bold">
      <a href="https://github.com/apache/airflow/blob/e107891e65432bf1ba1200c2ca744892db6622d5/airflow/task/task_runner/base_task_runner.py#L87">airflow/airflow/task/task_runner/base_task_runner.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
         Line 87
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/e107891e65432bf1ba1200c2ca744892db6622d5">e107891</a>
    </p>
  </div>
  <div itemprop="text" class="Box-body p-0 blob-wrapper blob-wrapper-embedded data">
    <table class="highlight tab-size mb-0 js-file-line-container" data-tab-size="8" data-paste-markdown-skip="">

        <tbody><tr class="border-0">
          <td id="L87" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="87"></td>
          <td id="LC87" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-s1">self</span>.<span class="pl-s1">_error_file</span> <span class="pl-c1">=</span> <span class="pl-v">NamedTemporaryFile</span>(<span class="pl-s1">delete</span><span class="pl-c1">=</span><span class="pl-c1">True</span>) </td>
        </tr>
    </tbody></table>
  </div>
</div>
<p></p>