return True
async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
-
self.log.debug(
"inspect kdu_model {} from (optional) repo: {}".format(kdu_model, repo_url)
)
return paths, env
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
self.log.debug("namespace not found")
async def _instances_list(self, cluster_id):
-
# init paths, env
paths, env = self._init_paths_env(
cluster_name=cluster_id, create_if_not_exist=True
yaml_format: bool = False,
show_error_log: bool = False,
) -> Union[str, dict]:
-
self.log.debug(
"status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
)
timeout,
kubeconfig,
) -> str:
-
timeout_str = ""
if timeout:
timeout_str = "--timeout {}".format(timeout)