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         _, cluster_id = self._get_namespace_cluster_id(cluster_uuid)
 101         self.log.debug("installing {} in cluster {}".format(kdu_model, cluster_id))
 102
 103         # sync local dir
 104         self.fs.sync(from_path=cluster_id)
 105
 106         # init env, paths
 107         paths, env = self._init_paths_env(
 108             cluster_name=cluster_id, 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             namespaces = await self._get_namespaces(cluster_id)
 114             if namespace not in namespaces:
 115                 await self._create_namespace(cluster_id, namespace)
 116
 117         await self._install_impl(
 118             cluster_id,
 119             kdu_model,
 120             paths,
 121             env,
 122             kdu_instance,
 123             atomic=atomic,
 124             timeout=timeout,
 125             params=params,
 126             db_dict=db_dict,
 127             kdu_name=kdu_name,
 128             namespace=namespace,
 129         )
 130
 131         # sync fs
 132         self.fs.reverse_sync(from_path=cluster_id)
 133
 134         self.log.debug("Returning kdu_instance {}".format(kdu_instance))
 135         return True
 136
 137     async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
 138
 139         self.log.debug(
 140             "inspect kdu_model {} from (optional) repo: {}".format(kdu_model, repo_url)
 141         )
 142
 143         return await self._exec_inspect_comand(
 144             inspect_command="all", kdu_model=kdu_model, repo_url=repo_url
 145         )
 146
 147     """
 148     ####################################################################################
 149     ################################### P R I V A T E ##################################
 150     ####################################################################################
 151     """
 152
 153     def _init_paths_env(self, cluster_name: str, create_if_not_exist: bool = True):
 154         """
 155         Creates and returns base cluster and kube dirs and returns them.
 156         Also created helm3 dirs according to new directory specification, paths are
 157         returned and also environment variables that must be provided to execute commands
 158
 159         Helm 3 directory specification uses XDG categories for variable support:
 160         - Cache: $XDG_CACHE_HOME, for example, ${HOME}/.cache/helm/
 161         - Configuration: $XDG_CONFIG_HOME, for example, ${HOME}/.config/helm/
 162         - Data: $XDG_DATA_HOME, for example ${HOME}/.local/share/helm
 163
 164         The variables assigned for this paths are:
 165         (In the documentation the variables names are $HELM_PATH_CACHE, $HELM_PATH_CONFIG,
 166         $HELM_PATH_DATA but looking and helm env the variable names are different)
 167         - Cache: $HELM_CACHE_HOME
 168         - Config: $HELM_CONFIG_HOME
 169         - Data: $HELM_DATA_HOME
 170         - helm kubeconfig: $KUBECONFIG
 171
 172         :param cluster_name:  cluster_name
 173         :return: Dictionary with config_paths and dictionary with helm environment variables
 174         """
 175
 176         base = self.fs.path
 177         if base.endswith("/") or base.endswith("\\"):
 178             base = base[:-1]
 179
 180         # base dir for cluster
 181         cluster_dir = base + "/" + cluster_name
 182
 183         # kube dir
 184         kube_dir = cluster_dir + "/" + ".kube"
 185         if create_if_not_exist and not os.path.exists(kube_dir):
 186             self.log.debug("Creating dir {}".format(kube_dir))
 187             os.makedirs(kube_dir)
 188
 189         helm_path_cache = cluster_dir + "/.cache/helm"
 190         if create_if_not_exist and not os.path.exists(helm_path_cache):
 191             self.log.debug("Creating dir {}".format(helm_path_cache))
 192             os.makedirs(helm_path_cache)
 193
 194         helm_path_config = cluster_dir + "/.config/helm"
 195         if create_if_not_exist and not os.path.exists(helm_path_config):
 196             self.log.debug("Creating dir {}".format(helm_path_config))
 197             os.makedirs(helm_path_config)
 198
 199         helm_path_data = cluster_dir + "/.local/share/helm"
 200         if create_if_not_exist and not os.path.exists(helm_path_data):
 201             self.log.debug("Creating dir {}".format(helm_path_data))
 202             os.makedirs(helm_path_data)
 203
 204         config_filename = kube_dir + "/config"
 205
 206         # 2 - Prepare dictionary with paths
 207         paths = {
 208             "kube_dir": kube_dir,
 209             "kube_config": config_filename,
 210             "cluster_dir": cluster_dir,
 211         }
 212
 213         # 3 - Prepare environment variables
 214         env = {
 215             "HELM_CACHE_HOME": helm_path_cache,
 216             "HELM_CONFIG_HOME": helm_path_config,
 217             "HELM_DATA_HOME": helm_path_data,
 218             "KUBECONFIG": config_filename,
 219         }
 220
 221         for file_name, file in paths.items():
 222             if "dir" in file_name and not os.path.exists(file):
 223                 err_msg = "{} dir does not exist".format(file)
 224                 self.log.error(err_msg)
 225                 raise K8sException(err_msg)
 226
 227         return paths, env
 228
 229     async def _get_namespaces(self, cluster_id: str):
 230
 231         self.log.debug("get namespaces cluster_id {}".format(cluster_id))
 232
 233         # init config, env
 234         paths, env = self._init_paths_env(
 235             cluster_name=cluster_id, create_if_not_exist=True
 236         )
 237
 238         command = "{} --kubeconfig={} get namespaces -o=yaml".format(
 239             self.kubectl_command, paths["kube_config"]
 240         )
 241         output, _rc = await self._local_async_exec(
 242             command=command, raise_exception_on_error=True, env=env
 243         )
 244
 245         data = yaml.load(output, Loader=yaml.SafeLoader)
 246         namespaces = [item["metadata"]["name"] for item in data["items"]]
 247         self.log.debug(f"namespaces {namespaces}")
 248
 249         return namespaces
 250
 251     async def _create_namespace(self, cluster_id: str, namespace: str):
 252
 253         self.log.debug(f"create namespace: {cluster_id} for cluster_id: {namespace}")
 254
 255         # init config, env
 256         paths, env = self._init_paths_env(
 257             cluster_name=cluster_id, create_if_not_exist=True
 258         )
 259
 260         command = "{} --kubeconfig={} create namespace {}".format(
 261             self.kubectl_command, paths["kube_config"], namespace
 262         )
 263         _, _rc = await self._local_async_exec(
 264             command=command, raise_exception_on_error=True, env=env
 265         )
 266         self.log.debug(f"namespace {namespace} created")
 267
 268         return _rc
 269
 270     async def _get_services(self, cluster_id: str, kdu_instance: str, namespace: str):
 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         command1 = "{} get manifest {} --namespace={}".format(
 278             self._helm_command, kdu_instance, namespace
 279         )
 280         command2 = "{} get --namespace={} -f -".format(self.kubectl_command, namespace)
 281         output, _rc = await self._local_async_exec_pipe(
 282             command1, command2, env=env, raise_exception_on_error=True
 283         )
 284         services = self._parse_services(output)
 285
 286         return services
 287
 288     async def _cluster_init(self, cluster_id, namespace, paths, env):
 289         """
 290         Implements the helm version dependent cluster initialization:
 291         For helm3 it creates the namespace if it is not created
 292         """
 293         if namespace != "kube-system":
 294             namespaces = await self._get_namespaces(cluster_id)
 295             if namespace not in namespaces:
 296                 await self._create_namespace(cluster_id, namespace)
 297
 298         # If default repo is not included add
 299         cluster_uuid = "{}:{}".format(namespace, cluster_id)
 300         repo_list = await self.repo_list(cluster_uuid)
 301         stable_repo = [repo for repo in repo_list if repo["name"] == "stable"]
 302         if not stable_repo and self._stable_repo_url:
 303             await self.repo_add(cluster_uuid, "stable", self._stable_repo_url)
 304
 305         # Returns False as no software needs to be uninstalled
 306         return False
 307
 308     async def _uninstall_sw(self, cluster_id: str, namespace: str):
 309         # nothing to do to uninstall sw
 310         pass
 311
 312     async def _instances_list(self, cluster_id: str):
 313
 314         # init paths, env
 315         paths, env = self._init_paths_env(
 316             cluster_name=cluster_id, create_if_not_exist=True
 317         )
 318
 319         command = "{} list --all-namespaces  --output yaml".format(self._helm_command)
 320         output, _rc = await self._local_async_exec(
 321             command=command, raise_exception_on_error=True, env=env
 322         )
 323
 324         if output and len(output) > 0:
 325             self.log.debug("instances list output: {}".format(output))
 326             return yaml.load(output, Loader=yaml.SafeLoader)
 327         else:
 328             return []
 329
 330     def _get_inspect_command(
 331         self, inspect_command: str, kdu_model: str, repo_str: str, version: str
 332     ):
 333         inspect_command = "{} show {} {}{} {}".format(
 334             self._helm_command, inspect_command, kdu_model, repo_str, version
 335         )
 336         return inspect_command
 337
 338     async def _status_kdu(
 339         self,
 340         cluster_id: str,
 341         kdu_instance: str,
 342         namespace: str = None,
 343         show_error_log: bool = False,
 344         return_text: bool = False,
 345     ):
 346
 347         self.log.debug(
 348             "status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
 349         )
 350
 351         if not namespace:
 352             namespace = "kube-system"
 353
 354         # init config, env
 355         paths, env = self._init_paths_env(
 356             cluster_name=cluster_id, create_if_not_exist=True
 357         )
 358         command = "{} status {} --namespace={} --output yaml".format(
 359             self._helm_command, kdu_instance, namespace
 360         )
 361
 362         output, rc = await self._local_async_exec(
 363             command=command,
 364             raise_exception_on_error=True,
 365             show_error_log=show_error_log,
 366             env=env,
 367         )
 368
 369         if return_text:
 370             return str(output)
 371
 372         if rc != 0:
 373             return None
 374
 375         data = yaml.load(output, Loader=yaml.SafeLoader)
 376
 377         # remove field 'notes' and manifest
 378         try:
 379             del data.get("info")["notes"]
 380             del data["manifest"]
 381         except KeyError:
 382             pass
 383
 384         # unable to parse 'resources' as currently it is not included in helm3
 385         return data
 386
 387     def _get_install_command(
 388         self,
 389         kdu_model: str,
 390         kdu_instance: str,
 391         namespace: str,
 392         params_str: str,
 393         version: str,
 394         atomic: bool,
 395         timeout: float,
 396     ) -> str:
 397
 398         timeout_str = ""
 399         if timeout:
 400             timeout_str = "--timeout {}s".format(timeout)
 401
 402         # atomic
 403         atomic_str = ""
 404         if atomic:
 405             atomic_str = "--atomic"
 406         # namespace
 407         namespace_str = ""
 408         if namespace:
 409             namespace_str = "--namespace {}".format(namespace)
 410
 411         # version
 412         version_str = ""
 413         if version:
 414             version_str = "--version {}".format(version)
 415
 416         command = (
 417             "{helm} install {name} {atomic} --output yaml  "
 418             "{params} {timeout} {ns} {model} {ver}".format(
 419                 helm=self._helm_command,
 420                 name=kdu_instance,
 421                 atomic=atomic_str,
 422                 params=params_str,
 423                 timeout=timeout_str,
 424                 ns=namespace_str,
 425                 model=kdu_model,
 426                 ver=version_str,
 427             )
 428         )
 429         return command
 430
 431     def _get_upgrade_command(
 432         self,
 433         kdu_model: str,
 434         kdu_instance: str,
 435         namespace: str,
 436         params_str: str,
 437         version: str,
 438         atomic: bool,
 439         timeout: float,
 440     ) -> str:
 441
 442         timeout_str = ""
 443         if timeout:
 444             timeout_str = "--timeout {}s".format(timeout)
 445
 446         # atomic
 447         atomic_str = ""
 448         if atomic:
 449             atomic_str = "--atomic"
 450
 451         # version
 452         version_str = ""
 453         if version:
 454             version_str = "--version {}".format(version)
 455
 456         # namespace
 457         namespace_str = ""
 458         if namespace:
 459             namespace_str = "--namespace {}".format(namespace)
 460
 461         command = (
 462             "{helm} upgrade {name} {model} {namespace} {atomic} --output yaml {params} "
 463             "{timeout}  {ver}".format(
 464                 helm=self._helm_command,
 465                 name=kdu_instance,
 466                 namespace=namespace_str,
 467                 atomic=atomic_str,
 468                 params=params_str,
 469                 timeout=timeout_str,
 470                 model=kdu_model,
 471                 ver=version_str,
 472             )
 473         )
 474         return command
 475
 476     def _get_rollback_command(
 477         self, kdu_instance: str, namespace: str, revision: float
 478     ) -> str:
 479         return "{} rollback {} {} --namespace={} --wait".format(
 480             self._helm_command, kdu_instance, revision, namespace
 481         )
 482
 483     def _get_uninstall_command(self, kdu_instance: str, namespace: str) -> str:
 484
 485         return "{} uninstall {} --namespace={}".format(
 486             self._helm_command, kdu_instance, namespace
 487         )
 488
 489     def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
 490         repo_ids = []
 491         cluster_filter = {"_admin.helm-chart-v3.id": cluster_uuid}
 492         cluster = self.db.get_one("k8sclusters", cluster_filter)
 493         if cluster:
 494             repo_ids = cluster.get("_admin").get("helm_chart_repos") or []
 495             return repo_ids
 496         else:
 497             raise K8sException(
 498                 "k8cluster with helm-id : {} not found".format(cluster_uuid)
 499             )