27152425961

Committed 08 Jun 2026 04:38PM UTC coverage: 82.105% (+0.06%) from 82.044%

Build # 27152425961

Build Type

Pull #108

github

Committed by

web-flow

Commit Message

Merge ed0b797b1 into 837aa7bf6

Pull Request Pull Request #108: Fix test-only JSON patches treated as non-empty mutations

Coverage Stats

1081 of 1496 branches covered (72.26%)

Branch coverage included in aggregate %.

4 of 4 new or added lines in 1 file covered. (100.0%)

2 existing lines in 1 file now uncovered.

4443 of 5232 relevant lines covered (84.92%)

4.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.35

/src/main/python/kubernator/plugins/k8s_api.py

# -*- coding: utf-8 -*-
#
#   Copyright 2020 Express Systems USA, Inc
#   Copyright 2021 Karellen, Inc.
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.
#

import json
import re
from collections import namedtuple
from collections.abc import Callable, Mapping, MutableMapping, Sequence
from enum import Enum, auto
from functools import partial, wraps
from pathlib import Path
from typing import Union, Optional

import yaml
from jsonschema.exceptions import ValidationError

from kubernator.api import load_file, FileType, load_remote_file, calling_frame_source, parse_yaml_docs


def api_exc_normalize_body(e):
    """Parse a raw ApiException body (JSON or YAML) into a Python object in-place.

    Idempotent: if e.body is already a parsed object, it is left unchanged.
    """
    if not isinstance(e.body, (str, bytes)):
        return
    if e.headers and "content-type" in e.headers:
        content_type = e.headers["content-type"]
        if content_type == "application/json" or content_type.endswith("+json"):
            e.body = json.loads(e.body)
        elif (content_type in ("application/yaml", "application/x-yaml", "text/yaml",
                               "text/x-yaml") or content_type.endswith("+yaml")):
            e.body = yaml.safe_load(e.body)


def api_exc_format_body(e):
    """Format an ApiException body back to a string for human-readable display.

    Idempotent: if e.body is already a string/bytes, it is left unchanged.
    """
    if not isinstance(e.body, (str, bytes)):
        e.body = json.dumps(e.body, indent=4)


def _normalize_api_exc(func):
    """Decorator: normalize ApiException body before propagating."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        from kubernetes.client import ApiException
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            api_exc_normalize_body(e)
            raise
    return wrapper


K8S_WARNING_HEADER = re.compile(r'(?:,\s*)?(\d{3})\s+(\S+)\s+"(.+?)(?<!\\)"(?:\s+\"(.+?)(?<!\\)\")?\s*')
UPPER_FOLLOWED_BY_LOWER_RE = re.compile(r"(.)([A-Z][a-z]+)")
LOWER_OR_NUM_FOLLOWED_BY_UPPER_RE = re.compile(r"([a-z0-9])([A-Z])")

CLUSTER_RESOURCE_PATH = re.compile(r"^/apis?/(?:[^/]+/){1,2}([^/]+)$")
NAMESPACED_RESOURCE_PATH = re.compile(r"^/apis?/(?:[^/]+/){1,2}namespaces/[^/]+/([^/]+)$")

PROJECT_ANNOTATION = "kubernator.io/project"


class K8SResourcePatchType(Enum):
    JSON_PATCH = auto()
    SERVER_SIDE_PATCH = auto()


class K8SPropagationPolicy(Enum):
    BACKGROUND = ("Background",)
    FOREGROUND = ("Foreground",)
    ORPHAN = ("Orphan",)

    def __init__(self, policy):
        self.policy = policy


def to_group_and_version(api_version):
    group, _, version = api_version.partition("/")
    if not version:
        version = group
        group = ""
    return group, version


def to_k8s_resource_def_key(manifest):
    return K8SResourceDefKey(*to_group_and_version(manifest["apiVersion"]),
                             manifest["kind"])


class K8SResourceDefKey(namedtuple("K8SResourceDefKey", ["group", "version", "kind"])):
    __slots__ = ()

    def __str__(self):
        return f"{self.group}{'/' if self.group else '/'}{self.version}/{self.kind}"


class K8SResourceDef:
    def __init__(self, key, singular, plural, namespaced, custom, schema):
        self.key = key
        self.singular = singular
        self.plural = plural
        self.namespaced = namespaced
        self.custom = custom
        self.schema = schema

        self._api_get = None
        self._api_create = None
        self._api_patch = None
        self._api_delete = None
        self._api_list = None

    @property
    def group(self) -> str:
        return self.key.group

    @property
    def version(self) -> str:
        return self.key.version

    @property
    def kind(self) -> str:
        return self.key.kind

    @property
    def has_api(self) -> bool:
        return self.custom or self.plural

    @property
    def get(self):
        return self._api_get

    @property
    def create(self):
        return self._api_create

    @property
    def patch(self):
        return self._api_patch

    @property
    def delete(self):
        return self._api_delete

    @property
    def list(self):
        return self._api_list

    def __eq__(self, o: object) -> bool:
        if not isinstance(o, K8SResourceDef):
            return False

        return (self.key == o.key and
                self.singular == o.singular and
                self.plural == o.plural and
                self.namespaced == o.namespaced and
                self.custom == o.custom)

    def __hash__(self) -> int:
        return self.key.__hash__()

    def __str__(self):
        return f"{self.key=}, {self.singular=}, {self.plural=}, {self.namespaced=}, {self.custom=}"

    @classmethod
    def from_manifest(cls, key: K8SResourceDefKey,
                      schema,
                      paths: Mapping[K8SResourceDefKey, Mapping[str, Mapping]]):
        singular = key.kind.lower()

        plural = None
        namespaced = False

        if singular == "namespace":
            plural = "namespaces"
        else:
            for path in paths.get(key, ()):
                if m := NAMESPACED_RESOURCE_PATH.fullmatch(path):
                    plural = m[1]
                    namespaced = True
                    break
                elif m := CLUSTER_RESOURCE_PATH.fullmatch(path):
                    plural = m[1]

        yield K8SResourceDef(key, singular, plural, namespaced, False, schema)

    @classmethod
    def from_resource(cls, resource: "K8SResource"):
        manifest = resource.manifest
        spec = manifest["spec"]
        group = spec["group"]
        names = spec["names"]
        kind = names["kind"]
        singular = names.get("singular", names["kind"].lower())
        plural = names["plural"]
        namespaced = spec["scope"] == "Namespaced"

        for version_spec in spec["versions"]:
            version = version_spec["name"]
            if resource.version == "v1":
                schema = version_spec["schema"]["openAPIV3Schema"]
            else:
                schema = spec["validation"]["openAPIV3Schema"]
            yield K8SResourceDef(K8SResourceDefKey(group, version, kind), singular, plural, namespaced, True, schema)

    def populate_api(self, k8s_client_module, k8s_client):
        if not self.has_api:
            raise RuntimeError(f"{self} has no API")

        if self._api_get:
            return

        group = self.group or "core"
        version = self.version
        kind = self.kind

        if self.custom:
            k8s_api = k8s_client_module.CustomObjectsApi(k8s_client)

            kwargs = {"group": group,
                      "version": version,
                      "plural": self.plural}
            if self.namespaced:
                self._api_get = partial(k8s_api.get_namespaced_custom_object, **kwargs)
                self._api_patch = partial(k8s_api.patch_namespaced_custom_object, **kwargs)
                self._api_create = partial(k8s_api.create_namespaced_custom_object, **kwargs)
                self._api_delete = partial(k8s_api.delete_namespaced_custom_object, **kwargs)
                self._api_list = partial(k8s_api.list_namespaced_custom_object, **kwargs)
            else:
                self._api_get = partial(k8s_api.get_cluster_custom_object, **kwargs)
                self._api_patch = partial(k8s_api.patch_cluster_custom_object, **kwargs)
                self._api_create = partial(k8s_api.create_cluster_custom_object, **kwargs)
                self._api_delete = partial(k8s_api.delete_cluster_custom_object, **kwargs)
                self._api_list = partial(k8s_api.list_cluster_custom_object, **kwargs)
        else:
            # Take care for the case e.g. api_type is "apiextensions.k8s.io"
            # Only replace the last instance
            group = "".join(group.rsplit(".k8s.io", 1))

            # convert group name from DNS subdomain format to
            # python class name convention
            group = "".join(word.capitalize() for word in group.split('.'))
            fcn_to_call = f"{group}{version.capitalize()}Api"
            k8s_api = getattr(k8s_client_module, fcn_to_call)(k8s_client)

            # Replace CamelCased action_type into snake_case
            kind = UPPER_FOLLOWED_BY_LOWER_RE.sub(r"\1_\2", kind)
            kind = LOWER_OR_NUM_FOLLOWED_BY_UPPER_RE.sub(r"\1_\2", kind).lower()

            if self.namespaced:
                self._api_get = getattr(k8s_api, f"read_namespaced_{kind}")
                self._api_patch = getattr(k8s_api, f"patch_namespaced_{kind}")
                self._api_create = getattr(k8s_api, f"create_namespaced_{kind}")
                self._api_delete = getattr(k8s_api, f"delete_namespaced_{kind}")
                self._api_list = getattr(k8s_api, f"list_namespaced_{kind}")
            else:
                self._api_get = getattr(k8s_api, f"read_{kind}")
                self._api_patch = getattr(k8s_api, f"patch_{kind}")
                self._api_create = getattr(k8s_api, f"create_{kind}")
                self._api_delete = getattr(k8s_api, f"delete_{kind}")
                self._api_list = getattr(k8s_api, f"list_{kind}")


class K8SResourceKey(namedtuple("K8SResourceKey", ["group", "kind", "name", "namespace"])):
    __slots__ = ()

    def __str__(self):
        return (f"{self.group}{'/' if self.group else 'v1/'}{self.kind}"
                f"/{self.name}{f'.{self.namespace}' if self.namespace else ''}")


class K8SResource:
    _k8s_client_version = None
    _k8s_field_validation = None
    _k8s_field_validation_patched = None
    _logger = None
    _api_warnings = None

    def __init__(self, manifest: dict, rdef: K8SResourceDef, source: Union[str, Path] = None):
        self.key = self.get_manifest_key(manifest)

        self.manifest = manifest
        self.rdef = rdef
        self.source = source

    @property
    def group(self) -> str:
        return self.key.group

    @property
    def version(self) -> str:
        return self.rdef.version

    @property
    def kind(self) -> str:
        return self.key.kind

    @property
    def name(self) -> str:
        return self.key.name

    @name.setter
    def name(self, value):
        self.manifest["metadata"]["name"] = value
        self.key = self.get_manifest_key(self.manifest)

    @property
    def namespace(self) -> Optional[str]:
        return self.key.namespace

    @namespace.setter
    def namespace(self, value):
        self.manifest["metadata"]["namespace"] = value
        self.key = self.get_manifest_key(self.manifest)

    @property
    def api_version(self) -> str:
        return self.manifest["apiVersion"]

    @property
    def schema(self) -> dict:
        return self.rdef.schema

    @property
    def is_crd(self):
        return self.group == "apiextensions.k8s.io" and self.kind == "CustomResourceDefinition"

    @property
    def project(self) -> Optional[str]:
        metadata = self.manifest.get("metadata") or {}
        annotations = metadata.get("annotations") or {}
        return annotations.get(PROJECT_ANNOTATION)

    def __str__(self):
        return f"{self.api_version}/{self.kind}/{self.name}{'.' + self.namespace if self.namespace else ''}"

    @_normalize_api_exc
    def get(self):
        rdef = self.rdef
        kwargs = {"name": self.name,
                  "_preload_content": False}
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        return json.loads(self.rdef.get(**kwargs).data)

    @_normalize_api_exc
    def create(self, dry_run=True):
        rdef = self.rdef
        kwargs = {"body": self.manifest,
                  "_preload_content": False,
                  "field_manager": "kubernator",
                  }

        # `and not self.rdef.custom` to be removed after solving https://github.com/kubernetes-client/gen/issues/259
        if self._k8s_field_validation_patched or not self.rdef.custom:
            kwargs["field_validation"] = self._k8s_field_validation
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        if dry_run:
            kwargs["dry_run"] = "All"
        resp = rdef.create(**kwargs)
        self._process_response_headers(resp)
        return json.loads(resp.data)

    @_normalize_api_exc
    def patch(self, json_patch, *, patch_type: K8SResourcePatchType, force=False, dry_run=True):
        rdef = self.rdef
        kwargs = {"name": self.name,
                  "body": json_patch,
                  "_preload_content": False,
                  "field_manager": "kubernator",
                  }

        # `and not self.rdef.custom` to be removed after solving https://github.com/kubernetes-client/gen/issues/259
        if self._k8s_field_validation_patched or not self.rdef.custom:
            kwargs["field_validation"] = self._k8s_field_validation
        if patch_type == K8SResourcePatchType.SERVER_SIDE_PATCH:
            kwargs["force"] = force
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        if dry_run:
            kwargs["dry_run"] = "All"

        def select_header_content_type_patch(content_types, method=None, body=None):
            if patch_type == K8SResourcePatchType.JSON_PATCH:
                return "application/json-patch+json"
            if patch_type == K8SResourcePatchType.SERVER_SIDE_PATCH:
                return "application/apply-patch+yaml"
            raise NotImplementedError

        if isinstance(rdef.patch, partial):
            api_client = rdef.patch.func.__self__.api_client
        else:
            api_client = rdef.patch.__self__.api_client

        old_func = api_client.select_header_content_type
        try:
            api_client.select_header_content_type = select_header_content_type_patch
            resp = rdef.patch(**kwargs)
            self._process_response_headers(resp)
            return json.loads(resp.data)
        finally:
            api_client.select_header_content_type = old_func

    @_normalize_api_exc
    def delete(self, *, dry_run=True, propagation_policy=K8SPropagationPolicy.BACKGROUND, wait=True):
        from kubernetes.client import ApiException
        rdef = self.rdef
        kwargs = {"name": self.name,
                  "_preload_content": False,
                  "propagation_policy": propagation_policy.policy
                  }
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        if dry_run:
            kwargs["dry_run"] = "All"

        result = json.loads(rdef.delete(**kwargs).data)

        if wait and not dry_run:
            # Wait for the resource to actually disappear by watching for the
            # DELETED event. If the resource is already gone before the watch
            # opens (race), the watch produces no events; loop with a short
            # server-side timeout and re-check existence via get() to break out.
            while True:
                for event in self.watch(timeout_seconds=10):
                    if event["type"] == "DELETED":
                        return result
                try:
                    self.get()
                except ApiException as e:
                    if e.status == 404:
                        return result
                    raise

        return result

    def watch(self, *, timeout_seconds=None):
        from kubernetes import watch as k8s_watch
        rdef = self.rdef
        kwargs = {"field_selector": f"metadata.name={self.name}"}
        if rdef.namespaced:
            kwargs["namespace"] = self.namespace
        if timeout_seconds is not None:
            kwargs["timeout_seconds"] = timeout_seconds
        return k8s_watch.Watch().stream(rdef.list, **kwargs)

    @staticmethod
    def get_manifest_key(manifest):
        return K8SResourceKey(to_group_and_version(manifest["apiVersion"])[0],
                              manifest["kind"],
                              manifest["metadata"]["name"],
                              manifest["metadata"].get("namespace"))

    @staticmethod
    def get_manifest_description(manifest: dict, source=None):
        api_version = manifest.get("apiVersion")
        kind = manifest.get("kind")
        metadata = manifest.get("metadata")
        name = None
        namespace = None
        if metadata:
            name = metadata.get("name")
            namespace = metadata.get("namespace")
        return (f"{api_version or 'unknown'}/{kind or '<unknown>'}/"
                f"{name or '<unknown>'}{'.' + namespace if namespace else ''}")

    def __eq__(self, other):
        if not isinstance(other, K8SResource):
            return False
        return self.key == other.key and self.manifest == other.manifest

    def _process_response_headers(self, resp):
        headers = resp.headers
        warn_headers = headers.get("Warning")
        if warn_headers:
            for warn in K8S_WARNING_HEADER.findall(warn_headers):
                code, _, msg, _ = warn
                code = int(code)
                msg = msg.encode("utf-8").decode("unicode_escape")
                if code == 299:
                    self._api_warnings(self, msg)
                else:
                    self._logger.warning("Unknown API warning received for resource %s from %s: code %d: %s",
                                         self, self.source, code, msg)


class K8SResourcePluginMixin:
    def __init__(self):
        self.validator = None
        self.resources: MutableMapping[K8SResourceKey, K8SResource] = {}

    def _require_validator(self):
        if self.validator is None:
            raise RuntimeError(
                "K8S validator not initialised; handle_start() must run first")
        return self.validator

    @property
    def resource_definitions(self) -> MutableMapping[K8SResourceDefKey, K8SResourceDef]:
        return self._require_validator().resource_definitions

    @property
    def resource_paths(self) -> MutableMapping[K8SResourceDefKey, MutableMapping[str, dict]]:
        return self._require_validator().resource_paths

    def add_resources(self, manifests: Union[str, list, dict], source: Union[str, Path] = None):
        if not source:
            source = calling_frame_source()

        if isinstance(manifests, str):
            manifests = list(parse_yaml_docs(manifests, source))

        if isinstance(manifests, (Mapping, dict)):
            return self.add_resource(manifests, source)
        else:
            return [self.add_resource(m, source) for m in manifests if m]

    def add_resource(self, manifest: dict, source: Union[str, Path] = None):
        if not source:
            source = calling_frame_source()
        resource = self._create_resource(manifest, source)

        try:
            trans_resource = self._transform_resource(list(self.resources.values()), resource)
        except Exception as e:
            self.logger.error("An error occurred running transformers on %s", resource, exc_info=e)
            raise

        errors = list(self._validate_resource(trans_resource.manifest, source))
        if errors:
            for error in errors:
                if source:
                    self.logger.error("Error detected in re-transformed K8S resource %s generated through %s",
                                      trans_resource, source, exc_info=error)
            raise errors[0]

        return self._add_resource(trans_resource, source)

    def add_crds(self, manifests: Union[str, list, dict], source: Union[str, Path] = None):
        if not source:
            source = calling_frame_source()

        if isinstance(manifests, str):
            manifests = list(parse_yaml_docs(manifests, source))

        if isinstance(manifests, (Mapping, dict)):
            return self.add_crd(manifests, source)
        else:
            return [self.add_crd(m, source) for m in manifests if m]

    def add_crd(self, manifest: dict, source: Union[str, Path] = None):
        if not source:
            source = calling_frame_source()
        resource = self._create_resource(manifest, source)
        if not resource.is_crd:
            resource_description = K8SResource.get_manifest_description(manifest, source)
            raise ValueError(f"K8S manifest {resource_description} from {source} is not a CRD")

        self._add_crd(resource)
        return resource

    def create_resource(self, manifest: dict, source: Union[str, Path] = None):
        """Create K8S resource without adding it"""
        if not source:
            source = calling_frame_source()

        return self._create_resource(manifest, source)

    def add_local_resources(self, path: Path, file_type: FileType, source: str = None):
        manifests = load_file(self.logger, path, file_type)

        return [self.add_resource(m, source or path) for m in manifests if m]

    def add_remote_resources(self, url: str, file_type: FileType, *, sub_category: Optional[str] = None,
                             source: str = None):
        manifests = load_remote_file(self.logger, url, file_type, sub_category=sub_category)

        return [self.add_resource(m, source or url) for m in manifests if m]

    def add_local_crds(self, path: Path, file_type: FileType, source: str = None):
        manifests = load_file(self.logger, path, file_type)

        return [self.add_crd(m, source or path) for m in manifests if m]

    def add_remote_crds(self, url: str, file_type: FileType, *, sub_category: Optional[str] = None,
                        source: str = None):
        manifests = load_remote_file(self.logger, url, file_type, sub_category=sub_category)

        return [self.add_crd(m, source or url) for m in manifests if m]

    def get_api_versions(self):
        return self.validator.api_versions()

    def _create_resource(self, manifest: dict, source: Union[str, Path] = None):
        resource_description = K8SResource.get_manifest_description(manifest, source)

        new_manifest = self._patch_manifest(manifest, resource_description)
        if new_manifest != manifest:
            manifest = new_manifest
            resource_description = K8SResource.get_manifest_description(manifest, source)

        self.logger.debug("Validating K8S manifest for %s", resource_description)
        errors = list(self._validate_resource(manifest, source))
        if errors:
            for error in errors:
                self.logger.error("Error detected in K8S manifest %s from %s: \n%s",
                                  resource_description, source or "<unknown>", yaml.safe_dump(manifest, None),
                                  exc_info=error)
            raise errors[0]

        rdef = self._get_manifest_rdef(manifest)
        return K8SResource(manifest, rdef, source)

    def _add_resource(self, resource: K8SResource, source):
        if resource.key in self.resources:
            existing_resource = self.resources[resource.key]
            if resource != existing_resource:
                raise ValidationError("resource %s from %s already exists and was added from %s" %
                                      (resource.key, resource.source, existing_resource.source))
            self.logger.trace("K8S resource for %s from %s is already present and is identical", resource, source)
            return existing_resource

        self.logger.info("Adding K8S resource for %s from %s", resource, source)
        self.resources[resource.key] = resource

        if resource.is_crd:
            self._add_crd(resource)

        return resource

    def _patch_manifest(self,
                        manifest: dict,
                        resource_description: str):
        return manifest

    def _transform_resource(self,
                            resources: Sequence[K8SResource],
                            resource: K8SResource) -> K8SResource:
        return resource

    def _filter_resources(self, func: Callable[[K8SResource], bool]):
        yield from filter(func, self.resources.values())

    def _validate_resource(self, manifest: dict, source: Union[str, Path] = None):
        yield from self._require_validator().iter_manifest_errors(manifest)

    def _get_manifest_rdef(self, manifest):
        return self._require_validator().get_manifest_rdef(manifest)

    def _add_crd(self, resource: K8SResource):
        for crd in K8SResourceDef.from_resource(resource):
            self.logger.info("Adding K8S CRD definition %s", crd.key)
            self.resource_definitions[crd.key] = crd

karellen / kubernator / 27152425961

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous