This is an automated email from the ASF dual-hosted git repository.
Lee-W 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 7483c23e9b7 Fix Edge workers not reporting the final state of deadline
callbacks (#74784)
7483c23e9b7 is described below
commit 7483c23e9b7b81f18ee0b60bf622a3b6c67544c9
Author: PoAn Yang <[email protected]>
AuthorDate: Thu Oct 22 15:26:01 2026 +0800
Fix Edge workers not reporting the final state of deadline callbacks
(#74784)
Co-authored-by: Wei Lee <[email protected]>
Signed-off-by: PoAn Yang <[email protected]>
---
.../src/airflow/providers/edge3/cli/worker.py | 25 ++++--
.../providers/edge3/worker_api/routes/logs.py | 35 +++++++-
.../edge3/worker_api/v2-edge-generated.yaml | 12 +++
.../edge3/tests/unit/edge3/cli/test_worker.py | 54 ++++++++++++
.../unit/edge3/worker_api/routes/test_logs.py | 95 +++++++++++++++++++++-
5 files changed, 213 insertions(+), 8 deletions(-)
diff --git a/providers/edge3/src/airflow/providers/edge3/cli/worker.py
b/providers/edge3/src/airflow/providers/edge3/cli/worker.py
index 2925ac4e1e9..da462d092e8 100644
--- a/providers/edge3/src/airflow/providers/edge3/cli/worker.py
+++ b/providers/edge3/src/airflow/providers/edge3/cli/worker.py
@@ -583,6 +583,13 @@ class EdgeWorker:
logger.exception("Log file %s vanished while reading, ignoring.",
job.logfile)
# Swallow the exception; the file may have been removed by log
rotation or cleanup while we were reading it.
# We'll catch up on the next heartbeat/log push or file was
uploaded by log integration in parallel.
+ except Exception:
+ # Log pushing is best effort. The job state must still reach the
central site, or the job
+ # stays RUNNING there until [scheduler]
task_instance_heartbeat_timeout passes.
+ logger.exception(
+ "Failed to push logs of %s to the central site. This part of
the log will not be sent again.",
+ job.edge_job.identifier,
+ )
async def start(self):
"""Start the execution in a loop until terminated."""
@@ -727,12 +734,18 @@ class EdgeWorker:
logger.error("Job failed: %s with:\n%s",
job.edge_job.identifier, ex_txt)
# Push it upwards to logs for better diagnostic as well
- await logs_push(
- task=job.edge_job.key,
- log_chunk_time=timezone.utcnow(),
- log_chunk_data=f"Error executing job:\n{ex_txt}",
- task_instance_id=job.edge_job.task_instance_id,
- )
+ try:
+ await logs_push(
+ task=job.edge_job.key,
+ log_chunk_time=timezone.utcnow(),
+ log_chunk_data=f"Error executing job:\n{ex_txt}",
+ task_instance_id=job.edge_job.task_instance_id,
+ )
+ except Exception:
+ logger.exception(
+ "Failed to push the error message of %s to the central
site.",
+ job.edge_job.identifier,
+ )
await jobs_set_state(
job.edge_job.key, TaskInstanceState.FAILED,
task_instance_id=job.edge_job.task_instance_id
)
diff --git
a/providers/edge3/src/airflow/providers/edge3/worker_api/routes/logs.py
b/providers/edge3/src/airflow/providers/edge3/worker_api/routes/logs.py
index a6e2fb1eb06..30fa69e2602 100644
--- a/providers/edge3/src/airflow/providers/edge3/worker_api/routes/logs.py
+++ b/providers/edge3/src/airflow/providers/edge3/worker_api/routes/logs.py
@@ -17,12 +17,13 @@
from __future__ import annotations
+import json
from functools import cache
from pathlib import Path
from typing import TYPE_CHECKING, Annotated
from uuid import UUID
-from fastapi import Body, Depends, Query, status
+from fastapi import Body, Depends, HTTPException, Query, status
from sqlalchemy import select
from airflow.api_fastapi.common.db.common import SessionDep # noqa: TC001
@@ -32,11 +33,15 @@ from airflow.models.taskinstance import TaskInstance
from airflow.providers.common.compat.sdk import TaskInstanceKey, conf
from airflow.providers.edge3.models.edge_job import EdgeJobModel
from airflow.providers.edge3.models.edge_logs import EdgeLogsModel
+from airflow.providers.edge3.models.types import is_callback_job
from airflow.providers.edge3.worker_api.auth import
jwt_token_authorization_rest
from airflow.providers.edge3.worker_api.datamodels import PushLogsBody,
WorkerApiDocs
from airflow.utils.log.file_task_handler import FileTaskHandler
from airflow.utils.session import NEW_SESSION, provide_session
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
+
logs_router = AirflowRouter(tags=["Logs"], prefix="/logs")
@@ -44,6 +49,8 @@ logs_router = AirflowRouter(tags=["Logs"], prefix="/logs")
@provide_session
def _logfile_path(task: TaskInstanceKey, task_instance_id: UUID | None = None,
*, session=NEW_SESSION) -> str:
"""Elaborate the (relative) path and filename to expect from task
execution."""
+ if is_callback_job(task.dag_id, task.task_id, task.run_id,
task.try_number, task.map_index):
+ return _get_callback_logfile_path(task, session=session)
if task_instance_id is None:
stored_id = session.scalar(
select(EdgeJobModel.task_instance_id).where(
@@ -74,6 +81,30 @@ def _logfile_path(task: TaskInstanceKey, task_instance_id:
UUID | None = None, *
return FileTaskHandler(".")._render_filename(ti, task.try_number)
+def _get_callback_logfile_path(task: TaskInstanceKey, *, session: Session) ->
str:
+ """
+ Return a callback's ``log_path`` from the workload JSON in the callback's
``edge_job`` row.
+
+ Executor callbacks have no task instance row, so their log path cannot be
rendered from a task
+ instance. When queuing this callback, the scheduler stored the workload
JSON in the ``command``
+ column of the callback's ``edge_job`` row. The ``log_path`` in that JSON
is a path relative to the
+ log folder, and the callback supervisor process writes the callback log at
that path on the Edge
+ worker. Raise HTTP 404 if the ``edge_job`` row is missing or the JSON has
no ``log_path``.
+ """
+ command = session.scalar(
+ select(EdgeJobModel.command).where(
+ EdgeJobModel.dag_id == task.dag_id,
+ EdgeJobModel.task_id == task.task_id,
+ EdgeJobModel.run_id == task.run_id,
+ EdgeJobModel.map_index == task.map_index,
+ EdgeJobModel.try_number == task.try_number,
+ )
+ )
+ if command and (log_path := json.loads(command).get("log_path")):
+ return log_path
+ raise HTTPException(status.HTTP_404_NOT_FOUND, f"No log path found for
callback {task.task_id}")
+
+
@logs_router.get(
"/logfile_path/{dag_id}/{task_id}/{run_id}/{try_number}/{map_index}",
dependencies=[Depends(jwt_token_authorization_rest)],
@@ -81,6 +112,7 @@ def _logfile_path(task: TaskInstanceKey, task_instance_id:
UUID | None = None, *
[
status.HTTP_400_BAD_REQUEST,
status.HTTP_403_FORBIDDEN,
+ status.HTTP_404_NOT_FOUND,
]
),
)
@@ -108,6 +140,7 @@ def logfile_path(
[
status.HTTP_400_BAD_REQUEST,
status.HTTP_403_FORBIDDEN,
+ status.HTTP_404_NOT_FOUND,
]
),
)
diff --git
a/providers/edge3/src/airflow/providers/edge3/worker_api/v2-edge-generated.yaml
b/providers/edge3/src/airflow/providers/edge3/worker_api/v2-edge-generated.yaml
index 3c3c355c2cd..ff2b9f1d5c8 100644
---
a/providers/edge3/src/airflow/providers/edge3/worker_api/v2-edge-generated.yaml
+++
b/providers/edge3/src/airflow/providers/edge3/worker_api/v2-edge-generated.yaml
@@ -269,6 +269,12 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
'422':
description: Validation Error
content:
@@ -360,6 +366,12 @@ paths:
schema:
$ref: '#/components/schemas/HTTPExceptionResponse'
description: Forbidden
+ '404':
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/HTTPExceptionResponse'
+ description: Not Found
'422':
description: Validation Error
content:
diff --git a/providers/edge3/tests/unit/edge3/cli/test_worker.py
b/providers/edge3/tests/unit/edge3/cli/test_worker.py
index fb805168762..744f71e5e4c 100644
--- a/providers/edge3/tests/unit/edge3/cli/test_worker.py
+++ b/providers/edge3/tests/unit/edge3/cli/test_worker.py
@@ -753,6 +753,60 @@ class TestEdgeWorker:
assert launched_job not in worker_with_job.jobs
assert not stderr_file_path.exists()
+ @pytest.mark.parametrize(
+ ("returncode", "expected_state", "expected_pushes"),
+ [
+ pytest.param(0, TaskInstanceState.SUCCESS, 1, id="success"),
+ pytest.param(1, TaskInstanceState.FAILED, 2, id="failure"),
+ ],
+ )
+ @patch("airflow.providers.edge3.cli.worker.jobs_fetch", autospec=True)
+ @patch("airflow.providers.edge3.cli.worker.jobs_set_state", autospec=True)
+ @patch("airflow.providers.edge3.cli.worker.logs_push", autospec=True)
+ @pytest.mark.asyncio
+ async def test_fetch_and_run_job_reports_state_when_log_push_fails(
+ self,
+ mock_logs_push,
+ mock_jobs_set_state,
+ mock_jobs_fetch,
+ returncode: int,
+ expected_state: TaskInstanceState,
+ expected_pushes: int,
+ tmp_path: Path,
+ worker_with_job: EdgeWorker,
+ ):
+ mock_logs_push.side_effect = ClientResponseError(
+ request_info=RequestInfo(url=URL("mock.com"), method="POST",
headers=None), # type:ignore[arg-type]
+ message="Internal Server Error",
+ status=500,
+ history=(),
+ )
+ edge_job = EdgeJobFetched(
+ dag_id="test",
+ task_id="test",
+ run_id="test",
+ map_index=-1,
+ try_number=1,
+ concurrency_slots=1,
+ command=MOCK_COMMAND, # type: ignore[arg-type]
+ )
+ mock_jobs_fetch.return_value = edge_job
+ worker_with_job.concurrency = 1
+ worker_with_job.push_logs = True
+ logfile = tmp_path / "mock.log"
+ logfile.write_text("some log content")
+ launched_job = Job(edge_job, _MockProcess(returncode=returncode),
logfile)
+
+ with patch.object(worker_with_job, "_launch_job",
return_value=launched_job):
+ await worker_with_job.fetch_and_run_job()
+
+ assert mock_logs_push.call_count == expected_pushes
+ assert mock_jobs_set_state.call_args_list == [
+ call(edge_job.key, TaskInstanceState.RUNNING,
task_instance_id=None),
+ call(edge_job.key, expected_state, task_instance_id=None),
+ ]
+ assert launched_job not in worker_with_job.jobs
+
@time_machine.travel(datetime.now(), tick=False)
@patch("airflow.providers.edge3.cli.worker.logs_push")
@pytest.mark.asyncio
diff --git a/providers/edge3/tests/unit/edge3/worker_api/routes/test_logs.py
b/providers/edge3/tests/unit/edge3/worker_api/routes/test_logs.py
index d69dc99b1c8..cba793c7390 100644
--- a/providers/edge3/tests/unit/edge3/worker_api/routes/test_logs.py
+++ b/providers/edge3/tests/unit/edge3/worker_api/routes/test_logs.py
@@ -16,20 +16,29 @@
# under the License.
from __future__ import annotations
+import json
from collections.abc import Sequence
from pathlib import Path
from typing import TYPE_CHECKING
import pytest
+from fastapi import HTTPException
from sqlalchemy import delete, select
from airflow.providers.common.compat.sdk import timezone
from airflow.providers.edge3.models.edge_job import EdgeJobModel
from airflow.providers.edge3.models.edge_logs import EdgeLogsModel
+from airflow.providers.edge3.models.types import (
+ CALLBACK_JOB_MAP_INDEX,
+ CALLBACK_JOB_TRY_NUMBER,
+ EXECUTE_CALLBACK_TAG,
+ build_callback_run_id,
+)
from airflow.providers.edge3.worker_api.datamodels import PushLogsBody
-from airflow.providers.edge3.worker_api.routes.logs import logfile_path,
push_logs
+from airflow.providers.edge3.worker_api.routes.logs import _logfile_path,
logfile_path, push_logs
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.utils.session import create_session
+from airflow.utils.state import TaskInstanceState
from tests_common.test_utils.config import conf_vars
from tests_common.test_utils.version_compat import AIRFLOW_V_3_4_PLUS
@@ -43,6 +52,8 @@ pytestmark = pytest.mark.db_test
DAG_ID = "my_dag"
TASK_ID = "my_task"
RUN_ID = "manual__2024-11-24"
+CALLBACK_ID = "12345678-1234-5678-1234-567812345678"
+CALLBACK_LOG_PATH = f"executor_callbacks/{DAG_ID}/{RUN_ID}/{CALLBACK_ID}"
class TestLogsApiRoutes:
@@ -218,3 +229,85 @@ class TestLogsApiRoutes:
assert logs[0].try_number == 1
assert logs[0].map_index == -1
assert "Lorem Ipsum" in logs[0].log_chunk_data
+
+ @pytest.fixture
+ def make_callback_job(self, session: Session):
+ def make(command: dict) -> EdgeJobModel:
+ job = EdgeJobModel(
+ dag_id=EXECUTE_CALLBACK_TAG,
+ task_id=CALLBACK_ID,
+ run_id=build_callback_run_id(CALLBACK_ID),
+ map_index=CALLBACK_JOB_MAP_INDEX,
+ try_number=CALLBACK_JOB_TRY_NUMBER,
+ state=TaskInstanceState.RUNNING,
+ queue="default",
+ concurrency_slots=1,
+ command=json.dumps(command),
+ )
+ session.add(job)
+ session.commit()
+ return job
+
+ _logfile_path.cache_clear()
+ yield make
+ session.execute(delete(EdgeJobModel))
+ session.commit()
+ _logfile_path.cache_clear()
+
+ def test_logfile_path_for_callback_job(self, make_callback_job):
+ job = make_callback_job({"type": EXECUTE_CALLBACK_TAG, "log_path":
CALLBACK_LOG_PATH})
+
+ path = logfile_path(
+ dag_id=job.dag_id,
+ task_id=job.task_id,
+ run_id=job.run_id,
+ try_number=job.try_number,
+ map_index=job.map_index,
+ )
+
+ assert path == CALLBACK_LOG_PATH
+
+ def test_push_logs_for_callback_job(self, make_callback_job, session:
Session, tmp_path: Path):
+ job = make_callback_job({"type": EXECUTE_CALLBACK_TAG, "log_path":
CALLBACK_LOG_PATH})
+ log_data = PushLogsBody(log_chunk_data="callback log line",
log_chunk_time=timezone.utcnow())
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path)}),
create_session() as api_session:
+ push_logs(
+ dag_id=job.dag_id,
+ task_id=job.task_id,
+ run_id=job.run_id,
+ try_number=job.try_number,
+ map_index=job.map_index,
+ body=log_data,
+ session=api_session,
+ )
+ logs: Sequence[EdgeLogsModel] =
session.scalars(select(EdgeLogsModel)).all()
+ assert [(log.dag_id, log.task_id, log.run_id, log.log_chunk_data) for
log in logs] == [
+ (EXECUTE_CALLBACK_TAG, CALLBACK_ID,
build_callback_run_id(CALLBACK_ID), "callback log line")
+ ]
+ assert (tmp_path / CALLBACK_LOG_PATH).read_text() == "callback log
line"
+
+ @pytest.mark.parametrize(
+ "stored_command",
+ [
+ pytest.param(None, id="no-edge-job"),
+ pytest.param({"type": EXECUTE_CALLBACK_TAG}, id="no-log-path"),
+ ],
+ )
+ def test_push_logs_for_callback_job_without_log_path(
+ self, make_callback_job, stored_command: dict | None, session: Session
+ ):
+ if stored_command is not None:
+ make_callback_job(stored_command)
+ log_data = PushLogsBody(log_chunk_data="callback log line",
log_chunk_time=timezone.utcnow())
+ with pytest.raises(HTTPException) as exc_info, create_session() as
api_session:
+ push_logs(
+ dag_id=EXECUTE_CALLBACK_TAG,
+ task_id=CALLBACK_ID,
+ run_id=build_callback_run_id(CALLBACK_ID),
+ try_number=CALLBACK_JOB_TRY_NUMBER,
+ map_index=CALLBACK_JOB_MAP_INDEX,
+ body=log_data,
+ session=api_session,
+ )
+ assert exc_info.value.status_code == 404
+ assert session.scalars(select(EdgeLogsModel)).all() == []
