return paths, env
async def _get_services(self, cluster_id, kdu_instance, namespace, kubeconfig):
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
)
command1 = "env KUBECONFIG={} {} get manifest {} ".format(
# init config, env
paths, env = self._init_paths_env(
cluster_name=cluster_id, create_if_not_exist=True
)
command1 = "env KUBECONFIG={} {} get manifest {} ".format(
- kubeconfig, self._helm_command, kdu_instance
+ kubeconfig, self._helm_command, quote(kdu_instance)
+ )
+ command2 = "{} get --namespace={} -f -".format(
+ self.kubectl_command, quote(namespace)
output, _rc = await self._local_async_exec_pipe(
command1, command2, env=env, raise_exception_on_error=True
)
output, _rc = await self._local_async_exec_pipe(
command1, command2, env=env, raise_exception_on_error=True
)
# check if tiller pod is up in cluster
command = "{} --kubeconfig={} --namespace={} get deployments".format(
# check if tiller pod is up in cluster
command = "{} --kubeconfig={} --namespace={} get deployments".format(
)
output, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=True, env=env
)
output, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=True, env=env
"Initializing helm in client and server: {}".format(cluster_id)
)
command = "{} --kubeconfig={} --namespace kube-system create serviceaccount {}".format(
"Initializing helm in client and server: {}".format(cluster_id)
)
command = "{} --kubeconfig={} --namespace kube-system create serviceaccount {}".format(
)
_, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=False, env=env
)
_, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=False, env=env
command = (
"{} --kubeconfig={} create clusterrolebinding osm-tiller-cluster-rule "
"--clusterrole=cluster-admin --serviceaccount=kube-system:{}"
command = (
"{} --kubeconfig={} create clusterrolebinding osm-tiller-cluster-rule "
"--clusterrole=cluster-admin --serviceaccount=kube-system:{}"
_, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=False, env=env
)
_, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=False, env=env
)
- namespace,
- paths["helm_dir"],
- self.service_account,
- "--stable-repo-url {}".format(self._stable_repo_url)
+ quote(namespace),
+ quote(paths["helm_dir"]),
+ quote(self.service_account),
+ "--stable-repo-url {}".format(quote(self._stable_repo_url))
if not namespace:
# find namespace for tiller pod
command = "{} --kubeconfig={} get deployments --all-namespaces".format(
if not namespace:
# find namespace for tiller pod
command = "{} --kubeconfig={} get deployments --all-namespaces".format(
)
output, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=False, env=env
)
output, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=False, env=env
# uninstall tiller from cluster
self.log.debug("Uninstalling tiller from cluster {}".format(cluster_id))
command = "{} --kubeconfig={} --home={} reset".format(
# uninstall tiller from cluster
self.log.debug("Uninstalling tiller from cluster {}".format(cluster_id))
command = "{} --kubeconfig={} --home={} reset".format(
- self._helm_command, paths["kube_config"], paths["helm_dir"]
+ self._helm_command,
+ quote(paths["kube_config"]),
+ quote(paths["helm_dir"]),
)
self.log.debug("resetting: {}".format(command))
output, _rc = await self._local_async_exec(
)
self.log.debug("resetting: {}".format(command))
output, _rc = await self._local_async_exec(
output, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=False, env=env
)
command = (
"{} --kubeconfig={} --namespace {} delete serviceaccount/{}".format(
self.kubectl_command,
output, _rc = await self._local_async_exec(
command=command, raise_exception_on_error=False, env=env
)
command = (
"{} --kubeconfig={} --namespace {} delete serviceaccount/{}".format(
self.kubectl_command,
self.log.debug("namespace not found")
async def _instances_list(self, cluster_id):
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
# init paths, env
paths, env = self._init_paths_env(
cluster_name=cluster_id, create_if_not_exist=True
self, show_command: str, kdu_model: str, repo_str: str, version: str
):
inspect_command = "{} inspect {} {}{} {}".format(
self, show_command: str, kdu_model: str, repo_str: str, version: str
):
inspect_command = "{} inspect {} {}{} {}".format(
- self._helm_command, show_command, kdu_model, repo_str, version
+ self._helm_command, show_command, quote(kdu_model), repo_str, version
self, get_command: str, kdu_instance: str, namespace: str, kubeconfig: str
):
get_command = "env KUBECONFIG={} {} get {} {} --output yaml".format(
self, get_command: str, kdu_instance: str, namespace: str, kubeconfig: str
):
get_command = "env KUBECONFIG={} {} get {} {} --output yaml".format(
- kubeconfig, self._helm_command, get_command, kdu_instance
+ kubeconfig, self._helm_command, get_command, quote(kdu_instance)
yaml_format: bool = False,
show_error_log: bool = False,
) -> Union[str, dict]:
yaml_format: bool = False,
show_error_log: bool = False,
) -> Union[str, dict]:
cluster_name=cluster_id, create_if_not_exist=True
)
command = ("env KUBECONFIG={} {} status {} --output yaml").format(
cluster_name=cluster_id, create_if_not_exist=True
)
command = ("env KUBECONFIG={} {} status {} --output yaml").format(
self, kdu_instance, namespace, revision, kubeconfig
) -> str:
return "env KUBECONFIG={} {} rollback {} {} --wait".format(
self, kdu_instance, namespace, revision, kubeconfig
) -> str:
return "env KUBECONFIG={} {} rollback {} {} --wait".format(
- kubeconfig, self._helm_command, kdu_instance, revision
+ kubeconfig, self._helm_command, quote(kdu_instance), revision
)
def _get_uninstall_command(
self, kdu_instance: str, namespace: str, kubeconfig: str
) -> str:
return "env KUBECONFIG={} {} delete --purge {}".format(
)
def _get_uninstall_command(
self, kdu_instance: str, namespace: str, kubeconfig: str
) -> str:
return "env KUBECONFIG={} {} delete --purge {}".format(