From 2c369ac75f2cf1bc82367a294578ae87214aae20 Mon Sep 17 00:00:00 2001 From: stroblme Date: Fri, 21 Aug 2026 21:54:10 +0200 Subject: [PATCH] Split the worker into a distribution of its own MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A cluster or GPU host installs `pip install fluksio-worker` and gets the agent and the runner, not psycopg, numpy and the MCP SDK. The engine depends on it as a workspace member, so the file it launches node code with is the same file a remote worker runs — which is what keeps a node unable to tell the difference. Copying the two files by hand still works. Co-Authored-By: Claude Opus 5 (1M context) --- Makefile | 1 + backend/Dockerfile | 5 +++ backend/fluksio/api/routes/workers.py | 4 +- backend/fluksio/flow/controller.py | 2 +- backend/fluksio/flow/workers.py | 6 ++- backend/fluksio/main.py | 2 +- backend/pyproject.toml | 15 ++++--- backend/tests/flow/test_workers.py | 2 +- pyproject.toml | 2 +- uv.lock | 14 +++++++ worker/README.md | 21 ++++++++++ worker/fluksio_worker/__init__.py | 6 +++ worker/fluksio_worker/__main__.py | 5 +++ .../fluksio_worker/agent.py | 40 +++++++------------ .../fluksio_worker/py.typed | 0 .../fluksio_worker}/worker_main.py | 8 ++-- worker/pyproject.toml | 31 ++++++++++++++ 17 files changed, 122 insertions(+), 42 deletions(-) create mode 100644 worker/README.md create mode 100644 worker/fluksio_worker/__init__.py create mode 100644 worker/fluksio_worker/__main__.py rename backend/fluksio/worker/fluksio_worker.py => worker/fluksio_worker/agent.py (90%) rename backend/fluksio/worker/__init__.py => worker/fluksio_worker/py.typed (100%) rename {backend/fluksio/flow => worker/fluksio_worker}/worker_main.py (98%) create mode 100644 worker/pyproject.toml diff --git a/Makefile b/Makefile index 1e8ecab..ff8a120 100644 --- a/Makefile +++ b/Makefile @@ -170,6 +170,7 @@ lint-backend: ## Lint backend with ruff + mypy cd backend && uv run ruff check . cd backend && uv run ruff format --check . cd backend && uv run mypy fluksio + cd worker && uv run --no-project --with mypy mypy fluksio_worker lint-frontend: ## Lint frontend with biome cd frontend && bun run lint diff --git a/backend/Dockerfile b/backend/Dockerfile index ad59a54..e122c3a 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -33,6 +33,11 @@ COPY ./backend/pyproject.toml ./backend/alembic.ini /app/backend/ COPY ./backend/fluksio /app/backend/fluksio +# The worker is a workspace member of its own, so the engine's sync needs it +# present to resolve the dependency on it. It is also what the engine runs +# node code with. +COPY ./worker /app/worker + # Sync the project # Ref: https://docs.astral.sh/uv/guides/integration/docker/#intermediate-layers RUN --mount=type=cache,target=/root/.cache/uv \ diff --git a/backend/fluksio/api/routes/workers.py b/backend/fluksio/api/routes/workers.py index d628c51..2c2520b 100644 --- a/backend/fluksio/api/routes/workers.py +++ b/backend/fluksio/api/routes/workers.py @@ -15,12 +15,12 @@ from typing import Any from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket from fastapi.responses import PlainTextResponse +from fluksio_worker import worker_main from jwt.exceptions import InvalidTokenError from pydantic import BaseModel, Field from fluksio.api.deps import get_current_active_superuser, get_current_user from fluksio.core import security -from fluksio.flow import worker_main from fluksio.flow.remote import PROTOCOL, RemoteWorker, RemoteWorkerHub logger = logging.getLogger(__name__) @@ -103,7 +103,7 @@ def read_runtime() -> str: """The worker's own code, so a fresh host installs by fetching one file. It is the same module the engine's local workers run — deliberately - standard library only, and with nothing of the app importable in it. + standard library only, and with nothing of the engine importable in it. """ return worker_main.__file__ and open(worker_main.__file__).read() diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index f362a79..d8d8b56 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -20,6 +20,7 @@ from typing import Any, cast from fastapi import FastAPI from fastapi.concurrency import run_in_threadpool +from fluksio_worker.worker_main import load_function from fluksio.core.config import settings from fluksio.flow.alerts import AlertManager @@ -60,7 +61,6 @@ from fluksio.flow.secrets import SecretNotFound, resolve_params from fluksio.flow.state import MemoryState, StateBackend from fluksio.flow.store import LIB_DIR, FlowNotFound, FlowStore, LibNotFound from fluksio.flow.supervision import Supervisor -from fluksio.flow.worker_main import load_function from fluksio.flow.workers import PythonWorkerPool logger = logging.getLogger(__name__) diff --git a/backend/fluksio/flow/workers.py b/backend/fluksio/flow/workers.py index aac81fa..068e6ca 100644 --- a/backend/fluksio/flow/workers.py +++ b/backend/fluksio/flow/workers.py @@ -26,11 +26,15 @@ from collections.abc import Callable from pathlib import Path from typing import Any +from fluksio_worker import worker_main as _worker_main + from fluksio.flow.events import EventBus logger = logging.getLogger(__name__) -WORKER_MAIN = Path(__file__).with_name("worker_main.py") +#: The runner, in the worker distribution — the same file a remote worker runs, +#: so a node cannot tell which kind of worker it is on. +WORKER_MAIN = Path(_worker_main.__file__) #: Importing what a node needs can be slow the first time; compiling is not #: something a person is watching a spinner for. diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 6d0967c..174c58d 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -8,6 +8,7 @@ from fastapi import FastAPI from fastapi.concurrency import run_in_threadpool from fastapi.responses import JSONResponse from fastapi.routing import APIRoute +from fluksio_worker.worker_main import ARTIFACT_DIR_ENV from starlette.middleware.cors import CORSMiddleware from fluksio.api.main import api_router @@ -33,7 +34,6 @@ from fluksio.flow.secrets import init_secrets from fluksio.flow.state import MemoryState, RedisState, StateBackend from fluksio.flow.store import FlowStore from fluksio.flow.watchdog import LoopWatchdog -from fluksio.flow.worker_main import ARTIFACT_DIR_ENV from fluksio.flow.workers import PythonWorkerPool diff --git a/backend/pyproject.toml b/backend/pyproject.toml index dad5a35..db06bec 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -28,8 +28,12 @@ dependencies = [ "influxdb-client[async]>=1.40.0", "croniter>=1.3.0", "mcp>=1.29,<2", + "fluksio-worker>=0.1,<0.2", ] +[tool.uv.sources] +fluksio-worker = { workspace = true } + [dependency-groups] dev = [ "pytest<8.0.0,>=7.4.3", @@ -42,6 +46,11 @@ dev = [ [tool.hatch.build.targets.wheel] packages = ["fluksio"] +# Named rather than excluded: this directory also holds a working tree's +# runtime data — `flow-data/` with the user venv in it — which is not source. +[tool.hatch.build.targets.sdist] +include = ["fluksio", "tests", "scripts", "alembic.ini", "pyproject.toml", "README.md"] + [build-system] requires = ["hatchling"] build-backend = "hatchling.build" @@ -92,12 +101,6 @@ ignore = [ # Printing is what this one is about: node code is user code, and `print` is # how it says things. "tests/flow/test_logs.py" = ["ARG001", "T201"] -# This one takes its own directory off sys.path before the rest of its imports -# run, which is the whole point of doing it there. -"fluksio/flow/worker_main.py" = ["E402"] -# A standalone script copied onto another machine: it has no logger configured -# before it tells you the one dependency it is missing. -"fluksio/worker/fluksio_worker.py" = ["T201"] [tool.ruff.lint.pyupgrade] # Preserve types, even if a file imports `from __future__ import annotations`. diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index 8f389b8..e877f20 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -6,10 +6,10 @@ import time from collections.abc import Iterator import pytest +from fluksio_worker.worker_main import ARTIFACT_DIR_ENV from fluksio.flow.artifacts import ArtifactStore from fluksio.flow.messages import DType, MessageSpec -from fluksio.flow.worker_main import ARTIFACT_DIR_ENV from fluksio.flow.workers import NodeCancelled, NodeTimeout, PythonWorkerPool diff --git a/pyproject.toml b/pyproject.toml index 3935549..1cc4d42 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,2 +1,2 @@ [tool.uv.workspace] -members = ["backend"] +members = ["backend", "worker"] diff --git a/uv.lock b/uv.lock index ce8fa8d..f0dfdad 100644 --- a/uv.lock +++ b/uv.lock @@ -9,6 +9,7 @@ resolution-markers = [ [manifest] members = [ "fluksio", + "fluksio-worker", ] [[package]] @@ -765,6 +766,7 @@ dependencies = [ { name = "email-validator" }, { name = "emails" }, { name = "fastapi", extra = ["standard"] }, + { name = "fluksio-worker" }, { name = "httpx" }, { name = "influxdb-client", extra = ["async"] }, { name = "jinja2" }, @@ -801,6 +803,7 @@ requires-dist = [ { name = "email-validator", specifier = ">=2.1.0.post1,<3.0.0.0" }, { name = "emails", specifier = ">=0.6,<1.0" }, { name = "fastapi", extras = ["standard"], specifier = ">=0.114.2,<1.0.0" }, + { name = "fluksio-worker", editable = "worker" }, { name = "httpx", specifier = ">=0.25.1,<1.0.0" }, { name = "influxdb-client", extras = ["async"], specifier = ">=1.40.0" }, { name = "jinja2", specifier = ">=3.1.4,<4.0.0" }, @@ -827,6 +830,17 @@ dev = [ { name = "ruff", specifier = ">=0.2.2,<1.0.0" }, ] +[[package]] +name = "fluksio-worker" +version = "0.1.0" +source = { editable = "worker" } +dependencies = [ + { name = "websockets" }, +] + +[package.metadata] +requires-dist = [{ name = "websockets", specifier = ">=12" }] + [[package]] name = "frozenlist" version = "1.8.0" diff --git a/worker/README.md b/worker/README.md new file mode 100644 index 0000000..d28143b --- /dev/null +++ b/worker/README.md @@ -0,0 +1,21 @@ +# fluksio-worker + +Runs [Fluksio](https://fluksio.com) nodes on a machine the engine cannot reach — +a GPU box, a cluster node, anything behind a NAT. It dials *out* to the engine +over one authenticated websocket, so nothing has to be exposed here. + +```sh +pip install fluksio-worker + +fluksio-worker \ + --url wss://api.example.com/api/v1/workers/attach \ + --token "$FLUKSIO_WORKER_TOKEN" \ + --labels gpu,cuda12 \ + --python /opt/torch-venv/bin/python +``` + +`--python` is the interpreter node code runs on, which is how this machine keeps +its own wheels without the engine ever installing them. Mint a token from the +engine with `POST /api/v1/workers/tokens`. + +Linux and macOS. diff --git a/worker/fluksio_worker/__init__.py b/worker/fluksio_worker/__init__.py new file mode 100644 index 0000000..0aa1cbc --- /dev/null +++ b/worker/fluksio_worker/__init__.py @@ -0,0 +1,6 @@ +"""The Fluksio worker: node execution on a machine the engine cannot reach. + +Deliberately import-free. The engine imports :mod:`fluksio_worker.worker_main` +to find the runner it launches as a subprocess, and paying for the agent's +imports on that path would be a cost for nothing. +""" diff --git a/worker/fluksio_worker/__main__.py b/worker/fluksio_worker/__main__.py new file mode 100644 index 0000000..740374d --- /dev/null +++ b/worker/fluksio_worker/__main__.py @@ -0,0 +1,5 @@ +"""``python -m fluksio_worker`` — the same entry point as the console script.""" + +from fluksio_worker.agent import main + +raise SystemExit(main()) diff --git a/backend/fluksio/worker/fluksio_worker.py b/worker/fluksio_worker/agent.py similarity index 90% rename from backend/fluksio/worker/fluksio_worker.py rename to worker/fluksio_worker/agent.py index 4124902..36befa9 100644 --- a/backend/fluksio/worker/fluksio_worker.py +++ b/worker/fluksio_worker/agent.py @@ -1,10 +1,9 @@ """The agent that runs Fluksio nodes on a machine the engine cannot reach. -Copy this file and ``worker_main.py`` onto the box with the GPU, point it at -the engine, and it dials in:: +Point it at the engine on the box with the GPU and it dials in:: - pip install websockets - python fluksio_worker.py --url wss://api.example.com/api/v1/workers/attach \\ + pip install fluksio-worker + fluksio-worker --url wss://api.example.com/api/v1/workers/attach \\ --token "$FLUKSIO_WORKER_TOKEN" --labels gpu --python /opt/venv/bin/python It connects *out*, so the engine needs no route back and nothing has to expose @@ -13,9 +12,11 @@ subprocesses running ``worker_main.py``, hand each call to one, and pass back everything that comes out — including the metrics a training loop reports while it is still running. -Deliberately one file with one dependency. Nothing of the app is imported +Deliberately two files with one dependency. Nothing of the engine is imported here; a worker host installs Python, ``websockets``, and whatever the nodes -themselves need. +themselves need. Where pip is not an option, copying this file and +``worker_main.py`` into one directory and running ``python agent.py`` is the +same thing — the engine serves the runner at ``GET /api/v1/workers/runtime``. """ from __future__ import annotations @@ -40,7 +41,6 @@ except ImportError: # pragma: no cover - the one dependency, named plainly ) raise SystemExit(1) from None -logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("fluksio-worker") PROTOCOL = 1 @@ -50,22 +50,9 @@ HEARTBEAT_S = 10.0 MAX_BACKOFF_S = 30.0 -def _worker_main() -> Path: - """The user-code runner: beside this file once deployed, or in a checkout. - - On a worker host the two files sit together, which is what the install - instructions say. Run straight from a clone and the runner is one directory - over, in the engine's own package — worth finding, so trying this out does - not start with copying files around. - """ - here = Path(__file__).resolve().parent - for candidate in (here / "worker_main.py", here.parent / "flow" / "worker_main.py"): - if candidate.exists(): - return candidate - return here / "worker_main.py" - - -WORKER_MAIN = _worker_main() +#: The user-code runner, beside this file — installed together, or copied +#: together onto a host where pip is not an option. +WORKER_MAIN = Path(__file__).resolve().with_name("worker_main.py") class Subprocess: @@ -267,7 +254,10 @@ def _venv_digest(python: str) -> str: return hashlib.sha256(listing.encode()).hexdigest()[:16] -def main() -> int: +def main(argv: list[str] | None = None) -> int: + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) parser = argparse.ArgumentParser(description="Run Fluksio nodes on this machine.") parser.add_argument("--url", required=True, help="wss://…/api/v1/workers/attach") parser.add_argument( @@ -288,7 +278,7 @@ def main() -> int: ) parser.add_argument("--parallel", type=int, default=1) parser.add_argument("--artifact-url", default="") - args = parser.parse_args() + args = parser.parse_args(argv) if not args.token: parser.error("a token is required (--token or FLUKSIO_WORKER_TOKEN)") diff --git a/backend/fluksio/worker/__init__.py b/worker/fluksio_worker/py.typed similarity index 100% rename from backend/fluksio/worker/__init__.py rename to worker/fluksio_worker/py.typed diff --git a/backend/fluksio/flow/worker_main.py b/worker/fluksio_worker/worker_main.py similarity index 98% rename from backend/fluksio/flow/worker_main.py rename to worker/fluksio_worker/worker_main.py index e52a04a..1a8a1ee 100644 --- a/backend/fluksio/flow/worker_main.py +++ b/worker/fluksio_worker/worker_main.py @@ -2,7 +2,7 @@ This runs under the *user* venv's interpreter, so a node's imports resolve against what the Modules page installed rather than against the engine's own -packages. Nothing of the app is importable here — the file is handed to the +packages. Nothing of the engine is importable here — the file is handed to the interpreter by path and is deliberately standard library only. The protocol is one JSON object per line: a request arrives on stdin, the reply @@ -33,7 +33,7 @@ import os import sys # Handed to the interpreter by path, so CPython puts this file's own directory -# — the engine's ``app/flow`` — on ``sys.path[0]``. A node doing ``import +# — ``fluksio_worker`` — on ``sys.path[0]``. A node doing ``import # queue`` would then get the engine's queue module rather than the standard # library's. Drop it before anything else can import. (``-P`` and # ``PYTHONSAFEPATH`` do this at startup, but both are 3.11+.) @@ -131,8 +131,8 @@ def _store_bytes(data: bytes, name: str, media_type: str) -> dict[str, Any]: A worker in the engine's own container writes the file; one on another host puts it over HTTP. Node code cannot tell the difference, which is the point — the same flow runs either place. The hashing is repeated from - ``fluksio.flow.artifacts`` rather than imported, because nothing of the app is - importable here. + ``fluksio.flow.artifacts`` rather than imported, because nothing of the engine + is importable here. """ digest = "sha256:" + hashlib.sha256(data).hexdigest() directory = os.environ.get(ARTIFACT_DIR_ENV) diff --git a/worker/pyproject.toml b/worker/pyproject.toml new file mode 100644 index 0000000..19e9f56 --- /dev/null +++ b/worker/pyproject.toml @@ -0,0 +1,31 @@ +[project] +name = "fluksio-worker" +version = "0.1.0" +description = "Runs Fluksio nodes on a machine the engine cannot reach" +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "websockets>=12", +] + +[project.scripts] +fluksio-worker = "fluksio_worker.agent:main" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["fluksio_worker"] + +[tool.ruff] +# The engine's config, so both distributions are linted the same way. +extend = "../backend/pyproject.toml" + +[tool.ruff.lint.per-file-ignores] +# This one takes its own directory off sys.path before the rest of its imports +# run, which is the whole point of doing it there. +"fluksio_worker/worker_main.py" = ["E402"] +# Copied onto another machine it has no logger configured before it tells you +# the one dependency it is missing. +"fluksio_worker/agent.py" = ["T201"]