18180360618

Committed 02 Oct 2025 01:15AM UTC coverage: 75.914% (-0.1%) from 76.019%

Build # 18180360618

Build Type

push

github

Committed by

arcivanov

Commit Message

Release 1.0.21

Run Details

615 of 961 branches covered (64.0%)

Branch coverage included in aggregate %.

2395 of 3004 relevant lines covered (79.73%)

4.78 hits per line

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

79.82

/src/main/python/kubernator/plugins/k8s.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 logging
import re
import sys
import types
from collections.abc import Mapping
from functools import partial
from importlib.metadata import version as pkg_version
from pathlib import Path
from typing import Iterable, Callable, Sequence

import jsonpatch
import yaml

from kubernator.api import (KubernatorPlugin,
                            Globs,
                            scan_dir,
                            load_file,
                            FileType,
                            load_remote_file,
                            StripNL,
                            install_python_k8s_client)
from kubernator.merge import extract_merge_instructions, apply_merge_instructions
from kubernator.plugins.k8s_api import (K8SResourcePluginMixin,
                                        K8SResource,
                                        K8SResourcePatchType,
                                        K8SPropagationPolicy)

logger = logging.getLogger("kubernator.k8s")
proc_logger = logger.getChild("proc")
stdout_logger = StripNL(proc_logger.info)
stderr_logger = StripNL(proc_logger.warning)

FIELD_VALIDATION_STRICT_MARKER = "strict decoding error: "
VALID_FIELD_VALIDATION = ("Ignore", "Warn", "Strict")


def final_resource_validator(resources: Sequence[K8SResource],
                             resource: K8SResource,
                             error: Callable[..., Exception]) -> Iterable[Exception]:
    final_key = resource.get_manifest_key(resource.manifest)
    if final_key != resource.key:
        yield error("Illegal change of identifiers of the resource "
                    "%s from %s have been changed to %s",
                    resource.key, resource.source, final_key)

    if resource.rdef.namespaced and not resource.namespace:
        yield error("Namespaced resource %s from %s is missing the required namespace",
                    resource, resource.source)


def normalize_pkg_version(v: str):
    v_split = v.split(".")
    rev = v_split[-1]
    if not rev.isdigit():
        new_rev = ""
        for c in rev:
            if not c.isdigit():
                break
            new_rev += c
        v_split[-1] = new_rev
    return tuple(map(int, v_split))


class KubernetesPlugin(KubernatorPlugin, K8SResourcePluginMixin):
    logger = logger

    _name = "k8s"

    def __init__(self):
        super().__init__()
        self.context = None

        self.embedded_pkg_version = self._get_kubernetes_client_version()

        self._transformers = []
        self._validators = []
        self._summary = 0, 0, 0

    def set_context(self, context):
        self.context = context

    def register(self,
                 field_validation="Warn",
                 field_validation_warn_fatal=True,
                 disable_client_patches=False):
        self.context.app.register_plugin("kubeconfig")

        if field_validation not in VALID_FIELD_VALIDATION:
            raise ValueError("'field_validation' must be one of %s" % (", ".join(VALID_FIELD_VALIDATION)))

        context = self.context
        context.globals.k8s = dict(patch_field_excludes=("^/metadata/managedFields",
                                                         "^/metadata/generation",
                                                         "^/metadata/creationTimestamp",
                                                         "^/metadata/resourceVersion",
                                                         ),
                                   immutable_changes={("apps", "DaemonSet"): K8SPropagationPolicy.BACKGROUND,
                                                      ("apps", "StatefulSet"): K8SPropagationPolicy.ORPHAN,
                                                      ("apps", "Deployment"): K8SPropagationPolicy.ORPHAN,
                                                      ("storage.k8s.io", "StorageClass"): K8SPropagationPolicy.ORPHAN,
                                                      },
                                   default_includes=Globs(["*.yaml", "*.yml"], True),
                                   default_excludes=Globs([".*"], True),
                                   add_resources=self.add_resources,
                                   load_resources=self.api_load_resources,
                                   load_remote_resources=self.api_load_remote_resources,
                                   load_crds=self.api_load_crds,
                                   load_remote_crds=self.api_load_remote_crds,
                                   add_transformer=self.api_add_transformer,
                                   remove_transformer=self.api_remove_transformer,
                                   add_validator=self.api_remove_validator,
                                   get_api_versions=self.get_api_versions,
                                   create_resource=self.create_resource,
                                   disable_client_patches=disable_client_patches,
                                   field_validation=field_validation,
                                   field_validation_warn_fatal=field_validation_warn_fatal,
                                   field_validation_warnings=0,
                                   _k8s=self,
                                   )
        context.k8s = dict(default_includes=Globs(context.globals.k8s.default_includes),
                           default_excludes=Globs(context.globals.k8s.default_excludes)
                           )
        self.api_add_validator(final_resource_validator)

    def handle_init(self):
        pass

    def handle_start(self):
        self.context.kubeconfig.register_change_notifier(self._kubeconfig_changed)
        self.setup_client()

    def _kubeconfig_changed(self):
        self.setup_client()

    def _get_kubernetes_client_version(self):
        return pkg_version("kubernetes").split(".")

    def setup_client(self):
        k8s = self.context.k8s
        if "server_version" not in k8s:
            self._setup_client()

        server_minor = k8s.server_version[1]

        logger.info("Using Kubernetes client version =~%s.0 for server version %s",
                    server_minor, ".".join(k8s.server_version))
        pkg_dir = install_python_k8s_client(self.context.app.run_passthrough_capturing, server_minor, logger,
                                            stdout_logger, stderr_logger, k8s.disable_client_patches)

        modules_to_delete = []
        for k, v in sys.modules.items():
            if k == "kubernetes" or k.startswith("kubernetes."):
                modules_to_delete.append(k)
        for k in modules_to_delete:
            del sys.modules[k]

        logger.info("Adding sys.path reference to %s", pkg_dir)
        sys.path.insert(0, str(pkg_dir))
        self.embedded_pkg_version = self._get_kubernetes_client_version()
        logger.info("Switching to Kubernetes client version %s", ".".join(self.embedded_pkg_version))
        self._setup_client()

        logger.debug("Reading Kubernetes OpenAPI spec for %s", k8s.server_git_version)

        k8s_def = load_remote_file(logger, f"https://raw.githubusercontent.com/kubernetes/kubernetes/"
                                           f"{k8s.server_git_version}/api/openapi-spec/swagger.json",
                                   FileType.JSON)
        self.resource_definitions_schema = k8s_def

        self._populate_resource_definitions()

    def _setup_client(self):
        from kubernetes import client

        context = self.context
        k8s = context.k8s

        k8s.client = self._setup_k8s_client()
        version = client.VersionApi(k8s.client).get_code()
        if "-eks-" in version.git_version:
            git_version = version.git_version.split("-")[0]
        else:
            git_version = version.git_version

        k8s.server_version = git_version[1:].split(".")
        k8s.server_git_version = git_version

        logger.info("Found Kubernetes %s on %s", k8s.server_git_version, k8s.client.configuration.host)

        K8SResource._k8s_client_version = normalize_pkg_version(pkg_version("kubernetes"))
        K8SResource._k8s_field_validation = k8s.field_validation
        K8SResource._k8s_field_validation_patched = not k8s.disable_client_patches
        K8SResource._logger = self.logger
        K8SResource._api_warnings = self._api_warnings

    def _api_warnings(self, resource, warn):
        k8s = self.context.k8s
        self.context.globals.k8s.field_validation_warnings += 1

        log = self.logger.warning
        if k8s.field_validation_warn_fatal:
            log = self.logger.error

        log("FAILED FIELD VALIDATION on resource %s from %s: %s", resource, resource.source, warn)

    def handle_before_dir(self, cwd: Path):
        context = self.context
        context.k8s.default_includes = Globs(context.k8s.default_includes)
        context.k8s.default_excludes = Globs(context.k8s.default_excludes)
        context.k8s.includes = Globs(context.k8s.default_includes)
        context.k8s.excludes = Globs(context.k8s.default_excludes)

    def handle_after_dir(self, cwd: Path):
        context = self.context
        k8s = context.k8s

        for f in scan_dir(logger, cwd, lambda d: d.is_file(), k8s.excludes, k8s.includes):
            p = cwd / f.name
            display_p = context.app.display_path(p)
            logger.debug("Adding Kubernetes manifest from %s", display_p)

            manifests = load_file(logger, p, FileType.YAML, display_p)

            for manifest in manifests:
                if manifest:
                    self.add_resource(manifest, display_p)

    def handle_apply(self):
        context = self.context
        k8s = context.k8s

        self._validate_resources()

        cmd = context.app.args.command
        file = context.app.args.file
        file_format = context.app.args.output_format
        dry_run = context.app.args.dry_run
        dump = cmd == "dump"

        status_msg = f"{' (dump only)' if dump else ' (dry run)' if dry_run else ''}"
        if dump:
            logger.info("Will dump the changes into a file %s in %s format", file, file_format)

        patch_field_excludes = [re.compile(e) for e in context.globals.k8s.patch_field_excludes]
        dump_results = []
        total_created, total_patched, total_deleted = 0, 0, 0
        for resource in self.resources.values():
            if dump:
                resource_id = {"apiVersion": resource.api_version,
                               "kind": resource.kind,
                               "name": resource.name
                               }

                def patch_func(patch):
                    if resource.rdef.namespaced:
                        resource_id["namespace"] = resource.namespace
                    method_descriptor = {"method": "patch",
                                         "resource": resource_id,
                                         "body": patch
                                         }
                    dump_results.append(method_descriptor)

                def create_func():
                    method_descriptor = {"method": "create",
                                         "body": resource.manifest}
                    dump_results.append(method_descriptor)

                def delete_func(*, propagation_policy):
                    method_descriptor = {"method": "delete",
                                         "resource": resource_id,
                                         "propagation_policy": propagation_policy.policy
                                         }
                    dump_results.append(method_descriptor)
            else:
                patch_func = partial(resource.patch, patch_type=K8SResourcePatchType.JSON_PATCH, dry_run=dry_run)
                create_func = partial(resource.create, dry_run=dry_run)
                delete_func = partial(resource.delete, dry_run=dry_run)

            created, patched, deleted = self._apply_resource(dry_run,
                                                             patch_field_excludes,
                                                             resource,
                                                             patch_func,
                                                             create_func,
                                                             delete_func,
                                                             status_msg)

            total_created += created
            total_patched += patched
            total_deleted += deleted

        if ((dump or dry_run) and
                k8s.field_validation_warn_fatal and self.context.globals.k8s.field_validation_warnings):
            msg = ("There were %d field validation warnings and the warnings are fatal!" %
                   self.context.globals.k8s.field_validation_warnings)
            logger.fatal(msg)
            raise RuntimeError(msg)

        if dump:
            if file_format in ("json", "json-pretty"):
                json.dump(dump_results, file, sort_keys=True,
                          indent=4 if file_format == "json-pretty" else None)
            else:
                yaml.safe_dump(dump_results, file)
        else:
            self._summary = total_created, total_patched, total_deleted

    def handle_summary(self):
        total_created, total_patched, total_deleted = self._summary
        logger.info("Created %d, patched %d, deleted %d resources", total_created, total_patched, total_deleted)

    def api_load_resources(self, path: Path, file_type: str):
        return self.add_local_resources(path, FileType[file_type.upper()])

    def api_load_remote_resources(self, url: str, file_type: str, file_category=None):
        return self.add_remote_resources(url, FileType[file_type.upper()], sub_category=file_category)

    def api_load_crds(self, path: Path, file_type: str):
        return self.add_local_crds(path, FileType[file_type.upper()])

    def api_load_remote_crds(self, url: str, file_type: str, file_category=None):
        return self.add_remote_crds(url, FileType[file_type.upper()], sub_category=file_category)

    def api_add_transformer(self, transformer):
        if transformer not in self._transformers:
            self._transformers.append(transformer)

    def api_add_validator(self, validator):
        if validator not in self._validators:
            self._validators.append(validator)

    def api_remove_transformer(self, transformer):
        if transformer in self._transformers:
            self._transformers.remove(transformer)

    def api_remove_validator(self, validator):
        if validator not in self._validators:
            self._validators.remove(validator)

    def api_validation_error(self, msg, *args):
        frame = sys._getframe().f_back
        tb = None
        while True:
            if not frame:
                break
            tb = types.TracebackType(tb, frame, frame.f_lasti, frame.f_lineno)
            frame = frame.f_back
        return ValueError((msg % args) if args else msg).with_traceback(tb)

    def _transform_resource(self, resources: Sequence[K8SResource], resource: K8SResource) -> K8SResource:
        for transformer in reversed(self._transformers):
            logger.debug("Applying transformer %s to %s from %s",
                         getattr(transformer, "__name__", transformer),
                         resource, resource.source)
            resource = transformer(resources, resource) or resource

        return resource

    def _validate_resources(self):
        errors: list[Exception] = []
        for resource in self.resources.values():
            for validator in reversed(self._validators):
                logger.debug("Applying validator %s to %s from %s",
                             getattr(validator, "__name__", validator),
                             resource, resource.source)
                errors.extend(validator(self.resources, resource, self.api_validation_error))
        if errors:
            for error in errors:
                logger.error("Validation error: %s", error)
            raise errors[0]

    def _apply_resource(self,
                        dry_run,
                        patch_field_excludes: Iterable[re.compile],
                        resource: K8SResource,
                        patch_func: Callable[[Iterable[dict]], None],
                        create_func: Callable[[], None],
                        delete_func: Callable[[K8SPropagationPolicy], None],
                        status_msg):
        from kubernetes import client
        from kubernetes.client.rest import ApiException

        rdef = resource.rdef
        rdef.populate_api(client, self.context.k8s.client)

        def handle_400_strict_validation_error(e: ApiException):
            if e.status == 400:
                status = json.loads(e.body)

                if status["status"] == "Failure":
                    if FIELD_VALIDATION_STRICT_MARKER in status["message"]:
                        message = status["message"]
                        messages = message[message.find(FIELD_VALIDATION_STRICT_MARKER) +
                                           len(FIELD_VALIDATION_STRICT_MARKER):].split(",")
                        for m in messages:
                            self._api_warnings(resource, m.strip())

                        raise e from None
                    else:
                        logger.error("FAILED MODIFYING resource %s from %s: %s",
                                     resource, resource.source, status["message"])
                        raise e from None

        def create(exists_ok=False):
            logger.info("Creating resource %s%s%s", resource, status_msg,
                        " (ignoring existing)" if exists_ok else "")
            try:
                create_func()
            except ApiException as e:
                if exists_ok:
                    if e.status == 409:
                        status = json.loads(e.body)
                        if status["reason"] == "AlreadyExists":
                            return
                raise

        merge_instrs, normalized_manifest = extract_merge_instructions(resource.manifest, resource)
        if merge_instrs:
            logger.trace("Normalized manifest (no merge instructions) for resource %s: %s", resource,
                         normalized_manifest)
        else:
            normalized_manifest = resource.manifest

        logger.debug("Applying resource %s%s", resource, status_msg)
        try:
            remote_resource = resource.get()
            logger.trace("Current resource %s: %s", resource, remote_resource)
        except ApiException as e:
            if e.status == 404:
                try:
                    create()
                    return 1, 0, 0
                except ApiException as e:
                    if not handle_400_strict_validation_error(e):
                        raise

            else:
                raise
        else:
            logger.trace("Attempting to retrieve a normalized patch for resource %s: %s", resource, normalized_manifest)
            try:
                merged_resource = resource.patch(normalized_manifest,
                                                 patch_type=K8SResourcePatchType.SERVER_SIDE_PATCH,
                                                 dry_run=True,
                                                 force=True)
            except ApiException as e:
                if e.status == 422:
                    status = json.loads(e.body)
                    details = status["details"]
                    immutable_key = details["group"], details["kind"]

                    try:
                        propagation_policy = self.context.k8s.immutable_changes[immutable_key]
                    except KeyError:
                        raise e from None
                    else:
                        for cause in details["causes"]:
                            if (
                                    cause["reason"] == "FieldValueInvalid" and
                                    "field is immutable" in cause["message"]
                                    or
                                    cause["reason"] == "FieldValueForbidden" and
                                    "Forbidden: updates to" in cause["message"]
                            ):
                                logger.info("Deleting resource %s (cascade %s)%s", resource,
                                            propagation_policy.policy,
                                            status_msg)
                                delete_func(propagation_policy=propagation_policy)
                                create(exists_ok=dry_run)
                                return 1, 0, 1
                        raise
                else:
                    if not handle_400_strict_validation_error(e):
                        raise
            else:
                logger.trace("Merged resource %s: %s", resource, merged_resource)
                if merge_instrs:
                    apply_merge_instructions(merge_instrs, normalized_manifest, merged_resource, logger, resource)

                patch = jsonpatch.make_patch(remote_resource, merged_resource)
                logger.trace("Resource %s initial patches are: %s", resource, patch)
                patch = self._filter_resource_patch(patch, patch_field_excludes)
                logger.trace("Resource %s final patches are: %s", resource, patch)
                if patch:
                    logger.info("Patching resource %s%s", resource, status_msg)
                    patch_func(patch)
                    return 0, 1, 0
                else:
                    logger.info("Nothing to patch for resource %s", resource)
                    return 0, 0, 0

    def _filter_resource_patch(self, patch: Iterable[Mapping], excludes: Iterable[re.compile]):
        result = []
        for op in patch:
            path = op["path"]
            excluded = False
            for exclude in excludes:
                if exclude.match(path):
                    logger.trace("Excluding %r from patch %s", op, patch)
                    excluded = True
                    break
            if excluded:
                continue
            result.append(op)
        return result

    def _setup_k8s_client(self):
        from kubernetes import client
        from kubernetes.config import load_incluster_config, load_kube_config, ConfigException

        try:
            logger.debug("Trying K8S in-cluster configuration")
            load_incluster_config()
            logger.info("Running K8S with in-cluster configuration")
        except ConfigException as e:
            logger.trace("K8S in-cluster configuration failed", exc_info=e)
            logger.debug("Initializing K8S with kubeconfig configuration")
            load_kube_config(config_file=self.context.kubeconfig.kubeconfig)

        k8s_client = client.ApiClient()

        # Patch the header content type selector to allow json patch
        k8s_client._select_header_content_type = k8s_client.select_header_content_type
        k8s_client.select_header_content_type = self._select_header_content_type_patch

        return k8s_client

    def _select_header_content_type_patch(self, content_types):
        """Returns `Content-Type` based on an array of content_types provided.
        :param content_types: List of content-types.
        :return: Content-Type (e.g. application/json).
        """

        content_type = self.context.k8s.client._select_header_content_type(content_types)
        if content_type == "application/merge-patch+json":
            return "application/json-patch+json"
        return content_type

    def __repr__(self):
        return "Kubernetes Plugin"

1	# -- coding: utf-8 --
2	#
3	# Copyright 2020 Express Systems USA, Inc
4	# Copyright 2021 Karellen, Inc.
5	#
6	# Licensed under the Apache License, Version 2.0 (the "License");
7	# you may not use this file except in compliance with the License.
8	# You may obtain a copy of the License at
9	#
10	# http://www.apache.org/licenses/LICENSE-2.0
11	#
12	# Unless required by applicable law or agreed to in writing, software
13	# distributed under the License is distributed on an "AS IS" BASIS,
14	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15	# See the License for the specific language governing permissions and
16	# limitations under the License.
17	#
18
19
20	import json	6✔
21	import logging	6✔
22	import re	6✔
23	import sys	6✔
24	import types	6✔
25	from collections.abc import Mapping	6✔
26	from functools import partial	6✔
27	from importlib.metadata import version as pkg_version	6✔
28	from pathlib import Path	6✔
29	from typing import Iterable, Callable, Sequence	6✔
30
31	import jsonpatch	6✔
32	import yaml	6✔
33
34	from kubernator.api import (KubernatorPlugin,	6✔
35	Globs,
36	scan_dir,
37	load_file,
38	FileType,
39	load_remote_file,
40	StripNL,
41	install_python_k8s_client)
42	from kubernator.merge import extract_merge_instructions, apply_merge_instructions	6✔
43	from kubernator.plugins.k8s_api import (K8SResourcePluginMixin,	6✔
44	K8SResource,
45	K8SResourcePatchType,
46	K8SPropagationPolicy)
47
48	logger = logging.getLogger("kubernator.k8s")	6✔
49	proc_logger = logger.getChild("proc")	6✔
50	stdout_logger = StripNL(proc_logger.info)	6✔
51	stderr_logger = StripNL(proc_logger.warning)	6✔
52
53	FIELD_VALIDATION_STRICT_MARKER = "strict decoding error: "	6✔
54	VALID_FIELD_VALIDATION = ("Ignore", "Warn", "Strict")	6✔
55
56
57	def final_resource_validator(resources: Sequence[K8SResource],	6✔
58	resource: K8SResource,
59	error: Callable[..., Exception]) -> Iterable[Exception]:
60	final_key = resource.get_manifest_key(resource.manifest)	6✔
61	if final_key != resource.key:	6!
62	yield error("Illegal change of identifiers of the resource "	×
63	"%s from %s have been changed to %s",
64	resource.key, resource.source, final_key)
65
66	if resource.rdef.namespaced and not resource.namespace:	6!
67	yield error("Namespaced resource %s from %s is missing the required namespace",	×
68	resource, resource.source)
69
70
71	def normalize_pkg_version(v: str):	6✔
72	v_split = v.split(".")	6✔
73	rev = v_split[-1]	6✔
74	if not rev.isdigit():	6✔
75	new_rev = ""	6✔
76	for c in rev:	6!
77	if not c.isdigit():	6✔
78	break	6✔
79	new_rev += c	6✔
80	v_split[-1] = new_rev	6✔
81	return tuple(map(int, v_split))	6✔
82
83
84	class KubernetesPlugin(KubernatorPlugin, K8SResourcePluginMixin):	6✔
85	logger = logger	6✔
86
87	_name = "k8s"	6✔
88
89	def __init__(self):	6✔
90	super().__init__()	6✔
91	self.context = None	6✔
92
93	self.embedded_pkg_version = self._get_kubernetes_client_version()	6✔
94
95	self._transformers = []	6✔
96	self._validators = []	6✔
97	self._summary = 0, 0, 0	6✔
98
99	def set_context(self, context):	6✔
100	self.context = context	6✔
101
102	def register(self,	6✔
103	field_validation="Warn",
104	field_validation_warn_fatal=True,
105	disable_client_patches=False):
106	self.context.app.register_plugin("kubeconfig")	6✔
107
108	if field_validation not in VALID_FIELD_VALIDATION:	6!
109	raise ValueError("'field_validation' must be one of %s" % (", ".join(VALID_FIELD_VALIDATION)))	×
110
111	context = self.context	6✔
112	context.globals.k8s = dict(patch_field_excludes=("^/metadata/managedFields",	6✔
113	"^/metadata/generation",
114	"^/metadata/creationTimestamp",
115	"^/metadata/resourceVersion",
116	),
117	immutable_changes={("apps", "DaemonSet"): K8SPropagationPolicy.BACKGROUND,
118	("apps", "StatefulSet"): K8SPropagationPolicy.ORPHAN,
119	("apps", "Deployment"): K8SPropagationPolicy.ORPHAN,
120	("storage.k8s.io", "StorageClass"): K8SPropagationPolicy.ORPHAN,
121	},
122	default_includes=Globs([".yaml", ".yml"], True),
123	default_excludes=Globs([".*"], True),
124	add_resources=self.add_resources,
125	load_resources=self.api_load_resources,
126	load_remote_resources=self.api_load_remote_resources,
127	load_crds=self.api_load_crds,
128	load_remote_crds=self.api_load_remote_crds,
129	add_transformer=self.api_add_transformer,
130	remove_transformer=self.api_remove_transformer,
131	add_validator=self.api_remove_validator,
132	get_api_versions=self.get_api_versions,
133	create_resource=self.create_resource,
134	disable_client_patches=disable_client_patches,
135	field_validation=field_validation,
136	field_validation_warn_fatal=field_validation_warn_fatal,
137	field_validation_warnings=0,
138	_k8s=self,
139	)
140	context.k8s = dict(default_includes=Globs(context.globals.k8s.default_includes),	6✔
141	default_excludes=Globs(context.globals.k8s.default_excludes)
142	)
143	self.api_add_validator(final_resource_validator)	6✔
144
145	def handle_init(self):	6✔
146	pass	6✔
147
148	def handle_start(self):	6✔
149	self.context.kubeconfig.register_change_notifier(self._kubeconfig_changed)	6✔
150	self.setup_client()	6✔
151
152	def _kubeconfig_changed(self):	6✔
153	self.setup_client()	×
154
155	def _get_kubernetes_client_version(self):	6✔
156	return pkg_version("kubernetes").split(".")	6✔
157
158	def setup_client(self):	6✔
159	k8s = self.context.k8s	6✔
160	if "server_version" not in k8s:	6!
161	self._setup_client()	6✔
162
163	server_minor = k8s.server_version[1]	6✔
164
165	logger.info("Using Kubernetes client version =~%s.0 for server version %s",	6✔
166	server_minor, ".".join(k8s.server_version))
167	pkg_dir = install_python_k8s_client(self.context.app.run_passthrough_capturing, server_minor, logger,	6✔
168	stdout_logger, stderr_logger, k8s.disable_client_patches)
169
170	modules_to_delete = []	6✔
171	for k, v in sys.modules.items():	6✔
172	if k == "kubernetes" or k.startswith("kubernetes."):	6✔
173	modules_to_delete.append(k)	6✔
174	for k in modules_to_delete:	6✔
175	del sys.modules[k]	6✔
176
177	logger.info("Adding sys.path reference to %s", pkg_dir)	6✔
178	sys.path.insert(0, str(pkg_dir))	6✔
179	self.embedded_pkg_version = self._get_kubernetes_client_version()	6✔
180	logger.info("Switching to Kubernetes client version %s", ".".join(self.embedded_pkg_version))	6✔
181	self._setup_client()	6✔
182
183	logger.debug("Reading Kubernetes OpenAPI spec for %s", k8s.server_git_version)	6✔
184
185	k8s_def = load_remote_file(logger, f"https://raw.githubusercontent.com/kubernetes/kubernetes/"	6✔
186	f"{k8s.server_git_version}/api/openapi-spec/swagger.json",
187	FileType.JSON)
188	self.resource_definitions_schema = k8s_def	6✔
189
190	self._populate_resource_definitions()	6✔
191
192	def _setup_client(self):	6✔
193	from kubernetes import client	6✔
194
195	context = self.context	6✔
196	k8s = context.k8s	6✔
197
198	k8s.client = self._setup_k8s_client()	6✔
199	version = client.VersionApi(k8s.client).get_code()	6✔
200	if "-eks-" in version.git_version:	6!
201	git_version = version.git_version.split("-")[0]	×
202	else:
203	git_version = version.git_version	6✔
204
205	k8s.server_version = git_version[1:].split(".")	6✔
206	k8s.server_git_version = git_version	6✔
207
208	logger.info("Found Kubernetes %s on %s", k8s.server_git_version, k8s.client.configuration.host)	6✔
209
210	K8SResource._k8s_client_version = normalize_pkg_version(pkg_version("kubernetes"))	6✔
211	K8SResource._k8s_field_validation = k8s.field_validation	6✔
212	K8SResource._k8s_field_validation_patched = not k8s.disable_client_patches	6✔
213	K8SResource._logger = self.logger	6✔
214	K8SResource._api_warnings = self._api_warnings	6✔
215
216	def _api_warnings(self, resource, warn):	6✔
217	k8s = self.context.k8s	6✔
218	self.context.globals.k8s.field_validation_warnings += 1	6✔
219
220	log = self.logger.warning	6✔
221	if k8s.field_validation_warn_fatal:	6✔
222	log = self.logger.error	6✔
223
224	log("FAILED FIELD VALIDATION on resource %s from %s: %s", resource, resource.source, warn)	6✔
225
226	def handle_before_dir(self, cwd: Path):	6✔
227	context = self.context	6✔
228	context.k8s.default_includes = Globs(context.k8s.default_includes)	6✔
229	context.k8s.default_excludes = Globs(context.k8s.default_excludes)	6✔
230	context.k8s.includes = Globs(context.k8s.default_includes)	6✔
231	context.k8s.excludes = Globs(context.k8s.default_excludes)	6✔
232
233	def handle_after_dir(self, cwd: Path):	6✔
234	context = self.context	6✔
235	k8s = context.k8s	6✔
236
237	for f in scan_dir(logger, cwd, lambda d: d.is_file(), k8s.excludes, k8s.includes):	6✔
238	p = cwd / f.name	6✔
239	display_p = context.app.display_path(p)	6✔
240	logger.debug("Adding Kubernetes manifest from %s", display_p)	6✔
241
242	manifests = load_file(logger, p, FileType.YAML, display_p)	6✔
243
244	for manifest in manifests:	6✔
245	if manifest:	6!
246	self.add_resource(manifest, display_p)	6✔
247
248	def handle_apply(self):	6✔
249	context = self.context	6✔
250	k8s = context.k8s	6✔
251
252	self._validate_resources()	6✔
253
254	cmd = context.app.args.command	6✔
255	file = context.app.args.file	6✔
256	file_format = context.app.args.output_format	6✔
257	dry_run = context.app.args.dry_run	6✔
258	dump = cmd == "dump"	6✔
259
260	status_msg = f"{' (dump only)' if dump else ' (dry run)' if dry_run else ''}"	6✔
261	if dump:	6✔
262	logger.info("Will dump the changes into a file %s in %s format", file, file_format)	6✔
263
264	patch_field_excludes = [re.compile(e) for e in context.globals.k8s.patch_field_excludes]	6✔
265	dump_results = []	6✔
266	total_created, total_patched, total_deleted = 0, 0, 0	6✔
267	for resource in self.resources.values():	6✔
268	if dump:	6✔
269	resource_id = {"apiVersion": resource.api_version,	6✔
270	"kind": resource.kind,
271	"name": resource.name
272	}
273
274	def patch_func(patch):	6✔
275	if resource.rdef.namespaced:	6!
276	resource_id["namespace"] = resource.namespace	×
277	method_descriptor = {"method": "patch",	6✔
278	"resource": resource_id,
279	"body": patch
280	}
281	dump_results.append(method_descriptor)	6✔
282
283	def create_func():	6✔
284	method_descriptor = {"method": "create",	6✔
285	"body": resource.manifest}
286	dump_results.append(method_descriptor)	6✔
287
288	def delete_func(*, propagation_policy):	6✔
289	method_descriptor = {"method": "delete",	×
290	"resource": resource_id,
291	"propagation_policy": propagation_policy.policy
292	}
293	dump_results.append(method_descriptor)	×
294	else:
295	patch_func = partial(resource.patch, patch_type=K8SResourcePatchType.JSON_PATCH, dry_run=dry_run)	6✔
296	create_func = partial(resource.create, dry_run=dry_run)	6✔
297	delete_func = partial(resource.delete, dry_run=dry_run)	6✔
298
299	created, patched, deleted = self._apply_resource(dry_run,	6✔
300	patch_field_excludes,
301	resource,
302	patch_func,
303	create_func,
304	delete_func,
305	status_msg)
306
307	total_created += created	6✔
308	total_patched += patched	6✔
309	total_deleted += deleted	6✔
310
311	if ((dump or dry_run) and	6✔
312	k8s.field_validation_warn_fatal and self.context.globals.k8s.field_validation_warnings):
313	msg = ("There were %d field validation warnings and the warnings are fatal!" %	6✔
314	self.context.globals.k8s.field_validation_warnings)
315	logger.fatal(msg)	6✔
316	raise RuntimeError(msg)	6✔
317
318	if dump:	6✔
319	if file_format in ("json", "json-pretty"):	6!
320	json.dump(dump_results, file, sort_keys=True,	×
321	indent=4 if file_format == "json-pretty" else None)
322	else:
323	yaml.safe_dump(dump_results, file)	6✔
324	else:
325	self._summary = total_created, total_patched, total_deleted	6✔
326
327	def handle_summary(self):	6✔
328	total_created, total_patched, total_deleted = self._summary	6✔
329	logger.info("Created %d, patched %d, deleted %d resources", total_created, total_patched, total_deleted)	6✔
330
331	def api_load_resources(self, path: Path, file_type: str):	6✔
332	return self.add_local_resources(path, FileType[file_type.upper()])	×
333
334	def api_load_remote_resources(self, url: str, file_type: str, file_category=None):	6✔
335	return self.add_remote_resources(url, FileType[file_type.upper()], sub_category=file_category)	×
336
337	def api_load_crds(self, path: Path, file_type: str):	6✔
338	return self.add_local_crds(path, FileType[file_type.upper()])	6✔
339
340	def api_load_remote_crds(self, url: str, file_type: str, file_category=None):	6✔
341	return self.add_remote_crds(url, FileType[file_type.upper()], sub_category=file_category)	6✔
342
343	def api_add_transformer(self, transformer):	6✔
344	if transformer not in self._transformers:	6!
345	self._transformers.append(transformer)	6✔
346
347	def api_add_validator(self, validator):	6✔
348	if validator not in self._validators:	6!
349	self._validators.append(validator)	6✔
350
351	def api_remove_transformer(self, transformer):	6✔
352	if transformer in self._transformers:	6!
353	self._transformers.remove(transformer)	6✔
354
355	def api_remove_validator(self, validator):	6✔
356	if validator not in self._validators:	×
357	self._validators.remove(validator)	×
358
359	def api_validation_error(self, msg, *args):	6✔
360	frame = sys._getframe().f_back	×
361	tb = None	×
362	while True:
363	if not frame:	×
364	break	×
365	tb = types.TracebackType(tb, frame, frame.f_lasti, frame.f_lineno)	×
366	frame = frame.f_back	×
367	return ValueError((msg % args) if args else msg).with_traceback(tb)	×
368
369	def _transform_resource(self, resources: Sequence[K8SResource], resource: K8SResource) -> K8SResource:	6✔
370	for transformer in reversed(self._transformers):	6✔
371	logger.debug("Applying transformer %s to %s from %s",	6✔
372	getattr(transformer, "__name__", transformer),
373	resource, resource.source)
374	resource = transformer(resources, resource) or resource	6✔
375
376	return resource	6✔
377
378	def _validate_resources(self):	6✔
379	errors: list[Exception] = []	6✔
380	for resource in self.resources.values():	6✔
381	for validator in reversed(self._validators):	6✔
382	logger.debug("Applying validator %s to %s from %s",	6✔
383	getattr(validator, "__name__", validator),
384	resource, resource.source)
385	errors.extend(validator(self.resources, resource, self.api_validation_error))	6✔
386	if errors:	6!
387	for error in errors:	×
388	logger.error("Validation error: %s", error)	×
389	raise errors[0]	×
390
391	def _apply_resource(self,	6✔
392	dry_run,
393	patch_field_excludes: Iterable[re.compile],
394	resource: K8SResource,
395	patch_func: Callable[[Iterable[dict]], None],
396	create_func: Callable[[], None],
397	delete_func: Callable[[K8SPropagationPolicy], None],
398	status_msg):
399	from kubernetes import client	6✔
400	from kubernetes.client.rest import ApiException	6✔
401
402	rdef = resource.rdef	6✔
403	rdef.populate_api(client, self.context.k8s.client)	6✔
404
405	def handle_400_strict_validation_error(e: ApiException):	6✔
406	if e.status == 400:	6!
407	status = json.loads(e.body)	6✔
408
409	if status["status"] == "Failure":	6!
410	if FIELD_VALIDATION_STRICT_MARKER in status["message"]:	6!
411	message = status["message"]	6✔
412	messages = message[message.find(FIELD_VALIDATION_STRICT_MARKER) +	6✔
413	len(FIELD_VALIDATION_STRICT_MARKER):].split(",")
414	for m in messages:	6✔
415	self._api_warnings(resource, m.strip())	6✔
416
417	raise e from None	6✔
418	else:
419	logger.error("FAILED MODIFYING resource %s from %s: %s",	×
420	resource, resource.source, status["message"])
421	raise e from None	×
422
423	def create(exists_ok=False):	6✔
424	logger.info("Creating resource %s%s%s", resource, status_msg,	6✔
425	" (ignoring existing)" if exists_ok else "")
426	try:	6✔
427	create_func()	6✔
428	except ApiException as e:	6✔
429	if exists_ok:	6!
430	if e.status == 409:	×
431	status = json.loads(e.body)	×
432	if status["reason"] == "AlreadyExists":	×
433	return	×
434	raise	6✔
435
436	merge_instrs, normalized_manifest = extract_merge_instructions(resource.manifest, resource)	6✔
437	if merge_instrs:	6✔
438	logger.trace("Normalized manifest (no merge instructions) for resource %s: %s", resource,	6✔
439	normalized_manifest)
440	else:
441	normalized_manifest = resource.manifest	6✔
442
443	logger.debug("Applying resource %s%s", resource, status_msg)	6✔
444	try:	6✔
445	remote_resource = resource.get()	6✔
446	logger.trace("Current resource %s: %s", resource, remote_resource)	6✔
447	except ApiException as e:	6✔
448	if e.status == 404:	6!
449	try:	6✔
450	create()	6✔
451	return 1, 0, 0	6✔
452	except ApiException as e:	6✔
453	if not handle_400_strict_validation_error(e):	6✔
454	raise	1✔
455
456	else:
457	raise	1✔
458	else:
459	logger.trace("Attempting to retrieve a normalized patch for resource %s: %s", resource, normalized_manifest)	6✔
460	try:	6✔
461	merged_resource = resource.patch(normalized_manifest,	6✔
462	patch_type=K8SResourcePatchType.SERVER_SIDE_PATCH,
463	dry_run=True,
464	force=True)
465	except ApiException as e:	×
466	if e.status == 422:	×
467	status = json.loads(e.body)	×
468	details = status["details"]	×
469	immutable_key = details["group"], details["kind"]	×
470
471	try:	×
472	propagation_policy = self.context.k8s.immutable_changes[immutable_key]	×
473	except KeyError:	×
474	raise e from None	×
475	else:
476	for cause in details["causes"]:	×
477	if (	×
478	cause["reason"] == "FieldValueInvalid" and
479	"field is immutable" in cause["message"]
480	or
481	cause["reason"] == "FieldValueForbidden" and
482	"Forbidden: updates to" in cause["message"]
483	):
484	logger.info("Deleting resource %s (cascade %s)%s", resource,	×
485	propagation_policy.policy,
486	status_msg)
487	delete_func(propagation_policy=propagation_policy)	×
488	create(exists_ok=dry_run)	×
489	return 1, 0, 1	×
490	raise	×
491	else:
492	if not handle_400_strict_validation_error(e):	×
493	raise	×
494	else:
495	logger.trace("Merged resource %s: %s", resource, merged_resource)	6✔
496	if merge_instrs:	6✔
497	apply_merge_instructions(merge_instrs, normalized_manifest, merged_resource, logger, resource)	6✔
498
499	patch = jsonpatch.make_patch(remote_resource, merged_resource)	6✔
500	logger.trace("Resource %s initial patches are: %s", resource, patch)	6✔
501	patch = self._filter_resource_patch(patch, patch_field_excludes)	6✔
502	logger.trace("Resource %s final patches are: %s", resource, patch)	6✔
503	if patch:	6✔
504	logger.info("Patching resource %s%s", resource, status_msg)	6✔
505	patch_func(patch)	6✔
506	return 0, 1, 0	6✔
507	else:
508	logger.info("Nothing to patch for resource %s", resource)	6✔
509	return 0, 0, 0	6✔
510
511	def _filter_resource_patch(self, patch: Iterable[Mapping], excludes: Iterable[re.compile]):	6✔
512	result = []	6✔
513	for op in patch:	6✔
514	path = op["path"]	6✔
515	excluded = False	6✔
516	for exclude in excludes:	6✔
517	if exclude.match(path):	6✔
518	logger.trace("Excluding %r from patch %s", op, patch)	6✔
519	excluded = True	6✔
520	break	6✔
521	if excluded:	6✔
522	continue	6✔
523	result.append(op)	6✔
524	return result	6✔
525
526	def _setup_k8s_client(self):	6✔
527	from kubernetes import client	6✔
528	from kubernetes.config import load_incluster_config, load_kube_config, ConfigException	6✔
529
530	try:	6✔
531	logger.debug("Trying K8S in-cluster configuration")	6✔
532	load_incluster_config()	6✔
533	logger.info("Running K8S with in-cluster configuration")	×
534	except ConfigException as e:	6✔
535	logger.trace("K8S in-cluster configuration failed", exc_info=e)	6✔
536	logger.debug("Initializing K8S with kubeconfig configuration")	6✔
537	load_kube_config(config_file=self.context.kubeconfig.kubeconfig)	6✔
538
539	k8s_client = client.ApiClient()	6✔
540
541	# Patch the header content type selector to allow json patch
542	k8s_client._select_header_content_type = k8s_client.select_header_content_type	6✔
543	k8s_client.select_header_content_type = self._select_header_content_type_patch	6✔
544
545	return k8s_client	6✔
546
547	def _select_header_content_type_patch(self, content_types):	6✔
548	"""Returns `Content-Type` based on an array of content_types provided.
549	:param content_types: List of content-types.
550	:return: Content-Type (e.g. application/json).
551	"""
552
553	content_type = self.context.k8s.client._select_header_content_type(content_types)	×
554	if content_type == "application/merge-patch+json":	×
555	return "application/json-patch+json"	×
556	return content_type	×
557
558	def __repr__(self):	6✔
559	return "Kubernetes Plugin"	6✔

karellen / kubernator / 18180360618

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