Coverage for scripts / live_release_validation / cleanup / transaction_search.py: 100.00%

1"""Put the account's Transaction Search state back the way the baseline found it. 

2 

3Runs as a retained-resource cleanup phase, i.e. only after every exact target 

4stack is proven absent, so no GCO component is left to write a span or to 

5re-run the enabler. Per regional Region, in dependency order: 

6 

71. **Destination.** If the baseline destination was ``XRay`` and the Region 

8 now sends segments to ``CloudWatchLogs``, switch it back with 

9 ``UpdateTraceSegmentDestination(Destination="XRay")`` (after a bounded wait 

10 for a still-``PENDING`` switch to settle), then wait, bounded, for the 

11 switch back to become ``ACTIVE``: while it is ``PENDING``, X-Ray may still 

12 deliver spans through the policy into the log groups. A destination that 

13 was already ``CloudWatchLogs`` at baseline is never touched. 

142. **Resource policy.** The ``gco-transaction-search-xray-access`` policy is 

15 deleted only when it did not exist at baseline, still has the X-Ray span 

16 delivery shape, and was last written during this run. 

173. **Span log groups.** ``aws/spans`` and ``/aws/application-signals/data`` 

18 are deleted only when they did not exist at baseline, the destination is 

19 ``XRay`` again (so nothing re-creates them), and the live generation was 

20 created during this run and carries no other owner's markers. The delete 

21 follows one exact identity read with nothing in between, and absence must 

22 then hold across repeated reads. When Transaction Search was already on at 

23 baseline, a new span log group is shared with every other span producer in 

24 the account and is retained (``final-inventory`` reports it as accepted). 

25 

26These groups are service-created, carry no run tags, and are not stack 

27resources, so they cannot join the tag-conditioned ``owned_log_groups`` 

28authority; the baseline absence proof plus the creation-time fence are their 

29ownership proof. Pre-existing state is never modified. A destination switch 

30that fails, or has not become ``ACTIVE`` by the deadline, stops that Region 

31before the policy or log groups are touched, because a Region that may still 

32deliver spans to ``CloudWatchLogs`` still needs both; resuming ``destroy`` 

33retries it. 

34""" 

35 

36from __future__ import annotations 

37 

38import copy 

39import json 

40import time 

41from datetime import datetime 

42from typing import Any 

43 

44from botocore.exceptions import ClientError 

45 

46from ..cleanup.log_groups import _log_group_adoption_blockers 

47from ..models import RunContext 

48from ..ownership.log_groups import _observe_log_group_stability 

49from ..ownership.stacks import _verify_target_stack_absence 

50from ..ownership.transaction_search import ( 

51 TRANSACTION_SEARCH_LOG_GROUPS, 

52 TRANSACTION_SEARCH_POLICY_NAME, 

53 _gco_resource_policy, 

54 _observe_transaction_search_state, 

55 _policy_summary, 

56 _read_trace_segment_destination, 

57 _validated_transaction_search_baseline, 

58) 

59 

60#: Bounded wait for a ``PENDING`` destination switch to settle before undoing it. 

61_DESTINATION_SETTLE_SECONDS = 600 

62_DESTINATION_POLL_SECONDS = 15 

63#: Reads that must agree before a delete, and absence reads after it. 

64_PRESENT_OBSERVATIONS = 2 

65_ABSENT_OBSERVATIONS = 3 

66 

67 

68class TransactionSearchRestoreError(RuntimeError): 

69 """A Region could not be restored; carries the partial evidence.""" 

70 

71 def __init__(self, message: str, details: dict[str, Any]) -> None: 

72 super().__init__(message) 

73 self.details = copy.deepcopy(details) 

74 

75 

76def _run_started_ms(ctx: RunContext) -> int: 

77 try: 

78 return int(datetime.fromisoformat(ctx.checkpoint.created_at).timestamp() * 1000) 

79 except (TypeError, ValueError) as exc: 

80 raise RuntimeError("Checkpoint created_at is not a valid timestamp") from exc 

81 

82 

83def _settled_destination(xray: Any) -> dict[str, str]: 

84 """Wait (bounded) for a ``PENDING`` destination to become ``ACTIVE``.""" 

85 deadline = time.monotonic() + _DESTINATION_SETTLE_SECONDS 

86 while True: 

87 state = _read_trace_segment_destination(xray) 

88 if state["status"] == "ACTIVE" or time.monotonic() >= deadline: 

89 return state 

90 time.sleep(_DESTINATION_POLL_SECONDS) 

91 

92 

93def _restore_destination( 

94 xray: Any, 

95 region: str, 

96 baseline: dict[str, Any], 

97 actions: list[dict[str, Any]], 

98) -> None: 

99 """Leave a baseline-``XRay`` Region on an ``ACTIVE`` ``XRay`` destination, or raise. 

100 

101 A Region that was on ``CloudWatchLogs`` at baseline is never touched. 

102 """ 

103 if baseline["destination"] != "XRay": 

104 return 

105 current = _settled_destination(xray) 

106 if current["destination"] == "CloudWatchLogs": 

107 response = xray.update_trace_segment_destination(Destination="XRay") 

108 actions.append( 

109 { 

110 "action": "update-trace-segment-destination", 

111 "from": "CloudWatchLogs", 

112 "to": "XRay", 

113 "status_before": current["status"], 

114 "response_status": str(response.get("Status") or ""), 

115 } 

116 ) 

117 current = _settled_destination(xray) 

118 if current["destination"] != "XRay" or current["status"] != "ACTIVE": 

119 raise RuntimeError( 

120 f"{region}: trace segment destination is {current['destination']} " 

121 f"({current['status']}), not ACTIVE on XRay, after restoring it; the resource " 

122 "policy and span log groups stay until it is (resume destroy to retry)" 

123 ) 

124 

125 

126def _restore_resource_policy( 

127 logs: Any, 

128 region: str, 

129 baseline: dict[str, Any], 

130 run_started_ms: int, 

131 actions: list[dict[str, Any]], 

132) -> None: 

133 """Delete the GCO policy only when this run wrote it.""" 

134 if baseline["resource_policy"]["present"]: 

135 return 

136 policy = _gco_resource_policy(logs) 

137 if policy is None: 

138 return 

139 summary = _policy_summary(policy) 

140 last_updated = summary["last_updated_time"] 

141 if not summary["grants_xray_span_delivery"]: 

142 raise RuntimeError( 

143 f"{region}: resource policy {TRANSACTION_SEARCH_POLICY_NAME} does not have the " 

144 "X-Ray span delivery shape; refusing to delete it" 

145 ) 

146 if last_updated is None or last_updated < run_started_ms: 

147 raise RuntimeError( 

148 f"{region}: resource policy {TRANSACTION_SEARCH_POLICY_NAME} was not written during " 

149 "this run; refusing to delete it" 

150 ) 

151 logs.delete_resource_policy(policyName=TRANSACTION_SEARCH_POLICY_NAME) 

152 if _gco_resource_policy(logs) is not None: 

153 raise RuntimeError( 

154 f"{region}: resource policy {TRANSACTION_SEARCH_POLICY_NAME} is still present " 

155 "after deletion" 

156 ) 

157 actions.append({"action": "delete-resource-policy", "policy": TRANSACTION_SEARCH_POLICY_NAME}) 

158 

159 

160def _restore_log_group( 

161 ctx: RunContext, 

162 logs: Any, 

163 region: str, 

164 name: str, 

165 *, 

166 baseline: dict[str, Any], 

167 run_started_ms: int, 

168) -> dict[str, Any]: 

169 """Return one span log group's disposition, deleting it only when the run owns it. 

170 

171 Only called once :func:`_restore_destination` has left a baseline-``XRay`` 

172 Region ``ACTIVE`` on ``XRay``, so nothing re-creates a deleted group. 

173 """ 

174 if baseline["log_groups"][name]["present"]: 

175 return {"disposition": "preexisting-untouched"} 

176 observed = _observe_log_group_stability( 

177 logs, 

178 region, 

179 name, 

180 required_present=_PRESENT_OBSERVATIONS, 

181 required_absent=_ABSENT_OBSERVATIONS, 

182 ) 

183 if observed["status"] == "absent": 

184 return {"disposition": "absent"} 

185 identity = observed.get("identity") 

186 if observed["status"] != "present" or not isinstance(identity, dict): 

187 raise RuntimeError( 

188 f"{region}:{name} did not settle before restoration: {observed['status']}" 

189 ) 

190 if baseline["destination"] != "XRay": 

191 # Transaction Search was already on: every span producer in the account 

192 # shares this group, whatever the destination reads now. 

193 return {"disposition": "retained-transaction-search-enabled-at-baseline"} 

194 if int(identity["creation_time"]) < run_started_ms: 

195 raise RuntimeError(f"{region}:{name} predates this validation run; refusing to delete it") 

196 blockers = _log_group_adoption_blockers( 

197 identity, 

198 run_id=ctx.settings.run_id, 

199 cleanup_token=str(ctx.checkpoint.state.get("log_group_cleanup_token") or ""), 

200 ) 

201 if blockers: 

202 raise RuntimeError( 

203 f"{region}:{name} carries another owner's markers: {', '.join(blockers)}" 

204 ) 

205 # Nothing may run between this exact read and the delete request. 

206 pre_delete = _observe_log_group_stability( 

207 logs, 

208 region, 

209 name, 

210 expected_identity=identity, 

211 required_present=1, 

212 required_absent=1, 

213 ) 

214 if pre_delete["status"] == "present": 

215 try: 

216 logs.delete_log_group(logGroupName=name) 

217 except ClientError as exc: 

218 if exc.response.get("Error", {}).get("Code") != "ResourceNotFoundException": 

219 raise 

220 elif pre_delete["status"] != "absent": 

221 raise RuntimeError(f"{region}:{name} changed immediately before deletion") 

222 absence = _observe_log_group_stability( 

223 logs, 

224 region, 

225 name, 

226 expected_identity=identity, 

227 required_present=None, 

228 required_absent=_ABSENT_OBSERVATIONS, 

229 ) 

230 if absence["status"] != "absent": 

231 raise RuntimeError(f"{region}:{name} did not stay absent after deletion") 

232 return { 

233 "disposition": "deleted" if pre_delete["status"] == "present" else "absent", 

234 "creation_time": int(identity["creation_time"]), 

235 } 

236 

237 

238def _restore_region( 

239 ctx: RunContext, 

240 region: str, 

241 baseline: dict[str, Any], 

242 run_started_ms: int, 

243) -> dict[str, Any]: 

244 xray = ctx.session.client("xray", region_name=region) 

245 logs = ctx.session.client("logs", region_name=region) 

246 before = _observe_transaction_search_state(ctx, region) 

247 actions: list[dict[str, Any]] = [] 

248 _restore_destination(xray, region, baseline, actions) 

249 _restore_resource_policy(logs, region, baseline, run_started_ms, actions) 

250 log_groups: dict[str, Any] = {} 

251 for name in TRANSACTION_SEARCH_LOG_GROUPS: 

252 log_groups[name] = _restore_log_group( 

253 ctx, 

254 logs, 

255 region, 

256 name, 

257 baseline=baseline, 

258 run_started_ms=run_started_ms, 

259 ) 

260 if log_groups[name]["disposition"] == "deleted": 

261 actions.append({"action": "delete-log-group", "log_group": name}) 

262 return { 

263 "baseline": copy.deepcopy(baseline), 

264 "before": before, 

265 "actions": actions, 

266 "log_groups": log_groups, 

267 "after": _observe_transaction_search_state(ctx, region), 

268 } 

269 

270 

271def _restore_transaction_search(ctx: RunContext) -> dict[str, Any]: 

272 """Restore every regional Region; one Region's failure never skips another.""" 

273 baseline = _validated_transaction_search_baseline(ctx) 

274 stack_absence = _verify_target_stack_absence(ctx) 

275 if not stack_absence["all_absent"]: 

276 raise RuntimeError( 

277 "Transaction Search restoration requires every exact target stack to be absent" 

278 ) 

279 run_started_ms = _run_started_ms(ctx) 

280 regions: dict[str, Any] = {} 

281 errors: list[dict[str, str]] = [] 

282 for region in sorted(baseline["regions"]): 

283 try: 

284 regions[region] = _restore_region( 

285 ctx, region, baseline["regions"][region], run_started_ms 

286 ) 

287 except Exception as exc: # restore every other Region, then fail with all evidence 

288 error = f"{type(exc).__name__}: {exc}" 

289 regions[region] = {"error": error} 

290 errors.append({"region": region, "error": error}) 

291 result = {"regions": regions, "errors": errors} 

292 with ctx.state_lock: 

293 ctx.checkpoint.state.setdefault("transaction_search_restore_attempts", []).append( 

294 copy.deepcopy(result) 

295 ) 

296 ctx.persist() 

297 if errors: 

298 raise TransactionSearchRestoreError( 

299 "Transaction Search restoration failed: " + json.dumps(errors, sort_keys=True), 

300 result, 

301 ) 

302 return result