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
 102         self.log.debug("installing {} in cluster {}".format(kdu_model, cluster_uuid))
 103
 104         # sync local dir
 105         self.fs.sync(from_path=cluster_uuid)
 106
 107         # init env, paths
 108         paths, env = self._init_paths_env(
 109             cluster_name=cluster_uuid, 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_uuid, namespace):
 115                 try:
 116                     await self._create_namespace(cluster_uuid, namespace)
 117                 except Exception as e:
 118                     if not await self._namespace_exists(cluster_uuid, 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_uuid,
 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_uuid)
 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         repo_list = await self.repo_list(cluster_id)
 319         stable_repo = [repo for repo in repo_list if repo["name"] == "stable"]
 320         if not stable_repo and self._stable_repo_url:
 321             await self.repo_add(cluster_id, "stable", self._stable_repo_url)
 322
 323         # Returns False as no software needs to be uninstalled
 324         return False
 325
 326     async def _uninstall_sw(self, cluster_id: str, namespace: str):
 327         # nothing to do to uninstall sw
 328         pass
 329
 330     async def _instances_list(self, cluster_id: str):
 331
 332         # init paths, env
 333         paths, env = self._init_paths_env(
 334             cluster_name=cluster_id, create_if_not_exist=True
 335         )
 336
 337         command = "{} list --all-namespaces  --output yaml".format(self._helm_command)
 338         output, _rc = await self._local_async_exec(
 339             command=command, raise_exception_on_error=True, env=env
 340         )
 341
 342         if output and len(output) > 0:
 343             self.log.debug("instances list output: {}".format(output))
 344             return yaml.load(output, Loader=yaml.SafeLoader)
 345         else:
 346             return []
 347
 348     def _get_inspect_command(
 349         self, inspect_command: str, kdu_model: str, repo_str: str, version: str
 350     ):
 351         inspect_command = "{} show {} {}{} {}".format(
 352             self._helm_command, inspect_command, kdu_model, repo_str, version
 353         )
 354         return inspect_command
 355
 356     async def _status_kdu(
 357         self,
 358         cluster_id: str,
 359         kdu_instance: str,
 360         namespace: str = None,
 361         yaml_format: bool = False,
 362         show_error_log: bool = False,
 363     ) -> Union[str, dict]:
 364
 365         self.log.debug(
 366             "status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
 367         )
 368
 369         if not namespace:
 370             namespace = "kube-system"
 371
 372         # init config, env
 373         paths, env = self._init_paths_env(
 374             cluster_name=cluster_id, create_if_not_exist=True
 375         )
 376         command = "env KUBECONFIG={} {} status {} --namespace={} --output yaml".format(
 377             paths["kube_config"], self._helm_command, kdu_instance, namespace
 378         )
 379
 380         output, rc = await self._local_async_exec(
 381             command=command,
 382             raise_exception_on_error=True,
 383             show_error_log=show_error_log,
 384             env=env,
 385         )
 386
 387         if yaml_format:
 388             return str(output)
 389
 390         if rc != 0:
 391             return None
 392
 393         data = yaml.load(output, Loader=yaml.SafeLoader)
 394
 395         # remove field 'notes' and manifest
 396         try:
 397             del data.get("info")["notes"]
 398         except KeyError:
 399             pass
 400
 401         # parse the manifest to a list of dictionaries
 402         if "manifest" in data:
 403             manifest_str = data.get("manifest")
 404             manifest_docs = yaml.load_all(manifest_str, Loader=yaml.SafeLoader)
 405
 406             data["manifest"] = []
 407             for doc in manifest_docs:
 408                 data["manifest"].append(doc)
 409
 410         return data
 411
 412     def _get_install_command(
 413         self,
 414         kdu_model: str,
 415         kdu_instance: str,
 416         namespace: str,
 417         params_str: str,
 418         version: str,
 419         atomic: bool,
 420         timeout: float,
 421         kubeconfig: str,
 422     ) -> str:
 423
 424         timeout_str = ""
 425         if timeout:
 426             timeout_str = "--timeout {}s".format(timeout)
 427
 428         # atomic
 429         atomic_str = ""
 430         if atomic:
 431             atomic_str = "--atomic"
 432         # namespace
 433         namespace_str = ""
 434         if namespace:
 435             namespace_str = "--namespace {}".format(namespace)
 436
 437         # version
 438         version_str = ""
 439         if version:
 440             version_str = "--version {}".format(version)
 441
 442         command = (
 443             "env KUBECONFIG={kubeconfig} {helm} install {name} {atomic} --output yaml  "
 444             "{params} {timeout} {ns} {model} {ver}".format(
 445                 kubeconfig=kubeconfig,
 446                 helm=self._helm_command,
 447                 name=kdu_instance,
 448                 atomic=atomic_str,
 449                 params=params_str,
 450                 timeout=timeout_str,
 451                 ns=namespace_str,
 452                 model=kdu_model,
 453                 ver=version_str,
 454             )
 455         )
 456         return command
 457
 458     def _get_upgrade_command(
 459         self,
 460         kdu_model: str,
 461         kdu_instance: str,
 462         namespace: str,
 463         params_str: str,
 464         version: str,
 465         atomic: bool,
 466         timeout: float,
 467         kubeconfig: str,
 468     ) -> str:
 469
 470         timeout_str = ""
 471         if timeout:
 472             timeout_str = "--timeout {}s".format(timeout)
 473
 474         # atomic
 475         atomic_str = ""
 476         if atomic:
 477             atomic_str = "--atomic"
 478
 479         # version
 480         version_str = ""
 481         if version:
 482             version_str = "--version {}".format(version)
 483
 484         # namespace
 485         namespace_str = ""
 486         if namespace:
 487             namespace_str = "--namespace {}".format(namespace)
 488
 489         command = (
 490             "env KUBECONFIG={kubeconfig} {helm} upgrade {name} {model} {namespace} {atomic} "
 491             "--output yaml {params} {timeout} {ver}"
 492         ).format(
 493             kubeconfig=kubeconfig,
 494             helm=self._helm_command,
 495             name=kdu_instance,
 496             namespace=namespace_str,
 497             atomic=atomic_str,
 498             params=params_str,
 499             timeout=timeout_str,
 500             model=kdu_model,
 501             ver=version_str,
 502         )
 503         return command
 504
 505     def _get_rollback_command(
 506         self, kdu_instance: str, namespace: str, revision: float, kubeconfig: str
 507     ) -> str:
 508         return "env KUBECONFIG={} {} rollback {} {} --namespace={} --wait".format(
 509             kubeconfig, self._helm_command, kdu_instance, revision, namespace
 510         )
 511
 512     def _get_uninstall_command(
 513         self, kdu_instance: str, namespace: str, kubeconfig: str
 514     ) -> str:
 515
 516         return "env KUBECONFIG={} {} uninstall {} --namespace={}".format(
 517             kubeconfig, self._helm_command, kdu_instance, namespace
 518         )
 519
 520     def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
 521         repo_ids = []
 522         cluster_filter = {"_admin.helm-chart-v3.id": cluster_uuid}
 523         cluster = self.db.get_one("k8sclusters", cluster_filter)
 524         if cluster:
 525             repo_ids = cluster.get("_admin").get("helm_chart_repos") or []
 526             return repo_ids
 527         else:
 528             raise K8sException(
 529                 "k8cluster with helm-id : {} not found".format(cluster_uuid)
 530             )