This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 7f0d50cffc6 Add task log message when the Execution API is unreachable
(#70290)
7f0d50cffc6 is described below
commit 7f0d50cffc61a48650590078f6832b349860322b
Author: Karen Hsieh <[email protected]>
AuthorDate: Mon Oct 12 16:14:20 2026 -0400
Add task log message when the Execution API is unreachable (#70290)
* Add task log message when the Execution API is unreachable
* Remove inline comments and spec the mocks in the unreachable API test
Drop the inline comments in the httpx.RequestError handler, as asked in
review, and give the test's coordinator, logger and log file mocks a
spec or autospec.
* Assert the unreachable API hint on the real task log file
Write the task log to a temporary folder and read it back, as the
launch-failure tests do, instead of mocking _configure_logging. The
test now also checks that the group is closed before the error.
---------
Co-authored-by: Kaxil Naik <[email protected]>
---
.../src/airflow/sdk/execution_time/supervisor.py | 11 +++++++
.../task_sdk/execution_time/test_supervisor.py | 36 ++++++++++++++++++++++
2 files changed, 47 insertions(+)
diff --git a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
index a5f164daa6d..9677c1e0607 100644
--- a/task-sdk/src/airflow/sdk/execution_time/supervisor.py
+++ b/task-sdk/src/airflow/sdk/execution_time/supervisor.py
@@ -3152,6 +3152,17 @@ def supervise_task(
logger=logger or
structlog.get_logger(logger_name="task").bind(),
error=e,
)
+ except httpx.RequestError as e:
+ if logger is not None:
+ logger.info("::endgroup::")
+ logger.error(
+ "Could not reach the Execution API server. Verify the
`[core] "
+ "execution_api_server_url` (or `[api] base_url`) setting
points at a reachable "
+ "Airflow API server.",
+ server=server,
+ error=str(e),
+ )
+ raise
finally:
if log_path and log_file_descriptor:
log_file_descriptor.close()
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index 0f507a87f8e..14e285e3cb3 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -274,6 +274,42 @@ class TestSupervisor:
with expectation:
supervise_task(**kw)
+ def test_supervise_task_logs_config_hint_on_connection_error(self, mocker,
tmp_path):
+ """Test that a connection failure reaching the Execution API is
surfaced in the task
+ log, naming the responsible config, and still propagates."""
+ ti = TaskInstance(
+ id=uuid7(),
+ task_id="b",
+ dag_id="c",
+ run_id="d",
+ try_number=1,
+ dag_version_id=uuid7(),
+ queue="default",
+ )
+ server = "http://nonexistent-api-server:8080/execution/";
+ coordinator = mocker.Mock(spec=BaseCoordinator)
+ coordinator.execute_task.side_effect = httpx.ConnectError("Name or
service not known")
+ mocker.patch(
+ "airflow.sdk.execution_time.supervisor.get_coordinator_manager",
autospec=True
+ ).return_value.for_queue.return_value = coordinator
+
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path)}),
pytest.raises(httpx.ConnectError):
+ supervise_task(
+ ti=ti,
+ dag_rel_path="c.py",
+ token="",
+ server=server,
+ client=mocker.Mock(spec=sdk_client.Client),
+ bundle_info=BundleInfo(name="my-bundle", version=None),
+ log_path="attempt=1.log",
+ )
+
+ entries = [json.loads(line) for line in (tmp_path /
"attempt=1.log").read_text().splitlines()]
+ assert [(e["level"], e["event"]) for e in entries] == [("info",
"::endgroup::"), ("error", mock.ANY)]
+ assert "execution_api_server_url" in entries[1]["event"]
+ assert "base_url" in entries[1]["event"]
+ assert entries[1]["server"] == server
+
def _response_error(status: int, detail: dict[str, Any]) ->
ServerResponseError:
error = ServerResponseError.from_response(
