diff --git a/backend/fluksio/alembic/versions/c7e2b9f34a15_flavor_table.py b/backend/fluksio/alembic/versions/c7e2b9f34a15_flavor_table.py new file mode 100644 index 0000000..2f9424b --- /dev/null +++ b/backend/fluksio/alembic/versions/c7e2b9f34a15_flavor_table.py @@ -0,0 +1,42 @@ +"""flavor + +A named amount of machine — cores, cards and memory — so a node can ask for one +by name. Seeded with a few sizes on first start, and editable afterwards: what +"gpu-small" means is a property of the machines an installation has, and those +change. + +Revision ID: c7e2b9f34a15 +Revises: b3f1a7c50d92 +Create Date: 2026-08-27 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "c7e2b9f34a15" +down_revision = "b3f1a7c50d92" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "flavor", + sa.Column("name", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), + sa.Column("cpus", sa.Integer(), nullable=False), + sa.Column("gpus", sa.Integer(), nullable=False), + sa.Column("ram", sa.Integer(), nullable=False), + sa.Column( + "description", + sqlmodel.sql.sqltypes.AutoString(length=255), + nullable=False, + ), + sa.PrimaryKeyConstraint("name"), + ) + + +def downgrade(): + op.drop_table("flavor") diff --git a/backend/fluksio/api/main.py b/backend/fluksio/api/main.py index 0efef1d..ba56eaa 100644 --- a/backend/fluksio/api/main.py +++ b/backend/fluksio/api/main.py @@ -5,6 +5,7 @@ from fluksio.api.routes import ( artifacts, cloud, dashboards, + flavors, flows, login, messages, @@ -24,6 +25,7 @@ api_router = APIRouter() api_router.include_router(login.router) api_router.include_router(users.router) api_router.include_router(utils.router) +api_router.include_router(flavors.router) api_router.include_router(flows.router) api_router.include_router(flows.ws_router) api_router.include_router(secrets.router) diff --git a/backend/fluksio/api/routes/flavors.py b/backend/fluksio/api/routes/flavors.py new file mode 100644 index 0000000..133ee0e --- /dev/null +++ b/backend/fluksio/api/routes/flavors.py @@ -0,0 +1,128 @@ +"""Named amounts of machine, for nodes to ask for by name. + +A node saying ``gpu-small`` keeps meaning something after the cluster it was +written against is replaced; one saying eight cores and sixteen gigabytes +quietly stops. So the sizes live here, editable, and a node refers to one. + +Reading them is open to anyone signed in — the node panel offers them in a +dropdown. Changing one is a superuser's, because it changes what every node +that names it gets. +""" + +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException +from fastapi.concurrency import run_in_threadpool +from sqlmodel import select + +from fluksio.api.deps import ( + FlowControllerDep, + SessionDep, + get_current_active_superuser, + get_current_user, +) +from fluksio.flow.resources import FLAVOR_NAME +from fluksio.models import ( + Flavor, + FlavorCreate, + FlavorPublic, + FlavorsPublic, + FlavorUpdate, + Message, +) + +router = APIRouter( + prefix="/flavors", tags=["flavors"], dependencies=[Depends(get_current_user)] +) + + +def _get(session: SessionDep, name: str) -> Flavor: + flavor = session.get(Flavor, name) + if flavor is None: + raise HTTPException(status_code=404, detail=f"No flavor named '{name}'") + return flavor + + +@router.get("/", response_model=FlavorsPublic) +def read_flavors(session: SessionDep) -> Any: + """Every size a node can ask for.""" + flavors = list(session.exec(select(Flavor).order_by(Flavor.name)).all()) + return FlavorsPublic(data=flavors, count=len(flavors)) + + +@router.post( + "/", + response_model=FlavorPublic, + dependencies=[Depends(get_current_active_superuser)], +) +def create_flavor(session: SessionDep, body: FlavorCreate) -> Any: + if not FLAVOR_NAME.match(body.name): + raise HTTPException( + status_code=422, + detail="A flavor name is lowercase letters, digits, dashes and " + "underscores, starting with a letter", + ) + if session.get(Flavor, body.name) is not None: + raise HTTPException( + status_code=409, detail=f"A flavor named '{body.name}' already exists" + ) + flavor = Flavor(**body.model_dump()) + session.add(flavor) + session.commit() + session.refresh(flavor) + return flavor + + +@router.patch( + "/{name}", + response_model=FlavorPublic, + dependencies=[Depends(get_current_active_superuser)], +) +def update_flavor(session: SessionDep, name: str, body: FlavorUpdate) -> Any: + """Change what a size means. Every node naming it gets the new one.""" + flavor = _get(session, name) + flavor.sqlmodel_update(body.model_dump(exclude_unset=True)) + session.add(flavor) + session.commit() + session.refresh(flavor) + return flavor + + +@router.delete( + "/{name}", + response_model=Message, + dependencies=[Depends(get_current_active_superuser)], +) +async def delete_flavor( + session: SessionDep, controller: FlowControllerDep, name: str +) -> Any: + """Remove a size, as long as no node still asks for it.""" + _get(session, name) + used_by = await run_in_threadpool(_usages, controller, name) + if used_by: + raise HTTPException( + status_code=409, + detail=f"'{name}' is still asked for by {', '.join(used_by)}", + ) + session.delete(session.get(Flavor, name)) + session.commit() + return Message(message=f"Deleted flavor '{name}'") + + +def _usages(controller: FlowControllerDep, name: str) -> list[str]: + """Which nodes name this flavor, published or in a draft.""" + used: list[str] = [] + for flow_name in controller.store.list_flows(): + seen = set() + for draft in (False, True): + if draft and not controller.store.has_draft(flow_name): + continue + try: + flow = controller.store.read_flow(flow_name, draft=draft) + except Exception: + continue + for node in flow.nodes: + if node.resources is not None and node.resources.flavor == name: + seen.add(f"{flow_name}.{node.id}") + used.extend(sorted(seen)) + return used diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 3d3be6e..5e440e5 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -57,7 +57,7 @@ from fluksio.flow.store import ( LibNotFound, StaleVersion, ) -from fluksio.models import Message, Run, RunArtifact, RunMetric, RunNode +from fluksio.models import Flavor, Message, Run, RunArtifact, RunMetric, RunNode router = APIRouter( prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)] @@ -368,6 +368,31 @@ def read_flow(name: str, controller: FlowControllerDep) -> Any: return _detail(controller, _read_flow(controller, name)) +def _check_flavors(definition: FlowDef) -> None: + """Refuse a size nobody stored, here rather than when the node is built. + + Catching it at the save covers the canvas and ``fluksio sync`` in one + place; a node that only finds out at build time is a red node somebody has + to go and look at. + """ + named = { + node.resources.flavor + for node in definition.nodes + if node.resources is not None and node.resources.flavor + } + if not named: + return + with Session(engine) as session: + known = set(session.exec(select(Flavor.name)).all()) + unknown = sorted(named - known) + if unknown: + raise HTTPException( + status_code=422, + detail=f"No flavor named '{unknown[0]}' " + f"(known: {', '.join(sorted(known)) or 'none'})", + ) + + @router.put("/{name}", response_model=FlowDetail) async def save_flow( name: str, @@ -390,6 +415,8 @@ async def save_flow( if len(duplicates) != len(definition.nodes): raise HTTPException(status_code=400, detail="Node names must be unique") + _check_flavors(definition) + try: stored = await run_in_threadpool( controller.store.write_draft, definition, definition.version diff --git a/backend/fluksio/core/db.py b/backend/fluksio/core/db.py index 9a0769c..0caa694 100644 --- a/backend/fluksio/core/db.py +++ b/backend/fluksio/core/db.py @@ -20,7 +20,7 @@ from sqlmodel import Session, SQLModel, create_engine, select from fluksio import crud from fluksio.core.config import settings -from fluksio.models import User, UserCreate +from fluksio.models import Flavor, User, UserCreate logger = logging.getLogger(__name__) @@ -110,8 +110,40 @@ def init_db(session: Session) -> None: user = crud.create_user(session=session, user_create=user_in) +#: The sizes an installation starts with. Written once, when there are none, +#: and editable from there — what a name means is a property of the machines +#: this installation has, and nothing here knows what those are. +SEED_FLAVORS: tuple[dict[str, Any], ...] = ( + {"name": "small", "cpus": 1, "ram": 2048, "description": "A poll, a threshold"}, + {"name": "medium", "cpus": 4, "ram": 8192, "description": "A step that computes"}, + {"name": "large", "cpus": 8, "ram": 16384, "description": "A heavy step"}, + { + "name": "gpu-small", + "cpus": 4, + "gpus": 1, + "ram": 16384, + "description": "One card, and cores to feed it", + }, +) + + +def seed_flavors(session: Session) -> None: + """Give a new installation sizes to pick from, once. + + Only when there are none at all: they are editable, and re-adding one that + somebody deliberately removed would be an argument nobody can win. + """ + if session.exec(select(Flavor)).first() is not None: + return + for row in SEED_FLAVORS: + session.add(Flavor(**row)) + session.commit() + logger.info("seeded %d resource flavors", len(SEED_FLAVORS)) + + def prepare(engine: Engine) -> None: """Everything that has to be true before the app serves a request.""" migrate(engine) with Session(engine) as session: init_db(session) + seed_flavors(session) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 22f2bc1..73d897a 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -64,7 +64,12 @@ from fluksio.flow.pipeline import ( ) from fluksio.flow.placement import Placer from fluksio.flow.remote import RemoteWorkerHub -from fluksio.flow.resources import ResourceAccountant, derive_env +from fluksio.flow.resources import ( + ResourceAccountant, + UnknownFlavor, + derive_env, + resolve_flavor, +) from fluksio.flow.schemas import ( BrainEdge, BrainGraph, @@ -869,11 +874,10 @@ class FlowController: # Building # ------------------------------------------------------------------------- - def _runs_elsewhere(self, node_def: NodeDef) -> bool: + def _runs_elsewhere(self, wanted: Resources | None) -> bool: """Whether this node asks for more than this machine could ever give.""" - if node_def.resources is None or self.placer is None: + if wanted is None or self.placer is None: return False - wanted = node_def.resources return not self.placer.local.fits(wanted.cpus, wanted.gpus, wanted.ram or 0) def _placed( @@ -1016,6 +1020,18 @@ class FlowController: # one importing torch is correct on the GPU box and a # missing module here, so checking it here would fail a # node that is fine. + try: + # A named size is read now rather than stored on the + # node, so editing the flavor edits the next run. + wanted = ( + resolve_flavor(node_def.resources) + if node_def.resources is not None + else None + ) + except UnknownFlavor as exc: + entry.status = NodeStatus.ERROR + entry.error = str(exc) + return entry remote_only = ( node_def.device and node_def.device_policy == "require" @@ -1025,7 +1041,7 @@ class FlowController: problem = self.remote.compile( node_def.device or "", owner, local, code ) - elif self._runs_elsewhere(node_def): + elif self._runs_elsewhere(wanted): # Asks for more than this machine has, so it will run on # one that has it. Same reason as a device: checking the # import here would fail a node that is fine there. @@ -1043,13 +1059,13 @@ class FlowController: if node_def.timeout is not None else settings.FLOW_NODE_TIMEOUT ) - if node_def.resources is not None and self.placer is not None: + if wanted is not None and self.placer is not None: # Says how much of a machine it takes, so which machine # and how much of it are one decision — including when # it also names a device, which used to mean the two # answers disagreed and nothing was accounted at all. function = self._placed( - node_def.resources, + wanted, node_def.device, node_def.device_policy, owner, diff --git a/backend/fluksio/flow/resources.py b/backend/fluksio/flow/resources.py index 3c03b07..3688f54 100644 --- a/backend/fluksio/flow/resources.py +++ b/backend/fluksio/flow/resources.py @@ -26,14 +26,21 @@ from __future__ import annotations import logging import os +import re import threading from collections.abc import Callable from dataclasses import dataclass +from sqlmodel import Session + from fluksio.flow.schemas import Resources logger = logging.getLogger(__name__) +#: What a flavor may be called. Dashes allowed, unlike a flow or node name — +#: "gpu-small" reads better than "gpu_small" on a dropdown. +FLAVOR_NAME = re.compile(r"^[a-z][a-z0-9_-]*$") + #: Every spelling of "how many threads may you use" that a scientific stack #: reads out of the environment at import. Set together, because a process #: usually pulls in more than one of them. @@ -175,6 +182,42 @@ def derive_env(wanted: Resources, allocation: Allocation) -> dict[str, str]: return env +class UnknownFlavor(ValueError): + """A node asks for a size that is not stored here.""" + + +def resolve_flavor(wanted: Resources) -> Resources: + """The concrete numbers behind a declaration. + + Read when the node is built rather than stored on it, so editing a flavor + changes what the next run gets. A node that names one that has been deleted + is an error rather than a default: running a training step against a size + nobody chose is worse than a node that says what is wrong with it. + """ + if not wanted.flavor: + return wanted + # Imported here: this module is the books, and the books have no business + # knowing about the database until somebody asks for a stored size. + from fluksio.core.db import engine + from fluksio.models import Flavor + + with Session(engine) as session: + row = session.get(Flavor, wanted.flavor) + if row is None: + raise UnknownFlavor( + f"flavor '{wanted.flavor}' does not exist — `fluksio flavors` lists them" + ) + # Built fresh rather than copied, so the flavor-and-numbers check runs on + # the result and this cannot quietly produce something invalid. + return Resources( + cpus=row.cpus, + gpus=row.gpus, + ram=row.ram, + env=wanted.env, + duration_s=wanted.duration_s, + ) + + def fair_share_env(cpus: int, workers: int) -> dict[str, str]: """Thread caps for the shared pool, where nothing declared anything. diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index 0a0f702..8a15757 100644 --- a/backend/fluksio/flow/schemas.py +++ b/backend/fluksio/flow/schemas.py @@ -9,12 +9,18 @@ from __future__ import annotations import re from typing import Any, Literal -from pydantic import BaseModel, ConfigDict, Field, field_validator +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from fluksio.flow.messages import MessageSpec NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]*$") +#: Megabytes, gigabytes, or a bare number already in megabytes. +_SIZE = re.compile(r"^\s*(\d+)\s*([mg]b?)?\s*$", re.IGNORECASE) +#: Seconds, minutes, hours or days — one unit, so there is one thing to read. +_SPAN = re.compile(r"^\s*(\d+)\s*([smhd])?\s*$", re.IGNORECASE) +_SPAN_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400} + def _validate_name(value: str) -> str: if not NAME_PATTERN.match(value): @@ -24,6 +30,27 @@ def _validate_name(value: str) -> str: return value +def _megabytes(value: Any) -> Any: + """``2G`` and ``512M`` and ``512`` all mean a number of megabytes.""" + if not isinstance(value, str): + return value + match = _SIZE.match(value) + if not match: + raise ValueError(f"'{value}' is not a size — write 512M, 2G, or 512") + amount, unit = int(match.group(1)), (match.group(2) or "m").lower() + return amount * 1024 if unit.startswith("g") else amount + + +def _seconds(value: Any) -> Any: + """``2h``, ``30m``, ``90s`` and ``90`` all mean a number of seconds.""" + if not isinstance(value, str): + return value + match = _SPAN.match(value) + if not match: + raise ValueError(f"'{value}' is not a duration — write 90s, 30m, 2h, or 90") + return int(match.group(1)) * _SPAN_SECONDS[(match.group(2) or "s").lower()] + + class Resources(BaseModel): """What one execution of a node needs to have to itself. @@ -64,10 +91,27 @@ class Resources(BaseModel): default=None, ge=1, description=( - "Megabytes held for the whole execution. Counted against machines " - "that said how much they have, and ignored by those that did not — " - "which is a machine with nothing to say about memory, not one with " - "none." + "Megabytes held for the whole execution; accepts '512M' or '2G'. " + "Counted against machines that said how much they have, and " + "ignored by those that did not — which is a machine with nothing " + "to say about memory, not one with none." + ), + ) + flavor: str | None = Field( + default=None, + description=( + "A stored size by name, standing in for cpus, gpus and ram. Read " + "again every time the node is built, so editing the flavor edits " + "what the next run gets." + ), + ) + duration_s: int | None = Field( + default=None, + ge=1, + description=( + "How long one execution is expected to take; accepts '30m' or " + "'2h'. A statement about the node for whoever is planning around " + "it, not a limit — the limit is `timeout`." ), ) env: dict[str, str] = Field( @@ -80,6 +124,25 @@ class Resources(BaseModel): ), ) + _parse_ram = field_validator("ram", mode="before")(_megabytes) + _parse_duration = field_validator("duration_s", mode="before")(_seconds) + + @model_validator(mode="after") + def _a_flavor_says_it_all(self) -> Resources: + """A flavor and a number for the same thing is two answers. + + Compared by value rather than by what was set, because an editor that + writes the whole object back sends the defaults with it — and a node + that says `flavor` and `cpus: 1` has not actually asked for anything + the flavor does not already cover. + """ + if self.flavor and (self.cpus != 1 or self.gpus != 0 or self.ram is not None): + raise ValueError( + f"flavor '{self.flavor}' already says how much — " + "drop cpus, gpus and ram, or drop the flavor" + ) + return self + class NodeDef(BaseModel): """A node as stored: identity, configuration and ports. diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 21eb296..82c9aae 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -462,3 +462,55 @@ class RunArtifact(SQLModel, table=True): default_factory=lambda: datetime.now(UTC), sa_type=UTCDateTime, ) + + +# ----------------------------------------------------------------------------- +# Flavors +# ----------------------------------------------------------------------------- + + +class FlavorBase(SQLModel): + """A named amount of machine, so a node can ask for one by name. + + The point is not to save typing. Cores and megabytes are a property of the + machines an installation actually has, and they change when those machines + do — so a node saying "gpu-small" keeps meaning something after the cluster + is replaced, where a node saying 8 and 16384 quietly stops. + """ + + cpus: int = Field(default=1, ge=1) + gpus: int = Field(default=0, ge=0) + #: Megabytes. A flavor always says, which is what a node asking for one is + #: buying — the raw form leaves memory unstated and unaccounted. + ram: int = Field(default=2048, ge=1) + description: str = Field(default="", max_length=255) + + +class Flavor(FlavorBase, table=True): + """A resource size somebody named, stored so nodes can refer to it.""" + + __tablename__ = "flavor" + + name: str = Field(primary_key=True, max_length=64) + + +class FlavorCreate(FlavorBase): + name: str = Field(max_length=64) + + +class FlavorUpdate(SQLModel): + """Everything but the name: renaming would orphan the nodes that ask.""" + + cpus: int | None = Field(default=None, ge=1) + gpus: int | None = Field(default=None, ge=0) + ram: int | None = Field(default=None, ge=1) + description: str | None = Field(default=None, max_length=255) + + +class FlavorPublic(FlavorBase): + name: str + + +class FlavorsPublic(SQLModel): + data: list[FlavorPublic] + count: int diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index 951cb56..facf75c 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -268,7 +268,7 @@ def node( device: str | None = None, device_policy: str = "require", cache: bool = True, - resources: dict[str, Any] | None = None, + resources: dict[str, Any] | str | None = None, ) -> Callable[[F], F]: """Mark a function as a node, declaring its ports. @@ -302,14 +302,22 @@ def node( library that sizes itself to every visible core is told otherwise. ``env`` is for the tuning the engine must not invent, such as ``XLA_FLAGS``. Declaring nothing is the default and changes nothing. + + A string is a stored flavor — ``resources="gpu-small"`` — which is the same + thing as ``{"flavor": "gpu-small"}`` and stands in for cpus, gpus and ram. + Sizes accept ``"2G"``; ``duration_s`` accepts ``"30m"``. """ if device_policy not in ("require", "prefer"): raise SyncError("device_policy is 'require' or 'prefer'") + if isinstance(resources, str): + resources = {"flavor": resources} if resources is not None: - unknown = sorted(set(resources) - {"cpus", "gpus", "env"}) + known = {"cpus", "gpus", "ram", "flavor", "duration_s", "env"} + unknown = sorted(set(resources) - known) if unknown: raise SyncError( - f"resources={{'{unknown[0]}': ...}} is not one of cpus, gpus, env" + f"resources={{'{unknown[0]}': ...}} is not one of " + f"{', '.join(sorted(known))}" ) if timeout is not None and timeout < 0: raise SyncError( diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 4a00537..5901b71 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -558,6 +558,35 @@ def _portal_phrase(portal: dict[str, Any]) -> tuple[str, str]: return "portal unreachable" + (f" ({trouble[:60]})" if trouble else ""), "red" +def _resource_lines(machines: dict[str, Any]) -> list[str]: + """One line per machine: what it holds, and what is queued for it.""" + targets = machines.get("targets") or [] + if not targets: + return [] + waiting = len(machines.get("waiting") or []) + lines = [] + for target in targets: + parts = [] + for label, level in ( + ("cpu", target.get("cpus")), + ("gpu", target.get("gpus")), + ): + # A machine with no card says nothing about cards. + if level and level.get("total"): + parts.append( + f"{label} {level['total'] - level['free']}/{level['total']}" + ) + ram = target.get("ram_mb") + if ram and ram.get("total"): + parts.append( + f"ram {(ram['total'] - ram['free']) // 1024}/{ram['total'] // 1024}G" + ) + lines.append(f"{str(target.get('target', '')):<16}" + " · ".join(parts)) + if waiting: + lines[-1] += f" {waiting} waiting" + return [f"resources {lines[0]}"] + [f" {line}" for line in lines[1:]] + + def _status_screen(client: Client) -> Any: """One frame: health, the flows, and the failures under them.""" from rich.console import Group @@ -576,6 +605,11 @@ def _status_screen(client: Client) -> Any: # Never enrolled, or an engine too old to answer. Neither is worth # failing a status screen over. portal = {} + try: + machines = client.resources() + except (SyncError, ApiError): + # An engine that accounts nothing, or one too old to answer. + machines = {} healthy = summary.get("status") == "ok" head = Text() @@ -625,7 +659,9 @@ def _status_screen(client: Client) -> Any: Text(" · ".join(notes), style="red" if flow.get("error_count") else "dim"), ) - parts: list[Any] = [head, facts, ""] + parts: list[Any] = [head, facts] + parts += [Text(line, style="dim") for line in _resource_lines(machines)] + parts.append("") parts.append( table if flows else Text("No flows yet. `fluksio sync` uploads yours.", "dim") ) @@ -727,6 +763,27 @@ def cmd_runs(args: argparse.Namespace) -> int: return 0 +def cmd_flavors(args: argparse.Namespace) -> int: + """The named sizes a node can ask for.""" + try: + with _client_for(args) as client: + rows = client.flavors() + except (SyncError, ApiError) as exc: + return _fail(str(exc)) + except httpx.HTTPError as exc: + return _unreachable(exc) + if not rows: + _say("No flavors. A node can still say cpus and gpus itself.") + return 0 + for row in rows: + cards = f"{row['gpus']:>3} gpu" if row.get("gpus") else " " * 7 + _say( + f"{row['name']:<14}{row['cpus']:>3} cpu {row['ram'] // 1024:>5} GB" + f"{cards} {row.get('description', '')}" + ) + return 0 + + def _grid( definition: dict[str, Any], values: list[str], seed: int | None ) -> list[dict[str, Any]]: @@ -909,6 +966,12 @@ def add_parsers(subparsers: Any) -> None: with_engine(parser, local=True) parser.set_defaults(func=cmd_runs) + parser = subparsers.add_parser( + "flavors", help="the named resource sizes a node can ask for" + ) + with_engine(parser) + parser.set_defaults(func=cmd_flavors) + parser = subparsers.add_parser( "sweep", help="one flow, once per combination of the parameters given" ) diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 138097c..02b52e2 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -272,6 +272,16 @@ class Client: result = self._call("GET", "/flows/") return list(result.get("data") or []) + def flavors(self) -> list[dict[str, Any]]: + """The named sizes a node can ask for.""" + result = self._call("GET", "/flavors/") + return list(result.get("data") or []) + + def resources(self) -> dict[str, Any]: + """Every machine, what is free of it, and which nodes are queued.""" + result: dict[str, Any] = self._call("GET", "/workers/resources") + return result + # -- how the engine is doing ------------------------------------------- def summary(self) -> dict[str, Any]: diff --git a/backend/scripts/bench_startup.py b/backend/scripts/bench_startup.py index 5b91065..8b9ce0e 100644 --- a/backend/scripts/bench_startup.py +++ b/backend/scripts/bench_startup.py @@ -37,7 +37,9 @@ log = logging.getLogger("bench") #: A flow that computes nothing, so what is timed is the getting-started. BENCH_FLOW = "bench_startup" -NODE_SOURCE = '"""Does nothing, on purpose."""\n\n\ndef process(n):\n return {"out": n}\n' +NODE_SOURCE = ( + '"""Does nothing, on purpose."""\n\n\ndef process(n):\n return {"out": n}\n' +) class Api: diff --git a/backend/tests/api/routes/test_flavors.py b/backend/tests/api/routes/test_flavors.py new file mode 100644 index 0000000..01bbfe1 --- /dev/null +++ b/backend/tests/api/routes/test_flavors.py @@ -0,0 +1,180 @@ +"""Named sizes: storing them, and what happens to a node that asks for one.""" + +import pytest +from fastapi.testclient import TestClient +from sqlmodel import Session, select + +from fluksio.core.config import settings +from fluksio.core.db import seed_flavors +from fluksio.models import Flavor + + +def test_an_installation_starts_with_sizes_to_pick_from( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + response = client.get( + f"{settings.API_V1_STR}/flavors/", headers=superuser_token_headers + ) + assert response.status_code == 200 + names = {row["name"] for row in response.json()["data"]} + assert {"small", "medium", "large", "gpu-small"} <= names + + +def test_seeding_leaves_an_installation_that_has_its_own_alone(db: Session) -> None: + """Re-adding a size somebody deliberately removed is an argument to avoid.""" + before = {row.name for row in db.exec(select(Flavor)).all()} + seed_flavors(db) + assert {row.name for row in db.exec(select(Flavor)).all()} == before + + +def test_a_size_can_be_added_changed_and_removed( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + created = client.post( + f"{settings.API_V1_STR}/flavors/", + headers=superuser_token_headers, + json={"name": "gpu-big", "cpus": 16, "gpus": 4, "ram": 65536}, + ) + assert created.status_code == 200 + assert created.json()["gpus"] == 4 + + again = client.post( + f"{settings.API_V1_STR}/flavors/", + headers=superuser_token_headers, + json={"name": "gpu-big", "cpus": 1}, + ) + assert again.status_code == 409 + + changed = client.patch( + f"{settings.API_V1_STR}/flavors/gpu-big", + headers=superuser_token_headers, + json={"cpus": 32}, + ) + assert changed.status_code == 200 + # Only what was sent: the rest of the size is untouched. + assert (changed.json()["cpus"], changed.json()["gpus"]) == (32, 4) + + removed = client.delete( + f"{settings.API_V1_STR}/flavors/gpu-big", headers=superuser_token_headers + ) + assert removed.status_code == 200 + listed = client.get( + f"{settings.API_V1_STR}/flavors/", headers=superuser_token_headers + ).json() + assert "gpu-big" not in {row["name"] for row in listed["data"]} + + +def test_changing_a_size_is_a_superusers( + client: TestClient, normal_user_token_headers: dict[str, str] +) -> None: + """Reading is open — the node panel offers them — and writing is not.""" + assert ( + client.get( + f"{settings.API_V1_STR}/flavors/", headers=normal_user_token_headers + ).status_code + == 200 + ) + refused = client.post( + f"{settings.API_V1_STR}/flavors/", + headers=normal_user_token_headers, + json={"name": "sneaky", "cpus": 99}, + ) + assert refused.status_code == 403 + + +def test_a_name_nobody_could_type_is_refused( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + refused = client.post( + f"{settings.API_V1_STR}/flavors/", + headers=superuser_token_headers, + json={"name": "GPU Big!", "cpus": 1}, + ) + assert refused.status_code == 422 + + +def test_a_flow_cannot_name_a_size_that_does_not_exist( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """Caught at the save, which covers the canvas and `fluksio sync` at once.""" + response = client.put( + f"{settings.API_V1_STR}/flows/flavor_check", + headers=superuser_token_headers, + json={ + "name": "flavor_check", + "version": 1, + "nodes": [ + {"id": "fit", "type": "python", "resources": {"flavor": "imaginary"}} + ], + }, + ) + assert response.status_code == 422 + assert "imaginary" in str(response.json()["detail"]) + + +def test_a_size_a_node_still_asks_for_is_not_deleted( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + client.post( + f"{settings.API_V1_STR}/flavors/", + headers=superuser_token_headers, + json={"name": "in-use", "cpus": 2}, + ) + client.put( + f"{settings.API_V1_STR}/flows/flavor_user", + headers=superuser_token_headers, + json={ + "name": "flavor_user", + "version": 1, + "nodes": [ + {"id": "fit", "type": "python", "resources": {"flavor": "in-use"}} + ], + }, + ) + try: + refused = client.delete( + f"{settings.API_V1_STR}/flavors/in-use", headers=superuser_token_headers + ) + assert refused.status_code == 409 + assert "flavor_user.fit" in refused.json()["detail"] + finally: + client.delete( + f"{settings.API_V1_STR}/flows/flavor_user", headers=superuser_token_headers + ) + client.delete( + f"{settings.API_V1_STR}/flavors/in-use", headers=superuser_token_headers + ) + + +def test_a_named_size_resolves_to_the_numbers_stored_for_it(db: Session) -> None: + from fluksio.flow.resources import UnknownFlavor, resolve_flavor + from fluksio.flow.schemas import Resources + + resolved = resolve_flavor(Resources(flavor="gpu-small", duration_s=60)) + seeded = db.get(Flavor, "gpu-small") + assert (resolved.cpus, resolved.gpus, resolved.ram) == ( + seeded.cpus, + seeded.gpus, + seeded.ram, + ) + # What the node said about itself survives; the flavor only sizes it. + assert resolved.duration_s == 60 + assert resolved.flavor is None + + # Read at build time, so editing a flavor edits what the next run gets. + seeded.cpus = 12 + db.add(seeded) + db.commit() + assert resolve_flavor(Resources(flavor="gpu-small")).cpus == 12 + + with pytest.raises(UnknownFlavor, match="does not exist"): + resolve_flavor(Resources(flavor="never-stored")) + + +def test_declaring_nothing_never_looks_anything_up() -> None: + """The common node pays nothing for a feature it does not use.""" + from fluksio.flow.resources import resolve_flavor + from fluksio.flow.schemas import Resources + + wanted = Resources(cpus=2) + assert resolve_flavor(wanted) is wanted diff --git a/backend/tests/flow/test_resources.py b/backend/tests/flow/test_resources.py index 7a043f6..b0e347b 100644 --- a/backend/tests/flow/test_resources.py +++ b/backend/tests/flow/test_resources.py @@ -150,3 +150,36 @@ def test_declaring_nothing_stays_exactly_as_it_was(): def test_a_misspelled_resource_is_refused(): with pytest.raises(ValueError): Resources(cpu=4) + + +# ----------------------------------------------------------------------------- +# Sizes with names, and sizes written out +# ----------------------------------------------------------------------------- + + +def test_a_size_can_be_written_the_way_people_write_sizes(): + assert Resources(ram="2G").ram == 2048 + assert Resources(ram="512M").ram == 512 + assert Resources(ram="512").ram == 512 + assert Resources(ram=512).ram == 512 + with pytest.raises(ValueError, match="not a size"): + Resources(ram="a lot") + + +def test_a_duration_is_one_unit_and_says_so_when_it_is_not(): + assert Resources(duration_s="2h").duration_s == 7200 + assert Resources(duration_s="30m").duration_s == 1800 + assert Resources(duration_s="90").duration_s == 90 + with pytest.raises(ValueError, match="not a duration"): + Resources(duration_s="1h30m") + + +def test_a_flavor_and_a_number_for_the_same_thing_is_two_answers(): + with pytest.raises(ValueError, match="already says how much"): + Resources(flavor="gpu-small", cpus=4) + + # The defaults are not an answer: an editor writing the whole object back + # sends them, and that must survive the round trip. + assert Resources(flavor="gpu-small", cpus=1, gpus=0).flavor == "gpu-small" + # Neither is a duration, which a flavor says nothing about. + assert Resources(flavor="gpu-small", duration_s="2h").duration_s == 7200