diff --git a/fleet/coord/publish.py b/fleet/coord/publish.py index 0bcd94344..5c3e7e4bc 100644 --- a/fleet/coord/publish.py +++ b/fleet/coord/publish.py @@ -8,16 +8,29 @@ import time from pathlib import Path +TOKEN = re.compile(r"[A-Za-z0-9.:_-]{1,64}") + + +class CommandFailed(RuntimeError): + def __init__(self, name, returncode): + super().__init__("command failed: " + name) + self.returncode = returncode + def command(args, *, body=None, env=None, allow_missing=False): result = subprocess.run(args, input=body, capture_output=True, text=True, env=env) if result.returncode: if allow_missing and "ImageNotFoundException" in result.stderr: return None - raise RuntimeError("command failed: " + args[0]) + raise CommandFailed(args[0], result.returncode) return result.stdout +def attribution(result): + """The release function's failed step and AWS error code: the only fields of its result repeated here.""" + return {k: result[k] for k in ("step", "code") if isinstance(result.get(k), str) and TOKEN.fullmatch(result[k])} + + def main(): commit, repo = os.environ["GITHUB_SHA"], os.environ["COORDINATOR_REPO"] if not re.fullmatch(r"[0-9a-f]{40}", commit): @@ -49,9 +62,20 @@ def main(): with tempfile.TemporaryDirectory(prefix="coordinator-release-") as tmp: output = Path(tmp) / "result.json" # Only status, task ARN and image digest; no credentials. for attempt in range(11): - meta = json.loads(command(["aws", "lambda", "invoke", "--region", "us-west-1", - "--function-name", os.environ["COORDINATOR_RELEASE"], "--cli-binary-format", "raw-in-base64-out", - "--payload", "file:///dev/stdin", "--cli-read-timeout", "910", str(output)], body=payload)) + try: + meta = json.loads(command(["aws", "lambda", "invoke", "--region", "us-west-1", + "--function-name", os.environ["COORDINATOR_RELEASE"], "--cli-binary-format", "raw-in-base64-out", + "--payload", "file:///dev/stdin", "--cli-read-timeout", "910", str(output)], body=payload)) + except CommandFailed as error: + # The call can end while the function keeps running (its connection was dropped, then the + # CLI's own retries were throttled by the function's single reserved slot). The release is + # serialized and idempotent: a later invocation reports busy, then the finished release. + if attempt == 10: + raise + print(f"The release invocation ended without a result (aws exit {error.returncode}); " + "retrying in 60 seconds", flush=True) + time.sleep(60) + continue if meta.get("FunctionError"): raise RuntimeError("release function failed") result = json.loads(output.read_text()) @@ -61,6 +85,8 @@ def main(): time.sleep(60) continue if status not in ("deployed", "unchanged") or result.get("digest") != images[0]["imageDigest"]: + if attribution(result): + print(json.dumps({"release": attribution(result)}), flush=True) raise RuntimeError("staging release did not complete") print(json.dumps({"status": status, "commit": commit, "digest": result["digest"]})) return diff --git a/fleet/coord/test_release.py b/fleet/coord/test_release.py index 2b8e3dc99..5f4193fbe 100644 --- a/fleet/coord/test_release.py +++ b/fleet/coord/test_release.py @@ -1,7 +1,9 @@ #!/usr/bin/env python3 """Offline checks for publishing and the live IPv4/name-regression probe.""" import importlib.util +import contextlib import fnmatch +import io import copy import json import os @@ -275,6 +277,61 @@ def test_malformed_commit_or_repository_never_runs_a_command(self): command.assert_not_called() +class ReleaseInvocationTests(unittest.TestCase): + """The Lambda invocation: what a failed release prints, and a call that ends without a result.""" + + DIGEST = "sha256:" + "c" * 64 + + def run_publish(self, results): + """Run main() with each `aws lambda invoke` taking the next entry: a result dict, or an exit status.""" + results, sleeps = list(results), [] + def command(args, body=None, env=None, allow_missing=False): + if args[1] == "ecr": + return json.dumps({"imageDetails": [{"imageDigest": self.DIGEST}]}) + outcome = results.pop(0) + if isinstance(outcome, int): + raise publish.CommandFailed(args[0], outcome) + Path(args[-1]).write_text(json.dumps(outcome)) + return json.dumps({"StatusCode": 200}) + env = {"GITHUB_SHA": "a" * 40, "COORDINATOR_REPO": "111122223333.dkr.ecr.us-east-2.amazonaws.com/fcvm-fleet/coordinator", + "COORDINATOR_RELEASE": "fcvm-coordinator-release"} + out = io.StringIO() + with patch.dict(os.environ, env), patch.object(publish, "command", command), \ + patch.object(publish.time, "sleep", sleeps.append), contextlib.redirect_stdout(out): + try: + publish.main() + error = None + except Exception as caught: + error = caught + self.assertEqual(results, [], "not every scripted invocation ran") + return out.getvalue(), error, sleeps + + def test_a_failed_release_prints_its_step_and_code_and_nothing_else(self): + printed, error, _ = self.run_publish([{ + "status": "failed", "errorType": "ClientError", "step": "unlock", + "code": "ConditionalCheckFailedException", "message": "secret detail", "taskDefinition": "secret"}]) + self.assertIsInstance(error, RuntimeError) + self.assertEqual([json.loads(line) for line in printed.splitlines()], + [{"release": {"step": "unlock", "code": "ConditionalCheckFailedException"}}]) + + def test_a_field_that_is_not_a_plain_token_is_not_printed(self): + printed, error, _ = self.run_publish([{"status": "failed", "step": "rollout", "code": "has spaces: secret"}]) + self.assertIsInstance(error, RuntimeError) + self.assertEqual([json.loads(line) for line in printed.splitlines()], [{"release": {"step": "rollout"}}]) + + def test_an_invocation_that_ends_without_a_result_is_retried_until_the_release_reports(self): + printed, error, sleeps = self.run_publish( + [255, 254, {"status": "busy"}, {"status": "unchanged", "digest": self.DIGEST}]) + self.assertIsNone(error) + self.assertEqual(sleeps, [60, 60, 60]) + self.assertEqual(json.loads(printed.splitlines()[-1])["status"], "unchanged") + + def test_invocations_that_never_return_a_result_still_fail(self): + printed, error, sleeps = self.run_publish([255] * 11) + self.assertIsInstance(error, publish.CommandFailed) + self.assertEqual(len(sleeps), 10) + + class WorkflowTests(unittest.TestCase): def test_changes_to_the_actual_cli_files_trigger_a_release(self): workflow = (HERE.parent.parent / ".github/workflows/fleet-coordinator-deploy.yml").read_text()