diff --git a/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/__init__.py b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/__init__.py index bf5ff6129..711e7630c 100644 --- a/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/__init__.py +++ b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/__init__.py @@ -3,6 +3,7 @@ from jumpstarter_cli_common.opt import opt_log_level from jumpstarter_cli_common.version import version +from .apply import apply from .create import create from .delete import delete from .get import get @@ -17,6 +18,7 @@ def admin(): admin.add_command(get) +admin.add_command(apply) admin.add_command(create) admin.add_command(delete) admin.add_command(import_res) diff --git a/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/apply.py b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/apply.py new file mode 100644 index 000000000..9d3d2290a --- /dev/null +++ b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/apply.py @@ -0,0 +1,95 @@ +from http import HTTPStatus +from typing import IO, Optional + +import click +from jumpstarter_cli_common.blocking import blocking +from jumpstarter_cli_common.opt import ( + OutputType, + opt_context, + opt_kubeconfig, + opt_namespace, + opt_output_all, +) +from jumpstarter_cli_common.print import model_print +from jumpstarter_kubernetes import ApplyV1Alpha1Api, ManifestError, load_manifests +from kubernetes_asyncio.client.exceptions import ApiException +from kubernetes_asyncio.config.config_exception import ConfigException + +from .k8s import ( + handle_k8s_api_exception, + handle_k8s_config_exception, +) + + +@click.command("apply") +@click.option( + "-f", + "--filename", + "filenames", + type=click.File("r"), + multiple=True, + required=True, + help="Manifest to apply, or '-' to read from stdin. Can be set multiple times.", +) +@click.option( + "--force-conflicts", + is_flag=True, + default=False, + help="Take ownership of fields another manager holds, instead of reporting the conflict.", +) +@click.option( + "--dry-run", + is_flag=True, + default=False, + help="Send the manifests to the server for validation without persisting them.", +) +@opt_namespace +@opt_kubeconfig +@opt_context +@opt_output_all +@blocking +async def apply( + filenames: tuple[IO, ...], + force_conflicts: bool, + dry_run: bool, + namespace: str, + kubeconfig: Optional[str], + context: Optional[str], + output: OutputType, +): + """Apply Jumpstarter manifests to a Kubernetes cluster + + Creates the resources in a manifest, or updates them in place when they + already exist. Only Jumpstarter resources (clients, exporters, exporter + sets, virtual target classes) can be applied. + """ + try: + manifests = [] + for file in filenames: + manifests.extend(load_manifests(file.read(), file.name)) + except ManifestError as e: + raise click.ClickException(str(e)) from e + + try: + async with ApplyV1Alpha1Api(namespace, kubeconfig, context) as api: + applied = await api.apply_all(manifests, dry_run=dry_run, force_conflicts=force_conflicts) + except ManifestError as e: + raise click.ClickException(str(e)) from e + except ApiException as e: + try: + handle_k8s_api_exception(e) + except click.ClickException as error: + if e.status == HTTPStatus.CONFLICT and not force_conflicts: + # The conflict names the fields; say how to win them on purpose. + error.message += "\nRe-run with --force-conflicts to take ownership of those fields." + raise + except ConfigException as e: + handle_k8s_config_exception(e) + + if output is None: + # Mirror kubectl: one line per resource saying what happened to it. + suffix = " (dry run)" if dry_run else "" + for resource in applied.items: + click.echo(f"{resource.qualified_name} {resource.action}{suffix}") + else: + model_print(applied, output, namespace=namespace) diff --git a/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/apply_test.py b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/apply_test.py new file mode 100644 index 000000000..b1763208c --- /dev/null +++ b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/apply_test.py @@ -0,0 +1,203 @@ +import json +from unittest.mock import AsyncMock, patch + +from click.testing import CliRunner +from jumpstarter_kubernetes import ( + ApplyV1Alpha1Api, + V1Alpha1AppliedResource, + V1Alpha1AppliedResourceList, +) +from kubernetes_asyncio.client.exceptions import ApiException + +from .apply import apply + +CLIENT_MANIFEST = """\ +apiVersion: jumpstarter.dev/v1alpha1 +kind: Client +metadata: + name: hello +""" + +EXPORTER_SET_MANIFEST = """\ +apiVersion: virtualtarget.jumpstarter.dev/v1alpha1 +kind: ExporterSet +metadata: + name: pool +""" + + +def applied(kind: str, name: str, action: str, api_version: str = "jumpstarter.dev/v1alpha1"): + return V1Alpha1AppliedResource( + apiVersion=api_version, + kind=kind, + name=name, + namespace="default", + action=action, + resource={"apiVersion": api_version, "kind": kind, "metadata": {"name": name}}, + ) + + +def run(args, files: dict[str, str], apply_all: AsyncMock, input: str | None = None): + runner = CliRunner() + with runner.isolated_filesystem(): + for filename, contents in files.items(): + with open(filename, "w") as f: + f.write(contents) + with ( + patch.object(ApplyV1Alpha1Api, "_load_kube_config", AsyncMock()), + patch.object(ApplyV1Alpha1Api, "apply_all", apply_all), + ): + return runner.invoke(apply, args, input=input) + + +def test_apply_reports_what_happened_to_each_resource(): + apply_all = AsyncMock( + return_value=V1Alpha1AppliedResourceList( + items=[ + applied("Client", "hello", "created"), + applied("ExporterSet", "pool", "configured", "virtualtarget.jumpstarter.dev/v1alpha1"), + ] + ) + ) + + result = run( + ["-f", "client.yaml", "-f", "set.yaml"], + {"client.yaml": CLIENT_MANIFEST, "set.yaml": EXPORTER_SET_MANIFEST}, + apply_all, + ) + + assert result.exit_code == 0 + assert "client.jumpstarter.dev/hello created" in result.output + assert "exporterset.virtualtarget.jumpstarter.dev/pool configured" in result.output + # Both files reach the cluster in the order they were given. + assert [m["kind"] for m in apply_all.call_args.args[0]] == ["Client", "ExporterSet"] + assert apply_all.call_args.kwargs["dry_run"] is False + + +def test_apply_reads_a_multi_document_manifest(): + apply_all = AsyncMock( + return_value=V1Alpha1AppliedResourceList( + items=[ + applied("Client", "hello", "created"), + applied("ExporterSet", "pool", "created", "virtualtarget.jumpstarter.dev/v1alpha1"), + ] + ) + ) + + result = run(["-f", "both.yaml"], {"both.yaml": f"{CLIENT_MANIFEST}---\n{EXPORTER_SET_MANIFEST}"}, apply_all) + + assert result.exit_code == 0 + assert [m["kind"] for m in apply_all.call_args.args[0]] == ["Client", "ExporterSet"] + + +def test_apply_says_when_nothing_was_persisted(): + apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "created")])) + + result = run(["-f", "client.yaml", "--dry-run"], {"client.yaml": CLIENT_MANIFEST}, apply_all) + + assert result.exit_code == 0 + assert "client.jumpstarter.dev/hello created (dry run)" in result.output + assert apply_all.call_args.kwargs["dry_run"] is True + + +def test_apply_prints_names_only(): + apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "created")])) + + result = run(["-f", "client.yaml", "-o", "name"], {"client.yaml": CLIENT_MANIFEST}, apply_all) + + assert result.exit_code == 0 + assert result.output.strip() == "client.jumpstarter.dev/hello" + + +def test_apply_prints_the_stored_resources_as_json(): + apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "created")])) + + result = run(["-f", "client.yaml", "-o", "json"], {"client.yaml": CLIENT_MANIFEST}, apply_all) + + assert result.exit_code == 0 + parsed = json.loads(result.output) + assert parsed["items"][0]["action"] == "created" + assert parsed["items"][0]["resource"]["metadata"]["name"] == "hello" + + +def test_apply_reads_a_manifest_from_stdin(): + apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "created")])) + + result = run(["-f", "-"], {}, apply_all, input=CLIENT_MANIFEST) + + assert result.exit_code == 0 + assert [m["kind"] for m in apply_all.call_args.args[0]] == ["Client"] + + +def test_apply_rejects_a_manifest_that_is_not_jumpstarters(): + apply_all = AsyncMock() + + result = run( + ["-f", "secret.yaml"], + {"secret.yaml": "apiVersion: v1\nkind: Secret\nmetadata:\n name: creds\n"}, + apply_all, + ) + + assert result.exit_code != 0 + assert "core Kubernetes resource" in result.output + # Nothing reaches the cluster when the manifest is refused. + apply_all.assert_not_awaited() + + +def test_apply_requires_a_manifest(): + result = run([], {}, AsyncMock()) + + assert result.exit_code != 0 + assert "Missing option" in result.output + + +CONFLICT_BODY = json.dumps( + { + "kind": "Status", + "reason": "Conflict", + "message": 'Apply failed with 1 conflict: conflict with "other-controller": .metadata.labels.owner', + } +) + + +def conflict() -> ApiException: + error = ApiException(status=409, reason="Conflict") + error.body = CONFLICT_BODY + return error + + +def test_apply_says_how_to_resolve_a_field_ownership_conflict(): + apply_all = AsyncMock(side_effect=conflict()) + + result = run(["-f", "client.yaml"], {"client.yaml": CLIENT_MANIFEST}, apply_all) + + assert result.exit_code != 0 + assert ".metadata.labels.owner" in result.output + assert "--force-conflicts" in result.output + + +def test_apply_does_not_suggest_a_flag_that_is_already_set(): + apply_all = AsyncMock(side_effect=conflict()) + + result = run(["--force-conflicts", "-f", "client.yaml"], {"client.yaml": CLIENT_MANIFEST}, apply_all) + + assert result.exit_code != 0 + assert "--force-conflicts" not in result.output + + +def test_apply_asks_the_cluster_to_take_ownership_when_told_to(): + apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "configured")])) + + result = run(["--force-conflicts", "-f", "client.yaml"], {"client.yaml": CLIENT_MANIFEST}, apply_all) + + assert result.exit_code == 0 + assert apply_all.call_args.kwargs["force_conflicts"] is True + + +def test_apply_leaves_conflicting_fields_alone_by_default(): + apply_all = AsyncMock(return_value=V1Alpha1AppliedResourceList(items=[applied("Client", "hello", "configured")])) + + result = run(["-f", "client.yaml"], {"client.yaml": CLIENT_MANIFEST}, apply_all) + + assert result.exit_code == 0 + assert apply_all.call_args.kwargs["force_conflicts"] is False diff --git a/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/k8s.py b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/k8s.py index 8e3b567d9..6f9524942 100644 --- a/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/k8s.py +++ b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/k8s.py @@ -10,10 +10,21 @@ def handle_k8s_api_exception(e: ApiException): # Try to parse the JSON response try: json_body = json.loads(e.body) - raise click.ClickException(f"Error from server ({json_body['reason']}): {json_body['message']}") from e - except json.decoder.JSONDecodeError: + except (json.decoder.JSONDecodeError, TypeError): raise click.ClickException(f"Server error: {e.body}") from e + # Valid JSON is not necessarily a Status: a proxy in front of the API server + # can answer with a bare string, a list, or null. + if not isinstance(json_body, dict): + raise click.ClickException(f"Server error: {e.body}") from e + + # Not every Status carries a reason: a 500 from an admission or schema + # check often has only a message, and losing it leaves nothing to act on. + message = json_body.get("message") or e.reason or "unknown error" + reason = json_body.get("reason") + prefix = f"Error from server ({reason})" if reason else "Error from server" + raise click.ClickException(f"{prefix}: {message}") from e + def handle_k8s_config_exception(e: ConfigException): """Handle a Kubernetes config exception""" diff --git a/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/k8s_test.py b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/k8s_test.py new file mode 100644 index 000000000..d3c9b4a4f --- /dev/null +++ b/python/packages/jumpstarter-cli-admin/jumpstarter_cli_admin/k8s_test.py @@ -0,0 +1,57 @@ +import click +import pytest +from kubernetes_asyncio.client.exceptions import ApiException + +from .k8s import handle_k8s_api_exception + + +def api_exception(status: int, body, reason: str | None = None) -> ApiException: + error = ApiException(status=status, reason=reason) + error.body = body + return error + + +def test_reports_the_reason_and_message_from_the_server(): + error = api_exception(404, '{"reason":"NotFound","message":"clients.jumpstarter.dev \\"nope\\" not found"}') + + with pytest.raises(click.ClickException, match=r"Error from server \(NotFound\): .*not found"): + handle_k8s_api_exception(error) + + +def test_reports_a_status_that_carries_no_reason(): + # Schema and admission failures answer with a message and nothing else. + error = api_exception(500, '{"kind":"Status","message":".provisioner: field not declared in schema","code":500}') + + with pytest.raises(click.ClickException, match="Error from server: .provisioner: field not declared in schema"): + handle_k8s_api_exception(error) + + +def test_falls_back_to_the_http_reason_when_the_body_says_nothing(): + error = api_exception(503, "{}", reason="Service Unavailable") + + with pytest.raises(click.ClickException, match="Error from server: Service Unavailable"): + handle_k8s_api_exception(error) + + +def test_passes_through_a_body_that_is_not_json(): + error = api_exception(502, "bad gateway") + + with pytest.raises(click.ClickException, match="Server error: bad gateway"): + handle_k8s_api_exception(error) + + +@pytest.mark.parametrize("body", ["null", '"just a string"', "[1,2,3]"]) +def test_passes_through_json_that_is_not_a_status(body): + # A proxy in front of the API server can answer with valid JSON that is not + # a Status object; reading it as one used to raise AttributeError. + error = api_exception(502, body) + + with pytest.raises(click.ClickException, match="Server error: "): + handle_k8s_api_exception(error) + + +def test_passes_through_a_missing_body(): + error = api_exception(500, None) + + with pytest.raises(click.ClickException, match="Server error: None"): + handle_k8s_api_exception(error) diff --git a/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/__init__.py b/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/__init__.py index dd785de75..48b6487af 100644 --- a/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/__init__.py +++ b/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/__init__.py @@ -1,3 +1,10 @@ +from .apply import ( + ApplyV1Alpha1Api, + ManifestError, + V1Alpha1AppliedResource, + V1Alpha1AppliedResourceList, + load_manifests, +) from .clients import ClientsV1Alpha1Api, V1Alpha1Client, V1Alpha1ClientList, V1Alpha1ClientStatus from .cluster import ( check_jumpstarter_installation, @@ -37,6 +44,11 @@ from .list import V1Alpha1List __all__ = [ + "ApplyV1Alpha1Api", + "ManifestError", + "V1Alpha1AppliedResource", + "V1Alpha1AppliedResourceList", + "load_manifests", "ClientsV1Alpha1Api", "V1Alpha1Client", "V1Alpha1ClientList", diff --git a/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/apply.py b/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/apply.py new file mode 100644 index 000000000..ee63436cd --- /dev/null +++ b/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/apply.py @@ -0,0 +1,327 @@ +"""Apply Jumpstarter manifests to a cluster. + +This is the write half of ``jmp admin get``: it takes the YAML for a Client, +Exporter, ExporterSet, VirtualTargetClass or any other Jumpstarter custom +resource and sends it to the cluster with a server-side apply, so a manifest +can be created and later re-applied without the caller tracking what changed. + +Only resources in the ``jumpstarter.dev`` API groups are accepted. This is a +Jumpstarter admin tool, not a general-purpose ``kubectl apply``. +""" + +import logging +from typing import Literal, Optional + +import yaml +from kubernetes_asyncio.client.exceptions import ApiException +from pydantic import Field + +from .exceptions import JumpstarterKubernetesError +from .json import JsonBaseModel +from .list import V1Alpha1List +from .util import AbstractAsyncCustomObjectApi + +logger = logging.getLogger(__name__) + +CORE_API_GROUP = "jumpstarter.dev" + +# Identifies this tool as the owner of the fields it applies, so a later apply +# of the same manifest can remove a field it no longer sets. +FIELD_MANAGER = "jumpstarter-admin" + +APPLY_PATCH_CONTENT_TYPE = "application/apply-patch+yaml" + + +class ManifestError(JumpstarterKubernetesError): + """Raised when a manifest cannot be applied as written.""" + + +def is_jumpstarter_group(group: str) -> bool: + """True for the Jumpstarter API group and its subgroups.""" + return group == CORE_API_GROUP or group.endswith("." + CORE_API_GROUP) + + +def validate_manifest(document: object, origin: str) -> dict: + """Check that a parsed YAML document is a Jumpstarter resource we can apply.""" + if not isinstance(document, dict): + raise ManifestError(f"{origin}: expected a resource, got {type(document).__name__}") + + api_version = document.get("apiVersion") + if not isinstance(api_version, str) or api_version == "": + raise ManifestError(f"{origin}: missing apiVersion") + group, _, version = api_version.partition("/") + if version == "": + raise ManifestError( + f"{origin}: apiVersion '{api_version}' is a core Kubernetes resource, " + "only Jumpstarter resources can be applied" + ) + if not is_jumpstarter_group(group): + raise ManifestError( + f"{origin}: apiVersion '{api_version}' is not a Jumpstarter API group, " + f"expected {CORE_API_GROUP} or a subgroup of it" + ) + + kind = document.get("kind") + if not isinstance(kind, str) or kind == "": + raise ManifestError(f"{origin}: missing kind") + + metadata = document.get("metadata") + if not isinstance(metadata, dict): + raise ManifestError(f"{origin}: missing metadata") + name = metadata.get("name") + if not isinstance(name, str) or name == "": + raise ManifestError(f"{origin}: missing metadata.name") + + return document + + +def load_manifests(source: str, origin: str = "manifest") -> list[dict]: + """Parse a YAML stream of Jumpstarter resources, empty documents skipped.""" + try: + documents = list(yaml.safe_load_all(source)) + except yaml.YAMLError as e: + raise ManifestError(f"{origin} is not valid YAML: {e}") from e + + resources = [] + for index, document in enumerate(documents): + if document is None: + continue + # Only number the documents when there is more than one to point at. + label = origin if len(documents) == 1 else f"{origin} (document {index + 1})" + resources.append(validate_manifest(document, label)) + + if len(resources) == 0: + raise ManifestError(f"{origin} contains no resources") + return resources + + +class V1Alpha1AppliedResource(JsonBaseModel): + """One resource as the cluster stored it, and what applying it did.""" + + api_version: str = Field(alias="apiVersion") + kind: str + name: str + namespace: Optional[str] = None + action: Literal["created", "configured", "unchanged"] + resource: dict + + @property + def qualified_name(self) -> str: + group = self.api_version.partition("/")[0] + return f"{self.kind.lower()}.{group}/{self.name}" + + @classmethod + def rich_add_columns(cls, table, **kwargs): + table.add_column("NAME", no_wrap=True) + table.add_column("KIND") + table.add_column("NAMESPACE") + table.add_column("ACTION") + + def rich_add_rows(self, table, **kwargs): + table.add_row(self.name, self.kind, self.namespace or "", self.action) + + def rich_add_names(self, names): + names.append(self.qualified_name) + + +class V1Alpha1AppliedResourceList(V1Alpha1List[V1Alpha1AppliedResource]): + kind: Literal["List"] = Field(default="List") + + @classmethod + def rich_add_columns(cls, table, **kwargs): + V1Alpha1AppliedResource.rich_add_columns(table, **kwargs) + + def rich_add_rows(self, table, **kwargs): + for applied in self.items: + applied.rich_add_rows(table, **kwargs) + + def rich_add_names(self, names): + for applied in self.items: + applied.rich_add_names(names) + + +class ApplyV1Alpha1Api(AbstractAsyncCustomObjectApi): + """Apply Jumpstarter manifests of any kind the cluster serves.""" + + def __init__(self, namespace: str, config_file: Optional[str] = None, context: Optional[str] = None): + super().__init__(namespace, config_file, context) + self._resources: dict[tuple[str, str, str], tuple[str, bool]] = {} + + async def _resolve_resource(self, group: str, version: str, kind: str) -> tuple[str, bool]: + """Look up a kind's plural name and scope in the cluster's discovery data. + + Discovery keeps this working for kinds this client was never taught, + which is what lets one command apply Clients, ExporterSets and whatever + the operator adds next. + """ + cached = self._resources.get((group, version, kind)) + if cached is not None: + return cached + + try: + listing = await self._client.call_api( + f"/apis/{group}/{version}", + "GET", + auth_settings=["BearerToken"], + header_params={"Accept": "application/json"}, + response_types_map={200: "object"}, + _return_http_data_only=True, + ) + except ApiException as e: + if e.status == 404: + raise ManifestError( + f"the cluster does not serve {group}/{version}, check that the Jumpstarter CRDs are installed" + ) from e + raise + + for resource in listing.get("resources", []): + # Subresources ("exportersets/status") are not kinds of their own. + if "/" in resource.get("name", ""): + continue + if resource.get("kind") == kind: + found = (resource["name"], bool(resource.get("namespaced", True))) + self._resources[(group, version, kind)] = found + return found + + raise ManifestError(f"the cluster does not serve kind {kind} in {group}/{version}") + + async def _read(self, group, version, plural, name, namespace) -> Optional[dict]: + try: + if namespace is None: + return await self.api.get_cluster_custom_object(group=group, version=version, plural=plural, name=name) + return await self.api.get_namespaced_custom_object( + group=group, version=version, plural=plural, name=name, namespace=namespace + ) + except ApiException as e: + if e.status == 404: + return None + raise + + async def _server_side_apply( + self, group, version, plural, name, namespace, manifest: dict, *, dry_run: bool, force: bool + ) -> dict: + """PATCH a resource with a server-side apply, creating it if needed. + + The request is built by hand because the generated custom object client + only deserializes a 200 response, and an apply that creates a resource + answers 201. + """ + if namespace is None: + path = "/apis/{group}/{version}/{plural}/{name}" + path_params = {"group": group, "version": version, "plural": plural, "name": name} + else: + path = "/apis/{group}/{version}/namespaces/{namespace}/{plural}/{name}" + path_params = { + "group": group, + "version": version, + "namespace": namespace, + "plural": plural, + "name": name, + } + + query_params = [("fieldManager", FIELD_MANAGER)] + if force: + # Take ownership of fields another manager holds. Off by default, + # as in kubectl: a conflict means something else is managing the + # field, which the caller should see rather than silently win. + query_params.append(("force", "true")) + if dry_run: + query_params.append(("dryRun", "All")) + + return await self._client.call_api( + path, + "PATCH", + path_params=path_params, + query_params=query_params, + header_params={"Accept": "application/json", "Content-Type": APPLY_PATCH_CONTENT_TYPE}, + body=manifest, + auth_settings=["BearerToken"], + response_types_map={200: "object", 201: "object"}, + _return_http_data_only=True, + ) + + async def apply( + self, manifest: dict, *, dry_run: bool = False, force_conflicts: bool = False + ) -> V1Alpha1AppliedResource: + """Server-side apply one resource and report what it did.""" + validate_manifest(manifest, f"{manifest.get('kind', 'resource')} manifest") + api_version = manifest["apiVersion"] + group, _, version = api_version.partition("/") + kind = manifest["kind"] + name = manifest["metadata"]["name"] + + plural, namespaced = await self._resolve_resource(group, version, kind) + namespace = None + if namespaced: + namespace = manifest["metadata"].get("namespace") or self.namespace + # The body has to agree with the URL the request is sent to. + manifest = {**manifest, "metadata": {**manifest["metadata"], "namespace": namespace}} + elif manifest["metadata"].get("namespace"): + raise ManifestError(f"{kind} is cluster scoped, remove metadata.namespace") + + existing = await self._read(group, version, plural, name, namespace) + applied = await self._server_side_apply( + group, version, plural, name, namespace, manifest, dry_run=dry_run, force=force_conflicts + ) + + return V1Alpha1AppliedResource( + apiVersion=api_version, + kind=kind, + name=name, + namespace=namespace, + action=_action_taken(existing, applied, dry_run=dry_run), + resource=applied, + ) + + async def apply_all( + self, manifests: list[dict], *, dry_run: bool = False, force_conflicts: bool = False + ) -> V1Alpha1AppliedResourceList: + """Apply resources in the order they were written.""" + applied = [] + for manifest in manifests: + applied.append(await self.apply(manifest, dry_run=dry_run, force_conflicts=force_conflicts)) + return V1Alpha1AppliedResourceList(items=applied) + + +def _action_taken( + existing: Optional[dict], applied: dict, *, dry_run: bool = False +) -> Literal["created", "configured", "unchanged"]: + """Describe an apply by what the cluster did, or would do, with it. + + A persisted apply that changes nothing leaves the resource version alone, + which is how an unchanged re-apply is told apart from a real update. A dry + run is never persisted, so its resource version never moves either — there + the answer has to come from comparing what the server says the object would + become against what it is now. + """ + if existing is None: + return "created" + if dry_run: + return "unchanged" if _same_content(existing, applied) else "configured" + before = existing.get("metadata", {}).get("resourceVersion") + after = applied.get("metadata", {}).get("resourceVersion") + if before is not None and before == after: + return "unchanged" + return "configured" + + +# Set by the server on every write, so they say nothing about whether the +# manifest changed anything. +_SERVER_OWNED_METADATA = frozenset( + {"resourceVersion", "generation", "managedFields", "creationTimestamp", "uid", "selfLink"} +) + + +def _same_content(before: dict, after: dict) -> bool: + """Whether two versions of a resource differ in anything the user wrote.""" + return _without_server_fields(before) == _without_server_fields(after) + + +def _without_server_fields(obj: dict) -> dict: + trimmed = {key: value for key, value in obj.items() if key != "status"} + trimmed["metadata"] = { + key: value + for key, value in (obj.get("metadata") or {}).items() + if key not in _SERVER_OWNED_METADATA + } + return trimmed diff --git a/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/apply_test.py b/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/apply_test.py new file mode 100644 index 000000000..b3fc2be1c --- /dev/null +++ b/python/packages/jumpstarter-kubernetes/jumpstarter_kubernetes/apply_test.py @@ -0,0 +1,305 @@ +from unittest.mock import AsyncMock, Mock + +import pytest +from kubernetes_asyncio.client.exceptions import ApiException + +from .apply import ( + APPLY_PATCH_CONTENT_TYPE, + FIELD_MANAGER, + ApplyV1Alpha1Api, + ManifestError, + load_manifests, + validate_manifest, +) + +CLIENT_MANIFEST = """ +apiVersion: jumpstarter.dev/v1alpha1 +kind: Client +metadata: + name: hello +""" + +EXPORTER_SET_MANIFEST = """ +apiVersion: virtualtarget.jumpstarter.dev/v1alpha1 +kind: ExporterSet +metadata: + name: pool +spec: + minReplicas: 1 +""" + + +def test_load_manifests_reads_every_document(): + manifests = load_manifests(f"{CLIENT_MANIFEST}\n---\n{EXPORTER_SET_MANIFEST}") + assert [m["kind"] for m in manifests] == ["Client", "ExporterSet"] + + +def test_load_manifests_skips_empty_documents(): + manifests = load_manifests(f"---\n\n---\n{CLIENT_MANIFEST}\n---\n# just a comment\n") + assert [m["kind"] for m in manifests] == ["Client"] + + +def test_load_manifests_rejects_an_empty_stream(): + with pytest.raises(ManifestError, match="contains no resources"): + load_manifests("\n# nothing here\n", "empty.yaml") + + +def test_load_manifests_rejects_invalid_yaml(): + with pytest.raises(ManifestError, match="not valid YAML"): + load_manifests("kind: [unterminated", "broken.yaml") + + +def test_load_manifests_names_the_document_that_failed(): + with pytest.raises(ManifestError, match=r"pair\.yaml \(document 2\): missing kind"): + load_manifests(f"{CLIENT_MANIFEST}\n---\napiVersion: jumpstarter.dev/v1alpha1\n", "pair.yaml") + + +@pytest.mark.parametrize( + ("document", "message"), + [ + ("a string", "expected a resource"), + ({"kind": "Client", "metadata": {"name": "x"}}, "missing apiVersion"), + ({"apiVersion": "v1", "kind": "Secret", "metadata": {"name": "x"}}, "core Kubernetes resource"), + ({"apiVersion": "apps/v1", "kind": "Deployment", "metadata": {"name": "x"}}, "not a Jumpstarter API group"), + ( + # A group that merely ends in the same letters is not a subgroup. + {"apiVersion": "notjumpstarter.dev/v1", "kind": "Client", "metadata": {"name": "x"}}, + "not a Jumpstarter API group", + ), + ({"apiVersion": "jumpstarter.dev/v1alpha1", "metadata": {"name": "x"}}, "missing kind"), + ({"apiVersion": "jumpstarter.dev/v1alpha1", "kind": "Client"}, "missing metadata"), + ({"apiVersion": "jumpstarter.dev/v1alpha1", "kind": "Client", "metadata": {}}, "missing metadata.name"), + ], +) +def test_validate_manifest_rejects(document, message): + with pytest.raises(ManifestError, match=message): + validate_manifest(document, "manifest") + + +def test_validate_manifest_accepts_a_subgroup(): + document = { + "apiVersion": "virtualtarget.jumpstarter.dev/v1alpha1", + "kind": "ExporterSet", + "metadata": {"name": "p"}, + } + assert validate_manifest(document, "manifest") is document + + +DISCOVERY = { + "resources": [ + {"name": "clients/status", "kind": "Client", "namespaced": True}, + {"name": "clients", "kind": "Client", "namespaced": True}, + {"name": "virtualtargetclasses", "kind": "VirtualTargetClass", "namespaced": False}, + ] +} + + +def make_api(*, discovery=None, existing=None, applied=None) -> ApplyV1Alpha1Api: + """An apply API wired to a fake cluster instead of a real one.""" + api = ApplyV1Alpha1Api("default") + api._client = Mock() + + async def call_api(path, method, **kwargs): + if method == "GET": + return discovery if discovery is not None else DISCOVERY + return applied or {"metadata": {"name": "hello", "resourceVersion": "2"}} + + api._client.call_api = AsyncMock(side_effect=call_api) + api.api = Mock() + api.api.get_namespaced_custom_object = AsyncMock(return_value=existing) + api.api.get_cluster_custom_object = AsyncMock(return_value=existing) + if existing is None: + api.api.get_namespaced_custom_object.side_effect = ApiException(status=404) + api.api.get_cluster_custom_object.side_effect = ApiException(status=404) + return api + + +def patch_call(api: ApplyV1Alpha1Api): + """The PATCH the API sent, as (path, path_params, query_params, body).""" + for call in api._client.call_api.await_args_list: + if call.args[1] == "PATCH": + return call + raise AssertionError("no apply request was sent") + + +@pytest.mark.asyncio +async def test_apply_creates_a_resource_the_cluster_does_not_have(): + api = make_api() + + applied = await api.apply(load_manifests(CLIENT_MANIFEST)[0]) + + assert applied.action == "created" + assert applied.qualified_name == "client.jumpstarter.dev/hello" + assert applied.namespace == "default" + call = patch_call(api) + assert call.args[0] == "/apis/{group}/{version}/namespaces/{namespace}/{plural}/{name}" + assert call.kwargs["path_params"] == { + "group": "jumpstarter.dev", + "version": "v1alpha1", + "namespace": "default", + "plural": "clients", + "name": "hello", + } + assert call.kwargs["header_params"]["Content-Type"] == APPLY_PATCH_CONTENT_TYPE + assert ("fieldManager", FIELD_MANAGER) in call.kwargs["query_params"] + # Conflicts are reported, not won, unless the caller asks for it. + assert not any(param == "force" for param, _ in call.kwargs["query_params"]) + assert not any(param == "dryRun" for param, _ in call.kwargs["query_params"]) + # The body has to carry the namespace it is being sent to. + assert call.kwargs["body"]["metadata"]["namespace"] == "default" + # An apply that creates a resource answers 201, which has to deserialize. + assert call.kwargs["response_types_map"] == {200: "object", 201: "object"} + + +@pytest.mark.asyncio +async def test_apply_reports_a_resource_it_changed_as_configured(): + api = make_api(existing={"metadata": {"name": "hello", "resourceVersion": "1"}}) + + applied = await api.apply(load_manifests(CLIENT_MANIFEST)[0]) + + assert applied.action == "configured" + + +@pytest.mark.asyncio +async def test_apply_reports_an_unchanged_resource_as_unchanged(): + # A server-side apply that changes nothing leaves the resource version be. + api = make_api( + existing={"metadata": {"name": "hello", "resourceVersion": "7"}}, + applied={"metadata": {"name": "hello", "resourceVersion": "7"}}, + ) + + applied = await api.apply(load_manifests(CLIENT_MANIFEST)[0]) + + assert applied.action == "unchanged" + + +@pytest.mark.asyncio +async def test_apply_keeps_the_namespace_the_manifest_asks_for(): + api = make_api() + manifest = load_manifests(CLIENT_MANIFEST)[0] + manifest["metadata"]["namespace"] = "lab" + + applied = await api.apply(manifest) + + assert applied.namespace == "lab" + assert patch_call(api).kwargs["path_params"]["namespace"] == "lab" + + +@pytest.mark.asyncio +async def test_apply_takes_ownership_of_conflicting_fields_when_asked(): + api = make_api() + + await api.apply(load_manifests(CLIENT_MANIFEST)[0], force_conflicts=True) + + assert ("force", "true") in patch_call(api).kwargs["query_params"] + + +@pytest.mark.asyncio +async def test_apply_passes_a_dry_run_through_to_the_server(): + api = make_api() + + await api.apply(load_manifests(CLIENT_MANIFEST)[0], dry_run=True) + + assert ("dryRun", "All") in patch_call(api).kwargs["query_params"] + + +@pytest.mark.asyncio +async def test_a_dry_run_reports_a_change_the_resource_version_cannot_show(): + # Nothing is persisted, so the resource version stays put even though the + # manifest would change the labels. The answer has to come from the content. + api = make_api( + existing={"metadata": {"name": "hello", "resourceVersion": "7", "labels": {"env": "dev"}}}, + applied={"metadata": {"name": "hello", "resourceVersion": "7", "labels": {"env": "prod"}}}, + ) + + applied = await api.apply(load_manifests(CLIENT_MANIFEST)[0], dry_run=True) + + assert applied.action == "configured" + + +@pytest.mark.asyncio +async def test_a_dry_run_ignores_fields_the_server_owns(): + api = make_api( + existing={ + "metadata": {"name": "hello", "resourceVersion": "7", "generation": 3, "managedFields": []}, + "spec": {"username": "hello"}, + "status": {"credential": {"name": "hello-client"}}, + }, + applied={ + "metadata": {"name": "hello", "resourceVersion": "7", "generation": 4, "managedFields": [{"a": 1}]}, + "spec": {"username": "hello"}, + "status": {}, + }, + ) + + applied = await api.apply(load_manifests(CLIENT_MANIFEST)[0], dry_run=True) + + assert applied.action == "unchanged" + + +@pytest.mark.asyncio +async def test_apply_uses_the_cluster_scoped_endpoint_for_a_cluster_scoped_kind(): + api = make_api() + manifest = { + "apiVersion": "jumpstarter.dev/v1alpha1", + "kind": "VirtualTargetClass", + "metadata": {"name": "qemu"}, + } + + applied = await api.apply(manifest) + + assert applied.namespace is None + call = patch_call(api) + assert call.args[0] == "/apis/{group}/{version}/{plural}/{name}" + assert call.kwargs["path_params"] == { + "group": "jumpstarter.dev", + "version": "v1alpha1", + "plural": "virtualtargetclasses", + "name": "qemu", + } + + +@pytest.mark.asyncio +async def test_apply_rejects_a_namespace_on_a_cluster_scoped_kind(): + api = make_api() + manifest = { + "apiVersion": "jumpstarter.dev/v1alpha1", + "kind": "VirtualTargetClass", + "metadata": {"name": "qemu", "namespace": "lab"}, + } + + with pytest.raises(ManifestError, match="cluster scoped"): + await api.apply(manifest) + + +@pytest.mark.asyncio +async def test_apply_rejects_a_kind_the_cluster_does_not_serve(): + api = make_api() + manifest = { + "apiVersion": "jumpstarter.dev/v1alpha1", + "kind": "Imaginary", + "metadata": {"name": "nope"}, + } + + with pytest.raises(ManifestError, match="does not serve kind Imaginary"): + await api.apply(manifest) + + +@pytest.mark.asyncio +async def test_apply_explains_a_missing_api_group(): + api = make_api() + api._client.call_api.side_effect = ApiException(status=404) + + with pytest.raises(ManifestError, match="CRDs are installed"): + await api.apply(load_manifests(CLIENT_MANIFEST)[0]) + + +@pytest.mark.asyncio +async def test_apply_all_looks_up_each_kind_once(): + api = make_api() + + applied = await api.apply_all(load_manifests(f"{CLIENT_MANIFEST}\n---\n{CLIENT_MANIFEST}")) + + assert [item.action for item in applied.items] == ["created", "created"] + discoveries = [call for call in api._client.call_api.await_args_list if call.args[1] == "GET"] + assert len(discoveries) == 1