From 148311e43d6e103cd2c703c75ad36f2a8e1b996c Mon Sep 17 00:00:00 2001 From: Romamo Date: Thu, 1 Oct 2026 20:33:19 +0300 Subject: [PATCH] feat!: health, backup, propose, approve and migrate run on treaty Phase 3 of moving the cloudfall CLI off argparse, for the commands that change servers and whose answer fits a dataclass. EngineContext and AgentConfig carry a step runner, subprocess by default for MCP and the argparse commands, so the treaty commands run every engine step and check-mode run through ctx.run in the project directory. operations approve and migrate keep --yes: without it they preview and answer effect noop, since treaty reads would_* only from its own dry run (treaty #197). deploy, rollback, restart and data migrate wait for treaty #183 (wouldRun). New exit codes 87 to 90 carry the result in data. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 15 + sdk/src/cloudfall/agent_tools.py | 5 + sdk/src/cloudfall/app.py | 844 ++++++++++++++++++++++++++++++- sdk/src/cloudfall/cli.py | 289 +---------- sdk/src/cloudfall/commands.py | 7 + sdk/src/cloudfall/decision.py | 10 +- sdk/src/cloudfall/lifecycle.py | 104 ++-- sdk/tests/test_app.py | 7 +- sdk/tests/test_arguments.py | 18 +- sdk/tests/test_decision.py | 10 +- sdk/tests/test_migrate.py | 11 +- 11 files changed, 976 insertions(+), 344 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b8f00db..a27cb7c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -80,6 +80,21 @@ Notable changes to Cloudfall. The format follows A missing playbook, inventory, or role directory exits 2 before Ansible starts. `playbook run --check` is now `--dry-run`; Ansible's play log streams to stderr. `cloudfall-engine manifest` describes every command +- **Breaking:** seven commands that run Ansible move to treaty: `health`, + `backup run`, `backup verify`, `operations propose`, `operations approve`, + `operator approve` and `migrate`. Ansible runs through treaty in the + project directory, so `--timeout` and Ctrl-C stop it, and `doctor` checks + ansible-playbook. `operations approve` and `migrate` still change nothing + without `--yes`; that preview answers `effect: noop` with + `data.status` `pending` or `plan`. New exit codes, each with the result + in `data`: 87 `ENGINE_STEP_FAILED` (an inventory render or playbook + failed, was 1), 88 `UNHEALTHY` (was 1), 89 `CHECK_FAILED` when check mode + fails in `propose` (was 1), 90 `NOT_VERIFIED` when an approved run fails + or does not verify (was 1), 3 `PARTIAL_FAILURE` for a paused `migrate`. + A failed `migrate` step's own code and message move from the top-level + `error` to `error.context`. `deploy`, `rollback`, `restart` and + `data migrate` stay on argparse until treaty #183, since their plan + answer has a `wouldRun` key - **Breaking:** six more `cloudfall` commands run on treaty: `init`, `add ssh-key`, `add server-type`, `add server`, `dashboard build` and `services inspect`. Their answer is the treaty envelope described below, diff --git a/sdk/src/cloudfall/agent_tools.py b/sdk/src/cloudfall/agent_tools.py index a657292..bcd94f7 100644 --- a/sdk/src/cloudfall/agent_tools.py +++ b/sdk/src/cloudfall/agent_tools.py @@ -40,6 +40,7 @@ DeployOptions, EngineContext, LifecycleError, + StepRunner, backup_service, build_release_artifact, deploy, @@ -48,6 +49,7 @@ restart, rollback, run_engine_playbook, + subprocess_step, verify_backup, ) from cloudfall.migrate import MigrateError, MigrateOptions, execute_migration @@ -125,6 +127,8 @@ class AgentConfig: gateway_ca_path: Path | None = None gateway_certificate_path: Path | None = None gateway_key_path: Path | None = None + run: StepRunner = subprocess_step + """Starts each engine step: ``subprocess`` here, ``ctx.run`` under treaty.""" def context(self) -> EngineContext: """Return the engine execution context shared by mutating tools.""" @@ -133,6 +137,7 @@ def context(self) -> EngineContext: schema_directory=self.schema_directory, engine_directory=self.engine_directory, inventory_file=self.inventory_file, + run=self.run, ) diff --git a/sdk/src/cloudfall/app.py b/sdk/src/cloudfall/app.py index d09052c..14af68f 100644 --- a/sdk/src/cloudfall/app.py +++ b/sdk/src/cloudfall/app.py @@ -35,6 +35,7 @@ Timeout, ) +from cloudfall.agent_tools import AgentConfig from cloudfall.ansible_api import ( AnsibleReadError, InventorySource, @@ -64,7 +65,20 @@ ) from cloudfall.commands import CLI_COMMANDS from cloudfall.dashboard import build_dashboard -from cloudfall.decision import DECISION_DIRECTORY, DecisionError, DecisionStore +from cloudfall.decision import ( + CHECK_TIMEOUT_SECONDS, + DECISION_DIRECTORY, + ERROR_DECISION_MISSING, + ApprovalRequest, + CheckRunner, + DecisionError, + DecisionStatus, + DecisionStore, + ProposalRequest, + Targets, + approve, + propose, +) from cloudfall.domain import ( ConnectionAddress, Hostname, @@ -73,9 +87,36 @@ TcpPort, ) from cloudfall.inventory import PlatformInventory +from cloudfall.lifecycle import ( + ERROR_EXECUTION_FAILED, + STEP_TIMEOUT_SECONDS, + EngineContext, + ExecutionStep, + LifecycleError, + StepRun, + StepRunner, + backup_service, + health, + verify_backup, +) +from cloudfall.migrate import MigrateError, MigrateOptions, execute_migration from cloudfall.observation import load_observations from cloudfall.operations import FleetOperations, UtcTimestamp, build_operations_view -from cloudfall.operator import ERROR_PROPOSAL_MISSING, OperatorError, ProposalStore +from cloudfall.operator import ( + ERROR_PROPOSAL_MISSING, + AlertFeed, + ApproveOptions, + OperatorError, + ProposalStatus, + ProposalStore, + TriggerKind, + alert_resolution_verifier, + drift_resolution_verifier, + engine_auditor, + engine_executor, + gateway_feed, +) +from cloudfall.operator import approve as approve_proposal from cloudfall.project import ( PROJECT_DIRECTORY_VARIABLE, InitOptions, @@ -89,7 +130,7 @@ resolve_installed_version, resolve_project_directory, ) -from cloudfall.resources import default_schema_directory +from cloudfall.resources import default_engine_directory, default_schema_directory from cloudfall.secrets import load_environment_receipts from cloudfall.service_evidence import ( DeploymentReceiptSet, @@ -108,7 +149,7 @@ ) if TYPE_CHECKING: - from collections.abc import Mapping + from collections.abc import Mapping, Sequence app = App( "cloudfall", @@ -1254,3 +1295,798 @@ def services_inspect( status="ok", observations=[str(path) for path in paths], ) + + +# Commands that change servers: health, backup run/verify, operations +# propose/approve, operator approve, migrate. Ansible runs through ctx.run, +# in the project directory as it did under argparse. + + +app.exit_code( + "ENGINE_STEP_FAILED", + 87, + description=( + "An engine step (inventory render or playbook run) failed; hosts may be " + "partly changed" + ), + retryable=False, + side_effects="partial", + suggestion="read error.message for the end of the step's output, fix it, rerun", +) +app.exit_code( + "UNHEALTHY", + 88, + description="A health check failed on a server; data holds the result", + retryable=True, + side_effects="none", + suggestion=( + "read data.detail for the failing server; restart the component or " + "redeploy it, then probe again" + ), +) +app.exit_code( + "CHECK_FAILED", + 89, + description=( + "Check mode failed; the decision is recorded with its diff, in data" + ), + retryable=False, + side_effects="none", + suggestion="read the diff the decision cites before proposing again", +) +app.exit_code( + "NOT_VERIFIED", + 90, + description=( + "The approved run failed or its verify step did not confirm it; data " + "holds the record" + ), + retryable=False, + side_effects="partial", +) + + +def _step_runner(ctx: Ctx, directory: Path) -> StepRunner: + """Run each engine step through ``ctx.run``, in the project directory.""" + + def run(step: ExecutionStep) -> StepRun: + return ctx.run( + step.argv, + env=step.environment, + cwd=directory, + timeout=Timeout(STEP_TIMEOUT_SECONDS), + check=False, + ) + + return run + + +def _check_runner(ctx: Ctx, directory: Path) -> CheckRunner: + """Run check mode through ``ctx.run``, keeping its output as the diff.""" + + def run(argv: Sequence[str], environment: Mapping[str, str], diff: Path) -> int: + done = ctx.run( + list(argv), + env=environment, + cwd=directory, + timeout=Timeout(CHECK_TIMEOUT_SECONDS), + check=False, + ) + diff.write_text(done.stdout + done.stderr, encoding="utf-8") + return done.returncode + + return run + + +def _lifecycle_failed(error: LifecycleError) -> Exception: + context = {"code": error.code} + if error.code == ERROR_EXECUTION_FAILED: + return Exit.ENGINE_STEP_FAILED(error.detail, context=context) + return Exit.PRECONDITION(error.detail, context=context) + + +@dataclass(frozen=True, slots=True, kw_only=True) +class EngineArgs(ProjectArgs): + """Options of a command that runs engine playbooks against the fleet.""" + + engine: Path = Flag( + default=default_engine_directory(), + description="Engine directory containing ansible contracts (default: bundled)", + ) + inventory_file: Path = Flag( + default=Path("tmp/ansible-inventory.json"), + description="Rendered inventory path (default: tmp/ansible-inventory.json)", + ) + + def __post_init__(self) -> None: + """Keep a relative inventory path inside the project.""" + _inside_project(self.inventory_file, "inventory-file") + + def context(self, fleet: Fleet, ctx: Ctx) -> EngineContext: + """Return the engine contract, with steps run through ``ctx.run``.""" + return EngineContext( + project_directory=fleet.directory, + schema_directory=self.schemas, + engine_directory=self.engine, + inventory_file=fleet.path(self.inventory_file), + run=_step_runner(ctx, fleet.directory), + ) + + +# health + + +@dataclass(frozen=True, slots=True, kw_only=True) +class HealthArgs(EngineArgs): + """Arguments of ``health``.""" + + component: ResourceId = Arg(description="Component to probe") + + +class HealthPayload(Payload): + """``health``: the probe's result on every server of the component.""" + + command = "health" + + +@app.command( + "health", + description=( + "Probe one component's declared health check; an unhealthy result " + "exits non-zero with the result in data" + ), + danger_level="safe", + exit_codes=[ + "PROJECT_INVALID", + "CONFIG_INVALID", + "PRECONDITION", + "UNHEALTHY", + ], + timeout=None, + subprocess=Subprocess("ansible-playbook"), + required_tools={"ansible-playbook": "2.21.0"}, + examples=[("Probe one component", "cloudfall health crm-backend")], +) +def health_command(args: HealthArgs, ctx: Ctx, fleet: Fleet) -> HealthPayload: + """Run the health playbook and answer whether every server passed.""" + try: + result = health(args.context(fleet, ctx), args.component) + except LifecycleError as error: + raise _lifecycle_failed(error) from error + payload = HealthPayload(result.as_dict()) + if not result.healthy: + message = f"{args.component} failed its health check" + raise Exit.UNHEALTHY(message, data=payload) + return payload + + +# backup run, verify + + +backup = app.group("backup", description="Run and prove declared service backups") + + +@dataclass(frozen=True, slots=True, kw_only=True) +class BackupArgs(EngineArgs): + """Arguments of ``backup run`` and ``backup verify``.""" + + service: ResourceId = Arg(description="Declared service to back up") + receipts: Path = Flag( + default=Path("tmp/backups"), + description="Backup receipt directory (default: tmp/backups)", + ) + + def __post_init__(self) -> None: + """Keep relative paths inside the project.""" + EngineArgs.__post_init__(self) + _inside_project(self.receipts, "receipts") + + +@dataclass(frozen=True, slots=True) +class BackedUp: + """The receipt one backup or restore check wrote.""" + + effect: str + status: str + receipt: dict[str, object] = Out(ordered=True) + path: Path = Path() + + +def _backed_up(body: Mapping[str, object]) -> BackedUp: + return BackedUp( + effect="created", + status=str(body["status"]), + receipt=dict(cast("Mapping[str, object]", body["receipt"])), + path=Path(str(body["path"])), + ) + + +_BACKUP_EXIT_CODES = [ + "PROJECT_INVALID", + "CONFIG_INVALID", + "PRECONDITION", + "ENGINE_STEP_FAILED", +] + + +@backup.command( + "run", + description="Run the declared backup for one service and keep its receipt", + danger_level="mutating", + exit_codes=_BACKUP_EXIT_CODES, + timeout=None, + supports_raw_payload=True, + subprocess=Subprocess("ansible-playbook"), + required_tools={"ansible-playbook": "2.21.0"}, + examples=[("Back up one database service", "cloudfall backup run postgresql-main")], +) +def backup_run(args: BackupArgs, ctx: Ctx, fleet: Fleet) -> BackedUp: + """Run the backup playbook for the service.""" + try: + body = backup_service( + args.context(fleet, ctx), args.service, fleet.path(args.receipts) + ) + except LifecycleError as error: + raise _lifecycle_failed(error) from error + return _backed_up(body) + + +@backup.command( + "verify", + description="Prove the newest backup restores for one service", + danger_level="mutating", + exit_codes=_BACKUP_EXIT_CODES, + timeout=None, + supports_raw_payload=True, + subprocess=Subprocess("ansible-playbook"), + required_tools={"ansible-playbook": "2.21.0"}, + examples=[ + ( + "Restore-check one database service", + "cloudfall backup verify postgresql-main", + ), + ], +) +def backup_verify(args: BackupArgs, ctx: Ctx, fleet: Fleet) -> BackedUp: + """Restore the newest backup into scratch and keep the proof.""" + try: + body = verify_backup( + args.context(fleet, ctx), args.service, fleet.path(args.receipts) + ) + except LifecycleError as error: + raise _lifecycle_failed(error) from error + return _backed_up(body) + + +# operations propose, approve + + +@dataclass(frozen=True, slots=True, kw_only=True) +class ProposeArgs(CatalogArgs): + """Arguments of ``operations propose``.""" + + operation: ResourceId = Arg(description="Operation id") + target: str | None = Flag( + default=None, + description=( + "Host or group the operation runs against, as its target scope requires" + ), + ) + input: tuple[str, ...] = Flag( + default=(), + description="Value for one declared input, as NAME=VALUE (repeatable)", + ) + observed: Path = Flag( + default=Path("tmp/observed"), + description=( + "Snapshot directory the proposal cites as its basis (default: tmp/observed)" + ), + ) + decisions: Path = Flag( + default=Path(DECISION_DIRECTORY), + description=( + f"Directory holding the decision records (default: {DECISION_DIRECTORY})" + ), + ) + + def __post_init__(self) -> None: + """Refuse a malformed input, and keep relative paths in the repository.""" + CatalogArgs.__post_init__(self) + _inside_project(self.observed, "observed") + _inside_project(self.decisions, "decisions") + self.inputs() + + def inputs(self) -> dict[str, object]: + """Return the declared inputs, parsed from ``NAME=VALUE``.""" + inputs: dict[str, object] = {} + for entry in self.input: + name, separator, value = entry.partition("=") + if not separator or not name: + message = f"input must be given as NAME=VALUE, got {entry!r}" + raise ParseError(message, context={"flag": "input"}) + inputs[name] = value + return inputs + + +@dataclass(frozen=True, slots=True, kw_only=True) +class ApproveDecisionArgs(DecisionsArgs): + """Arguments of ``operations approve``.""" + + decision: ResourceId = Arg(description="Decision id from `operations propose`") + approver: str | None = Flag( + default=None, + description="Who is approving (default: the USER environment variable)", + ) + yes: bool = Flag( + default=False, + description=( + "Change the servers; without it the command shows the recorded " + "proposal and runs nothing" + ), + ) + + +@dataclass(frozen=True, slots=True) +class Decided: + """One decision record, and what to do next when it waits for approval.""" + + effect: str + status: str + decision: dict[str, object] = Out(ordered=True) + next: list[str] = Out(ordered=True) + + +def _decision_failed(error: DecisionError) -> Exception: + return Exit.PRECONDITION(error.detail, context={"code": error.code}) + + +@operations.command( + "propose", + description=( + "Run one operation in check mode and record what it would do; nothing " + "on the fleet changes" + ), + danger_level="mutating", + exit_codes=["CONFIG_INVALID", "NOT_FOUND", "PRECONDITION", "CHECK_FAILED"], + timeout=None, + subprocess=Subprocess("ansible-playbook"), + required_tools={"ansible-playbook": "2.21.0"}, + supports_raw_payload=True, + examples=[ + ( + "Propose a restart of nginx on one host", + "cloudfall operations propose restart-nginx --target h1", + ), + ], +) +def operations_propose(args: ProposeArgs, ctx: Ctx) -> Decided: + """Record a proposal with the diff check mode produced.""" + catalog = args.catalog() + try: + operation = catalog.get(args.operation) + except ConfigValidationError as error: + if error.issue.code == ERROR_OPERATION_UNDECLARED: + raise Exit.NOT_FOUND( + error.issue.message, context=error.issue.as_dict() + ) from error + raise _config_invalid(error) from error + request = ProposalRequest( + operation=operation, + targets=Targets(scope=operation.targets, pattern=args.target), + inputs=args.inputs(), + repository=args.root, + observations=args.root / args.observed, + ) + store = DecisionStore( + directory=args.root / args.decisions, catalog=SchemaCatalog(args.schemas) + ) + try: + decision = propose(request, store, run=_check_runner(ctx, args.root)) + except DecisionError as error: + raise _decision_failed(error) from error + decided = Decided( + effect="created", status="ok", decision=decision.as_document(), next=[] + ) + if decision.check.exit_code != 0: + message = ( + f"check mode exited {decision.check.exit_code}; the decision is " + "recorded with its diff" + ) + raise Exit.CHECK_FAILED(message, data=decided) + return decided + + +@operations.command( + "approve", + description=( + "Approve one recorded proposal, run it, and verify it; without --yes it " + "shows the proposal and runs nothing" + ), + danger_level="mutating", + exit_codes=["NOT_FOUND", "PRECONDITION", "RECORD_INVALID", "NOT_VERIFIED"], + timeout=None, + supports_raw_payload=True, + subprocess=Subprocess("ansible-playbook"), + required_tools={"ansible-playbook": "2.21.0"}, + examples=[ + ("Review a proposal", "cloudfall operations approve restart-nginx-20260101"), + ("Run it", "cloudfall operations approve restart-nginx-20260101 --yes"), + ], +) +def operations_approve(args: ApproveDecisionArgs, ctx: Ctx) -> Decided: + """Run what was proposed, as recorded, and record how it ended.""" + store = DecisionStore( + directory=args.root / args.decisions, catalog=SchemaCatalog(args.schemas) + ) + try: + decision = store.load(args.decision) + except DecisionError as error: + if error.code == ERROR_DECISION_MISSING: + raise Exit.NOT_FOUND(error.detail, context={"code": error.code}) from error + raise _record_invalid(error) from error + if not args.yes: + # Nothing ran. treaty reads would_* only from a dry run it switched, + # and --yes is the inverse of its --dry-run (treaty #197). + return Decided( + effect="noop", + status="pending", + decision=decision.as_document(), + next=[ + f"review the recorded diff at {decision.check.diff.path}", + "approve with --yes to run it", + ], + ) + approver = args.approver if args.approver is not None else ctx.env.get("USER", "") + try: + approved = approve( + ApprovalRequest(decision=decision, approver=approver, repository=args.root), + store, + run=_check_runner(ctx, args.root), + ) + except DecisionError as error: + raise _decision_failed(error) from error + decided = Decided( + effect="updated", status="ok", decision=approved.as_document(), next=[] + ) + if approved.status is DecisionStatus.FAILED: + message = "the approved run failed or its verify step did not confirm it" + raise Exit.NOT_VERIFIED(message, data=decided) + return decided + + +# operator approve + + +@dataclass(frozen=True, slots=True, kw_only=True) +class OperatorApproveArgs(ProposalArgs): + """Arguments of ``operator approve``.""" + + proposal: ResourceId = Arg(description="Proposal id") + engine: Path = Flag( + default=default_engine_directory(), + description="Engine directory containing ansible contracts (default: bundled)", + ) + inventory_file: Path = Flag( + default=Path("tmp/ansible-inventory.json"), + description="Rendered inventory path (default: tmp/ansible-inventory.json)", + ) + observed: Path = Flag( + default=Path("tmp/operator/observed"), + description=( + "Observation directory for drift verification " + "(default: tmp/operator/observed)" + ), + ) + verify_timeout: float = Flag( + default=180.0, + description="Seconds to wait for the trigger to resolve (default: 180)", + ) + gateway_url: str | None = Flag( + default=None, + description="Alerts endpoint (default: derived from the declared gateway)", + ) + gateway_ca: Path | None = Flag(default=None, description="Gateway CA file") + gateway_cert: Path | None = Flag( + default=None, description="Client certificate for the gateway" + ) + gateway_key: Path | None = Flag( + default=None, description="Client key for the gateway", secret=False + ) + + def __post_init__(self) -> None: + """Keep relative paths inside the project.""" + ProposalArgs.__post_init__(self) + _inside_project(self.inventory_file, "inventory-file") + _inside_project(self.observed, "observed") + + def context(self, fleet: Fleet, ctx: Ctx) -> EngineContext: + """Return the engine contract, with steps run through ``ctx.run``.""" + return EngineContext( + project_directory=fleet.directory, + schema_directory=self.schemas, + engine_directory=self.engine, + inventory_file=fleet.path(self.inventory_file), + run=_step_runner(ctx, fleet.directory), + ) + + def feed(self, fleet: Fleet) -> AlertFeed: + """Return the gateway's alert feed, which an alert proposal needs.""" + if ( + self.gateway_ca is None + or self.gateway_cert is None + or self.gateway_key is None + ): + message = ( + "approving an alert-triggered proposal requires --gateway-ca, " + "--gateway-cert, and --gateway-key" + ) + raise Exit.PRECONDITION( + message, context={"code": "operator_gateway_material_missing"} + ) + return gateway_feed( + fleet.inventory, + ca_path=self.gateway_ca, + certificate_path=self.gateway_cert, + key_path=self.gateway_key, + url_override=self.gateway_url, + ) + + +@dataclass(frozen=True, slots=True) +class Approved: + """One proposal receipt after its approved run.""" + + effect: str + status: str + proposal: dict[str, object] = Out(ordered=True) + + +@operator.command( + "approve", + description="Execute a proposal and verify its trigger resolves", + danger_level="mutating", + exit_codes=[ + "PROJECT_INVALID", + "CONFIG_INVALID", + "NOT_FOUND", + "PRECONDITION", + "RECORD_INVALID", + "ENGINE_STEP_FAILED", + "NOT_VERIFIED", + ], + timeout=None, + supports_raw_payload=True, + subprocess=Subprocess("ansible-playbook"), + required_tools={"ansible-playbook": "2.21.0"}, + examples=[ + ("Approve one proposal", "cloudfall operator approve nginx-down-20260101"), + ], +) +def operator_approve( + args: OperatorApproveArgs, ctx: Ctx, fleet: Fleet +) -> Approved: + """Run the proposal's operation, then wait for its trigger to resolve.""" + store = args.store(fleet) + context = args.context(fleet, ctx) + try: + pending = store.load(args.proposal) + if pending.trigger_kind is TriggerKind.ALERT: + verifier = alert_resolution_verifier(args.feed(fleet)) + else: + verifier = drift_resolution_verifier( + engine_auditor(context, fleet.inventory, fleet.path(args.observed)) + ) + proposal = approve_proposal( + store, + args.proposal, + engine_executor(context), + verifier, + ApproveOptions(verify_timeout_seconds=args.verify_timeout), + ) + except OperatorError as error: + raise _record_invalid(error) from error + except LifecycleError as error: + raise _lifecycle_failed(error) from error + verified = proposal.status is ProposalStatus.VERIFIED + approved = Approved( + effect="updated", + status="ok" if verified else "failed", + proposal=proposal.as_document(), + ) + if not verified: + message = "the run finished but its trigger did not resolve" + raise Exit.NOT_VERIFIED(message, data=approved) + return approved + + +# migrate + + +@dataclass(frozen=True, slots=True, kw_only=True) +class MigrateArgs(EngineArgs): + """Arguments of ``migrate``.""" + + observed: Path = Flag( + default=Path("tmp/observed"), + description="Server observation directory (default: tmp/observed)", + ) + service_observed: Path = Flag( + default=Path("tmp/observed-services"), + description="Domain observation directory (default: tmp/observed-services)", + ) + deployments: Path = Flag( + default=Path("tmp/deployments"), + description="Deployment receipt directory (default: tmp/deployments)", + ) + receipts: Path = Flag( + default=Path("tmp/releases"), + description="Release receipt directory (default: tmp/releases)", + ) + artifacts: Path = Flag( + default=Path("tmp/artifacts"), + description="Artifact directory (default: tmp/artifacts)", + ) + plan_file: Path = Flag( + default=Path("tmp/migrate/plan.json"), + description="Persisted migration plan (default: tmp/migrate/plan.json)", + ) + build: tuple[str, ...] = Flag( + default=(), + description="Build a component from a git ref, as COMPONENT=REF (repeatable)", + ) + release: tuple[str, ...] = Flag( + default=(), + description="Deploy an existing release, as COMPONENT=RELEASE (repeatable)", + ) + env_file: tuple[str, ...] = Flag( + default=(), + description="Environment file for a component, as COMPONENT=PATH (repeatable)", + ) + data: tuple[str, ...] = Flag( + default=(), + description=( + "Source URL file for a database, as DATABASE=PATH (repeatable)" + ), + ) + yes: bool = Flag( + default=False, + description=( + "Change the servers; without it the command prints the plan and " + "runs nothing" + ), + ) + restart: bool = Flag( + default=False, + description="Discard the persisted plan and start again", + ) + + def __post_init__(self) -> None: + """Refuse a malformed pair, and keep relative paths in the project.""" + EngineArgs.__post_init__(self) + for flag in ( + "observed", + "service_observed", + "deployments", + "receipts", + "artifacts", + "plan_file", + ): + _inside_project(getattr(self, flag), flag.replace("_", "-")) + for option, entries in ( + ("build", self.build), + ("release", self.release), + ("env-file", self.env_file), + ("data", self.data), + ): + _pairs(entries, option) + + def options(self, fleet: Fleet) -> MigrateOptions: + """Return what this run builds, deploys and migrates.""" + return MigrateOptions( + plan_file=fleet.path(self.plan_file), + builds=_pairs(self.build, "build"), + releases=_pairs(self.release, "release"), + environment_files={ + component: Path(value) + for component, value in _pairs(self.env_file, "env-file").items() + }, + data_migrations={ + database: Path(value) + for database, value in _pairs(self.data, "data").items() + }, + execute=self.yes, + restart=self.restart, + ) + + def config(self, fleet: Fleet, ctx: Ctx) -> AgentConfig: + """Return the filesystem contract the migration runs under.""" + return AgentConfig( + project_directory=fleet.directory, + schema_directory=self.schemas, + engine_directory=self.engine, + inventory_file=fleet.path(self.inventory_file), + observed_directory=fleet.path(self.observed), + service_observed_directory=fleet.path(self.service_observed), + deployments_directory=fleet.path(self.deployments), + releases_directory=fleet.path(self.receipts), + artifacts_directory=fleet.path(self.artifacts), + run=_step_runner(ctx, fleet.directory), + ) + + +def _pairs(entries: tuple[str, ...], option: str) -> dict[str, str]: + pairs: dict[str, str] = {} + for entry in entries: + key, separator, value = entry.partition("=") + if not separator or not key or not value: + message = f"--{option} expects NAME=VALUE, got {entry!r}" + raise ParseError(message, context={"flag": option}) + pairs[key] = value + return pairs + + +@dataclass(frozen=True, slots=True) +class Migration: + """The migration plan's steps and how far the run got.""" + + effect: str + status: str + steps: list[dict[str, object]] = Out(ordered=True) + completed: int = 0 + next: str | None = None + step: str | None = None + + +@app.command( + "migrate", + description=( + "Run the resumable end-to-end migration; without --yes it prints the plan " + "and runs nothing" + ), + danger_level="mutating", + exit_codes=[ + "PROJECT_INVALID", + "CONFIG_INVALID", + "PRECONDITION", + "PARTIAL_FAILURE", + "ENGINE_STEP_FAILED", + ], + timeout=None, + supports_raw_payload=True, + subprocess=Subprocess("ansible-playbook"), + required_tools={"ansible-playbook": "2.21.0", "git": "2.24.0"}, + examples=[ + ("Print the plan", "cloudfall migrate --build crm-backend=main"), + ("Run it", "cloudfall migrate --build crm-backend=main --yes"), + ], +) +def migrate(args: MigrateArgs, ctx: Ctx, fleet: Fleet) -> Migration: + """Run each pending step, saving progress after every one.""" + try: + result = execute_migration(args.config(fleet, ctx), args.options(fleet)) + except MigrateError as error: + raise Exit.PRECONDITION( + error.detail, context={"code": error.code} + ) from error + status = str(result["status"]) + migration = Migration( + # A plan changes nothing: noop, as without --yes in `operations approve`. + effect="noop" if status == "plan" else "updated", + status=status, + steps=list(cast("list[dict[str, object]]", result["steps"])), + completed=int(cast("int", result["completed"])), + next=cast("str | None", result.get("next")), + step=cast("str | None", result.get("step")), + ) + # The step's own error goes in error.context, beside the plan in data. + cause = { + "step": migration.step, + **cast("Mapping[str, object]", result.get("error", {})), + } + if status == "paused": + message = f"the migration paused at {migration.step}; resume it with --yes" + raise Exit.PARTIAL_FAILURE(message, context=cause, data=migration) + if status == "error": + message = f"the migration failed at {migration.step}" + raise Exit.ENGINE_STEP_FAILED(message, context=cause, data=migration) + return migration diff --git a/sdk/src/cloudfall/cli.py b/sdk/src/cloudfall/cli.py index 5687ebc..a2d2a58 100644 --- a/sdk/src/cloudfall/cli.py +++ b/sdk/src/cloudfall/cli.py @@ -14,11 +14,9 @@ from argparse import Namespace from collections.abc import Callable, Mapping, Sequence - from cloudfall.domain import ResourceId from cloudfall.operations import FleetOperations from cloudfall.operator import AlertFeed, OperatorProposal -from cloudfall.agent_tools import AgentConfig from cloudfall.ansible_api import ( ANSIBLE_CONFIG_FILE, AnsibleReadError, @@ -38,7 +36,7 @@ root_parser, schema_version_argument, ) -from cloudfall.catalog import CATALOG_DIRECTORY, OperationCatalog, load_catalog +from cloudfall.catalog import CATALOG_DIRECTORY from cloudfall.dashboard import RefreshInterval from cloudfall.dashboard_server import ( EvidenceSources, @@ -47,14 +45,8 @@ ) from cloudfall.decision import ( DECISION_DIRECTORY, - ApprovalRequest, DecisionError, - DecisionStatus, DecisionStore, - ProposalRequest, - Targets, - approve, - propose, ) from cloudfall.importer import ( ImportTargets, @@ -67,9 +59,7 @@ EngineContext, LifecycleError, LifecyclePreview, - backup_service, deploy, - health, migrate_data, preview_data_migration, preview_deploy, @@ -77,9 +67,7 @@ preview_rollback, restart, rollback, - verify_backup, ) -from cloudfall.migrate import MigrateError, MigrateOptions, execute_migration from cloudfall.observation import load_observations from cloudfall.observe import ( ObservationRequest, @@ -89,9 +77,7 @@ ) from cloudfall.operations import UtcTimestamp, build_operations_view from cloudfall.operator import ( - ApproveOptions, OperatorError, - ProposalStatus, ProposalStore, TriggerKind, alert_resolution_verifier, @@ -102,9 +88,6 @@ engine_executor, gateway_feed, ) -from cloudfall.operator import ( - approve as operator_approve, -) from cloudfall.operator import ( run_once as operator_run_once, ) @@ -1381,16 +1364,11 @@ def _dispatch( arguments: Namespace, state: ValidatedConfig, schema_directory: Path ) -> int: handlers: dict[str, Callable[[Namespace, ValidatedConfig, Path], int]] = { - "backup:run": _run_backup_run, - "backup:verify": _run_backup_verify, "secrets:render": _run_secrets_render, "data:migrate": _run_data_migrate, "dashboard:serve": _run_dashboard_serve, "deploy": _run_deploy, - "health": _run_health, "observe": _run_observe, - "migrate": _run_migrate, - "operator:approve": _run_operator_approve, "operator:run": _run_operator_run, "restart": _run_restart, "rollback": _run_rollback, @@ -1420,30 +1398,6 @@ def _command_key(arguments: Namespace) -> str: return f"{arguments.command}:{subcommand}" -def _run_operations(arguments: Namespace) -> int: - """Propose or approve against the catalog, which needs no fleet. - - An agent needs its tool list from a repository Cloudfall cannot read a - fleet from yet, so the catalog resolves against the repository it lives - in and nothing else. - """ - repository = ( - Path(arguments.repository) - if arguments.repository is not None - else Path.cwd() - ) - if arguments.operations_command == "approve": - return _approve_decision(arguments, repository) - # list, show and decisions run on treaty (cloudfall.app); propose is - # the only other subcommand that reaches argparse. - catalog = load_catalog( - repository, - Path(arguments.schemas), - repository / Path(arguments.operations), - ) - return _propose_operation(arguments, repository, catalog) - - def _run_why(arguments: Namespace) -> int: """Answer the question from the record alone: no catalog, no fleet. @@ -1470,6 +1424,13 @@ def _run_why(arguments: Namespace) -> int: return 0 +def _decision_store(arguments: Namespace, repository: Path) -> DecisionStore: + return DecisionStore( + directory=repository / Path(arguments.decisions), + catalog=SchemaCatalog(Path(arguments.schemas)), + ) + + def _run_changelog(arguments: Namespace) -> int: write_result( { @@ -1488,86 +1449,11 @@ def _run_changelog(arguments: Namespace) -> int: _RECORD_COMMANDS: Mapping[str, Callable[[Namespace], int]] = { - "operations": _run_operations, "why": _run_why, } """The commands that read the repository and nothing else: no fleet, no project.""" -def _propose_operation( - arguments: Namespace, repository: Path, catalog: OperationCatalog -) -> int: - """Run one operation in check mode and record what it would do.""" - operation = catalog.get(arguments.operation) - request = ProposalRequest( - operation=operation, - targets=Targets(scope=operation.targets, pattern=arguments.target), - inputs=_declared_inputs(arguments.input), - repository=repository, - observations=repository / Path(arguments.observed), - ) - decision = propose(request, _decision_store(arguments, repository)) - code = 0 if decision.check.exit_code == 0 else 1 - write_result(decision.as_dict(), exit_code=code) - return code - - -def _approve_decision(arguments: Namespace, repository: Path) -> int: - """Approve one recorded proposal, run it, and verify it. - - The approval needs no catalog: it acts on the record, so what a human - read is what runs. - """ - store = _decision_store(arguments, repository) - decision = store.load(arguments.decision) - if not arguments.yes: - write_result( - { - "status": "pending", - "decision": decision.as_document(), - "next": [ - "review the recorded diff at " - f"{decision.check.diff.path}", - "approve with --yes to run it", - ], - } - ) - return 0 - approver = ( - arguments.approver - if arguments.approver is not None - else os.environ.get("USER", "") - ) - approved = approve( - ApprovalRequest( - decision=decision, approver=approver, repository=repository - ), - store, - ) - code = 0 if approved.status is not DecisionStatus.FAILED else 1 - write_result(approved.as_dict(), exit_code=code) - return code - - -def _decision_store(arguments: Namespace, repository: Path) -> DecisionStore: - return DecisionStore( - directory=repository / Path(arguments.decisions), - catalog=SchemaCatalog(Path(arguments.schemas)), - ) - - -def _declared_inputs(declared: Sequence[str]) -> dict[str, object]: - """Parse repeated ``NAME=VALUE`` arguments into input values.""" - inputs: dict[str, object] = {} - for entry in declared: - name, separator, value = entry.partition("=") - if not separator or not name: - message = f"input must be given as NAME=VALUE, got {entry!r}" - raise ValueError(message) - inputs[name] = value - return inputs - - def _run_observe( arguments: Namespace, state: ValidatedConfig, _schema_directory: Path ) -> int: @@ -1622,35 +1508,6 @@ def _run_secrets_render( return 0 -def _run_backup_run( - arguments: Namespace, _state: ValidatedConfig, schema_directory: Path -) -> int: - return _run_backup_operation(arguments, schema_directory, backup_service) - - -def _run_backup_verify( - arguments: Namespace, _state: ValidatedConfig, schema_directory: Path -) -> int: - return _run_backup_operation(arguments, schema_directory, verify_backup) - - -def _run_backup_operation( - arguments: Namespace, - schema_directory: Path, - operation: Callable[[EngineContext, ResourceId, Path], dict[str, object]], -) -> int: - try: - result = operation( - _engine_context(arguments, schema_directory), - arguments.service, - Path(arguments.receipts), - ) - except LifecycleError as error: - return _lifecycle_exit(error) - write_result(result) - return 0 - - def _operator_store(arguments: Namespace, schema_directory: Path) -> ProposalStore: return ProposalStore( directory=Path(arguments.proposals), @@ -1668,20 +1525,6 @@ def _operator_feed(arguments: Namespace, inventory: PlatformInventory) -> AlertF ) -def _require_gateway_material(arguments: Namespace) -> None: - if ( - arguments.gateway_ca is None - or arguments.gateway_cert is None - or arguments.gateway_key is None - ): - code = "operator_gateway_material_missing" - message = ( - "approving an alert-triggered proposal requires --gateway-ca, " - "--gateway-cert, and --gateway-key" - ) - raise OperatorError(code, message) - - def _operator_exit(error: OperatorError) -> int: return write_error(error.as_dict(), 2) @@ -1746,44 +1589,6 @@ def _verifier_for( return _operator_exit(error) -def _run_operator_approve( - arguments: Namespace, state: ValidatedConfig, schema_directory: Path -) -> int: - inventory = PlatformInventory.from_state(state) - try: - store = _operator_store(arguments, schema_directory) - proposal_id = arguments.proposal - pending = store.load(proposal_id) - context = _engine_context(arguments, schema_directory) - if pending.trigger_kind is TriggerKind.ALERT: - _require_gateway_material(arguments) - verifier = alert_resolution_verifier(_operator_feed(arguments, inventory)) - else: - verifier = drift_resolution_verifier( - engine_auditor(context, inventory, Path(arguments.observed)) - ) - proposal = operator_approve( - store, - proposal_id, - engine_executor(context), - verifier, - ApproveOptions(verify_timeout_seconds=arguments.verify_timeout), - ) - except OperatorError as error: - return _operator_exit(error) - except LifecycleError as error: - return _lifecycle_exit(error) - verified = proposal.status is ProposalStatus.VERIFIED - write_result( - { - "status": "ok" if verified else "failed", - "proposal": proposal.as_document(), - }, - exit_code=0 if verified else 1, - ) - return 0 if verified else 1 - - def _run_dashboard_serve( arguments: Namespace, _state: ValidatedConfig, schema_directory: Path ) -> int: @@ -1946,84 +1751,6 @@ def _run_restart( return 0 -def _run_health( - arguments: Namespace, _state: ValidatedConfig, schema_directory: Path -) -> int: - try: - result = health( - _engine_context(arguments, schema_directory), - arguments.component, - ) - except LifecycleError as error: - return _lifecycle_exit(error) - code = 0 if result.healthy else 1 - write_result(result.as_dict(), exit_code=code) - return code - - -def _run_migrate( - arguments: Namespace, _state: ValidatedConfig, schema_directory: Path -) -> int: - try: - builds = _key_value_pairs(arguments.build, "--build") - releases = _key_value_pairs(arguments.release, "--release") - environment_files = { - component: Path(value) - for component, value in _key_value_pairs( - arguments.env_file, "--env-file" - ).items() - } - data_migrations = { - database: Path(value) - for database, value in _key_value_pairs(arguments.data, "--data").items() - } - except ValueError as error: - payload = { - "status": "error", - "error": {"code": "invalid_argument", "message": str(error)}, - } - return write_error(payload, 2) - config = AgentConfig( - project_directory=Path(arguments.project_directory), - schema_directory=schema_directory, - engine_directory=Path(arguments.engine), - inventory_file=Path(arguments.inventory_file), - observed_directory=Path(arguments.observed), - service_observed_directory=Path(arguments.service_observed), - deployments_directory=Path(arguments.deployments), - releases_directory=Path(arguments.receipts), - artifacts_directory=Path(arguments.artifacts), - ) - options = MigrateOptions( - plan_file=Path(arguments.plan_file), - builds=builds, - releases=releases, - environment_files=environment_files, - data_migrations=data_migrations, - execute=arguments.yes, - restart=arguments.restart, - ) - try: - result = execute_migration(config, options) - except MigrateError as error: - return write_error(error.as_dict(), 2) - status = str(result["status"]) - code = 0 if status in {"ok", "plan"} else 3 if status == "paused" else 1 - write_result(result, exit_code=code) - return code - - -def _key_value_pairs(entries: list[str], option: str) -> dict[str, str]: - pairs: dict[str, str] = {} - for entry in entries: - component, separator, value = entry.partition("=") - if not separator or not component or not value: - message = f"{option} expects COMPONENT=VALUE, got {entry!r}" - raise ValueError(message) - pairs[component] = value - return pairs - - def _write_plan(preview: LifecyclePreview) -> int: write_result( { diff --git a/sdk/src/cloudfall/commands.py b/sdk/src/cloudfall/commands.py index 3bc74b3..2700be2 100644 --- a/sdk/src/cloudfall/commands.py +++ b/sdk/src/cloudfall/commands.py @@ -348,6 +348,7 @@ def _plan(subject: str, *extra: str) -> OutputShape: CommandEffect.READ, "run one operation in check mode and record what it would change", output=(_shape("status", "decision"),), + treaty=True, ), CommandContract( "cloudfall", @@ -359,6 +360,7 @@ def _plan(subject: str, *extra: str) -> OutputShape: _shape("status", "decision", "next", when="without `--yes`"), _shape("status", "decision", when="with `--yes`"), ), + treaty=True, ), CommandContract( "cloudfall", @@ -442,6 +444,7 @@ def _plan(subject: str, *extra: str) -> OutputShape: when="`detail` only when the component is unhealthy", ), ), + treaty=True, ), CommandContract( "cloudfall", @@ -603,6 +606,7 @@ def _plan(subject: str, *extra: str) -> OutputShape: when="`paused` (exit 3) or `error` (exit 1) at `step`", ), ), + treaty=True, ), CommandContract( "cloudfall", @@ -611,6 +615,7 @@ def _plan(subject: str, *extra: str) -> OutputShape: "run the declared backup for one service on its server", gate="the service must declare the backup; a receipt is written", output=_BACKUP, + treaty=True, ), CommandContract( "cloudfall", @@ -619,6 +624,7 @@ def _plan(subject: str, *extra: str) -> OutputShape: "restore the newest backup into a scratch database on the server", gate="the service must declare the backup; a receipt is written", output=_BACKUP, + treaty=True, ), CommandContract( "cloudfall", @@ -627,6 +633,7 @@ def _plan(subject: str, *extra: str) -> OutputShape: "execute one proposal and verify that its trigger resolves", gate="the proposal id names a receipt a human has reviewed", output=_PROPOSAL, + treaty=True, ), ) diff --git a/sdk/src/cloudfall/decision.py b/sdk/src/cloudfall/decision.py index 4fc4cd2..c24410a 100644 --- a/sdk/src/cloudfall/decision.py +++ b/sdk/src/cloudfall/decision.py @@ -55,7 +55,7 @@ _TREE_DIRECTORY = "tree" _DIFF_SUFFIX = ".diff" _ANSIBLE_CONFIG_FILE = "ansible.cfg" -_CHECK_TIMEOUT_SECONDS = 1800 +CHECK_TIMEOUT_SECONDS = 1800 _CHANGED_KEY = "changed" _ERROR_ANSIBLE_MISSING = "decision_ansible_missing" @@ -65,7 +65,7 @@ _ERROR_INPUT_MISSING = "decision_input_missing" _ERROR_INPUT_TYPE = "decision_input_type" _ERROR_DECISION_EXISTS = "decision_exists" -_ERROR_DECISION_MISSING = "decision_missing" +ERROR_DECISION_MISSING = "decision_missing" _ERROR_NOT_PROPOSED = "decision_not_proposed" _ERROR_APPROVER_UNKNOWN = "decision_approver_unknown" @@ -316,7 +316,7 @@ def update(self, decision: Decision) -> Path: path = self._path(decision.decision_id) if not path.exists(): message = f"decision does not exist: {path}" - raise DecisionError(_ERROR_DECISION_MISSING, message) + raise DecisionError(ERROR_DECISION_MISSING, message) return self._write(path, decision) def load(self, decision_id: ResourceId) -> Decision: @@ -324,7 +324,7 @@ def load(self, decision_id: ResourceId) -> Decision: path = self._path(decision_id) if not path.is_file(): message = f"decision does not exist: {path}" - raise DecisionError(_ERROR_DECISION_MISSING, message) + raise DecisionError(ERROR_DECISION_MISSING, message) document = cast( "dict[str, object]", json.loads(path.read_text(encoding="utf-8")) ) @@ -770,7 +770,7 @@ def _run_check( stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, check=False, - timeout=_CHECK_TIMEOUT_SECONDS, + timeout=CHECK_TIMEOUT_SECONDS, ) return completed.returncode diff --git a/sdk/src/cloudfall/lifecycle.py b/sdk/src/cloudfall/lifecycle.py index f9c773f..4e9e6ef 100644 --- a/sdk/src/cloudfall/lifecycle.py +++ b/sdk/src/cloudfall/lifecycle.py @@ -16,7 +16,7 @@ import sys from dataclasses import dataclass from enum import StrEnum -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Protocol, cast from jsonschema.exceptions import ValidationError @@ -24,14 +24,13 @@ from cloudfall.validation import SchemaCatalog, validate_config if TYPE_CHECKING: - from collections.abc import Mapping + from collections.abc import Callable, Mapping from pathlib import Path from cloudfall.domain import ReleaseId, ResourceId from cloudfall.inventory import ComponentInventory, ServiceInventory _ARTIFACT_SCHEMA = "artifact.schema.json" -_STEP_TIMEOUT_SECONDS = 3600 _OUTPUT_TAIL_CHARACTERS = 2000 _ERROR_COMPONENT_MISSING = "lifecycle_component_missing" _ERROR_ARTIFACT_MISSING = "lifecycle_artifact_missing" @@ -40,7 +39,7 @@ _ERROR_ARTIFACT_DIGEST = "lifecycle_artifact_digest_mismatch" _ERROR_ANSIBLE_MISSING = "lifecycle_ansible_missing" _ERROR_ENGINE_MISSING = "lifecycle_engine_missing" -_ERROR_EXECUTION_FAILED = "lifecycle_execution_failed" +ERROR_EXECUTION_FAILED = "lifecycle_execution_failed" _ERROR_RECEIPT_MISSING = "lifecycle_receipt_missing" _ERROR_SERVICE_MISSING = "lifecycle_service_missing" _ERROR_DATABASE_MISSING = "lifecycle_database_missing" @@ -73,6 +72,48 @@ class ExecutionStep: environment: Mapping[str, str] +class StepRun(Protocol): + """How one finished step reads: ``subprocess`` or treaty's ``ctx.run``.""" + + @property + def returncode(self) -> int: + """The exit code.""" + ... + + @property + def stdout(self) -> str: + """What the step printed on stdout.""" + ... + + @property + def stderr(self) -> str: + """What the step printed on stderr.""" + ... + + +type StepRunner = Callable[[ExecutionStep], StepRun] +"""Runs one step to completion without raising on a non-zero exit.""" + +STEP_TIMEOUT_SECONDS = 3600 +"""How long one step (an inventory render or a playbook run) may take.""" + + +def subprocess_step(step: ExecutionStep) -> StepRun: + """Run one step as a child process, the runner outside treaty commands.""" + try: + return subprocess.run( # noqa: S603 - argv from typed values. + list(step.argv), + check=False, + capture_output=True, + text=True, + timeout=STEP_TIMEOUT_SECONDS, + env={**os.environ, **step.environment}, + ) + except subprocess.TimeoutExpired as error: + detail = f"{step.description} exceeded {STEP_TIMEOUT_SECONDS} seconds" + raise LifecycleError(ERROR_EXECUTION_FAILED, detail) from error + + @dataclass(frozen=True, slots=True) class ExecutionPlan: """Ordered subprocess steps for one lifecycle action.""" @@ -174,6 +215,8 @@ class EngineContext: schema_directory: Path engine_directory: Path inventory_file: Path + run: StepRunner = subprocess_step + """Starts each step: ``subprocess`` by default, ``ctx.run`` under treaty.""" def playbook(self, name: str) -> Path: """Return the path of one engine playbook, requiring it to exist.""" @@ -347,9 +390,9 @@ def plan_health( ) -def execute_plan(plan: ExecutionPlan) -> None: +def execute_plan(plan: ExecutionPlan, run: StepRunner) -> None: """Run every step of a plan, failing fast on the first error.""" - _execute_steps(plan.steps) + _execute_steps(plan.steps, run) def run_engine_playbook( @@ -362,7 +405,8 @@ def run_engine_playbook( ( _render_inventory_step(context), _playbook_step(context, playbook_name, extra_vars), - ) + ), + context.run, ) @@ -394,44 +438,30 @@ def build_release_artifact( argv=argv, environment={}, ) - output = _execute_step_with_output(step) + output = _execute_step_with_output(step, context.run) # The engine answers one treaty envelope: the build is under ``data``. parsed = cast("object", json.loads(output)) if not isinstance(parsed, dict) or parsed.get("ok") is not True: detail = "artifact builder returned no successful envelope" - raise LifecycleError(_ERROR_EXECUTION_FAILED, detail) + raise LifecycleError(ERROR_EXECUTION_FAILED, detail) data = cast("object", parsed.get("data")) if not isinstance(data, dict) or not all(isinstance(key, str) for key in data): detail = "artifact builder returned a non-object payload" - raise LifecycleError(_ERROR_EXECUTION_FAILED, detail) + raise LifecycleError(ERROR_EXECUTION_FAILED, detail) return {"status": "ok", **cast("dict[str, object]", data)} -def _execute_steps(steps: tuple[ExecutionStep, ...]) -> None: +def _execute_steps(steps: tuple[ExecutionStep, ...], run: StepRunner) -> None: for step in steps: - _execute_step_with_output(step) + _execute_step_with_output(step, run) -def _execute_step_with_output(step: ExecutionStep) -> str: - environment = {**os.environ, **step.environment} - try: - completed = subprocess.run( # noqa: S603 - argv from typed values. - list(step.argv), - check=True, - capture_output=True, - text=True, - timeout=_STEP_TIMEOUT_SECONDS, - env=environment, - ) - except subprocess.CalledProcessError as error: - tail = f"{error.stdout}\n{error.stderr}"[-_OUTPUT_TAIL_CHARACTERS:] +def _execute_step_with_output(step: ExecutionStep, run: StepRunner) -> str: + completed = run(step) + if completed.returncode != 0: + tail = f"{completed.stdout}\n{completed.stderr}"[-_OUTPUT_TAIL_CHARACTERS:] detail = f"{step.description} failed: {tail.strip()}" - raise LifecycleError(_ERROR_EXECUTION_FAILED, detail) from error - except subprocess.TimeoutExpired as error: - detail = ( - f"{step.description} exceeded {_STEP_TIMEOUT_SECONDS} seconds" - ) - raise LifecycleError(_ERROR_EXECUTION_FAILED, detail) from error + raise LifecycleError(ERROR_EXECUTION_FAILED, detail) return completed.stdout @@ -455,7 +485,7 @@ def deploy( artifact, resolved_options, ) - execute_plan(plan) + execute_plan(plan, context.run) receipt = None if resolved_options.receipt_directory is not None: receipt = ( @@ -553,7 +583,7 @@ def rollback( """Switch one component back to an existing release, health gated.""" component = _component(context, component_id) plan = plan_rollback(context, component_id, release) - execute_plan(plan) + execute_plan(plan, context.run) return LifecycleResult( action="rollback", component_id=component_id, @@ -571,7 +601,7 @@ def restart( """Restart one component and require its declared health check.""" component = _component(context, component_id) plan = plan_restart(context, component_id) - execute_plan(plan) + execute_plan(plan, context.run) return LifecycleResult( action="restart", component_id=component_id, @@ -592,9 +622,9 @@ def health( healthy = True detail: str | None = None try: - execute_plan(plan) + execute_plan(plan, context.run) except LifecycleError as error: - if error.code != _ERROR_EXECUTION_FAILED: + if error.code != ERROR_EXECUTION_FAILED: raise healthy = False detail = error.detail @@ -654,7 +684,7 @@ def migrate_data( plan = plan_data_migration( context, service_id, database, source_url_file, receipt_directory ) - execute_plan(plan) + execute_plan(plan, context.run) result: dict[str, object] = { "status": "ok", "action": "data-migration", diff --git a/sdk/tests/test_app.py b/sdk/tests/test_app.py index 6edc8ff..c38812d 100644 --- a/sdk/tests/test_app.py +++ b/sdk/tests/test_app.py @@ -57,11 +57,14 @@ def test_a_moved_command_keeps_its_keys_with_status_under_data( def test_a_half_moved_group_sends_the_rest_to_argparse( capsys: pytest.CaptureFixture[str], ) -> None: - code = main(["operations", "propose", "restart", "--repository", str(EXAMPLES)]) + """`operator list` runs on treaty; `operator run` is still argparse's.""" + with pytest.raises(SystemExit) as exit_info: + main(["operator", "run", "--project", str(EXAMPLES)]) error = json.loads(capsys.readouterr().err) - assert code == 2 + assert exit_info.value.code == 2 assert error["status"] == "error" + assert "--gateway-ca" in error["error"]["message"] @pytest.mark.xfail( diff --git a/sdk/tests/test_arguments.py b/sdk/tests/test_arguments.py index 77a6852..c76aec7 100644 --- a/sdk/tests/test_arguments.py +++ b/sdk/tests/test_arguments.py @@ -106,9 +106,7 @@ def test_missing_required_argument_is_a_json_usage_error( @pytest.mark.parametrize( "argv", [ - ["health", "Bad ID!"], ["rollback", "crm-backend", "--release", "r1"], - ["backup", "run", "acme%2Fdb"], ], ) def test_invalid_identifiers_fail_before_dispatch( @@ -124,17 +122,23 @@ def test_invalid_identifiers_fail_before_dispatch( assert "invalid" in str(error["message"]) +@pytest.mark.parametrize( + ("argv", "field"), + [ + (["operator", "show", "../ghost"], "proposal"), + (["health", "Bad ID!"], "component"), + (["backup", "run", "acme%2Fdb"], "service"), + ], +) def test_an_invalid_identifier_fails_before_a_treaty_command_runs( - capsys: pytest.CaptureFixture[str], + argv: list[str], field: str, capsys: pytest.CaptureFixture[str] ) -> None: - payload = _treaty_usage_error( - ["operator", "show", "../ghost", "--project", str(EXAMPLES)], capsys - ) + payload = _treaty_usage_error([*argv, "--project", str(EXAMPLES)], capsys) error = payload["error"] assert isinstance(error, dict) assert error["code"] == "ARG_ERROR" - assert "proposal" in str(error["message"]) + assert field in str(error["message"]) def test_import_render_rejects_an_invalid_application_id( diff --git a/sdk/tests/test_decision.py b/sdk/tests/test_decision.py index 8a2ad2c..6efa6ab 100644 --- a/sdk/tests/test_decision.py +++ b/sdk/tests/test_decision.py @@ -411,9 +411,10 @@ def test_the_cli_reports_a_gate_refusal_as_json( ] ) - payload = json.loads(capsys.readouterr().err) - assert exit_code == 2 - assert payload["error"]["code"] == "decision_target_required" + payload = json.loads(capsys.readouterr().out) + assert exit_code == 4 + assert payload["error"]["code"] == "PRECONDITION" + assert payload["error"]["context"]["code"] == "decision_target_required" def test_the_record_is_readable_without_a_catalog( @@ -624,7 +625,8 @@ def test_the_cli_shows_the_proposal_before_it_runs_anything( payload = json.loads(capsys.readouterr().out) assert exit_code == 0 - assert payload["status"] == "pending" + assert payload["data"]["status"] == "pending" + assert payload["data"]["effect"] == "noop" assert payload["data"]["decision"]["spec"]["status"] == "proposed" assert store.load(proposed.decision_id).status is DecisionStatus.PROPOSED diff --git a/sdk/tests/test_migrate.py b/sdk/tests/test_migrate.py index ebc399c..b5170a1 100644 --- a/sdk/tests/test_migrate.py +++ b/sdk/tests/test_migrate.py @@ -221,7 +221,8 @@ def test_cli_migrate_previews_without_executing( captured = capsys.readouterr() payload = json.loads(captured.out) assert exit_code == 0 - assert payload["status"] == "plan" + assert payload["data"]["status"] == "plan" + assert payload["data"]["effect"] == "noop" assert payload["data"]["next"] == "baseline" @@ -243,7 +244,9 @@ def test_cli_migrate_rejects_malformed_component_pairs( ] ) - captured = capsys.readouterr() - payload = json.loads(captured.err) + payload = json.loads(capsys.readouterr().out) assert exit_code == 2 - assert payload["error"]["code"] == "invalid_argument" + assert payload["error"]["code"] == "ARG_ERROR" + assert "COMPONENT" in payload["error"]["message"] or "NAME=VALUE" in ( + payload["error"]["message"] + )