n2vc/k8s_helm3_conn.py

   1 ##
   2 # Copyright 2019 Telefonica Investigacion y Desarrollo, S.A.U.
   3 # This file is part of OSM
   4 # All Rights Reserved.
   5 #
   6 # Licensed under the Apache License, Version 2.0 (the "License");
   7 # you may not use this file except in compliance with the License.
   8 # You may obtain a copy of the License at
   9 #
  10 #    http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  15 # implied.
  16 # See the License for the specific language governing permissions and
  17 # limitations under the License.
  18 #
  19 # For those usages not covered by the Apache License, Version 2.0 please
  20 # contact with: nfvlabs@tid.es
  21 ##
  22 import os
  23 import yaml
  24
  25 from n2vc.k8s_helm_base_conn import K8sHelmBaseConnector
  26 from n2vc.exceptions import K8sException
  27
  28
  29 class K8sHelm3Connector(K8sHelmBaseConnector):
  30
  31     """
  32     ####################################################################################
  33     ################################### P U B L I C ####################################
  34     ####################################################################################
  35     """
  36
  37     def __init__(
  38         self,
  39         fs: object,
  40         db: object,
  41         kubectl_command: str = "/usr/bin/kubectl",
  42         helm_command: str = "/usr/bin/helm3",
  43         log: object = None,
  44         on_update_db=None,
  45     ):
  46         """
  47         Initializes helm connector for helm v3
  48
  49         :param fs: file system for kubernetes and helm configuration
  50         :param db: database object to write current operation status
  51         :param kubectl_command: path to kubectl executable
  52         :param helm_command: path to helm executable
  53         :param log: logger
  54         :param on_update_db: callback called when k8s connector updates database
  55         """
  56
  57         # parent class
  58         K8sHelmBaseConnector.__init__(
  59             self,
  60             db=db,
  61             log=log,
  62             fs=fs,
  63             kubectl_command=kubectl_command,
  64             helm_command=helm_command,
  65             on_update_db=on_update_db,
  66         )
  67
  68         self.log.info("K8S Helm3 connector initialized")
  69
  70     async def install(
  71         self,
  72         cluster_uuid: str,
  73         kdu_model: str,
  74         kdu_instance: str,
  75         atomic: bool = True,
  76         timeout: float = 300,
  77         params: dict = None,
  78         db_dict: dict = None,
  79         kdu_name: str = None,
  80         namespace: str = None,
  81         **kwargs,
  82     ):
  83         """Install a helm chart
  84
  85         :param cluster_uuid str: The UUID of the cluster to install to
  86         :param kdu_model str: The name or path of a bundle to install
  87         :param kdu_instance: Kdu instance name
  88         :param atomic bool: If set, waits until the model is active and resets
  89                             the cluster on failure.
  90         :param timeout int: The time, in seconds, to wait for the install
  91                             to finish
  92         :param params dict: Key-value pairs of instantiation parameters
  93         :param kdu_name: Name of the KDU instance to be installed
  94         :param namespace: K8s namespace to use for the KDU instance
  95
  96         :param kwargs: Additional parameters (None yet)
  97
  98         :return: True if successful
  99         """
 100
 101         self.log.debug("installing {} in cluster {}".format(kdu_model, cluster_uuid))
 102
 103         # sync local dir
 104         self.fs.sync(from_path=cluster_uuid)
 105
 106         # init env, paths
 107         paths, env = self._init_paths_env(
 108             cluster_name=cluster_uuid, create_if_not_exist=True
 109         )
 110
 111         # for helm3 if namespace does not exist must create it
 112         if namespace and namespace != "kube-system":
 113             if not await self._namespace_exists(cluster_uuid, namespace):
 114                 try:
 115                     await self._create_namespace(cluster_uuid, namespace)
 116                 except Exception as e:
 117                     if not await self._namespace_exists(cluster_uuid, namespace):
 118                         err_msg = (
 119                             "namespace {} does not exist in cluster_id {} "
 120                             "error message: ".format(namespace, e)
 121                         )
 122                         self.log.error(err_msg)
 123                         raise K8sException(err_msg)
 124
 125         await self._install_impl(
 126             cluster_uuid,
 127             kdu_model,
 128             paths,
 129             env,
 130             kdu_instance,
 131             atomic=atomic,
 132             timeout=timeout,
 133             params=params,
 134             db_dict=db_dict,
 135             kdu_name=kdu_name,
 136             namespace=namespace,
 137         )
 138
 139         # sync fs
 140         self.fs.reverse_sync(from_path=cluster_uuid)
 141
 142         self.log.debug("Returning kdu_instance {}".format(kdu_instance))
 143         return True
 144
 145     async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
 146
 147         self.log.debug(
 148             "inspect kdu_model {} from (optional) repo: {}".format(kdu_model, repo_url)
 149         )
 150
 151         return await self._exec_inspect_comand(
 152             inspect_command="all", kdu_model=kdu_model, repo_url=repo_url
 153         )
 154
 155     """
 156     ####################################################################################
 157     ################################### P R I V A T E ##################################
 158     ####################################################################################
 159     """
 160
 161     def _init_paths_env(self, cluster_name: str, create_if_not_exist: bool = True):
 162         """
 163         Creates and returns base cluster and kube dirs and returns them.
 164         Also created helm3 dirs according to new directory specification, paths are
 165         returned and also environment variables that must be provided to execute commands
 166
 167         Helm 3 directory specification uses XDG categories for variable support:
 168         - Cache: $XDG_CACHE_HOME, for example, ${HOME}/.cache/helm/
 169         - Configuration: $XDG_CONFIG_HOME, for example, ${HOME}/.config/helm/
 170         - Data: $XDG_DATA_HOME, for example ${HOME}/.local/share/helm
 171
 172         The variables assigned for this paths are:
 173         (In the documentation the variables names are $HELM_PATH_CACHE, $HELM_PATH_CONFIG,
 174         $HELM_PATH_DATA but looking and helm env the variable names are different)
 175         - Cache: $HELM_CACHE_HOME
 176         - Config: $HELM_CONFIG_HOME
 177         - Data: $HELM_DATA_HOME
 178         - helm kubeconfig: $KUBECONFIG
 179
 180         :param cluster_name:  cluster_name
 181         :return: Dictionary with config_paths and dictionary with helm environment variables
 182         """
 183
 184         base = self.fs.path
 185         if base.endswith("/") or base.endswith("\\"):
 186             base = base[:-1]
 187
 188         # base dir for cluster
 189         cluster_dir = base + "/" + cluster_name
 190
 191         # kube dir
 192         kube_dir = cluster_dir + "/" + ".kube"
 193         if create_if_not_exist and not os.path.exists(kube_dir):
 194             self.log.debug("Creating dir {}".format(kube_dir))
 195             os.makedirs(kube_dir)
 196
 197         helm_path_cache = cluster_dir + "/.cache/helm"
 198         if create_if_not_exist and not os.path.exists(helm_path_cache):
 199             self.log.debug("Creating dir {}".format(helm_path_cache))
 200             os.makedirs(helm_path_cache)
 201
 202         helm_path_config = cluster_dir + "/.config/helm"
 203         if create_if_not_exist and not os.path.exists(helm_path_config):
 204             self.log.debug("Creating dir {}".format(helm_path_config))
 205             os.makedirs(helm_path_config)
 206
 207         helm_path_data = cluster_dir + "/.local/share/helm"
 208         if create_if_not_exist and not os.path.exists(helm_path_data):
 209             self.log.debug("Creating dir {}".format(helm_path_data))
 210             os.makedirs(helm_path_data)
 211
 212         config_filename = kube_dir + "/config"
 213
 214         # 2 - Prepare dictionary with paths
 215         paths = {
 216             "kube_dir": kube_dir,
 217             "kube_config": config_filename,
 218             "cluster_dir": cluster_dir,
 219         }
 220
 221         # 3 - Prepare environment variables
 222         env = {
 223             "HELM_CACHE_HOME": helm_path_cache,
 224             "HELM_CONFIG_HOME": helm_path_config,
 225             "HELM_DATA_HOME": helm_path_data,
 226             "KUBECONFIG": config_filename,
 227         }
 228
 229         for file_name, file in paths.items():
 230             if "dir" in file_name and not os.path.exists(file):
 231                 err_msg = "{} dir does not exist".format(file)
 232                 self.log.error(err_msg)
 233                 raise K8sException(err_msg)
 234
 235         return paths, env
 236
 237     async def _namespace_exists(self, cluster_id, namespace) -> bool:
 238         self.log.debug(
 239             "checking if namespace {} exists cluster_id {}".format(
 240                 namespace, cluster_id
 241             )
 242         )
 243         namespaces = await self._get_namespaces(cluster_id)
 244         return namespace in namespaces if namespaces else False
 245
 246     async def _get_namespaces(self, cluster_id: str):
 247
 248         self.log.debug("get namespaces cluster_id {}".format(cluster_id))
 249
 250         # init config, env
 251         paths, env = self._init_paths_env(
 252             cluster_name=cluster_id, create_if_not_exist=True
 253         )
 254
 255         command = "{} --kubeconfig={} get namespaces -o=yaml".format(
 256             self.kubectl_command, paths["kube_config"]
 257         )
 258         output, _rc = await self._local_async_exec(
 259             command=command, raise_exception_on_error=True, env=env
 260         )
 261
 262         data = yaml.load(output, Loader=yaml.SafeLoader)
 263         namespaces = [item["metadata"]["name"] for item in data["items"]]
 264         self.log.debug(f"namespaces {namespaces}")
 265
 266         return namespaces
 267
 268     async def _create_namespace(self, cluster_id: str, namespace: str):
 269
 270         self.log.debug(f"create namespace: {cluster_id} for cluster_id: {namespace}")
 271
 272         # init config, env
 273         paths, env = self._init_paths_env(
 274             cluster_name=cluster_id, create_if_not_exist=True
 275         )
 276
 277         command = "{} --kubeconfig={} create namespace {}".format(
 278             self.kubectl_command, paths["kube_config"], namespace
 279         )
 280         _, _rc = await self._local_async_exec(
 281             command=command, raise_exception_on_error=True, env=env
 282         )
 283         self.log.debug(f"namespace {namespace} created")
 284
 285         return _rc
 286
 287     async def _get_services(
 288         self, cluster_id: str, kdu_instance: str, namespace: str, kubeconfig: str
 289     ):
 290
 291         # init config, env
 292         paths, env = self._init_paths_env(
 293             cluster_name=cluster_id, create_if_not_exist=True
 294         )
 295
 296         command1 = "env KUBECONFIG={} {} get manifest {} --namespace={}".format(
 297             kubeconfig, self._helm_command, kdu_instance, namespace
 298         )
 299         command2 = "{} get --namespace={} -f -".format(self.kubectl_command, namespace)
 300         output, _rc = await self._local_async_exec_pipe(
 301             command1, command2, env=env, raise_exception_on_error=True
 302         )
 303         services = self._parse_services(output)
 304
 305         return services
 306
 307     async def _cluster_init(self, cluster_id, namespace, paths, env):
 308         """
 309         Implements the helm version dependent cluster initialization:
 310         For helm3 it creates the namespace if it is not created
 311         """
 312         if namespace != "kube-system":
 313             namespaces = await self._get_namespaces(cluster_id)
 314             if namespace not in namespaces:
 315                 await self._create_namespace(cluster_id, namespace)
 316
 317         repo_list = await self.repo_list(cluster_id)
 318         stable_repo = [repo for repo in repo_list if repo["name"] == "stable"]
 319         if not stable_repo and self._stable_repo_url:
 320             await self.repo_add(cluster_id, "stable", self._stable_repo_url)
 321
 322         # Returns False as no software needs to be uninstalled
 323         return False
 324
 325     async def _uninstall_sw(self, cluster_id: str, namespace: str):
 326         # nothing to do to uninstall sw
 327         pass
 328
 329     async def _instances_list(self, cluster_id: str):
 330
 331         # init paths, env
 332         paths, env = self._init_paths_env(
 333             cluster_name=cluster_id, create_if_not_exist=True
 334         )
 335
 336         command = "{} list --all-namespaces  --output yaml".format(self._helm_command)
 337         output, _rc = await self._local_async_exec(
 338             command=command, raise_exception_on_error=True, env=env
 339         )
 340
 341         if output and len(output) > 0:
 342             self.log.debug("instances list output: {}".format(output))
 343             return yaml.load(output, Loader=yaml.SafeLoader)
 344         else:
 345             return []
 346
 347     def _get_inspect_command(
 348         self, inspect_command: str, kdu_model: str, repo_str: str, version: str
 349     ):
 350         inspect_command = "{} show {} {}{} {}".format(
 351             self._helm_command, inspect_command, kdu_model, repo_str, version
 352         )
 353         return inspect_command
 354
 355     async def _status_kdu(
 356         self,
 357         cluster_id: str,
 358         kdu_instance: str,
 359         namespace: str = None,
 360         show_error_log: bool = False,
 361         return_text: bool = False,
 362     ):
 363
 364         self.log.debug(
 365             "status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
 366         )
 367
 368         if not namespace:
 369             namespace = "kube-system"
 370
 371         # init config, env
 372         paths, env = self._init_paths_env(
 373             cluster_name=cluster_id, create_if_not_exist=True
 374         )
 375         command = "env KUBECONFIG={} {} status {} --namespace={} --output yaml".format(
 376             paths["kube_config"], self._helm_command, kdu_instance, namespace
 377         )
 378
 379         output, rc = await self._local_async_exec(
 380             command=command,
 381             raise_exception_on_error=True,
 382             show_error_log=show_error_log,
 383             env=env,
 384         )
 385
 386         if return_text:
 387             return str(output)
 388
 389         if rc != 0:
 390             return None
 391
 392         data = yaml.load(output, Loader=yaml.SafeLoader)
 393
 394         # remove field 'notes' and manifest
 395         try:
 396             del data.get("info")["notes"]
 397             del data["manifest"]
 398         except KeyError:
 399             pass
 400
 401         # unable to parse 'resources' as currently it is not included in helm3
 402         return data
 403
 404     def _get_install_command(
 405         self,
 406         kdu_model: str,
 407         kdu_instance: str,
 408         namespace: str,
 409         params_str: str,
 410         version: str,
 411         atomic: bool,
 412         timeout: float,
 413         kubeconfig: str,
 414     ) -> str:
 415
 416         timeout_str = ""
 417         if timeout:
 418             timeout_str = "--timeout {}s".format(timeout)
 419
 420         # atomic
 421         atomic_str = ""
 422         if atomic:
 423             atomic_str = "--atomic"
 424         # namespace
 425         namespace_str = ""
 426         if namespace:
 427             namespace_str = "--namespace {}".format(namespace)
 428
 429         # version
 430         version_str = ""
 431         if version:
 432             version_str = "--version {}".format(version)
 433
 434         command = (
 435             "env KUBECONFIG={kubeconfig} {helm} install {name} {atomic} --output yaml  "
 436             "{params} {timeout} {ns} {model} {ver}".format(
 437                 kubeconfig=kubeconfig,
 438                 helm=self._helm_command,
 439                 name=kdu_instance,
 440                 atomic=atomic_str,
 441                 params=params_str,
 442                 timeout=timeout_str,
 443                 ns=namespace_str,
 444                 model=kdu_model,
 445                 ver=version_str,
 446             )
 447         )
 448         return command
 449
 450     def _get_upgrade_command(
 451         self,
 452         kdu_model: str,
 453         kdu_instance: str,
 454         namespace: str,
 455         params_str: str,
 456         version: str,
 457         atomic: bool,
 458         timeout: float,
 459         kubeconfig: str,
 460     ) -> str:
 461
 462         timeout_str = ""
 463         if timeout:
 464             timeout_str = "--timeout {}s".format(timeout)
 465
 466         # atomic
 467         atomic_str = ""
 468         if atomic:
 469             atomic_str = "--atomic"
 470
 471         # version
 472         version_str = ""
 473         if version:
 474             version_str = "--version {}".format(version)
 475
 476         # namespace
 477         namespace_str = ""
 478         if namespace:
 479             namespace_str = "--namespace {}".format(namespace)
 480
 481         command = (
 482             "env KUBECONFIG={kubeconfig} {helm} upgrade {name} {model} {namespace} {atomic} "
 483             "--output yaml {params} {timeout} {ver}"
 484         ).format(
 485             kubeconfig=kubeconfig,
 486             helm=self._helm_command,
 487             name=kdu_instance,
 488             namespace=namespace_str,
 489             atomic=atomic_str,
 490             params=params_str,
 491             timeout=timeout_str,
 492             model=kdu_model,
 493             ver=version_str,
 494         )
 495         return command
 496
 497     def _get_rollback_command(
 498         self, kdu_instance: str, namespace: str, revision: float, kubeconfig: str
 499     ) -> str:
 500         return "env KUBECONFIG={} {} rollback {} {} --namespace={} --wait".format(
 501             kubeconfig, self._helm_command, kdu_instance, revision, namespace
 502         )
 503
 504     def _get_uninstall_command(
 505         self, kdu_instance: str, namespace: str, kubeconfig: str
 506     ) -> str:
 507
 508         return "env KUBECONFIG={} {} uninstall {} --namespace={}".format(
 509             kubeconfig, self._helm_command, kdu_instance, namespace
 510         )
 511
 512     def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
 513         repo_ids = []
 514         cluster_filter = {"_admin.helm-chart-v3.id": cluster_uuid}
 515         cluster = self.db.get_one("k8sclusters", cluster_filter)
 516         if cluster:
 517             repo_ids = cluster.get("_admin").get("helm_chart_repos") or []
 518             return repo_ids
 519         else:
 520             raise K8sException(
 521                 "k8cluster with helm-id : {} not found".format(cluster_uuid)
 522             )