# For those usages not covered by the Apache License, Version 2.0 please
# contact with: nfvlabs@tid.es
##
+from typing import Union
import os
import yaml
"""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_model str: chart/reference (string), which can be either
+ of these options:
+ - a name of chart available via the repos known by OSM
+ (e.g. stable/openldap, stable/openldap:1.2.4)
+ - a path to a packaged chart (e.g. mychart.tgz)
+ - a path to an unpacked chart directory or a URL (e.g. mychart)
:param kdu_instance: Kdu instance name
:param atomic bool: If set, waits until the model is active and resets
the cluster on failure.
:return: True if successful
"""
- _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
- self.log.debug("installing {} in cluster {}".format(kdu_model, cluster_id))
+
+ self.log.debug("installing {} in cluster {}".format(kdu_model, cluster_uuid))
# sync local dir
- self.fs.sync(from_path=cluster_id)
+ self.fs.sync(from_path=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
)
# for helm3 if namespace does not exist must create it
if namespace and namespace != "kube-system":
- if not await self._namespace_exists(cluster_id, namespace):
+ if not await self._namespace_exists(cluster_uuid, namespace):
try:
- await self._create_namespace(cluster_id, namespace)
+ await self._create_namespace(cluster_uuid, namespace)
except Exception as e:
- if not await self._namespace_exists(cluster_id, namespace):
+ if not await self._namespace_exists(cluster_uuid, namespace):
err_msg = (
"namespace {} does not exist in cluster_id {} "
"error message: ".format(namespace, e)
raise K8sException(err_msg)
await self._install_impl(
- cluster_id,
+ cluster_uuid,
kdu_model,
paths,
env,
)
# sync fs
- self.fs.reverse_sync(from_path=cluster_id)
+ self.fs.reverse_sync(from_path=cluster_uuid)
self.log.debug("Returning kdu_instance {}".format(kdu_instance))
return True
if namespace not in namespaces:
await self._create_namespace(cluster_id, namespace)
- # If default repo is not included add
- cluster_uuid = "{}:{}".format(namespace, cluster_id)
- repo_list = await self.repo_list(cluster_uuid)
+ repo_list = await self.repo_list(cluster_id)
stable_repo = [repo for repo in repo_list if repo["name"] == "stable"]
if not stable_repo and self._stable_repo_url:
- await self.repo_add(cluster_uuid, "stable", self._stable_repo_url)
+ await self.repo_add(cluster_id, "stable", self._stable_repo_url)
# Returns False as no software needs to be uninstalled
return False
def _get_inspect_command(
self, inspect_command: str, kdu_model: str, repo_str: str, version: str
):
+ """Generates the command to obtain the information about an Helm Chart package
+ (´helm show ...´ command)
+
+ Args:
+ show_command: the second part of the command (`helm show <show_command>`)
+ kdu_model: The name or path of an Helm Chart
+ repo_url: Helm Chart repository url
+ version: constraint with specific version of the Chart to use
+
+ Returns:
+ str: the generated Helm Chart command
+ """
+
inspect_command = "{} show {} {}{} {}".format(
self._helm_command, inspect_command, kdu_model, repo_str, version
)
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]:
self.log.debug(
"status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
env=env,
)
- if return_text:
+ if yaml_format:
return str(output)
if rc != 0:
# remove field 'notes' and manifest
try:
del data.get("info")["notes"]
- del data["manifest"]
except KeyError:
pass
- # unable to parse 'resources' as currently it is not included in helm3
+ # parse the manifest to a list of dictionaries
+ if "manifest" in data:
+ manifest_str = data.get("manifest")
+ manifest_docs = yaml.load_all(manifest_str, Loader=yaml.SafeLoader)
+
+ data["manifest"] = []
+ for doc in manifest_docs:
+ data["manifest"].append(doc)
+
return data
def _get_install_command(