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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
+13
-6
@@ -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
|
||||
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -1534,6 +1534,9 @@ export type WorkerInfo = {
|
||||
cpus?: number;
|
||||
gpus?: number;
|
||||
ram_mb?: (number | null);
|
||||
protocol?: number;
|
||||
version?: string;
|
||||
behind?: boolean;
|
||||
};
|
||||
|
||||
/**
|
||||
|
||||
@@ -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() {
|
||||
</span>
|
||||
) : null}
|
||||
</span>
|
||||
<span className="pl-6 font-mono text-xs text-muted-foreground">
|
||||
{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
|
||||
</span>
|
||||
{worker.behind ? (
|
||||
<span className="pl-6 text-xs text-muted-foreground">
|
||||
Refused for an older protocol. Upgrade it with{" "}
|
||||
<code className="font-mono">pip install -U fluksio-worker</code>
|
||||
</span>
|
||||
) : (
|
||||
<span className="pl-6 font-mono text-xs text-muted-foreground">
|
||||
{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
|
||||
</span>
|
||||
)}
|
||||
</span>
|
||||
<span className="text-xs text-muted-foreground">
|
||||
seen {ago(worker.last_seen ?? 0)}
|
||||
<span className="flex items-center gap-3 text-xs text-muted-foreground">
|
||||
{worker.behind ? (
|
||||
<Badge
|
||||
variant="outline"
|
||||
className="border-transparent bg-destructive/15 text-destructive"
|
||||
>
|
||||
Needs upgrade
|
||||
</Badge>
|
||||
) : null}
|
||||
<span className="font-mono">
|
||||
{worker.version ? `v${worker.version} · ` : ""}protocol{" "}
|
||||
{worker.protocol}
|
||||
</span>
|
||||
<span>
|
||||
{worker.behind ? "refused" : "seen"} {ago(worker.last_seen ?? 0)}
|
||||
</span>
|
||||
</span>
|
||||
</div>
|
||||
))}
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user