##
import abc
import asyncio
+from typing import Union
import random
import time
import shlex
# sync local dir
self.fs.sync(from_path=cluster_uuid)
- # helm repo update
- command = "env KUBECONFIG={} {} repo update".format(
- paths["kube_config"], self._helm_command
- )
- self.log.debug("updating repo: {}".format(command))
- await self._local_async_exec(
- command=command, raise_exception_on_error=False, env=env
- )
-
# helm repo add name url
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_id), "temp.crt"
+ 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:
command=command, raise_exception_on_error=True, env=env
)
+ # helm repo update
+ command = "env KUBECONFIG={} {} repo update".format(
+ paths["kube_config"], self._helm_command
+ )
+ self.log.debug("updating repo: {}".format(command))
+ await self._local_async_exec(
+ command=command, raise_exception_on_error=False, env=env
+ )
+
# sync fs
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=cluster_uuid,
kdu_instance=kdu_instance,
namespace=instance["namespace"],
+ yaml_format=yaml_format,
show_error_log=True,
- return_text=True,
)
# sync fs
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))