##
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
+from n2vc.config import EnvironConfig
from n2vc.exceptions import K8sException
from n2vc.k8s_conn import K8sConnector
################################### P U B L I C ####################################
####################################################################################
"""
+
service_account = "osm"
- _STABLE_REPO_URL = "https://charts.helm.sh/stable"
def __init__(
self,
helm_command: str = "/usr/bin/helm",
log: object = None,
on_update_db=None,
- vca_config: dict = None,
):
"""
self.log.info("Initializing K8S Helm connector")
+ self.config = EnvironConfig()
# random numbers for release name generation
random.seed(time.time())
self._check_file_exists(filename=helm_command, exception_if_not_exists=True)
# obtain stable repo url from config or apply default
- if not vca_config or not vca_config.get("stablerepourl"):
- self._stable_repo_url = self._STABLE_REPO_URL
- else:
- self._stable_repo_url = vca_config.get("stablerepourl")
+ self._stable_repo_url = self.config.get("stablerepourl")
+ 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, 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
"""
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))
+ 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
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"
+ 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))
-
- # sync local dir
- self.fs.sync(from_path=cluster_id)
+ self.log.debug(
+ "Cluster {}, adding {} repository {}. URL: {}".format(
+ cluster_uuid, repo_type, name, url
+ )
+ )
# 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)
+ 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
+ # 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=False, env=env
)
- self.log.debug("adding repo: {}".format(command))
- 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 repo_list(self, cluster_uuid: str) -> list:
"""
: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
)
# 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:
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
)
- 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, 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
: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)
- self.log.debug("Resetting K8s environment. cluster uuid: {} uninstall={}"
- .format(cluster_id, uninstall_sw))
+ namespace = self._get_namespace(cluster_uuid=cluster_uuid)
+ self.log.debug(
+ "Resetting K8s environment. cluster uuid: {} uninstall={}".format(
+ 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:
# 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)
+ ).format(cluster_uuid)
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)
+ 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
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,
):
+ # 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 = str(parts[1])
kdu_model = parts[0]
- command = self._get_install_command(kdu_model, kdu_instance, namespace,
- params_str, version, atomic, timeout)
+ 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,
+ paths["kube_config"],
+ )
self.log.debug("installing: {}".format(command))
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)
# 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
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)
+ repo = self._split_repo(kdu_model)
+ if repo:
+ self.repo_update(cluster_uuid, repo)
+
+ command = self._get_upgrade_command(
+ kdu_model,
+ kdu_instance,
+ instance_info["namespace"],
+ params_str,
+ version,
+ atomic,
+ timeout,
+ paths["kube_config"],
+ )
self.log.debug("upgrading: {}".format(command))
# 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,
# 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,
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(
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
):
-
- _, 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)
# 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_rollback_command(kdu_instance, instance_info["namespace"],
- revision)
+ # sync local dir
+ self.fs.sync(from_path=cluster_uuid)
+
+ command = self._get_rollback_command(
+ kdu_instance, instance_info["namespace"], revision, paths["kube_config"]
+ )
self.log.debug("rolling_back: {}".format(command))
# 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,
# 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,
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(
: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
+ 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)
: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
"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.
- `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
- 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)
-
# 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
: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:
)
)
- _, 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
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_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
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(
"""
@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
"""
@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,
+ yaml_format: bool = False,
+ show_error_log: 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) -> str:
+ def _get_install_command(
+ 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_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,
+ 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
"""
@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
"""
of dictionaries
"""
new_list = []
- for dictionary in input_list:
- new_dict = dict((k.lower(), v) for k, v in dictionary.items())
- new_list.append(new_dict)
+ if input_list:
+ for dictionary in input_list:
+ new_dict = dict((k.lower(), v) for k, v in dictionary.items())
+ 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,
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)
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
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)
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()
"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"))
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
)
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,
+ yaml_format=False,
+ namespace=namespace,
)
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,
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:
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)
# 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:
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