X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=n2vc%2Fk8s_helm_base_conn.py;h=20fa337ddd8d681adc11eab405e0e0005edca745;hb=refs%2Ftags%2Fv10.1.0;hp=b3d7bda3c83f40812ccf03d2a196ea247099dc48;hpb=82b591ceed704c798ead2d9104085a08e75b511b;p=osm%2FN2VC.git diff --git a/n2vc/k8s_helm_base_conn.py b/n2vc/k8s_helm_base_conn.py index b3d7bda..20fa337 100644 --- a/n2vc/k8s_helm_base_conn.py +++ b/n2vc/k8s_helm_base_conn.py @@ -26,11 +26,11 @@ import time import shlex import shutil import stat -import subprocess import os import yaml from uuid import uuid4 +from n2vc.config import EnvironConfig from n2vc.exceptions import K8sException from n2vc.k8s_conn import K8sConnector @@ -44,7 +44,6 @@ class K8sHelmBaseConnector(K8sConnector): """ service_account = "osm" - _STABLE_REPO_URL = "https://charts.helm.sh/stable" def __init__( self, @@ -54,7 +53,6 @@ class K8sHelmBaseConnector(K8sConnector): helm_command: str = "/usr/bin/helm", log: object = None, on_update_db=None, - vca_config: dict = None, ): """ @@ -71,6 +69,7 @@ class K8sHelmBaseConnector(K8sConnector): self.log.info("Initializing K8S Helm connector") + self.config = EnvironConfig() # random numbers for release name generation random.seed(time.time()) @@ -86,10 +85,9 @@ class K8sHelmBaseConnector(K8sConnector): self._check_file_exists(filename=helm_command, exception_if_not_exists=True) # obtain stable repo url from config or apply default - if not vca_config or not vca_config.get("stablerepourl"): - self._stable_repo_url = self._STABLE_REPO_URL - else: - self._stable_repo_url = vca_config.get("stablerepourl") + self._stable_repo_url = self.config.get("stablerepourl") + if self._stable_repo_url == "None": + self._stable_repo_url = None @staticmethod def _get_namespace_cluster_id(cluster_uuid: str) -> (str, str): @@ -160,23 +158,27 @@ class K8sHelmBaseConnector(K8sConnector): ) ) - # sync local dir - self.fs.sync(from_path=cluster_id) - # init_env paths, env = self._init_paths_env( cluster_name=cluster_id, create_if_not_exist=True ) + # sync local dir + self.fs.sync(from_path=cluster_id) + # helm repo update - command = "{} repo update".format(self._helm_command) + command = "env KUBECONFIG={} {} repo update".format( + paths["kube_config"], self._helm_command + ) self.log.debug("updating repo: {}".format(command)) await self._local_async_exec( command=command, raise_exception_on_error=False, env=env ) # helm repo add name url - command = "{} repo add {} {}".format(self._helm_command, name, url) + command = "env KUBECONFIG={} {} repo add {} {}".format( + paths["kube_config"], self._helm_command, name, url + ) self.log.debug("adding repo: {}".format(command)) await self._local_async_exec( command=command, raise_exception_on_error=True, env=env @@ -195,15 +197,17 @@ class K8sHelmBaseConnector(K8sConnector): _, cluster_id = self._get_namespace_cluster_id(cluster_uuid) self.log.debug("list repositories for cluster {}".format(cluster_id)) - # sync local dir - self.fs.sync(from_path=cluster_id) - # config filename paths, env = self._init_paths_env( cluster_name=cluster_id, create_if_not_exist=True ) - command = "{} repo list --output yaml".format(self._helm_command) + # sync local dir + self.fs.sync(from_path=cluster_id) + + command = "env KUBECONFIG={} {} repo list --output yaml".format( + paths["kube_config"], self._helm_command + ) # Set exception to false because if there are no repos just want an empty list output, _rc = await self._local_async_exec( @@ -228,15 +232,17 @@ class K8sHelmBaseConnector(K8sConnector): _, cluster_id = self._get_namespace_cluster_id(cluster_uuid) self.log.debug("remove {} repositories for cluster {}".format(name, cluster_id)) - # sync local dir - self.fs.sync(from_path=cluster_id) - # init env, paths paths, env = self._init_paths_env( cluster_name=cluster_id, create_if_not_exist=True ) - command = "{} repo remove {}".format(self._helm_command, name) + # sync local dir + self.fs.sync(from_path=cluster_id) + + command = "env KUBECONFIG={} {} repo remove {}".format( + paths["kube_config"], self._helm_command, name + ) await self._local_async_exec( command=command, raise_exception_on_error=True, env=env ) @@ -330,6 +336,11 @@ class K8sHelmBaseConnector(K8sConnector): kdu_name: str = None, namespace: str = None, ): + # init env, paths + paths, env = self._init_paths_env( + cluster_name=cluster_id, create_if_not_exist=True + ) + # params to str params_str, file_to_delete = self._params_to_file_option( cluster_id=cluster_id, params=params @@ -344,7 +355,14 @@ class K8sHelmBaseConnector(K8sConnector): kdu_model = parts[0] command = self._get_install_command( - kdu_model, kdu_instance, namespace, params_str, version, atomic, timeout + kdu_model, + kdu_instance, + namespace, + params_str, + version, + atomic, + timeout, + paths["kube_config"], ) self.log.debug("installing: {}".format(command)) @@ -429,6 +447,9 @@ class K8sHelmBaseConnector(K8sConnector): cluster_name=cluster_id, create_if_not_exist=True ) + # sync local dir + self.fs.sync(from_path=cluster_id) + # params to str params_str, file_to_delete = self._params_to_file_option( cluster_id=cluster_id, params=params @@ -450,6 +471,7 @@ class K8sHelmBaseConnector(K8sConnector): version, atomic, timeout, + paths["kube_config"], ) self.log.debug("upgrading: {}".format(command)) @@ -563,8 +585,11 @@ class K8sHelmBaseConnector(K8sConnector): cluster_name=cluster_id, create_if_not_exist=True ) + # sync local dir + self.fs.sync(from_path=cluster_id) + command = self._get_rollback_command( - kdu_instance, instance_info["namespace"], revision + kdu_instance, instance_info["namespace"], revision, paths["kube_config"] ) self.log.debug("rolling_back: {}".format(command)) @@ -648,14 +673,19 @@ class K8sHelmBaseConnector(K8sConnector): # look for instance to obtain namespace instance_info = await self.get_instance_info(cluster_uuid, kdu_instance) if not instance_info: - raise K8sException("kdu_instance {} not found".format(kdu_instance)) - + self.log.warning(("kdu_instance {} not found".format(kdu_instance))) + return True # init env, paths paths, env = self._init_paths_env( cluster_name=cluster_id, create_if_not_exist=True ) - command = self._get_uninstall_command(kdu_instance, instance_info["namespace"]) + # sync local dir + self.fs.sync(from_path=cluster_id) + + command = self._get_uninstall_command( + kdu_instance, instance_info["namespace"], paths["kube_config"] + ) output, _rc = await self._local_async_exec( command=command, raise_exception_on_error=True, env=env ) @@ -748,11 +778,18 @@ class K8sHelmBaseConnector(K8sConnector): ) ) + # init env, paths + paths, env = self._init_paths_env( + cluster_name=cluster_id, create_if_not_exist=True + ) + # sync local dir self.fs.sync(from_path=cluster_id) # get list of services names for kdu - service_names = await self._get_services(cluster_id, kdu_instance, namespace) + service_names = await self._get_services( + cluster_id, kdu_instance, namespace, paths["kube_config"] + ) service_list = [] for service in service_names: @@ -972,7 +1009,7 @@ class K8sHelmBaseConnector(K8sConnector): """ @abc.abstractmethod - async def _get_services(self, cluster_id, kdu_instance, namespace): + async def _get_services(self, cluster_id, kdu_instance, namespace, kubeconfig): """ Implements the helm version dependent method to obtain services from a helm instance """ @@ -992,7 +1029,15 @@ class K8sHelmBaseConnector(K8sConnector): @abc.abstractmethod def _get_install_command( - self, kdu_model, kdu_instance, namespace, params_str, version, atomic, timeout + self, + kdu_model, + kdu_instance, + namespace, + params_str, + version, + atomic, + timeout, + kubeconfig, ) -> str: """ Obtain command to be executed to delete the indicated instance @@ -1000,20 +1045,32 @@ class K8sHelmBaseConnector(K8sConnector): @abc.abstractmethod def _get_upgrade_command( - self, kdu_model, kdu_instance, namespace, params_str, version, atomic, timeout + self, + kdu_model, + kdu_instance, + namespace, + params_str, + version, + atomic, + timeout, + kubeconfig, ) -> str: """ Obtain command to be executed to upgrade the indicated instance """ @abc.abstractmethod - def _get_rollback_command(self, kdu_instance, namespace, revision) -> str: + def _get_rollback_command( + self, kdu_instance, namespace, revision, kubeconfig + ) -> str: """ Obtain command to be executed to rollback the indicated instance """ @abc.abstractmethod - def _get_uninstall_command(self, kdu_instance: str, namespace: str) -> str: + def _get_uninstall_command( + self, kdu_instance: str, namespace: str, kubeconfig: str + ) -> str: """ Obtain command to be executed to delete the indicated instance """ @@ -1137,27 +1194,12 @@ class K8sHelmBaseConnector(K8sConnector): of dictionaries """ new_list = [] - for dictionary in input_list: - new_dict = dict((k.lower(), v) for k, v in dictionary.items()) - new_list.append(new_dict) + if input_list: + for dictionary in input_list: + new_dict = dict((k.lower(), v) for k, v in dictionary.items()) + new_list.append(new_dict) return new_list - def _local_exec(self, command: str) -> (str, int): - command = self._remove_multiple_spaces(command) - self.log.debug("Executing sync local command: {}".format(command)) - # raise exception if fails - output = "" - try: - output = subprocess.check_output( - command, shell=True, universal_newlines=True - ) - return_code = 0 - self.log.debug(output) - except Exception: - return_code = 1 - - return output, return_code - async def _local_async_exec( self, command: str,