This is an automated email from the ASF dual-hosted git repository.
jscheffl 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 3e076e8a64e Edge: Return 400 instead of 500 on invalid worker version
(#72925)
3e076e8a64e is described below
commit 3e076e8a64e3eb189b871a3541272c2d7185d5a5
Author: Stefan Wang <[email protected]>
AuthorDate: Fri Nov 6 10:23:41 2026 -0800
Edge: Return 400 instead of 500 on invalid worker version (#72925)
* Edge provider: Return 400 instead of 500 on invalid worker version
When minimum acceptable worker versions were configured, an edge worker
submitting an unparseable version string caused packaging.version.Version to
raise InvalidVersion. Because this exception was unhandled, it escaped as an
HTTP 500 Internal Server Error.
Catch InvalidVersion in _version() and return a clean 400 Bad Request error
matching the route error specification.
Signed-off-by: 1fanwang <[email protected]>
* Keep invalid server version settings out of worker input errors
A malformed server minimum must not be reported as bad worker input. Cover
both registration and heartbeat through authenticated HTTP requests.
Signed-off-by: 1fanwang <[email protected]>
* Bind the sysinfo payload to its own typed variable in the version test
The test body is a heterogeneous dict literal, so mypy infers its value type
as object and rejects the nested assignment into body["sysinfo"]. Building
sysinfo first keeps its dict[str, str | int | float | datetime] type while
the body still references the same object, so the mutation reaches the
request payload exactly as before.
Signed-off-by: 1fanwang <[email protected]>
* Reject non-string Edge worker versions with HTTP 400
The worker sysinfo payload accepts numbers as well as strings, so a worker
can report its version as 3 or 3.5. packaging releases before 26.3 raise
TypeError instead of InvalidVersion for such values, which still escaped as
HTTP 500 whenever a minimum acceptable worker version was configured.
Signed-off-by: 1fanwang <[email protected]>
---------
Signed-off-by: 1fanwang <[email protected]>
---
.../providers/edge3/worker_api/routes/worker.py | 33 ++++++++---
.../unit/edge3/worker_api/routes/test_worker.py | 68 +++++++++++++++++++++-
2 files changed, 90 insertions(+), 11 deletions(-)
diff --git
a/providers/edge3/src/airflow/providers/edge3/worker_api/routes/worker.py
b/providers/edge3/src/airflow/providers/edge3/worker_api/routes/worker.py
index 2e42ae8902b..80900ba6d62 100644
--- a/providers/edge3/src/airflow/providers/edge3/worker_api/routes/worker.py
+++ b/providers/edge3/src/airflow/providers/edge3/worker_api/routes/worker.py
@@ -21,7 +21,7 @@ from datetime import datetime
from typing import Annotated
from fastapi import Body, Depends, HTTPException, Path, status
-from packaging.version import Version
+from packaging.version import InvalidVersion, Version
from sqlalchemy import select
from airflow import __version__ as airflow_version
@@ -53,9 +53,20 @@ worker_router = AirflowRouter(
)
-def _version(version_str: str) -> tuple[int, int, int]:
+def _version(version_str: str | int | float | datetime) -> tuple[int, int,
int]:
"""Convert a version string into a tuple of integers for comparison."""
- version = Version(version_str)
+ if not isinstance(version_str, str):
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid version {version_str!r}: expected a string, got
{type(version_str).__name__}",
+ )
+ try:
+ version = Version(version_str)
+ except InvalidVersion as e:
+ raise HTTPException(
+ status.HTTP_400_BAD_REQUEST,
+ f"Invalid version string '{version_str}': {e}",
+ ) from e
return version.major, version.minor, version.micro
@@ -63,14 +74,16 @@ def _assert_version(sysinfo: dict[str, str | int | float |
datetime]) -> bool:
"""Check if the Edge Worker version matches the central API site."""
versions_match = True
if "airflow_version" in sysinfo:
- airflow_on_worker: str = sysinfo["airflow_version"] # type: ignore
+ airflow_on_worker = sysinfo["airflow_version"]
if airflow_on_worker != airflow_version:
versions_match = False
minimum_acceptable_core_version_for_workers = conf.get(
"edge", "minimum_acceptable_core_version_for_workers",
fallback=None
)
- if not minimum_acceptable_core_version_for_workers or
_version(airflow_on_worker) < _version(
- minimum_acceptable_core_version_for_workers
+ if (
+ not minimum_acceptable_core_version_for_workers
+ or _version(airflow_on_worker)
+ <
Version(minimum_acceptable_core_version_for_workers).release[:3]
):
raise HTTPException(
status.HTTP_400_BAD_REQUEST,
@@ -83,14 +96,16 @@ def _assert_version(sysinfo: dict[str, str | int | float |
datetime]) -> bool:
)
if "edge_provider_version" in sysinfo:
- provider_on_worker: str = sysinfo["edge_provider_version"] # type:
ignore
+ provider_on_worker = sysinfo["edge_provider_version"]
if provider_on_worker != edge_provider_version:
versions_match = False
minimum_acceptable_edge_version_for_workers = conf.get(
"edge", "minimum_acceptable_edge_version_for_workers",
fallback=None
)
- if not minimum_acceptable_edge_version_for_workers or
_version(provider_on_worker) < _version(
- minimum_acceptable_edge_version_for_workers
+ if (
+ not minimum_acceptable_edge_version_for_workers
+ or _version(provider_on_worker)
+ <
Version(minimum_acceptable_edge_version_for_workers).release[:3]
):
raise HTTPException(
status.HTTP_400_BAD_REQUEST,
diff --git a/providers/edge3/tests/unit/edge3/worker_api/routes/test_worker.py
b/providers/edge3/tests/unit/edge3/worker_api/routes/test_worker.py
index 939cf5d5f54..2ad10f409c1 100644
--- a/providers/edge3/tests/unit/edge3/worker_api/routes/test_worker.py
+++ b/providers/edge3/tests/unit/edge3/worker_api/routes/test_worker.py
@@ -23,11 +23,13 @@ from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
-from fastapi import HTTPException
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
from sqlalchemy import delete, select
from airflow import __version__ as airflow_version
-from airflow.providers.common.compat.sdk import Stats, timezone
+from airflow.api_fastapi.auth.tokens import JWTGenerator
+from airflow.providers.common.compat.sdk import Stats, conf, timezone
from airflow.providers.edge3 import __version__ as edge_provider_version
from airflow.providers.edge3.cli.worker import EdgeWorker
from airflow.providers.edge3.models.edge_worker import (
@@ -35,6 +37,7 @@ from airflow.providers.edge3.models.edge_worker import (
EdgeWorkerState,
set_worker_concurrency,
)
+from airflow.providers.edge3.worker_api.auth import jwt_validator
from airflow.providers.edge3.worker_api.datamodels import
WorkerQueueUpdateBody, WorkerStateBody
from airflow.providers.edge3.worker_api.routes.worker import (
_assert_version,
@@ -42,6 +45,7 @@ from airflow.providers.edge3.worker_api.routes.worker import (
register,
set_state,
update_queues,
+ worker_router,
)
from tests_common.test_utils.config import conf_vars
@@ -77,6 +81,66 @@ class TestWorkerApiRoutes:
assert _version("1.2.3") == (1, 2, 3)
assert _version("1.2.3rc1") == (1, 2, 3)
assert _version("1.2.3.dev0") == (1, 2, 3)
+ with pytest.raises(HTTPException, match="expected a string, got
datetime") as exc_info:
+ _version(datetime(2026, 9, 15))
+ assert exc_info.value.status_code == 400
+
+ @conf_vars({("api_auth", "jwt_secret"): "edge-version-test-secret"})
+ @pytest.mark.parametrize(
+ ("field", "minimum_key"),
+ [
+ ("airflow_version", "minimum_acceptable_core_version_for_workers"),
+ ("edge_provider_version",
"minimum_acceptable_edge_version_for_workers"),
+ ],
+ )
+ @pytest.mark.parametrize("method", ["POST", "PATCH"])
+ @pytest.mark.parametrize(
+ ("worker_version", "minimum", "expected_status"),
+ [
+ pytest.param("invalid-version", "0.0.0", 400,
id="worker-invalid-string"),
+ pytest.param("2026-09-15T19:53:06", "0.0.0", 400,
id="worker-datetime-string"),
+ pytest.param(3, "0.0.0", 400, id="worker-int"),
+ pytest.param(3.5, "0.0.0", 400, id="worker-float"),
+ pytest.param(True, "0.0.0", 400, id="worker-bool"),
+ pytest.param("0.0.0", "invalid-version", 500,
id="server-invalid-minimum"),
+ ],
+ )
+ def test_invalid_version_http_status(
+ self,
+ session: Session,
+ field: str,
+ minimum_key: str,
+ method: str,
+ worker_version: str | int | float,
+ minimum: str,
+ expected_status: int,
+ ):
+ session.commit()
+ app = FastAPI()
+ app.include_router(worker_router, prefix="/edge_worker/v1")
+ path = "/edge_worker/v1/worker/version_test"
+ token = JWTGenerator(
+ secret_key=conf.get("api_auth", "jwt_secret"), valid_for=60,
audience="api"
+ ).generate(extras={"method": "worker/version_test"})
+ sysinfo = dict(self.MOCK_SYSINFO)
+ body = {
+ "state": EdgeWorkerState.STARTING,
+ "jobs_active": 0,
+ "queues": ["default"],
+ "sysinfo": sysinfo,
+ }
+ jwt_validator.cache_clear()
+ try:
+ with TestClient(app, raise_server_exceptions=False,
headers={"Authorization": token}) as client:
+ if method == "PATCH":
+ assert client.post(path, json=body).status_code == 200
+ sysinfo[field] = worker_version
+ with conf_vars({("edge", minimum_key): minimum}):
+ response = client.request(method=method, url=path,
json=body)
+ finally:
+ jwt_validator.cache_clear()
+
+ assert response.status_code == expected_status
def test_assert_version(self):
from airflow import __version__ as airflow_version
