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