# contact with: nfvlabs@tid.es
##
import asyncio
+from typing import Union
import os
import yaml
helm_command: str = "/usr/bin/helm",
log: object = None,
on_update_db=None,
- vca_config: dict = None,
):
"""
Initializes helm connector for helm v2
kubectl_command=kubectl_command,
helm_command=helm_command,
on_update_db=on_update_db,
- vca_config=vca_config,
)
self.log.info("Initializing K8S Helm2 connector")
# initialize helm client-only
self.log.debug("Initializing helm client-only...")
- command = "{} init --client-only --stable-repo-url {} ".format(
- self._helm_command, self._stable_repo_url
+ command = "{} init --client-only {} ".format(
+ self._helm_command,
+ "--stable-repo-url {}".format(self._stable_repo_url)
+ if self._stable_repo_url
+ else "--skip-repos",
)
try:
asyncio.ensure_future(
return paths, env
- async def _get_services(self, cluster_id, kdu_instance, namespace):
+ async def _get_services(self, cluster_id, kdu_instance, namespace, kubeconfig):
# init config, env
paths, env = self._init_paths_env(
cluster_name=cluster_id, create_if_not_exist=True
)
- command1 = "{} get manifest {} ".format(self._helm_command, kdu_instance)
+ command1 = "env KUBECONFIG={} {} get manifest {} ".format(
+ kubeconfig, self._helm_command, kdu_instance
+ )
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
)
command = (
- "{} --kubeconfig={} --tiller-namespace={} --home={} --service-account {} "
- " --stable-repo-url {} init"
+ "{} init --kubeconfig={} --tiller-namespace={} --home={} --service-account {} "
+ " {}"
).format(
self._helm_command,
paths["kube_config"],
namespace,
paths["helm_dir"],
self.service_account,
- self._stable_repo_url,
+ "--stable-repo-url {}".format(self._stable_repo_url)
+ if self._stable_repo_url
+ else "--skip-repos",
)
_, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=True, env=env
):
self.log.info("Initializing helm in client: {}".format(cluster_id))
command = (
- "{} --kubeconfig={} --tiller-namespace={} "
- "--home={} init --client-only --stable-repo-url {} "
+ "{} init --kubeconfig={} --tiller-namespace={} "
+ "--home={} --client-only {} "
).format(
self._helm_command,
paths["kube_config"],
namespace,
paths["helm_dir"],
- self._stable_repo_url,
+ "--stable-repo-url {}".format(self._stable_repo_url)
+ if self._stable_repo_url
+ else "--skip-repos",
)
output, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=True, env=env
if repo["name"] == "stable" and repo["url"] != self._stable_repo_url:
self.log.debug("Add new stable repo url: {}")
await self.repo_remove(cluster_uuid, "stable")
- await self.repo_add(cluster_uuid, "stable", self._stable_repo_url)
+ if self._stable_repo_url:
+ await self.repo_add(cluster_uuid, "stable", self._stable_repo_url)
break
return n2vc_installed_sw
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)
paths, env = self._init_paths_env(
cluster_name=cluster_id, create_if_not_exist=True
)
- command = "{} status {} --output yaml".format(self._helm_command, kdu_instance)
+ command = ("env KUBECONFIG={} {} status {} --output yaml").format(
+ paths["kube_config"], self._helm_command, kdu_instance
+ )
output, rc = await self._local_async_exec(
command=command,
raise_exception_on_error=True,
env=env,
)
- if return_text:
+ if yaml_format:
return str(output)
if rc != 0:
except KeyError:
pass
+ # 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)
+
# parse field 'resources'
try:
resources = str(data.get("info").get("status").get("resources"))
)
async def _is_install_completed(self, cluster_id: str, kdu_instance: str) -> bool:
+ # init config, env
+ paths, env = self._init_paths_env(
+ cluster_name=cluster_id, create_if_not_exist=True
+ )
status = await self._status_kdu(
- cluster_id=cluster_id, kdu_instance=kdu_instance, return_text=False
+ cluster_id=cluster_id, kdu_instance=kdu_instance, yaml_format=False
)
# extract info.status.resources-> str
return ready
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:
timeout_str = ""
version_str = version_str = "--version {}".format(version)
command = (
- "{helm} install {atomic} --output yaml "
+ "env KUBECONFIG={kubeconfig} {helm} install {atomic} --output yaml "
"{params} {timeout} --name={name} {ns} {model} {ver}".format(
+ kubeconfig=kubeconfig,
helm=self._helm_command,
atomic=atomic_str,
params=params_str,
return command
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:
timeout_str = ""
if version:
version_str = "--version {}".format(version)
- command = "{helm} upgrade {atomic} --output yaml {params} {timeout} {name} {model} {ver}".format(
+ command = (
+ "env KUBECONFIG={kubeconfig} {helm} upgrade {atomic} --output yaml {params} {timeout} {name} {model} {ver}"
+ ).format(
+ kubeconfig=kubeconfig,
helm=self._helm_command,
atomic=atomic_str,
params=params_str,
)
return command
- def _get_rollback_command(self, kdu_instance, namespace, revision) -> str:
- return "{} rollback {} {} --wait".format(
- self._helm_command, kdu_instance, revision
+ def _get_rollback_command(
+ self, kdu_instance, namespace, revision, kubeconfig
+ ) -> str:
+ return "env KUBECONFIG={} {} rollback {} {} --wait".format(
+ kubeconfig, self._helm_command, kdu_instance, revision
)
- def _get_uninstall_command(self, kdu_instance: str, namespace: str) -> str:
- return "{} delete --purge {}".format(self._helm_command, kdu_instance)
+ def _get_uninstall_command(
+ self, kdu_instance: str, namespace: str, kubeconfig: str
+ ) -> str:
+ return "env KUBECONFIG={} {} delete --purge {}".format(
+ kubeconfig, self._helm_command, kdu_instance
+ )