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         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         self.log.debug("get namespaces cluster_id {}".format(cluster_id))
 248
 249         # init config, env
 250         paths, env = self._init_paths_env(
 251             cluster_name=cluster_id, create_if_not_exist=True
 252         )
 253
 254         command = "{} --kubeconfig={} get namespaces -o=yaml".format(
 255             self.kubectl_command, paths["kube_config"]
 256         )
 257         output, _rc = await self._local_async_exec(
 258             command=command, raise_exception_on_error=True, env=env
 259         )
 260
 261         data = yaml.load(output, Loader=yaml.SafeLoader)
 262         namespaces = [item["metadata"]["name"] for item in data["items"]]
 263         self.log.debug(f"namespaces {namespaces}")
 264
 265         return namespaces
 266
 267     async def _create_namespace(self, cluster_id: str, namespace: str):
 268         self.log.debug(f"create namespace: {cluster_id} for cluster_id: {namespace}")
 269
 270         # init config, env
 271         paths, env = self._init_paths_env(
 272             cluster_name=cluster_id, create_if_not_exist=True
 273         )
 274
 275         command = "{} --kubeconfig={} create namespace {}".format(
 276             self.kubectl_command, paths["kube_config"], namespace
 277         )
 278         _, _rc = await self._local_async_exec(
 279             command=command, raise_exception_on_error=True, env=env
 280         )
 281         self.log.debug(f"namespace {namespace} created")
 282
 283         return _rc
 284
 285     async def _get_services(
 286         self, cluster_id: str, kdu_instance: str, namespace: str, kubeconfig: str
 287     ):
 288         # init config, env
 289         paths, env = self._init_paths_env(
 290             cluster_name=cluster_id, create_if_not_exist=True
 291         )
 292
 293         command1 = "env KUBECONFIG={} {} get manifest {} --namespace={}".format(
 294             kubeconfig, self._helm_command, kdu_instance, namespace
 295         )
 296         command2 = "{} get --namespace={} -f -".format(self.kubectl_command, namespace)
 297         output, _rc = await self._local_async_exec_pipe(
 298             command1, command2, env=env, raise_exception_on_error=True
 299         )
 300         services = self._parse_services(output)
 301
 302         return services
 303
 304     async def _cluster_init(self, cluster_id, namespace, paths, env):
 305         """
 306         Implements the helm version dependent cluster initialization:
 307         For helm3 it creates the namespace if it is not created
 308         """
 309         if namespace != "kube-system":
 310             namespaces = await self._get_namespaces(cluster_id)
 311             if namespace not in namespaces:
 312                 await self._create_namespace(cluster_id, namespace)
 313
 314         # If default repo is not included add
 315         cluster_uuid = "{}:{}".format(namespace, cluster_id)
 316         repo_list = await self.repo_list(cluster_uuid)
 317         stable_repo = [repo for repo in repo_list if repo["name"] == "stable"]
 318         if not stable_repo and self._stable_repo_url:
 319             await self.repo_add(cluster_uuid, "stable", self._stable_repo_url)
 320
 321         # Returns False as no software needs to be uninstalled
 322         return False
 323
 324     async def _uninstall_sw(self, cluster_id: str, namespace: str):
 325         # nothing to do to uninstall sw
 326         pass
 327
 328     async def _instances_list(self, cluster_id: str):
 329         # init paths, env
 330         paths, env = self._init_paths_env(
 331             cluster_name=cluster_id, create_if_not_exist=True
 332         )
 333
 334         command = "{} list --all-namespaces  --output yaml".format(self._helm_command)
 335         output, _rc = await self._local_async_exec(
 336             command=command, raise_exception_on_error=True, env=env
 337         )
 338
 339         if output and len(output) > 0:
 340             self.log.debug("instances list output: {}".format(output))
 341             return yaml.load(output, Loader=yaml.SafeLoader)
 342         else:
 343             return []
 344
 345     def _get_inspect_command(
 346         self, inspect_command: str, kdu_model: str, repo_str: str, version: str
 347     ):
 348         inspect_command = "{} show {} {}{} {}".format(
 349             self._helm_command, inspect_command, kdu_model, repo_str, version
 350         )
 351         return inspect_command
 352
 353     async def _status_kdu(
 354         self,
 355         cluster_id: str,
 356         kdu_instance: str,
 357         namespace: str = None,
 358         yaml_format: bool = False,
 359         show_error_log: bool = False,
 360     ) -> Union[str, dict]:
 361         self.log.debug(
 362             "status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
 363         )
 364
 365         if not namespace:
 366             namespace = "kube-system"
 367
 368         # init config, env
 369         paths, env = self._init_paths_env(
 370             cluster_name=cluster_id, create_if_not_exist=True
 371         )
 372         command = "env KUBECONFIG={} {} status {} --namespace={} --output yaml".format(
 373             paths["kube_config"], self._helm_command, kdu_instance, namespace
 374         )
 375
 376         output, rc = await self._local_async_exec(
 377             command=command,
 378             raise_exception_on_error=True,
 379             show_error_log=show_error_log,
 380             env=env,
 381         )
 382
 383         if yaml_format:
 384             return str(output)
 385
 386         if rc != 0:
 387             return None
 388
 389         data = yaml.load(output, Loader=yaml.SafeLoader)
 390
 391         # remove field 'notes' and manifest
 392         try:
 393             del data.get("info")["notes"]
 394         except KeyError:
 395             pass
 396
 397         # parse the manifest to a list of dictionaries
 398         if "manifest" in data:
 399             manifest_str = data.get("manifest")
 400             manifest_docs = yaml.load_all(manifest_str, Loader=yaml.SafeLoader)
 401
 402             data["manifest"] = []
 403             for doc in manifest_docs:
 404                 data["manifest"].append(doc)
 405
 406         return data
 407
 408     def _get_install_command(
 409         self,
 410         kdu_model: str,
 411         kdu_instance: str,
 412         namespace: str,
 413         params_str: str,
 414         version: str,
 415         atomic: bool,
 416         timeout: float,
 417         kubeconfig: str,
 418     ) -> str:
 419         timeout_str = ""
 420         if timeout:
 421             timeout_str = "--timeout {}s".format(timeout)
 422
 423         # atomic
 424         atomic_str = ""
 425         if atomic:
 426             atomic_str = "--atomic"
 427         # namespace
 428         namespace_str = ""
 429         if namespace:
 430             namespace_str = "--namespace {}".format(namespace)
 431
 432         # version
 433         version_str = ""
 434         if version:
 435             version_str = "--version {}".format(version)
 436
 437         command = (
 438             "env KUBECONFIG={kubeconfig} {helm} install {name} {atomic} --output yaml  "
 439             "{params} {timeout} {ns} {model} {ver}".format(
 440                 kubeconfig=kubeconfig,
 441                 helm=self._helm_command,
 442                 name=kdu_instance,
 443                 atomic=atomic_str,
 444                 params=params_str,
 445                 timeout=timeout_str,
 446                 ns=namespace_str,
 447                 model=kdu_model,
 448                 ver=version_str,
 449             )
 450         )
 451         return command
 452
 453     def _get_upgrade_command(
 454         self,
 455         kdu_model: str,
 456         kdu_instance: str,
 457         namespace: str,
 458         params_str: str,
 459         version: str,
 460         atomic: bool,
 461         timeout: float,
 462         kubeconfig: str,
 463     ) -> str:
 464         timeout_str = ""
 465         if timeout:
 466             timeout_str = "--timeout {}s".format(timeout)
 467
 468         # atomic
 469         atomic_str = ""
 470         if atomic:
 471             atomic_str = "--atomic"
 472
 473         # version
 474         version_str = ""
 475         if version:
 476             version_str = "--version {}".format(version)
 477
 478         # namespace
 479         namespace_str = ""
 480         if namespace:
 481             namespace_str = "--namespace {}".format(namespace)
 482
 483         command = (
 484             "env KUBECONFIG={kubeconfig} {helm} upgrade {name} {model} {namespace} {atomic} "
 485             "--output yaml {params} {timeout} {ver}"
 486         ).format(
 487             kubeconfig=kubeconfig,
 488             helm=self._helm_command,
 489             name=kdu_instance,
 490             namespace=namespace_str,
 491             atomic=atomic_str,
 492             params=params_str,
 493             timeout=timeout_str,
 494             model=kdu_model,
 495             ver=version_str,
 496         )
 497         return command
 498
 499     def _get_rollback_command(
 500         self, kdu_instance: str, namespace: str, revision: float, kubeconfig: str
 501     ) -> str:
 502         return "env KUBECONFIG={} {} rollback {} {} --namespace={} --wait".format(
 503             kubeconfig, self._helm_command, kdu_instance, revision, namespace
 504         )
 505
 506     def _get_uninstall_command(
 507         self, kdu_instance: str, namespace: str, kubeconfig: str
 508     ) -> str:
 509         return "env KUBECONFIG={} {} uninstall {} --namespace={}".format(
 510             kubeconfig, self._helm_command, kdu_instance, namespace
 511         )
 512
 513     def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
 514         repo_ids = []
 515         cluster_filter = {"_admin.helm-chart-v3.id": cluster_uuid}
 516         cluster = self.db.get_one("k8sclusters", cluster_filter)
 517         if cluster:
 518             repo_ids = cluster.get("_admin").get("helm_chart_repos") or []
 519             return repo_ids
 520         else:
 521             raise K8sException(
 522                 "k8cluster with helm-id : {} not found".format(cluster_uuid)
 523             )