Update implementation of the cluster-stack remote api

Signed-off-by: michal.gubricky <[email protected]>

Tests/kaas/plugin/plugin_cluster_stacks_remote_api.py

@@ -1,13 +1,24 @@
 import os
 import yaml
 import subprocess
-import base64
 import time
 import logging
 from interface import KubernetesClusterPlugin
 
 logger = logging.getLogger("PluginClusterStacks")
 
+
+# Default configuration values
+DEFAULTS = {
+    'cs_name': 'scs',
+}
+
+# Keys needed for environment variables
+ENV_KEYS = {'cs_name', 'cs_version', 'cs_channel', 'cs_secretname', 'cs_class_name',
+            'cs_namespace', 'cs_pod_cidr', 'cs_service_cidr', 'cs_external_id', 'cs_k8s_patch_version',
+            'cs_cluster_name', 'cs_k8s_version'}
+
+
 # Helper functions
 def wait_for_pods(self, namespaces, timeout=240, interval=15, kubeconfig=None):
     """
@@ -27,13 +38,20 @@ def wait_for_pods(self, namespaces, timeout=240, interval=15, kubeconfig=None):
         for namespace in namespaces:
             try:
                 # Get pod status in the namespace
-                wait_pods_command = (
-                    f"kubectl wait -n {namespace} --for=condition=Ready --timeout={timeout}s pod --all"
+                wait_pods_command = f"kubectl wait -n {namespace} --for=condition=Ready --timeout={timeout}s pod --all"
+                result = self._run_subprocess(
+                    wait_pods_command,
+                    f"Error fetching pods in {namespace}",
+                    shell=True,
+                    capture_output=True,
+                    text=True,
+                    kubeconfig=kubeconfig
                 )
-                result = self._run_subprocess(wait_pods_command, f"Error fetching pods in {namespace}", shell=True, capture_output=True, text=True, kubeconfig=kubeconfig)
 
                 if result.returncode != 0:
-                    logger.warning(f"Not all pods in namespace {namespace} are ready. Details: {result.stderr}")
+                    logger.warning(
+                        f"Not all pods in namespace {namespace} are ready. Details: {result.stderr}"
+                    )
                     all_pods_ready = False
                 else:
                     logger.info(f"All pods in namespace {namespace} are ready.")
@@ -49,69 +67,121 @@ def wait_for_pods(self, namespaces, timeout=240, interval=15, kubeconfig=None):
         logger.info("Waiting for all pods in specified namespaces to become ready...")
         time.sleep(interval)
 
-    raise TimeoutError(f"Timed out after {timeout} seconds waiting for pods in namespaces {namespaces} to become ready.")
+    raise TimeoutError(
+        f"Timed out after {timeout} seconds waiting for pods in namespaces {namespaces} to become ready."
+    )
 
 
 def load_config(config_path):
     """
     Loads the configuration from a YAML file.
     """
 
-    with open(config_path, 'r') as file:
+    with open(config_path, "r") as file:
         config = yaml.safe_load(file) or {}
+
+        base_dir = os.path.dirname(config_path)
+        if 'kubeconfig' in config:
+            config['kubeconfig'] = os.path.join(base_dir, config['kubeconfig'])
+        if 'workloadcluster' in config:
+            config['workloadcluster'] = os.path.join(base_dir, config['workloadcluster'])
+        if 'clusterstack' in config:
+            config['clusterstack'] = os.path.join(base_dir, config['clusterstack'])
+
         return config
 
+
+def setup_environment_variables(self):
+    """
+    Constructs and returns a dictionary of required environment variables
+    based on the configuration.
+
+    :raises ValueError: If the `GIT_ACCESS_TOKEN` environment variable is not set.
+
+    :return: A dictionary of required environment variables with necessary values and
+             encodings for Kubernetes and Git-related configurations.
+    """
+    # Calculate values that need to be set dynamically
+    if hasattr(self, 'cluster_version'):
+        self.config['cs_k8s_version'] = self.cluster_version
+        self.config['cs_namespace'] = self.cs_namespace
+        self.config['cs_class_name'] = (
+            f"openstack-{self.config['cs_name']}-{str(self.config['cs_k8s_version']).replace('.', '-')}-"
+            f"{self.config['cs_version']}"
+        )
+    if hasattr(self, 'cluster_name'):
+        self.config['cs_cluster_name'] = self.cluster_name
+
+    # Construct general environment variables
+    required_env = {key.upper(): value for key, value in self.config.items() if key in ENV_KEYS}
+
+    return required_env
+
+
 class PluginClusterStacksRemoteAPI(KubernetesClusterPlugin):
-    def __init__(self, config_file=None):
+    def __init__(self, config_file):
         self.config = load_config(config_file) if config_file else {}
         logger.debug(self.config)
         self.working_directory = os.getcwd()
         logger.debug(f"Working from {self.working_directory}")
-        self.kubeconfig_mgmnt = self.config['kubeconfig']
-        self.workloadclusters  = self.config['workloadcluster']
-        self.cs_namespace = self.config['namespace']
-
-
-    def create_cluster(self, cluster_name=None, version=None, kubeconfig_filepath=None):
+        for key, value in DEFAULTS.items():
+            self.config.setdefault(key, value)
+        self.kubeconfig_mgmnt = self.config["kubeconfig"]
+        self.workloadclusters = self.config["workloadcluster"]
+        self.clusterstack = self.config["clusterstack"]
+        self.cs_namespace = self.config["namespace"]
+
+    def create_cluster(self, cluster_name, version, kubeconfig_filepath):
         self.cluster_name = cluster_name
         self.cluster_version = version
         self.kubeconfig_cs_cluster = kubeconfig_filepath
 
-        # Create workload cluster
-        self._apply_yaml(self.workloadclusters, "Error applying cluster.yaml", kubeconfig=self.kubeconfig_mgmnt)
+        # Create cluster-stack resource
+        self._apply_yaml(self.clusterstack, "Error applying clusterstack.yaml", kubeconfig=self.kubeconfig_mgmnt)
 
-        #TODO:!!! We also need to introduce a waiting function here
-
-        print("retrieve kubeconfig to path")
+        # Create workload cluster
+        self._apply_yaml(
+            self.workloadclusters,
+            "Error applying cluster.yaml",
+            kubeconfig=self.kubeconfig_mgmnt,
+        )
+
+        # Get and wait on kubeadmcontrolplane and retrieve workload cluster kubeconfig
+        kcp_name = self._get_kubeadm_control_plane_name(namespace=self.cs_namespace, kubeconfig=self.kubeconfig_mgmnt)
+        self._wait_kcp_ready(kcp_name, namespace=self.cs_namespace, kubeconfig=self.kubeconfig_mgmnt)
         self._retrieve_kubeconfig(namespace=self.cs_namespace, kubeconfig=self.kubeconfig_mgmnt)
 
         # Wait for workload system pods to be ready
-        # wait_for_workload_pods_ready(kubeconfig_path=self.kubeconfig_cs_cluster)
-        # ~ wait_for_pods(self, ["kube-system"], timeout=600, interval=15, kubeconfig=self.kubeconfig_cs_cluster)
-
+        wait_for_pods(self, ["kube-system"], timeout=600, interval=15, kubeconfig=self.kubeconfig_cs_cluster)
 
-    def delete_cluster(self, cluster_name=None): #TODO:!!! need to adjust delete method
-        self.cluster_name = cluster_name
-        #Get the name of the workloadcluster from the config file
-        workload_cluster_config = load_config(self.workloadclusters)
-        workload_cluster_name = workload_cluster_config['metadata']['name']
+    def delete_cluster(self, cluster_name):
         try:
             # Check if the cluster exists
-            check_cluster_command = f"kubectl --kubeconfig={self.kubeconfig_mgmnt} get cluster {workload_cluster_name} -n {self.cs_namespace}"
-            result = self._run_subprocess(check_cluster_command, "Failed to get cluster resource", shell=True, capture_output=True, text=True)
+            check_cluster_command = f"kubectl get cluster {cluster_name} -n {self.cs_namespace}"
+            result = self._run_subprocess(
+                check_cluster_command,
+                "Failed to get cluster resource",
+                shell=True,
+                capture_output=True,
+                text=True,
+                kubeconfig=self.kubeconfig_mgmnt
+            )
 
             # Proceed with deletion only if the cluster exists
             if result.returncode == 0:
-                delete_command = f"kubectl --kubeconfig={self.kubeconfig_mgmnt} delete cluster {workload_cluster_name} --timeout=600s -n {self.cs_namespace}"
-                self._run_subprocess(delete_command, "Timeout while deleting the cluster", shell=True)
+                delete_command = f"kubectl delete cluster {cluster_name} --timeout=600s -n {self.cs_namespace}"
+                self._run_subprocess(
+                    delete_command, "Timeout while deleting the cluster", shell=True, kubeconfig=self.kubeconfig_mgmnt
+                )
 
         except subprocess.CalledProcessError as error:
             if "NotFound" in error.stderr:
-                logger.info(f"Cluster {workload_cluster_name} not found. Skipping deletion.")
+                logger.info(
+                    f"Cluster {cluster_name} not found. Skipping deletion."
+                )
             else:
                 raise RuntimeError(f"Error checking for cluster existence: {error}")
 
-
     def _apply_yaml(self, yaml_file, error_msg, kubeconfig=None):
         """
         Applies a Kubernetes YAML configuration file to the cluster, substituting environment variables as needed.
@@ -123,15 +193,63 @@ def _apply_yaml(self, yaml_file, error_msg, kubeconfig=None):
         try:
             # Determine if the file is a local path or a URL
             if os.path.isfile(yaml_file):
-                command = f"kubectl --kubeconfig={self.kubeconfig_mgmnt} apply -f {yaml_file} -n {self.cs_namespace}"
+                command = f"/tmp/envsubst < {yaml_file} | kubectl apply -f -"
             else:
                 raise ValueError(f"Unknown file: {yaml_file}")
 
-            self._run_subprocess(command, error_msg, shell=True)
+            self._run_subprocess(command, error_msg, shell=True, kubeconfig=kubeconfig)
 
         except subprocess.CalledProcessError as error:
             raise RuntimeError(f"{error_msg}: {error}")
 
+    def _get_kubeadm_control_plane_name(self, namespace="default", kubeconfig=None):
+        """
+        Retrieves the name of the KubeadmControlPlane resource for the Kubernetes cluster
+        in the specified namespace.
+
+        :param namespace: The namespace to search for the KubeadmControlPlane resource.
+        :param kubeconfig: Optional path to the kubeconfig file for the target Kubernetes cluster.
+
+        :return: The name of the KubeadmControlPlane resource as a string.
+        """
+        max_retries = 6
+        delay_between_retries = 20
+        for _ in range(max_retries):
+            try:
+                kcp_command = (
+                    f"kubectl get kubeadmcontrolplane -n {namespace} "
+                    "-o=jsonpath='{.items[0].metadata.name}'"
+                )
+                kcp_name = self._run_subprocess(kcp_command, "Error retrieving kcp_name", shell=True, capture_output=True, text=True, kubeconfig=kubeconfig)
+                logger.info(kcp_name)
+                kcp_name_stdout = kcp_name.stdout.strip()
+                if kcp_name_stdout:
+                    print(f"KubeadmControlPlane name: {kcp_name_stdout}")
+                    return kcp_name_stdout
+            except subprocess.CalledProcessError as error:
+                print(f"Error getting kubeadmcontrolplane name: {error}")
+            # Wait before retrying
+            time.sleep(delay_between_retries)
+        else:
+            raise RuntimeError("Failed to get kubeadmcontrolplane name")
+
+    def _wait_kcp_ready(self, kcp_name, namespace="default", kubeconfig=None):
+        """
+        Waits for the specified KubeadmControlPlane resource to become 'Available'.
+
+        :param kcp_name: The name of the KubeadmControlPlane resource to check for availability.
+        :param namespace: The namespace where the KubeadmControlPlane resource is.
+        :param kubeconfig: Optional path to the kubeconfig file for the target Kubernetes cluster.
+        """
+        try:
+            self._run_subprocess(
+                f"kubectl wait kubeadmcontrolplane/{kcp_name} --for=condition=Available --timeout=600s -n {namespace}",
+                "Error waiting for kubeadmcontrolplane availability",
+                shell=True,
+                kubeconfig=kubeconfig
+            )
+        except subprocess.CalledProcessError as error:
+            raise RuntimeError(f"Error waiting for kubeadmcontrolplane to be ready: {error}")
 
     def _retrieve_kubeconfig(self, namespace="default", kubeconfig=None):
         """
@@ -140,26 +258,12 @@ def _retrieve_kubeconfig(self, namespace="default", kubeconfig=None):
         :param namespace: The namespace of the cluster to retrieve the kubeconfig for.
         :param kubeconfig: Optional path to the kubeconfig file for the target Kubernetes cluster.
         """
+        kubeconfig_command = (
+            f"sudo -E clusterctl get kubeconfig {self.cluster_name} -n {namespace} > {self.kubeconfig_cs_cluster}"
+        )
+        self._run_subprocess(kubeconfig_command, "Error retrieving kubeconfig", shell=True, kubeconfig=kubeconfig)
 
-        #Get the name of the workloadcluster from the config file
-        workload_cluster_config = load_config(self.workloadclusters)
-        workload_cluster_name = workload_cluster_config['metadata']['name']
-
-        command_args = [
-            "kubectl ",
-            f"--kubeconfig={self.kubeconfig_mgmnt}",
-            f"-n {self.cs_namespace}",
-            f"get secret {workload_cluster_name}-kubeconfig", 
-            "-o go-template='{{.data.value|base64decode}}'",
-            f"> {self.kubeconfig_cs_cluster}",
-        ]
-        kubeconfig_command = ""
-        for entry in command_args:
-            kubeconfig_command += entry + " "
-        self._run_subprocess(kubeconfig_command, "Error retrieving kubeconfig", shell=True)
-
-
-    def _run_subprocess(self, command, error_msg, shell=False, capture_output=False, text=False):
+    def _run_subprocess(self, command, error_msg, shell=False, capture_output=False, text=False, kubeconfig=None):
         """
         Executes a subprocess command with the specified environment variables and parameters.
 
@@ -168,12 +272,24 @@ def _run_subprocess(self, command, error_msg, shell=False, capture_output=False,
         :param shell: Whether to execute the command through the shell (default: `False`).
         :param capture_output: Whether to capture the command's standard output and standard error (default: `False`).
         :param text: Whether to treat the command's output and error as text (default: `False`).
+        :param kubeconfig: Optional path to the kubeconfig file for the target Kubernetes cluster.
+
         :return: The result of the `subprocess.run` command
         """
         try:
-            # Run the subprocess
-            result = subprocess.run(command, shell=shell, capture_output=capture_output, text=text, check=True)
+            env = setup_environment_variables(self)
+            env['PATH'] = f'/usr/local/bin:/usr/bin:{self.working_directory}'
+            # Set env variable DISPLAY which you need to open the oidc window automatically
+            env['DISPLAY'] = ':0'
+            env['HOME'] = self.working_directory
+            if kubeconfig:
+                env['KUBECONFIG'] = kubeconfig
+
+            # Run the subprocess with the environment
+            result = subprocess.run(command, shell=shell, capture_output=capture_output, text=text, check=True, env=env)
+
             return result
+
         except subprocess.CalledProcessError as error:
             logger.error(f"{error_msg}: {error}")
             raise

playbooks/k8s_configs/moin_cluster_clusterstack.yaml

@@ -0,0 +1,30 @@
+# Cluster-stack and OpenStack cluster-stack release resource templates
+apiVersion: clusterstack.x-k8s.io/v1alpha1
+kind: ClusterStack
+metadata:
+  name: clusterstack
+  namespace: ${CS_NAMESPACE}
+spec:
+  provider: openstack
+  name: ${CS_NAME}
+  kubernetesVersion: "${CS_K8S_VERSION}"
+  channel: ${CS_CHANNEL}
+  autoSubscribe: false
+  providerRef:
+    apiVersion: infrastructure.clusterstack.x-k8s.io/v1alpha1
+    kind: OpenStackClusterStackReleaseTemplate
+    name: cspotemplate
+  versions:
+    - ${CS_VERSION}
+---
+apiVersion: infrastructure.clusterstack.x-k8s.io/v1alpha1
+kind: OpenStackClusterStackReleaseTemplate
+metadata:
+  name: cspotemplate
+  namespace: ${CS_NAMESPACE}
+spec:
+  template:
+    spec:
+      identityRef:
+        kind: Secret
+        name: ${CS_SECRETNAME}

playbooks/k8s_configs/moin_cluster_config.yaml

@@ -1,3 +1,16 @@
-kubeconfig: "../../playbooks/k8s_configs/moin_cluster_kubeconfig.yaml"
-workloadcluster: "../../playbooks/k8s_configs/moin_cluster_workloadcluster.yaml"
-namespace: "kaas-playground1"
+kubeconfig: "moin_cluster_kubeconfig.yaml"
+workloadcluster: "moin_cluster_workloadcluster.yaml"
+clusterstack: "moin_cluster_clusterstack.yaml"
+namespace: "kaas-playground8"
+
+# Cluster-stack related configuration
+cs_name: "scs"  # Cluster Stack Name
+cs_version: "v1"  # Cluster Stack Version
+cs_channel: "stable"  # Release channel
+cs_secretname: "openstack"  # Cloud name from OpenStack clouds.yaml
+
+# Cluster Information
+cs_pod_cidr: "192.168.0.0/16"  # Pod CIDR for networking
+cs_service_cidr: "10.96.0.0/12"  # Service CIDR for networking
+cs_external_id: "ebfe5546-f09f-4f42-ab54-094e457d42ec"  # External ID for the Cluster Stack
+cs_k8s_patch_version: "9"  # Kubernetes patch version to use