##
import abc
import asyncio
+from typing import Union
import random
import time
import shlex
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",
+ cert: str = None,
+ user: str = None,
+ password: str = None,
):
self.log.debug(
"Cluster {}, adding {} repository {}. URL: {}".format(
)
# helm repo add name url
- command = "env KUBECONFIG={} {} repo add {} {}".format(
+ command = ("env KUBECONFIG={} {} repo add {} {}").format(
paths["kube_config"], self._helm_command, name, url
)
+
+ if cert:
+ temp_cert_file = os.path.join(
+ self.fs.path, "{}/helmcerts/".format(cluster_uuid), "temp.crt"
+ )
+ os.makedirs(os.path.dirname(temp_cert_file), exist_ok=True)
+ with open(temp_cert_file, "w") as the_cert:
+ the_cert.write(cert)
+ command += " --ca-file {}".format(temp_cert_file)
+
+ if user:
+ command += " --username={}".format(user)
+
+ if password:
+ command += " --password={}".format(password)
+
self.log.debug("adding repo: {}".format(command))
await self._local_async_exec(
command=command, raise_exception_on_error=True, env=env
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=cluster_uuid,
kdu_instance=kdu_instance,
namespace=instance["namespace"],
+ yaml_format=yaml_format,
show_error_log=True,
- return_text=True,
)
# sync fs
# add repo
self.log.debug("add repo {}".format(db_repo["name"]))
- await self.repo_add(
- cluster_uuid, db_repo["name"], db_repo["url"]
- )
+ if "ca_cert" in db_repo:
+ await self.repo_add(
+ cluster_uuid,
+ db_repo["name"],
+ db_repo["url"],
+ cert=db_repo["ca_cert"],
+ )
+ else:
+ 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(
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]:
"""
Implements the helm version dependent method to obtain status of a helm instance
"""
detailed_status = await self._status_kdu(
cluster_id=cluster_id,
kdu_instance=kdu_instance,
+ yaml_format=False,
namespace=namespace,
- return_text=False,
)
status = detailed_status.get("info").get("description")
self.log.debug("KDU {} STATUS: {}.".format(kdu_instance, status))