Name the sizes a node can ask for

Raw cpus and gpus are a property of the machines an installation has, so a node
written against a cluster quietly stops meaning anything when the cluster is
replaced. A node says "gpu-small" instead, and what that is stored here —
editable, and read again every time the node is built, so changing the flavor
changes what the next run gets.

Memory joins the schema properly (`ram`, in MB, accepting "2G"), along with
`duration_s` for how long a node is expected to take. That one is recorded and
shown and nothing else yet: a statement for whoever is planning around the node,
not a limit — the limit is still `timeout`.

A flavor and a number for the same thing is refused, compared by value so an
editor writing the whole object back with its defaults still round-trips. A name
nothing stores is refused at the save, which covers the canvas and `fluksio
sync` at once, and deleting one a node still asks for says which node.

Four sizes are seeded on an installation that has none, and never re-seeded:
re-adding one somebody deliberately removed is an argument nobody wins.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A6HeySA27EkGANZN95QySW
This commit is contained in:
2026-08-27 08:59:10 +02:00
co-authored by Claude Opus 5
parent 6ff56533f5
commit a82f88cf0a
15 changed files with 720 additions and 19 deletions
@@ -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")
+2
View File
@@ -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)
+128
View File
@@ -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
+28 -1
View File
@@ -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
+33 -1
View File
@@ -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)
+23 -7
View File
@@ -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,
+43
View File
@@ -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.
+68 -5
View File
@@ -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.
+52
View File
@@ -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
+11 -3
View File
@@ -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(
+64 -1
View File
@@ -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"
)
+10
View File
@@ -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]:
+3 -1
View File
@@ -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:
+180
View File
@@ -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
+33
View File
@@ -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