####################################################################################
"""
service_account = "osm"
+ _STABLE_REPO_URL = "https://charts.helm.sh/stable"
def __init__(
self,
helm_command: str = "/usr/bin/helm",
log: object = None,
on_update_db=None,
+ vca_config: dict = None,
):
"""
self._helm_command = helm_command
self._check_file_exists(filename=helm_command, exception_if_not_exists=True)
+ # obtain stable repo url from config or apply default
+ if not vca_config or not vca_config.get("stablerepourl"):
+ self._stable_repo_url = self._STABLE_REPO_URL
+ else:
+ self._stable_repo_url = vca_config.get("stablerepourl")
+
@staticmethod
def _get_namespace_cluster_id(cluster_uuid: str) -> (str, str):
"""
return True
- async def install(
+ async def _install_impl(
self,
- cluster_uuid: str,
+ cluster_id: str,
kdu_model: str,
+ paths: dict,
+ env: dict,
atomic: bool = True,
timeout: float = 300,
params: dict = None,
kdu_name: str = None,
namespace: str = None,
):
- _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
- self.log.debug("installing {} in cluster {}".format(kdu_model, cluster_id))
-
- # sync local dir
- self.fs.sync(from_path=cluster_id)
-
- # init env, paths
- paths, env = self._init_paths_env(
- cluster_name=cluster_id, create_if_not_exist=True
- )
-
# params to str
params_str, file_to_delete = self._params_to_file_option(
cluster_id=cluster_id, params=params
self.log.error(msg)
raise K8sException(msg)
- # sync fs
- self.fs.reverse_sync(from_path=cluster_id)
-
- self.log.debug("Returning kdu_instance {}".format(kdu_instance))
return kdu_instance
async def upgrade(
self.log.error("Error synchronizing repos: {}".format(e))
raise Exception("Error synchronizing repos: {}".format(e))
- def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
- repo_ids = []
- cluster_filter = {"_admin.helm-chart.id": cluster_uuid}
- cluster = self.db.get_one("k8sclusters", cluster_filter)
- if cluster:
- repo_ids = cluster.get("_admin").get("helm_chart_repos") or []
- return repo_ids
- else:
- raise K8sException(
- "k8cluster with helm-id : {} not found".format(cluster_uuid)
- )
-
def _get_db_repos_dict(self, repo_ids: list):
db_repos_dict = {}
for repo_id in repo_ids:
For Helm v3 it does nothing and does not need to be callled
"""
+ @abc.abstractmethod
+ def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
+ """
+ Obtains the cluster repos identifiers
+ """
+
"""
####################################################################################
################################### P R I V A T E ##################################