Coverage for scripts / live_release_validation / actions / deploy.py: 100.00%

1"""deploy: deploy the configured GCO topology.""" 

2 

3from __future__ import annotations 

4 

5import time 

6from collections.abc import Mapping 

7from typing import Any 

8 

9from ..cleanup.local_images import prune_local_cdk_asset_images_safely 

10from ..constants import ( 

11 _RUN_STACK_TAG, 

12) 

13from ..inventory import ( 

14 describe_stack, 

15) 

16from ..models import RunContext 

17from ..ownership.ecr import ( 

18 _checkpoint_new_ecr_images, 

19 _checkpoint_new_ecr_repositories, 

20 _record_ecr_repository_creation, 

21) 

22from ..ownership.kms import ( 

23 _checkpoint_retained_kms_keys, 

24) 

25from ..ownership.stacks import ( 

26 _authorize_owned_stack, 

27 _owned_stack_record, 

28 _prepared_change_set_authority, 

29 _reconcile_stack_ownership, 

30 _record_prepared_stack_identity, 

31 _record_stack_identity, 

32) 

33 

34 

35def action_deploy(ctx: RunContext) -> dict[str, Any]: 

36 """Deploy the exact checked-out CDK graph and checkpoint every AWS identity.""" 

37 if ctx.checkpoint.baseline is None: 

38 raise RuntimeError("A protected-resource baseline is required before deployment") 

39 ctx.checkpoint.deployment_attempted = True 

40 ctx.checkpoint.destroyed = False 

41 ctx.persist() 

42 

43 events: list[dict[str, Any]] = list(ctx.checkpoint.state.get("deploy_events", [])) 

44 

45 def on_start(stack_name: str) -> None: 

46 with ctx.state_lock: 

47 events.append({"stack": stack_name, "event": "started", "at": time.time()}) 

48 ctx.checkpoint.state["deploy_events"] = events 

49 ctx.persist_callback(ctx.checkpoint) 

50 

51 def on_complete(stack_name: str, success: bool) -> None: 

52 with ctx.state_lock: 

53 events.append( 

54 { 

55 "stack": stack_name, 

56 "event": "completed", 

57 "success": success, 

58 "at": time.time(), 

59 } 

60 ) 

61 ctx.checkpoint.state["deploy_events"] = events 

62 region = str(ctx.checkpoint.state["target_stack_regions"][stack_name]) 

63 stack = describe_stack(ctx.session, region, stack_name) 

64 if stack is None and success: 

65 raise RuntimeError(f"CDK reported success but {region}:{stack_name} is absent") 

66 if stack is not None: 

67 _record_stack_identity(ctx, stack_name, region, stack) 

68 ctx.persist_callback(ctx.checkpoint) 

69 

70 def on_prepared( 

71 stack_name: str, 

72 region: str, 

73 stack_id: str, 

74 change_set_id: str, 

75 change_set_type: str, 

76 ) -> None: 

77 _record_prepared_stack_identity( 

78 ctx, 

79 stack_name, 

80 region, 

81 stack_id, 

82 change_set_id, 

83 change_set_type, 

84 ) 

85 prepared_change_sets.setdefault(stack_name, {})[change_set_id] = { 

86 "change_set_id": change_set_id, 

87 "stack_id": stack_id, 

88 "change_set_type": change_set_type, 

89 } 

90 expected_stack_ids[stack_name] = stack_id 

91 

92 def on_repository_created(region: str, repository: Mapping[str, Any]) -> None: 

93 _record_ecr_repository_creation(ctx, region, repository) 

94 

95 expected_stack_ids = { 

96 name: ( 

97 str(record["stack_id"]) 

98 if (record := _owned_stack_record(ctx, str(region), name)) is not None 

99 else None 

100 ) 

101 for name, region in ctx.checkpoint.state["target_stack_regions"].items() 

102 } 

103 prepared_change_sets = _prepared_change_set_authority(ctx) 

104 

105 try: 

106 overall, successful, failed = ctx.stack_manager.deploy_orchestrated( 

107 require_approval=False, 

108 tags={_RUN_STACK_TAG: ctx.settings.run_id}, 

109 progress="events", 

110 on_stack_start=on_start, 

111 on_stack_complete=on_complete, 

112 parallel=False, 

113 max_workers=1, 

114 allow_bootstrap=False, 

115 bootstrap_stacks=ctx.checkpoint.state["bootstrap_stacks"], 

116 expected_stack_ids=expected_stack_ids, 

117 prepared_change_sets=prepared_change_sets, 

118 authorize_stack=lambda name, region, stack_id: _authorize_owned_stack( 

119 ctx, 

120 name, 

121 region, 

122 stack_id, 

123 ), 

124 strict_deployment_token=ctx.settings.run_id, 

125 on_change_set_prepared=on_prepared, 

126 on_ecr_repository_created=on_repository_created, 

127 ) 

128 finally: 

129 _reconcile_stack_ownership(ctx) 

130 _checkpoint_new_ecr_repositories(ctx) 

131 _checkpoint_new_ecr_images(ctx) 

132 _checkpoint_retained_kms_keys(ctx) 

133 # Published asset images have no further local use; reclaim the disk 

134 # they occupy so successive runs cannot fill the host (which fails an 

135 # image build mid-deploy and then the checkpoint write cleanup needs). 

136 local_image_prune = prune_local_cdk_asset_images_safely() 

137 ctx.checkpoint.state["local_image_prune"] = local_image_prune 

138 

139 result = { 

140 "overall_success": overall, 

141 "successful_stacks": successful, 

142 "failed_stacks": failed, 

143 "events": events, 

144 "local_image_prune": local_image_prune, 

145 "owned_stacks": ctx.checkpoint.state.get("owned_stacks", {}), 

146 "owned_ecr_repositories": ctx.checkpoint.state.get("created_ecr_repositories", []), 

147 "owned_ecr_images": [], 

148 "retained_ecr_image_deltas": ctx.checkpoint.state.get("retained_ecr_image_deltas", []), 

149 "owned_kms_keys": ctx.checkpoint.state.get("owned_kms_keys", []), 

150 } 

151 ctx.checkpoint.state["deploy_result"] = result 

152 ctx.persist() 

153 if not overall: 

154 raise RuntimeError(f"Orchestrated deployment failed for: {', '.join(failed) or 'unknown'}") 

155 return result