<p dir="auto">This test fails sometimes:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="___________ TestLocalTaskJob.test_process_sigterm_works_with_retries ___________
  
  self = &lt;tests.jobs.test_local_task_job.TestLocalTaskJob object at 0x7f3d380ae1f0&gt;
  dag_maker = &lt;tests.conftest.dag_maker.&lt;locals&gt;.DagFactory object at 0x7f3d38917760&gt;
  
      def test_process_sigterm_works_with_retries(self, dag_maker):
          &quot;&quot;&quot;
          Test that ensures that task runner sets tasks to retry when they(task runner)
           receive sigterm
          &quot;&quot;&quot;
          # use shared memory value so we can properly track value change even if
          # it's been updated across processes.
          retry_callback_called = Value('i', 0)
          task_terminated_externally = Value('i', 1)
          shared_mem_lock = Lock()
      
          def retry_callback(context):
              with shared_mem_lock:
                  retry_callback_called.value += 1
              assert context['dag_run'].dag_id == 'test_mark_failure_2'
      
          def task_function(ti):
              time.sleep(60)
              # This should not happen -- the state change should be noticed and the task should get killed
              with shared_mem_lock:
                  task_terminated_externally.value = 0
      
          with dag_maker(dag_id='test_mark_failure_2'):
              task = PythonOperator(
                  task_id='test_on_failure',
                  python_callable=task_function,
                  retries=1,
                  retry_delay=timedelta(seconds=2),
                  on_retry_callback=retry_callback,
              )
          ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)
          ti.refresh_from_db()
          job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
          job1.task_runner = StandardTaskRunner(job1)
          job1.task_runner.start()
          settings.engine.dispose()
          process = multiprocessing.Process(target=job1.run)
          process.start()
          for _ in range(0, 25):
              ti.refresh_from_db()
              if ti.state == State.RUNNING and ti.pid is not None:
                  break
              time.sleep(0.2)
          os.kill(process.pid, signal.SIGTERM)
          process.join(timeout=10)
          ti.refresh_from_db()
  &gt;       assert ti.state == State.UP_FOR_RETRY
  E       AssertionError: assert None == &lt;TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'&gt;
  E        +  where None = &lt;TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01 00:00:00+00:00 [None]&gt;.state
  E        +  and   &lt;TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'&gt; = State.UP_FOR_RETRY
  
  tests/jobs/test_local_task_job.py:828: AssertionError
  ----------------------------- Captured stdout call -----------------------------
  Running &lt;TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01T00:00:00+00:00 [None]&gt; on host 67aa517c450e
  ----------------------------- Captured stderr call -----------------------------
  Process Process-113:
  Traceback (most recent call last):
    File &quot;/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py&quot;, line 1276, in _execute_context
      self.dialect.do_execute(
    File &quot;/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/default.py&quot;, line 608, in do_execute
      cursor.execute(statement, parameters)
    File &quot;/usr/local/lib/python3.8/site-packages/MySQLdb/cursors.py&quot;, line 206, in execute
      res = self._query(query)
    File &quot;/usr/local/lib/python3.8/site-packages/MySQLdb/cursors.py&quot;, line 319, in _query
      db.query(q)
    File &quot;/usr/local/lib/python3.8/site-packages/MySQLdb/connections.py&quot;, line 259, in query
      _mysql.connection.query(self, query)
  MySQLdb._exceptions.IntegrityError: (1062, &quot;Duplicate entry 'test_on_failure-test_mark_failure_2-2016-01-01 00:00:00.000000' for key 'task_instance.PRIMARY'&quot;)"><pre class="notranslate"><code class="notranslate">___________ TestLocalTaskJob.test_process_sigterm_works_with_retries ___________
  
  self = &lt;tests.jobs.test_local_task_job.TestLocalTaskJob object at 0x7f3d380ae1f0&gt;
  dag_maker = &lt;tests.conftest.dag_maker.&lt;locals&gt;.DagFactory object at 0x7f3d38917760&gt;
  
      def test_process_sigterm_works_with_retries(self, dag_maker):
          """
          Test that ensures that task runner sets tasks to retry when they(task runner)
           receive sigterm
          """
          # use shared memory value so we can properly track value change even if
          # it's been updated across processes.
          retry_callback_called = Value('i', 0)
          task_terminated_externally = Value('i', 1)
          shared_mem_lock = Lock()
      
          def retry_callback(context):
              with shared_mem_lock:
                  retry_callback_called.value += 1
              assert context['dag_run'].dag_id == 'test_mark_failure_2'
      
          def task_function(ti):
              time.sleep(60)
              # This should not happen -- the state change should be noticed and the task should get killed
              with shared_mem_lock:
                  task_terminated_externally.value = 0
      
          with dag_maker(dag_id='test_mark_failure_2'):
              task = PythonOperator(
                  task_id='test_on_failure',
                  python_callable=task_function,
                  retries=1,
                  retry_delay=timedelta(seconds=2),
                  on_retry_callback=retry_callback,
              )
          ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)
          ti.refresh_from_db()
          job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True, executor=SequentialExecutor())
          job1.task_runner = StandardTaskRunner(job1)
          job1.task_runner.start()
          settings.engine.dispose()
          process = multiprocessing.Process(target=job1.run)
          process.start()
          for _ in range(0, 25):
              ti.refresh_from_db()
              if ti.state == State.RUNNING and ti.pid is not None:
                  break
              time.sleep(0.2)
          os.kill(process.pid, signal.SIGTERM)
          process.join(timeout=10)
          ti.refresh_from_db()
  &gt;       assert ti.state == State.UP_FOR_RETRY
  E       AssertionError: assert None == &lt;TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'&gt;
  E        +  where None = &lt;TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01 00:00:00+00:00 [None]&gt;.state
  E        +  and   &lt;TaskInstanceState.UP_FOR_RETRY: 'up_for_retry'&gt; = State.UP_FOR_RETRY
  
  tests/jobs/test_local_task_job.py:828: AssertionError
  ----------------------------- Captured stdout call -----------------------------
  Running &lt;TaskInstance: test_mark_failure_2.test_on_failure 2016-01-01T00:00:00+00:00 [None]&gt; on host 67aa517c450e
  ----------------------------- Captured stderr call -----------------------------
  Process Process-113:
  Traceback (most recent call last):
    File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1276, in _execute_context
      self.dialect.do_execute(
    File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
      cursor.execute(statement, parameters)
    File "/usr/local/lib/python3.8/site-packages/MySQLdb/cursors.py", line 206, in execute
      res = self._query(query)
    File "/usr/local/lib/python3.8/site-packages/MySQLdb/cursors.py", line 319, in _query
      db.query(q)
    File "/usr/local/lib/python3.8/site-packages/MySQLdb/connections.py", line 259, in query
      _mysql.connection.query(self, query)
  MySQLdb._exceptions.IntegrityError: (1062, "Duplicate entry 'test_on_failure-test_mark_failure_2-2016-01-01 00:00:00.000000' for key 'task_instance.PRIMARY'")
</code></pre></div>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">Other Airflow 2 version</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">If a task gets scheduled on a different worker host from a previous run, logs from that previous run will be unavailable from the webserver.</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/7269927/187462675-010a2b6f-e2e5-4784-9b19-315f3b22ecf5.png"><img src="https://user-images.githubusercontent.com/7269927/187462675-010a2b6f-e2e5-4784-9b19-315f3b22ecf5.png" alt="Screenshot from 2022-08-30 09-22-39" style="max-width: 100%;"></a></p>
<p dir="auto">From my observation, it's pretty clear that this is happening because Airflow is looking for logs at the <em>most recent</em> hostname, not at the hostname that hosted the historical run.  You can watch this happen "live" with the example DAG below - each time the task retries on a different worker, the URL that it attempts to use for historical logs changes to use the most recent hostname.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">All previous logs should be available from the webserver (as long as the log file still exists on disk).</p>
<h3 dir="auto">How to reproduce</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="#!/usr/bin/env python3
import datetime
import logging
import socket

from airflow.decorators import dag, task


logger = logging.getLogger(__name__)


@dag(
    schedule_interval='@daily',
    start_date=datetime.datetime(2022, 8, 30),
    default_args={
        'retries': 9,
        'retry_delay': 10.0,
    },
)
def test_logs():
    @task
    def log():
        logger.info(f'running from {socket.gethostname()}')
        raise RuntimeError('force fail')

    log()


dag = test_logs()


if __name__ == '__main__':
    dag.cli()"><pre class="notranslate"><code class="notranslate">#!/usr/bin/env python3
import datetime
import logging
import socket

from airflow.decorators import dag, task


logger = logging.getLogger(__name__)


@dag(
    schedule_interval='@daily',
    start_date=datetime.datetime(2022, 8, 30),
    default_args={
        'retries': 9,
        'retry_delay': 10.0,
    },
)
def test_logs():
    @task
    def log():
        logger.info(f'running from {socket.gethostname()}')
        raise RuntimeError('force fail')

    log()


dag = test_logs()


if __name__ == '__main__':
    dag.cli()
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">CentOS Stream 8</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">None</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Airflow v2.3.3<br>
Self-hosted<br>
Postgres DB backend<br>
CeleryExecutor</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Possibly related:</p>
<ul dir="auto">
<li><a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="778813083" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/13483" data-hovercard-type="issue" data-hovercard-url="/apache/airflow/issues/13483/hovercard" href="https://github.com/apache/airflow/issues/13483">#13483</a></li>
<li><a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="113511652" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/561" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/561/hovercard" href="https://github.com/apache/airflow/pull/561">#561</a></li>
</ul>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>