This is an automated email from the ASF dual-hosted git repository.
dheerajturaga 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 5dc89069c36 Fix airflow edge list-workers always showing null
concurrency (#72959)
5dc89069c36 is described below
commit 5dc89069c36650ce0382dfb9c76c0fbdca3f42c8
Author: Y-C <[email protected]>
AuthorDate: Wed Sep 16 02:36:47 2026 +0800
Fix airflow edge list-workers always showing null concurrency (#72959)
The sysinfo column became a JSON type in edge3 3.5.0, so SQLAlchemy now
hands the CLI a dict. The listing still ran json.loads on it, the
resulting TypeError was swallowed, and concurrency/free_concurrency were
reported as null for every worker. Read the dict directly, keeping the
"bad data shows null" behaviour instead of aborting the whole command.
Co-authored-by: Eason09053360
<[email protected]>
---
.../airflow/providers/edge3/cli/edge_command.py | 24 +++++-----------------
.../edge3/tests/unit/edge3/cli/test_worker.py | 4 ++++
2 files changed, 9 insertions(+), 19 deletions(-)
diff --git a/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
b/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
index 7daf1b58a69..6a11c876a69 100644
--- a/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
+++ b/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
@@ -266,28 +266,14 @@ def list_edge_workers(args) -> None:
queues=args.queues.split(",") if args.queues else None,
)
# Format and print worker info on the screen
- fields = [
- "worker_name",
- "state",
- "queues",
- "jobs_active",
- "concurrency",
- "free_concurrency",
- "maintenance_comment",
- ]
+ fields = ["worker_name", "state", "queues", "jobs_active",
"maintenance_comment"]
all_hosts = []
for host in all_hosts_iter:
- host_data = {
- f: getattr(host, f, None) for f in fields if f not in
("concurrency", "free_concurrency")
- }
- try:
- sysinfo = json.loads(host.sysinfo or "{}")
- host_data["concurrency"] = sysinfo.get("concurrency")
- host_data["free_concurrency"] = sysinfo.get("free_concurrency")
- except (json.JSONDecodeError, TypeError):
- host_data["concurrency"] = None
- host_data["free_concurrency"] = None
+ host_data = {f: getattr(host, f, None) for f in fields}
+ sysinfo = host.sysinfo if isinstance(host.sysinfo, dict) else {}
+ host_data["concurrency"] = sysinfo.get("concurrency")
+ host_data["free_concurrency"] = sysinfo.get("free_concurrency")
all_hosts.append(host_data)
AirflowConsole().print_as(data=all_hosts, output=args.output)
diff --git a/providers/edge3/tests/unit/edge3/cli/test_worker.py
b/providers/edge3/tests/unit/edge3/cli/test_worker.py
index 2a7c2675bb7..fb5cd1eb448 100644
--- a/providers/edge3/tests/unit/edge3/cli/test_worker.py
+++ b/providers/edge3/tests/unit/edge3/cli/test_worker.py
@@ -1088,6 +1088,7 @@ class TestEdgeWorker:
@pytest.mark.db_test
def test_list_edge_workers(self, mock_edgeworker: EdgeWorkerModel):
+ mock_edgeworker.sysinfo = {"concurrency": 8, "free_concurrency": 3}
args = self.parser.parse_args(["edge", "list-workers", "--output",
"json"])
with contextlib.redirect_stdout(StringIO()) as temp_stdout:
with (
@@ -1113,6 +1114,9 @@ class TestEdgeWorker:
]:
assert key in edge_workers[0]
assert any("test_edge_worker" in h["worker_name"] for h in
edge_workers)
+ # print_as stringifies scalars for json output
+ assert edge_workers[0]["concurrency"] == "8"
+ assert edge_workers[0]["free_concurrency"] == "3"
@pytest.mark.db_test
def test_list_edge_workers_passes_name_pattern(self, mock_edgeworker:
EdgeWorkerModel):
