15104716107

Committed 19 May 2025 05:02AM UTC coverage: 76.294% (-0.03%) from 76.325%

Build # 15104716107

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #74 from karellen/upgrade_dependencies

Upgrade jsonpath-ng, jsonpatch and jsonschema dependencies

Run Details

615 of 951 branches covered (64.67%)

Branch coverage included in aggregate %.

9 of 9 new or added lines in 3 files covered. (100.0%)

4 existing lines in 2 files now uncovered.

2378 of 2972 relevant lines covered (80.01%)

3.99 hits per line

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

17.18

/src/main/python/kubernator/plugins/kops.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
from pathlib import Path
from tempfile import TemporaryDirectory

import yaml

from kubernator.api import (KubernatorPlugin, scan_dir,
                            FileType,
                            load_remote_file,
                            io_StringIO,
                            TemplateEngine,
                            StripNL,
                            Globs)
from kubernator.plugins.k8s_api import K8SResourcePluginMixin
from kubernator.proc import CalledProcessError

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

KOPS_CRDS = ["kops.k8s.io_clusters.yaml",
             "kops.k8s.io_instancegroups.yaml",
             "kops.k8s.io_keysets.yaml",
             "kops.k8s.io_sshcredentials.yaml"]

OBJECT_SCHEMA_VERSION = "1.20.6"


class KopsPlugin(KubernatorPlugin, K8SResourcePluginMixin):
    logger = logger

    _name = "kops"

    def __init__(self):
        self.context = None
        self.kops_stanza = ["kops"]
        self.version = None
        self.kops_dir = None
        self.kops_path = None

        self.template_engine = TemplateEngine(logger)
        super().__init__()

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

    def register(self, helm_version=None):
        raise RuntimeError("Currently disabled, please don't use")
        self.context.app.register_plugin("kubeconfig")
        self.context.app.register_plugin("kubectl")

    def handle_init(self):
        context = self.context

        # Temp dir
        self.kops_dir = TemporaryDirectory()
        context.app.register_cleanup(self.kops_dir)
        self.kops_path = kops_path = Path(self.kops_dir.name)
        kops_config = kops_path / ".kops.yaml"
        kops_config.touch()
        self.kops_stanza.extend(("--config", str(kops_config)))

        # Kube config dir
        kubeconfig_dir = kops_path / ".kube"
        kubeconfig_dir.mkdir()
        self.kubeconfig_path = kubeconfig_dir / "config"

        # if logger.getEffectiveLevel() < logging.DEBUG:
        #    self.kops_stanza.append("-v5")
        # elif logger.getEffectiveLevel() < logging.INFO:
        #    self.kops_stanza.append("-v2")

        context.globals.kops = dict(walk_remote=self.walk_remote,
                                    walk_local=self.walk_local,
                                    update=self.update,
                                    export=self.export,
                                    master_interval="8m",
                                    node_interval="8m"
                                    )
        context.kops = dict()

        self.version = context.app.run_capturing_out(["kops", "version", "--short"], stderr_logger).strip()
        logger.info("Found kOps version %s", self.version)

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

        common_url_path = f"https://raw.githubusercontent.com/kubernetes/kops/v{self.version}/k8s/crds"
        for kops_crd in KOPS_CRDS:
            url = f"{common_url_path}/{kops_crd}"
            self.add_remote_crds(url, FileType.YAML, sub_category="kops")

    def handle_start(self):
        context = self.context

        # Exclude Kops YAMLs from K8S resource loading
        context.k8s.default_excludes.add("*.kops.yaml")
        context.k8s.default_excludes.add("*.kops.yml")

    def walk_remote(self, url, *paths: Path,
                    excludes=(".*",), includes=("*.kops.yaml", "*.kops.yml")):
        repository = self.context.app.repository(url)
        for path in paths:
            self.walk_local(repository.local_dir / path, excludes, includes)

    def walk_local(self, path: Path, excludes=(".*",), includes=("*.kops.yaml", "*.kops.yml")):
        context = self.context
        path = Path(path)

        for f in scan_dir(logger, path, lambda d: d.is_file(), Globs(excludes), Globs(includes)):
            p = path / f.name
            display_p = context.app.display_path(p)
            logger.debug("Adding Kops resources from %s", display_p)

            with open(p, "rt") as file:
                template = self.template_engine.from_string(file.read())

            self.add_resources(template.render({"ktor": context}), display_p)

    def update(self):
        context = self.context
        run = context.app.run
        run_capturing_out = context.app.run_capturing_out

        os.environ["KUBECONFIG"] = str(self.kubeconfig_path)
        os.environ["KOPS_CLUSTER_NAME"] = context.kops.cluster_name
        os.environ["KOPS_STATE_STORE"] = context.kops.state_store

        kops_extra_args = ["--name", context.kops.cluster_name, "--state", context.kops.state_store]

        cmd = context.app.args.command
        dry_run = context.app.args.dry_run

        for resource in self.resources.values():
            logger.info("Replacing/creating kOps resource %s", resource)
            resource_out = io_StringIO()
            yaml.dump(resource.manifest, resource_out)
            if cmd != "apply" or dry_run:
                logger.info("Would replace kOps resource if not for dry-run mode: %s", resource_out.getvalue())
            else:
                run(self.kops_stanza + ["replace", "--force", "-f", "-"] + kops_extra_args,
                    stdout_logger,
                    stderr_logger,
                    resource_out.getvalue()).wait()

        logger.info("Staging kOps update")
        update_cmd = self.kops_stanza + ["update", "cluster"] + kops_extra_args
        result = run_capturing_out(update_cmd, stderr_logger)
        proc_logger.info(result)
        if "Must specify --yes to apply changes" in result:
            logger.info("kOps update would make changes")
            if cmd != "apply" or dry_run:
                logger.info("Skipping actual kOps update due to dry-run mode")
            else:
                logger.info("Running kOps update")
                run(update_cmd + ["--yes"],
                    stdout_logger,
                    stderr_logger).wait()

        self.export()

        if cmd != "apply":
            logger.info("Skipping cluster validation since not in apply mode")
        else:
            validation_failed = False
            try:
                output = run_capturing_out(self.kops_stanza + ["validate", "cluster", "-o", "json"],
                                           stderr_logger)
            except CalledProcessError as e:
                validation_failed = True
                output = e.output

            if validation_failed:
                if output:
                    validation_results = json.loads(output)
                    for failure in validation_results["failures"]:
                        logger.error("%s %s failed validation: %s", failure["type"], failure["name"],
                                     failure["message"])
                raise RuntimeError("Cluster validation failed!")
            else:
                logger.info("Cluster validation successful!")

        logger.info("Staging kOps cluster rolling update")
        rolling_update_cmd = self.kops_stanza + ["rolling-update", "cluster",
                                                 "--master-interval", context.kops.master_interval,
                                                 "--node-interval", context.kops.node_interval] + kops_extra_args
        result = run_capturing_out(rolling_update_cmd, stderr_logger)
        proc_logger.info(result)
        if "Must specify --yes to rolling-update" in result:
            logger.info("kOps cluster rolling update would make changes")
            if cmd != "apply" or dry_run:
                logger.info("Skipping actual kOps cluster rolling update due to dry-run")
            else:
                logger.info("Running kOps cluster rolling update")
                run(rolling_update_cmd + ["--yes"],
                    stdout_logger,
                    stderr_logger).wait()

    def export(self):
        context = self.context
        run = context.app.run

        kops_extra_args = ["--name", context.kops.cluster_name, "--state", context.kops.state_store]
        logger.info("Exporting kubeconfig from kOps")
        run(self.kops_stanza + ["export", "kubecfg", "--admin"] + kops_extra_args,
            stdout_logger,
            stderr_logger).wait()

        if not self.kubeconfig_path.exists():
            raise RuntimeError("kOps failed to export kubeconfig for unknown reason - check AWS credentials")

    def __repr__(self):
        return "kOps 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	5✔
20	import logging	5✔
21	import os	5✔
22	from pathlib import Path	5✔
23	from tempfile import TemporaryDirectory	5✔
24
25	import yaml	5✔
26
27	from kubernator.api import (KubernatorPlugin, scan_dir,	5✔
28	FileType,
29	load_remote_file,
30	io_StringIO,
31	TemplateEngine,
32	StripNL,
33	Globs)
34	from kubernator.plugins.k8s_api import K8SResourcePluginMixin	5✔
35	from kubernator.proc import CalledProcessError	5✔
36
37	logger = logging.getLogger("kubernator.kops")	5✔
38	proc_logger = logger.getChild("kops")	5✔
39	stdout_logger = StripNL(proc_logger.info)	5✔
40	stderr_logger = StripNL(proc_logger.warning)	5✔
41
42	KOPS_CRDS = ["kops.k8s.io_clusters.yaml",	5✔
43	"kops.k8s.io_instancegroups.yaml",
44	"kops.k8s.io_keysets.yaml",
45	"kops.k8s.io_sshcredentials.yaml"]
46
47	OBJECT_SCHEMA_VERSION = "1.20.6"	5✔
48
49
50	class KopsPlugin(KubernatorPlugin, K8SResourcePluginMixin):	5✔
51	logger = logger	5✔
52
53	_name = "kops"	5✔
54
55	def __init__(self):	5✔
56	self.context = None	×
57	self.kops_stanza = ["kops"]	×
58	self.version = None	×
59	self.kops_dir = None	×
60	self.kops_path = None	×
61
62	self.template_engine = TemplateEngine(logger)	×
63	super().__init__()	×
64
65	def set_context(self, context):	5✔
66	self.context = context	×
67
68	def register(self, helm_version=None):	5✔
69	raise RuntimeError("Currently disabled, please don't use")	×
UNCOV 70	self.context.app.register_plugin("kubeconfig")
UNCOV 71	self.context.app.register_plugin("kubectl")
72
73	def handle_init(self):	5✔
74	context = self.context	×
75
76	# Temp dir
77	self.kops_dir = TemporaryDirectory()	×
78	context.app.register_cleanup(self.kops_dir)	×
79	self.kops_path = kops_path = Path(self.kops_dir.name)	×
80	kops_config = kops_path / ".kops.yaml"	×
81	kops_config.touch()	×
82	self.kops_stanza.extend(("--config", str(kops_config)))	×
83
84	# Kube config dir
85	kubeconfig_dir = kops_path / ".kube"	×
86	kubeconfig_dir.mkdir()	×
87	self.kubeconfig_path = kubeconfig_dir / "config"	×
88
89	# if logger.getEffectiveLevel() < logging.DEBUG:
90	# self.kops_stanza.append("-v5")
91	# elif logger.getEffectiveLevel() < logging.INFO:
92	# self.kops_stanza.append("-v2")
93
94	context.globals.kops = dict(walk_remote=self.walk_remote,	×
95	walk_local=self.walk_local,
96	update=self.update,
97	export=self.export,
98	master_interval="8m",
99	node_interval="8m"
100	)
101	context.kops = dict()	×
102
103	self.version = context.app.run_capturing_out(["kops", "version", "--short"], stderr_logger).strip()	×
104	logger.info("Found kOps version %s", self.version)	×
105
106	self.resource_definitions_schema = load_remote_file(logger,	×
107	f"https://raw.githubusercontent.com/kubernetes/kubernetes/"
108	f"v{OBJECT_SCHEMA_VERSION}/api/openapi-spec/swagger.json",
109	FileType.JSON)
110	self._populate_resource_definitions()	×
111
112	common_url_path = f"https://raw.githubusercontent.com/kubernetes/kops/v{self.version}/k8s/crds"	×
113	for kops_crd in KOPS_CRDS:	×
114	url = f"{common_url_path}/{kops_crd}"	×
115	self.add_remote_crds(url, FileType.YAML, sub_category="kops")	×
116
117	def handle_start(self):	5✔
118	context = self.context	×
119
120	# Exclude Kops YAMLs from K8S resource loading
121	context.k8s.default_excludes.add("*.kops.yaml")	×
122	context.k8s.default_excludes.add("*.kops.yml")	×
123
124	def walk_remote(self, url, *paths: Path,	5✔
125	excludes=(".",), includes=(".kops.yaml", "*.kops.yml")):
126	repository = self.context.app.repository(url)	×
127	for path in paths:	×
128	self.walk_local(repository.local_dir / path, excludes, includes)	×
129
130	def walk_local(self, path: Path, excludes=(".",), includes=(".kops.yaml", "*.kops.yml")):	5✔
131	context = self.context	×
132	path = Path(path)	×
133
134	for f in scan_dir(logger, path, lambda d: d.is_file(), Globs(excludes), Globs(includes)):	×
135	p = path / f.name	×
136	display_p = context.app.display_path(p)	×
137	logger.debug("Adding Kops resources from %s", display_p)	×
138
139	with open(p, "rt") as file:	×
140	template = self.template_engine.from_string(file.read())	×
141
142	self.add_resources(template.render({"ktor": context}), display_p)	×
143
144	def update(self):	5✔
145	context = self.context	×
146	run = context.app.run	×
147	run_capturing_out = context.app.run_capturing_out	×
148
149	os.environ["KUBECONFIG"] = str(self.kubeconfig_path)	×
150	os.environ["KOPS_CLUSTER_NAME"] = context.kops.cluster_name	×
151	os.environ["KOPS_STATE_STORE"] = context.kops.state_store	×
152
153	kops_extra_args = ["--name", context.kops.cluster_name, "--state", context.kops.state_store]	×
154
155	cmd = context.app.args.command	×
156	dry_run = context.app.args.dry_run	×
157
158	for resource in self.resources.values():	×
159	logger.info("Replacing/creating kOps resource %s", resource)	×
160	resource_out = io_StringIO()	×
161	yaml.dump(resource.manifest, resource_out)	×
162	if cmd != "apply" or dry_run:	×
163	logger.info("Would replace kOps resource if not for dry-run mode: %s", resource_out.getvalue())	×
164	else:
165	run(self.kops_stanza + ["replace", "--force", "-f", "-"] + kops_extra_args,	×
166	stdout_logger,
167	stderr_logger,
168	resource_out.getvalue()).wait()
169
170	logger.info("Staging kOps update")	×
171	update_cmd = self.kops_stanza + ["update", "cluster"] + kops_extra_args	×
172	result = run_capturing_out(update_cmd, stderr_logger)	×
173	proc_logger.info(result)	×
174	if "Must specify --yes to apply changes" in result:	×
175	logger.info("kOps update would make changes")	×
176	if cmd != "apply" or dry_run:	×
177	logger.info("Skipping actual kOps update due to dry-run mode")	×
178	else:
179	logger.info("Running kOps update")	×
180	run(update_cmd + ["--yes"],	×
181	stdout_logger,
182	stderr_logger).wait()
183
184	self.export()	×
185
186	if cmd != "apply":	×
187	logger.info("Skipping cluster validation since not in apply mode")	×
188	else:
189	validation_failed = False	×
190	try:	×
191	output = run_capturing_out(self.kops_stanza + ["validate", "cluster", "-o", "json"],	×
192	stderr_logger)
193	except CalledProcessError as e:	×
194	validation_failed = True	×
195	output = e.output	×
196
197	if validation_failed:	×
198	if output:	×
199	validation_results = json.loads(output)	×
200	for failure in validation_results["failures"]:	×
201	logger.error("%s %s failed validation: %s", failure["type"], failure["name"],	×
202	failure["message"])
203	raise RuntimeError("Cluster validation failed!")	×
204	else:
205	logger.info("Cluster validation successful!")	×
206
207	logger.info("Staging kOps cluster rolling update")	×
208	rolling_update_cmd = self.kops_stanza + ["rolling-update", "cluster",	×
209	"--master-interval", context.kops.master_interval,
210	"--node-interval", context.kops.node_interval] + kops_extra_args
211	result = run_capturing_out(rolling_update_cmd, stderr_logger)	×
212	proc_logger.info(result)	×
213	if "Must specify --yes to rolling-update" in result:	×
214	logger.info("kOps cluster rolling update would make changes")	×
215	if cmd != "apply" or dry_run:	×
216	logger.info("Skipping actual kOps cluster rolling update due to dry-run")	×
217	else:
218	logger.info("Running kOps cluster rolling update")	×
219	run(rolling_update_cmd + ["--yes"],	×
220	stdout_logger,
221	stderr_logger).wait()
222
223	def export(self):	5✔
224	context = self.context	×
225	run = context.app.run	×
226
227	kops_extra_args = ["--name", context.kops.cluster_name, "--state", context.kops.state_store]	×
228	logger.info("Exporting kubeconfig from kOps")	×
229	run(self.kops_stanza + ["export", "kubecfg", "--admin"] + kops_extra_args,	×
230	stdout_logger,
231	stderr_logger).wait()
232
233	if not self.kubeconfig_path.exists():	×
234	raise RuntimeError("kOps failed to export kubeconfig for unknown reason - check AWS credentials")	×
235
236	def __repr__(self):	5✔
237	return "kOps Plugin"	×

karellen / kubernator / 15104716107

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