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

1"""topology: verify stacks, EKS, HTTPS ALB targets, APIs, queues, and DynamoDB.""" 

2 

3from __future__ import annotations 

4 

5import json 

6from typing import Any 

7 

8from ..checks.alb_tls import _alb_https_target_evidence 

9from ..checks.topology import ( 

10 _bounded_topology_evidence, 

11 _converge_region_addons, 

12 _health_stability_samples, 

13 _health_warmup_samples, 

14 _metrics_reachability_samples, 

15 _queue_counts, 

16) 

17from ..constants import ( 

18 _HEALTHY_STACK_STATUSES, 

19) 

20from ..context import ( 

21 _direct_regional_access_enabled, 

22) 

23from ..inventory import ( 

24 describe_stack, 

25) 

26from ..models import RunContext, utc_now 

27from ..ownership.stacks import ( 

28 _reconcile_stack_ownership, 

29 _record_stack_identity, 

30) 

31 

32 

33def action_topology(ctx: RunContext) -> dict[str, Any]: 

34 """Verify deterministic add-on convergence before stable API and data-plane health.""" 

35 _reconcile_stack_ownership(ctx) 

36 stack_details: dict[str, Any] = {} 

37 target_stack_regions = ctx.checkpoint.state["target_stack_regions"] 

38 for stack_name, region in target_stack_regions.items(): 

39 stack = describe_stack(ctx.session, str(region), stack_name) 

40 if stack is None: 

41 raise RuntimeError(f"Expected deployed stack is absent: {stack_name} ({region})") 

42 _record_stack_identity(ctx, stack_name, str(region), stack) 

43 if stack["status"] not in _HEALTHY_STACK_STATUSES: 

44 raise RuntimeError( 

45 f"Stack {stack_name} is {stack['status']}, expected one of " 

46 f"{sorted(_HEALTHY_STACK_STATUSES)}" 

47 ) 

48 stack_details[stack_name] = stack 

49 

50 convergence: dict[str, Any] = { 

51 "started_at": utc_now(), 

52 "status": "running", 

53 "regions": {}, 

54 } 

55 ctx.checkpoint.state["topology_convergence"] = convergence 

56 ctx.persist() 

57 for region in ctx.deployment_regions: 

58 stack_name = f"{ctx.config.project_name}-{region}" 

59 evidence: dict[str, Any] = { 

60 "region": region, 

61 "stack_name": stack_name, 

62 "result": "pending", 

63 "observations": [], 

64 } 

65 convergence["regions"][region] = evidence 

66 try: 

67 if target_stack_regions.get(stack_name) != region: 

68 raise RuntimeError( 

69 f"Checkpoint does not bind exact regional stack {stack_name} to {region}" 

70 ) 

71 stack = stack_details.get(stack_name) 

72 if not isinstance(stack, dict): 

73 raise RuntimeError(f"Exact regional stack was not described: {region}:{stack_name}") 

74 _converge_region_addons( 

75 ctx, 

76 region=region, 

77 stack_name=stack_name, 

78 stack=stack, 

79 evidence=evidence, 

80 ) 

81 except Exception as exc: 

82 evidence["result"] = "failed" 

83 evidence["completed_at"] = utc_now() 

84 evidence["error"] = _bounded_topology_evidence(f"{type(exc).__name__}: {exc}") 

85 convergence["status"] = "failed" 

86 convergence["completed_at"] = utc_now() 

87 ctx.persist() 

88 raise 

89 evidence["result"] = "succeeded" 

90 evidence["completed_at"] = utc_now() 

91 ctx.persist() 

92 convergence["status"] = "succeeded" 

93 convergence["completed_at"] = utc_now() 

94 ctx.persist() 

95 

96 clusters: dict[str, Any] = {} 

97 for region in ctx.deployment_regions: 

98 name = f"{ctx.config.project_name}-{region}" 

99 cluster = ctx.session.client("eks", region_name=region).describe_cluster(name=name)[ 

100 "cluster" 

101 ] 

102 if cluster.get("status") != "ACTIVE": 

103 raise RuntimeError(f"EKS cluster {name} is not ACTIVE: {cluster.get('status')}") 

104 clusters[region] = { 

105 "name": name, 

106 "arn": cluster.get("arn"), 

107 "status": cluster.get("status"), 

108 "version": cluster.get("version"), 

109 "endpoint_public_access": (cluster.get("resourcesVpcConfig") or {}).get( 

110 "endpointPublicAccess" 

111 ), 

112 "endpoint_private_access": (cluster.get("resourcesVpcConfig") or {}).get( 

113 "endpointPrivateAccess" 

114 ), 

115 } 

116 

117 alb_https_targets = { 

118 region: _alb_https_target_evidence( 

119 ctx, 

120 region=region, 

121 cluster_name=f"{ctx.config.project_name}-{region}", 

122 ) 

123 for region in ctx.deployment_regions 

124 } 

125 

126 global_endpoint = ctx.aws_client.get_api_endpoint(force_refresh=True) 

127 global_url = str(getattr(global_endpoint, "url", "") or "") 

128 if not global_url: 

129 raise RuntimeError("Global API endpoint has no URL") 

130 direct_regional_access = _direct_regional_access_enabled(ctx) 

131 regional_urls: dict[str, str] = {} 

132 if direct_regional_access: 

133 for region in ctx.deployment_regions: 

134 endpoint = ctx.aws_client.get_regional_api_endpoint(region, force_refresh=True) 

135 endpoint_url = str(getattr(endpoint, "url", "") or "") 

136 if not endpoint_url: 

137 raise RuntimeError(f"Direct regional API endpoint is absent in {region}") 

138 regional_urls[region] = endpoint_url 

139 

140 health_warmup_samples = _health_warmup_samples( 

141 ctx, 

142 global_url=global_url, 

143 regional_urls=regional_urls, 

144 ) 

145 health_samples = _health_stability_samples( 

146 ctx, 

147 global_url=global_url, 

148 regional_urls=regional_urls, 

149 ) 

150 metrics_samples = _metrics_reachability_samples( 

151 ctx, 

152 global_url=global_url, 

153 regional_urls=regional_urls, 

154 ) 

155 global_samples = [sample for sample in health_samples if sample["scope"] == "global"] 

156 global_api = { 

157 "url": global_url, 

158 "health": global_samples[-1]["payload"], 

159 "samples": global_samples, 

160 } 

161 if direct_regional_access: 

162 regional_endpoints = { 

163 region: { 

164 "url": regional_urls[region], 

165 "health": next( 

166 sample["payload"] 

167 for sample in reversed(health_samples) 

168 if sample["region"] == region 

169 ), 

170 "samples": [sample for sample in health_samples if sample["region"] == region], 

171 } 

172 for region in ctx.deployment_regions 

173 } 

174 else: 

175 regional_endpoints = { 

176 region: { 

177 "skipped": True, 

178 "reason": "direct caller access is disabled by cdk.json", 

179 "samples": [], 

180 } 

181 for region in ctx.deployment_regions 

182 } 

183 

184 queue_baseline: dict[str, Any] = {} 

185 for region in ctx.deployment_regions: 

186 status = ctx.job_manager.get_queue_status(region) 

187 counts = _queue_counts(status) 

188 if any(counts.values()): 

189 raise RuntimeError( 

190 f"Fresh queue in {region} is not empty: {json.dumps(counts, sort_keys=True)}" 

191 ) 

192 queue_baseline[region] = status 

193 

194 table_name = f"{ctx.config.project_name}-jobs" 

195 table = ctx.session.client("dynamodb", region_name=ctx.config.global_region).describe_table( 

196 TableName=table_name 

197 )["Table"] 

198 if table.get("TableStatus") != "ACTIVE": 

199 raise RuntimeError(f"DynamoDB table {table_name} is not ACTIVE") 

200 

201 ctx.checkpoint.state["queue_baseline"] = queue_baseline 

202 ctx.persist() 

203 return { 

204 "stacks": stack_details, 

205 "clusters": clusters, 

206 "convergence": convergence, 

207 "alb_https_targets": alb_https_targets, 

208 "health_warmup_samples": health_warmup_samples, 

209 "health_samples": health_samples, 

210 "metrics_samples": metrics_samples, 

211 "global_api": global_api, 

212 "regional_apis": regional_endpoints, 

213 "queue_baseline": queue_baseline, 

214 "jobs_table": { 

215 "name": table_name, 

216 "arn": table.get("TableArn"), 

217 "status": table.get("TableStatus"), 

218 }, 

219 }