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

1"""Reclaim the local container images CDK asset publishing leaves behind. 

2 

3Every ``deploy`` builds each service image locally, tags it ``cdkasset-<hash>`` 

4plus the ECR asset-repository reference it was pushed under, and never removes 

5either. Hashes change with the source, so successive runs accumulate one full 

6image set each — a validation host filled its disk with hundreds of stale asset 

7images, which failed an image build mid-deploy and then the checkpoint write 

8that guaranteed cleanup depends on. Once an image is published to ECR the local 

9copy has no further use (CDK checks ECR before rebuilding), so the deploy 

10action prunes exactly the CDK asset images plus dangling build layers, and 

11nothing else the operator keeps in the local store. 

12""" 

13 

14from __future__ import annotations 

15 

16import re 

17import subprocess 

18from collections.abc import Callable 

19from typing import Any 

20 

21from cli._container_runtime import detect_container_runtime 

22 

23#: Local tag CDK assigns to every built asset image. 

24_ASSET_TAG_PREFIX = "cdkasset-" 

25#: The bootstrap ECR repository CDK pushes asset images to 

26#: (``cdk-<qualifier>-container-assets-<account>-<region>``), optionally 

27#: prefixed by the registry host. 

28_ASSET_REPOSITORY_PATTERN = re.compile(r"(?:^|/)cdk-[a-z0-9]+-container-assets-\d{12}-[a-z0-9-]+$") 

29_COMMAND_TIMEOUT_SECONDS = 300 

30#: Cap on the error text retained from a failed runtime command. 

31_MAX_ERROR_CHARS = 500 

32 

33RunCommand = Callable[..., "subprocess.CompletedProcess[str]"] 

34 

35 

36def _is_cdk_asset_repository(repository: str) -> bool: 

37 """Whether ``repository`` is a CDK asset image reference (either tag form).""" 

38 if repository.rsplit("/", 1)[-1].startswith(_ASSET_TAG_PREFIX): 

39 return True 

40 return _ASSET_REPOSITORY_PATTERN.search(repository) is not None 

41 

42 

43def _run(runtime: str, arguments: list[str], run: RunCommand) -> subprocess.CompletedProcess[str]: 

44 return run( 

45 [runtime, *arguments], 

46 capture_output=True, 

47 text=True, 

48 check=False, 

49 timeout=_COMMAND_TIMEOUT_SECONDS, 

50 ) 

51 

52 

53def prune_cdk_asset_images( 

54 *, 

55 runtime: str | None = None, 

56 run: RunCommand = subprocess.run, 

57) -> dict[str, Any]: 

58 """Remove local CDK asset images and dangling layers; never raise. 

59 

60 Returns evidence for the deploy record: the runtime used, every image 

61 reference removed, whether dangling layers were pruned, and any runtime 

62 error text. Absence of a runtime or a failing command is reported, not 

63 raised — reclaiming local disk must never change a deploy's verdict. 

64 """ 

65 runtime = runtime or detect_container_runtime() 

66 if runtime is None: 

67 return { 

68 "runtime": None, 

69 "removed_images": [], 

70 "dangling_pruned": False, 

71 "errors": [], 

72 "skipped": "no container runtime detected", 

73 } 

74 

75 errors: list[str] = [] 

76 removed: list[str] = [] 

77 listing = _run(runtime, ["images", "--format", "{{.Repository}}:{{.Tag}} {{.ID}}"], run) 

78 if listing.returncode != 0: 

79 errors.append(f"images: {(listing.stderr or '').strip()[:_MAX_ERROR_CHARS]}") 

80 else: 

81 image_ids: set[str] = set() 

82 for line in (listing.stdout or "").splitlines(): 

83 parts = line.split() 

84 if len(parts) != 2: 

85 continue 

86 reference, image_id = parts 

87 repository = reference.rsplit(":", 1)[0] 

88 if _is_cdk_asset_repository(repository): 

89 removed.append(reference) 

90 image_ids.add(image_id) 

91 if image_ids: 

92 result = _run(runtime, ["rmi", "-f", *sorted(image_ids)], run) 

93 if result.returncode != 0: 

94 errors.append(f"rmi: {(result.stderr or '').strip()[:_MAX_ERROR_CHARS]}") 

95 

96 prune = _run(runtime, ["image", "prune", "-f"], run) 

97 if prune.returncode != 0: 

98 errors.append(f"image prune: {(prune.stderr or '').strip()[:_MAX_ERROR_CHARS]}") 

99 

100 return { 

101 "runtime": runtime, 

102 "removed_images": sorted(removed), 

103 "dangling_pruned": prune.returncode == 0, 

104 "errors": errors, 

105 } 

106 

107 

108def prune_local_cdk_asset_images_safely() -> dict[str, Any]: 

109 """``prune_cdk_asset_images`` that also absorbs unexpected exceptions.""" 

110 try: 

111 return prune_cdk_asset_images() 

112 except Exception as exc: # noqa: BLE001 - disk reclamation is best-effort by contract 

113 return { 

114 "runtime": None, 

115 "removed_images": [], 

116 "dangling_pruned": False, 

117 "errors": [f"{type(exc).__name__}: {exc}"[:_MAX_ERROR_CHARS]], 

118 }