11758973854

Committed 09 Nov 2024 07:45PM UTC coverage: 59.786% (-0.3%) from 60.037%

Build # 11758973854

Build Type

Pull #281

github

Committed by

web-flow

Commit Message

Merge 45d0a8c13 into e92f56ef4

Pull Request Pull Request #281: [BV-599] Kubeconfig helper

Run Details

190 of 458 branches covered (41.48%)

Branch coverage included in aggregate %.

42 of 76 new or added lines in 4 files covered. (55.26%)

61 existing lines in 5 files now uncovered.

2498 of 4038 relevant lines covered (61.86%)

0.62 hits per line

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

73.03

/leverage/containers/kubectl.py

import os
from dataclasses import dataclass
from pathlib import Path

from click.exceptions import Exit
from docker.types import Mount
import ruamel.yaml
from simple_term_menu import TerminalMenu

from leverage import logger
from leverage._utils import AwsCredsEntryPoint, ExitError, CustomEntryPoint
from leverage.container import TerraformContainer


@dataclass
class ClusterInfo:
    cluster_name: str
    profile: str
    region: str


class KubeCtlContainer(TerraformContainer):
    """Container specifically tailored to run kubectl commands."""

    KUBECTL_CLI_BINARY = "/usr/local/bin/kubectl"
    KUBECTL_CONFIG_PATH = Path(f"/home/{TerraformContainer.CONTAINER_USER}/.kube")
    KUBECTL_CONFIG_FILE = KUBECTL_CONFIG_PATH / Path("config")
    METADATA_FILENAME = "metadata.yaml"

    def __init__(self, client):
        super().__init__(client)

        self.entrypoint = self.KUBECTL_CLI_BINARY

        self.host_kubectl_config_dir = Path.home() / Path(f".kube/{self.project}")
        if not self.host_kubectl_config_dir.exists():
            # make sure the folder exists before mounting it
            self.host_kubectl_config_dir.mkdir(parents=True)

        self.container_config["host_config"]["Mounts"].append(
            # the container is expecting a file named "config" here
            Mount(
                source=str(self.host_kubectl_config_dir),
                target=str(self.KUBECTL_CONFIG_PATH),
                type="bind",
            )
        )

    def start_shell(self):
        with AwsCredsEntryPoint(self, override_entrypoint=""):
            self._start(self.SHELL)

    def configure(self, ci: ClusterInfo = None):
        """
        Add the given EKS cluster configuration to the .kube/ files.
        """
        if ci:
            # if you have the details, generate the command right away
            cmd = f"aws eks update-kubeconfig --region {ci.region} --name {ci.cluster_name} --profile {ci.profile}"
        else:
            # otherwise go get them from the layer
            logger.info("Retrieving k8s cluster information...")
            with CustomEntryPoint(self, entrypoint=""):
                cmd = self._get_eks_kube_config()

        logger.info("Configuring context...")
        with AwsCredsEntryPoint(self, override_entrypoint=""):
            exit_code = self._start(cmd)

        if exit_code:
            raise Exit(exit_code)

        logger.info("Done.")

    def _get_eks_kube_config(self) -> str:
        exit_code, output = self._start_with_output(f"{self.TF_BINARY} output -no-color")  # TODO: override on CM?
        if exit_code:
            raise ExitError(exit_code, output)

        aws_eks_cmd = next(op for op in output.split("\r\n") if op.startswith("aws eks update-kubeconfig"))
        return aws_eks_cmd + f" --region {self.region}"

    def _scan_clusters(self):
        """
        Scan all the subdirectories in search of "cluster" metadata files.
        """
        for root, dirs, files in os.walk(self.paths.cwd):
            # exclude hidden directories
            dirs[:] = [d for d in dirs if not d[0] == "."]

            for file in files:
                if file != self.METADATA_FILENAME:
                    continue

                cluster_file = Path(root) / file
                try:
                    data = ruamel.yaml.load(cluster_file)
                    assert data["type"] == "k8s-eks-cluster"
                except AssertionError:
                    continue
                except Exception as exc:
                    logger.warning(exc)
                    continue
                else:
                    yield Path(root), data

    def discover(self):
        """
        TODO: documentation
        """
        cluster_files = [(path, data) for path, data in self._scan_clusters()]
        if not cluster_files:
            raise ExitError(1, "No clusters found.")

        terminal_menu = TerminalMenu(
            [f"{c[1]['data']['cluster_name']}: {str(c[0])}" for c in cluster_files], title="Clusters found:"
        )
        menu_entry_index = terminal_menu.show()
        if menu_entry_index is None:
            # selection cancelled
            return

        layer_path = cluster_files[menu_entry_index][0]
        cluster_data = cluster_files[menu_entry_index][1]
        cluster_info = ClusterInfo(
            cluster_name=cluster_data["data"]["cluster_name"],
            profile=cluster_data["data"]["profile"],
            region=cluster_data["data"]["region"],
        )

        # cluster is the host path, so in order to be able to run commands in that layer
        # we need to convert it into a relative inside the container
        self.container_config["working_dir"] = (
            self.paths.guest_base_path / layer_path.relative_to(self.paths.cwd)
        ).as_posix()

        # TODO: rather than overriding property by propery, maybe a custom .paths object pointing to cluster_path?
        self.paths.cwd = layer_path
        self.paths.account_config_dir = self.paths._account_config_dir(layer_path)
        self.paths.account_conf = self.paths.account_conf_from_layer(layer_path)

        self.configure(cluster_info)

1	import os	1✔
2	from dataclasses import dataclass	1✔
3	from pathlib import Path	1✔
4
5	from click.exceptions import Exit	1✔
6	from docker.types import Mount	1✔
7	import ruamel.yaml	1✔
8	from simple_term_menu import TerminalMenu	1✔
9
10	from leverage import logger	1✔
11	from leverage._utils import AwsCredsEntryPoint, ExitError, CustomEntryPoint	1✔
12	from leverage.container import TerraformContainer	1✔
13
14
15	@dataclass	1✔
16	class ClusterInfo:	1✔
17	cluster_name: str	1✔
18	profile: str	1✔
19	region: str	1✔
20
21
22	class KubeCtlContainer(TerraformContainer):	1✔
23	"""Container specifically tailored to run kubectl commands."""
24
25	KUBECTL_CLI_BINARY = "/usr/local/bin/kubectl"	1✔
26	KUBECTL_CONFIG_PATH = Path(f"/home/{TerraformContainer.CONTAINER_USER}/.kube")	1✔
27	KUBECTL_CONFIG_FILE = KUBECTL_CONFIG_PATH / Path("config")	1✔
28	METADATA_FILENAME = "metadata.yaml"	1✔
29
30	def __init__(self, client):	1✔
31	super().__init__(client)	1✔
32
33	self.entrypoint = self.KUBECTL_CLI_BINARY	1✔
34
35	self.host_kubectl_config_dir = Path.home() / Path(f".kube/{self.project}")	1✔
36	if not self.host_kubectl_config_dir.exists():	1✔
37	# make sure the folder exists before mounting it
38	self.host_kubectl_config_dir.mkdir(parents=True)	1✔
39
40	self.container_config["host_config"]["Mounts"].append(	1✔
41	# the container is expecting a file named "config" here
42	Mount(
43	source=str(self.host_kubectl_config_dir),
44	target=str(self.KUBECTL_CONFIG_PATH),
45	type="bind",
46	)
47	)
48
49	def start_shell(self):	1✔
50	with AwsCredsEntryPoint(self, override_entrypoint=""):	1✔
51	self._start(self.SHELL)	1✔
52
53	def configure(self, ci: ClusterInfo = None):	1✔
54	"""
55	Add the given EKS cluster configuration to the .kube/ files.
56	"""
57	if ci:	1✔
58	# if you have the details, generate the command right away
NEW 59	cmd = f"aws eks update-kubeconfig --region {ci.region} --name {ci.cluster_name} --profile {ci.profile}"	×
60	else:
61	# otherwise go get them from the layer
62	logger.info("Retrieving k8s cluster information...")	1✔
63	with CustomEntryPoint(self, entrypoint=""):	1✔
64	cmd = self._get_eks_kube_config()	1✔
65
66	logger.info("Configuring context...")	1✔
67	with AwsCredsEntryPoint(self, override_entrypoint=""):	1✔
68	exit_code = self._start(cmd)	1✔
69
70	if exit_code:	1✔
71	raise Exit(exit_code)	×
72
73	logger.info("Done.")	1✔
74
75	def _get_eks_kube_config(self) -> str:	1✔
76	exit_code, output = self._start_with_output(f"{self.TF_BINARY} output -no-color") # TODO: override on CM?	1✔
77	if exit_code:	1✔
78	raise ExitError(exit_code, output)	1✔
79
80	aws_eks_cmd = next(op for op in output.split("\r\n") if op.startswith("aws eks update-kubeconfig"))	1✔
81	return aws_eks_cmd + f" --region {self.region}"	1✔
82
83	def _scan_clusters(self):	1✔
84	"""
85	Scan all the subdirectories in search of "cluster" metadata files.
86	"""
87	for root, dirs, files in os.walk(self.paths.cwd):	1✔
88	# exclude hidden directories
89	dirs[:] = [d for d in dirs if not d[0] == "."]	1✔
90
91	for file in files:	1✔
92	if file != self.METADATA_FILENAME:	1✔
93	continue	1✔
94
95	cluster_file = Path(root) / file	1✔
96	try:	1✔
97	data = ruamel.yaml.load(cluster_file)	1✔
98	assert data["type"] == "k8s-eks-cluster"	1✔
NEW 99	except AssertionError:	×
NEW 100	continue	×
NEW 101	except Exception as exc:	×
NEW 102	logger.warning(exc)	×
NEW 103	continue	×
104	else:
105	yield Path(root), data	1✔
106
107	def discover(self):	1✔
108	"""
109	TODO: documentation
110	"""
NEW 111	cluster_files = [(path, data) for path, data in self._scan_clusters()]	×
NEW 112	if not cluster_files:	×
NEW 113	raise ExitError(1, "No clusters found.")	×
114
NEW 115	terminal_menu = TerminalMenu(	×
116	[f"{c[1]['data']['cluster_name']}: {str(c[0])}" for c in cluster_files], title="Clusters found:"
117	)
NEW 118	menu_entry_index = terminal_menu.show()	×
NEW 119	if menu_entry_index is None:	×
120	# selection cancelled
NEW 121	return	×
122
NEW 123	layer_path = cluster_files[menu_entry_index][0]	×
NEW 124	cluster_data = cluster_files[menu_entry_index][1]	×
NEW 125	cluster_info = ClusterInfo(	×
126	cluster_name=cluster_data["data"]["cluster_name"],
127	profile=cluster_data["data"]["profile"],
128	region=cluster_data["data"]["region"],
129	)
130
131	# cluster is the host path, so in order to be able to run commands in that layer
132	# we need to convert it into a relative inside the container
NEW 133	self.container_config["working_dir"] = (	×
134	self.paths.guest_base_path / layer_path.relative_to(self.paths.cwd)
135	).as_posix()
136
137	# TODO: rather than overriding property by propery, maybe a custom .paths object pointing to cluster_path?
NEW 138	self.paths.cwd = layer_path	×
NEW 139	self.paths.account_config_dir = self.paths._account_config_dir(layer_path)	×
NEW 140	self.paths.account_conf = self.paths.account_conf_from_layer(layer_path)	×
141
NEW 142	self.configure(cluster_info)	×

binbashar / leverage / 11758973854

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