X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=n2vc%2Fk8s_helm_base_conn.py;h=6ddf1186b72a41030bea9af97d903ff5ded33125;hb=0631399f81ccd9823aa611911ecf2fd1cbcd4c0c;hp=952630a5e9f11b3a4e0de2e359246cfa7d41323b;hpb=8070c3c8260010f052ee9fe546c85bed4aa6b2eb;p=osm%2FN2VC.git diff --git a/n2vc/k8s_helm_base_conn.py b/n2vc/k8s_helm_base_conn.py index 952630a..6ddf118 100644 --- a/n2vc/k8s_helm_base_conn.py +++ b/n2vc/k8s_helm_base_conn.py @@ -90,6 +90,9 @@ class K8sHelmBaseConnector(K8sConnector): if self._stable_repo_url == "None": self._stable_repo_url = None + # Lock to avoid concurrent execution of helm commands + self.cmd_lock = asyncio.Lock() + def _get_namespace(self, cluster_uuid: str) -> str: """ Obtains the namespace used by the cluster with the uuid passed by argument @@ -371,6 +374,9 @@ class K8sHelmBaseConnector(K8sConnector): return True + def _is_helm_chart_a_file(self, chart_name: str): + return chart_name.count("/") > 1 + async def _install_impl( self, cluster_id: str, @@ -398,9 +404,9 @@ class K8sHelmBaseConnector(K8sConnector): # version kdu_model, version = self._split_version(kdu_model) - repo = self._split_repo(kdu_model) + _, repo = self._split_repo(kdu_model) if repo: - self.repo_update(cluster_id, repo) + await self.repo_update(cluster_id, repo) command = self._get_install_command( kdu_model, @@ -431,7 +437,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=namespace, db_dict=db_dict, operation="install", - run_once=False, ) ) @@ -460,8 +465,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=namespace, db_dict=db_dict, operation="install", - run_once=True, - check_every=0, ) if rc != 0: @@ -505,9 +508,9 @@ class K8sHelmBaseConnector(K8sConnector): # version kdu_model, version = self._split_version(kdu_model) - repo = self._split_repo(kdu_model) + _, repo = self._split_repo(kdu_model) if repo: - self.repo_update(cluster_uuid, repo) + await self.repo_update(cluster_uuid, repo) command = self._get_upgrade_command( kdu_model, @@ -538,7 +541,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="upgrade", - run_once=False, ) ) @@ -566,8 +568,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="upgrade", - run_once=True, - check_every=0, ) if rc != 0: @@ -642,10 +642,6 @@ class K8sHelmBaseConnector(K8sConnector): kdu_model, version = self._split_version(kdu_model) repo_url = await self._find_repo(kdu_model, cluster_uuid) - if not repo_url: - raise K8sException( - "Repository not found for kdu_model {}".format(kdu_model) - ) _, replica_str = await self._get_replica_count_url( kdu_model, repo_url, resource_name @@ -681,7 +677,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="scale", - run_once=False, ) ) @@ -704,8 +699,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="scale", - run_once=True, - check_every=0, ) if rc != 0: @@ -732,7 +725,7 @@ class K8sHelmBaseConnector(K8sConnector): cluster_uuid: The UUID of the cluster resource_name: Resource name kdu_instance: KDU instance name - kdu_model: The name or path of a bundle + kdu_model: The name or path of an Helm Chart kwargs: Additional parameters Returns: @@ -749,30 +742,42 @@ class K8sHelmBaseConnector(K8sConnector): raise K8sException("kdu_instance {} not found".format(kdu_instance)) # init env, paths - paths, env = self._init_paths_env( + paths, _ = self._init_paths_env( cluster_name=cluster_uuid, create_if_not_exist=True ) replicas = await self._get_replica_count_instance( - kdu_instance, instance_info["namespace"], paths["kube_config"] + kdu_instance=kdu_instance, + namespace=instance_info["namespace"], + kubeconfig=paths["kube_config"], + resource_name=resource_name, ) - # Get default value if scale count is not found from provided values - if not replicas: - repo_url = await self._find_repo(kdu_model, cluster_uuid) - if not repo_url: - raise K8sException( - "Repository not found for kdu_model {}".format(kdu_model) - ) + self.log.debug( + f"Number of replicas of the KDU instance {kdu_instance} and resource {resource_name} obtained: {replicas}" + ) + # Get default value if scale count is not found from provided values + # Important note: this piece of code shall only be executed in the first scaling operation, + # since it is expected that the _get_replica_count_instance is able to obtain the number of + # replicas when a scale operation was already conducted previously for this KDU/resource! + if replicas is None: + repo_url = await self._find_repo( + kdu_model=kdu_model, cluster_uuid=cluster_uuid + ) replicas, _ = await self._get_replica_count_url( - kdu_model, repo_url, resource_name + kdu_model=kdu_model, repo_url=repo_url, resource_name=resource_name ) - if not replicas: - msg = "Replica count not found. Cannot be scaled" - self.log.error(msg) - raise K8sException(msg) + self.log.debug( + f"Number of replicas of the Helm Chart package for KDU instance {kdu_instance} and resource " + f"{resource_name} obtained: {replicas}" + ) + + if replicas is None: + msg = "Replica count not found. Cannot be scaled" + self.log.error(msg) + raise K8sException(msg) return int(replicas) @@ -821,7 +826,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="rollback", - run_once=False, ) ) @@ -840,8 +844,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="rollback", - run_once=True, - check_every=0, ) if rc != 0: @@ -1132,6 +1134,15 @@ class K8sHelmBaseConnector(K8sConnector): ) async def values_kdu(self, kdu_model: str, repo_url: str = None) -> str: + """Method to obtain the Helm Chart package's values + + Args: + kdu_model: The name or path of an Helm Chart + repo_url: Helm Chart repository url + + Returns: + str: the values of the Helm Chart package + """ self.log.debug( "inspect kdu_model values {} from (optional) repo: {}".format( @@ -1352,8 +1363,17 @@ class K8sHelmBaseConnector(K8sConnector): def _get_inspect_command( self, show_command: str, kdu_model: str, repo_str: str, version: str ): - """ - Obtain command to be executed to obtain information about the kdu + """Generates the command to obtain the information about an Helm Chart package + (´helm show ...´ command) + + Args: + show_command: the second part of the command (`helm show `) + kdu_model: The name or path of an Helm Chart + repo_url: Helm Chart repository url + version: constraint with specific version of the Chart to use + + Returns: + str: the generated Helm Chart command """ @abc.abstractmethod @@ -1501,17 +1521,18 @@ class K8sHelmBaseConnector(K8sConnector): environ.update(env) try: - process = await asyncio.create_subprocess_exec( - *command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - env=environ, - ) + async with self.cmd_lock: + process = await asyncio.create_subprocess_exec( + *command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=environ, + ) - # wait for command terminate - stdout, stderr = await process.communicate() + # wait for command terminate + stdout, stderr = await process.communicate() - return_code = process.returncode + return_code = process.returncode output = "" if stdout: @@ -1538,6 +1559,9 @@ class K8sHelmBaseConnector(K8sConnector): return output, return_code except asyncio.CancelledError: + # first, kill the process if it is still running + if process.returncode is None: + process.kill() raise except K8sException: raise @@ -1575,16 +1599,19 @@ class K8sHelmBaseConnector(K8sConnector): environ.update(env) try: - read, write = os.pipe() - await asyncio.create_subprocess_exec(*command1, stdout=write, env=environ) - os.close(write) - process_2 = await asyncio.create_subprocess_exec( - *command2, stdin=read, stdout=asyncio.subprocess.PIPE, env=environ - ) - os.close(read) - stdout, stderr = await process_2.communicate() + async with self.cmd_lock: + read, write = os.pipe() + process_1 = await asyncio.create_subprocess_exec( + *command1, stdout=write, env=environ + ) + os.close(write) + process_2 = await asyncio.create_subprocess_exec( + *command2, stdin=read, stdout=asyncio.subprocess.PIPE, env=environ + ) + os.close(read) + stdout, stderr = await process_2.communicate() - return_code = process_2.returncode + return_code = process_2.returncode output = "" if stdout: @@ -1610,6 +1637,10 @@ class K8sHelmBaseConnector(K8sConnector): return output, return_code except asyncio.CancelledError: + # first, kill the processes if they are still running + for process in (process_1, process_2): + if process.returncode is None: + process.kill() raise except K8sException: raise @@ -1681,16 +1712,23 @@ class K8sHelmBaseConnector(K8sConnector): async def _exec_inspect_command( self, inspect_command: str, kdu_model: str, repo_url: str = None ): - """Obtains information about a kdu, no cluster (no env).""" + """Obtains information about an Helm Chart package (´helm show´ command) + + Args: + inspect_command: the Helm sub command (`helm show ...`) + kdu_model: The name or path of an Helm Chart + repo_url: Helm Chart repository url + + Returns: + str: the requested info about the Helm Chart package + """ repo_str = "" if repo_url: repo_str = " --repo {}".format(repo_url) - idx = kdu_model.find("/") - if idx >= 0: - idx += 1 - kdu_model = kdu_model[idx:] + # Obtain the Chart's name and store it in the var kdu_model + kdu_model, _ = self._split_repo(kdu_model=kdu_model) kdu_model, version = self._split_version(kdu_model) if version: @@ -1699,34 +1737,42 @@ class K8sHelmBaseConnector(K8sConnector): version_str = "" full_command = self._get_inspect_command( - inspect_command, kdu_model, repo_str, version_str + show_command=inspect_command, + kdu_model=kdu_model, + repo_str=repo_str, + version=version_str, ) - output, _rc = await self._local_async_exec(command=full_command) + output, _ = await self._local_async_exec(command=full_command) return output async def _get_replica_count_url( self, kdu_model: str, - repo_url: str, + repo_url: str = None, resource_name: str = None, - ): + ) -> (int, str): """Get the replica count value in the Helm Chart Values. Args: - kdu_model: The name or path of a bundle + kdu_model: The name or path of an Helm Chart repo_url: Helm Chart repository url resource_name: Resource name Returns: - True if replicas, False replicaCount + A tuple with: + - The number of replicas of the specific instance; if not found, returns None; and + - The string corresponding to the replica count key in the Helm values """ kdu_values = yaml.load( - await self.values_kdu(kdu_model, repo_url), Loader=yaml.SafeLoader + await self.values_kdu(kdu_model=kdu_model, repo_url=repo_url), + Loader=yaml.SafeLoader, ) + self.log.debug(f"Obtained the Helm package values for the KDU: {kdu_values}") + if not kdu_values: raise K8sException( "kdu_values not found for kdu_model {}".format(kdu_model) @@ -1747,10 +1793,10 @@ class K8sHelmBaseConnector(K8sConnector): replica_str = "" replicas = None - if kdu_values.get("replicaCount", None): + if kdu_values.get("replicaCount") is not None: replicas = kdu_values["replicaCount"] replica_str = "replicaCount" - elif kdu_values.get("replicas", None): + elif kdu_values.get("replicas") is not None: duplicate_check = True replicas = kdu_values["replicas"] replica_str = "replicas" @@ -1789,7 +1835,7 @@ class K8sHelmBaseConnector(K8sConnector): namespace: str, kubeconfig: str, resource_name: str = None, - ): + ) -> int: """Get the replica count value in the instance. Args: @@ -1799,7 +1845,7 @@ class K8sHelmBaseConnector(K8sConnector): resource_name: Resource name Returns: - True if replicas, False replicaCount + The number of replicas of the specific instance; if not found, returns None """ kdu_values = yaml.load( @@ -1807,23 +1853,23 @@ class K8sHelmBaseConnector(K8sConnector): Loader=yaml.SafeLoader, ) + self.log.debug(f"Obtained the Helm values for the KDU instance: {kdu_values}") + replicas = None if kdu_values: resource_values = ( kdu_values.get(resource_name, None) if resource_name else None ) - replicas = ( - ( - resource_values.get("replicaCount", None) - or resource_values.get("replicas", None) - ) - if resource_values - else ( - kdu_values.get("replicaCount", None) - or kdu_values.get("replicas", None) - ) - ) + + for replica_str in ("replicaCount", "replicas"): + if resource_values: + replicas = resource_values.get(replica_str) + else: + replicas = kdu_values.get(replica_str) + + if replicas is not None: + break return replicas @@ -1833,42 +1879,51 @@ class K8sHelmBaseConnector(K8sConnector): operation: str, kdu_instance: str, namespace: str = None, - check_every: float = 10, db_dict: dict = None, - run_once: bool = False, - ): - while True: - try: - await asyncio.sleep(check_every) - detailed_status = await self._status_kdu( - cluster_id=cluster_id, - kdu_instance=kdu_instance, - yaml_format=False, - namespace=namespace, - ) - status = detailed_status.get("info").get("description") - self.log.debug("KDU {} STATUS: {}.".format(kdu_instance, status)) - # write status to db - result = await self.write_app_status_to_db( - db_dict=db_dict, - status=str(status), - detailed_status=str(detailed_status), - operation=operation, - ) - if not result: - self.log.info("Error writing in database. Task exiting...") - return - except asyncio.CancelledError: - self.log.debug("Task cancelled") - return - except Exception as e: - self.log.debug( - "_store_status exception: {}".format(str(e)), exc_info=True - ) - pass - finally: - if run_once: - return + ) -> None: + """ + Obtains the status of the KDU instance based on Helm Charts, and stores it in the database. + + :param cluster_id (str): the cluster where the KDU instance is deployed + :param operation (str): The operation related to the status to be updated (for instance, "install" or "upgrade") + :param kdu_instance (str): The KDU instance in relation to which the status is obtained + :param namespace (str): The Kubernetes namespace where the KDU instance was deployed. Defaults to None + :param db_dict (dict): A dictionary with the database necessary information. It shall contain the + values for the keys: + - "collection": The Mongo DB collection to write to + - "filter": The query filter to use in the update process + - "path": The dot separated keys which targets the object to be updated + Defaults to None. + """ + + try: + detailed_status = await self._status_kdu( + cluster_id=cluster_id, + kdu_instance=kdu_instance, + yaml_format=False, + namespace=namespace, + ) + + status = detailed_status.get("info").get("description") + self.log.debug(f"Status for KDU {kdu_instance} obtained: {status}.") + + # write status to db + result = await self.write_app_status_to_db( + db_dict=db_dict, + status=str(status), + detailed_status=str(detailed_status), + operation=operation, + ) + + if not result: + self.log.info("Error writing in database. Task exiting...") + + except asyncio.CancelledError as e: + self.log.warning( + f"Exception in method {self._store_status.__name__} (task cancelled): {e}" + ) + except Exception as e: + self.log.warning(f"Exception in method {self._store_status.__name__}: {e}") # params for use in -f file # returns values file option and filename (in order to delete it at the end) @@ -1888,7 +1943,7 @@ class K8sHelmBaseConnector(K8sConnector): for key in params: value = params.get(key) if "!!yaml" in str(value): - value = yaml.load(value[7:]) + value = yaml.safe_load(value[7:]) params2[key] = value values_file = get_random_number() + ".yaml" @@ -1954,27 +2009,55 @@ class K8sHelmBaseConnector(K8sConnector): def _split_version(self, kdu_model: str) -> (str, str): version = None - if ":" in kdu_model: + if not self._is_helm_chart_a_file(kdu_model) and ":" in kdu_model: parts = kdu_model.split(sep=":") if len(parts) == 2: version = str(parts[1]) kdu_model = parts[0] return kdu_model, version - async def _split_repo(self, kdu_model: str) -> str: + def _split_repo(self, kdu_model: str) -> (str, str): + """Obtain the Helm Chart's repository and Chart's names from the KDU model + + Args: + kdu_model (str): Associated KDU model + + Returns: + (str, str): Tuple with the Chart name in index 0, and the repo name + in index 2; if there was a problem finding them, return None + for both + """ + + chart_name = None repo_name = None + idx = kdu_model.find("/") if idx >= 0: + chart_name = kdu_model[idx + 1 :] repo_name = kdu_model[:idx] - return repo_name + + return chart_name, repo_name async def _find_repo(self, kdu_model: str, cluster_uuid: str) -> str: + """Obtain the Helm repository for an Helm Chart + + Args: + kdu_model (str): the KDU model associated with the Helm Chart instantiation + cluster_uuid (str): The cluster UUID associated with the Helm Chart instantiation + + Returns: + str: the repository URL; if Helm Chart is a local one, the function returns None + """ + + _, repo_name = self._split_repo(kdu_model=kdu_model) + repo_url = None - idx = kdu_model.find("/") - if idx >= 0: - repo_name = kdu_model[:idx] + if repo_name: # Find repository link local_repo_list = await self.repo_list(cluster_uuid) for repo in local_repo_list: - repo_url = repo["url"] if repo["name"] == repo_name else None + if repo["name"] == repo_name: + repo_url = repo["url"] + break # it is not necessary to continue the loop if the repo link was found... + return repo_url