Coverage for scripts / live_release_validation / ownership / log_groups.py: 100.00%

1"""CloudWatch log-group identity, stability observation, and ownership checkpointing.""" 

2 

3from __future__ import annotations 

4 

5import copy 

6import re 

7import time 

8import uuid 

9from collections.abc import Mapping 

10from datetime import datetime 

11from typing import Any, cast 

12 

13from botocore.exceptions import ClientError 

14 

15from ..constants import ( 

16 _EKS_LOG_GROUP_SUFFIXES, 

17 _LOG_CLEANUP_TOKEN_TAG, 

18 _LOG_GROUP_CHECKPOINT_STABLE_OBSERVATIONS, 

19 _LOG_GROUP_CLEANUP_STABLE_OBSERVATIONS, 

20 _LOG_GROUP_OBSERVATION_ATTEMPTS, 

21 _LOG_GROUP_OBSERVATION_HISTORY_LIMIT, 

22 _LOG_GROUP_OBSERVATION_POLL_SECONDS, 

23 _LOG_GROUP_RETRYABLE_OBSERVATION_CODES, 

24 _LOG_GROUP_SOURCE_TYPES, 

25 _RUN_STACK_TAG, 

26) 

27from ..inventory import ( 

28 describe_stack, 

29) 

30from ..models import RunContext, utc_now 

31from ..ownership.stacks import ( 

32 _owned_stack_ids, 

33 _owned_stack_record, 

34) 

35 

36 

37def _derived_log_group_names(resource_type: str, physical_id: str) -> tuple[str, ...]: 

38 if resource_type == "AWS::Logs::LogGroup": 

39 return (physical_id,) 

40 if resource_type == "AWS::Lambda::Function": 

41 return (f"/aws/lambda/{physical_id}",) 

42 if resource_type == "AWS::EKS::Cluster": 

43 return ( 

44 f"/aws/eks/{physical_id}/cluster", 

45 *( 

46 f"/aws/containerinsights/{physical_id}/{suffix}" 

47 for suffix in _EKS_LOG_GROUP_SUFFIXES 

48 ), 

49 ) 

50 return () 

51 

52 

53def _live_eks_cluster_identity( 

54 ctx: RunContext, 

55 region: str, 

56 cluster_name: str, 

57) -> dict[str, str]: 

58 """Require the exact ACTIVE service-side cluster before deriving log authority.""" 

59 partition = ctx.session.get_partition_for_region(region) 

60 if not partition: 

61 raise RuntimeError(f"Could not resolve AWS partition for EKS cluster in {region}") 

62 expected_arn = ( 

63 f"arn:{partition}:eks:{region}:{ctx.settings.expected_account}:cluster/{cluster_name}" 

64 ) 

65 response = ctx.session.client("eks", region_name=region).describe_cluster(name=cluster_name) 

66 cluster = response.get("cluster") 

67 if not isinstance(cluster, dict): 

68 raise RuntimeError(f"EKS omitted cluster identity for {region}:{cluster_name}") 

69 identity = { 

70 "name": str(cluster.get("name") or ""), 

71 "arn": str(cluster.get("arn") or ""), 

72 "status": str(cluster.get("status") or ""), 

73 } 

74 if identity != {"name": cluster_name, "arn": expected_arn, "status": "ACTIVE"}: 

75 raise RuntimeError( 

76 f"EKS cluster identity is not exact and ACTIVE for {region}:{cluster_name}" 

77 ) 

78 return identity 

79 

80 

81def _eks_cluster_log_authority_identity( 

82 ctx: RunContext, 

83 region: str, 

84 cluster_name: str, 

85 *, 

86 allow_deleted: bool, 

87) -> dict[str, str]: 

88 """Resolve EKS log authority, tolerating a rolled-back (deleted) cluster. 

89 

90 A create rollback deletes the cluster itself while its control-plane and 

91 Container Insights log groups survive. The DELETED tombstone identity is 

92 only ever derived from this run's own stack resource record. 

93 """ 

94 if not allow_deleted: 

95 return _live_eks_cluster_identity(ctx, region, cluster_name) 

96 partition = ctx.session.get_partition_for_region(region) 

97 if not partition: 

98 raise RuntimeError(f"Could not resolve AWS partition for EKS cluster in {region}") 

99 expected_arn = ( 

100 f"arn:{partition}:eks:{region}:{ctx.settings.expected_account}:cluster/{cluster_name}" 

101 ) 

102 try: 

103 return _live_eks_cluster_identity(ctx, region, cluster_name) 

104 except ClientError as exc: 

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

106 raise 

107 return {"name": cluster_name, "arn": expected_arn, "status": "DELETED"} 

108 

109 

110def _validated_owned_log_group_identity( 

111 ctx: RunContext, 

112 record: Mapping[str, Any], 

113) -> tuple[str, str]: 

114 """Validate an exact log-group name derived from a checkpointed stack resource.""" 

115 region = str(record.get("region") or "") 

116 name = str(record.get("name") or "") 

117 stack_name = str(record.get("stack_name") or "") 

118 stack_id = str(record.get("stack_id") or "") 

119 resource_type = str(record.get("source_resource_type") or "") 

120 logical_id = str(record.get("source_logical_id") or "") 

121 physical_id = str(record.get("source_physical_id") or "") 

122 target_regions = ctx.checkpoint.state.get("target_stack_regions") 

123 if not isinstance(target_regions, dict) or str(target_regions.get(stack_name) or "") != region: 

124 raise RuntimeError(f"Log-group checkpoint target stack is invalid for {region}:{name}") 

125 partition = ctx.session.get_partition_for_region(region) 

126 if not partition: 

127 raise RuntimeError(f"Could not resolve AWS partition for log group in {region}") 

128 expected_stack_prefix = ( 

129 f"arn:{partition}:cloudformation:{region}:{ctx.settings.expected_account}:" 

130 f"stack/{stack_name}/" 

131 ) 

132 owned_stack_record = _owned_stack_record(ctx, region, stack_name) 

133 # The source stack may be the current generation or, after an adopted 

134 # in-place upgrade, one it replaced (its retained log groups outlive it). 

135 owned_stack_ids = _owned_stack_ids(ctx, region, stack_name) 

136 if ( 

137 not name 

138 or not logical_id 

139 or resource_type not in _LOG_GROUP_SOURCE_TYPES 

140 or name not in _derived_log_group_names(resource_type, physical_id) 

141 ): 

142 raise RuntimeError(f"Log-group checkpoint source is invalid for {region}:{name}") 

143 source_service_identity = record.get("source_service_identity") 

144 if resource_type == "AWS::EKS::Cluster": 

145 expected_arn = ( 

146 f"arn:{partition}:eks:{region}:{ctx.settings.expected_account}:cluster/{physical_id}" 

147 ) 

148 # ACTIVE is the pre-destroy authority; DELETED is the exact tombstone 

149 # recorded when a rolled-back create removed the cluster but left its 

150 # control-plane and Container Insights log groups behind. 

151 accepted_source_identities = tuple( 

152 {"name": physical_id, "arn": expected_arn, "status": status} 

153 for status in ("ACTIVE", "DELETED") 

154 ) 

155 if source_service_identity not in accepted_source_identities: 

156 raise RuntimeError( 

157 f"Log-group checkpoint lacks exact live EKS identity for {region}:{name}" 

158 ) 

159 elif source_service_identity not in (None, {}): 

160 raise RuntimeError(f"Unexpected service identity for log group {region}:{name}") 

161 cleanup_token = str(record.get("cleanup_token") or "") 

162 expected_cleanup_token = str(ctx.checkpoint.state.get("log_group_cleanup_token") or "") 

163 if ( 

164 not stack_id.startswith(expected_stack_prefix) 

165 or stack_id not in owned_stack_ids 

166 or (owned_stack_record or {}).get("run_tag") != ctx.settings.run_id 

167 or record.get("run_tag") != ctx.settings.run_id 

168 or record.get("ownership_authority") != "cloudformation-stack-resource-derived" 

169 or record.get("authority_phase") != "pre-destroy" 

170 or not cleanup_token 

171 or cleanup_token != expected_cleanup_token 

172 ): 

173 raise RuntimeError(f"Log-group checkpoint authority is invalid for {region}:{name}") 

174 return region, name 

175 

176 

177def _describe_exact_log_group(client: Any, name: str) -> dict[str, Any] | None: 

178 kwargs: dict[str, Any] = {"logGroupNamePrefix": name, "limit": 50} 

179 while True: 

180 response = client.describe_log_groups(**kwargs) 

181 for log_group in response.get("logGroups", []): 

182 if not isinstance(log_group, Mapping): 

183 raise RuntimeError("CloudWatch Logs returned a non-object log-group record") 

184 candidate = cast(Mapping[str, Any], log_group) 

185 if str(candidate.get("logGroupName") or "") == name: 

186 return {str(key): value for key, value in candidate.items()} 

187 token = response.get("nextToken") 

188 if not token: 

189 return None 

190 kwargs["nextToken"] = token 

191 

192 

193def _log_group_identity(client: Any, region: str, name: str) -> dict[str, Any] | None: 

194 log_group = _describe_exact_log_group(client, name) 

195 if log_group is None: 

196 return None 

197 arn = str(log_group.get("logGroupArn") or log_group.get("arn") or "").removesuffix(":*") 

198 creation_time = log_group.get("creationTime") 

199 if not arn or not isinstance(creation_time, int): 

200 raise RuntimeError(f"CloudWatch Logs omitted identity for {region}:{name}") 

201 tags = client.list_tags_for_resource(resourceArn=arn).get("tags") or {} 

202 return { 

203 "arn": arn, 

204 "creation_time": creation_time, 

205 "tags": {str(key): str(value) for key, value in tags.items()}, 

206 } 

207 

208 

209def _log_group_generation(identity: Mapping[str, Any]) -> dict[str, Any]: 

210 """Return the immutable fields that distinguish same-name log generations.""" 

211 arn = str(identity.get("arn") or "") 

212 creation_time = identity.get("creation_time") 

213 if not arn or not isinstance(creation_time, int): 

214 raise RuntimeError("Checkpointed CloudWatch log-group identity is malformed") 

215 return {"arn": arn, "creation_time": creation_time} 

216 

217 

218def _observe_log_group_stability( 

219 client: Any, 

220 region: str, 

221 name: str, 

222 *, 

223 expected_identity: Mapping[str, Any] | None = None, 

224 expected_tags: Mapping[str, str] | None = None, 

225 required_present: int | None, 

226 required_absent: int | None, 

227 attempts: int = _LOG_GROUP_OBSERVATION_ATTEMPTS, 

228 poll_seconds: float = _LOG_GROUP_OBSERVATION_POLL_SECONDS, 

229) -> dict[str, Any]: 

230 """Bound identity reads until presence/absence is stable or a fence is crossed.""" 

231 for label, value in ( 

232 ("attempts", attempts), 

233 ("required_present", required_present), 

234 ("required_absent", required_absent), 

235 ): 

236 if value is not None and ( 

237 isinstance(value, bool) or not isinstance(value, int) or value <= 0 

238 ): 

239 raise ValueError(f"{label} must be a positive integer or None") 

240 if required_present is None and required_absent is None: 

241 raise ValueError("At least one stable log-group outcome must be requested") 

242 if poll_seconds < 0: 

243 raise ValueError("poll_seconds must be non-negative") 

244 

245 expected_generation = ( 

246 _log_group_generation(expected_identity) if expected_identity is not None else None 

247 ) 

248 expected_authority_tags = {str(key): str(value) for key, value in (expected_tags or {}).items()} 

249 observations: list[dict[str, Any]] = [] 

250 seen_generations: list[dict[str, Any]] = [] 

251 present_streak = 0 

252 absent_streak = 0 

253 replacement_streak = 0 

254 replacement_generation: dict[str, Any] | None = None 

255 last_identity: dict[str, Any] | None = None 

256 

257 def result(status: str, **extra: Any) -> dict[str, Any]: 

258 return { 

259 "status": status, 

260 "region": region, 

261 "name": name, 

262 "attempt_count": len(observations), 

263 "observations": observations, 

264 "identity": copy.deepcopy(last_identity), 

265 **extra, 

266 } 

267 

268 for attempt in range(1, attempts + 1): 

269 observed_at = utc_now() 

270 try: 

271 identity = _log_group_identity(client, region, name) 

272 except ClientError as exc: 

273 code = str(exc.response.get("Error", {}).get("Code") or "") 

274 if code not in _LOG_GROUP_RETRYABLE_OBSERVATION_CODES: 

275 raise 

276 observations.append( 

277 { 

278 "attempt": attempt, 

279 "observed_at": observed_at, 

280 "status": "retryable-error", 

281 "error_code": code, 

282 "error": str(exc), 

283 } 

284 ) 

285 present_streak = 0 

286 absent_streak = 0 

287 replacement_streak = 0 

288 replacement_generation = None 

289 else: 

290 last_identity = copy.deepcopy(identity) 

291 if identity is None: 

292 if expected_generation is None and seen_generations: 

293 observations.append( 

294 { 

295 "attempt": attempt, 

296 "observed_at": observed_at, 

297 "status": "replacement", 

298 "observed_generation": None, 

299 } 

300 ) 

301 return result( 

302 "replacement", 

303 expected_generation=seen_generations[-1], 

304 observed_generation=None, 

305 ) 

306 replacement_streak = 0 

307 replacement_generation = None 

308 absent_streak += 1 

309 present_streak = 0 

310 observations.append( 

311 { 

312 "attempt": attempt, 

313 "observed_at": observed_at, 

314 "status": "absent", 

315 "consecutive": absent_streak, 

316 } 

317 ) 

318 if required_absent is not None and absent_streak >= required_absent: 

319 return result("absent", consecutive=absent_streak) 

320 else: 

321 generation = _log_group_generation(identity) 

322 if generation not in seen_generations: 

323 seen_generations.append(generation) 

324 observations.append( 

325 { 

326 "attempt": attempt, 

327 "observed_at": observed_at, 

328 "status": "present", 

329 "generation": copy.deepcopy(generation), 

330 } 

331 ) 

332 if expected_generation is not None and generation != expected_generation: 

333 if generation == replacement_generation: 

334 replacement_streak += 1 

335 else: 

336 replacement_generation = generation 

337 replacement_streak = 1 

338 observations[-1]["status"] = "replacement-candidate" 

339 observations[-1]["consecutive"] = replacement_streak 

340 present_streak = 0 

341 absent_streak = 0 

342 if replacement_streak >= _LOG_GROUP_CLEANUP_STABLE_OBSERVATIONS: 

343 observations[-1]["status"] = "replacement" 

344 return result( 

345 "replacement", 

346 expected_generation=expected_generation, 

347 observed_generation=generation, 

348 consecutive=replacement_streak, 

349 ) 

350 if attempt < attempts: 

351 time.sleep(poll_seconds) 

352 continue 

353 replacement_streak = 0 

354 replacement_generation = None 

355 if expected_generation is None and len(seen_generations) > 1: 

356 observations[-1]["status"] = "replacement" 

357 return result( 

358 "replacement", 

359 expected_generation=seen_generations[0], 

360 observed_generation=generation, 

361 ) 

362 tags = identity.get("tags") or {} 

363 tag_drift = { 

364 key: {"expected": value, "observed": tags.get(key)} 

365 for key, value in expected_authority_tags.items() 

366 if tags.get(key) != value 

367 } 

368 if tag_drift: 

369 observations[-1]["status"] = "tag-drift" 

370 observations[-1]["tag_drift"] = copy.deepcopy(tag_drift) 

371 return result("tag-drift", tag_drift=tag_drift) 

372 present_streak += 1 

373 absent_streak = 0 

374 observations[-1]["consecutive"] = present_streak 

375 if required_present is not None and present_streak >= required_present: 

376 return result("present", consecutive=present_streak) 

377 if attempt < attempts: 

378 time.sleep(poll_seconds) 

379 

380 return result( 

381 "unsettled", 

382 required_present=required_present, 

383 required_absent=required_absent, 

384 present_streak=present_streak, 

385 absent_streak=absent_streak, 

386 ) 

387 

388 

389def _record_log_group_observation( 

390 ctx: RunContext, 

391 record: dict[str, Any], 

392 *, 

393 phase: str, 

394 outcome: Mapping[str, Any], 

395) -> dict[str, Any]: 

396 """Persist bounded identity evidence and any confirmed replacement generation.""" 

397 entry = {"phase": phase, "recorded_at": utc_now(), **copy.deepcopy(dict(outcome))} 

398 with ctx.state_lock: 

399 history = record.setdefault("identity_observation_history", []) 

400 if not isinstance(history, list): 

401 raise RuntimeError("Log-group identity_observation_history must be a list") 

402 history.append(entry) 

403 del history[:-_LOG_GROUP_OBSERVATION_HISTORY_LIMIT] 

404 if outcome.get("status") == "replacement": 

405 replacements = record.setdefault("replacement_evidence", []) 

406 if not isinstance(replacements, list): 

407 raise RuntimeError("Log-group replacement_evidence must be a list") 

408 replacements.append(copy.deepcopy(entry)) 

409 ctx.persist_callback(ctx.checkpoint) 

410 return entry 

411 

412 

413def _record_log_group_checkpoint_incident( 

414 ctx: RunContext, 

415 candidate: Mapping[str, Any], 

416 *, 

417 phase: str, 

418 outcome: Mapping[str, Any], 

419) -> None: 

420 """Preserve failed pre-authority observations without adopting the generation.""" 

421 with ctx.state_lock: 

422 incidents = ctx.checkpoint.state.setdefault("log_group_checkpoint_incidents", []) 

423 if not isinstance(incidents, list): 

424 raise RuntimeError("Checkpoint log_group_checkpoint_incidents must be a list") 

425 incidents.append( 

426 { 

427 "phase": phase, 

428 "recorded_at": utc_now(), 

429 "candidate": copy.deepcopy(dict(candidate)), 

430 "outcome": copy.deepcopy(dict(outcome)), 

431 } 

432 ) 

433 ctx.persist_callback(ctx.checkpoint) 

434 

435 

436def _set_log_group_disposition( 

437 ctx: RunContext, 

438 record: dict[str, Any], 

439 *, 

440 status: str, 

441 phase: str, 

442 outcome: Mapping[str, Any], 

443) -> dict[str, Any]: 

444 disposition = { 

445 "status": status, 

446 "phase": phase, 

447 "recorded_at": utc_now(), 

448 "original_identity": copy.deepcopy(record.get("observed_identity")), 

449 "last_observation_status": str(outcome.get("status") or ""), 

450 } 

451 with ctx.state_lock: 

452 record["original_generation_disposition"] = disposition 

453 ctx.persist_callback(ctx.checkpoint) 

454 return disposition 

455 

456 

457#: Checkpoint list of records a recreated stack superseded (see 

458#: :func:`_carry_log_group_across_generations`), kept as evidence only. 

459_SUPERSEDED_LOG_GROUPS_KEY = "superseded_log_groups" 

460 

461#: Record fields a same-name log group keeps when its stack is recreated. 

462_GENERATION_INVARIANT_FIELDS = ( 

463 "region", 

464 "name", 

465 "stack_name", 

466 "source_resource_type", 

467 "source_physical_id", 

468 "ownership_authority", 

469 "authority_phase", 

470 "run_tag", 

471 "cleanup_token", 

472) 

473 

474#: Record fields that bind a log group to one stack generation. The logical ID 

475#: is among them because a later release may rename the construct that owns a 

476#: fixed physical name. 

477_GENERATION_BINDING_FIELDS = ("stack_id", "source_logical_id", "source_service_identity") 

478 

479 

480def _replaced_generation_record( 

481 ctx: RunContext, 

482 previous: Mapping[str, Any], 

483 candidate: Mapping[str, Any], 

484) -> bool: 

485 """Whether ``previous`` was derived from a stack generation this run replaced. 

486 

487 Only run-tag adoption (the upgrade harness) records replaced generations. 

488 A recreated stack keeps its fixed resource names, so the EKS cluster's and 

489 each named Lambda function's log groups are derived again, under the same 

490 names, from the generation that replaced it. 

491 """ 

492 previous_stack = str(previous.get("stack_id") or "") 

493 owned = _owned_stack_ids(ctx, str(candidate["region"]), str(candidate["stack_name"])) 

494 return ( 

495 previous_stack != candidate["stack_id"] 

496 and previous_stack in owned 

497 and all(previous.get(field) == candidate[field] for field in _GENERATION_INVARIANT_FIELDS) 

498 ) 

499 

500 

501def _carry_log_group_across_generations( 

502 ctx: RunContext, 

503 records: list[Any], 

504 previous: dict[str, Any], 

505 candidate: Mapping[str, Any], 

506 *, 

507 logs: Any, 

508 authority_tags: Mapping[str, str], 

509) -> bool: 

510 """Hand a replaced generation's same-name record to the stack that replaced it. 

511 

512 Returns ``True`` when the recorded group generation outlived its stack: 

513 the record is rebound to the new stack generation in place, its earlier 

514 binding kept under ``stack_generations``. Returns ``False`` when that 

515 generation is gone (``gco``'s teardown deletes the implicit groups of the 

516 stacks it destroys): the record moves to ``superseded_log_groups`` and the 

517 caller checkpoints the name afresh from the new generation. A deleted 

518 log-group generation never returns, since its creation time is part of its 

519 identity, so a stable different generation or stable absence proves it 

520 gone. Anything else fails closed. 

521 """ 

522 region = str(candidate["region"]) 

523 name = str(candidate["name"]) 

524 _validated_owned_log_group_identity(ctx, previous) 

525 observed = previous.get("observed_identity") 

526 if not isinstance(observed, dict): 

527 raise RuntimeError(f"Log-group checkpoint identity is malformed for {region}:{name}") 

528 phase = "checkpoint-stack-generation" 

529 outcome = _observe_log_group_stability( 

530 logs, 

531 region, 

532 name, 

533 expected_identity=observed, 

534 expected_tags=authority_tags, 

535 required_present=_LOG_GROUP_CHECKPOINT_STABLE_OBSERVATIONS, 

536 required_absent=_LOG_GROUP_CHECKPOINT_STABLE_OBSERVATIONS, 

537 ) 

538 _record_log_group_observation(ctx, previous, phase=phase, outcome=outcome) 

539 if outcome["status"] == "present": 

540 with ctx.state_lock: 

541 binding = { 

542 field: copy.deepcopy(previous[field]) 

543 for field in _GENERATION_BINDING_FIELDS 

544 if field in previous 

545 } 

546 generations = previous.setdefault("stack_generations", []) 

547 if not isinstance(generations, list): 

548 raise RuntimeError("Log-group stack_generations must be a list") 

549 generations.append( 

550 {**binding, "rebound_to": candidate["stack_id"], "rebound_at": utc_now()} 

551 ) 

552 # The resource type is invariant, so both records carry the same 

553 # binding fields (only an EKS cluster's has a service identity). 

554 for field in _GENERATION_BINDING_FIELDS: 

555 if field in candidate: 

556 previous[field] = copy.deepcopy(candidate[field]) 

557 ctx.persist_callback(ctx.checkpoint) 

558 return True 

559 if outcome["status"] not in {"replacement", "absent"}: 

560 _set_log_group_disposition( 

561 ctx, 

562 previous, 

563 status="checkpoint-generation-not-stable", 

564 phase=phase, 

565 outcome=outcome, 

566 ) 

567 raise RuntimeError( 

568 f"Log-group checkpoint generation is not stable for {region}:{name}: " 

569 f"{outcome['status']}" 

570 ) 

571 _set_log_group_disposition( 

572 ctx, 

573 previous, 

574 status="superseded-by-stack-generation", 

575 phase=phase, 

576 outcome=outcome, 

577 ) 

578 with ctx.state_lock: 

579 superseded = ctx.checkpoint.state.setdefault(_SUPERSEDED_LOG_GROUPS_KEY, []) 

580 if not isinstance(superseded, list): 

581 raise RuntimeError("Checkpoint superseded_log_groups must be a list") 

582 superseded.append( 

583 { 

584 **copy.deepcopy(previous), 

585 "superseded_by_stack_id": candidate["stack_id"], 

586 "superseded_at": utc_now(), 

587 } 

588 ) 

589 records[:] = [item for item in records if item is not previous] 

590 ctx.persist_callback(ctx.checkpoint) 

591 return False 

592 

593 

594def _checkpoint_owned_log_groups(ctx: RunContext) -> list[dict[str, Any]]: 

595 """Fence, tag, and checkpoint exact generations while source stacks are live.""" 

596 target_regions = ctx.checkpoint.state.get("target_stack_regions") 

597 if not isinstance(target_regions, dict): 

598 raise RuntimeError("Checkpoint target_stack_regions must be an object") 

599 try: 

600 run_started_ms = int(datetime.fromisoformat(ctx.checkpoint.created_at).timestamp() * 1000) 

601 except ValueError as exc: 

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

603 

604 with ctx.state_lock: 

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

606 if not cleanup_token: 

607 cleanup_token = uuid.uuid4().hex 

608 ctx.checkpoint.state["log_group_cleanup_token"] = cleanup_token 

609 if not re.fullmatch(r"[0-9a-f]{32}", cleanup_token): 

610 raise RuntimeError("Checkpoint log-group cleanup token is malformed") 

611 records = ctx.checkpoint.state.setdefault("owned_log_groups", []) 

612 if not isinstance(records, list): 

613 raise RuntimeError("Checkpoint owned_log_groups must be a list") 

614 by_identity = { 

615 (str(item.get("region") or ""), str(item.get("name") or "")): item 

616 for item in records 

617 if isinstance(item, dict) 

618 } 

619 authority_tags = { 

620 _RUN_STACK_TAG: ctx.settings.run_id, 

621 _LOG_CLEANUP_TOKEN_TAG: cleanup_token, 

622 } 

623 for stack_name, raw_region in sorted(target_regions.items()): 

624 region = str(raw_region) 

625 stack_record = _owned_stack_record(ctx, region, str(stack_name)) 

626 if stack_record is None: 

627 continue 

628 live_stack = describe_stack(ctx.session, region, stack_record["stack_id"]) 

629 if ( 

630 live_stack is None 

631 or str(live_stack.get("status") or "").startswith("DELETE") 

632 or (live_stack.get("tags") or {}).get(_RUN_STACK_TAG) != ctx.settings.run_id 

633 ): 

634 # Destructive authority is never created or completed from a 

635 # deleted-stack tombstone. Existing pre-destroy records remain usable. 

636 continue 

637 # A rolled-back create leaves log groups behind: retained LogGroup 

638 # resources, Lambda-created default groups, and EKS control-plane 

639 # groups all survive resource deletion. Their stack resources read 

640 # DELETE_COMPLETE while the stack itself is still describable, so 

641 # rollback statuses widen the resource filter to those tombstones. 

642 rolled_back = str(live_stack.get("status") or "") in { 

643 "ROLLBACK_COMPLETE", 

644 "ROLLBACK_FAILED", 

645 "UPDATE_ROLLBACK_COMPLETE", 

646 "UPDATE_ROLLBACK_FAILED", 

647 } 

648 allowed_resource_statuses = {"CREATE_COMPLETE", "UPDATE_COMPLETE"} 

649 if rolled_back: 

650 allowed_resource_statuses |= {"DELETE_COMPLETE", "DELETE_FAILED", "DELETE_SKIPPED"} 

651 cfn = ctx.session.client("cloudformation", region_name=region) 

652 pages = cfn.get_paginator("list_stack_resources").paginate( 

653 StackName=stack_record["stack_id"] 

654 ) 

655 resources = [ 

656 item 

657 for page in pages 

658 for item in page.get("StackResourceSummaries", []) 

659 if str(item.get("ResourceType") or "") in _LOG_GROUP_SOURCE_TYPES 

660 and item.get("LogicalResourceId") 

661 and item.get("PhysicalResourceId") 

662 and str(item.get("ResourceStatus") or "") in allowed_resource_statuses 

663 ] 

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

665 lambda_client = ctx.session.client("lambda", region_name=region) 

666 for resource in resources: 

667 resource_type = str(resource["ResourceType"]) 

668 physical_id = str(resource["PhysicalResourceId"]) 

669 source_service_identity = None 

670 if resource_type == "AWS::EKS::Cluster": 

671 source_service_identity = _eks_cluster_log_authority_identity( 

672 ctx, 

673 region, 

674 physical_id, 

675 allow_deleted=rolled_back, 

676 ) 

677 names = _derived_log_group_names(resource_type, physical_id) 

678 if resource_type == "AWS::Lambda::Function": 

679 default_name = f"/aws/lambda/{physical_id}" 

680 try: 

681 function = lambda_client.get_function_configuration( 

682 FunctionName=physical_id 

683 ) 

684 except ClientError as exc: 

685 error_code = exc.response.get("Error", {}).get("Code", "") 

686 if not (rolled_back and error_code == "ResourceNotFoundException"): 

687 raise 

688 # The rolled-back function is gone; only its default 

689 # log group can remain. 

690 names = (default_name,) 

691 else: 

692 configured_name = str( 

693 (function.get("LoggingConfig") or {}).get("LogGroup") or default_name 

694 ) 

695 names = (default_name,) if configured_name == default_name else () 

696 for name in names: 

697 key = (region, name) 

698 candidate = { 

699 "region": region, 

700 "name": name, 

701 "stack_name": str(stack_name), 

702 "stack_id": stack_record["stack_id"], 

703 "source_resource_type": resource_type, 

704 "source_logical_id": str(resource["LogicalResourceId"]), 

705 "source_physical_id": physical_id, 

706 "ownership_authority": "cloudformation-stack-resource-derived", 

707 "authority_phase": "pre-destroy", 

708 "run_tag": ctx.settings.run_id, 

709 "cleanup_token": cleanup_token, 

710 } 

711 if source_service_identity is not None: 

712 candidate["source_service_identity"] = source_service_identity 

713 _validated_owned_log_group_identity(ctx, candidate) 

714 

715 previous = by_identity.get(key) 

716 immutable = tuple(candidate) 

717 expected_identity: Mapping[str, Any] | None = None 

718 if previous is not None and any( 

719 previous.get(field) != candidate[field] for field in immutable 

720 ): 

721 if not _replaced_generation_record(ctx, previous, candidate): 

722 raise RuntimeError(f"Log-group ownership changed for {region}:{name}") 

723 if _carry_log_group_across_generations( 

724 ctx, 

725 records, 

726 previous, 

727 candidate, 

728 logs=logs, 

729 authority_tags=authority_tags, 

730 ): 

731 continue 

732 del by_identity[key] 

733 previous = None 

734 if previous is not None: 

735 observed = previous.get("observed_identity") 

736 if not isinstance(observed, dict): 

737 raise RuntimeError( 

738 f"Log-group checkpoint identity is malformed for {region}:{name}" 

739 ) 

740 expected_identity = observed 

741 

742 initial = _observe_log_group_stability( 

743 logs, 

744 region, 

745 name, 

746 expected_identity=expected_identity, 

747 expected_tags=authority_tags if previous is not None else None, 

748 required_present=_LOG_GROUP_CHECKPOINT_STABLE_OBSERVATIONS, 

749 required_absent=_LOG_GROUP_CHECKPOINT_STABLE_OBSERVATIONS, 

750 ) 

751 if previous is not None: 

752 _record_log_group_observation( 

753 ctx, 

754 previous,