X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Fk8s_helm_base_conn.py;h=273b206ddb13eb8027773db228e8f45b1884bc02;hp=4a43ee214320cb786bfcbf62e6b003f6dad03c64;hb=156677967e417604682fda05ed0099f07f1a3129;hpb=4395cfa6c7d0d80980c00d9f078440e0333fd826

diff --git a/n2vc/k8s_helm_base_conn.py b/n2vc/k8s_helm_base_conn.py
index 4a43ee2..273b206 100644
--- a/n2vc/k8s_helm_base_conn.py
+++ b/n2vc/k8s_helm_base_conn.py
@@ -26,7 +26,6 @@ import time
 import shlex
 import shutil
 import stat
-import subprocess
 import os
 import yaml
 from uuid import uuid4
@@ -159,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
@@ -194,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(
@@ -227,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
         )
@@ -329,21 +336,28 @@ 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
         )
 
         # version
-        version = None
-        if ":" in kdu_model:
-            parts = kdu_model.split(sep=":")
-            if len(parts) == 2:
-                version = str(parts[1])
-                kdu_model = parts[0]
+        kdu_model, version = self._split_version(kdu_model)
 
         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))
@@ -428,18 +442,16 @@ 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
         )
 
         # version
-        version = None
-        if ":" in kdu_model:
-            parts = kdu_model.split(sep=":")
-            if len(parts) == 2:
-                version = str(parts[1])
-                kdu_model = parts[0]
+        kdu_model, version = self._split_version(kdu_model)
 
         command = self._get_upgrade_command(
             kdu_model,
@@ -449,6 +461,7 @@ class K8sHelmBaseConnector(K8sConnector):
             version,
             atomic,
             timeout,
+            paths["kube_config"],
         )
 
         self.log.debug("upgrading: {}".format(command))
@@ -526,17 +539,189 @@ class K8sHelmBaseConnector(K8sConnector):
         scale: int,
         resource_name: str,
         total_timeout: float = 1800,
+        cluster_uuid: str = None,
+        kdu_model: str = None,
+        atomic: bool = True,
+        db_dict: dict = None,
         **kwargs,
     ):
-        raise NotImplementedError("Method not implemented")
+        """Scale a resource in a Helm Chart.
+
+        Args:
+            kdu_instance: KDU instance name
+            scale: Scale to which to set the resource
+            resource_name: Resource name
+            total_timeout: The time, in seconds, to wait
+            cluster_uuid: The UUID of the cluster
+            kdu_model: The chart reference
+            atomic: if set, upgrade process rolls back changes made in case of failed upgrade.
+                The --wait flag will be set automatically if --atomic is used
+            db_dict: Dictionary for any additional data
+            kwargs: Additional parameters
+
+        Returns:
+            True if successful, False otherwise
+        """
+
+        _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
+
+        debug_mgs = "scaling {} in cluster {}".format(kdu_model, cluster_id)
+        if resource_name:
+            debug_mgs = "scaling resource {} in model {} (cluster {})".format(
+                resource_name, kdu_model, cluster_id
+            )
+
+        self.log.debug(debug_mgs)
+
+        # look for instance to obtain namespace
+        # get_instance_info function calls the sync command
+        instance_info = await self.get_instance_info(cluster_uuid, kdu_instance)
+        if not instance_info:
+            raise K8sException("kdu_instance {} not found".format(kdu_instance))
+
+        # init env, paths
+        paths, env = self._init_paths_env(
+            cluster_name=cluster_id, create_if_not_exist=True
+        )
+
+        # version
+        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
+        )
+
+        command = self._get_upgrade_scale_command(
+            kdu_model,
+            kdu_instance,
+            instance_info["namespace"],
+            scale,
+            version,
+            atomic,
+            replica_str,
+            total_timeout,
+            resource_name,
+            paths["kube_config"],
+        )
+
+        self.log.debug("scaling: {}".format(command))
+
+        if atomic:
+            # exec helm in a task
+            exec_task = asyncio.ensure_future(
+                coro_or_future=self._local_async_exec(
+                    command=command, raise_exception_on_error=False, env=env
+                )
+            )
+            # write status in another task
+            status_task = asyncio.ensure_future(
+                coro_or_future=self._store_status(
+                    cluster_id=cluster_id,
+                    kdu_instance=kdu_instance,
+                    namespace=instance_info["namespace"],
+                    db_dict=db_dict,
+                    operation="scale",
+                    run_once=False,
+                )
+            )
+
+            # wait for execution task
+            await asyncio.wait([exec_task])
+
+            # cancel status task
+            status_task.cancel()
+            output, rc = exec_task.result()
+
+        else:
+            output, rc = await self._local_async_exec(
+                command=command, raise_exception_on_error=False, env=env
+            )
+
+        # write final status
+        await self._store_status(
+            cluster_id=cluster_id,
+            kdu_instance=kdu_instance,
+            namespace=instance_info["namespace"],
+            db_dict=db_dict,
+            operation="scale",
+            run_once=True,
+            check_every=0,
+        )
+
+        if rc != 0:
+            msg = "Error executing command: {}\nOutput: {}".format(command, output)
+            self.log.error(msg)
+            raise K8sException(msg)
+
+        # sync fs
+        self.fs.reverse_sync(from_path=cluster_id)
+
+        return True
 
     async def get_scale_count(
         self,
         resource_name: str,
         kdu_instance: str,
+        cluster_uuid: str,
+        kdu_model: str,
         **kwargs,
-    ):
-        raise NotImplementedError("Method not implemented")
+    ) -> int:
+        """Get a resource scale count.
+
+        Args:
+            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
+            kwargs: Additional parameters
+
+        Returns:
+            Resource instance count
+        """
+
+        _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
+        self.log.debug(
+            "getting scale count for {} in cluster {}".format(kdu_model, cluster_id)
+        )
+
+        # 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))
+
+        # init env, paths
+        paths, env = self._init_paths_env(
+            cluster_name=cluster_id, create_if_not_exist=True
+        )
+
+        replicas = await self._get_replica_count_instance(
+            kdu_instance, instance_info["namespace"], paths["kube_config"]
+        )
+
+        # 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)
+                )
+
+            replicas, _ = await self._get_replica_count_url(
+                kdu_model, repo_url, resource_name
+            )
+
+        if not replicas:
+            msg = "Replica count not found. Cannot be scaled"
+            self.log.error(msg)
+            raise K8sException(msg)
+
+        return int(replicas)
 
     async def rollback(
         self, cluster_uuid: str, kdu_instance: str, revision=0, db_dict: dict = None
@@ -562,8 +747,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))
@@ -647,14 +835,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
         )
@@ -747,11 +940,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:
@@ -846,6 +1046,19 @@ class K8sHelmBaseConnector(K8sConnector):
 
         return status
 
+    async def get_values_kdu(
+        self, kdu_instance: str, namespace: str, kubeconfig: str
+    ) -> str:
+
+        self.log.debug("get kdu_instance values {}".format(kdu_instance))
+
+        return await self._exec_get_command(
+            get_command="values",
+            kdu_instance=kdu_instance,
+            namespace=namespace,
+            kubeconfig=kubeconfig,
+        )
+
     async def values_kdu(self, kdu_model: str, repo_url: str = None) -> str:
 
         self.log.debug(
@@ -854,7 +1067,7 @@ class K8sHelmBaseConnector(K8sConnector):
             )
         )
 
-        return await self._exec_inspect_comand(
+        return await self._exec_inspect_command(
             inspect_command="values", kdu_model=kdu_model, repo_url=repo_url
         )
 
@@ -864,7 +1077,7 @@ class K8sHelmBaseConnector(K8sConnector):
             "inspect kdu_model {} readme.md from repo: {}".format(kdu_model, repo_url)
         )
 
-        return await self._exec_inspect_comand(
+        return await self._exec_inspect_command(
             inspect_command="readme", kdu_model=kdu_model, repo_url=repo_url
         )
 
@@ -971,7 +1184,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
         """
@@ -991,28 +1204,64 @@ 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
         """
 
+    @abc.abstractmethod
+    def _get_upgrade_scale_command(
+        self,
+        kdu_model,
+        kdu_instance,
+        namespace,
+        count,
+        version,
+        atomic,
+        replicas,
+        timeout,
+        resource_name,
+        kubeconfig,
+    ) -> str:
+        """Obtain command to be executed to upgrade the indicated instance."""
+
     @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
         """
@@ -1025,6 +1274,12 @@ class K8sHelmBaseConnector(K8sConnector):
         Obtain command to be executed to obtain information about the kdu
         """
 
+    @abc.abstractmethod
+    def _get_get_command(
+        self, get_command: str, kdu_instance: str, namespace: str, kubeconfig: str
+    ):
+        """Obtain command to be executed to get information about the kdu instance."""
+
     @abc.abstractmethod
     async def _uninstall_sw(self, cluster_id: str, namespace: str):
         """
@@ -1142,22 +1397,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,
@@ -1344,12 +1583,23 @@ class K8sHelmBaseConnector(K8sConnector):
 
         return service
 
-    async def _exec_inspect_comand(
+    async def _exec_get_command(
+        self, get_command: str, kdu_instance: str, namespace: str, kubeconfig: str
+    ):
+        """Obtains information about the kdu instance."""
+
+        full_command = self._get_get_command(
+            get_command, kdu_instance, namespace, kubeconfig
+        )
+
+        output, _rc = await self._local_async_exec(command=full_command)
+
+        return output
+
+    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 a kdu, no cluster (no env)."""
 
         repo_str = ""
         if repo_url:
@@ -1360,22 +1610,141 @@ class K8sHelmBaseConnector(K8sConnector):
             idx += 1
             kdu_model = kdu_model[idx:]
 
-        version = ""
-        if ":" in kdu_model:
-            parts = kdu_model.split(sep=":")
-            if len(parts) == 2:
-                version = "--version {}".format(str(parts[1]))
-                kdu_model = parts[0]
+        kdu_model, version = self._split_version(kdu_model)
+        if version:
+            version_str = "--version {}".format(version)
+        else:
+            version_str = ""
 
         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
+            inspect_command, kdu_model, repo_str, version_str
         )
 
+        output, _rc = await self._local_async_exec(command=full_command)
+
         return output
 
+    async def _get_replica_count_url(
+        self,
+        kdu_model: str,
+        repo_url: str,
+        resource_name: str = None,
+    ):
+        """Get the replica count value in the Helm Chart Values.
+
+        Args:
+            kdu_model: The name or path of a bundle
+            repo_url: Helm Chart repository url
+            resource_name: Resource name
+
+        Returns:
+            True if replicas, False replicaCount
+        """
+
+        kdu_values = yaml.load(
+            await self.values_kdu(kdu_model, repo_url), Loader=yaml.SafeLoader
+        )
+
+        if not kdu_values:
+            raise K8sException(
+                "kdu_values not found for kdu_model {}".format(kdu_model)
+            )
+
+        if resource_name:
+            kdu_values = kdu_values.get(resource_name, None)
+
+        if not kdu_values:
+            msg = "resource {} not found in the values in model {}".format(
+                resource_name, kdu_model
+            )
+            self.log.error(msg)
+            raise K8sException(msg)
+
+        duplicate_check = False
+
+        replica_str = ""
+        replicas = None
+
+        if kdu_values.get("replicaCount", None):
+            replicas = kdu_values["replicaCount"]
+            replica_str = "replicaCount"
+        elif kdu_values.get("replicas", None):
+            duplicate_check = True
+            replicas = kdu_values["replicas"]
+            replica_str = "replicas"
+        else:
+            if resource_name:
+                msg = (
+                    "replicaCount or replicas not found in the resource"
+                    "{} values in model {}. Cannot be scaled".format(
+                        resource_name, kdu_model
+                    )
+                )
+            else:
+                msg = (
+                    "replicaCount or replicas not found in the values"
+                    "in model {}. Cannot be scaled".format(kdu_model)
+                )
+            self.log.error(msg)
+            raise K8sException(msg)
+
+        # Control if replicas and replicaCount exists at the same time
+        msg = "replicaCount and replicas are exists at the same time"
+        if duplicate_check:
+            if "replicaCount" in kdu_values:
+                self.log.error(msg)
+                raise K8sException(msg)
+        else:
+            if "replicas" in kdu_values:
+                self.log.error(msg)
+                raise K8sException(msg)
+
+        return replicas, replica_str
+
+    async def _get_replica_count_instance(
+        self,
+        kdu_instance: str,
+        namespace: str,
+        kubeconfig: str,
+        resource_name: str = None,
+    ):
+        """Get the replica count value in the instance.
+
+        Args:
+            kdu_instance: The name of the KDU instance
+            namespace: KDU instance namespace
+            kubeconfig:
+            resource_name: Resource name
+
+        Returns:
+            True if replicas, False replicaCount
+        """
+
+        kdu_values = yaml.load(
+            await self.get_values_kdu(kdu_instance, namespace, kubeconfig),
+            Loader=yaml.SafeLoader,
+        )
+
+        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)
+                )
+            )
+
+        return replicas
+
     async def _store_status(
         self,
         cluster_id: str,
@@ -1500,3 +1869,23 @@ 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 _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