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: