[BV-599] Kubeconfig helper

leverage/containers/kubectl.py

@@ -1,19 +1,36 @@
+import os
+from dataclasses import dataclass
+from enum import Enum
 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
+from leverage._utils import AwsCredsEntryPoint, ExitError, CustomEntryPoint
 from leverage.container import TerraformContainer
 
 
+@dataclass
+class ClusterInfo:
+    cluster_name: str
+    profile: str
+    region: str
+
+
+class MetadataTypes(Enum):
+    K8S_CLUSTER = "k8s-eks-cluster"
+
+
 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)
@@ -38,18 +55,23 @@ def start_shell(self):
         with AwsCredsEntryPoint(self, override_entrypoint=""):
             self._start(self.SHELL)
 
-    def configure(self):
-        # make sure we are on the cluster layer
-        self.paths.check_for_cluster_layer()
-
-        logger.info("Retrieving k8s cluster information...")
-        # generate the command that will configure the new cluster
-        with AwsCredsEntryPoint(self, override_entrypoint=""):
-            add_eks_cluster_cmd = self._get_eks_kube_config()
+    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(add_eks_cluster_cmd)
+            exit_code = self._start(cmd)
+
         if exit_code:
             raise Exit(exit_code)
 
@@ -62,3 +84,61 @@ def _get_eks_kube_config(self) -> str:
 
         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 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.get("type") != MetadataTypes.K8S_CLUSTER.value:
+                        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)

leverage/modules/auth.py

@@ -65,7 +65,7 @@ def get_profiles(cli):
     # these are files from the layer we are currently on
     for name in ("config.tf", "locals.tf"):
         try:
-            with open(name) as tf_file:
+            with open(cli.paths.cwd / name) as tf_file:
                 tf_config = hcl2.load(tf_file)
         except FileNotFoundError:
             continue

leverage/modules/kubectl.py

@@ -18,19 +18,26 @@ def kubectl(context, state, args):
     """Run Kubectl commands in a custom containerized environment."""
     state.container = KubeCtlContainer(get_docker_client())
     state.container.ensure_image()
-    state.container.paths.check_for_layer_location()
     _handle_subcommand(context=context, cli_container=state.container, args=args)
 
 
 @kubectl.command(context_settings=CONTEXT_SETTINGS)
 @pass_container
 def shell(kctl: KubeCtlContainer):
     """Spawn a shell with the kubectl credentials pre-configured."""
+    kctl.paths.check_for_layer_location()
     kctl.start_shell()
 
 
 @kubectl.command(context_settings=CONTEXT_SETTINGS)
 @pass_container
 def configure(kctl: KubeCtlContainer):
     """Automatically add the EKS cluster from the layer into your kubectl config file."""
+    kctl.paths.check_for_cluster_layer()
     kctl.configure()
+
+
+@kubectl.command(context_settings=CONTEXT_SETTINGS)
+@pass_container
+def discover(kctl: KubeCtlContainer):
+    kctl.discover()

leverage/path.py

@@ -2,6 +2,7 @@
     Utilities to obtain relevant files' and directories' locations
 """
 import os
+import pathlib
 from pathlib import Path
 from subprocess import CalledProcessError
 from subprocess import PIPE
@@ -169,6 +170,14 @@ def __init__(self, env_conf: dict, container_user: str):
             self.host_aws_credentials_dir.mkdir(parents=True)
         self.sso_cache = self.host_aws_credentials_dir / "sso" / "cache"
 
+    def update_cwd(self, new_cwd):
+        self.cwd = new_cwd
+        acc_folder = new_cwd.relative_to(self.root_dir).parts[0]
+
+        self.account_config_dir = self.root_dir / acc_folder / "config"
+        account_config_path = self.account_config_dir / self.ACCOUNT_TF_VARS
+        self.account_conf = hcl2.loads(account_config_path.read_text())
+
     @property
     def guest_account_base_path(self):
         return f"{self.guest_base_path}/{self.account_dir.relative_to(self.root_dir).as_posix()}"
@@ -257,25 +266,27 @@ def guest_config_file(self, file):
     def tf_cache_dir(self):
         return os.getenv("TF_PLUGIN_CACHE_DIR")
 
-    def check_for_layer_location(self):
-        """Make sure the command is being ran at layer level. If not, bail."""
-        if self.cwd in (self.common_config_dir, self.account_config_dir):
+    def check_for_layer_location(self, path: Path = None):
+        """Make sure the command is being run at layer level. If not, bail."""
+        path = path or self.cwd
+        if path in (self.common_config_dir, self.account_config_dir):
             raise ExitError(1, "Currently in a configuration directory, no Terraform command can be run here.")
 
-        if self.cwd in (self.root_dir, self.account_dir):
+        if path in (self.root_dir, self.account_dir):
             raise ExitError(
                 1,
                 "Terraform commands cannot run neither in the root of the project or in"
                 " the root directory of an account.",
             )
 
-        if not list(self.cwd.glob("*.tf")):
+        if not list(path.glob("*.tf")):
             raise ExitError(1, "This command can only run at [bold]layer[/bold] level.")
 
-    def check_for_cluster_layer(self):
-        self.check_for_layer_location()
+    def check_for_cluster_layer(self, path: Path = None):
+        path = path or self.cwd
+        self.check_for_layer_location(path)
         # assuming the "cluster" layer will contain the expected EKS outputs
-        if self.cwd.parts[-1] != "cluster":
+        if path.parts[-1] != "cluster":
             raise ExitError(1, "This command can only run at the [bold]cluster layer[/bold].")