X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=n2vc%2Fk8s_helm_base_conn.py;h=149b064ba5ed57fb5fbccac5ae7705eba074d07f;hb=60a3a96717d7c36ba7a65573da59a6bc039f5e28;hp=b6aa7e9683628d4c57f1627aa3a1588b37a32dfb;hpb=2a10e43076ffb8cf44a5a1a617028a56d8e1e6e2;p=osm%2FN2VC.git diff --git a/n2vc/k8s_helm_base_conn.py b/n2vc/k8s_helm_base_conn.py index b6aa7e9..149b064 100644 --- a/n2vc/k8s_helm_base_conn.py +++ b/n2vc/k8s_helm_base_conn.py @@ -186,22 +186,20 @@ class K8sHelmBaseConnector(K8sConnector): ) # sync fs - self.fs.reverse_sync(from_path=cluster_uuid) + self.fs.reverse_sync(from_path=cluster_id) - async def repo_update(self, cluster_uuid: str, name: str, repo_type: str = "chart"): + async def repo_update(self, cluster_id: str, name: str, repo_type: str = "chart"): self.log.debug( - "Cluster {}, updating {} repository {}".format( - cluster_uuid, repo_type, name - ) + "Cluster {}, updating {} repository {}".format(cluster_id, repo_type, name) ) # init_env paths, env = self._init_paths_env( - cluster_name=cluster_uuid, create_if_not_exist=True + cluster_name=cluster_id, create_if_not_exist=True ) # sync local dir - self.fs.sync(from_path=cluster_uuid) + self.fs.sync(from_path=cluster_id) # helm repo update command = "{} repo update {}".format(self._helm_command, name) @@ -211,7 +209,7 @@ class K8sHelmBaseConnector(K8sConnector): ) # sync fs - self.fs.reverse_sync(from_path=cluster_uuid) + self.fs.reverse_sync(from_path=cluster_id) async def repo_list(self, cluster_uuid: str) -> list: """ @@ -254,7 +252,6 @@ class K8sHelmBaseConnector(K8sConnector): return [] async def repo_remove(self, cluster_uuid: str, name: str): - _, cluster_id = self._get_namespace_cluster_id(cluster_uuid) self.log.debug("remove {} repositories for cluster {}".format(name, cluster_id)) @@ -382,7 +379,7 @@ class K8sHelmBaseConnector(K8sConnector): 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, @@ -413,7 +410,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=namespace, db_dict=db_dict, operation="install", - run_once=False, ) ) @@ -426,7 +422,6 @@ class K8sHelmBaseConnector(K8sConnector): output, rc = exec_task.result() else: - output, rc = await self._local_async_exec( command=command, raise_exception_on_error=False, env=env ) @@ -442,8 +437,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=namespace, db_dict=db_dict, operation="install", - run_once=True, - check_every=0, ) if rc != 0: @@ -495,7 +488,7 @@ class K8sHelmBaseConnector(K8sConnector): repo = self._split_repo(kdu_model) if repo: - self.repo_update(cluster_uuid, repo) + await self.repo_update(cluster_id, repo) command = self._get_upgrade_command( kdu_model, @@ -511,7 +504,6 @@ class K8sHelmBaseConnector(K8sConnector): self.log.debug("upgrading: {}".format(command)) if atomic: - # exec helm in a task exec_task = asyncio.ensure_future( coro_or_future=self._local_async_exec( @@ -526,7 +518,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="upgrade", - run_once=False, ) ) @@ -538,7 +529,6 @@ class K8sHelmBaseConnector(K8sConnector): output, rc = exec_task.result() else: - output, rc = await self._local_async_exec( command=command, raise_exception_on_error=False, env=env ) @@ -554,8 +544,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="upgrade", - run_once=True, - check_every=0, ) if rc != 0: @@ -598,7 +586,6 @@ class K8sHelmBaseConnector(K8sConnector): async def rollback( self, cluster_uuid: str, kdu_instance: str, revision=0, db_dict: dict = None ): - _, cluster_id = self._get_namespace_cluster_id(cluster_uuid) self.log.debug( "rollback kdu_instance {} to revision {} from cluster {}".format( @@ -642,7 +629,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="rollback", - run_once=False, ) ) @@ -661,8 +647,6 @@ class K8sHelmBaseConnector(K8sConnector): namespace=instance_info["namespace"], db_dict=db_dict, operation="rollback", - run_once=True, - check_every=0, ) if rc != 0: @@ -838,7 +822,6 @@ class K8sHelmBaseConnector(K8sConnector): async def get_service( self, cluster_uuid: str, service_name: str, namespace: str ) -> object: - self.log.debug( "get service, service_name: {}, namespace: {}, cluster_uuid: {}".format( service_name, namespace, cluster_uuid @@ -923,7 +906,6 @@ class K8sHelmBaseConnector(K8sConnector): return status async def values_kdu(self, kdu_model: str, repo_url: str = None) -> str: - self.log.debug( "inspect kdu_model values {} from (optional) repo: {}".format( kdu_model, repo_url @@ -935,7 +917,6 @@ class K8sHelmBaseConnector(K8sConnector): ) async def help_kdu(self, kdu_model: str, repo_url: str = None) -> str: - self.log.debug( "inspect kdu_model {} readme.md from repo: {}".format(kdu_model, repo_url) ) @@ -945,7 +926,6 @@ class K8sHelmBaseConnector(K8sConnector): ) async def synchronize_repos(self, cluster_uuid: str): - self.log.debug("synchronize repos for cluster helm-id: {}".format(cluster_uuid)) try: db_repo_ids = self._get_helm_chart_repos_ids(cluster_uuid) @@ -1246,7 +1226,6 @@ class K8sHelmBaseConnector(K8sConnector): encode_utf8: bool = False, env: dict = None, ) -> (str, int): - command = K8sHelmBaseConnector._remove_multiple_spaces(command) self.log.debug( "Executing async local command: {}, env: {}".format(command, env) @@ -1317,7 +1296,6 @@ class K8sHelmBaseConnector(K8sConnector): encode_utf8: bool = False, env: dict = None, ): - command1 = K8sHelmBaseConnector._remove_multiple_spaces(command1) command2 = K8sHelmBaseConnector._remove_multiple_spaces(command2) command = "{} | {}".format(command1, command2) @@ -1462,47 +1440,55 @@ 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) def _params_to_file_option(self, cluster_id: str, params: dict) -> (str, str): - if params and len(params) > 0: self._init_paths_env(cluster_name=cluster_id, create_if_not_exist=True) @@ -1581,7 +1567,7 @@ class K8sHelmBaseConnector(K8sConnector): name = name + get_random_number() return name.lower() - async def _split_repo(self, kdu_model: str) -> str: + def _split_repo(self, kdu_model: str) -> str: repo_name = None idx = kdu_model.find("/") if idx >= 0: