diff --git a/AGENTS.md b/AGENTS.md index ffa6b0fd..633882e0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1734,8 +1734,11 @@ trusts anything in `fcvm-fleet`, and no peering, shared subnet or resource polic software release pointer and tracks the latest active task revision; no task/service fields are ignored. Before its first ECS mutation the function increments the pointer's parameter version. An ephemeral apply-time check rejects a saved plan made before that increment, - including when a failed release restores the previous image. Infrastructure still changes - only through Terraform from a jumpbox. + including when a failed release restores the previous image. A failed release returns only + `status`, `errorType`, the `step` that failed and, for an AWS error or a failed release check, + its `code`; never a message or an SDK payload. A lock the function no longer holds at the end + (deleted, or replaced after a force-unlock) is left alone, and the release's own result stands. + Infrastructure still changes only through Terraform from a jumpbox. - **GPU hosts** (ejc3/fcvm#1257): no auto scaling group. The coordinator launches each host itself, spot first (an instant EC2 Fleet over the region's host templates and host subnets, `FLEET_HOST_TEMPLATES`, `FLEET_HOST_SUBNETS`), on demand in the same pass when spot is refused for capacity, at most `local.fcvm_fleet_on_demand_max` = 8 on-demand hosts across diff --git a/fcvm-coordinator-release/release.py b/fcvm-coordinator-release/release.py index 4b7cb43d..62ffba28 100644 --- a/fcvm-coordinator-release/release.py +++ b/fcvm-coordinator-release/release.py @@ -11,6 +11,32 @@ from botocore.config import Config from botocore.exceptions import ClientError +CONDITION_FAILED = "ConditionalCheckFailedException" + + +class Refused(RuntimeError): + """A release check that failed; `code` names the check in the failure result.""" + + def __init__(self, code, message): + super().__init__(message) + self.code = code + + +def error_code(error): + """The AWS error code or the release check that failed, never a message or a payload.""" + if isinstance(error, ClientError): + code = error.response.get("Error", {}).get("Code", "") + elif isinstance(error, Refused): + code = error.code + else: + return None + return code if isinstance(code, str) and re.fullmatch(r"[A-Za-z0-9.:_-]{1,64}", code) else None + + +def lock_holder(error): + """The lock's Info when a conditional write was refused (ALL_OLD), or None when no lock exists.""" + return error.response.get("Item", {}).get("Info", {}).get("S") + def task_input(response, image): """Preserve the service's task configuration; replace exactly one container image.""" @@ -43,7 +69,7 @@ def image_of(response): def service(ecs): result = ecs.describe_services(cluster="fcvm-fleet", services=["fcvm-fleet-coordinator"]) if result.get("failures") or len(result.get("services", [])) != 1: - raise RuntimeError("staging service unavailable") + raise Refused("service-unavailable", "staging service unavailable") return result["services"][0] @@ -51,10 +77,10 @@ def wait_for(ecs, arn, deadline): while time.monotonic() < deadline: current = service(ecs) if current["taskDefinition"] != arn: - raise RuntimeError("staging service changed during release") + raise Refused("service-changed", "staging service changed during release") deployments = current.get("deployments", []) if any(d.get("rolloutState") == "FAILED" for d in deployments): - raise RuntimeError("ECS rollout failed") + raise Refused("rollout-failed", "ECS rollout failed") if (len(deployments) == 1 and deployments[0].get("rolloutState") == "COMPLETED" and current["runningCount"] == current["desiredCount"] == 1 and current["pendingCount"] == 0): @@ -63,7 +89,10 @@ def wait_for(ecs, arn, deadline): raise TimeoutError("ECS rollout did not become stable") -def deploy(event, ddb, ecs, ssm, ecr, *, remaining_ms=lambda: 850_000): +def deploy(event, ddb, ecs, ssm, ecr, *, remaining_ms=lambda: 850_000, progress=None): + """Release one image. `progress["step"]` names the step that raised, for the failure result.""" + progress = {} if progress is None else progress + progress["step"] = "input" commit, digest = event.get("commit", ""), event.get("digest", "") if not re.fullmatch(r"[0-9a-f]{40}", commit) or not re.fullmatch(r"sha256:[0-9a-f]{64}", digest): raise ValueError("a full commit and image digest are required") @@ -77,67 +106,105 @@ def deploy(event, ddb, ecs, ssm, ecr, *, remaining_ms=lambda: 850_000): info = json.dumps({"ID": token, "Operation": "fcvm coordinator release", "Info": commit, "Who": "fleet-stage-ci", "Version": "1", "Path": lock_id, "Created": datetime.datetime.now(datetime.timezone.utc).isoformat()}) + progress["step"] = "lock" try: + # ALL_OLD: a refusal carries the lock it was checked against, so a retry of a write that + # already landed (the SDK retries a timed-out request) can tell its own lock from another's. ddb.put_item(TableName=table, Item={"LockID": {"S": lock_id}, "Info": {"S": info}}, - ConditionExpression="attribute_not_exists(LockID)") + ConditionExpression="attribute_not_exists(LockID)", + ReturnValuesOnConditionCheckFailure="ALL_OLD") except ClientError as error: - if error.response["Error"]["Code"] == "ConditionalCheckFailedException": + if error_code(error) != CONDITION_FAILED: + raise + if lock_holder(error) != info: return {"status": "busy"} - raise + # Our own token: an earlier attempt of this request took the lock and its reply was lost. new_arn, previous = None, None try: + progress["step"] = "inspect" current = service(ecs) if current["desiredCount"] != 1 or len(current.get("deployments", [])) != 1: - raise RuntimeError("staging is disabled or already rolling out") + raise Refused("not-idle", "staging is disabled or already rolling out") old_arn = current["taskDefinition"] old = ecs.describe_task_definition(taskDefinition=old_arn, include=["TAGS"]) previous = ssm.get_parameter(Name=parameter)["Parameter"]["Value"] if image_of(old) != previous: - raise RuntimeError("release pointer and running task disagree") + raise Refused("pointer-mismatch", "release pointer and running task disagree") if previous == image: + progress["step"] = "rollout" wait_for(ecs, old_arn, time.monotonic() + 60) return {"status": "unchanged", "taskDefinition": old_arn, "digest": digest} definition = task_input(old, image) # Bump the parameter version BEFORE the first ECS change. The ephemeral # Terraform guard then rejects older saved plans, even after rollback or # a failed registration that leaves the image value unchanged. + progress["step"] = "register" ssm.put_parameter(Name=parameter, Value=previous, Type="String", Overwrite=True) new_arn = ecs.register_task_definition(**definition)["taskDefinition"]["taskDefinitionArn"] + progress["step"] = "update" ecs.update_service(cluster="fcvm-fleet", service="fcvm-fleet-coordinator", taskDefinition=new_arn) # Reserve time for a rollback and releasing the backend lock on every invocation. + progress["step"] = "rollout" wait_for(ecs, new_arn, time.monotonic() + min(480, max(5, remaining_ms() / 1000 - 240))) + progress["step"] = "pointer" ssm.put_parameter(Name=parameter, Value=image, Type="String", Overwrite=True) return {"status": "deployed", "taskDefinition": new_arn, "digest": digest} except Exception: if new_arn: + failed_step, progress["step"] = progress["step"], "rollback" # Never overwrite a deployment another operator made while we were running. if service(ecs)["taskDefinition"] in (old_arn, new_arn): ecs.update_service(cluster="fcvm-fleet", service="fcvm-fleet-coordinator", taskDefinition=old_arn) wait_for(ecs, old_arn, time.monotonic() + 150) ssm.put_parameter(Name=parameter, Value=previous, Type="String", Overwrite=True) ecs.deregister_task_definition(taskDefinition=new_arn) + progress["step"] = failed_step raise finally: - # The exact Info value is the ownership token; a stale invocation cannot unlock another owner. + unlock(ddb, table, lock_id, info, progress) + + +def unlock(ddb, table, lock_id, info, progress): + """Delete the lock only while it holds this release's exact Info, the ownership token. + + A refusal means the lock is not ours: an earlier attempt of this same delete removed it and its + reply was lost, or it was force-released (and possibly taken) by someone else. Either way there + is nothing of ours left to delete, and the release's own outcome stands. + """ + failed_step, progress["step"] = progress.get("step"), "unlock" + try: ddb.delete_item(TableName=table, Key={"LockID": {"S": lock_id}}, ConditionExpression="#info = :owner", ExpressionAttributeNames={"#info": "Info"}, - ExpressionAttributeValues={":owner": {"S": info}}) + ExpressionAttributeValues={":owner": {"S": info}}, + ReturnValuesOnConditionCheckFailure="ALL_OLD") + except ClientError as error: + if error_code(error) != CONDITION_FAILED: + raise + print(json.dumps({"step": "unlock", "code": CONDITION_FAILED, + "lock": "held by another owner" if lock_holder(error) else "absent"})) + progress["step"] = failed_step def lambda_handler(event, context): - config = Config(connect_timeout=5, read_timeout=15, retries={"max_attempts": 2}) - credentials = boto3.client("sts", config=config).assume_role( - RoleArn=os.environ["RELEASE_ROLE"], RoleSessionName="fcvm-stage-release")["Credentials"] - session = boto3.Session(aws_access_key_id=credentials["AccessKeyId"], - aws_secret_access_key=credentials["SecretAccessKey"], - aws_session_token=credentials["SessionToken"], region_name="us-east-2") + progress = {"step": "credentials"} try: + config = Config(connect_timeout=5, read_timeout=15, retries={"max_attempts": 2}) + credentials = boto3.client("sts", config=config).assume_role( + RoleArn=os.environ["RELEASE_ROLE"], RoleSessionName="fcvm-stage-release")["Credentials"] + session = boto3.Session(aws_access_key_id=credentials["AccessKeyId"], + aws_secret_access_key=credentials["SecretAccessKey"], + aws_session_token=credentials["SessionToken"], region_name="us-east-2") result = deploy(event, boto3.client("dynamodb", config=config), session.client("ecs", config=config), session.client("ssm", config=config), session.client("ecr", config=config), - remaining_ms=context.get_remaining_time_in_millis) + remaining_ms=context.get_remaining_time_in_millis, progress=progress) print(json.dumps(result)) return result except Exception as error: - # Never return SDK responses or task definitions: configuration can contain credentials. - print(json.dumps({"status": "failed", "errorType": type(error).__name__})) - return {"status": "failed", "errorType": type(error).__name__} + # Never return messages, SDK responses or task definitions: configuration can contain + # credentials. The step and the AWS error code (or the release check's code) are fixed tokens. + result = {"status": "failed", "errorType": type(error).__name__, "step": progress["step"]} + code = error_code(error) + if code: + result["code"] = code + print(json.dumps(result)) + return result diff --git a/scripts/test-fcvm-coordinator-release.py b/scripts/test-fcvm-coordinator-release.py index 309d5c4f..48ecfd54 100644 --- a/scripts/test-fcvm-coordinator-release.py +++ b/scripts/test-fcvm-coordinator-release.py @@ -56,16 +56,23 @@ def __init__(self): def describe_images(self, **kwargs): return {"imageDetails": [{"imageDigest": DIGEST}]} + def refused(self, kwargs): + """DynamoDB's refusal; with ALL_OLD it carries the item the condition was checked against.""" + response = {"Error": {"Code": "ConditionalCheckFailedException", "Message": "secret detail"}} + if kwargs.get("ReturnValuesOnConditionCheckFailure") == "ALL_OLD" and self.lock: + response["Item"] = {"LockID": {"S": ENV["LOCK_ID"]}, "Info": {"S": self.lock}} + return ClientError(response) + def put_item(self, **kwargs): if self.lock and kwargs.get("ConditionExpression") == "attribute_not_exists(LockID)": - raise ClientError({"Error": {"Code": "ConditionalCheckFailedException"}}) + raise self.refused(kwargs) self.lock = kwargs["Item"]["Info"]["S"] self.calls.append("lock") def delete_item(self, **kwargs): if (kwargs.get("ConditionExpression") == "#info = :owner" and kwargs["ExpressionAttributeValues"][":owner"]["S"] != self.lock): - raise ClientError({"Error": {"Code": "ConditionalCheckFailedException"}}) + raise self.refused(kwargs) self.lock = None self.calls.append("unlock") @@ -186,15 +193,135 @@ def test_a_failed_rollout_restores_the_previous_task_and_pointer(self): self.assertEqual(self.aws.image, old) self.assertEqual(self.aws.calls, ["lock", "pointer", "register", NEW, OLD, "pointer", "deregister:" + NEW, "unlock"]) - def test_the_lock_delete_checks_the_unique_owner(self): + def test_a_lock_taken_over_during_the_release_is_left_alone_and_the_release_reports_deployed(self): real = self.aws.update_service def changed(**kwargs): real(**kwargs) - self.aws.lock = "new owner" + self.aws.lock = "new owner" # force-released and taken by Terraform while the rollout ran self.aws.update_service = changed - with self.assertRaises(ClientError): - self.run_release() + self.assertEqual(self.run_release()["status"], "deployed") self.assertEqual(self.aws.lock, "new owner") + self.assertEqual(self.aws.image, REPO + "@" + DIGEST) + + def test_a_retried_unlock_whose_first_attempt_landed_reports_deployed(self): + real = self.aws.delete_item + def retried(**kwargs): + real(**kwargs) # the first attempt deletes the lock, its reply is lost, the SDK retries + real(**kwargs) + self.aws.delete_item = retried + result = self.run_release() + self.assertEqual(result["status"], "deployed") + self.assertIsNone(self.aws.lock) + + def test_a_retried_lock_write_whose_first_attempt_landed_owns_the_lock(self): + real = self.aws.put_item + def retried(**kwargs): + real(**kwargs) # the first attempt writes the lock, its reply is lost, the SDK retries + real(**kwargs) + self.aws.put_item = retried + self.assertEqual(self.run_release()["status"], "deployed") + self.assertIsNone(self.aws.lock, "the release left its own lock behind") + + def test_a_failed_unlock_of_a_lock_still_held_is_reported(self): + def throttled(**kwargs): + raise ClientError({"Error": {"Code": "ProvisionedThroughputExceededException"}}) + self.aws.delete_item = throttled + progress = {} + with self.assertRaises(ClientError): + release.deploy(EVENT, self.aws, self.aws, self.aws, self.aws, progress=progress) + self.assertEqual(progress["step"], "unlock") + + +class Session: + def __init__(self, aws): + self.aws = aws + + def client(self, name, config=None): + return self.aws + + +class HandlerTests(unittest.TestCase): + """The failure result names the step and the AWS error code, and nothing else.""" + + def setUp(self): + self.aws = AWS() + self.aws.assume_role = lambda **kwargs: {"Credentials": { + "AccessKeyId": "id", "SecretAccessKey": "secret", "SessionToken": "token"}} + fake = types.SimpleNamespace(client=lambda name, config=None: self.aws, + Session=lambda **kwargs: Session(self.aws)) + self.patches = [patch.object(release, "boto3", fake), patch.object(release, "Config", lambda **kwargs: None), + patch.dict(os.environ, {**ENV, "RELEASE_ROLE": "arn:example:role/release"})] + for p in self.patches: + p.start() + self.addCleanup(p.stop) + self.context = types.SimpleNamespace(get_remaining_time_in_millis=lambda: 850_000) + + def break_call(self, method, code="AccessDeniedException"): + def raises(**kwargs): + raise ClientError({"Error": {"Code": code, "Message": "secret detail"}, + "ResponseMetadata": {"HTTPHeaders": {"x-secret": "value"}}}) + setattr(self.aws, method, raises) + + def handle(self): + result = release.lambda_handler(EVENT, self.context) + self.assertNotIn("secret", json.dumps(result)) + return result + + def test_each_failing_step_is_named_with_its_code(self): + cases = [("assume_role", "credentials"), ("put_item", "lock"), + ("describe_task_definition", "inspect"), ("register_task_definition", "register"), + ("delete_item", "unlock")] + for method, step in cases: + with self.subTest(step=step): + self.setUp() + self.break_call(method) + self.assertEqual(self.handle(), {"status": "failed", "errorType": "ClientError", + "step": step, "code": "AccessDeniedException"}) + + def test_a_failed_service_update_is_named_after_its_rollback(self): + real = self.aws.update_service + def forward_update_fails(**kwargs): + real(**kwargs) + if kwargs["taskDefinition"] == NEW: + raise ClientError({"Error": {"Code": "InvalidParameterException", "Message": "secret detail"}}) + self.aws.update_service = forward_update_fails + self.assertEqual(self.handle(), {"status": "failed", "errorType": "ClientError", + "step": "update", "code": "InvalidParameterException"}) + self.assertEqual(self.aws.arn, OLD) + + def test_the_failed_pointer_write_is_named(self): + real = self.aws.put_parameter + def second_write_fails(**kwargs): + if kwargs["Value"] == REPO + "@" + DIGEST: + raise ClientError({"Error": {"Code": "ThrottlingException", "Message": "secret detail"}}) + real(**kwargs) + self.aws.put_parameter = second_write_fails + self.assertEqual(self.handle(), {"status": "failed", "errorType": "ClientError", + "step": "pointer", "code": "ThrottlingException"}) + + def test_a_failed_rollout_is_named_after_a_successful_rollback(self): + self.aws.fail_rollout = True + self.assertEqual(self.handle(), {"status": "failed", "errorType": "Refused", + "step": "rollout", "code": "rollout-failed"}) + self.assertEqual(self.aws.arn, OLD) + + def test_a_failed_rollback_is_named(self): + self.aws.fail_rollout = True + real = self.aws.update_service + def rollback_fails(**kwargs): + if kwargs["taskDefinition"] == OLD: + raise ClientError({"Error": {"Code": "ServiceNotActiveException"}}) + real(**kwargs) + self.aws.update_service = rollback_fails + self.assertEqual(self.handle(), {"status": "failed", "errorType": "ClientError", + "step": "rollback", "code": "ServiceNotActiveException"}) + + def test_an_error_without_a_code_names_only_its_step(self): + self.aws.describe_task_definition = lambda **kwargs: (_ for _ in ()).throw(TimeoutError("secret detail")) + self.assertEqual(self.handle(), {"status": "failed", "errorType": "TimeoutError", "step": "inspect"}) + + def test_a_completed_release_reports_deployed(self): + self.assertEqual(self.handle()["status"], "deployed") if __name__ == "__main__":