"""
def __init__(
- self,
- fs: object,
- db: object,
- kubectl_command: str = "/usr/bin/kubectl",
- helm_command: str = "/usr/bin/helm3",
- log: object = None,
- on_update_db=None,
+ self,
+ fs: object,
+ db: object,
+ kubectl_command: str = "/usr/bin/kubectl",
+ helm_command: str = "/usr/bin/helm3",
+ log: object = None,
+ on_update_db=None,
+ vca_config: dict = None,
):
"""
Initializes helm connector for helm v3
"""
# parent class
- K8sHelmBaseConnector.__init__(self,
- db=db,
- log=log,
- fs=fs,
- kubectl_command=kubectl_command,
- helm_command=helm_command,
- on_update_db=on_update_db)
+ K8sHelmBaseConnector.__init__(
+ self,
+ db=db,
+ log=log,
+ fs=fs,
+ kubectl_command=kubectl_command,
+ helm_command=helm_command,
+ on_update_db=on_update_db,
+ vca_config=vca_config,
+ )
self.log.info("K8S Helm3 connector initialized")
async def install(
- self,
- cluster_uuid: str,
- kdu_model: str,
- atomic: bool = True,
- timeout: float = 300,
- params: dict = None,
- db_dict: dict = None,
- kdu_name: str = None,
- namespace: str = None,
+ self,
+ cluster_uuid: str,
+ kdu_model: str,
+ kdu_instance: str,
+ atomic: bool = True,
+ timeout: float = 300,
+ params: dict = None,
+ db_dict: dict = None,
+ kdu_name: str = None,
+ namespace: str = None,
+ **kwargs,
):
+ """Install a helm chart
+
+ :param cluster_uuid str: The UUID of the cluster to install to
+ :param kdu_model str: The name or path of a bundle to install
+ :param kdu_instance: Kdu instance name
+ :param atomic bool: If set, waits until the model is active and resets
+ the cluster on failure.
+ :param timeout int: The time, in seconds, to wait for the install
+ to finish
+ :param params dict: Key-value pairs of instantiation parameters
+ :param kdu_name: Name of the KDU instance to be installed
+ :param namespace: K8s namespace to use for the KDU instance
+
+ :param kwargs: Additional parameters (None yet)
+
+ :return: True if successful
+ """
_, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
self.log.debug("installing {} in cluster {}".format(kdu_model, cluster_id))
if namespace not in namespaces:
await self._create_namespace(cluster_id, namespace)
- kdu_instance = await self._install_impl(cluster_id,
- kdu_model,
- paths,
- env,
- atomic=atomic,
- timeout=timeout,
- params=params,
- db_dict=db_dict,
- kdu_name=kdu_name,
- namespace=namespace)
+ await self._install_impl(
+ cluster_id,
+ kdu_model,
+ paths,
+ env,
+ kdu_instance,
+ atomic=atomic,
+ timeout=timeout,
+ params=params,
+ db_dict=db_dict,
+ kdu_name=kdu_name,
+ namespace=namespace,
+ )
# sync fs
self.fs.reverse_sync(from_path=cluster_id)
self.log.debug("Returning kdu_instance {}".format(kdu_instance))
- return kdu_instance
+ return True
async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
paths = {
"kube_dir": kube_dir,
"kube_config": config_filename,
- "cluster_dir": cluster_dir
+ "cluster_dir": cluster_dir,
}
# 3 - Prepare environment variables
"HELM_CACHE_HOME": helm_path_cache,
"HELM_CONFIG_HOME": helm_path_config,
"HELM_DATA_HOME": helm_path_data,
- "KUBECONFIG": config_filename
+ "KUBECONFIG": config_filename,
}
for file_name, file in paths.items():
return paths, env
- async def _get_namespaces(self,
- cluster_id: str):
+ async def _get_namespaces(self, cluster_id: str):
self.log.debug("get namespaces cluster_id {}".format(cluster_id))
return namespaces
- async def _create_namespace(self,
- cluster_id: str,
- namespace: str):
+ async def _create_namespace(self, cluster_id: str, namespace: str):
self.log.debug(f"create namespace: {cluster_id} for cluster_id: {namespace}")
command1 = "{} get manifest {} --namespace={}".format(
self._helm_command, kdu_instance, namespace
)
- command2 = "{} get --namespace={} -f -".format(
- self.kubectl_command, namespace
- )
+ command2 = "{} get --namespace={} -f -".format(self.kubectl_command, namespace)
output, _rc = await self._local_async_exec_pipe(
command1, command2, env=env, raise_exception_on_error=True
)
self.log.debug("Default repo already present")
break
else:
- await self.repo_add(cluster_uuid,
- "stable",
- "https://kubernetes-charts.storage.googleapis.com/")
+ await self.repo_add(cluster_uuid, "stable", self._stable_repo_url)
# Returns False as no software needs to be uninstalled
return False
cluster_name=cluster_id, create_if_not_exist=True
)
- command = "{} list --all-namespaces --output yaml".format(
- self._helm_command
- )
+ command = "{} list --all-namespaces --output yaml".format(self._helm_command)
output, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=True, env=env
)
else:
return []
- def _get_inspect_command(self, inspect_command: str, kdu_model: str, repo_str: str,
- version: str):
+ def _get_inspect_command(
+ self, inspect_command: str, kdu_model: str, repo_str: str, version: str
+ ):
inspect_command = "{} show {} {}{} {}".format(
self._helm_command, inspect_command, kdu_model, repo_str, version
)
return_text: bool = False,
):
- self.log.debug("status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace))
+ self.log.debug(
+ "status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
+ )
if not namespace:
namespace = "kube-system"
command=command,
raise_exception_on_error=True,
show_error_log=show_error_log,
- env=env
+ env=env,
)
if return_text:
# unable to parse 'resources' as currently it is not included in helm3
return data
- def _get_install_command(self, kdu_model: str, kdu_instance: str, namespace: str,
- params_str: str, version: str, atomic: bool, timeout: float) -> str:
+ def _get_install_command(
+ self,
+ kdu_model: str,
+ kdu_instance: str,
+ namespace: str,
+ params_str: str,
+ version: str,
+ atomic: bool,
+ timeout: float,
+ ) -> str:
timeout_str = ""
if timeout:
)
return command
- def _get_upgrade_command(self, kdu_model: str, kdu_instance: str, namespace: str,
- params_str: str, version: str, atomic: bool, timeout: float) -> str:
+ def _get_upgrade_command(
+ self,
+ kdu_model: str,
+ kdu_instance: str,
+ namespace: str,
+ params_str: str,
+ version: str,
+ atomic: bool,
+ timeout: float,
+ ) -> str:
timeout_str = ""
if timeout:
)
return command
- def _get_rollback_command(self, kdu_instance: str, namespace: str, revision: float) -> str:
+ def _get_rollback_command(
+ self, kdu_instance: str, namespace: str, revision: float
+ ) -> str:
return "{} rollback {} {} --namespace={} --wait".format(
self._helm_command, kdu_instance, revision, namespace
)
def _get_uninstall_command(self, kdu_instance: str, namespace: str) -> str:
return "{} uninstall {} --namespace={}".format(
- self._helm_command, kdu_instance, namespace)
+ self._helm_command, kdu_instance, namespace
+ )
def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
repo_ids = []
projects / osm / N2VC.git / blobdiff