X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Fk8s_helm_base_conn.py;h=b3d7bda3c83f40812ccf03d2a196ea247099dc48;hp=0d001ee24208d57751bf5919256cc5d52ad9d05a;hb=82b591ceed704c798ead2d9104085a08e75b511b;hpb=eb8943a887e2fb8cce0240382811f9e504f3c7fb

diff --git a/n2vc/k8s_helm_base_conn.py b/n2vc/k8s_helm_base_conn.py
index 0d001ee..b3d7bda 100644
--- a/n2vc/k8s_helm_base_conn.py
+++ b/n2vc/k8s_helm_base_conn.py
@@ -42,6 +42,7 @@ class K8sHelmBaseConnector(K8sConnector):
     ################################### P U B L I C ####################################
     ####################################################################################
     """
+
     service_account = "osm"
     _STABLE_REPO_URL = "https://charts.helm.sh/stable"
 
@@ -96,11 +97,15 @@ class K8sHelmBaseConnector(K8sConnector):
         Parses cluster_uuid stored at database that can be either 'namespace:cluster_id' or only
         cluster_id for backward compatibility
         """
-        namespace, _, cluster_id = cluster_uuid.rpartition(':')
+        namespace, _, cluster_id = cluster_uuid.rpartition(":")
         return namespace, cluster_id
 
     async def init_env(
-        self, k8s_creds: str, namespace: str = "kube-system", reuse_cluster_uuid=None, **kwargs,
+        self,
+        k8s_creds: str,
+        namespace: str = "kube-system",
+        reuse_cluster_uuid=None,
+        **kwargs,
     ) -> (str, bool):
         """
         It prepares a given K8s cluster environment to run Charts
@@ -123,7 +128,9 @@ class K8sHelmBaseConnector(K8sConnector):
             cluster_id = str(uuid4())
         cluster_uuid = "{}:{}".format(namespace, cluster_id)
 
-        self.log.debug("Initializing K8S Cluster {}. namespace: {}".format(cluster_id, namespace))
+        self.log.debug(
+            "Initializing K8S Cluster {}. namespace: {}".format(cluster_id, namespace)
+        )
 
         paths, env = self._init_paths_env(
             cluster_name=cluster_id, create_if_not_exist=True
@@ -144,11 +151,14 @@ class K8sHelmBaseConnector(K8sConnector):
         return cluster_uuid, n2vc_installed_sw
 
     async def repo_add(
-            self, cluster_uuid: str, name: str, url: str, repo_type: str = "chart"
+        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))
+        self.log.debug(
+            "Cluster {}, adding {} repository {}. URL: {}".format(
+                cluster_id, repo_type, name, url
+            )
+        )
 
         # sync local dir
         self.fs.sync(from_path=cluster_id)
@@ -159,18 +169,18 @@ class K8sHelmBaseConnector(K8sConnector):
         )
 
         # helm repo update
-        command = "{} repo update".format(
-            self._helm_command
-        )
+        command = "{} repo update".format(self._helm_command)
         self.log.debug("updating repo: {}".format(command))
-        await self._local_async_exec(command=command, raise_exception_on_error=False, env=env)
+        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 = "{} repo add {} {}".format(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)
+        await self._local_async_exec(
+            command=command, raise_exception_on_error=True, env=env
+        )
 
         # sync fs
         self.fs.reverse_sync(from_path=cluster_id)
@@ -193,9 +203,7 @@ class K8sHelmBaseConnector(K8sConnector):
             cluster_name=cluster_id, create_if_not_exist=True
         )
 
-        command = "{} repo list --output yaml".format(
-            self._helm_command
-        )
+        command = "{} repo list --output yaml".format(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,16 +236,20 @@ class K8sHelmBaseConnector(K8sConnector):
             cluster_name=cluster_id, create_if_not_exist=True
         )
 
-        command = "{} repo remove {}".format(
-            self._helm_command, name
+        command = "{} repo remove {}".format(self._helm_command, name)
+        await self._local_async_exec(
+            command=command, raise_exception_on_error=True, env=env
         )
-        await self._local_async_exec(command=command, raise_exception_on_error=True, env=env)
 
         # sync fs
         self.fs.reverse_sync(from_path=cluster_id)
 
     async def reset(
-            self, cluster_uuid: str, force: bool = False, uninstall_sw: bool = False, **kwargs
+        self,
+        cluster_uuid: str,
+        force: bool = False,
+        uninstall_sw: bool = False,
+        **kwargs,
     ) -> bool:
         """Reset a cluster
 
@@ -250,8 +262,11 @@ class K8sHelmBaseConnector(K8sConnector):
         :return: Returns True if successful or raises an exception.
         """
         namespace, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
-        self.log.debug("Resetting K8s environment. cluster uuid: {} uninstall={}"
-                       .format(cluster_id, uninstall_sw))
+        self.log.debug(
+            "Resetting K8s environment. cluster uuid: {} uninstall={}".format(
+                cluster_id, uninstall_sw
+            )
+        )
 
         # sync local dir
         self.fs.sync(from_path=cluster_id)
@@ -276,14 +291,18 @@ class K8sHelmBaseConnector(K8sConnector):
                             # that in some cases of previously installed helm releases it
                             # raised an error
                             self.log.warn(
-                                "Error uninstalling release {}: {}".format(kdu_instance, e)
+                                "Error uninstalling release {}: {}".format(
+                                    kdu_instance, e
+                                )
                             )
                 else:
                     msg = (
                         "Cluster uuid: {} has releases and not force. Leaving K8s helm environment"
                     ).format(cluster_id)
                     self.log.warn(msg)
-                    uninstall_sw = False  # Allow to remove k8s cluster without removing Tiller
+                    uninstall_sw = (
+                        False  # Allow to remove k8s cluster without removing Tiller
+                    )
 
         if uninstall_sw:
             await self._uninstall_sw(cluster_id, namespace)
@@ -298,18 +317,18 @@ class K8sHelmBaseConnector(K8sConnector):
         return True
 
     async def _install_impl(
-            self,
-            cluster_id: str,
-            kdu_model: str,
-            paths: dict,
-            env: dict,
-            kdu_instance: str,
-            atomic: bool = True,
-            timeout: float = 300,
-            params: dict = None,
-            db_dict: dict = None,
-            kdu_name: str = None,
-            namespace: str = None,
+        self,
+        cluster_id: str,
+        kdu_model: str,
+        paths: dict,
+        env: dict,
+        kdu_instance: str,
+        atomic: bool = True,
+        timeout: float = 300,
+        params: dict = None,
+        db_dict: dict = None,
+        kdu_name: str = None,
+        namespace: str = None,
     ):
         # params to str
         params_str, file_to_delete = self._params_to_file_option(
@@ -324,8 +343,9 @@ class K8sHelmBaseConnector(K8sConnector):
                 version = str(parts[1])
                 kdu_model = parts[0]
 
-        command = self._get_install_command(kdu_model, kdu_instance, namespace,
-                                            params_str, version, atomic, timeout)
+        command = self._get_install_command(
+            kdu_model, kdu_instance, namespace, params_str, version, atomic, timeout
+        )
 
         self.log.debug("installing: {}".format(command))
 
@@ -422,8 +442,15 @@ class K8sHelmBaseConnector(K8sConnector):
                 version = str(parts[1])
                 kdu_model = parts[0]
 
-        command = self._get_upgrade_command(kdu_model, kdu_instance, instance_info["namespace"],
-                                            params_str, version, atomic, timeout)
+        command = self._get_upgrade_command(
+            kdu_model,
+            kdu_instance,
+            instance_info["namespace"],
+            params_str,
+            version,
+            atomic,
+            timeout,
+        )
 
         self.log.debug("upgrading: {}".format(command))
 
@@ -494,6 +521,24 @@ class K8sHelmBaseConnector(K8sConnector):
         else:
             return 0
 
+    async def scale(
+        self,
+        kdu_instance: str,
+        scale: int,
+        resource_name: str,
+        total_timeout: float = 1800,
+        **kwargs,
+    ):
+        raise NotImplementedError("Method not implemented")
+
+    async def get_scale_count(
+        self,
+        resource_name: str,
+        kdu_instance: str,
+        **kwargs,
+    ):
+        raise NotImplementedError("Method not implemented")
+
     async def rollback(
         self, cluster_uuid: str, kdu_instance: str, revision=0, db_dict: dict = None
     ):
@@ -518,8 +563,9 @@ class K8sHelmBaseConnector(K8sConnector):
             cluster_name=cluster_id, create_if_not_exist=True
         )
 
-        command = self._get_rollback_command(kdu_instance, instance_info["namespace"],
-                                             revision)
+        command = self._get_rollback_command(
+            kdu_instance, instance_info["namespace"], revision
+        )
 
         self.log.debug("rolling_back: {}".format(command))
 
@@ -593,9 +639,7 @@ class K8sHelmBaseConnector(K8sConnector):
 
         _, 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_id)
         )
 
         # sync local dir
@@ -678,10 +722,9 @@ class K8sHelmBaseConnector(K8sConnector):
             "different from rollback, upgrade and status"
         )
 
-    async def get_services(self,
-                           cluster_uuid: str,
-                           kdu_instance: str,
-                           namespace: str) -> list:
+    async def get_services(
+        self, cluster_uuid: str, kdu_instance: str, namespace: str
+    ) -> list:
         """
         Returns a list of services defined for the specified kdu instance.
 
@@ -721,14 +764,14 @@ class K8sHelmBaseConnector(K8sConnector):
 
         return service_list
 
-    async def get_service(self,
-                          cluster_uuid: str,
-                          service_name: str,
-                          namespace: str) -> object:
+    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)
+                service_name, namespace, cluster_uuid
+            )
         )
 
         _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
@@ -785,8 +828,11 @@ class K8sHelmBaseConnector(K8sConnector):
                 break
         else:
             # instance does not exist
-            raise K8sException("Instance name: {} not found in cluster: {}".format(
-                kdu_instance, cluster_id))
+            raise K8sException(
+                "Instance name: {} not found in cluster: {}".format(
+                    kdu_instance, cluster_id
+                )
+            )
 
         status = await self._status_kdu(
             cluster_id=cluster_id,
@@ -845,14 +891,19 @@ class K8sHelmBaseConnector(K8sConnector):
                     repo_id = db_repo.get("_id")
                     if curr_repo_url != db_repo["url"]:
                         if curr_repo_url:
-                            self.log.debug("repo {} url changed, delete and and again".format(
-                                db_repo["url"]))
+                            self.log.debug(
+                                "repo {} url changed, delete and and again".format(
+                                    db_repo["url"]
+                                )
+                            )
                             await self.repo_remove(cluster_uuid, db_repo["name"])
                             deleted_repo_list.append(repo_id)
 
                         # add repo
                         self.log.debug("add repo {}".format(db_repo["name"]))
-                        await self.repo_add(cluster_uuid, db_repo["name"], db_repo["url"])
+                        await self.repo_add(
+                            cluster_uuid, db_repo["name"], db_repo["url"]
+                        )
                         added_repo_dict[repo_id] = db_repo["name"]
                 except Exception as e:
                     raise K8sException(
@@ -927,22 +978,30 @@ class K8sHelmBaseConnector(K8sConnector):
         """
 
     @abc.abstractmethod
-    async def _status_kdu(self, cluster_id: str, kdu_instance: str, namespace: str = None,
-                          show_error_log: bool = False, return_text: bool = False):
+    async def _status_kdu(
+        self,
+        cluster_id: str,
+        kdu_instance: str,
+        namespace: str = None,
+        show_error_log: bool = False,
+        return_text: bool = False,
+    ):
         """
         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) -> str:
+    def _get_install_command(
+        self, kdu_model, kdu_instance, namespace, params_str, version, atomic, timeout
+    ) -> str:
         """
         Obtain command to be executed to delete the indicated instance
         """
 
     @abc.abstractmethod
-    def _get_upgrade_command(self, kdu_model, kdu_instance, namespace,
-                             params_str, version, atomic, timeout) -> str:
+    def _get_upgrade_command(
+        self, kdu_model, kdu_instance, namespace, params_str, version, atomic, timeout
+    ) -> str:
         """
         Obtain command to be executed to upgrade the indicated instance
         """
@@ -960,8 +1019,9 @@ class K8sHelmBaseConnector(K8sConnector):
         """
 
     @abc.abstractmethod
-    def _get_inspect_command(self, show_command: str, kdu_model: str, repo_str: str,
-                             version: str):
+    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
         """
@@ -1104,11 +1164,13 @@ class K8sHelmBaseConnector(K8sConnector):
         raise_exception_on_error: bool = False,
         show_error_log: bool = True,
         encode_utf8: bool = False,
-        env: dict = None
+        env: dict = None,
     ) -> (str, int):
 
         command = K8sHelmBaseConnector._remove_multiple_spaces(command)
-        self.log.debug("Executing async local command: {}, env: {}".format(command, env))
+        self.log.debug(
+            "Executing async local command: {}, env: {}".format(command, env)
+        )
 
         # split command
         command = shlex.split(command)
@@ -1119,8 +1181,10 @@ class K8sHelmBaseConnector(K8sConnector):
 
         try:
             process = await asyncio.create_subprocess_exec(
-                *command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
-                env=environ
+                *command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                env=environ,
             )
 
             # wait for command terminate
@@ -1164,18 +1228,22 @@ class K8sHelmBaseConnector(K8sConnector):
             else:
                 return "", -1
 
-    async def _local_async_exec_pipe(self,
-                                     command1: str,
-                                     command2: str,
-                                     raise_exception_on_error: bool = True,
-                                     show_error_log: bool = True,
-                                     encode_utf8: bool = False,
-                                     env: dict = None):
+    async def _local_async_exec_pipe(
+        self,
+        command1: str,
+        command2: str,
+        raise_exception_on_error: bool = True,
+        show_error_log: bool = True,
+        encode_utf8: bool = False,
+        env: dict = None,
+    ):
 
         command1 = K8sHelmBaseConnector._remove_multiple_spaces(command1)
         command2 = K8sHelmBaseConnector._remove_multiple_spaces(command2)
         command = "{} | {}".format(command1, command2)
-        self.log.debug("Executing async local command: {}, env: {}".format(command, env))
+        self.log.debug(
+            "Executing async local command: {}, env: {}".format(command, env)
+        )
 
         # split command
         command1 = shlex.split(command1)
@@ -1189,9 +1257,9 @@ class K8sHelmBaseConnector(K8sConnector):
             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)
+            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()
 
@@ -1267,7 +1335,7 @@ class K8sHelmBaseConnector(K8sConnector):
             "name": service_name,
             "type": self._get_deep(data, ("spec", "type")),
             "ports": self._get_deep(data, ("spec", "ports")),
-            "cluster_ip": self._get_deep(data, ("spec", "clusterIP"))
+            "cluster_ip": self._get_deep(data, ("spec", "clusterIP")),
         }
         if service["type"] == "LoadBalancer":
             ip_map_list = self._get_deep(data, ("status", "loadBalancer", "ingress"))
@@ -1299,7 +1367,9 @@ class K8sHelmBaseConnector(K8sConnector):
                 version = "--version {}".format(str(parts[1]))
                 kdu_model = parts[0]
 
-        full_command = self._get_inspect_command(inspect_command, kdu_model, repo_str, version)
+        full_command = self._get_inspect_command(
+            inspect_command, kdu_model, repo_str, version
+        )
         output, _rc = await self._local_async_exec(
             command=full_command, encode_utf8=True
         )
@@ -1320,11 +1390,13 @@ class K8sHelmBaseConnector(K8sConnector):
             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
+                    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))
+                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,
@@ -1339,7 +1411,9 @@ class K8sHelmBaseConnector(K8sConnector):
                 self.log.debug("Task cancelled")
                 return
             except Exception as e:
-                self.log.debug("_store_status exception: {}".format(str(e)), exc_info=True)
+                self.log.debug(
+                    "_store_status exception: {}".format(str(e)), exc_info=True
+                )
                 pass
             finally:
                 if run_once:
@@ -1350,9 +1424,7 @@ class K8sHelmBaseConnector(K8sConnector):
     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
-            )
+            self._init_paths_env(cluster_name=cluster_id, create_if_not_exist=True)
 
             def get_random_number():
                 r = random.randrange(start=1, stop=99999999)
@@ -1399,11 +1471,11 @@ class K8sHelmBaseConnector(K8sConnector):
         # check embeded chart (file or dir)
         if chart_name.startswith("/"):
             # extract file or directory name
-            chart_name = chart_name[chart_name.rfind("/") + 1:]
+            chart_name = chart_name[chart_name.rfind("/") + 1 :]
         # check URL
         elif "://" in chart_name:
             # extract last portion of URL
-            chart_name = chart_name[chart_name.rfind("/") + 1:]
+            chart_name = chart_name[chart_name.rfind("/") + 1 :]
 
         name = ""
         for c in chart_name: