11999658197

Committed 24 Nov 2024 09:16PM UTC coverage: 60.419% (+0.4%) from 60.037%

Build # 11999658197

Build Type

Pull #281

github

Committed by

web-flow

Commit Message

Merge 7e2f57c24 into e92f56ef4

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

Run Details

196 of 460 branches covered (42.61%)

Branch coverage included in aggregate %.

52 of 73 new or added lines in 4 files covered. (71.23%)

47 existing lines in 5 files now uncovered.

2518 of 4032 relevant lines covered (62.45%)

0.62 hits per line

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

88.89

/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
import simple_term_menu

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:
                    with open(cluster_file) as cluster_yaml_file:
                        data = ruamel.yaml.safe_load(cluster_yaml_file)
                    if data["type"] != "k8s-eks-cluster":
                        continue
                except KeyError:
                    continue
                except Exception as exc:
                    logger.warning(exc)
                    continue
                else:
                    yield Path(root), data

    def discover(self):
        """
        Do a scan down the tree of subdirectories looking for k8s clusters metadata files.
        Open up a menu with all the found items, where you can pick up and configure it on your .kubeconfig file.
        """
        cluster_files = [(path, data) for path, data in self._scan_clusters()]
        if not cluster_files:
            raise ExitError(1, "No clusters found.")

        terminal_menu = simple_term_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()
        # now simulate we are standing on the chosen layer folder
        self.paths.update_cwd(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	import simple_term_menu	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	with open(cluster_file) as cluster_yaml_file:	1✔
98	data = ruamel.yaml.safe_load(cluster_yaml_file)	1✔
99	if data["type"] != "k8s-eks-cluster":	1✔
NEW 100	continue	×
NEW 101	except KeyError:	×
NEW 102	continue	×
NEW 103	except Exception as exc:	×
NEW 104	logger.warning(exc)	×
NEW 105	continue	×
106	else:
107	yield Path(root), data	1✔
108
109	def discover(self):	1✔
110	"""
111	Do a scan down the tree of subdirectories looking for k8s clusters metadata files.
112	Open up a menu with all the found items, where you can pick up and configure it on your .kubeconfig file.
113	"""
114	cluster_files = [(path, data) for path, data in self._scan_clusters()]	1✔
115	if not cluster_files:	1✔
NEW 116	raise ExitError(1, "No clusters found.")	×
117
118	terminal_menu = simple_term_menu.TerminalMenu(	1✔
119	[f"{c[1]['data']['cluster_name']}: {str(c[0])}" for c in cluster_files], title="Clusters found:"
120	)
121	menu_entry_index = terminal_menu.show()	1✔
122	if menu_entry_index is None:	1✔
123	# selection cancelled
NEW 124	return	×
125
126	layer_path = cluster_files[menu_entry_index][0]	1✔
127	cluster_data = cluster_files[menu_entry_index][1]	1✔
128	cluster_info = ClusterInfo(	1✔
129	cluster_name=cluster_data["data"]["cluster_name"],
130	profile=cluster_data["data"]["profile"],
131	region=cluster_data["data"]["region"],
132	)
133
134	# cluster is the host path, so in order to be able to run commands in that layer
135	# we need to convert it into a relative inside the container
136	self.container_config["working_dir"] = (	1✔
137	self.paths.guest_base_path / layer_path.relative_to(self.paths.cwd)
138	).as_posix()
139	# now simulate we are standing on the chosen layer folder
140	self.paths.update_cwd(layer_path)	1✔
141	self.configure(cluster_info)	1✔

binbashar / leverage / 11999658197

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