18262235457

Committed 05 Oct 2025 05:48PM UTC coverage: 75.964% (+0.05%) from 75.914%

Build # 18262235457

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #78 from karellen/templates_everywhere

Basic (non-reproducible) templating is now available in regular k8s manifests and Helm.

Run Details

614 of 958 branches covered (64.09%)

Branch coverage included in aggregate %.

42 of 48 new or added lines in 5 files covered. (87.5%)

1 existing line in 1 file now uncovered.

2401 of 3011 relevant lines covered (79.74%)

4.78 hits per line

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

80.11

/src/main/python/kubernator/plugins/helm.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 os
import tarfile
import tempfile
from hashlib import sha256
from pathlib import Path
from shutil import which, copy
from typing import Sequence

import yaml
from jsonschema import Draft7Validator

from kubernator.api import (KubernatorPlugin, Globs, StripNL,
                            scan_dir,
                            load_file,
                            FileType,
                            calling_frame_source,
                            validator_with_defaults,
                            get_golang_os,
                            get_golang_machine,
                            prepend_os_path, TemplateEngine
                            )
from kubernator.plugins.k8s_api import K8SResource
from kubernator.proc import DEVNULL

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

HELM_SCHEMA = {
    "properties": {
        "repository": {
            "type": "string",
            "description": "repository hosting the charts"
        },
        "chart": {
            "type": "string",
            "description": "chart name within the repository"
        },
        "version": {
            "type": "string",
            "description": "chart version"
        },
        "name": {
            "type": "string",
            "description": "name of the particular release"
        },
        "namespace": {
            "type": "string",
            "description": "namespace where the chart is going to be deployed"
        },
        "include-crds": {
            "type": "boolean",
            "default": True,
            "description": "generate CRDs for the helm"
        },
        "values": {
            "type": "object",
            "additionalProperties": True,
            "description": "inline Helm chart values"
        },
        "values-file": {
            "type": "string",
            "description": "reference to the file containing Helm chart values"
        }
    },
    "type": "object",
    "required": ["chart", "name", "namespace"]
}

Draft7Validator.check_schema(HELM_SCHEMA)
HELM_VALIDATOR_CLS = validator_with_defaults(Draft7Validator)
HELM_VALIDATOR = HELM_VALIDATOR_CLS(HELM_SCHEMA, format_checker=Draft7Validator.FORMAT_CHECKER)


class HelmPlugin(KubernatorPlugin):
    logger = logger

    _name = "helm"

    def __init__(self):
        self.context = None
        self.repositories = set()
        self.helm_dir = None
        self.template_engine = TemplateEngine(logger)

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

    def stanza(self):
        context = self.context
        stanza = [context.helm.helm_file, f"--kubeconfig={context.kubeconfig.kubeconfig}"]
        if logger.getEffectiveLevel() < logging.INFO:
            stanza.append("--debug")
        return stanza

    def register(self, version=None):
        context = self.context
        context.app.register_plugin("kubeconfig")
        context.app.register_plugin("k8s")

        if version:
            # Download and use specific version
            helm_url = f"https://get.helm.sh/helm-v{version}-{get_golang_os()}-{get_golang_machine()}.tar.gz"
            helm_file_dl, _ = context.app.download_remote_file(logger, helm_url, "bin")
            helm_file_dl = str(helm_file_dl)
            self.helm_dir = tempfile.TemporaryDirectory()
            context.app.register_cleanup(self.helm_dir)

            helm_file = str(Path(self.helm_dir.name) / "helm")
            helm_tar = tarfile.open(helm_file_dl)
            helm_tar.extractall(self.helm_dir.name)

            copy(Path(self.helm_dir.name) / f"{get_golang_os()}-{get_golang_machine()}" / "helm", helm_file)

            os.chmod(helm_file, 0o500)
            prepend_os_path(self.helm_dir.name)
        else:
            # Use current version
            helm_file = which("helm")
            if not helm_file:
                raise RuntimeError("`helm` cannot be found and no version has been specified")

            logger.debug("Found Helm in %r", helm_file)

        context.globals.helm = dict(default_includes=Globs(["*.helm.yaml", "*.helm.yml"], True),
                                    default_excludes=Globs([".*"], True),
                                    namespace_transformer=True,
                                    helm_file=helm_file,
                                    stanza=self.stanza,
                                    add_helm_template=self.add_helm_template,
                                    add_helm=self.add_helm,
                                    )

    def handle_init(self):
        version = self.context.app.run_capturing_out(self.stanza() + ["version", "--template", "{{.Version}}"],
                                                     logger.error)
        logger.info("Found Helm version %s", version)

    def handle_start(self):
        pass

    def handle_before_dir(self, cwd: Path):
        context = self.context

        context.helm.default_includes = Globs(context.helm.default_includes)
        context.helm.default_excludes = Globs(context.helm.default_excludes)
        context.helm.includes = Globs(context.helm.default_includes)
        context.helm.excludes = Globs(context.helm.default_excludes)

        # Exclude Helm YAMLs from K8S resource loading
        context.k8s.excludes.add("*.helm.yaml")
        context.k8s.excludes.add("*.helm.yml")

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

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

            helm_templates = load_file(logger, p, FileType.YAML, display_p,
                                       self.template_engine,
                                       {"ktor": context})

            for helm_template in helm_templates:
                self._add_helm(helm_template, display_p)

    def add_helm_template(self, template):
        return self._add_helm(template, calling_frame_source())

    def add_helm(self, **kwargs):
        return self._internal_add_helm(calling_frame_source(), **kwargs)

    def _add_helm(self, template, source):
        errors = list(HELM_VALIDATOR.iter_errors(template))
        if errors:
            for error in errors:
                self.logger.error("Error detected in Helm template from %s", source, exc_info=error)
            raise errors[0]

        return self._internal_add_helm(source, **{k.replace("-", "_"): v for k, v in template.items()})

    def _add_repository(self, repository: str):
        repository_hash = sha256(repository.encode("UTF-8")).hexdigest()
        logger.debug("Repository %s mapping to %s", repository, repository_hash)
        if repository_hash not in self.repositories:
            logger.info("Adding and updating repository %s mapping to %s", repository, repository_hash)
            self.context.app.run(self.stanza() + ["repo", "add", repository_hash, repository],
                                 stdout_logger,
                                 stderr_logger).wait()
            self.context.app.run(self.stanza() + ["repo", "update"],
                                 stdout_logger,
                                 stderr_logger).wait()
            self.repositories.add(repository_hash)

        return repository_hash

    def _internal_add_helm(self, source, *, chart, name, namespace, include_crds,
                           values=None, values_file=None, repository=None, version=None):
        if values and values_file:
            raise RuntimeError(f"In {source} either values or values file may be specified, but not both")

        if (repository and chart and chart.startswith("oci://") or
                not repository and chart and not chart.startswith("oci://")):
            raise RuntimeError(
                f"In {source} either repository must be specified or OCI-chart must be used, but not both")

        if not version and repository:
            raise RuntimeError(f"In {source} version must be specified unless OCI-chart is used")

        if values_file:
            values_file = Path(values_file)
            if not values_file.is_absolute():
                values_file = self.context.app.cwd / values_file
            values = list(load_file(logger, values_file, FileType.YAML,
                                    template_engine=self.template_engine,
                                    template_context={"ktor": self.context,
                                                      "helm": {"chart": chart,
                                                               "name": name,
                                                               "namespace": namespace,
                                                               "include_crds": include_crds,
                                                               "repository": repository,
                                                               "version": version,
                                                               }}))
            values = values[0] if values else {}

        version_spec = []
        if repository:
            repository_hash = self._add_repository(repository)
            chart_name = f"{repository_hash}/{chart}"
        else:
            chart_name = chart

        if version:
            version_spec = ["--version", version]

        stdin = DEVNULL

        if values:
            def write_stdin():
                return json.dumps(values)

            stdin = write_stdin

        resources = self.context.app.run_capturing_out(self.stanza() +
                                                       ["template",
                                                        name,
                                                        chart_name,
                                                        "-n", namespace,
                                                        "-a", ",".join(self.context.k8s.get_api_versions())
                                                        ] +
                                                       version_spec +
                                                       (["--include-crds"] if include_crds else []) +
                                                       ["-f", "-"],
                                                       stderr_logger,
                                                       stdin=stdin,
                                                       )

        def helm_namespace_transformer(resources: Sequence[K8SResource],
                                       resource: K8SResource):
            if resource.rdef.namespaced and not resource.namespace:
                resource.namespace = namespace
                return resource

        if self.context.helm.namespace_transformer:
            self.context.k8s.add_transformer(helm_namespace_transformer)

        self.context.k8s.add_resources(yaml.safe_load_all(resources), source)

        if self.context.helm.namespace_transformer:
            self.context.k8s.remove_transformer(helm_namespace_transformer)

    def __repr__(self):
        return "Helm 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	import json	6✔
20	import logging	6✔
21	import os	6✔
22	import tarfile	6✔
23	import tempfile	6✔
24	from hashlib import sha256	6✔
25	from pathlib import Path	6✔
26	from shutil import which, copy	6✔
27	from typing import Sequence	6✔
28
29	import yaml	6✔
30	from jsonschema import Draft7Validator	6✔
31
32	from kubernator.api import (KubernatorPlugin, Globs, StripNL,	6✔
33	scan_dir,
34	load_file,
35	FileType,
36	calling_frame_source,
37	validator_with_defaults,
38	get_golang_os,
39	get_golang_machine,
40	prepend_os_path, TemplateEngine
41	)
42	from kubernator.plugins.k8s_api import K8SResource	6✔
43	from kubernator.proc import DEVNULL	6✔
44
45	logger = logging.getLogger("kubernator.helm")	6✔
46	proc_logger = logger.getChild("proc")	6✔
47	stdout_logger = StripNL(proc_logger.info)	6✔
48	stderr_logger = StripNL(proc_logger.warning)	6✔
49
50	HELM_SCHEMA = {	6✔
51	"properties": {
52	"repository": {
53	"type": "string",
54	"description": "repository hosting the charts"
55	},
56	"chart": {
57	"type": "string",
58	"description": "chart name within the repository"
59	},
60	"version": {
61	"type": "string",
62	"description": "chart version"
63	},
64	"name": {
65	"type": "string",
66	"description": "name of the particular release"
67	},
68	"namespace": {
69	"type": "string",
70	"description": "namespace where the chart is going to be deployed"
71	},
72	"include-crds": {
73	"type": "boolean",
74	"default": True,
75	"description": "generate CRDs for the helm"
76	},
77	"values": {
78	"type": "object",
79	"additionalProperties": True,
80	"description": "inline Helm chart values"
81	},
82	"values-file": {
83	"type": "string",
84	"description": "reference to the file containing Helm chart values"
85	}
86	},
87	"type": "object",
88	"required": ["chart", "name", "namespace"]
89	}
90
91	Draft7Validator.check_schema(HELM_SCHEMA)	6✔
92	HELM_VALIDATOR_CLS = validator_with_defaults(Draft7Validator)	6✔
93	HELM_VALIDATOR = HELM_VALIDATOR_CLS(HELM_SCHEMA, format_checker=Draft7Validator.FORMAT_CHECKER)	6✔
94
95
96	class HelmPlugin(KubernatorPlugin):	6✔
97	logger = logger	6✔
98
99	_name = "helm"	6✔
100
101	def __init__(self):	6✔
102	self.context = None	6✔
103	self.repositories = set()	6✔
104	self.helm_dir = None	6✔
105	self.template_engine = TemplateEngine(logger)	6✔
106
107	def set_context(self, context):	6✔
108	self.context = context	6✔
109
110	def stanza(self):	6✔
111	context = self.context	6✔
112	stanza = [context.helm.helm_file, f"--kubeconfig={context.kubeconfig.kubeconfig}"]	6✔
113	if logger.getEffectiveLevel() < logging.INFO:	6!
114	stanza.append("--debug")	6✔
115	return stanza	6✔
116
117	def register(self, version=None):	6✔
118	context = self.context	6✔
119	context.app.register_plugin("kubeconfig")	6✔
120	context.app.register_plugin("k8s")	6✔
121
122	if version:	6!
123	# Download and use specific version
124	helm_url = f"https://get.helm.sh/helm-v{version}-{get_golang_os()}-{get_golang_machine()}.tar.gz"	6✔
125	helm_file_dl, _ = context.app.download_remote_file(logger, helm_url, "bin")	6✔
126	helm_file_dl = str(helm_file_dl)	6✔
127	self.helm_dir = tempfile.TemporaryDirectory()	6✔
128	context.app.register_cleanup(self.helm_dir)	6✔
129
130	helm_file = str(Path(self.helm_dir.name) / "helm")	6✔
131	helm_tar = tarfile.open(helm_file_dl)	6✔
132	helm_tar.extractall(self.helm_dir.name)	6✔
133
134	copy(Path(self.helm_dir.name) / f"{get_golang_os()}-{get_golang_machine()}" / "helm", helm_file)	6✔
135
136	os.chmod(helm_file, 0o500)	6✔
137	prepend_os_path(self.helm_dir.name)	6✔
138	else:
139	# Use current version
140	helm_file = which("helm")	×
141	if not helm_file:	×
142	raise RuntimeError("`helm` cannot be found and no version has been specified")	×
143
144	logger.debug("Found Helm in %r", helm_file)	×
145
146	context.globals.helm = dict(default_includes=Globs([".helm.yaml", ".helm.yml"], True),	6✔
147	default_excludes=Globs([".*"], True),
148	namespace_transformer=True,
149	helm_file=helm_file,
150	stanza=self.stanza,
151	add_helm_template=self.add_helm_template,
152	add_helm=self.add_helm,
153	)
154
155	def handle_init(self):	6✔
156	version = self.context.app.run_capturing_out(self.stanza() + ["version", "--template", "{{.Version}}"],	6✔
157	logger.error)
158	logger.info("Found Helm version %s", version)	6✔
159
160	def handle_start(self):	6✔
161	pass	6✔
162
163	def handle_before_dir(self, cwd: Path):	6✔
164	context = self.context	6✔
165
166	context.helm.default_includes = Globs(context.helm.default_includes)	6✔
167	context.helm.default_excludes = Globs(context.helm.default_excludes)	6✔
168	context.helm.includes = Globs(context.helm.default_includes)	6✔
169	context.helm.excludes = Globs(context.helm.default_excludes)	6✔
170
171	# Exclude Helm YAMLs from K8S resource loading
172	context.k8s.excludes.add("*.helm.yaml")	6✔
173	context.k8s.excludes.add("*.helm.yml")	6✔
174
175	def handle_after_dir(self, cwd: Path):	6✔
176	context = self.context	6✔
177	helm = context.helm	6✔
178
179	for f in scan_dir(logger, cwd, lambda d: d.is_file(), helm.excludes, helm.includes):	6✔
180	p = cwd / f.name	6✔
181	display_p = context.app.display_path(p)	6✔
182	logger.debug("Adding Helm template from %s", display_p)	6✔
183
184	helm_templates = load_file(logger, p, FileType.YAML, display_p,	6✔
185	self.template_engine,
186	{"ktor": context})
187
188	for helm_template in helm_templates:	6✔
189	self._add_helm(helm_template, display_p)	6✔
190
191	def add_helm_template(self, template):	6✔
192	return self._add_helm(template, calling_frame_source())	×
193
194	def add_helm(self, **kwargs):	6✔
195	return self._internal_add_helm(calling_frame_source(), **kwargs)	×
196
197	def _add_helm(self, template, source):	6✔
198	errors = list(HELM_VALIDATOR.iter_errors(template))	6✔
199	if errors:	6!
200	for error in errors:	×
201	self.logger.error("Error detected in Helm template from %s", source, exc_info=error)	×
202	raise errors[0]	×
203
204	return self._internal_add_helm(source, **{k.replace("-", "_"): v for k, v in template.items()})	6✔
205
206	def _add_repository(self, repository: str):	6✔
207	repository_hash = sha256(repository.encode("UTF-8")).hexdigest()	6✔
208	logger.debug("Repository %s mapping to %s", repository, repository_hash)	6✔
209	if repository_hash not in self.repositories:	6!
210	logger.info("Adding and updating repository %s mapping to %s", repository, repository_hash)	6✔
211	self.context.app.run(self.stanza() + ["repo", "add", repository_hash, repository],	6✔
212	stdout_logger,
213	stderr_logger).wait()
214	self.context.app.run(self.stanza() + ["repo", "update"],	6✔
215	stdout_logger,
216	stderr_logger).wait()
217	self.repositories.add(repository_hash)	6✔
218
219	return repository_hash	6✔
220
221	def _internal_add_helm(self, source, *, chart, name, namespace, include_crds,	6✔
222	values=None, values_file=None, repository=None, version=None):
223	if values and values_file:	6!
224	raise RuntimeError(f"In {source} either values or values file may be specified, but not both")	×
225
226	if (repository and chart and chart.startswith("oci://") or	6✔
227	not repository and chart and not chart.startswith("oci://")):
228	raise RuntimeError(	6✔
229	f"In {source} either repository must be specified or OCI-chart must be used, but not both")
230
231	if not version and repository:	6✔
232	raise RuntimeError(f"In {source} version must be specified unless OCI-chart is used")	6✔
233
234	if values_file:	6!
235	values_file = Path(values_file)	×
236	if not values_file.is_absolute():	×
237	values_file = self.context.app.cwd / values_file	×
NEW 238	values = list(load_file(logger, values_file, FileType.YAML,	×
239	template_engine=self.template_engine,
240	template_context={"ktor": self.context,
241	"helm": {"chart": chart,
242	"name": name,
243	"namespace": namespace,
244	"include_crds": include_crds,
245	"repository": repository,
246	"version": version,
247	}}))
NEW 248	values = values[0] if values else {}	×
249
250	version_spec = []	6✔
251	if repository:	6✔
252	repository_hash = self._add_repository(repository)	6✔
253	chart_name = f"{repository_hash}/{chart}"	6✔
254	else:
255	chart_name = chart	6✔
256
257	if version:	6✔
258	version_spec = ["--version", version]	6✔
259
260	stdin = DEVNULL	6✔
261
262	if values:	6!
263	def write_stdin():	×
264	return json.dumps(values)	×
265
266	stdin = write_stdin	×
267
268	resources = self.context.app.run_capturing_out(self.stanza() +	6✔
269	["template",
270	name,
271	chart_name,
272	"-n", namespace,
273	"-a", ",".join(self.context.k8s.get_api_versions())
274	] +
275	version_spec +
276	(["--include-crds"] if include_crds else []) +
277	["-f", "-"],
278	stderr_logger,
279	stdin=stdin,
280	)
281
282	def helm_namespace_transformer(resources: Sequence[K8SResource],	6✔
283	resource: K8SResource):
284	if resource.rdef.namespaced and not resource.namespace:	6!
285	resource.namespace = namespace	×
286	return resource	×
287
288	if self.context.helm.namespace_transformer:	6!
289	self.context.k8s.add_transformer(helm_namespace_transformer)	6✔
290
291	self.context.k8s.add_resources(yaml.safe_load_all(resources), source)	6✔
292
293	if self.context.helm.namespace_transformer:	6!
294	self.context.k8s.remove_transformer(helm_namespace_transformer)	6✔
295
296	def __repr__(self):	6✔
297	return "Helm Plugin"	6✔

karellen / kubernator / 18262235457

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