• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

karellen / kubernator / 20020563524

08 Dec 2025 07:46AM UTC coverage: 77.296% (-0.2%) from 77.467%
20020563524

push

github

web-flow
Merge pull request #94 from karellen/add_import_cluster_crds

Add import_cluster_crds function

666 of 1020 branches covered (65.29%)

Branch coverage included in aggregate %.

19 of 30 new or added lines in 2 files covered. (63.33%)

2 existing lines in 1 file now uncovered.

2541 of 3129 relevant lines covered (81.21%)

4.87 hits per line

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

74.9
/src/main/python/kubernator/plugins/k8s.py
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
from kubernetes.client import ApiException
6✔
34

35
from kubernator.api import (KubernatorPlugin,
6✔
36
                            Globs,
37
                            scan_dir,
38
                            load_file,
39
                            FileType,
40
                            load_remote_file,
41
                            StripNL,
42
                            install_python_k8s_client,
43
                            TemplateEngine,
44
                            sleep)
45
from kubernator.merge import extract_merge_instructions, apply_merge_instructions
6✔
46
from kubernator.plugins.k8s_api import (K8SResourcePluginMixin,
6✔
47
                                        K8SResource,
48
                                        K8SResourcePatchType,
49
                                        K8SPropagationPolicy)
50

51
logger = logging.getLogger("kubernator.k8s")
6✔
52
proc_logger = logger.getChild("proc")
6✔
53
stdout_logger = StripNL(proc_logger.info)
6✔
54
stderr_logger = StripNL(proc_logger.warning)
6✔
55

56
FIELD_VALIDATION_STRICT_MARKER = "strict decoding error: "
6✔
57
VALID_FIELD_VALIDATION = ("Ignore", "Warn", "Strict")
6✔
58

59

60
def final_resource_validator(resources: Sequence[K8SResource],
6✔
61
                             resource: K8SResource,
62
                             error: Callable[..., Exception]) -> Iterable[Exception]:
63
    final_key = resource.get_manifest_key(resource.manifest)
6✔
64
    if final_key != resource.key:
6!
65
        yield error("Illegal change of identifiers of the resource "
×
66
                    "%s from %s have been changed to %s",
67
                    resource.key, resource.source, final_key)
68

69
    if resource.rdef.namespaced and not resource.namespace:
6!
70
        yield error("Namespaced resource %s from %s is missing the required namespace",
×
71
                    resource, resource.source)
72

73

74
def normalize_pkg_version(v: str):
6✔
75
    v_split = v.split(".")
6✔
76
    rev = v_split[-1]
6✔
77
    if not rev.isdigit():
6✔
78
        new_rev = ""
6✔
79
        for c in rev:
6!
80
            if not c.isdigit():
6✔
81
                break
6✔
82
            new_rev += c
6✔
83
        v_split[-1] = new_rev
6✔
84
    return tuple(map(int, v_split))
6✔
85

86

87
def api_exc_normalize_body(e: "ApiException"):
6✔
88
    if e.headers and "content-type" in e.headers:
6!
89
        content_type = e.headers["content-type"]
6✔
90
        if content_type == "application/json" or content_type.endswith("+json"):
6!
91
            e.body = json.loads(e.body)
6✔
92
        elif (content_type in ("application/yaml", "application/x-yaml", "text/yaml",
×
93
                               "text/x-yaml") or content_type.endswith("+yaml")):
94
            e.body = yaml.safe_load(e.body)
×
95

96

97
def api_exc_format_body(e: ApiException):
6✔
98
    if not isinstance(e.body, (str, bytes)):
6!
99
        e.body = json.dumps(e.body, indent=4)
6✔
100

101

102
class KubernetesPlugin(KubernatorPlugin, K8SResourcePluginMixin):
6✔
103
    logger = logger
6✔
104

105
    _name = "k8s"
6✔
106

107
    def __init__(self):
6✔
108
        super().__init__()
6✔
109
        self.context = None
6✔
110

111
        self.embedded_pkg_version = self._get_kubernetes_client_version()
6✔
112

113
        self._transformers = []
6✔
114
        self._validators = []
6✔
115
        self._manifest_patchers = []
6✔
116
        self._summary = 0, 0, 0
6✔
117
        self._template_engine = TemplateEngine(logger)
6✔
118

119
    def set_context(self, context):
6✔
120
        self.context = context
6✔
121

122
    def register(self,
6✔
123
                 field_validation="Warn",
124
                 field_validation_warn_fatal=True,
125
                 disable_client_patches=False):
126
        self.context.app.register_plugin("kubeconfig")
6✔
127

128
        if field_validation not in VALID_FIELD_VALIDATION:
6!
129
            raise ValueError("'field_validation' must be one of %s" % (", ".join(VALID_FIELD_VALIDATION)))
×
130

131
        context = self.context
6✔
132
        context.globals.k8s = dict(patch_field_excludes=("^/metadata/managedFields",
6✔
133
                                                         "^/metadata/generation",
134
                                                         "^/metadata/creationTimestamp",
135
                                                         "^/metadata/resourceVersion",
136
                                                         ),
137
                                   immutable_changes={("apps", "DaemonSet"): K8SPropagationPolicy.BACKGROUND,
138
                                                      ("apps", "StatefulSet"): K8SPropagationPolicy.ORPHAN,
139
                                                      ("apps", "Deployment"): K8SPropagationPolicy.ORPHAN,
140
                                                      ("storage.k8s.io", "StorageClass"): K8SPropagationPolicy.ORPHAN,
141
                                                      (None, "Pod"): K8SPropagationPolicy.BACKGROUND,
142
                                                      ("batch", "Job"): K8SPropagationPolicy.ORPHAN,
143
                                                      },
144
                                   default_includes=Globs(["*.yaml", "*.yml"], True),
145
                                   default_excludes=Globs([".*"], True),
146
                                   add_resources=self.add_resources,
147
                                   load_resources=self.api_load_resources,
148
                                   load_remote_resources=self.api_load_remote_resources,
149
                                   load_crds=self.api_load_crds,
150
                                   import_cluster_crds=self.api_import_cluster_crds,
151
                                   load_remote_crds=self.api_load_remote_crds,
152
                                   add_transformer=self.api_add_transformer,
153
                                   remove_transformer=self.api_remove_transformer,
154
                                   add_validator=self.api_remove_validator,
155
                                   add_manifest_patcher=self.api_add_manifest_patcher,
156
                                   get_api_versions=self.get_api_versions,
157
                                   create_resource=self.create_resource,
158
                                   disable_client_patches=disable_client_patches,
159
                                   field_validation=field_validation,
160
                                   field_validation_warn_fatal=field_validation_warn_fatal,
161
                                   field_validation_warnings=0,
162
                                   conflict_retry_delay=0.3,
163
                                   _k8s=self,
164
                                   )
165
        context.k8s = dict(default_includes=Globs(context.globals.k8s.default_includes),
6✔
166
                           default_excludes=Globs(context.globals.k8s.default_excludes)
167
                           )
168
        self.api_add_validator(final_resource_validator)
6✔
169

170
    def handle_init(self):
6✔
171
        pass
6✔
172

173
    def handle_start(self):
6✔
174
        self.context.kubeconfig.register_change_notifier(self._kubeconfig_changed)
6✔
175
        self.setup_client()
6✔
176

177
    def _kubeconfig_changed(self):
6✔
178
        self.setup_client()
×
179

180
    def _get_kubernetes_client_version(self):
6✔
181
        return pkg_version("kubernetes").split(".")
6✔
182

183
    def setup_client(self):
6✔
184
        k8s = self.context.k8s
6✔
185
        if "server_version" not in k8s:
6!
186
            self._setup_client()
6✔
187

188
        server_minor = k8s.server_version[1]
6✔
189

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

195
        modules_to_delete = []
6✔
196
        for k, v in sys.modules.items():
6✔
197
            if k == "kubernetes" or k.startswith("kubernetes."):
6✔
198
                modules_to_delete.append(k)
6✔
199
        for k in modules_to_delete:
6✔
200
            del sys.modules[k]
6✔
201

202
        logger.info("Adding sys.path reference to %s", pkg_dir)
6✔
203
        sys.path.insert(0, str(pkg_dir))
6✔
204
        self.embedded_pkg_version = self._get_kubernetes_client_version()
6✔
205
        logger.info("Switching to Kubernetes client version %s", ".".join(self.embedded_pkg_version))
6✔
206
        self._setup_client()
6✔
207

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

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

215
        self._populate_resource_definitions()
6✔
216

217
    def _setup_client(self):
6✔
218
        from kubernetes import client
6✔
219

220
        context = self.context
6✔
221
        k8s = context.k8s
6✔
222

223
        k8s.client = self._setup_k8s_client()
6✔
224
        version = client.VersionApi(k8s.client).get_code()
6✔
225
        if "-eks-" or "-gke" in version.git_version:
6!
226
            git_version = version.git_version.split("-")[0]
6✔
227
        else:
228
            git_version = version.git_version
229

230
        k8s.server_version = git_version[1:].split(".")
6✔
231
        k8s.server_git_version = git_version
6✔
232

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

235
        K8SResource._k8s_client_version = normalize_pkg_version(pkg_version("kubernetes"))
6✔
236
        K8SResource._k8s_field_validation = k8s.field_validation
6✔
237
        K8SResource._k8s_field_validation_patched = not k8s.disable_client_patches
6✔
238
        K8SResource._logger = self.logger
6✔
239
        K8SResource._api_warnings = self._api_warnings
6✔
240

241
    def _api_warnings(self, resource, warn):
6✔
242
        k8s = self.context.k8s
6✔
243
        self.context.globals.k8s.field_validation_warnings += 1
6✔
244

245
        log = self.logger.warning
6✔
246
        if k8s.field_validation_warn_fatal:
6✔
247
            log = self.logger.error
6✔
248

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

251
    def handle_before_dir(self, cwd: Path):
6✔
252
        context = self.context
6✔
253
        context.k8s.default_includes = Globs(context.k8s.default_includes)
6✔
254
        context.k8s.default_excludes = Globs(context.k8s.default_excludes)
6✔
255
        context.k8s.includes = Globs(context.k8s.default_includes)
6✔
256
        context.k8s.excludes = Globs(context.k8s.default_excludes)
6✔
257

258
    def handle_after_dir(self, cwd: Path):
6✔
259
        context = self.context
6✔
260
        k8s = context.k8s
6✔
261

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

267
            manifests = load_file(logger, p, FileType.YAML, display_p,
6✔
268
                                  self._template_engine,
269
                                  {"ktor": context}
270
                                  )
271

272
            for manifest in manifests:
6✔
273
                if manifest:
6!
274
                    self.add_resource(manifest, display_p)
6✔
275

276
    def handle_apply(self):
6✔
277
        context = self.context
6✔
278
        k8s = context.k8s
6✔
279

280
        self._validate_resources()
6✔
281

282
        cmd = context.app.args.command
6✔
283
        file = context.app.args.file
6✔
284
        file_format = context.app.args.output_format
6✔
285
        dry_run = context.app.args.dry_run
6✔
286
        dump = cmd == "dump"
6✔
287

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

292
        patch_field_excludes = [re.compile(e) for e in context.globals.k8s.patch_field_excludes]
6✔
293
        dump_results = []
6✔
294
        total_created, total_patched, total_deleted = 0, 0, 0
6✔
295
        for resource in self.resources.values():
6✔
296
            if dump:
6✔
297
                resource_id = {"apiVersion": resource.api_version,
6✔
298
                               "kind": resource.kind,
299
                               "name": resource.name
300
                               }
301

302
                def patch_func(patch):
6✔
303
                    if resource.rdef.namespaced:
6!
304
                        resource_id["namespace"] = resource.namespace
×
305
                    method_descriptor = {"method": "patch",
6✔
306
                                         "resource": resource_id,
307
                                         "body": patch
308
                                         }
309
                    dump_results.append(method_descriptor)
6✔
310

311
                def create_func():
6✔
312
                    method_descriptor = {"method": "create",
6✔
313
                                         "body": resource.manifest}
314
                    dump_results.append(method_descriptor)
6✔
315

316
                def delete_func(*, propagation_policy):
6✔
317
                    method_descriptor = {"method": "delete",
×
318
                                         "resource": resource_id,
319
                                         "propagation_policy": propagation_policy.policy
320
                                         }
321
                    dump_results.append(method_descriptor)
×
322
            else:
323
                patch_func = partial(resource.patch, patch_type=K8SResourcePatchType.JSON_PATCH, dry_run=dry_run)
6✔
324
                create_func = partial(resource.create, dry_run=dry_run)
6✔
325
                delete_func = partial(resource.delete, dry_run=dry_run)
6✔
326

327
            created, patched, deleted = self._apply_resource(dry_run,
6✔
328
                                                             patch_field_excludes,
329
                                                             resource,
330
                                                             patch_func,
331
                                                             create_func,
332
                                                             delete_func,
333
                                                             status_msg)
334

335
            total_created += created
6✔
336
            total_patched += patched
6✔
337
            total_deleted += deleted
6✔
338

339
        if ((dump or dry_run) and
6✔
340
                k8s.field_validation_warn_fatal and self.context.globals.k8s.field_validation_warnings):
341
            msg = ("There were %d field validation warnings and the warnings are fatal!" %
6✔
342
                   self.context.globals.k8s.field_validation_warnings)
343
            logger.fatal(msg)
6✔
344
            raise RuntimeError(msg)
6✔
345

346
        if dump:
6✔
347
            if file_format in ("json", "json-pretty"):
6!
348
                json.dump(dump_results, file, sort_keys=True,
×
349
                          indent=4 if file_format == "json-pretty" else None)
350
            else:
351
                yaml.safe_dump(dump_results, file)
6✔
352
        else:
353
            self._summary = total_created, total_patched, total_deleted
6✔
354

355
    def handle_summary(self):
6✔
356
        total_created, total_patched, total_deleted = self._summary
6✔
357
        logger.info("Created %d, patched %d, deleted %d resources", total_created, total_patched, total_deleted)
6✔
358

359
    def api_load_resources(self, path: Path, file_type: str):
6✔
360
        return self.add_local_resources(path, FileType[file_type.upper()])
×
361

362
    def api_load_remote_resources(self, url: str, file_type: str, file_category=None):
6✔
363
        return self.add_remote_resources(url, FileType[file_type.upper()], sub_category=file_category)
×
364

365
    def api_load_crds(self, path: Path, file_type: str):
6✔
366
        return self.add_local_crds(path, FileType[file_type.upper()])
6✔
367

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

371
    def api_import_cluster_crds(self):
6✔
NEW
372
        context = self.context
×
NEW
373
        k8s = context.k8s
×
NEW
374
        client = k8s.client
×
NEW
375
        from kubernetes import client as client_module
×
376

NEW
377
        api = client_module.ApiextensionsV1Api(client)
×
NEW
378
        crds = api.list_custom_resource_definition(watch=False)
×
NEW
379
        for crd in crds.items:
×
NEW
380
            manifest = client.sanitize_for_serialization(crd)
×
NEW
381
            manifest["apiVersion"] = "apiextensions.k8s.io/v1"
×
NEW
382
            manifest["kind"] = "CustomResourceDefinition"
×
NEW
383
            self.add_crd(manifest)
×
384

385
    def api_add_transformer(self, transformer):
6✔
386
        if transformer not in self._transformers:
6!
387
            self._transformers.append(transformer)
6✔
388

389
    def api_add_validator(self, validator):
6✔
390
        if validator not in self._validators:
6!
391
            self._validators.append(validator)
6✔
392

393
    def api_add_manifest_patcher(self, patcher):
6✔
394
        if patcher not in self._manifest_patchers:
×
395
            self._manifest_patchers.append(patcher)
×
396

397
    def api_remove_transformer(self, transformer):
6✔
398
        if transformer in self._transformers:
6!
399
            self._transformers.remove(transformer)
6✔
400

401
    def api_remove_validator(self, validator):
6✔
402
        if validator not in self._validators:
×
403
            self._validators.remove(validator)
×
404

405
    def api_validation_error(self, msg, *args):
6✔
406
        frame = sys._getframe().f_back
×
407
        tb = None
×
408
        while True:
409
            if not frame:
×
410
                break
×
411
            tb = types.TracebackType(tb, frame, frame.f_lasti, frame.f_lineno)
×
412
            frame = frame.f_back
×
413
        return ValueError((msg % args) if args else msg).with_traceback(tb)
×
414

415
    def _patch_manifest(self,
6✔
416
                        manifest: dict,
417
                        resource_description: str):
418
        for patcher in reversed(self._manifest_patchers):
6!
419
            logger.debug("Applying patcher %s to %s",
×
420
                         getattr(patcher, "__name__", patcher),
421
                         resource_description)
422
            manifest = patcher(manifest, resource_description) or manifest
×
423

424
        return manifest
6✔
425

426
    def _transform_resource(self, resources: Sequence[K8SResource], resource: K8SResource) -> K8SResource:
6✔
427
        for transformer in reversed(self._transformers):
6✔
428
            logger.debug("Applying transformer %s to %s from %s",
6✔
429
                         getattr(transformer, "__name__", transformer),
430
                         resource, resource.source)
431
            resource = transformer(resources, resource) or resource
6✔
432

433
        return resource
6✔
434

435
    def _validate_resources(self):
6✔
436
        errors: list[Exception] = []
6✔
437
        for resource in self.resources.values():
6✔
438
            for validator in reversed(self._validators):
6✔
439
                logger.debug("Applying validator %s to %s from %s",
6✔
440
                             getattr(validator, "__name__", validator),
441
                             resource, resource.source)
442
                errors.extend(validator(self.resources, resource, self.api_validation_error))
6✔
443
        if errors:
6!
444
            for error in errors:
×
445
                logger.error("Validation error: %s", error)
×
446
            raise errors[0]
×
447

448
    def _apply_resource(self,
6✔
449
                        dry_run,
450
                        patch_field_excludes: Iterable[re.compile],
451
                        resource: K8SResource,
452
                        patch_func: Callable[[Iterable[dict]], None],
453
                        create_func: Callable[[], None],
454
                        delete_func: Callable[[K8SPropagationPolicy], None],
455
                        status_msg):
456
        from kubernetes import client
6✔
457
        from kubernetes.client.rest import ApiException
6✔
458

459
        rdef = resource.rdef
6✔
460
        rdef.populate_api(client, self.context.k8s.client)
6✔
461

462
        def handle_400_strict_validation_error(e: ApiException):
6✔
463
            if e.status == 400:
6!
464
                # Assumes the body has been parsed
465
                status = e.body
6✔
466
                if status["status"] == "Failure":
6!
467
                    if FIELD_VALIDATION_STRICT_MARKER in status["message"]:
6!
468
                        message = status["message"]
6✔
469
                        messages = message[message.find(FIELD_VALIDATION_STRICT_MARKER) +
6✔
470
                                           len(FIELD_VALIDATION_STRICT_MARKER):].split(",")
471
                        for m in messages:
6✔
472
                            self._api_warnings(resource, m.strip())
6✔
473

474
                        raise e from None
6✔
475
                    else:
476
                        logger.error("FAILED MODIFYING resource %s from %s: %s",
×
477
                                     resource, resource.source, status["message"])
478
                        raise e from None
×
479

480
        def create(exists_ok=False, wait_for_delete=False):
6✔
481
            logger.info("Creating resource %s%s%s", resource, status_msg,
6✔
482
                        " (ignoring existing)" if exists_ok else "")
483
            while True:
5✔
484
                try:
6✔
485
                    create_func()
6✔
486
                    return
6✔
487
                except ApiException as __e:
6✔
488
                    api_exc_normalize_body(__e)
6✔
489
                    try:
6✔
490
                        if exists_ok or wait_for_delete:
6!
491
                            if __e.status == 409:
×
492
                                status = __e.body
×
493
                                if status["reason"] == "AlreadyExists":
×
494
                                    if wait_for_delete:
×
495
                                        sleep(self.context.k8s.conflict_retry_delay)
×
496
                                        logger.info("Retry creating resource %s%s%s", resource, status_msg,
×
497
                                                    " (ignoring existing)" if exists_ok else "")
498
                                        continue
×
499
                                    else:
500
                                        return
×
501
                        raise
6✔
502
                    except ApiException as ___e:
6✔
503
                        api_exc_format_body(___e)
6✔
504
                        raise
6✔
505

506
        merge_instrs, normalized_manifest = extract_merge_instructions(resource.manifest, resource)
6✔
507
        if merge_instrs:
6✔
508
            logger.trace("Normalized manifest (no merge instructions) for resource %s: %s", resource,
6✔
509
                         normalized_manifest)
510
        else:
511
            normalized_manifest = resource.manifest
6✔
512

513
        logger.debug("Applying resource %s%s", resource, status_msg)
6✔
514
        try:
6✔
515
            remote_resource = resource.get()
6✔
516
            logger.trace("Current resource %s: %s", resource, remote_resource)
6✔
517
        except ApiException as e:
6✔
518
            api_exc_normalize_body(e)
6✔
519
            try:
6✔
520
                if e.status == 404:
6!
521
                    try:
6✔
522
                        create()
6✔
523
                        return 1, 0, 0
6✔
524
                    except ApiException as e:
6✔
525
                        api_exc_normalize_body(e)
6✔
526
                        if not handle_400_strict_validation_error(e):
6!
527
                            raise
1✔
528
                else:
529
                    raise
×
530
            except ApiException as _e:
6✔
531
                api_exc_format_body(_e)
6✔
532
                raise
6✔
533
        else:
534
            logger.trace("Attempting to retrieve a normalized patch for resource %s: %s", resource, normalized_manifest)
6✔
535
            try:
6✔
536
                merged_resource = resource.patch(normalized_manifest,
6✔
537
                                                 patch_type=K8SResourcePatchType.SERVER_SIDE_PATCH,
538
                                                 dry_run=True,
539
                                                 force=True)
540
            except ApiException as e:
×
541
                try:
×
542
                    api_exc_normalize_body(e)
×
543

544
                    if e.status == 422:
×
545
                        status = e.body
×
546
                        # Assumes the body has been unmarshalled
547
                        details = status["details"]
×
548
                        immutable_key = details.get("group"), details["kind"]
×
549

550
                        try:
×
551
                            propagation_policy = self.context.k8s.immutable_changes[immutable_key]
×
552
                        except KeyError:
×
553
                            raise e from None
×
554
                        else:
555
                            for cause in details["causes"]:
×
556
                                if (
×
557
                                        cause["reason"] == "FieldValueInvalid" and
558
                                        "field is immutable" in cause["message"]
559
                                        or
560
                                        cause["reason"] == "FieldValueForbidden" and
561
                                        ("Forbidden: updates to" in cause["message"]
562
                                         or
563
                                         "Forbidden: pod updates" in cause["message"])
564
                                ):
565
                                    logger.info("Deleting resource %s (cascade %s)%s", resource,
×
566
                                                propagation_policy.policy,
567
                                                status_msg)
568
                                    delete_func(propagation_policy=propagation_policy)
×
569
                                    create(exists_ok=dry_run, wait_for_delete=not dry_run)
×
570
                                    return 1, 0, 1
×
571
                            raise
×
572
                    else:
573
                        if not handle_400_strict_validation_error(e):
×
574
                            raise
×
575
                except ApiException as _e:
×
576
                    api_exc_format_body(_e)
×
577
                    raise
×
578

579
            else:
580
                logger.trace("Merged resource %s: %s", resource, merged_resource)
6✔
581
                if merge_instrs:
6✔
582
                    apply_merge_instructions(merge_instrs, normalized_manifest, merged_resource, logger, resource)
6✔
583

584
                patch = jsonpatch.make_patch(remote_resource, merged_resource)
6✔
585
                logger.trace("Resource %s initial patches are: %s", resource, patch)
6✔
586
                patch = self._filter_resource_patch(patch, patch_field_excludes)
6✔
587
                logger.trace("Resource %s final patches are: %s", resource, patch)
6✔
588
                if patch:
6✔
589
                    logger.info("Patching resource %s%s", resource, status_msg)
6✔
590
                    patch_func(patch)
6✔
591
                    return 0, 1, 0
6✔
592
                else:
593
                    logger.info("Nothing to patch for resource %s", resource)
6✔
594
                    return 0, 0, 0
6✔
595

596
    def _filter_resource_patch(self, patch: Iterable[Mapping], excludes: Iterable[re.compile]):
6✔
597
        result = []
6✔
598
        for op in patch:
6✔
599
            path = op["path"]
6✔
600
            excluded = False
6✔
601
            for exclude in excludes:
6✔
602
                if exclude.match(path):
6✔
603
                    logger.trace("Excluding %r from patch %s", op, patch)
6✔
604
                    excluded = True
6✔
605
                    break
6✔
606
            if excluded:
6✔
607
                continue
6✔
608
            result.append(op)
6✔
609
        return result
6✔
610

611
    def _setup_k8s_client(self):
6✔
612
        from kubernetes import client
6✔
613
        from kubernetes.config import load_incluster_config, load_kube_config, ConfigException
6✔
614

615
        try:
6✔
616
            logger.debug("Trying K8S in-cluster configuration")
6✔
617
            load_incluster_config()
6✔
618
            logger.info("Running K8S with in-cluster configuration")
×
619
        except ConfigException as e:
6✔
620
            logger.trace("K8S in-cluster configuration failed", exc_info=e)
6✔
621
            logger.debug("Initializing K8S with kubeconfig configuration")
6✔
622
            load_kube_config(config_file=self.context.kubeconfig.kubeconfig)
6✔
623

624
        k8s_client = client.ApiClient()
6✔
625

626
        # Patch the header content type selector to allow json patch
627
        k8s_client._select_header_content_type = k8s_client.select_header_content_type
6✔
628
        k8s_client.select_header_content_type = self._select_header_content_type_patch
6✔
629

630
        return k8s_client
6✔
631

632
    def _select_header_content_type_patch(self, content_types):
6✔
633
        """Returns `Content-Type` based on an array of content_types provided.
634
        :param content_types: List of content-types.
635
        :return: Content-Type (e.g. application/json).
636
        """
637

638
        content_type = self.context.k8s.client._select_header_content_type(content_types)
×
639
        if content_type == "application/merge-patch+json":
×
640
            return "application/json-patch+json"
×
641
        return content_type
×
642

643
    def __repr__(self):
6✔
644
        return "Kubernetes Plugin"
6✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc