From a6736bf484254b28003aa7a3a8dbbdb5c5d33d1b Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 15 Oct 2026 11:09:54 +0200 Subject: [PATCH] Workers screen shows each agent's version and protocol, and who is behind The hello carries the fluksio-worker version. An agent refused for an older protocol stays listed as Needs upgrade until it attaches again. Docs gain an Upgrading workers section. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx --- backend/fluksio/api/routes/workers.py | 32 +++++++++++++++++-- backend/tests/api/routes/test_workers.py | 21 +++++++++++++ docs/code/workers.md | 19 +++++++---- frontend/src/client/schemas.gen.ts | 15 +++++++++ frontend/src/client/types.gen.ts | 3 ++ frontend/src/routes/_layout/workers.tsx | 40 +++++++++++++++++++----- worker/fluksio_worker/agent.py | 10 ++++++ 7 files changed, 124 insertions(+), 16 deletions(-) diff --git a/backend/fluksio/api/routes/workers.py b/backend/fluksio/api/routes/workers.py index ca6ef74..2388ef9 100644 --- a/backend/fluksio/api/routes/workers.py +++ b/backend/fluksio/api/routes/workers.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio import logging +import time import uuid from datetime import UTC, datetime, timedelta from typing import Any @@ -55,6 +56,11 @@ TOKEN_DAYS = 365 #: The sockets attached with each token, by `jti`, so revoking one drops them. _attached: dict[str, set[WebSocket]] = {} +#: Workers refused for an older protocol, by name, until one attaches again. +#: Such an agent exits on the refusal, so this is the one place left that says +#: which host still needs `pip install -U fluksio-worker`. +_behind: dict[str, WorkerInfo] = {} + class WorkerInfo(BaseModel): name: str @@ -68,6 +74,11 @@ class WorkerInfo(BaseModel): cpus: int = 1 gpus: int = 0 ram_mb: int | None = None + protocol: int = 0 + #: The fluksio-worker it runs; empty for an agent too old to say. + version: str = "" + #: Refused for speaking an older protocol than this engine, so not attached. + behind: bool = False class TokenRequest(BaseModel): @@ -109,9 +120,11 @@ def read_workers(request: Request) -> Any: cpus=worker.inventory.cpus, gpus=worker.inventory.gpus, ram_mb=worker.inventory.ram_mb, + protocol=int(worker.info.get("protocol") or 0), + version=str(worker.info.get("version") or ""), ) for worker in _hub(request.app).workers() - ] + ] + list(_behind.values()) class ResourceLevel(BaseModel): @@ -274,7 +287,19 @@ async def _attached_session(websocket: WebSocket, claims: dict[str, Any]) -> Non await websocket.close(code=1002) return - if hello.get("op") != "hello" or int(hello.get("protocol", 0)) != PROTOCOL: + protocol = int(hello.get("protocol", 0)) + if hello.get("op") != "hello" or protocol != PROTOCOL: + if hello.get("op") == "hello" and protocol < PROTOCOL: + name = str(claims.get("sub") or hello.get("name") or "worker") + _behind[name] = WorkerInfo( + name=name, + labels=sorted(str(label) for label in claims.get("labels") or []), + last_seen=time.time(), + python=str(hello.get("python") or ""), + protocol=protocol, + version=str(hello.get("version") or ""), + behind=True, + ) await websocket.send_json( {"op": "refused", "reason": f"this engine speaks protocol {PROTOCOL}"} ) @@ -308,10 +333,13 @@ async def _attached_session(websocket: WebSocket, claims: dict[str, Any]) -> Non info={ "python": hello.get("python"), "venv_digest": hello.get("venv_digest"), + "protocol": protocol, + "version": hello.get("version"), }, inventory=WorkerInventory.from_hello(hello.get("inventory")), ) hub.attach(worker) + _behind.pop(name, None) await websocket.send_json({"op": "welcome", "protocol": PROTOCOL, "name": name}) try: diff --git a/backend/tests/api/routes/test_workers.py b/backend/tests/api/routes/test_workers.py index 2a5b6bb..884e8a8 100644 --- a/backend/tests/api/routes/test_workers.py +++ b/backend/tests/api/routes/test_workers.py @@ -38,6 +38,27 @@ def test_the_token_names_the_worker_and_its_labels( assert [(w["name"], w["labels"]) for w in attached] == [("gpu-box", ["gpu"])] +def test_the_list_names_each_version_and_who_is_behind( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + issued = mint(client, superuser_token_headers) + url = f"{PREFIX}/attach?token={issued['token']}" + + def listed() -> list[tuple[object, ...]]: + answer = client.get(PREFIX, headers=superuser_token_headers).json() + return [(w["protocol"], w["version"], w["behind"]) for w in answer] + + with client.websocket_connect(url) as ws: + ws.send_json({**hello(), "protocol": PROTOCOL - 1, "version": "0.1.5"}) + assert ws.receive_json()["op"] == "refused" + assert listed() == [(PROTOCOL - 1, "0.1.5", True)] + + with client.websocket_connect(url) as ws: + ws.send_json({**hello(), "version": "0.2.0"}) + ws.receive_json() + assert listed() == [(PROTOCOL, "0.2.0", False)] + + def test_a_worker_cannot_claim_a_label_its_token_lacks( client: TestClient, superuser_token_headers: dict[str, str] ) -> None: diff --git a/docs/code/workers.md b/docs/code/workers.md index 810c501..b996bda 100644 --- a/docs/code/workers.md +++ b/docs/code/workers.md @@ -127,7 +127,7 @@ curl "$API_URL/workers" -H "Authorization: Bearer $TOKEN" curl "$API_URL/workers/resources" -H "Authorization: Bearer $TOKEN" ``` -## Revoke or upgrade +## Revoke a token List tokens and revoke one by its `jti`: @@ -138,12 +138,19 @@ curl -X POST "$API_URL/workers/tokens/$JTI/revoke" \ ``` Revocation also disconnects attached workers and blocks artifact access. -Replacing the OAuth signing key revokes all worker tokens. +Replacing the OAuth signing key revokes all worker tokens. Older tokens without +signed labels and a `jti` need to be replaced. -Upgrade the engine before its workers, then run `pip install -U fluksio-worker` -on each host. Engine and worker must speak the same protocol; a mismatch is -reported and refused. Older tokens without signed labels and a `jti` need to -be replaced. +## Upgrading workers + +Engine and worker must speak the same protocol. A worker on another protocol is +refused and exits, so a release that bumps the protocol needs +`pip install -U fluksio-worker` on every worker host. Upgrade the engine first, +then the workers. + +The **Workers** screen shows each worker's package version and protocol. A +worker refused for an older protocol stays listed as **Needs upgrade** until it +attaches again, or until the engine restarts. ## Slurm provisioning diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 3b7a8c7..badc07c 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -4677,6 +4677,21 @@ export const WorkerInfoSchema = { } ], title: 'Ram Mb' + }, + protocol: { + type: 'integer', + title: 'Protocol', + default: 0 + }, + version: { + type: 'string', + title: 'Version', + default: '' + }, + behind: { + type: 'boolean', + title: 'Behind', + default: false } }, type: 'object', diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 1734eb0..3f1300e 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -1534,6 +1534,9 @@ export type WorkerInfo = { cpus?: number; gpus?: number; ram_mb?: (number | null); + protocol?: number; + version?: string; + behind?: boolean; }; /** diff --git a/frontend/src/routes/_layout/workers.tsx b/frontend/src/routes/_layout/workers.tsx index 7a0ca8d..7abb438 100644 --- a/frontend/src/routes/_layout/workers.tsx +++ b/frontend/src/routes/_layout/workers.tsx @@ -9,6 +9,7 @@ import { FlavorsService, WorkersService, } from "@/client" +import { Badge } from "@/components/ui/badge" import { Button } from "@/components/ui/button" import { Dialog, @@ -172,15 +173,38 @@ function Attached() { ) : null} - - {worker.cpus} cpu - {worker.gpus ? ` · ${worker.gpus} gpu` : ""} - {worker.ram_mb ? ` · ${Math.round(worker.ram_mb / 1024)} GB` : ""}{" "} - · {worker.in_flight}/{worker.max_parallel} in flight - + {worker.behind ? ( + + Refused for an older protocol. Upgrade it with{" "} + pip install -U fluksio-worker + + ) : ( + + {worker.cpus} cpu + {worker.gpus ? ` · ${worker.gpus} gpu` : ""} + {worker.ram_mb + ? ` · ${Math.round(worker.ram_mb / 1024)} GB` + : ""}{" "} + · {worker.in_flight}/{worker.max_parallel} in flight + + )} - - seen {ago(worker.last_seen ?? 0)} + + {worker.behind ? ( + + Needs upgrade + + ) : null} + + {worker.version ? `v${worker.version} · ` : ""}protocol{" "} + {worker.protocol} + + + {worker.behind ? "refused" : "seen"} {ago(worker.last_seen ?? 0)} + ))} diff --git a/worker/fluksio_worker/agent.py b/worker/fluksio_worker/agent.py index 8c91ca5..3a428f2 100644 --- a/worker/fluksio_worker/agent.py +++ b/worker/fluksio_worker/agent.py @@ -28,6 +28,7 @@ import argparse import asyncio import contextlib import glob +import importlib.metadata import ipaddress import json import logging @@ -167,6 +168,7 @@ class Agent: { "op": "hello", "protocol": PROTOCOL, + "version": _version(), "name": self.args.name, "labels": self.args.labels, "python": self.args.python, @@ -421,6 +423,14 @@ def _remap_cuda(mask: str, requested: str) -> str: return ",".join(out) +def _version() -> str: + """The installed fluksio-worker, or empty when this file was copied over.""" + try: + return importlib.metadata.version("fluksio-worker") + except importlib.metadata.PackageNotFoundError: + return "" + + def _venv_digest(python: str) -> str: """What is installed here, so the engine can say when it has drifted.""" try: