diff --git a/backend/fluksio/api/routes/cloud.py b/backend/fluksio/api/routes/cloud.py index 70f5513..cb7d641 100644 --- a/backend/fluksio/api/routes/cloud.py +++ b/backend/fluksio/api/routes/cloud.py @@ -12,7 +12,6 @@ chain stops with whoever a superuser here typed a code for. from __future__ import annotations -import asyncio import logging import secrets from typing import Any @@ -224,13 +223,6 @@ def disconnect(request: Request) -> Message: def _start_connector(app: Any) -> None: - from fluksio.cloud.connector import CloudConnector + from fluksio.cloud.connector import start - existing = getattr(app.state, "cloud_task", None) - if existing is not None: - existing.cancel() - connector = CloudConnector(app) - app.state.cloud_connector = connector - app.state.cloud_task = asyncio.create_task( - connector.serve_forever(), name="cloud-connector" - ) + start(app) diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index fd81b49..eedd9b8 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -16,6 +16,7 @@ before any of that happens, which is what `_configure_environment` is for. from __future__ import annotations import argparse +import copy import os import secrets import sys @@ -285,16 +286,48 @@ def cmd_serve(args: argparse.Namespace) -> int: # One process: it holds the flow engine, and a second worker would be a # second engine — duplicated subscriptions, cron ticks and webhooks. - uvicorn.run(app, host=args.host, port=args.port, log_level=args.log_level) + uvicorn.run( + app, + host=args.host, + port=args.port, + log_level=args.log_level, + log_config=_log_config(args.log_level), + ) return 0 +def _log_config(level: str) -> dict[str, Any]: + """Uvicorn's logging, with the engine's own loggers drawn the same way. + + Without this the engine's lines go to the root logger, which has no handler + configured and falls back to `INFO:fluksio.cloud.connector:...` — beside + uvicorn's own aligned, coloured output it reads like something went wrong. + """ + from uvicorn.config import LOGGING_CONFIG + + config = copy.deepcopy(LOGGING_CONFIG) + # Named rather than configuring the root: everything else that logs — httpx + # on every portal call, for one — is at INFO too, and today none of it is + # printed at all. Styling the root would turn all of it on. + for name in ("fluksio", "alembic"): + config["loggers"][name] = { + "handlers": ["default"], + "level": level.upper(), + # It has a handler of its own now; propagating would print each + # line twice the moment anything configures the root. + "propagate": False, + } + return config + + def cmd_enroll(args: argparse.Namespace) -> int: data_dir = _data_dir(args.data_dir, args.shared) _prepare(data_dir) result = _enroll(args.portal, args.code, args.as_email) if result == 0: - _say("Start it with `fluksio serve`; it dials the portal as it comes up.") + # True either way round: one already serving notices within seconds, + # and one not yet started dials as it comes up. + _say("An engine already running picks this up; otherwise `fluksio serve`.") return result diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index 60918aa..d022284 100644 --- a/backend/fluksio/cloud/connector.py +++ b/backend/fluksio/cloud/connector.py @@ -50,6 +50,49 @@ MAX_WS_MESSAGE = 1024 * 1024 #: Only the versioned API is served over the tunnel. The MCP mount and the #: OAuth endpoints live outside it and stay local-only. ALLOWED_PREFIX = "/api/v1/" +#: How often to look for a config that appeared while the engine was up. +#: Enrolment is something a person does and then waits on, so this is the +#: delay they sit through — short enough not to be worth a restart. +ENROL_POLL_S = 3.0 + + +def start(app: FastAPI) -> None: + """Dial the portal, replacing any link already up.""" + existing = getattr(app.state, "cloud_task", None) + if existing is not None: + existing.cancel() + connector = CloudConnector(app) + app.state.cloud_connector = connector + app.state.cloud_task = asyncio.create_task( + connector.serve_forever(), name="cloud-connector" + ) + + +async def watch_enrolment(app: FastAPI) -> None: + """Notice an enrolment that happened outside this process. + + ``fluksio enroll`` writes the config against the database directly, with no + idea whether an engine is running — so without this, pairing an installation + that is already serving would take a restart to come into effect. Enrolling + through the API starts the link itself and this sees a task already there. + """ + while True: + await asyncio.sleep(ENROL_POLL_S) + task = getattr(app.state, "cloud_task", None) + if task is not None and task.done(): + # It returns of its own accord when the config goes away, which is + # what `fluksio disconnect` and the portal's own Disconnect do. + app.state.cloud_task = None + app.state.cloud_connector = None + task = None + # `load()`, not `exists()`: a file that cannot be read is not an + # enrolment, and the connector would give up on it the instant it + # started — which, started from here, is a restart every few seconds. + # A config that is fine but unreachable keeps its task, and the + # retrying belongs to the connector rather than to this. + if task is None and cloud_config.load() is not None: + logger.info("Enrolled while running; dialling the portal") + start(app) class CloudConnector: diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index c601c5b..d440a93 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -174,15 +174,17 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: cloud_task: asyncio.Task[None] | None = None app.state.cloud_connector = None app.state.cloud_task = None - if cloud_config.exists(): - from fluksio.cloud.connector import CloudConnector + from fluksio.cloud import connector as cloud_connector - connector = CloudConnector(app) - app.state.cloud_connector = connector - cloud_task = asyncio.create_task( - connector.serve_forever(), name="cloud-connector" - ) - app.state.cloud_task = cloud_task + if cloud_config.exists(): + cloud_connector.start(app) + cloud_task = app.state.cloud_task + # Watched whether or not one exists now: enrolling from the CLI writes the + # config from another process entirely, and an engine already serving + # should pick it up rather than need restarting. + enrol_task = asyncio.create_task( + cloud_connector.watch_enrolment(app), name="cloud-enrolment-watch" + ) try: # A mounted sub-app gets no lifespan of its own, so the MCP session # manager is entered here; without it every /mcp request fails. @@ -192,6 +194,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: watchdog_task.cancel() alerts_task.cancel() metrics_task.cancel() + enrol_task.cancel() # Re-read from app.state: enrolling at runtime replaces this. running_cloud = getattr(app.state, "cloud_task", None) or cloud_task if running_cloud is not None: diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 5863b3f..b6739e4 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -505,6 +505,22 @@ def _flow_state(flow: dict[str, Any]) -> str: return "running" +def _portal_phrase(portal: dict[str, Any]) -> tuple[str, str]: + """Where this installation stands with its portal, and how to colour it. + + Three states worth telling apart: never paired, paired and linked, and + paired but not reaching it — the last being the one somebody needs to know + about, since the dashboard is served from the other end. + """ + if not portal.get("enrolled"): + return "no portal", "dim" + if portal.get("connected"): + host = str(portal.get("portal_url") or "").split("//")[-1].rstrip("/") + return f"portal {host}", "green" + trouble = str(portal.get("last_error") or "").strip() + return "portal unreachable" + (f" ({trouble[:60]})" if trouble else ""), "red" + + def _status_screen(client: Client) -> Any: """One frame: health, the flows, and the failures under them.""" from rich.console import Group @@ -517,6 +533,12 @@ def _status_screen(client: Client) -> Any: # fails as an engine event, while a batch run fails on its own row. failures = client.events(kind="failure", limit=5) runs = client.runs(limit=5) + try: + portal = client.cloud_status() + except (SyncError, ApiError): + # Never enrolled, or an engine too old to answer. Neither is worth + # failing a status screen over. + portal = {} healthy = summary.get("status") == "ok" head = Text() @@ -527,6 +549,8 @@ def _status_screen(client: Client) -> Any: problems = summary.get("problems") or [] if problems: head.append(" " + " · ".join(str(p) for p in problems), style="yellow") + phrase, style = _portal_phrase(portal) + head.append(" " + phrase, style=style) counts = summary.get("flows") or {} nodes = summary.get("nodes") or {} @@ -758,14 +782,32 @@ def add_parsers(subparsers: Any) -> None: "run", help="sync this directory, then start a run of one of its flows" ) parser.add_argument("flow") - parser.add_argument("--seed", type=int, default=None) + parser.add_argument( + "--seed", + type=int, + default=None, + help=( + "the run's seed: recorded on it, part of what tells two runs of one " + "configuration apart, and passed to an input named 'seed' when the " + "flow declares one" + ), + ) parser.add_argument("--wait", action="store_true", help="block until it finishes") parser.add_argument( "--follow", action="store_true", help="wait, printing the numbers it reports as they arrive", ) - parser.add_argument("--timeout", type=float, default=0.0) + parser.add_argument( + "--timeout", + type=float, + default=0.0, + metavar="SECONDS", + help=( + "give up waiting after this long and leave the run going; 0, the " + "default, waits as long as it takes. Not the node timeout" + ), + ) parser.add_argument( "--no-sync", action="store_true", @@ -814,10 +856,29 @@ def add_parsers(subparsers: Any) -> None: metavar="NAME=V1,V2", help="an input and the values to try; repeat for a grid", ) - parser.add_argument("--seed", type=int, default=None) + parser.add_argument( + "--seed", + type=int, + default=None, + help="the seed every run in the sweep gets; vary it with --param seed=1,2", + ) parser.add_argument("--wait", action="store_true", help="block until all finish") - parser.add_argument("--timeout", type=float, default=0.0) - parser.add_argument("--no-sync", action="store_true") - parser.add_argument("--no-cache", action="store_true") + parser.add_argument( + "--timeout", + type=float, + default=0.0, + metavar="SECONDS", + help="give up waiting after this long; 0, the default, waits them out", + ) + parser.add_argument( + "--no-sync", + action="store_true", + help="run what is already on the engine, without uploading first", + ) + parser.add_argument( + "--no-cache", + action="store_true", + help="execute every node, even one an earlier run already answered", + ) with_engine(parser, local=True) parser.set_defaults(func=cmd_sweep) diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index db3a956..b48e802 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -206,6 +206,11 @@ class Client: result: dict[str, Any] = self._call("GET", "/observability/summary") return result + def cloud_status(self) -> dict[str, Any]: + """Whether this installation is enrolled with a portal, and linked.""" + result: dict[str, Any] = self._call("GET", "/cloud/status") + return result + def events(self, kind: str = "failure", limit: int = 10) -> list[dict[str, Any]]: """What went wrong, or who changed what. Newest first.""" result = self._call( diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py index 1b192ec..5e35fc8 100644 --- a/backend/tests/test_cloud.py +++ b/backend/tests/test_cloud.py @@ -9,6 +9,7 @@ nothing, which is what makes deleting that local account a revocation. from __future__ import annotations +import asyncio import uuid from dataclasses import replace from unittest.mock import Mock, patch @@ -358,3 +359,50 @@ def test_the_owner_is_adopted_when_the_enrolling_row_is_gone( healed = cloud_config.load() assert healed is not None assert healed.local_user_id == str(enrolled.id) + + +# ----------------------------------------------------------------------------- +# Enrolling an engine that is already running +# +# `fluksio enroll` writes the config against the database from another process +# entirely, with no idea whether an engine is up. Noticing that is what saves a +# restart. +# ----------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_a_config_that_appears_while_running_is_dialled(monkeypatch) -> None: + from fluksio.cloud import connector + + app = Mock() + app.state = Mock(cloud_task=None, cloud_connector=None) + started: list[object] = [] + monkeypatch.setattr(connector, "ENROL_POLL_S", 0.01) + monkeypatch.setattr(connector, "start", lambda one: started.append(one)) + monkeypatch.setattr(cloud_config, "load", lambda: object()) + + watcher = asyncio.create_task(connector.watch_enrolment(app)) + await asyncio.sleep(0.05) + watcher.cancel() + + assert started + + +@pytest.mark.anyio +async def test_a_config_that_cannot_be_read_is_not_dialled(monkeypatch) -> None: + """Otherwise the connector gives up at once and this restarts it forever.""" + from fluksio.cloud import connector + + app = Mock() + app.state = Mock(cloud_task=None, cloud_connector=None) + started: list[object] = [] + monkeypatch.setattr(connector, "ENROL_POLL_S", 0.01) + monkeypatch.setattr(connector, "start", lambda one: started.append(one)) + # The file is there; it just does not parse, which `load` reports as None. + monkeypatch.setattr(cloud_config, "load", lambda: None) + + watcher = asyncio.create_task(connector.watch_enrolment(app)) + await asyncio.sleep(0.05) + watcher.cancel() + + assert not started diff --git a/docs/code/cli.md b/docs/code/cli.md index 226596a..b0c6f59 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -180,7 +180,18 @@ the numbers the run reports as they arrive: ``` Ctrl-C while either is waiting cancels the run on the engine rather than only -stopping the watching, and exits 130. +stopping the watching, and exits 130. `--timeout SECONDS` gives up *waiting* +after that long and leaves the run going; it is nothing to do with a node's own +timeout. + +`--seed N` is the experiment's seed, and it does three things. It is recorded +on the run, so what a result came from is answerable later. It goes into the +digest that identifies a run's inputs, so two runs of one configuration with +different seeds are different runs rather than a cache hit. And if the flow +declares an input named `seed`, that is what fills it — so the number the run +is labelled with is the number your code actually drew from, instead of merely +looking like it. A flow that declares no such input still records it, and +nothing reads it. Sweep over seeds with `--param seed=1,2,3`. An input declared as an `artifact` takes the file a previous run produced, named rather than typed out: @@ -234,8 +245,14 @@ fluksio status [--watch] ``` The home screen's top half in a terminal: whether the engine is healthy and -what is wrong if not, then every flow with its state, its node count and -whether it has unpublished changes, and the last few failures under them. +what is wrong if not, whether it is paired with a portal and reaching it, then +every flow with its state, its node count and whether it has unpublished +changes, and the last few runs and failures under them. + +The portal reads one of three ways. `no portal` means this installation was +never enrolled. `portal hub.fluksio.com` means the link is up. `portal +unreachable` names the error, and is the one worth acting on — the dashboard is +served from the other end, so nobody can reach it while that is showing. `--watch` keeps it on screen and refreshes every five seconds until Ctrl-C — the cadence the dashboard polls at, since nothing here moves faster. It needs a