X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=n2vc%2Fk8s_helm_base_conn.py;h=b0d8e6d52624a034a9aa328e9870867d273fac88;hb=refs%2Fchanges%2F26%2F11926%2F3;hp=4a43ee214320cb786bfcbf62e6b003f6dad03c64;hpb=4395cfa6c7d0d80980c00d9f078440e0333fd826;p=osm%2FN2VC.git

diff --git a/n2vc/k8s_helm_base_conn.py b/n2vc/k8s_helm_base_conn.py
index 4a43ee2..b0d8e6d 100644
--- a/n2vc/k8s_helm_base_conn.py
+++ b/n2vc/k8s_helm_base_conn.py
@@ -21,12 +21,12 @@
 ##
 import abc
 import asyncio
+from typing import Union
 import random
 import time
 import shlex
 import shutil
 import stat
-import subprocess
 import os
 import yaml
 from uuid import uuid4
@@ -90,14 +90,18 @@ class K8sHelmBaseConnector(K8sConnector):
         if self._stable_repo_url == "None":
             self._stable_repo_url = None
 
-    @staticmethod
-    def _get_namespace_cluster_id(cluster_uuid: str) -> (str, str):
+    def _get_namespace(self, cluster_uuid: str) -> str:
         """
-        Parses cluster_uuid stored at database that can be either 'namespace:cluster_id' or only
-        cluster_id for backward compatibility
+        Obtains the namespace used by the cluster with the uuid passed by argument
+
+        param: cluster_uuid: cluster's uuid
         """
-        namespace, _, cluster_id = cluster_uuid.rpartition(":")
-        return namespace, cluster_id
+
+        # first, obtain the cluster corresponding to the uuid passed by argument
+        k8scluster = self.db.get_one(
+            "k8sclusters", q_filter={"_id": cluster_uuid}, fail_on_empty=False
+        )
+        return k8scluster.get("namespace")
 
     async def init_env(
         self,
@@ -121,11 +125,9 @@ class K8sHelmBaseConnector(K8sConnector):
         """
 
         if reuse_cluster_uuid:
-            namespace_, cluster_id = self._get_namespace_cluster_id(reuse_cluster_uuid)
-            namespace = namespace_ or namespace
+            cluster_id = reuse_cluster_uuid
         else:
             cluster_id = str(uuid4())
-        cluster_uuid = "{}:{}".format(namespace, cluster_id)
 
         self.log.debug(
             "Initializing K8S Cluster {}. namespace: {}".format(cluster_id, namespace)
@@ -147,42 +149,70 @@ class K8sHelmBaseConnector(K8sConnector):
 
         self.log.info("Cluster {} initialized".format(cluster_id))
 
-        return cluster_uuid, n2vc_installed_sw
+        return cluster_id, n2vc_installed_sw
 
     async def repo_add(
         self, cluster_uuid: str, name: str, url: str, repo_type: str = "chart"
     ):
-        _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
         self.log.debug(
             "Cluster {}, adding {} repository {}. URL: {}".format(
-                cluster_id, repo_type, name, url
+                cluster_uuid, repo_type, name, url
             )
         )
 
-        # 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
+            cluster_name=cluster_uuid, create_if_not_exist=True
+        )
+
+        # sync local dir
+        self.fs.sync(from_path=cluster_uuid)
+
+        # helm repo add 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
         )
 
         # helm repo update
-        command = "{} repo update".format(self._helm_command)
+        command = "env KUBECONFIG={} {} repo update {}".format(
+            paths["kube_config"], self._helm_command, name
+        )
         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)
-        self.log.debug("adding repo: {}".format(command))
+        # sync fs
+        self.fs.reverse_sync(from_path=cluster_uuid)
+
+    async def repo_update(self, cluster_uuid: str, name: str, repo_type: str = "chart"):
+        self.log.debug(
+            "Cluster {}, updating {} repository {}".format(
+                cluster_uuid, repo_type, name
+            )
+        )
+
+        # init_env
+        paths, env = self._init_paths_env(
+            cluster_name=cluster_uuid, create_if_not_exist=True
+        )
+
+        # sync local dir
+        self.fs.sync(from_path=cluster_uuid)
+
+        # helm repo update
+        command = "{} repo update {}".format(self._helm_command, name)
+        self.log.debug("updating repo: {}".format(command))
         await self._local_async_exec(
-            command=command, raise_exception_on_error=True, env=env
+            command=command, raise_exception_on_error=False, env=env
         )
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
     async def repo_list(self, cluster_uuid: str) -> list:
         """
@@ -191,18 +221,19 @@ class K8sHelmBaseConnector(K8sConnector):
         :return: list of registered repositories: [ (name, url) .... ]
         """
 
-        _, 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)
+        self.log.debug("list repositories for cluster {}".format(cluster_uuid))
 
         # config filename
         paths, env = self._init_paths_env(
-            cluster_name=cluster_id, create_if_not_exist=True
+            cluster_name=cluster_uuid, create_if_not_exist=True
         )
 
-        command = "{} repo list --output yaml".format(self._helm_command)
+        # sync local dir
+        self.fs.sync(from_path=cluster_uuid)
+
+        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(
@@ -210,7 +241,7 @@ class K8sHelmBaseConnector(K8sConnector):
         )
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
         if _rc == 0:
             if output and len(output) > 0:
@@ -223,25 +254,27 @@ 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))
-
-        # sync local dir
-        self.fs.sync(from_path=cluster_id)
+        self.log.debug(
+            "remove {} repositories for cluster {}".format(name, cluster_uuid)
+        )
 
         # init env, paths
         paths, env = self._init_paths_env(
-            cluster_name=cluster_id, create_if_not_exist=True
+            cluster_name=cluster_uuid, create_if_not_exist=True
         )
 
-        command = "{} repo remove {}".format(self._helm_command, name)
+        # sync local dir
+        self.fs.sync(from_path=cluster_uuid)
+
+        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
         )
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
     async def reset(
         self,
@@ -260,15 +293,15 @@ class K8sHelmBaseConnector(K8sConnector):
         :param kwargs: Additional parameters (None yet)
         :return: Returns True if successful or raises an exception.
         """
-        namespace, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
+        namespace = self._get_namespace(cluster_uuid=cluster_uuid)
         self.log.debug(
             "Resetting K8s environment. cluster uuid: {} uninstall={}".format(
-                cluster_id, uninstall_sw
+                cluster_uuid, uninstall_sw
             )
         )
 
         # sync local dir
-        self.fs.sync(from_path=cluster_id)
+        self.fs.sync(from_path=cluster_uuid)
 
         # uninstall releases if needed.
         if uninstall_sw:
@@ -297,20 +330,20 @@ class K8sHelmBaseConnector(K8sConnector):
                 else:
                     msg = (
                         "Cluster uuid: {} has releases and not force. Leaving K8s helm environment"
-                    ).format(cluster_id)
+                    ).format(cluster_uuid)
                     self.log.warn(msg)
                     uninstall_sw = (
                         False  # Allow to remove k8s cluster without removing Tiller
                     )
 
         if uninstall_sw:
-            await self._uninstall_sw(cluster_id, namespace)
+            await self._uninstall_sw(cluster_id=cluster_uuid, namespace=namespace)
 
         # delete cluster directory
-        self.log.debug("Removing directory {}".format(cluster_id))
-        self.fs.file_delete(cluster_id, ignore_non_exist=True)
+        self.log.debug("Removing directory {}".format(cluster_uuid))
+        self.fs.file_delete(cluster_uuid, ignore_non_exist=True)
         # Remove also local directorio if still exist
-        direct = self.fs.path + "/" + cluster_id
+        direct = self.fs.path + "/" + cluster_uuid
         shutil.rmtree(direct, ignore_errors=True)
 
         return True
@@ -329,6 +362,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
@@ -342,8 +380,19 @@ class K8sHelmBaseConnector(K8sConnector):
                 version = str(parts[1])
                 kdu_model = parts[0]
 
+        repo = self._split_repo(kdu_model)
+        if repo:
+            self.repo_update(cluster_id, repo)
+
         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))
@@ -364,7 +413,6 @@ class K8sHelmBaseConnector(K8sConnector):
                     namespace=namespace,
                     db_dict=db_dict,
                     operation="install",
-                    run_once=False,
                 )
             )
 
@@ -393,8 +441,6 @@ class K8sHelmBaseConnector(K8sConnector):
             namespace=namespace,
             db_dict=db_dict,
             operation="install",
-            run_once=True,
-            check_every=0,
         )
 
         if rc != 0:
@@ -412,11 +458,10 @@ class K8sHelmBaseConnector(K8sConnector):
         params: dict = None,
         db_dict: dict = None,
     ):
-        _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
-        self.log.debug("upgrading {} in cluster {}".format(kdu_model, cluster_id))
+        self.log.debug("upgrading {} in cluster {}".format(kdu_model, cluster_uuid))
 
         # sync local dir
-        self.fs.sync(from_path=cluster_id)
+        self.fs.sync(from_path=cluster_uuid)
 
         # look for instance to obtain namespace
         instance_info = await self.get_instance_info(cluster_uuid, kdu_instance)
@@ -425,12 +470,15 @@ class K8sHelmBaseConnector(K8sConnector):
 
         # init env, paths
         paths, env = self._init_paths_env(
-            cluster_name=cluster_id, create_if_not_exist=True
+            cluster_name=cluster_uuid, create_if_not_exist=True
         )
 
+        # sync local dir
+        self.fs.sync(from_path=cluster_uuid)
+
         # params to str
         params_str, file_to_delete = self._params_to_file_option(
-            cluster_id=cluster_id, params=params
+            cluster_id=cluster_uuid, params=params
         )
 
         # version
@@ -441,6 +489,10 @@ class K8sHelmBaseConnector(K8sConnector):
                 version = str(parts[1])
                 kdu_model = parts[0]
 
+        repo = self._split_repo(kdu_model)
+        if repo:
+            self.repo_update(cluster_uuid, repo)
+
         command = self._get_upgrade_command(
             kdu_model,
             kdu_instance,
@@ -449,6 +501,7 @@ class K8sHelmBaseConnector(K8sConnector):
             version,
             atomic,
             timeout,
+            paths["kube_config"],
         )
 
         self.log.debug("upgrading: {}".format(command))
@@ -464,12 +517,11 @@ class K8sHelmBaseConnector(K8sConnector):
             # write status in another task
             status_task = asyncio.ensure_future(
                 coro_or_future=self._store_status(
-                    cluster_id=cluster_id,
+                    cluster_id=cluster_uuid,
                     kdu_instance=kdu_instance,
                     namespace=instance_info["namespace"],
                     db_dict=db_dict,
                     operation="upgrade",
-                    run_once=False,
                 )
             )
 
@@ -492,13 +544,11 @@ class K8sHelmBaseConnector(K8sConnector):
 
         # write final status
         await self._store_status(
-            cluster_id=cluster_id,
+            cluster_id=cluster_uuid,
             kdu_instance=kdu_instance,
             namespace=instance_info["namespace"],
             db_dict=db_dict,
             operation="upgrade",
-            run_once=True,
-            check_every=0,
         )
 
         if rc != 0:
@@ -507,7 +557,7 @@ class K8sHelmBaseConnector(K8sConnector):
             raise K8sException(msg)
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
         # return new revision number
         instance = await self.get_instance_info(
@@ -541,16 +591,14 @@ 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(
-                kdu_instance, revision, cluster_id
+                kdu_instance, revision, cluster_uuid
             )
         )
 
         # sync local dir
-        self.fs.sync(from_path=cluster_id)
+        self.fs.sync(from_path=cluster_uuid)
 
         # look for instance to obtain namespace
         instance_info = await self.get_instance_info(cluster_uuid, kdu_instance)
@@ -559,11 +607,14 @@ class K8sHelmBaseConnector(K8sConnector):
 
         # init env, paths
         paths, env = self._init_paths_env(
-            cluster_name=cluster_id, create_if_not_exist=True
+            cluster_name=cluster_uuid, create_if_not_exist=True
         )
 
+        # sync local dir
+        self.fs.sync(from_path=cluster_uuid)
+
         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))
@@ -577,12 +628,11 @@ class K8sHelmBaseConnector(K8sConnector):
         # write status in another task
         status_task = asyncio.ensure_future(
             coro_or_future=self._store_status(
-                cluster_id=cluster_id,
+                cluster_id=cluster_uuid,
                 kdu_instance=kdu_instance,
                 namespace=instance_info["namespace"],
                 db_dict=db_dict,
                 operation="rollback",
-                run_once=False,
             )
         )
 
@@ -596,13 +646,11 @@ class K8sHelmBaseConnector(K8sConnector):
 
         # write final status
         await self._store_status(
-            cluster_id=cluster_id,
+            cluster_id=cluster_uuid,
             kdu_instance=kdu_instance,
             namespace=instance_info["namespace"],
             db_dict=db_dict,
             operation="rollback",
-            run_once=True,
-            check_every=0,
         )
 
         if rc != 0:
@@ -611,7 +659,7 @@ class K8sHelmBaseConnector(K8sConnector):
             raise K8sException(msg)
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
         # return new revision number
         instance = await self.get_instance_info(
@@ -636,31 +684,37 @@ class K8sHelmBaseConnector(K8sConnector):
         :return: True if successful
         """
 
-        _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
         self.log.debug(
-            "uninstall kdu_instance {} from cluster {}".format(kdu_instance, cluster_id)
+            "uninstall kdu_instance {} from cluster {}".format(
+                kdu_instance, cluster_uuid
+            )
         )
 
         # sync local dir
-        self.fs.sync(from_path=cluster_id)
+        self.fs.sync(from_path=cluster_uuid)
 
         # 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
+            cluster_name=cluster_uuid, create_if_not_exist=True
         )
 
-        command = self._get_uninstall_command(kdu_instance, instance_info["namespace"])
+        # sync local dir
+        self.fs.sync(from_path=cluster_uuid)
+
+        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
         )
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
         return self._output_to_table(output)
 
@@ -672,17 +726,16 @@ class K8sHelmBaseConnector(K8sConnector):
         :return:
         """
 
-        _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
-        self.log.debug("list releases for cluster {}".format(cluster_id))
+        self.log.debug("list releases for cluster {}".format(cluster_uuid))
 
         # sync local dir
-        self.fs.sync(from_path=cluster_id)
+        self.fs.sync(from_path=cluster_uuid)
 
         # execute internal command
-        result = await self._instances_list(cluster_id)
+        result = await self._instances_list(cluster_uuid)
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
         return result
 
@@ -740,26 +793,32 @@ class K8sHelmBaseConnector(K8sConnector):
         - `external_ip` List of external ips (in case they are available)
         """
 
-        _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
         self.log.debug(
             "get_services: cluster_uuid: {}, kdu_instance: {}".format(
                 cluster_uuid, kdu_instance
             )
         )
 
+        # init env, paths
+        paths, env = self._init_paths_env(
+            cluster_name=cluster_uuid, create_if_not_exist=True
+        )
+
         # sync local dir
-        self.fs.sync(from_path=cluster_id)
+        self.fs.sync(from_path=cluster_uuid)
 
         # 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_uuid, kdu_instance, namespace, paths["kube_config"]
+        )
 
         service_list = []
         for service in service_names:
-            service = await self._get_service(cluster_id, service, namespace)
+            service = await self._get_service(cluster_uuid, service, namespace)
             service_list.append(service)
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
         return service_list
 
@@ -773,19 +832,19 @@ class K8sHelmBaseConnector(K8sConnector):
             )
         )
 
-        _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
-
         # sync local dir
-        self.fs.sync(from_path=cluster_id)
+        self.fs.sync(from_path=cluster_uuid)
 
-        service = await self._get_service(cluster_id, service_name, namespace)
+        service = await self._get_service(cluster_uuid, service_name, namespace)
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
         return service
 
-    async def status_kdu(self, cluster_uuid: str, kdu_instance: str, **kwargs) -> str:
+    async def status_kdu(
+        self, cluster_uuid: str, kdu_instance: str, yaml_format: str = False, **kwargs
+    ) -> Union[str, dict]:
         """
         This call would retrieve tha current state of a given KDU instance. It would be
         would allow to retrieve the _composition_ (i.e. K8s objects) and _specific
@@ -795,6 +854,8 @@ class K8sHelmBaseConnector(K8sConnector):
         :param cluster_uuid: UUID of a K8s cluster known by OSM
         :param kdu_instance: unique name for the KDU instance
         :param kwargs: Additional parameters (None yet)
+        :param yaml_format: if the return shall be returned as an YAML string or as a
+                                dictionary
         :return: If successful, it will return the following vector of arguments:
         - K8s `namespace` in the cluster where the KDU lives
         - `state` of the KDU instance. It can be:
@@ -815,13 +876,11 @@ class K8sHelmBaseConnector(K8sConnector):
             )
         )
 
-        _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
-
         # sync local dir
-        self.fs.sync(from_path=cluster_id)
+        self.fs.sync(from_path=cluster_uuid)
 
         # get instance: needed to obtain namespace
-        instances = await self._instances_list(cluster_id=cluster_id)
+        instances = await self._instances_list(cluster_id=cluster_uuid)
         for instance in instances:
             if instance.get("name") == kdu_instance:
                 break
@@ -829,20 +888,20 @@ class K8sHelmBaseConnector(K8sConnector):
             # instance does not exist
             raise K8sException(
                 "Instance name: {} not found in cluster: {}".format(
-                    kdu_instance, cluster_id
+                    kdu_instance, cluster_uuid
                 )
             )
 
         status = await self._status_kdu(
-            cluster_id=cluster_id,
+            cluster_id=cluster_uuid,
             kdu_instance=kdu_instance,
             namespace=instance["namespace"],
+            yaml_format=yaml_format,
             show_error_log=True,
-            return_text=True,
         )
 
         # sync fs
-        self.fs.reverse_sync(from_path=cluster_id)
+        self.fs.reverse_sync(from_path=cluster_uuid)
 
         return status
 
@@ -971,7 +1030,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
         """
@@ -982,16 +1041,24 @@ class K8sHelmBaseConnector(K8sConnector):
         cluster_id: str,
         kdu_instance: str,
         namespace: str = None,
+        yaml_format: bool = False,
         show_error_log: bool = False,
-        return_text: bool = False,
-    ):
+    ) -> Union[str, dict]:
         """
         Implements the helm version dependent method to obtain status of a helm instance
         """
 
     @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
@@ -999,20 +1066,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
         """
@@ -1142,22 +1221,6 @@ class K8sHelmBaseConnector(K8sConnector):
                 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,
@@ -1382,42 +1445,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,
-                    namespace=namespace,
-                    return_text=False,
-                )
-                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)
@@ -1500,3 +1572,30 @@ class K8sHelmBaseConnector(K8sConnector):
 
         name = name + get_random_number()
         return name.lower()
+
+    def _split_version(self, kdu_model: str) -> (str, str):
+        version = None
+        if ":" 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:
+        repo_name = None
+        idx = kdu_model.find("/")
+        if idx >= 0:
+            repo_name = kdu_model[:idx]
+        return repo_name
+
+    async def _find_repo(self, kdu_model: str, cluster_uuid: str) -> str:
+        repo_url = None
+        idx = kdu_model.find("/")
+        if idx >= 0:
+            repo_name = kdu_model[:idx]
+            # 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
+        return repo_url