Verify Kubernetes context before install (#1739)

* Add method to confirm kubectl context

* Set development server name to thecombine.localhost

* Add function to verify required environment variables are set

* Update Python dependencies

* Improve user feedback importing semantic domains

* Confirm kubernetes context & AWS environment when installing The Combine

Co-authored-by: D. Ror <imnasnainaec@gmail.com>

deploy/scripts/aws_env.py

@@ -0,0 +1,92 @@
+#! /usr/bin/env python3
+"""Set AWS Environment variables from aws cli profiles."""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Dict, List, Optional
+
+from utils import choose_from_list, run_cmd
+
+
+def aws_version() -> Optional[int]:
+    """Test if the aws cli version 2 is installed."""
+    try:
+        result = run_cmd(["aws", "--version"], check_results=False, chomp=True)
+    except FileNotFoundError:
+        print("AWS CLI version 2 is not installed.")
+        return None
+    else:
+        if result.returncode == 0:
+            # get major version number from stdout
+            match = re.match(r"aws-cli/(\d+)\..*", result.stdout)
+            if match:
+                return int(match.group(1))
+    return None
+
+
+def list_aws_profiles() -> List[str]:
+    aws_ver = aws_version()
+    if aws_ver is not None and aws_ver == 2:
+        result = run_cmd(["aws", "configure", "list-profiles"], chomp=True)
+        return result.stdout.split("\n")
+    return []
+
+
+def get_profile_var(profile: str, var_name: str) -> str:
+    result = run_cmd(["aws", "configure", "--profile", profile, "get", var_name], chomp=True)
+    return result.stdout
+
+
+def init_aws_environment() -> None:
+    profile_list = list_aws_profiles()
+    # Build a map for looking up a profile name from the access key id.  This
+    # algorithm assumes:
+    #  - the 'default' profile will be processed first
+    #  - if there are profiles using the same access key id, only the last
+    #    one will be put into the map
+    if len(profile_list) == 0:
+        return
+    profile_map: Dict[str, str] = {}
+    for profile in profile_list:
+        key_id = get_profile_var(profile, "aws_access_key_id")
+        profile_map[key_id] = profile
+    curr_access_key = os.getenv("AWS_ACCESS_KEY_ID", "")
+    if curr_access_key in profile_map:
+        curr_profile = profile_map[curr_access_key]
+    else:
+        curr_profile = None
+    aws_profile = choose_from_list("AWS Environment", curr_profile, profile_list)
+    if aws_profile is not None and aws_profile != curr_profile:
+        os.environ["AWS_PROFILE"] = aws_profile
+        os.environ["AWS_ACCESS_KEY_ID"] = get_profile_var(aws_profile, "aws_access_key_id")
+        os.environ["AWS_SECRET_ACCESS_KEY"] = get_profile_var(aws_profile, "aws_secret_access_key")
+        os.environ["AWS_DEFAULT_REGION"] = get_profile_var(aws_profile, "region")
+        result = run_cmd(
+            [
+                "aws",
+                "sts",
+                "--profile",
+                aws_profile,
+                "get-caller-identity",
+                "--query",
+                "Account",
+                "--output",
+                "text",
+            ],
+            chomp=True,
+        )
+        os.environ["AWS_ACCOUNT"] = result.stdout
+
+
+if __name__ == "__main__":
+    init_aws_environment()
+    print("AWS Environment:")
+    for env_var in [
+        "AWS_ACCOUNT",
+        "AWS_DEFAULT_REGION",
+        "AWS_ACCESS_KEY_ID",
+        "AWS_SECRET_ACCESS_KEY",
+    ]:
+        print(f"{env_var}: {os.getenv(env_var, None)}")

deploy/scripts/kube_env.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""Manage the Kubernetes environment for kubectl & helm."""
+
+from __future__ import annotations
+
+import argparse
+from typing import List, Optional
+
+from utils import choose_from_list, run_cmd
+
+
+class KubernetesEnvironment:
+    def __init__(self, args: argparse.Namespace) -> None:
+        if "kubeconfig" in args and args.kubeconfig is not None:
+            self.kubeconfig = args.kubeconfig
+        else:
+            self.kubeconfig = None
+        if "context" in args and args.context is not None:
+            # if the user specified a context, use that one.
+            self.kubecontext = args.context
+        else:
+            context_list: List[str] = []
+
+            result = run_cmd(
+                ["kubectl"] + self.get_kubeconfig() + ["config", "get-contexts", "--no-headers"],
+                check_results=True,
+            )
+            curr_context: Optional[str] = None
+            for line in result.stdout.splitlines():
+                if line[0] == "*":
+                    curr_context = line.split()[1]
+                    context_list.append(curr_context)
+                else:
+                    context_list.append(line.split()[0])
+
+            # If  there is more than one context available, prompt the user to make sure
+            # that the intended context will be used.
+            curr_context = choose_from_list("context", curr_context, context_list)
+            if curr_context:
+                self.kubecontext = curr_context
+            else:
+                self.kubecontext = None
+        if "debug" in args:
+            self.debug = args.debug
+        else:
+            self.debug = False
+
+    def get_kubeconfig(self) -> List[str]:
+        if self.kubeconfig is not None:
+            return ["--kubeconfig", self.kubeconfig]
+        return []
+
+    def get_helm_opts(self) -> List[str]:
+        """
+        Create list of general helm options.
+        """
+        helm_opts = self.get_kubeconfig()
+
+        if self.kubecontext is not None:
+            helm_opts.extend(["--kube-context", self.kubecontext])
+        if self.debug:
+            helm_opts.append("--debug")
+        return helm_opts
+
+    def get_kubectl_opts(self) -> List[str]:
+        """
+        Create list of general kubectl options.
+        """
+        kubectl_opts = self.get_kubeconfig()
+
+        if self.kubecontext is not None:
+            kubectl_opts.extend(["--context", self.kubecontext])
+        return kubectl_opts
+
+
+def add_kube_opts(parser: argparse.ArgumentParser) -> None:
+    """Add commandline arguments for Kubernetes tools."""
+    parser.add_argument(
+        "--context",
+        help="Context in kubectl configuration file to be used.",
+    )
+    parser.add_argument(
+        "--debug",
+        action="store_true",
+        help="Enable debugging output for helm commands.",
+    )
+    parser.add_argument(
+        "--kubeconfig",
+        help="Specify the kubectl configuration file to be used.",
+    )
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Generate Helm Charts for The Combine.",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    add_kube_opts(parser)
+    args = parser.parse_args()
+
+    kube_env = KubernetesEnvironment(args)
+    print(f"kubectl {kube_env.get_kubectl_opts()} ...")
+    print(f"helm {kube_env.get_helm_opts()} ...")

deploy/scripts/setup_cluster.py

@@ -11,7 +11,8 @@
 from typing import Any, Dict, List
 
 from enum_types import ExitStatus, HelmAction
-from utils import add_helm_opts, add_namespace, get_helm_opts, run_cmd
+from kube_env import KubernetesEnvironment, add_kube_opts
+from utils import add_namespace, run_cmd
 import yaml
 
 scripts_dir = Path(__file__).resolve().parent
@@ -24,7 +25,7 @@ def parse_args() -> argparse.Namespace:
         description="Build containerd container images for project.",
         formatter_class=argparse.ArgumentDefaultsHelpFormatter,
     )
-    add_helm_opts(parser)
+    add_kube_opts(parser)
     parser.add_argument(
         "--type",
         "-t",
@@ -90,13 +91,15 @@ def main() -> None:
     for chart in yaml.safe_load(chart_list_results.stdout):
         curr_charts.append(chart["name"])
 
+    # Verify the Kubernetes/Helm environment
+    kube_env = KubernetesEnvironment(args)
     # Install the required charts
     for chart_descr in this_cluster:
         chart_spec = config[chart_descr]["chart"]
         # add namespace if needed
-        add_namespace(chart_spec["namespace"])
+        add_namespace(chart_spec["namespace"], kube_env.get_kubectl_opts())
         # install the chart
-        helm_cmd = ["helm"] + get_helm_opts(args)
+        helm_cmd = ["helm"] + kube_env.get_helm_opts()
         if chart_spec["name"] in curr_charts:
             helm_action = HelmAction.UPGRADE
         else:

deploy/scripts/setup_combine.py

@@ -26,9 +26,11 @@
 from typing import Any, Dict, List
 
 from app_release import get_release
+from aws_env import init_aws_environment
 import combine_charts
 from enum_types import ExitStatus, HelmAction
-from utils import add_helm_opts, add_namespace, get_helm_opts, run_cmd
+from kube_env import KubernetesEnvironment, add_kube_opts
+from utils import add_namespace, run_cmd
 import yaml
 
 scripts_dir = Path(__file__).resolve().parent
@@ -41,7 +43,7 @@ def parse_args() -> argparse.Namespace:
         description="Generate Helm Charts for The Combine.",
         formatter_class=argparse.ArgumentDefaultsHelpFormatter,
     )
-    add_helm_opts(parser)
+    add_kube_opts(parser)
     parser.add_argument(
         "--clean", action="store_true", help="Delete chart, if it exists, before installing."
     )
@@ -213,9 +215,13 @@ def main() -> None:
     else:
         profile = args.profile
 
-    # Create a base helm command for commands used to alter
+    # Verify the Kubernetes/Helm environment
+    kube_env = KubernetesEnvironment(args)
+    # Cache options for helm commands used to alter
     # the target cluster
-    helm_opts = get_helm_opts(args)
+    helm_opts = kube_env.get_helm_opts()
+    # Check AWS Environment Variables
+    init_aws_environment()
 
     # create list of target specific variable values
     target_vars = [
@@ -242,7 +248,7 @@ def main() -> None:
         for chart in config["profiles"][profile]["charts"]:
             # create the chart namespace if it does not exist
             chart_namespace = config["charts"][chart]["namespace"]
-            if add_namespace(chart_namespace):
+            if add_namespace(chart_namespace, kube_env.get_kubectl_opts()):
                 installed_charts: List[str] = []
             else:
                 # get list of charts in target namespace

deploy/scripts/setup_files/combine_config.yaml

@@ -17,7 +17,7 @@ targets:
       # override values for 'thecombine' chart
       thecombine:
         global:
-          serverName: thecombine.local
+          serverName: thecombine.localhost
   nuc1:
     profile: nuc
     env_vars_required: true

deploy/scripts/utils.py

@@ -4,7 +4,6 @@
 
 from __future__ import annotations
 
-import argparse
 import subprocess
 import sys
 from typing import List, Optional
@@ -44,49 +43,61 @@ def run_cmd(
         sys.exit(err.returncode)
 
 
-def add_namespace(namespace: str) -> bool:
+def add_namespace(namespace: str, kube_opts: List[str]) -> bool:
     """
     Create a Kubernetes namespace if and only if it does not exist.
 
     Returns True if the namespace was added.
     """
-    lookup_results = run_cmd(["kubectl", "get", "namespace", namespace], check_results=False)
+    lookup_results = run_cmd(
+        ["kubectl"] + kube_opts + ["get", "namespace", namespace], check_results=False
+    )
     if lookup_results.returncode != 0:
-        run_cmd(["kubectl", "create", "namespace", namespace])
+        run_cmd(["kubectl"] + kube_opts + ["create", "namespace", namespace])
         return True
     return False
 
 
-def add_helm_opts(parser: argparse.ArgumentParser) -> None:
-    """
-    Add commandline arguments that are shared between scripts calling helm.
-
-    Sets up '--verbose' as the equivalent of '--debug'.
+def choose_from_list(
+    name: str, curr_selection: Optional[str], options: List[str]
+) -> Optional[str]:
     """
-    parser.add_argument(
-        "--context",
-        help="Context in kubectl configuration file to be used.",
-    )
-    parser.add_argument(
-        "--debug",
-        action="store_true",
-        help="Enable debugging output for helm commands.",
-    )
-    parser.add_argument(
-        "--kubeconfig",
-        help="Specify the kubectl configuration file to be used.",
-    )
+    Prompt user to choose/confirm a selection from a list.
 
-
-def get_helm_opts(args: argparse.Namespace) -> List[str]:
-    """
-    Create list of general helm options based on argparse Namespace.
+    The curr_selection is automatically chosen if the options List is empty
+    or has the curr_selection as its only member.
     """
-    helm_opts = []
-    if args.kubeconfig:
-        helm_opts.extend(["--kubeconfig", args.kubeconfig])
-    if args.context:
-        helm_opts.extend(["--kube-context", args.context])
-    if args.debug:
-        helm_opts.append("--debug")
-    return helm_opts
+    if len(options) == 1 and curr_selection is not None and curr_selection == options[0]:
+        return curr_selection
+    if len(options) >= 1:
+        while True:
+            print(f"Choose {name} from:")
+            for index, option in enumerate(options):
+                print(f"\t{index+1}: {option}")
+            if curr_selection is None:
+                prompt_str = f"Enter {name}: "
+            else:
+                prompt_str = f"Enter {name} (Default: {curr_selection}): "
+            try:
+                reply = input(prompt_str)
+            except KeyboardInterrupt:
+                print("\nCancelled.")
+                sys.exit(1)
+            else:
+                if not reply:
+                    break
+                elif reply in options:
+                    curr_selection = reply
+                    break
+                else:
+                    try:
+                        index = int(reply)
+                        if index > 0 and index <= len(options):
+                            curr_selection = options[index - 1]
+                            break
+                        else:
+                            curr_selection = None
+                    except ValueError:
+                        curr_selection = None
+                    print(f"{reply} is not in the list.  Please re-enter.")
+    return curr_selection