app/docs/code/workers.md

Remote workers

The engine runs where the automations are. The GPU is somewhere else, the Raspberry Pi with the relays is in a shed, and neither of them is on the same network as the other.

A worker is a process that runs the code of nodes marked for it. It dials out to the engine over one authenticated websocket, so nothing on that machine has to be reachable, and nothing has to expose the engine's state backend across hosts, which it never should.

Install and attach

pip install fluksio-worker

fluksio-worker \
  --url wss://api.${DOMAIN}/api/v1/workers/attach \
  --token "$FLUKSIO_WORKER_TOKEN" \
  --labels gpu,cuda12 \
  --python /opt/torch-venv/bin/python

fluksio-worker is its own distribution: the agent, the node runner, and websockets. Nothing of the engine, so a GPU box does not install a database driver in order to run a training step. An engine host already has it, and fluksio worker … is the same program.

Option	Default	What it does
--url	required	wss://…/api/v1/workers/attach
--token	$FLUKSIO_WORKER_TOKEN	the credential, minted on the engine
--name	this host's name	sent along; the worker list shows the token's name
--labels	the token's	comma-separated; must match the labels the token was minted with
--python	this interpreter	the interpreter node code runs on
--parallel	1	how many node calls it will take at once
--artifact-url	derived from --url	where the artifact store is, if not beside the socket
--cpus	what the job or the machine has	cores to advertise
--gpus	what the job says, else counted	GPUs to advertise
--ram-mb	what the job or the machine has	memory to advertise, in MB
--max-idle	never	stop after this many seconds with nothing running
--insecure	off	allow ws:// and http:// to an engine on another machine

The worker refuses a plain ws:// or http:// address unless the engine is on this machine (localhost, 127.0.0.0/8, ::1): its token and every secret a node is handed would cross the network in clear. --insecure is for a network you trust, such as an engine on the same LAN without TLS. Node code gets this host's environment less anything named like a credential (*TOKEN*, *SECRET*, *PASSWORD*), the same as on the engine; hand a node a credential as a secret instead.

--python is the important one. It is how this machine keeps its own wheels (the CUDA build, the vendor SDK, the thing that will not install anywhere else) without the engine ever installing them or knowing about them.

What it says it has

A worker reports its inventory when it attaches (cores, GPUs and memory) and the engine schedules against it: a node asking for two cores and a GPU goes to a machine that has them free, not merely to one carrying the right label.

Cores and memory are read off the machine, or off the batch job that started this worker (SLURM_CPUS_ON_NODE, SLURM_MEM_PER_NODE). What the job says it was given always wins for GPUs (SLURM_GPUS_ON_NODE, SLURM_JOB_GPUS, or FLUKSIO_WORKER_GPUS): a node with eight cards may have granted this job one, and advertising eight would be a lie the scheduler acts on. With nothing said, NVIDIA's device nodes are counted, the same as the engine does for its own machine, and --gpus overrides either. No vendor tool is asked, which is what keeps the one dependency from becoming two. A worker that reports nothing still attaches and is scheduled by its label alone, as every worker was before any of them reported anything.

The engine tells each call what it may use: thread caps, and the devices it may see. The worker starts a process per call, so it applies them at the only moment a numerical library still reads them: before the import.

--max-idle is for a worker something else started for one job, a batch scheduler say. It exits when nothing has run for that long, so the allocation goes back rather than idling until its walltime.

Mint the token

On the engine, as a superuser:

curl -X POST $FLUKSIO/workers/tokens -H "Authorization: Bearer $TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"name": "gpu-dev", "labels": ["gpu", "cuda12"]}'

Shown once, valid for a year, since a worker is a machine somebody sets up and leaves running. The name and labels are signed into it: labels decide which nodes, with their source and secrets, are sent to a worker, so a worker cannot choose its own. It shows up under the token's name, and a worker whose --labels differ from the token's is refused.

curl -s $FLUKSIO/workers/tokens -H "Authorization: Bearer $TOKEN" | jq
curl -X POST $FLUKSIO/workers/tokens/$JTI/revoke -H "Authorization: Bearer $TOKEN"

The first lists what was minted, by jti (the tokens themselves are not kept); the second refuses one from then on, here and at the artifact endpoint, and drops a worker attached with it. Rotating the keypair agent tokens are signed with still revokes every worker at once. A token minted before tokens carried their labels is refused: mint a new one.

??? note "A host where pip is not an option"

The two files work copied into one directory and run with `python agent.py
…`. The engine serves the runner itself at `GET /api/v1/workers/runtime`.
It is the same module its own local workers run, deliberately standard
library only.

Send a node to it

A node declares the label of the machine it needs:

{
  "id": "train",
  "device": "gpu",
  "device_policy": "require",
  "timeout": 7200
}

device_policy	Behaviour when nothing carrying the label is attached
require (default)	the run stays queued and says what it is waiting for
prefer	it runs on the engine instead

prefer is what makes a flow work before the GPU box exists. require is what you want once it does.

!!! note "Not in the panel yet"

`device` and `device_policy`, which machine a node's code runs on, are set
through the API rather than the panel, with `PUT /flows/{name}`.

What follows from this

• The node's source travels with every call. Nothing has to be deployed to the worker, and changing a node's code takes effect on the next execution.
• A node bound to a device is compiled on that machine. A node importing torch is correct on the GPU box and a missing module on the engine, so checking it here would fail something that is fine.
• import fluksio inside a node is the worker's own reporter. emit, save_artifact, load_artifact, installed before your code runs, so an installed fluksio package on that box never shadows it.
• Cancelling a run kills what it is executing, there or here, and leaves other runs of the same node alone.
• If the worker disappears mid-call, the run fails in seconds with worker went away mid-call rather than waiting out its timeout.
• A worker sends a heartbeat every ten seconds while it executes, so a long node is distinguishable from a dead socket. Ninety seconds of nothing at all, not even a heartbeat, fails the call as gone. A heartbeat says the agent is alive and nothing about the node, so it never satisfies a node's own timeout: one set to thirty seconds fires after thirty seconds of the node reporting nothing, wherever it runs.

Artifacts across machines

An artifact reference names content by its hash, not a location, so it stays valid wherever the store is reachable from. A worker that shares the engine's filesystem writes to it directly; one that does not fetches and uploads over HTTP, using the artifact endpoint beside the socket it already has. Either way your node code is the same two calls.

A fetch is cached on the worker by digest, since content addressing means an entry is never stale. Nothing expires on its own, so the cache is bounded by size and the oldest fall out: FLUKSIO_ARTIFACT_CACHE says where it lives (default a directory in the temporary directory) and FLUKSIO_ARTIFACT_CACHE_BYTES how much it holds (default 1 GiB). Worth raising where a worker reads the same large inputs repeatedly, and worth leaving alone where it reads a media stream — those are chunks nothing asks for twice.

Seeing what is attached

curl -s $FLUKSIO/workers -H "Authorization: Bearer $TOKEN" | jq

Name, labels, how many calls it will take at once, how many are in flight, when it attached, when it was last seen, its Python version, a digest of its environment, and what it says it has: cores, GPUs and memory.

Upgrading

The engine and the worker speak a version-matched protocol, and a worker announcing anything else is refused rather than half-understood. Protocol 2, the one that carries inventory, is fluksio-worker 0.2.0. An older agent is told so on the socket and stops, rather than retrying against an engine that will never accept it; pip install -U fluksio-worker on that host is the whole upgrade. Nothing changed in the runner served at GET /api/v1/workers/runtime, so a host that copies its two files copies the same one as before.

A token minted before tokens carried their labels and a jti is refused. Mint each existing worker host a new one with POST /workers/tokens, naming the labels it should carry, and swap it in for FLUKSIO_WORKER_TOKEN.

Upgrade the engine before its workers. A worker sends its token as an Authorization header on the handshake, where it stays out of access logs, and an engine from before that reads it only from the URL: it refuses the newer worker at the handshake. The engine still takes the old ?token= for one release, so older workers keep attaching to a newer engine meanwhile.

Machines from a batch scheduler

A cluster is not a machine that attaches and stays; it is a queue somebody else owns. So Fluksio does not submit nodes to Slurm. It submits a job whose payload is an ordinary worker dialling back in, and from there everything works the way it already does: the same protocol, the same artifacts, the same cancellation.

Write the clusters into provisioners.json beside the flows:

[{
  "type": "slurm",
  "name": "hpc",
  "login": "me@login.cluster",
  "ssh_key": "/secrets/hpc_ed25519",
  "engine_url": "wss://api.${DOMAIN}/api/v1/workers/attach",
  "max_idle_s": 300,
  "provision_timeout_s": 900,
  "profiles": [{
    "name": "gpu-small",
    "cpus": 8, "gpus": 1, "ram_mb": 65536,
    "labels": ["gpu"],
    "sbatch": ["--partition=gpu", "--gres=gpu:1", "--time=04:00:00"],
    "prerun": ["module load cuda/12", "source ~/venvs/flux/bin/activate"]
  }]
}]

A profile is what the scheduler is asked for, where a flavor is what a node asks for. They are separate on purpose, and agree when you set them up to.

When a node needs a machine nothing attached can give, and a profile fits, the engine sbatches one over the system ssh, and the run waits meanwhile, saying so. prerun owns the environment: a module load, or a venv with fluksio-worker already in it. The generated script runs no pip install. It hands the worker a token of its own, good for 24 hours, as FLUKSIO_WORKER_TOKEN rather than --token, so other users of a shared compute node cannot read it from ps.

One outstanding request per profile, however often it is asked for. A job that never attaches within provision_timeout_s is scancelled, as is anything outstanding when the engine stops. --max-idle is what ends the job at the other end, so an allocation goes back rather than idling to its walltime.

Both take 0 for "no limit": provision_timeout_s: 0 waits for as long as the queue does, which is what a cluster that queues overnight needs, and max_idle_s: 0 keeps the machine for the job's whole walltime.

GET /workers/resources reports what is outstanding and what last went wrong.

!!! note "It needs a route out"

A compute node must be able to open a connection to the engine. That is
true of most clusters and false of air-gapped ones; there is no staging
path today.

What a worker is not

It is not a second engine. Subscriptions, schedules, webhooks, the dashboards and the run queue all stay in one process, which keeps a message having one definition and a cron tick happening once. A worker executes node bodies.

Running two engines against one data directory is not supported. Distribute work with workers.

See also

• Runs: pipelines that finish
• Writing node code
• Getting started: data science