24362925690

Committed 13 Apr 2026 07:35PM UTC coverage: 76.88%. First build

Build # 24362925690

Build Type

Pull #100

github

Committed by

web-flow

Commit Message

Merge 8e96c3f7a into 9b2b0c4e8

Pull Request Pull Request #100: Add kind plugin; migrate internal integration tests from minikube

Coverage Stats

539 of 890 branches covered (60.56%)

Branch coverage included in aggregate %.

231 of 264 new or added lines in 1 file covered. (87.5%)

2763 of 3405 relevant lines covered (81.15%)

4.06 hits per line

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

82.37

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

# -*- coding: utf-8 -*-
#
#   Copyright 2020 Express Systems USA, Inc
#   Copyright 2023 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 http.client
import logging
import os
import socket
import ssl
import tempfile
import time
from pathlib import Path
from urllib.parse import urlparse

import yaml

from kubernator.api import (KubernatorPlugin,
                            StripNL,
                            get_golang_os,
                            get_golang_machine,
                            prepend_os_path,
                            get_cache_dir,
                            CalledProcessError,
                            )

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

KIND_CLUSTER_LABEL = "io.x-k8s.kind.cluster"

DEFAULT_NODE_IMAGE_REGISTRY = "ghcr.io/karellen/kindest-node"


class KindPlugin(KubernatorPlugin):
    logger = logger

    _name = "kind"

    def __init__(self):
        self.context = None
        self.kind_dir = None
        self.kubeconfig_dir = None
        self._config_path = None
        super().__init__()

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

    def _resolve_latest_tag(self, repo_url):
        """Return highest v<major>.<minor>.<patch> tag from a GitHub repo via git ls-remote.
        Skips pre-release tags like v0.1.0-alpha, v0.2.0-rc.1."""
        versions = self.context.app.run_capturing_out(
            ["git", "ls-remote", "-t", "--refs", repo_url, "v*"],
            stderr_logger,
        )
        tuples = []
        for line in versions.splitlines():
            if not line:
                continue
            # "<sha>\trefs/tags/v1.2.3" -> "1.2.3"
            tag = line.split()[1][11:]
            parts = tag.split(".")
            if len(parts) != 3:
                continue
            if not all(p.isdigit() for p in parts):
                continue
            tuples.append(tuple(int(p) for p in parts))
        if not tuples:
            raise RuntimeError(f"No numeric v<major>.<minor>.<patch> tags found at {repo_url}")
        return ".".join(str(x) for x in sorted(tuples)[-1])

    def get_latest_kind_version(self):
        return self._resolve_latest_tag("https://github.com/kubernetes-sigs/kind")

    def cmd(self, *extra_args):
        stanza, env = self._stanza(list(extra_args))
        return self.context.app.run(stanza, stdout_logger, stderr_logger, env=env).wait()

    def cmd_out(self, *extra_args):
        stanza, env = self._stanza(list(extra_args))
        return self.context.app.run_capturing_out(stanza, stderr_logger, env=env)

    def _stanza(self, extra_args):
        context = self.context
        kind = context.kind
        stanza = [kind.kind_file] + extra_args
        env = dict(os.environ)
        env["KUBECONFIG"] = str(kind.kubeconfig)
        if kind.provider == "podman":
            env["KIND_EXPERIMENTAL_PROVIDER"] = "podman"
        return stanza, env

    def _docker_ps(self, *filters, all_containers=True):
        args = ["docker", "ps", "-q"]
        if all_containers:
            args.append("-a")
        for f in filters:
            args += ["--filter", f]
        out = self.context.app.run_capturing_out(args, stderr_logger).strip()
        return [line for line in out.splitlines() if line]

    def _cluster_containers(self, profile, running=None, all_containers=True):
        filters = [f"label={KIND_CLUSTER_LABEL}={profile}"]
        if running is True:
            filters.append("status=running")
        elif running is False:
            filters.append("status=exited")
        return self._docker_ps(*filters, all_containers=all_containers)

    def _cluster_exists(self, profile):
        out = self.cmd_out("get", "clusters")
        return profile in {line.strip() for line in out.splitlines() if line.strip()}

    def _detect_provider(self, provider):
        context = self.context
        cmd_debug_logger = StripNL(proc_logger.debug)

        def probe(binary):
            try:
                context.app.run([binary, "info"], cmd_debug_logger, cmd_debug_logger).wait()
                return True
            except (FileNotFoundError, CalledProcessError) as e:
                logger.trace("%s is NOT functional", binary, exc_info=e)
                return False

        if provider:
            if not probe(provider):
                raise RuntimeError(f"Requested kind provider {provider!r} is not functional")
            return provider

        if probe("docker"):
            logger.info("Docker is functional, selecting 'docker' as the kind provider")
            return "docker"
        if probe("podman"):
            logger.info("Podman is functional, selecting 'podman' as the kind provider")
            return "podman"
        raise RuntimeError("No kind provider is functional. Tried 'docker' and 'podman'.")

    def register(self,
                 kind_version=None,
                 profile="default",
                 k8s_version=None,
                 node_image=None,
                 node_image_registry=DEFAULT_NODE_IMAGE_REGISTRY,
                 keep_running=False,
                 start_fresh=False,
                 nodes=1,
                 control_plane_nodes=1,
                 provider=None,
                 config=None,
                 extra_port_mappings=None,
                 feature_gates=None,
                 runtime_config=None):
        context = self.context

        context.app.register_plugin("kubeconfig")

        if not k8s_version and not node_image:
            msg = "Either k8s_version or node_image must be specified for kind"
            logger.critical(msg)
            raise RuntimeError(msg)

        if nodes < 1:
            raise RuntimeError(f"kind requires nodes >= 1, got {nodes}")
        if control_plane_nodes < 1:
            raise RuntimeError(f"kind requires control_plane_nodes >= 1, got {control_plane_nodes}")
        if control_plane_nodes > nodes:
            raise RuntimeError(
                f"control_plane_nodes ({control_plane_nodes}) cannot exceed nodes ({nodes})")

        k8s_version_tuple = tuple(map(int, k8s_version.split("."))) if k8s_version else None

        if not kind_version:
            kind_version = self.get_latest_kind_version()
            logger.info("No kind version is specified, latest is %s", kind_version)

        kind_dl_file, _ = context.app.download_remote_file(
            logger,
            f"https://github.com/kubernetes-sigs/kind/releases/download/v{kind_version}/"
            f"kind-{get_golang_os()}-{get_golang_machine()}",
            "bin",
        )
        os.chmod(kind_dl_file, 0o500)

        self.kind_dir = tempfile.TemporaryDirectory()
        context.app.register_cleanup(self.kind_dir)
        kind_file = Path(self.kind_dir.name) / "kind"
        kind_file.symlink_to(kind_dl_file)
        prepend_os_path(self.kind_dir.name)

        version_out: str = context.app.run_capturing_out(
            [str(kind_file), "version"], stderr_logger).strip()
        # "kind v0.31.0 go1.22.1 linux/amd64"
        version = version_out.split()[1].lstrip("v") if version_out else kind_version
        logger.info("Found kind %s in %s", version, kind_file)

        profile_dir = get_cache_dir("kind")
        self.kubeconfig_dir = profile_dir / ".kube" / profile
        self.kubeconfig_dir.mkdir(parents=True, exist_ok=True)
        kubeconfig_path = self.kubeconfig_dir / "config"

        resolved_provider = self._detect_provider(provider)

        if not node_image and k8s_version:
            node_image = f"{node_image_registry}:v{k8s_version}"

        context.globals.kind = dict(
            version=version,
            kind_file=str(kind_file),
            profile=profile,
            k8s_version=k8s_version,
            k8s_version_tuple=k8s_version_tuple,
            node_image=node_image,
            node_image_registry=node_image_registry,
            start_fresh=start_fresh,
            keep_running=keep_running,
            nodes=nodes,
            control_plane_nodes=control_plane_nodes,
            provider=resolved_provider,
            config=config,
            extra_port_mappings=list(extra_port_mappings) if extra_port_mappings else [],
            feature_gates=dict(feature_gates) if feature_gates else {},
            runtime_config=dict(runtime_config) if runtime_config else {},
            kubeconfig=str(kubeconfig_path),
            cmd=self.cmd,
            cmd_out=self.cmd_out,
        )
        context.kubeconfig.kubeconfig = context.kind.kubeconfig

        logger.info("Kind kubeconfig is %s", context.kind.kubeconfig)
        logger.info("Kind node image is %s", node_image)

    def _generate_cluster_config(self):
        kind = self.context.kind
        if kind.config:
            return kind.config

        needs_config = (kind.nodes > 1
                        or kind.control_plane_nodes > 1
                        or kind.extra_port_mappings
                        or kind.feature_gates
                        or kind.runtime_config)
        if not needs_config:
            return None

        doc = {
            "kind": "Cluster",
            "apiVersion": "kind.x-k8s.io/v1alpha4",
        }
        if kind.feature_gates:
            doc["featureGates"] = {k: bool(v) for k, v in kind.feature_gates.items()}
        if kind.runtime_config:
            doc["runtimeConfig"] = {k: str(v) for k, v in kind.runtime_config.items()}

        node_list = []
        for i in range(kind.control_plane_nodes):
            entry = {"role": "control-plane"}
            if i == 0 and kind.extra_port_mappings:
                entry["extraPortMappings"] = [dict(m) for m in kind.extra_port_mappings]
            node_list.append(entry)
        for _ in range(kind.nodes - kind.control_plane_nodes):
            node_list.append({"role": "worker"})
        doc["nodes"] = node_list

        return yaml.safe_dump(doc, sort_keys=False)

    def _write_cluster_config(self):
        config_yaml = self._generate_cluster_config()
        if not config_yaml:
            self._config_path = None
            return None
        self._config_path = Path(self.kind_dir.name) / "cluster.yaml"
        self._config_path.write_text(config_yaml)
        logger.debug("Wrote kind cluster config to %s:\n%s", self._config_path, config_yaml)
        return self._config_path

    def _export_kubeconfig(self):
        kind = self.context.kind
        self.cmd("export", "kubeconfig",
                 "--name", kind.profile,
                 "--kubeconfig", str(kind.kubeconfig))
        logger.info("Exported kubeconfig for cluster %r to %s", kind.profile, kind.kubeconfig)

    def _docker(self, *args, capture=False):
        cmd_args = ["docker", *args]
        if capture:
            return self.context.app.run_capturing_out(cmd_args, stderr_logger)
        return self.context.app.run(cmd_args, stdout_logger, stderr_logger).wait()

    def kind_create(self):
        kind = self.context.kind
        args = ["create", "cluster", "--name", kind.profile, "--wait", "120s"]
        if kind.node_image:
            args += ["--image", kind.node_image]
        config_path = self._write_cluster_config()
        if config_path:
            args += ["--config", str(config_path)]
        logger.info("Creating kind cluster %r (image=%s, nodes=%d, control_plane_nodes=%d)",
                    kind.profile, kind.node_image, kind.nodes, kind.control_plane_nodes)
        self.cmd(*args)

    def kind_delete(self):
        kind = self.context.kind
        logger.warning("Deleting kind cluster %r", kind.profile)
        try:
            self.cmd("delete", "cluster", "--name", kind.profile)
        except CalledProcessError as e:
            logger.warning("kind delete failed for %r: %s", kind.profile, e)

    def kind_stop(self):
        kind = self.context.kind
        running = self._cluster_containers(kind.profile, running=True)
        if not running:
            logger.info("Kind cluster %r has no running containers", kind.profile)
            return
        logger.info("Stopping %d container(s) for kind cluster %r", len(running), kind.profile)
        self._docker("stop", *running)

    def kind_start(self):
        kind = self.context.kind
        resumed = False
        if self._cluster_exists(kind.profile):
            stopped = self._cluster_containers(kind.profile, running=False)
            running = self._cluster_containers(kind.profile, running=True)
            if stopped and not running:
                logger.info("Starting %d stopped container(s) for kind cluster %r",
                            len(stopped), kind.profile)
                self._docker("start", *stopped)
                resumed = True
            elif stopped and running:
                logger.info("Resuming %d stopped container(s) for kind cluster %r",
                            len(stopped), kind.profile)
                self._docker("start", *stopped)
                resumed = True
            else:
                logger.info("Kind cluster %r is already running", kind.profile)
        else:
            self.kind_create()

        self._export_kubeconfig()

        # On resume, kind's --wait flag did not run; static pods take a few
        # seconds to become reachable. Poll /readyz until the apiserver answers
        # so downstream plugins don't see SSL handshake failures.
        if resumed:
            self._wait_for_apiserver()

        context = self.context
        context.app.register_plugin("kubectl", version=kind.k8s_version)

    def _wait_for_apiserver(self, timeout=120):
        with open(self.context.kind.kubeconfig) as f:
            cfg = yaml.safe_load(f)
        server_url = cfg["clusters"][0]["cluster"]["server"]
        parsed = urlparse(server_url)
        ssl_ctx = ssl._create_unverified_context()

        logger.info("Waiting up to %ds for apiserver at %s to be ready",
                    timeout, server_url)
        deadline = time.monotonic() + timeout
        last_err = None
        while time.monotonic() < deadline:
            try:
                conn = http.client.HTTPSConnection(parsed.hostname, parsed.port,
                                                   context=ssl_ctx, timeout=5)
                try:
                    conn.request("GET", "/readyz")
                    resp = conn.getresponse()
                    status = resp.status
                    resp.read()
                finally:
                    conn.close()
                if status == 200:
                    logger.info("Apiserver at %s is ready", server_url)
                    return
                last_err = f"HTTP {status}"
            except (OSError, socket.error, ssl.SSLError,
                    http.client.HTTPException) as e:
                last_err = f"{type(e).__name__}: {e}"
            time.sleep(2)
        raise RuntimeError(
            f"Apiserver at {server_url} did not become ready within {timeout}s: {last_err}")

    def handle_start(self):
        kind = self.context.kind
        if kind.start_fresh:
            if self._cluster_exists(kind.profile):
                self.kind_delete()
        self.kind_start()

    def handle_shutdown(self):
        kind = self.context.kind
        if kind.keep_running:
            logger.warning("Keeping kind cluster %r running", kind.profile)
            return
        self.kind_stop()

    def __repr__(self):
        return "Kind Plugin"

1	# -- coding: utf-8 --
2	#
3	# Copyright 2020 Express Systems USA, Inc
4	# Copyright 2023 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	import http.client	5✔
19	import logging	5✔
20	import os	5✔
21	import socket	5✔
22	import ssl	5✔
23	import tempfile	5✔
24	import time	5✔
25	from pathlib import Path	5✔
26	from urllib.parse import urlparse	5✔
27
28	import yaml	5✔
29
30	from kubernator.api import (KubernatorPlugin,	5✔
31	StripNL,
32	get_golang_os,
33	get_golang_machine,
34	prepend_os_path,
35	get_cache_dir,
36	CalledProcessError,
37	)
38
39	logger = logging.getLogger("kubernator.kind")	5✔
40	proc_logger = logger.getChild("proc")	5✔
41	stdout_logger = StripNL(proc_logger.info)	5✔
42	stderr_logger = StripNL(proc_logger.warning)	5✔
43
44	KIND_CLUSTER_LABEL = "io.x-k8s.kind.cluster"	5✔
45
46	DEFAULT_NODE_IMAGE_REGISTRY = "ghcr.io/karellen/kindest-node"	5✔
47
48
49	class KindPlugin(KubernatorPlugin):	5✔
50	logger = logger	5✔
51
52	_name = "kind"	5✔
53
54	def __init__(self):	5✔
55	self.context = None	5✔
56	self.kind_dir = None	5✔
57	self.kubeconfig_dir = None	5✔
58	self._config_path = None	5✔
59	super().__init__()	5✔
60
61	def set_context(self, context):	5✔
62	self.context = context	5✔
63
64	def _resolve_latest_tag(self, repo_url):	5✔
65	"""Return highest v<major>.<minor>.<patch> tag from a GitHub repo via git ls-remote.
66	Skips pre-release tags like v0.1.0-alpha, v0.2.0-rc.1."""
67	versions = self.context.app.run_capturing_out(	5✔
68	["git", "ls-remote", "-t", "--refs", repo_url, "v*"],
69	stderr_logger,
70	)
71	tuples = []	5✔
72	for line in versions.splitlines():	5✔
73	if not line:	5!
NEW 74	continue	×
75	# "<sha>\trefs/tags/v1.2.3" -> "1.2.3"
76	tag = line.split()[1][11:]	5✔
77	parts = tag.split(".")	5✔
78	if len(parts) != 3:	5!
NEW 79	continue	×
80	if not all(p.isdigit() for p in parts):	5✔
81	continue	5✔
82	tuples.append(tuple(int(p) for p in parts))	5✔
83	if not tuples:	5!
NEW 84	raise RuntimeError(f"No numeric v<major>.<minor>.<patch> tags found at {repo_url}")	×
85	return ".".join(str(x) for x in sorted(tuples)[-1])	5✔
86
87	def get_latest_kind_version(self):	5✔
88	return self._resolve_latest_tag("https://github.com/kubernetes-sigs/kind")	5✔
89
90	def cmd(self, *extra_args):	5✔
91	stanza, env = self._stanza(list(extra_args))	5✔
92	return self.context.app.run(stanza, stdout_logger, stderr_logger, env=env).wait()	5✔
93
94	def cmd_out(self, *extra_args):	5✔
95	stanza, env = self._stanza(list(extra_args))	5✔
96	return self.context.app.run_capturing_out(stanza, stderr_logger, env=env)	5✔
97
98	def _stanza(self, extra_args):	5✔
99	context = self.context	5✔
100	kind = context.kind	5✔
101	stanza = [kind.kind_file] + extra_args	5✔
102	env = dict(os.environ)	5✔
103	env["KUBECONFIG"] = str(kind.kubeconfig)	5✔
104	if kind.provider == "podman":	5!
NEW 105	env["KIND_EXPERIMENTAL_PROVIDER"] = "podman"	×
106	return stanza, env	5✔
107
108	def _docker_ps(self, *filters, all_containers=True):	5✔
109	args = ["docker", "ps", "-q"]	5✔
110	if all_containers:	5!
111	args.append("-a")	5✔
112	for f in filters:	5✔
113	args += ["--filter", f]	5✔
114	out = self.context.app.run_capturing_out(args, stderr_logger).strip()	5✔
115	return [line for line in out.splitlines() if line]	5✔
116
117	def _cluster_containers(self, profile, running=None, all_containers=True):	5✔
118	filters = [f"label={KIND_CLUSTER_LABEL}={profile}"]	5✔
119	if running is True:	5✔
120	filters.append("status=running")	5✔
121	elif running is False:	5!
122	filters.append("status=exited")	5✔
123	return self._docker_ps(*filters, all_containers=all_containers)	5✔
124
125	def _cluster_exists(self, profile):	5✔
126	out = self.cmd_out("get", "clusters")	5✔
127	return profile in {line.strip() for line in out.splitlines() if line.strip()}	5✔
128
129	def _detect_provider(self, provider):	5✔
130	context = self.context	5✔
131	cmd_debug_logger = StripNL(proc_logger.debug)	5✔
132
133	def probe(binary):	5✔
134	try:	5✔
135	context.app.run([binary, "info"], cmd_debug_logger, cmd_debug_logger).wait()	5✔
136	return True	5✔
NEW 137	except (FileNotFoundError, CalledProcessError) as e:	×
NEW 138	logger.trace("%s is NOT functional", binary, exc_info=e)	×
NEW 139	return False	×
140
141	if provider:	5!
NEW 142	if not probe(provider):	×
NEW 143	raise RuntimeError(f"Requested kind provider {provider!r} is not functional")	×
NEW 144	return provider	×
145
146	if probe("docker"):	5!
147	logger.info("Docker is functional, selecting 'docker' as the kind provider")	5✔
148	return "docker"	5✔
NEW 149	if probe("podman"):	×
NEW 150	logger.info("Podman is functional, selecting 'podman' as the kind provider")	×
NEW 151	return "podman"	×
NEW 152	raise RuntimeError("No kind provider is functional. Tried 'docker' and 'podman'.")	×
153
154	def register(self,	5✔
155	kind_version=None,
156	profile="default",
157	k8s_version=None,
158	node_image=None,
159	node_image_registry=DEFAULT_NODE_IMAGE_REGISTRY,
160	keep_running=False,
161	start_fresh=False,
162	nodes=1,
163	control_plane_nodes=1,
164	provider=None,
165	config=None,
166	extra_port_mappings=None,
167	feature_gates=None,
168	runtime_config=None):
169	context = self.context	5✔
170
171	context.app.register_plugin("kubeconfig")	5✔
172
173	if not k8s_version and not node_image:	5!
NEW 174	msg = "Either k8s_version or node_image must be specified for kind"	×
NEW 175	logger.critical(msg)	×
NEW 176	raise RuntimeError(msg)	×
177
178	if nodes < 1:	5!
NEW 179	raise RuntimeError(f"kind requires nodes >= 1, got {nodes}")	×
180	if control_plane_nodes < 1:	5!
NEW 181	raise RuntimeError(f"kind requires control_plane_nodes >= 1, got {control_plane_nodes}")	×
182	if control_plane_nodes > nodes:	5!
NEW 183	raise RuntimeError(	×
184	f"control_plane_nodes ({control_plane_nodes}) cannot exceed nodes ({nodes})")
185
186	k8s_version_tuple = tuple(map(int, k8s_version.split("."))) if k8s_version else None	5✔
187
188	if not kind_version:	5!
189	kind_version = self.get_latest_kind_version()	5✔
190	logger.info("No kind version is specified, latest is %s", kind_version)	5✔
191
192	kind_dl_file, _ = context.app.download_remote_file(	5✔
193	logger,
194	f"https://github.com/kubernetes-sigs/kind/releases/download/v{kind_version}/"
195	f"kind-{get_golang_os()}-{get_golang_machine()}",
196	"bin",
197	)
198	os.chmod(kind_dl_file, 0o500)	5✔
199
200	self.kind_dir = tempfile.TemporaryDirectory()	5✔
201	context.app.register_cleanup(self.kind_dir)	5✔
202	kind_file = Path(self.kind_dir.name) / "kind"	5✔
203	kind_file.symlink_to(kind_dl_file)	5✔
204	prepend_os_path(self.kind_dir.name)	5✔
205
206	version_out: str = context.app.run_capturing_out(	5✔
207	[str(kind_file), "version"], stderr_logger).strip()
208	# "kind v0.31.0 go1.22.1 linux/amd64"
209	version = version_out.split()[1].lstrip("v") if version_out else kind_version	5✔
210	logger.info("Found kind %s in %s", version, kind_file)	5✔
211
212	profile_dir = get_cache_dir("kind")	5✔
213	self.kubeconfig_dir = profile_dir / ".kube" / profile	5✔
214	self.kubeconfig_dir.mkdir(parents=True, exist_ok=True)	5✔
215	kubeconfig_path = self.kubeconfig_dir / "config"	5✔
216
217	resolved_provider = self._detect_provider(provider)	5✔
218
219	if not node_image and k8s_version:	5!
220	node_image = f"{node_image_registry}:v{k8s_version}"	5✔
221
222	context.globals.kind = dict(	5✔
223	version=version,
224	kind_file=str(kind_file),
225	profile=profile,
226	k8s_version=k8s_version,
227	k8s_version_tuple=k8s_version_tuple,
228	node_image=node_image,
229	node_image_registry=node_image_registry,
230	start_fresh=start_fresh,
231	keep_running=keep_running,
232	nodes=nodes,
233	control_plane_nodes=control_plane_nodes,
234	provider=resolved_provider,
235	config=config,
236	extra_port_mappings=list(extra_port_mappings) if extra_port_mappings else [],
237	feature_gates=dict(feature_gates) if feature_gates else {},
238	runtime_config=dict(runtime_config) if runtime_config else {},
239	kubeconfig=str(kubeconfig_path),
240	cmd=self.cmd,
241	cmd_out=self.cmd_out,
242	)
243	context.kubeconfig.kubeconfig = context.kind.kubeconfig	5✔
244
245	logger.info("Kind kubeconfig is %s", context.kind.kubeconfig)	5✔
246	logger.info("Kind node image is %s", node_image)	5✔
247
248	def _generate_cluster_config(self):	5✔
249	kind = self.context.kind	5✔
250	if kind.config:	5!
NEW 251	return kind.config	×
252
253	needs_config = (kind.nodes > 1	5✔
254	or kind.control_plane_nodes > 1
255	or kind.extra_port_mappings
256	or kind.feature_gates
257	or kind.runtime_config)
258	if not needs_config:	5✔
259	return None	5✔
260
261	doc = {	5✔
262	"kind": "Cluster",
263	"apiVersion": "kind.x-k8s.io/v1alpha4",
264	}
265	if kind.feature_gates:	5!
NEW 266	doc["featureGates"] = {k: bool(v) for k, v in kind.feature_gates.items()}	×
267	if kind.runtime_config:	5!
NEW 268	doc["runtimeConfig"] = {k: str(v) for k, v in kind.runtime_config.items()}	×
269
270	node_list = []	5✔
271	for i in range(kind.control_plane_nodes):	5✔
272	entry = {"role": "control-plane"}	5✔
273	if i == 0 and kind.extra_port_mappings:	5!
NEW 274	entry["extraPortMappings"] = [dict(m) for m in kind.extra_port_mappings]	×
275	node_list.append(entry)	5✔
276	for _ in range(kind.nodes - kind.control_plane_nodes):	5✔
277	node_list.append({"role": "worker"})	5✔
278	doc["nodes"] = node_list	5✔
279
280	return yaml.safe_dump(doc, sort_keys=False)	5✔
281
282	def _write_cluster_config(self):	5✔
283	config_yaml = self._generate_cluster_config()	5✔
284	if not config_yaml:	5✔
285	self._config_path = None	5✔
286	return None	5✔
287	self._config_path = Path(self.kind_dir.name) / "cluster.yaml"	5✔
288	self._config_path.write_text(config_yaml)	5✔
289	logger.debug("Wrote kind cluster config to %s:\n%s", self._config_path, config_yaml)	5✔
290	return self._config_path	5✔
291
292	def _export_kubeconfig(self):	5✔
293	kind = self.context.kind	5✔
294	self.cmd("export", "kubeconfig",	5✔
295	"--name", kind.profile,
296	"--kubeconfig", str(kind.kubeconfig))
297	logger.info("Exported kubeconfig for cluster %r to %s", kind.profile, kind.kubeconfig)	5✔
298
299	def _docker(self, *args, capture=False):	5✔
300	cmd_args = ["docker", *args]	5✔
301	if capture:	5!
NEW 302	return self.context.app.run_capturing_out(cmd_args, stderr_logger)	×
303	return self.context.app.run(cmd_args, stdout_logger, stderr_logger).wait()	5✔
304
305	def kind_create(self):	5✔
306	kind = self.context.kind	5✔
307	args = ["create", "cluster", "--name", kind.profile, "--wait", "120s"]	5✔
308	if kind.node_image:	5!
309	args += ["--image", kind.node_image]	5✔
310	config_path = self._write_cluster_config()	5✔
311	if config_path:	5✔
312	args += ["--config", str(config_path)]	5✔
313	logger.info("Creating kind cluster %r (image=%s, nodes=%d, control_plane_nodes=%d)",	5✔
314	kind.profile, kind.node_image, kind.nodes, kind.control_plane_nodes)
315	self.cmd(*args)	5✔
316
317	def kind_delete(self):	5✔
318	kind = self.context.kind	5✔
319	logger.warning("Deleting kind cluster %r", kind.profile)	5✔
320	try:	5✔
321	self.cmd("delete", "cluster", "--name", kind.profile)	5✔
NEW 322	except CalledProcessError as e:	×
NEW 323	logger.warning("kind delete failed for %r: %s", kind.profile, e)	×
324
325	def kind_stop(self):	5✔
326	kind = self.context.kind	5✔
327	running = self._cluster_containers(kind.profile, running=True)	5✔
328	if not running:	5!
NEW 329	logger.info("Kind cluster %r has no running containers", kind.profile)	×
NEW 330	return	×
331	logger.info("Stopping %d container(s) for kind cluster %r", len(running), kind.profile)	5✔
332	self._docker("stop", *running)	5✔
333
334	def kind_start(self):	5✔
335	kind = self.context.kind	5✔
336	resumed = False	5✔
337	if self._cluster_exists(kind.profile):	5✔
338	stopped = self._cluster_containers(kind.profile, running=False)	5✔
339	running = self._cluster_containers(kind.profile, running=True)	5✔
340	if stopped and not running:	5✔
341	logger.info("Starting %d stopped container(s) for kind cluster %r",	5✔
342	len(stopped), kind.profile)
343	self._docker("start", *stopped)	5✔
344	resumed = True	5✔
345	elif stopped and running:	5!
NEW 346	logger.info("Resuming %d stopped container(s) for kind cluster %r",	×
347	len(stopped), kind.profile)
NEW 348	self._docker("start", *stopped)	×
NEW 349	resumed = True	×
350	else:
351	logger.info("Kind cluster %r is already running", kind.profile)	5✔
352	else:
353	self.kind_create()	5✔
354
355	self._export_kubeconfig()	5✔
356
357	# On resume, kind's --wait flag did not run; static pods take a few
358	# seconds to become reachable. Poll /readyz until the apiserver answers
359	# so downstream plugins don't see SSL handshake failures.
360	if resumed:	5✔
361	self._wait_for_apiserver()	5✔
362
363	context = self.context	5✔
364	context.app.register_plugin("kubectl", version=kind.k8s_version)	5✔
365
366	def _wait_for_apiserver(self, timeout=120):	5✔
367	with open(self.context.kind.kubeconfig) as f:	5✔
368	cfg = yaml.safe_load(f)	5✔
369	server_url = cfg["clusters"][0]["cluster"]["server"]	5✔
370	parsed = urlparse(server_url)	5✔
371	ssl_ctx = ssl._create_unverified_context()	5✔
372
373	logger.info("Waiting up to %ds for apiserver at %s to be ready",	5✔
374	timeout, server_url)
375	deadline = time.monotonic() + timeout	5✔
376	last_err = None	5✔
377	while time.monotonic() < deadline:	5!
378	try:	5✔
379	conn = http.client.HTTPSConnection(parsed.hostname, parsed.port,	5✔
380	context=ssl_ctx, timeout=5)
381	try:	5✔
382	conn.request("GET", "/readyz")	5✔
383	resp = conn.getresponse()	5✔
384	status = resp.status	5✔
385	resp.read()	5✔
386	finally:
387	conn.close()	5✔
388	if status == 200:	5✔
389	logger.info("Apiserver at %s is ready", server_url)	5✔
390	return	5✔
391	last_err = f"HTTP {status}"	5✔
392	except (OSError, socket.error, ssl.SSLError,	5✔
393	http.client.HTTPException) as e:
394	last_err = f"{type(e).__name__}: {e}"	5✔
395	time.sleep(2)	5✔
NEW 396	raise RuntimeError(	×
397	f"Apiserver at {server_url} did not become ready within {timeout}s: {last_err}")
398
399	def handle_start(self):	5✔
400	kind = self.context.kind	5✔
401	if kind.start_fresh:	5✔
402	if self._cluster_exists(kind.profile):	5!
403	self.kind_delete()	5✔
404	self.kind_start()	5✔
405
406	def handle_shutdown(self):	5✔
407	kind = self.context.kind	5✔
408	if kind.keep_running:	5✔
409	logger.warning("Keeping kind cluster %r running", kind.profile)	5✔
410	return	5✔
411	self.kind_stop()	5✔
412
413	def __repr__(self):	5✔
414	return "Kind Plugin"	5✔

karellen / kubernator / 24362925690

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