From 37a7df9d24d6265953644325ac01e03309534803 Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 27 Aug 2026 14:17:51 +0200 Subject: [PATCH] Let a sweep run more than four at a time, and name the run a failure was in MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Concurrent runs sat at 4 whatever FLOW_MAX_CASCADES said: that setting bounds cascades, and the run drivers read a hardcoded MAX_PARALLEL nobody could reach. FLOW_MAX_RUNS is the knob they read now, --max-runs/--max-cascades/--max-workers are the same three as flags on serve, and the engine says which numbers it started with — which is the only way to tell that a settings file was read. Events keep the run they happened in. The payload always carried it and the persist path dropped it, so reading one run's failures meant filtering the engine-wide list; a batch run's id reaches those events now too, since a run has no journaled item to name itself by. Also: a provisioner's 0 means "no deadline" rather than "cancel on the next reconcile", and a command that reaches no engine says how to start one. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_015sbYeYaVgYQqm1sbx7wPdL --- .../versions/d4a71e9c2b58_engine_event_run.py | 40 +++++++++++++++++ backend/fluksio/api/routes/observability.py | 7 +++ backend/fluksio/cli.py | 35 +++++++++++++++ backend/fluksio/core/config.py | 4 ++ backend/fluksio/flow/controller.py | 1 + backend/fluksio/flow/metrics.py | 9 +++- backend/fluksio/flow/pipeline.py | 13 ++++-- backend/fluksio/flow/provision.py | 10 ++++- backend/fluksio/flow/runs.py | 16 +++++-- backend/fluksio/main.py | 9 ++++ backend/fluksio/models.py | 3 ++ backend/fluksio/sdk/cli.py | 9 +++- backend/fluksio/sdk/client.py | 10 +++-- .../tests/api/routes/test_observability.py | 27 ++++++++++++ backend/tests/api/routes/test_runs.py | 44 +++++++++++++++++++ backend/tests/flow/test_provision.py | 21 ++++++++- backend/tests/test_metrics.py | 24 ++++++++++ docs/code/api.md | 2 +- docs/code/workers.md | 4 ++ docs/reference/configuration.md | 5 +++ frontend/src/client/schemas.gen.ts | 6 ++- frontend/src/client/sdk.gen.ts | 5 +++ frontend/src/client/types.gen.ts | 2 + 23 files changed, 288 insertions(+), 18 deletions(-) create mode 100644 backend/fluksio/alembic/versions/d4a71e9c2b58_engine_event_run.py diff --git a/backend/fluksio/alembic/versions/d4a71e9c2b58_engine_event_run.py b/backend/fluksio/alembic/versions/d4a71e9c2b58_engine_event_run.py new file mode 100644 index 0000000..e697b85 --- /dev/null +++ b/backend/fluksio/alembic/versions/d4a71e9c2b58_engine_event_run.py @@ -0,0 +1,40 @@ +"""engine_event.run + +Which run an event happened in. The payload always carried it and the persist +path dropped it, so reading one run's failures meant filtering the global list. +Rows written before this carry an empty string, the same as anything that +belongs to no run at all. + +Revision ID: d4a71e9c2b58 +Revises: c7e2b9f34a15 +Create Date: 2026-08-27 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "d4a71e9c2b58" +down_revision = "c7e2b9f34a15" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "engine_event", + sa.Column( + "run", + sqlmodel.sql.sqltypes.AutoString(length=64), + nullable=False, + server_default="", + ), + ) + op.create_index("ix_engine_event_run", "engine_event", ["run"]) + + +def downgrade(): + op.drop_index("ix_engine_event_run", table_name="engine_event") + op.drop_column("engine_event", "run") diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index 5653153..1333baf 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -108,6 +108,7 @@ class EventRow(BaseModel): node: str detail: str actor: str + run: str class DeadLetter(BaseModel): @@ -389,6 +390,7 @@ def read_events( session: SessionDep, kind: Literal["failure", "audit"] = "failure", flow: str | None = None, + run: str | None = None, since: datetime | None = None, until: datetime | None = None, limit: int = 100, @@ -397,6 +399,9 @@ def read_events( ``since`` is inclusive and ``until`` exclusive, the same window ``/runs`` takes, so a list can cover the span the charts beside it are drawn from. + ``run`` narrows to one run — a batch run's id, or the journaled item a live + cascade came from. Rows recorded before the column existed carry none, so + an old failure answers no run at all rather than the wrong one. """ statement = select(EngineEvent).order_by(col(EngineEvent.ts).desc()) if kind == "audit": @@ -405,6 +410,8 @@ def read_events( statement = statement.where(col(EngineEvent.type) != "audit") if flow: statement = statement.where(col(EngineEvent.flow) == flow) + if run: + statement = statement.where(col(EngineEvent.run) == run) if since: statement = statement.where(col(EngineEvent.ts) >= _aware(since)) if until: diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index eedd9b8..1429945 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -220,8 +220,22 @@ def _sign_in(admin_id: Any, url: str, data_dir: Path) -> Path: return write_config(url, token, data_dir) +#: The `serve` flags that are settings under another name, and the setting each +#: one writes. A flag is a real environment variable, which outranks the env +#: file the settings read — so the order is flag, environment, `/env`. +CONCURRENCY_FLAGS = { + "max_workers": "FLOW_MAX_WORKERS", + "max_cascades": "FLOW_MAX_CASCADES", + "max_runs": "FLOW_MAX_RUNS", +} + + def cmd_serve(args: argparse.Namespace) -> int: data_dir = _data_dir(args.data_dir, args.shared) + for flag, name in CONCURRENCY_FLAGS.items(): + value = getattr(args, flag, None) + if value is not None: + os.environ[name] = str(value) _prepare(data_dir) from sqlmodel import Session @@ -370,6 +384,27 @@ def _parser() -> argparse.ArgumentParser: metavar="URL", help=f"the portal --enroll redeems at (default {DEFAULT_PORTAL})", ) + serve.add_argument( + "--max-runs", + type=int, + default=None, + metavar="N", + help="batch runs driven at once (default 4, FLOW_MAX_RUNS)", + ) + serve.add_argument( + "--max-cascades", + type=int, + default=None, + metavar="N", + help="cascades in flight at once (default 4, FLOW_MAX_CASCADES)", + ) + serve.add_argument( + "--max-workers", + type=int, + default=None, + metavar="N", + help="python worker processes (default 4, FLOW_MAX_WORKERS)", + ) serve.set_defaults(func=cmd_serve) enroll = subparsers.add_parser( diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index f52b64e..1e8729b 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -102,6 +102,10 @@ class Settings(BaseSettings): # over the mean cascade time, so an installation whose nodes wait on the # network rather than on a CPU wants it higher than the core count. FLOW_MAX_CASCADES: int = 4 + # How many batch runs are driven at once. A different limit from the one + # above: a run drives a whole graph, and its nodes are bounded by the worker + # pool rather than by cascade slots. A sweep is what this governs. + FLOW_MAX_RUNS: int = 4 # How long a python node may be silent before its worker is killed, unless # the node sets its own. 0, the default, disables it: a dead worker still # fails fast, and a slow one is left to finish. Set it where silence means diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 73d897a..06850a0 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -1612,6 +1612,7 @@ class FlowController: observer=observer, emission_observer=emission_observer, run_cache=run_cache, + run_id=run.run_id if run is not None else "", ) pipeline.history_limits = self.history_limits return pipeline diff --git a/backend/fluksio/flow/metrics.py b/backend/fluksio/flow/metrics.py index 4adf66c..9de6085 100644 --- a/backend/fluksio/flow/metrics.py +++ b/backend/fluksio/flow/metrics.py @@ -155,7 +155,11 @@ class MetricsCollector: """Fold one event in. Synchronous: this is arithmetic on dicts.""" kind = str(event.get("type") or "") ts = float(event.get("ts") or time.time()) - run = self._runs.get(str(event.get("run") or "")) + # Two different things: the id as the event carries it, which is what a + # recorded row keeps, and the live cascade it belongs to — which a batch + # run has none of, since nothing here started one for it. + run_id = str(event.get("run") or "")[:64] + run = self._runs.get(run_id) if kind == "node_executed": bucket = self._bucket(event) @@ -199,6 +203,7 @@ class MetricsCollector: detail=(f"{error}\n{traceback}" if traceback else error)[ :DETAIL_CAP ], + run=run_id, ) ) return @@ -223,6 +228,7 @@ class MetricsCollector: flow=str(event.get("flow") or ""), node=str(event.get("node") or ""), detail=_detail(event) or "Reported itself down.", + run=run_id, ) ) return @@ -247,6 +253,7 @@ class MetricsCollector: flow=str(event.get("flow") or ""), node=str(event.get("node") or event.get("task") or ""), detail=_detail(event), + run=run_id, ) ) diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 7b0418b..be80ea0 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -237,6 +237,7 @@ class Pipeline: "observer", "emission_observer", "run_cache", + "run_id", ) def __init__( @@ -252,6 +253,7 @@ class Pipeline: observer: Callable[[NodeOutcome], None] | None = None, emission_observer: Callable[[str, dict[str, Any]], None] | None = None, run_cache: RunCacheLookup | None = None, + run_id: str = "", ) -> None: self._nodes = nodes or [] # Stopped flows are stored and survive a restart; paused ones are a @@ -286,6 +288,11 @@ class Pipeline: # none: a cascade is about what just happened, not about what a node # once returned for the same inputs. self.run_cache = run_cache + # The batch run this pipeline belongs to, if any. A live cascade names + # the journaled item it came from instead, which is what the events + # below carry; a run has no such item, so without this its failures + # would be recorded belonging to nothing. + self.run_id = run_id # How deep to keep each message's series; a chart asking for more # than the default puts its message in here. Swapped, never mutated. self.history_limits: dict[str, int] = {} @@ -853,7 +860,7 @@ class Pipeline: "flow": node.flow, "node": node.id, "error": error, - "run": entry_id, + "run": entry_id or self.run_id, "ts": time.time(), } ) @@ -916,7 +923,7 @@ class Pipeline: "node": node.id, "outputs": len(outputs or {}), "duration_ms": 0.0, - "run": entry_id, + "run": entry_id or self.run_id, "ts": time.time(), } ) @@ -1007,7 +1014,7 @@ class Pipeline: # which is a different thing to show than one that emitted. "outputs": len(result or {}), "duration_ms": duration_ms, - "run": entry_id, + "run": entry_id or self.run_id, "ts": time.time(), } ) diff --git a/backend/fluksio/flow/provision.py b/backend/fluksio/flow/provision.py index 573c7a9..204d4eb 100644 --- a/backend/fluksio/flow/provision.py +++ b/backend/fluksio/flow/provision.py @@ -125,7 +125,12 @@ class SlurmProvisioner: profiles: list[SlurmProfile], ssh_key: str = "", artifact_url: str = "", + #: How long a machine sits idle before it gives itself back; 0 keeps it + #: for as long as the job runs, which is what a queue paid for in hours + #: of wall time wants. max_idle_s: float = 300.0, + #: How long a submitted job may take to attach before it is cancelled; + #: 0 waits for as long as the scheduler makes it wait. provision_timeout_s: float = 900.0, events: EventBus | None = None, ) -> None: @@ -255,7 +260,10 @@ class SlurmProvisioner: if job.worker and job.worker in attached: # It arrived. Asking again is somebody else's decision. del self._outstanding[name] - elif time.monotonic() - job.since > self.provision_timeout_s: + elif ( + self.provision_timeout_s > 0 + and time.monotonic() - job.since > self.provision_timeout_s + ): del self._outstanding[name] if job.job_id: expired.append(job.job_id) diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index d34efd0..72bf8e6 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -68,9 +68,12 @@ LEASE_INTERVAL_S = 20.0 LEASE_STALE_S = 90.0 #: How often stale leases are looked for. SWEEP_INTERVAL_S = 30.0 -#: Runs driven at once. Node bodies are bounded by the worker pool anyway; -#: this only bounds how many graphs are in flight. +#: Runs driven at once, unless `FLOW_MAX_RUNS` says otherwise. Node bodies are +#: bounded by the worker pool anyway; this only bounds how many graphs are in +#: flight. MAX_PARALLEL = 4 +#: Runs claimed per poll, at least. Raising the pool raises this with it, so a +#: queue of ready runs fills the drivers in one pass rather than four a second. CLAIM_COUNT = 4 CLAIM_BLOCK_MS = 1000 ERROR_CAP = 2000 @@ -706,7 +709,10 @@ class RunService: # the isolation it wants, minus surviving the process. self._state_factory = state_factory or (lambda _ns: MemoryState()) self.engine_name = f"{socket.gethostname()}-{os.getpid()}"[:64] - self._pool = ThreadPoolExecutor(max_workers=parallel, thread_name_prefix="run") + self.parallel = max(1, parallel) + self._pool = ThreadPoolExecutor( + max_workers=self.parallel, thread_name_prefix="run" + ) self._stop = threading.Event() self._consumer: threading.Thread | None = None self._keeper: threading.Thread | None = None @@ -882,7 +888,9 @@ class RunService: # Runs put back to wait for a worker come due here. The claim # below blocks for a second, so this is about once a second. self.queue.move_due(time.time()) - items = self.queue.claim(CLAIM_COUNT, CLAIM_BLOCK_MS) + items = self.queue.claim( + max(CLAIM_COUNT, self.parallel), CLAIM_BLOCK_MS + ) failures = 0 except Exception as exc: failures += 1 diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 3c09b76..4ef3024 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -206,8 +206,17 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: queue=_work_queue("run"), state_factory=_run_state, artifacts=artifacts, + parallel=settings.FLOW_MAX_RUNS, ) app.state.run_service = run_service + # Said out loud because it is the only way to tell that a settings file was + # read at all — the numbers are what someone raising them is looking for. + logger.info( + "Engine limits: workers=%s cascades=%s runs=%s", + settings.FLOW_MAX_WORKERS, + execution.max_cascades, + run_service.parallel, + ) watchdog = LoopWatchdog(event_bus) app.state.watchdog = watchdog watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog") diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 82c9aae..06ede43 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -273,6 +273,9 @@ class EngineEvent(SQLModel, table=True): detail: str = "" #: Who did it, on audit rows. actor: str = "" + #: The run this happened in: a batch run's id, or the journaled item a live + #: cascade came from. Empty for everything that belongs to neither. + run: str = Field(default="", index=True, max_length=64) class FlowRun(SQLModel, table=True): diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 5901b71..7b69f1c 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -59,7 +59,12 @@ def _fail(message: str) -> int: return 1 -def _unreachable(exc: Exception, note: str = "") -> int: +#: What to try when nothing answered at all. By far the commonest reason is +#: that no engine is running, and the message said only that it was not. +NO_ENGINE = "Is one running? `fluksio serve` starts one." + + +def _unreachable(exc: Exception, note: str = NO_ENGINE) -> int: """The engine did not answer. Say so as a sentence, not a traceback.""" return _fail( f"engine not answering ({type(exc).__name__}: {exc})" @@ -514,7 +519,7 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: # The run is the engine's, not this command's: it carries on, and its # id is how to find it again. return _unreachable( - exc, f"Run {handle.id} is still on the engine." if handle else "" + exc, f"Run {handle.id} is still on the engine." if handle else NO_ENGINE ) diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 02b52e2..973cd76 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -299,18 +299,22 @@ class Client: kind: str = "failure", limit: int = 10, flow: str = "", + run: str = "", since: Any = None, until: Any = None, ) -> list[dict[str, Any]]: """What went wrong, or who changed what. Newest first. - Engine-wide, and narrowed by flow or by time. What *one run* did is a - question about that run: :attr:`RunHandle.failures` answers it from the - run's own node rows, which carry the traceback anyway. + Engine-wide, and narrowed by flow, by run or by time. ``run`` takes a + run id and answers what the engine recorded during it. For the failure + that ended a run, :attr:`RunHandle.failures` is still the shorter road: + it reads the run's own node rows, which carry the traceback too. """ query: dict[str, Any] = {"kind": kind, "limit": limit} if flow: query["flow"] = flow + if run: + query["run"] = run for name, value in (("since", since), ("until", until)): if value is not None: query[name] = ( diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index 29e0b45..95331a5 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -312,3 +312,30 @@ def test_events_narrow_to_one_minute( # The upper bound is exclusive, so the failure a minute later is not in it. assert [event["detail"] for event in events] == ["minute-in"] + + +def test_events_narrow_to_one_run( + client: TestClient, superuser_token_headers: dict[str, str], db: Session +) -> None: + """Reading one run's failures without filtering the engine-wide list.""" + ts = datetime.now(UTC) - timedelta(hours=6) + db.add( + EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="mine", run="run-mine") + ) + db.add( + EngineEvent( + ts=ts, type="node_error", flow=FLOW, detail="theirs", run="run-theirs" + ) + ) + # What a live cascade unrelated to any run leaves, and what every row + # written before the column existed looks like. + db.add(EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="neither")) + db.commit() + + events = client.get( + f"{PREFIX}/events", + headers=superuser_token_headers, + params={"flow": FLOW, "run": "run-mine"}, + ).json() + + assert [event["detail"] for event in events] == ["mine"] diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 95c382c..1bfac14 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -598,3 +598,47 @@ def test_a_step_the_x_metric_never_reached_is_left_out( ).json() assert [point[0] for point in answer["lines"][0]["points"]] == [10.0, 20.0, 30.0] + + +def test_a_run_records_the_code_it_started_with_not_the_code_it_was_queued_with(): + """A sweep queues every run at once and the tree moves while it waits. + + `_restamp` is what runs at claim time, so the digest the record keeps is + the one the run actually executed. + """ + service = RunService(controller=_Unusable(), queue=_Unusable()) + with Session(db_engine) as session: + session.add( + Run( + id="stamp-1", + flow="study", + status="queued", + code_digest="at-submit", + created_at=datetime.now(UTC), + ) + ) + session.commit() + run = session.get(Run, "stamp-1") + + assert service._restamp(run, "at-claim") == "at-claim" + + with Session(db_engine) as session: + assert session.get(Run, "stamp-1").code_digest == "at-claim" + + +def test_a_tree_that_did_not_move_is_not_written_again(): + service = RunService(controller=_Unusable(), queue=_Unusable()) + with Session(db_engine) as session: + session.add( + Run( + id="stamp-2", + flow="study", + status="queued", + code_digest="same", + created_at=datetime.now(UTC), + ) + ) + session.commit() + run = session.get(Run, "stamp-2") + + assert service._restamp(run, "same") == "same" diff --git a/backend/tests/flow/test_provision.py b/backend/tests/flow/test_provision.py index 7ade8b8..816eb39 100644 --- a/backend/tests/flow/test_provision.py +++ b/backend/tests/flow/test_provision.py @@ -8,6 +8,7 @@ it should, and that a job nobody ever attached does not sit in the queue. import subprocess import threading +import time from fluksio.flow.placement import Placer from fluksio.flow.provision import SlurmProfile, SlurmProvisioner, load_provisioners @@ -116,6 +117,22 @@ def test_a_machine_that_arrives_clears_the_way_for_the_next_ask(monkeypatch): def test_a_job_that_never_attaches_is_cancelled(monkeypatch): + ssh = FakeSsh() + monkeypatch.setattr(subprocess, "run", ssh) + hpc = cluster(provision_timeout_s=0.01) + + hpc.provision(cpus=2, gpus=0, ram_mb=0) + assert ssh.ran.wait(5) + # Slept rather than set to zero: zero is what says "no deadline" now. + time.sleep(0.02) + hpc.reconcile(set()) + + assert hpc.status()["outstanding"] == [] + assert ["scancel", "4711"] == ssh.calls[-1][0][-2:] + + +def test_no_deadline_waits_for_as_long_as_the_queue_does(monkeypatch): + """0 is "never give up", not "give up now": a cluster can queue for days.""" ssh = FakeSsh() monkeypatch.setattr(subprocess, "run", ssh) hpc = cluster(provision_timeout_s=0) @@ -124,8 +141,8 @@ def test_a_job_that_never_attaches_is_cancelled(monkeypatch): assert ssh.ran.wait(5) hpc.reconcile(set()) - assert hpc.status()["outstanding"] == [] - assert ["scancel", "4711"] == ssh.calls[-1][0][-2:] + assert hpc.status()["outstanding"][0]["job"] == "4711" + assert [argv for argv, _ in ssh.calls if "scancel" in argv] == [] def test_a_refused_submission_says_what_the_cluster_said(monkeypatch): diff --git a/backend/tests/test_metrics.py b/backend/tests/test_metrics.py index 0b78a3a..a628ba5 100644 --- a/backend/tests/test_metrics.py +++ b/backend/tests/test_metrics.py @@ -224,3 +224,27 @@ def test_a_traceback_no_failure_ever_claims_is_dropped() -> None: asyncio.run(collector.flush()) assert collector._tracebacks == {} + + +def test_a_failure_keeps_the_run_it_happened_in(db: Session) -> None: + """The payload always carried it; the row used to drop it. + + A batch run publishes no `cascade_started`, so there is no `FlowRun` beside + this — which is exactly the case that had no way of being asked about. + """ + collector = MetricsCollector(EventBus()) + ts = datetime.now(UTC).replace(second=0, microsecond=0).timestamp() + collector.handle( + { + "type": "node_error", + "flow": FLOW, + "node": NODE, + "error": "ValueError: in a run", + "run": "run-abc123", + "ts": ts, + } + ) + asyncio.run(collector.flush()) + + failure = db.exec(select(EngineEvent).where(EngineEvent.run == "run-abc123")).one() + assert failure.detail.startswith("ValueError: in a run") diff --git a/docs/code/api.md b/docs/code/api.md index 33baa41..39d51f6 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -176,7 +176,7 @@ read one back. | `/observability/timeseries` | executions and failures over a window | | `/observability/flows` | per-flow rollups with a 60-slice trend | | `/observability/runs` | recent cascades, with `?flow=`, `?since=`, `?until=` | -| `/observability/events?kind=failure\|audit` | what went wrong, or who changed what | +| `/observability/events?kind=failure\|audit` | what went wrong, or who changed what; narrows by `?flow=`, `?run=`, `?since=`, `?until=` | | `/observability/dead-letter` | work the engine gave up on | `GET /utils/health/` is the deep health check the container probe uses: it diff --git a/docs/code/workers.md b/docs/code/workers.md index e3c35bc..c1049fe 100644 --- a/docs/code/workers.md +++ b/docs/code/workers.md @@ -204,6 +204,10 @@ never attaches within `provision_timeout_s` is `scancel`led, as is anything outstanding when the engine stops. `--max-idle` is what ends the job at the other end, so an allocation goes back rather than idling to its walltime. +Both take `0` for "no limit": `provision_timeout_s: 0` waits for as long as the +queue does, which is what a cluster that queues overnight needs, and +`max_idle_s: 0` keeps the machine for the job's whole walltime. + `GET /workers/resources` reports what is outstanding and what last went wrong. !!! note "It needs a route out" diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index c367e79..914efce 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -125,6 +125,7 @@ warning into a refusal to start. |---|---|---| | `FLOW_MAX_WORKERS` | `4` | node-code subprocesses run in parallel | | `FLOW_MAX_CASCADES` | `4` | cascades in flight at once; throughput is this over the mean cascade time, so raise it where nodes wait on a network rather than a CPU | +| `FLOW_MAX_RUNS` | `4` | batch runs driven at once. A different limit from the one above: a run drives a whole graph, and its nodes are bounded by `FLOW_MAX_WORKERS`. This is what a sweep queues behind | | `FLOW_NODE_TIMEOUT` | `0` | seconds a node may be silent, unless it sets its own; 0 is no limit | | `FLOW_CPUS` | `0` | cores nodes that declare `resources` may be given; 0 works it out as every core but two, which are what keeps the engine answering while the machine is busy | | `FLOW_GPUS` | `0` | GPUs on this machine, each held by one node at a time. Not detected — say how many there are | @@ -132,6 +133,10 @@ warning into a refusal to start. | `ARTIFACT_GC_INTERVAL_S` | `3600` | how often artifact bytes nothing refers to are swept away; 0 never sweeps | | `ARTIFACT_GC_GRACE_S` | `3600` | how long a freshly written artifact is spared, whatever refers to it | +The three concurrency limits are also flags on `fluksio serve` — `--max-workers`, +`--max-cascades`, `--max-runs` — which outrank the file, and the engine says +which numbers it started with in its first lines. + An artifact is referred to by a run that recorded it or by a message currently holding it; anything else is what a camera published four hours ago, and the sweep is what keeps a flow streaming media from filling the disk. It stands diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 5e38b81..64ea629 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -715,10 +715,14 @@ export const EventRowSchema = { actor: { type: 'string', title: 'Actor' + }, + run: { + type: 'string', + title: 'Run' } }, type: 'object', - required: ['id', 'ts', 'type', 'flow', 'node', 'detail', 'actor'], + required: ['id', 'ts', 'type', 'flow', 'node', 'detail', 'actor', 'run'], title: 'EventRow' } as const; diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 0cb2aba..105c39a 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -1548,9 +1548,13 @@ export class ObservabilityService { * * ``since`` is inclusive and ``until`` exclusive, the same window ``/runs`` * takes, so a list can cover the span the charts beside it are drawn from. + * ``run`` narrows to one run — a batch run's id, or the journaled item a live + * cascade came from. Rows recorded before the column existed carry none, so + * an old failure answers no run at all rather than the wrong one. * @param data The data for the request. * @param data.kind * @param data.flow + * @param data.run * @param data.since * @param data.until * @param data.limit @@ -1564,6 +1568,7 @@ export class ObservabilityService { query: { kind: data.kind, flow: data.flow, + run: data.run, since: data.since, until: data.until, limit: data.limit diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 888eecd..c11f714 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -222,6 +222,7 @@ export type EventRow = { node: string; detail: string; actor: string; + run: string; }; export type FlavorCreate = { @@ -1662,6 +1663,7 @@ export type ObservabilityReadEventsData = { flow?: (string | null); kind?: 'failure' | 'audit'; limit?: number; + run?: (string | null); since?: (string | null); until?: (string | null); };