|
| 1 | +# Copyright 2024 Canonical Ltd. |
| 2 | +# |
| 3 | +# This program is free software; you can redistribute it and/or |
| 4 | +# modify it under the terms of the GNU Lesser General Public |
| 5 | +# License version 3 as published by the Free Software Foundation. |
| 6 | +# |
| 7 | +# This program is distributed in the hope that it will be useful, |
| 8 | +# but WITHOUT ANY WARRANTY; without even the implied warranty of |
| 9 | +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| 10 | +# Lesser General Public License for more details. |
| 11 | +# |
| 12 | +# You should have received a copy of the GNU Lesser General Public License |
| 13 | +# along with this program; if not, write to the Free Software Foundation, |
| 14 | +# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. |
| 15 | +# |
| 16 | + |
| 17 | +"""Utilities for use in snap hooks. |
| 18 | +
|
| 19 | +A base instance's full name may look like this: |
| 20 | + base-instance-whatevercraft-buildd-base-v7-craft-com.ubuntu.cloud-buildd-daily-core24 |
| 21 | +
|
| 22 | +From that, the thing we care most about is the compatibility tag: |
| 23 | + whatevercraft-buildd-base-v7 |
| 24 | +""" |
| 25 | + |
| 26 | +import dataclasses |
| 27 | +import json |
| 28 | +import re |
| 29 | +import subprocess |
| 30 | +import sys |
| 31 | +from typing import Any |
| 32 | + |
| 33 | +from typing_extensions import Self |
| 34 | + |
| 35 | +from craft_providers import Base, lxd |
| 36 | + |
| 37 | +_BASE_INSTANCE_START_STRING = "base-instance" |
| 38 | +_CURRENT_COMPATIBILITY_TAG_REGEX = re.compile( |
| 39 | + f"^{_BASE_INSTANCE_START_STRING}.*-{Base.compatibility_tag}-.*" |
| 40 | +) |
| 41 | + |
| 42 | + |
| 43 | +class HookError(Exception): |
| 44 | + """Hook logic cannot continue. Hooks themselves should not exit nonzero.""" |
| 45 | + |
| 46 | + |
| 47 | +@dataclasses.dataclass |
| 48 | +class LXDInstance: |
| 49 | + """Represents an lxc instance.""" |
| 50 | + |
| 51 | + name: str |
| 52 | + expanded_config: dict[str, str] |
| 53 | + |
| 54 | + def base_instance_name(self) -> str: |
| 55 | + """Get the full name of the base instance this instance was created from.""" |
| 56 | + try: |
| 57 | + return self.expanded_config["image.description"] |
| 58 | + except KeyError as e: |
| 59 | + # Unexpected, cannot continue |
| 60 | + raise HookError("Could not get full base name from {self.name}") from e |
| 61 | + |
| 62 | + def is_current_base_instance(self) -> bool: |
| 63 | + """Return true if this is a base instance with the current compat tag.""" |
| 64 | + return bool(re.match(_CURRENT_COMPATIBILITY_TAG_REGEX, self.name)) |
| 65 | + |
| 66 | + def is_base_instance(self) -> bool: |
| 67 | + """Return true if this is a base instance.""" |
| 68 | + return self.name.startswith(_BASE_INSTANCE_START_STRING) |
| 69 | + |
| 70 | + @classmethod |
| 71 | + def unmarshal( |
| 72 | + cls, |
| 73 | + src: dict[str, str], |
| 74 | + ) -> Self: |
| 75 | + """Use this rather than init - the lxc output has a lot of extra fields.""" |
| 76 | + return cls( |
| 77 | + **{ # type: ignore[arg-type] |
| 78 | + k: v |
| 79 | + for k, v in src.items() |
| 80 | + if k in {f.name for f in dataclasses.fields(cls)} |
| 81 | + } |
| 82 | + ) |
| 83 | + |
| 84 | + |
| 85 | +class HookHelper: |
| 86 | + """Hook business logic.""" |
| 87 | + |
| 88 | + def __init__(self, *, project_name: str, simulate: bool, debug: bool) -> None: |
| 89 | + self.simulate = simulate |
| 90 | + self.debug = debug |
| 91 | + self._project_name = project_name |
| 92 | + |
| 93 | + self._check_has_lxd() |
| 94 | + self._check_project_exists() |
| 95 | + |
| 96 | + def _check_has_lxd(self) -> None: |
| 97 | + """Check if LXD is installed before doing anything. |
| 98 | +
|
| 99 | + On recent Ubuntu systems, "lxc" might be "/usr/sbin/lxc", which is provided by the |
| 100 | + "lxd-installer" package and will install the LXD snap if it's not installed. This |
| 101 | + installation can then take a long time if the store is having issues. For the |
| 102 | + purposes of the configure and remove hooks we don't want to install LXD just to |
| 103 | + check that it has no stale images. |
| 104 | + """ |
| 105 | + if not lxd.is_installed(): |
| 106 | + raise HookError("LXD is not installed.") |
| 107 | + |
| 108 | + def _check_project_exists(self) -> None: |
| 109 | + """Raise HookError if lxc doesn't know about this app.""" |
| 110 | + for project in self.lxc("project", "list", proj=False): |
| 111 | + if project["name"] == self._project_name: |
| 112 | + return |
| 113 | + |
| 114 | + # Didn't find our project name |
| 115 | + raise HookError(f"Project {self._project_name} does not exist in LXD.") |
| 116 | + |
| 117 | + def dprint(self, *args: Any, **kwargs: Any) -> None: # noqa: ANN401 |
| 118 | + """Print messages to stderr if debug=True. |
| 119 | +
|
| 120 | + Can treat this like normal print(), except can also pass an instance |
| 121 | + dict as the first argument for some automatic formatting. |
| 122 | + """ |
| 123 | + if not self.debug: |
| 124 | + return |
| 125 | + if "file" not in kwargs: |
| 126 | + kwargs["file"] = sys.stderr |
| 127 | + |
| 128 | + print_args = list(args) |
| 129 | + if len(args) >= 1 and isinstance(args[0], LXDInstance): |
| 130 | + # First arg quacks like an instance object |
| 131 | + instance = print_args.pop(0) |
| 132 | + print_args += [":", instance.name] |
| 133 | + |
| 134 | + print(*print_args, **kwargs) |
| 135 | + |
| 136 | + def lxc( |
| 137 | + self, |
| 138 | + *args: Any, # noqa: ANN401 |
| 139 | + fail_msg: str | None = None, |
| 140 | + proj: bool = True, |
| 141 | + json_out: bool = True, |
| 142 | + ) -> Any: # noqa: ANN401 |
| 143 | + """Run lxc commands specified in *args. |
| 144 | +
|
| 145 | + :param fail_msg: Print this if the command returns nonzero. |
| 146 | + :param proj: Set to False to not specify lxc project. |
| 147 | + :param json_out: If set to False, don't ask lxc for JSON output. |
| 148 | + """ |
| 149 | + lxc_args = ["lxc"] |
| 150 | + if json_out: |
| 151 | + lxc_args += ["--format", "json"] |
| 152 | + if proj: |
| 153 | + lxc_args += ["--project", self._project_name] |
| 154 | + lxc_args += args |
| 155 | + |
| 156 | + try: |
| 157 | + out = subprocess.run( |
| 158 | + lxc_args, |
| 159 | + check=True, |
| 160 | + text=True, |
| 161 | + capture_output=True, |
| 162 | + ).stdout |
| 163 | + except FileNotFoundError: |
| 164 | + raise HookError("LXD is not installed.") |
| 165 | + except subprocess.CalledProcessError as e: |
| 166 | + if not fail_msg: |
| 167 | + fail_msg = e.stderr |
| 168 | + raise HookError(fail_msg) |
| 169 | + else: |
| 170 | + if not json_out: |
| 171 | + return out |
| 172 | + try: |
| 173 | + return json.loads(out) |
| 174 | + except json.decoder.JSONDecodeError as e: |
| 175 | + raise HookError(f"Didn't get back JSON: {out}") from e |
| 176 | + |
| 177 | + def delete_instance(self, instance: LXDInstance) -> None: |
| 178 | + """Delete the specified lxc instance.""" |
| 179 | + print( |
| 180 | + f" > Removing instance {instance.name} in LXD {self._project_name} project..." |
| 181 | + ) |
| 182 | + if self.simulate: |
| 183 | + return |
| 184 | + self.lxc( |
| 185 | + "delete", |
| 186 | + "--force", |
| 187 | + instance.name, |
| 188 | + fail_msg=f"Failed to remove LXD instance {instance.name}.", |
| 189 | + json_out=False, |
| 190 | + ) |
| 191 | + |
| 192 | + def _delete_image(self, image_fingerprint: str) -> None: |
| 193 | + """Remove the image.""" |
| 194 | + self.lxc("image", "delete", image_fingerprint, json_out=False) |
| 195 | + |
| 196 | + def delete_all_images(self) -> None: |
| 197 | + """Delete all images of the lxc project.""" |
| 198 | + for image_fingerprint in self._list_images(): |
| 199 | + self._delete_image(image_fingerprint) |
| 200 | + |
| 201 | + def delete_project(self) -> None: |
| 202 | + """Delete this lxc project.""" |
| 203 | + print(f"Removing project {self._project_name}") |
| 204 | + if self.simulate: |
| 205 | + return |
| 206 | + self.lxc( |
| 207 | + "project", |
| 208 | + "delete", |
| 209 | + self._project_name, |
| 210 | + proj=False, |
| 211 | + json_out=False, |
| 212 | + ) |
| 213 | + |
| 214 | + def _list_images(self) -> list[str]: |
| 215 | + """Return fingerprints of all images associated with the lxc project.""" |
| 216 | + return [image["fingerprint"] for image in self.lxc("image", "list")] |
| 217 | + |
| 218 | + def list_instances(self) -> list[LXDInstance]: |
| 219 | + """Return a list of all instance objects for the project.""" |
| 220 | + return [LXDInstance.unmarshal(instance) for instance in self.lxc("list")] |
| 221 | + |
| 222 | + def list_base_instances(self) -> list[LXDInstance]: |
| 223 | + """Return a list of all base instance objects for the project.""" |
| 224 | + base_instances = [] |
| 225 | + for instance in self.list_instances(): |
| 226 | + if not instance.is_base_instance(): |
| 227 | + self.dprint(instance, "Not a base instance") |
| 228 | + continue |
| 229 | + |
| 230 | + base_instances.append(instance) |
| 231 | + return base_instances |
| 232 | + |
| 233 | + |
| 234 | +def configure_hook(lxc: HookHelper) -> None: |
| 235 | + """Cleanup hook run on snap configure.""" |
| 236 | + # Keep the newest base instance with the most recent compatibility tag. |
| 237 | + delete_base_full_names = set() |
| 238 | + for instance in lxc.list_base_instances(): |
| 239 | + if instance.is_current_base_instance(): |
| 240 | + lxc.dprint(instance, "Base instance is current") |
| 241 | + continue |
| 242 | + |
| 243 | + # This is a base instance but it doesn't match the compat tag, assume it's |
| 244 | + # old (not future) and delete it. |
| 245 | + lxc.dprint(instance, "Base instance uses old compatibility tag, deleting") |
| 246 | + lxc.delete_instance(instance) |
| 247 | + delete_base_full_names.add(instance.base_instance_name()) |
| 248 | + |
| 249 | + if not delete_base_full_names: |
| 250 | + lxc.dprint("No base instances were deleted, so no derived instances to delete") |
| 251 | + return |
| 252 | + |
| 253 | + # Find the child instances of the bases we deleted and delete them too |
| 254 | + did_delete = False |
| 255 | + for instance in lxc.list_instances(): |
| 256 | + if instance.base_instance_name() not in delete_base_full_names: |
| 257 | + continue |
| 258 | + lxc.dprint(instance, "Base instance was deleted, deleting derived instance") |
| 259 | + lxc.delete_instance(instance) |
| 260 | + did_delete = True |
| 261 | + if not did_delete: |
| 262 | + lxc.dprint("Found no instances derived from deleted base instances") |
| 263 | + |
| 264 | + |
| 265 | +def remove_hook(lxc: HookHelper) -> None: |
| 266 | + """Cleanup hook run on snap removal.""" |
| 267 | + for instance in lxc.list_instances(): |
| 268 | + lxc.delete_instance(instance) |
| 269 | + |
| 270 | + # Project deletion will fail if images aren't all deleted first |
| 271 | + lxc.delete_all_images() |
| 272 | + |
| 273 | + lxc.delete_project() |
0 commit comments