From 408f34a54aea150d27b539a6ea063dc926d71e41 Mon Sep 17 00:00:00 2001 From: Ryan Hatter Date: Sat, 8 Jun 2024 10:04:40 -0400 Subject: [PATCH 1/5] revamp was it killed externally question --- airflow/jobs/backfill_job_runner.py | 6 ++++-- airflow/jobs/scheduler_job_runner.py | 6 ++++-- tests/jobs/test_scheduler_job.py | 5 ++--- 3 files changed, 10 insertions(+), 7 deletions(-) diff --git a/airflow/jobs/backfill_job_runner.py b/airflow/jobs/backfill_job_runner.py index 4fd6b24bbfc2a..379d074e7545c 100644 --- a/airflow/jobs/backfill_job_runner.py +++ b/airflow/jobs/backfill_job_runner.py @@ -313,9 +313,11 @@ def _manage_executor_state( and ti.state in self.STATES_COUNT_AS_RUNNING ): msg = ( - f"Executor reports task instance {ti} finished ({state}) although the task says its " - f"{ti.state}. Was the task killed externally? Info: {info}" + f"The executor reported that the task instance {ti} finished with state {state}, " + f"but the task instance's state attribute is {ti.state}." ) + if info is not None: + msg += f" Extra info: {info}" self.log.error(msg) ti.handle_failure(error=msg) continue diff --git a/airflow/jobs/scheduler_job_runner.py b/airflow/jobs/scheduler_job_runner.py index 3d81aaeb56276..853866e39cd9c 100644 --- a/airflow/jobs/scheduler_job_runner.py +++ b/airflow/jobs/scheduler_job_runner.py @@ -774,9 +774,11 @@ def _process_executor_events(self, session: Session) -> int: tags={"dag_id": ti.dag_id, "task_id": ti.task_id}, ) msg = ( - "Executor reports task instance %s finished (%s) although the " - "task says it's %s. (Info: %s) Was the task killed externally?" + f"The executor reported that the task instance {ti} finished with state {state}, " + f"but the task instance's state attribute is {ti.state}." ) + if info is not None: + msg += f" Extra info: {info}" self._task_context_logger.error(msg, ti, state, ti.state, info, ti=ti) # Get task from the Serialized DAG diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py index 9fcdba5ac5abb..4960437ecb49b 100644 --- a/tests/jobs/test_scheduler_job.py +++ b/tests/jobs/test_scheduler_job.py @@ -400,10 +400,9 @@ def test_process_executor_events_with_callback(self, mock_stats_incr, mock_task_ full_filepath=dag.fileloc, simple_task_instance=mock.ANY, processor_subdir=None, - msg="Executor reports task instance " + msg="The executor reported that the task instance " " " - "finished (failed) although the task says it's queued. (Info: None) " - "Was the task killed externally?", + "finished with state failedbut the task instance's state attribute is queued." ) scheduler_job.executor.callback_sink.send.assert_called_once_with(task_callback) scheduler_job.executor.callback_sink.reset_mock() From 22b6976587da410cfa273cee986e0681d2444843 Mon Sep 17 00:00:00 2001 From: Ryan Hatter Date: Sat, 8 Jun 2024 11:15:24 -0400 Subject: [PATCH 2/5] static checks --- tests/jobs/test_scheduler_job.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py index 4960437ecb49b..0ce62f19bf050 100644 --- a/tests/jobs/test_scheduler_job.py +++ b/tests/jobs/test_scheduler_job.py @@ -402,7 +402,7 @@ def test_process_executor_events_with_callback(self, mock_stats_incr, mock_task_ processor_subdir=None, msg="The executor reported that the task instance " " " - "finished with state failedbut the task instance's state attribute is queued." + "finished with state failed, but the task instance's state attribute is queued.", ) scheduler_job.executor.callback_sink.send.assert_called_once_with(task_callback) scheduler_job.executor.callback_sink.reset_mock() From 22afde3c5c22aec22ecee441461b5a2297e65c27 Mon Sep 17 00:00:00 2001 From: Ryan Hatter Date: Sun, 9 Jun 2024 09:46:21 -0400 Subject: [PATCH 3/5] revert to % strings --- airflow/jobs/scheduler_job_runner.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/airflow/jobs/scheduler_job_runner.py b/airflow/jobs/scheduler_job_runner.py index 853866e39cd9c..82f1797facf75 100644 --- a/airflow/jobs/scheduler_job_runner.py +++ b/airflow/jobs/scheduler_job_runner.py @@ -773,10 +773,7 @@ def _process_executor_events(self, session: Session) -> int: "scheduler.tasks.killed_externally", tags={"dag_id": ti.dag_id, "task_id": ti.task_id}, ) - msg = ( - f"The executor reported that the task instance {ti} finished with state {state}, " - f"but the task instance's state attribute is {ti.state}." - ) + msg = f"The executor reported that the task instance {ti} finished with state {state}, but the task instance's state attribute is {ti.state}." if info is not None: msg += f" Extra info: {info}" self._task_context_logger.error(msg, ti, state, ti.state, info, ti=ti) @@ -794,12 +791,12 @@ def _process_executor_events(self, session: Session) -> int: request = TaskCallbackRequest( full_filepath=ti.dag_model.fileloc, simple_task_instance=SimpleTaskInstance.from_ti(ti), - msg=msg % (ti, state, ti.state, info), + msg=msg, processor_subdir=ti.dag_model.processor_subdir, ) self.job.executor.send_callback(request) else: - ti.handle_failure(error=msg % (ti, state, ti.state, info), session=session) + ti.handle_failure(error=msg, session=session) return len(event_buffer) From 4491a725996ece377210a1bb20ca63d1beccc628 Mon Sep 17 00:00:00 2001 From: Ryan Hatter Date: Sun, 9 Jun 2024 13:15:54 -0400 Subject: [PATCH 4/5] add more useful info to log --- airflow/jobs/backfill_job_runner.py | 3 +++ airflow/jobs/scheduler_job_runner.py | 2 +- tests/jobs/test_scheduler_job.py | 5 ++++- 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/airflow/jobs/backfill_job_runner.py b/airflow/jobs/backfill_job_runner.py index 379d074e7545c..d2e99d0f457c2 100644 --- a/airflow/jobs/backfill_job_runner.py +++ b/airflow/jobs/backfill_job_runner.py @@ -315,6 +315,9 @@ def _manage_executor_state( msg = ( f"The executor reported that the task instance {ti} finished with state {state}, " f"but the task instance's state attribute is {ti.state}." + "This indicates that the task was marked failed by something other than the scheduler. " + "The task might have been marked failed by a user, by the task_queued_timeout configuration, " + "or it might have been killed by something else." ) if info is not None: msg += f" Extra info: {info}" diff --git a/airflow/jobs/scheduler_job_runner.py b/airflow/jobs/scheduler_job_runner.py index 82f1797facf75..72ae4cb13ebd4 100644 --- a/airflow/jobs/scheduler_job_runner.py +++ b/airflow/jobs/scheduler_job_runner.py @@ -773,7 +773,7 @@ def _process_executor_events(self, session: Session) -> int: "scheduler.tasks.killed_externally", tags={"dag_id": ti.dag_id, "task_id": ti.task_id}, ) - msg = f"The executor reported that the task instance {ti} finished with state {state}, but the task instance's state attribute is {ti.state}." + msg = f"The executor reported that the task instance {ti} finished with state {state}, but the task instance's state attribute is {ti.state}. This indicates that the task was marked failed by something other than the scheduler. The task might have been marked failed by a user, by the task_queued_timeout configuration, or it might have been killed by something else." if info is not None: msg += f" Extra info: {info}" self._task_context_logger.error(msg, ti, state, ti.state, info, ti=ti) diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py index 0ce62f19bf050..fd871b8c6e102 100644 --- a/tests/jobs/test_scheduler_job.py +++ b/tests/jobs/test_scheduler_job.py @@ -402,7 +402,10 @@ def test_process_executor_events_with_callback(self, mock_stats_incr, mock_task_ processor_subdir=None, msg="The executor reported that the task instance " " " - "finished with state failed, but the task instance's state attribute is queued.", + "finished with state failed, but the task instance's state attribute is queued." + "This indicates that the task was marked failed by something other than the scheduler. " + "The task might have been marked failed by a user, by the task_queued_timeout configuration, " + "or it might have been killed by something else.", ) scheduler_job.executor.callback_sink.send.assert_called_once_with(task_callback) scheduler_job.executor.callback_sink.reset_mock() From e849b5dad7dba70a30f6865ef570309c0d50e9fa Mon Sep 17 00:00:00 2001 From: Ryan Hatter Date: Sun, 9 Jun 2024 13:19:49 -0400 Subject: [PATCH 5/5] fix string --- airflow/jobs/backfill_job_runner.py | 2 +- tests/jobs/test_scheduler_job.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/airflow/jobs/backfill_job_runner.py b/airflow/jobs/backfill_job_runner.py index d2e99d0f457c2..0b24dde8f02a0 100644 --- a/airflow/jobs/backfill_job_runner.py +++ b/airflow/jobs/backfill_job_runner.py @@ -314,7 +314,7 @@ def _manage_executor_state( ): msg = ( f"The executor reported that the task instance {ti} finished with state {state}, " - f"but the task instance's state attribute is {ti.state}." + f"but the task instance's state attribute is {ti.state}. " "This indicates that the task was marked failed by something other than the scheduler. " "The task might have been marked failed by a user, by the task_queued_timeout configuration, " "or it might have been killed by something else." diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py index fd871b8c6e102..1ff520c7198b2 100644 --- a/tests/jobs/test_scheduler_job.py +++ b/tests/jobs/test_scheduler_job.py @@ -402,7 +402,7 @@ def test_process_executor_events_with_callback(self, mock_stats_incr, mock_task_ processor_subdir=None, msg="The executor reported that the task instance " " " - "finished with state failed, but the task instance's state attribute is queued." + "finished with state failed, but the task instance's state attribute is queued. " "This indicates that the task was marked failed by something other than the scheduler. " "The task might have been marked failed by a user, by the task_queued_timeout configuration, " "or it might have been killed by something else.",