stroblme
e3171f87e2
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.
477 lines
17 KiB
Python
477 lines
17 KiB
Python
"""A pool of subprocesses that run the code people write in python nodes.
|
|
|
|
User code used to be ``exec``'d in the engine process, where an ``os._exit``,
|
|
a segfaulting C extension or a ``while True`` took the whole engine with it.
|
|
Here each node call is an RPC to a long-lived worker: a crash costs one
|
|
subprocess, a timeout is a kill, and a cancel is the same kill on request.
|
|
|
|
The workers run the user venv's interpreter, so what the Modules page installs
|
|
is what a node can import. Only JSON crosses the boundary, which the typed
|
|
message contract already guarantees for everything a node consumes or provides.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import queue
|
|
import select
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
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__)
|
|
|
|
#: 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.
|
|
COMPILE_TIMEOUT = 60.0
|
|
|
|
#: Environment the worker is not given. ``SECRET_KEY`` decrypts every stored
|
|
#: secret, not only the ones bound to the node asking.
|
|
ENV_DENY_PREFIXES = ("POSTGRES_", "FIRST_SUPERUSER", "SENTRY_DSN")
|
|
ENV_DENY_WORDS = ("PASSWORD", "SECRET", "TOKEN")
|
|
|
|
|
|
def worker_env(extra: dict[str, str] | None = None) -> dict[str, str]:
|
|
"""The engine's environment with its credentials taken out.
|
|
|
|
A denylist, so ``PATH``, ``HOME``, the locale and whatever else a venv
|
|
needs still reach the worker. This is not a sandbox: the worker is the same
|
|
user in the same container, with the same filesystem and the same network.
|
|
It only means node code cannot read the deployment's secrets straight out
|
|
of its own environment.
|
|
|
|
``extra`` is put back afterwards, deliberately: the artifact store's
|
|
address is something the worker has to be told, and one of its names would
|
|
otherwise be caught by the denylist for containing "TOKEN".
|
|
"""
|
|
env = {
|
|
key: value
|
|
for key, value in os.environ.items()
|
|
if not key.startswith(ENV_DENY_PREFIXES)
|
|
and not any(word in key.upper() for word in ENV_DENY_WORDS)
|
|
}
|
|
env.update(extra or {})
|
|
return env
|
|
|
|
|
|
class RemoteError(Exception):
|
|
"""Something that went wrong inside a worker, re-raised on this side.
|
|
|
|
``remote_traceback`` is the worker's own traceback, trimmed to the node's
|
|
code — the frames here are the RPC, which the node's author did not write.
|
|
"""
|
|
|
|
def __init__(self, message: str, remote_traceback: str = "") -> None:
|
|
super().__init__(message)
|
|
self.remote_traceback = remote_traceback
|
|
|
|
|
|
class NodeTimeout(RemoteError):
|
|
"""The node ran past its timeout, so its worker was killed."""
|
|
|
|
|
|
class NodeCancelled(RemoteError):
|
|
"""Someone asked for this node to stop while it was running."""
|
|
|
|
|
|
_remote_types: dict[str, type[RemoteError]] = {}
|
|
|
|
|
|
def _remote_class(name: str) -> type[RemoteError]:
|
|
"""A ``RemoteError`` wearing the remote exception's name.
|
|
|
|
The engine renders a node failure as ``f"{type(exc).__name__}: {exc}"``, so
|
|
the author still reads ``ValueError: bad input`` rather than the name of
|
|
the transport that carried it.
|
|
"""
|
|
cls = _remote_types.get(name)
|
|
if cls is None:
|
|
cls = type(name, (RemoteError,), {})
|
|
_remote_types[name] = cls
|
|
return cls
|
|
|
|
|
|
class _Worker:
|
|
"""One subprocess, and the framing of one request/response over its pipes."""
|
|
|
|
def __init__(
|
|
self, python: str, generation: int, env: dict[str, str] | None = None
|
|
) -> None:
|
|
self.generation = generation
|
|
self.cancelled = False
|
|
# What a read took past the end of a line. A node reporting quickly
|
|
# puts several lines in one chunk, and the next one is the caller's to
|
|
# read — dropping it would lose a metric, returning it with the first
|
|
# would be unparseable.
|
|
self._buffer = bytearray()
|
|
self.proc = subprocess.Popen(
|
|
[python, str(WORKER_MAIN)],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
close_fds=True,
|
|
env=worker_env(env),
|
|
)
|
|
|
|
def alive(self) -> bool:
|
|
return self.proc.poll() is None
|
|
|
|
def send(self, request: dict[str, Any]) -> None:
|
|
assert self.proc.stdin is not None
|
|
self.proc.stdin.write((json.dumps(request) + "\n").encode())
|
|
self.proc.stdin.flush()
|
|
|
|
def read_line(self, deadline: float) -> str | None:
|
|
"""One line; ``None`` past the deadline, ``""`` if the pipe closed."""
|
|
assert self.proc.stdout is not None
|
|
fd = self.proc.stdout.fileno()
|
|
while True:
|
|
end = self._buffer.find(b"\n")
|
|
if end >= 0:
|
|
line = bytes(self._buffer[: end + 1])
|
|
del self._buffer[: end + 1]
|
|
return line.decode(errors="replace")
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
return None
|
|
ready, _, _ = select.select([fd], [], [], remaining)
|
|
if not ready:
|
|
return None
|
|
chunk = os.read(fd, 65536)
|
|
if not chunk:
|
|
# Partial output before the pipe closed is a half-written
|
|
# line, which is no more use than none at all.
|
|
self._buffer.clear()
|
|
return ""
|
|
self._buffer += chunk
|
|
|
|
def kill(self) -> None:
|
|
"""SIGKILL: user code has no cleanup we can trust to run.
|
|
|
|
The pipes are deliberately left alone — a cancel runs on another
|
|
thread than the one blocked reading this worker, and closing the fd
|
|
out from under it is how you get a reader on somebody else's socket.
|
|
Killing the process closes the far end, which is what wakes the reader.
|
|
"""
|
|
try:
|
|
self.proc.send_signal(signal.SIGKILL)
|
|
except OSError:
|
|
pass
|
|
# Reap it, so a killed worker does not linger as a zombie.
|
|
try:
|
|
self.proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
logger.warning("Worker %s did not die", self.proc.pid)
|
|
|
|
|
|
class PythonWorkerPool:
|
|
"""Fixed set of worker slots, handed out one call at a time.
|
|
|
|
A slot holds a live worker or nothing; a slot that is empty, dead or from
|
|
before the last ``respawn_all`` spawns a fresh process the next time it is
|
|
taken. There is no background thread — the pool only does work while a node
|
|
is calling it.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
python: str,
|
|
size: int = 4,
|
|
events: EventBus | None = None,
|
|
env: dict[str, str] | None = None,
|
|
) -> None:
|
|
self.python = python
|
|
self.size = size
|
|
self.events = events
|
|
# Put into every worker's environment past the denylist — where the
|
|
# artifact store is, which node code needs and cannot guess.
|
|
self.env = env or {}
|
|
self._idle: queue.Queue[_Worker | None] = queue.Queue()
|
|
# Keyed by (run, node): a sweep has the same node executing in several
|
|
# runs at once, and cancelling one of them must not kill the others.
|
|
# The live pipeline's own executions carry an empty run.
|
|
self._running: dict[tuple[str, str], _Worker] = {}
|
|
self._generation = 0
|
|
self._stopped = False
|
|
self._lock = threading.Lock()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Lifecycle
|
|
# -------------------------------------------------------------------------
|
|
|
|
def start(self) -> None:
|
|
"""Open the slots. Processes are spawned by the first call that needs one."""
|
|
for _ in range(self.size):
|
|
self._idle.put(None)
|
|
|
|
def stop(self) -> None:
|
|
self._stopped = True
|
|
self._generation += 1
|
|
for worker in list(self._running.values()):
|
|
worker.kill()
|
|
for slot in self._drain():
|
|
if slot is not None:
|
|
slot.kill()
|
|
# A node thread can be blocked on a slot right now, and the threads are
|
|
# not daemons — without a sentinel each it waits for a worker that is
|
|
# never coming back, and the interpreter cannot exit.
|
|
for _ in range(self.size):
|
|
self._idle.put(None)
|
|
|
|
def respawn_all(self) -> None:
|
|
"""Retire every worker, so the next call picks up a changed venv.
|
|
|
|
Idle ones go now; a busy one is replaced when it comes back, because
|
|
its generation no longer matches.
|
|
"""
|
|
with self._lock:
|
|
self._generation += 1
|
|
for slot in self._drain():
|
|
if slot is not None:
|
|
slot.kill()
|
|
self._idle.put(None)
|
|
|
|
def _drain(self) -> list[_Worker | None]:
|
|
slots = []
|
|
while True:
|
|
try:
|
|
slots.append(self._idle.get_nowait())
|
|
except queue.Empty:
|
|
return slots
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Slots
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _acquire(self) -> _Worker:
|
|
"""Take a slot, blocking while every worker is busy — that is the backpressure."""
|
|
if self._stopped:
|
|
raise RemoteError("the worker pool is shutting down")
|
|
slot = self._idle.get()
|
|
if self._stopped:
|
|
# The sentinel stop() put back, so this is a wake-up rather than a slot.
|
|
raise RemoteError("the worker pool is shutting down")
|
|
if (
|
|
slot is not None
|
|
and slot.alive()
|
|
and slot.generation == self._generation
|
|
and not slot.cancelled
|
|
):
|
|
return slot
|
|
if slot is not None:
|
|
slot.kill()
|
|
try:
|
|
return _Worker(self.python, self._generation, self.env)
|
|
except Exception as exc:
|
|
self._idle.put(None)
|
|
raise RemoteError(f"worker unavailable: {exc}") from exc
|
|
|
|
def _release(self, worker: _Worker) -> None:
|
|
reusable = (
|
|
worker.alive()
|
|
and worker.generation == self._generation
|
|
and not worker.cancelled
|
|
)
|
|
if reusable:
|
|
self._idle.put(worker)
|
|
return
|
|
worker.kill()
|
|
self._idle.put(None)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Calls
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _request(
|
|
self,
|
|
payload: dict[str, Any],
|
|
timeout: float,
|
|
node_id: str = "",
|
|
run_id: str = "",
|
|
on_event: Callable[[dict[str, Any]], None] | None = None,
|
|
) -> dict[str, Any]:
|
|
worker = self._acquire()
|
|
key = (run_id, node_id)
|
|
if node_id:
|
|
self._running[key] = worker
|
|
self._publish(
|
|
{
|
|
"type": "node_started",
|
|
# A flow name cannot contain a dot, so this is exact.
|
|
"flow": node_id.split(".", 1)[0],
|
|
"node": node_id,
|
|
"run": run_id,
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
try:
|
|
try:
|
|
worker.send(payload)
|
|
except OSError as exc:
|
|
raise RemoteError(f"worker died: {exc}") from exc
|
|
|
|
while True:
|
|
# Reset per line, so the timeout measures how long the node has
|
|
# been silent rather than how long it has been working. A node
|
|
# that reports nothing is still held to it, which is what keeps
|
|
# the deadline meaningful for the ones that never report.
|
|
line = worker.read_line(time.monotonic() + timeout)
|
|
if line:
|
|
try:
|
|
message = dict(json.loads(line))
|
|
except (TypeError, ValueError) as exc:
|
|
# A reply we cannot read leaves this worker out of step:
|
|
# whatever is still in its pipe would be taken by the
|
|
# next request on this slot as its own answer. Retire it.
|
|
worker.cancelled = True
|
|
raise RemoteError(
|
|
f"unreadable reply from the worker: {exc}"
|
|
) from exc
|
|
if message.get("event"):
|
|
if on_event is not None:
|
|
try:
|
|
on_event(message)
|
|
except Exception:
|
|
logger.exception("Could not record a worker event")
|
|
continue
|
|
return message
|
|
if worker.cancelled:
|
|
raise NodeCancelled("cancelled while it was running")
|
|
if line is None:
|
|
worker.kill()
|
|
raise NodeTimeout(f"was silent for {timeout}s and was killed")
|
|
raise RemoteError("worker died")
|
|
finally:
|
|
# Under the lock, so a cancel that has already read this worker out
|
|
# of _running cannot kill it after another node has taken the slot.
|
|
with self._lock:
|
|
if node_id:
|
|
self._running.pop(key, None)
|
|
self._release(worker)
|
|
|
|
def compile(self, flow: str, node: str, source: str) -> str | None:
|
|
"""Load this source in a worker. Returns what to show the author, or None."""
|
|
try:
|
|
response = self._request(
|
|
{"op": "compile", "flow": flow, "node": node, "source": source},
|
|
timeout=COMPILE_TIMEOUT,
|
|
)
|
|
except RemoteError as exc:
|
|
return f"{type(exc).__name__}: {exc}"
|
|
if response.get("ok"):
|
|
return None
|
|
error = response.get("error") or {}
|
|
return str(error.get("short") or "The node could not be loaded.")
|
|
|
|
def run(
|
|
self,
|
|
flow: str,
|
|
node: str,
|
|
source: str,
|
|
kwargs: dict[str, Any],
|
|
node_id: str,
|
|
timeout: float,
|
|
run_id: str = "",
|
|
on_event: Callable[[dict[str, Any]], None] | None = None,
|
|
) -> Any:
|
|
response = self._request(
|
|
{
|
|
"op": "run",
|
|
"call_id": f"{run_id}:{node_id}" if run_id else node_id,
|
|
"flow": flow,
|
|
"node": node,
|
|
"source": source,
|
|
"kwargs": kwargs,
|
|
"run": {"id": run_id} if run_id else None,
|
|
},
|
|
timeout=timeout,
|
|
node_id=node_id,
|
|
run_id=run_id,
|
|
on_event=on_event,
|
|
)
|
|
# Into the tee, from the thread the engine is capturing on: this is
|
|
# what puts a node's prints in the log panel, so it has to happen
|
|
# before the value comes back or the error goes up.
|
|
logs = response.get("logs")
|
|
if logs:
|
|
sys.stdout.write(logs)
|
|
if response.get("ok"):
|
|
return response.get("result")
|
|
error = response.get("error") or {}
|
|
raise _remote_class(str(error.get("type") or "RemoteError"))(
|
|
str(error.get("message") or "the node failed"),
|
|
str(error.get("traceback") or ""),
|
|
)
|
|
|
|
def proxy(
|
|
self,
|
|
flow: str,
|
|
node: str,
|
|
source: str,
|
|
node_id: str,
|
|
timeout: float,
|
|
run_id: str = "",
|
|
on_event: Callable[[dict[str, Any]], None] | None = None,
|
|
) -> Callable[..., Any]:
|
|
"""The callable a python node runs instead of its own compiled function.
|
|
|
|
A run builds its own nodes, so the run this proxy belongs to is bound
|
|
here rather than looked up — which is also what lets two runs of one
|
|
node be told apart when one of them is cancelled.
|
|
"""
|
|
|
|
def call(**kwargs: Any) -> Any:
|
|
return self.run(
|
|
flow,
|
|
node,
|
|
source,
|
|
kwargs,
|
|
node_id,
|
|
timeout,
|
|
run_id=run_id,
|
|
on_event=on_event,
|
|
)
|
|
|
|
return call
|
|
|
|
def cancel(self, node_id: str, run_id: str = "") -> bool:
|
|
"""Stop a node that is running now. False when there was nothing to stop."""
|
|
with self._lock:
|
|
worker = self._running.get((run_id, node_id))
|
|
if worker is None:
|
|
return False
|
|
worker.cancelled = True
|
|
worker.kill()
|
|
return True
|
|
|
|
def cancel_run(self, run_id: str) -> int:
|
|
"""Stop every node this run has in a worker right now."""
|
|
with self._lock:
|
|
workers = [
|
|
worker
|
|
for (owner, _node), worker in self._running.items()
|
|
if owner == run_id
|
|
]
|
|
for worker in workers:
|
|
worker.cancelled = True
|
|
worker.kill()
|
|
return len(workers)
|
|
|
|
def _publish(self, event: dict[str, Any]) -> None:
|
|
if self.events is not None:
|
|
self.events.publish(event)
|