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             if not await self._namespace_exists(cluster_id, namespace):
 114                 try:
 115                     await self._create_namespace(cluster_id, namespace)
 116                 except Exception as e:
 117                     if not await self._namespace_exists(cluster_id, 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_id,
 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_id)
 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         # If default repo is not included add
 318         cluster_uuid = "{}:{}".format(namespace, cluster_id)
 319         repo_list = await self.repo_list(cluster_uuid)
 320         stable_repo = [repo for repo in repo_list if repo["name"] == "stable"]
 321         if not stable_repo and self._stable_repo_url:
 322             await self.repo_add(cluster_uuid, "stable", self._stable_repo_url)
 323
 324         # Returns False as no software needs to be uninstalled
 325         return False
 326
 327     async def _uninstall_sw(self, cluster_id: str, namespace: str):
 328         # nothing to do to uninstall sw
 329         pass
 330
 331     async def _instances_list(self, cluster_id: str):
 332
 333         # init paths, env
 334         paths, env = self._init_paths_env(
 335             cluster_name=cluster_id, create_if_not_exist=True
 336         )
 337
 338         command = "{} list --all-namespaces  --output yaml".format(self._helm_command)
 339         output, _rc = await self._local_async_exec(
 340             command=command, raise_exception_on_error=True, env=env
 341         )
 342
 343         if output and len(output) > 0:
 344             self.log.debug("instances list output: {}".format(output))
 345             return yaml.load(output, Loader=yaml.SafeLoader)
 346         else:
 347             return []
 348
 349     def _get_inspect_command(
 350         self, inspect_command: str, kdu_model: str, repo_str: str, version: str
 351     ):
 352         inspect_command = "{} show {} {}{} {}".format(
 353             self._helm_command, inspect_command, kdu_model, repo_str, version
 354         )
 355         return inspect_command
 356
 357     async def _status_kdu(
 358         self,
 359         cluster_id: str,
 360         kdu_instance: str,
 361         namespace: str = None,
 362         show_error_log: bool = False,
 363         return_text: bool = False,
 364     ):
 365
 366         self.log.debug(
 367             "status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
 368         )
 369
 370         if not namespace:
 371             namespace = "kube-system"
 372
 373         # init config, env
 374         paths, env = self._init_paths_env(
 375             cluster_name=cluster_id, create_if_not_exist=True
 376         )
 377         command = "env KUBECONFIG={} {} status {} --namespace={} --output yaml".format(
 378             paths["kube_config"], self._helm_command, kdu_instance, namespace
 379         )
 380
 381         output, rc = await self._local_async_exec(
 382             command=command,
 383             raise_exception_on_error=True,
 384             show_error_log=show_error_log,
 385             env=env,
 386         )
 387
 388         if return_text:
 389             return str(output)
 390
 391         if rc != 0:
 392             return None
 393
 394         data = yaml.load(output, Loader=yaml.SafeLoader)
 395
 396         # remove field 'notes' and manifest
 397         try:
 398             del data.get("info")["notes"]
 399             del data["manifest"]
 400         except KeyError:
 401             pass
 402
 403         # unable to parse 'resources' as currently it is not included in helm3
 404         return data
 405
 406     def _get_install_command(
 407         self,
 408         kdu_model: str,
 409         kdu_instance: str,
 410         namespace: str,
 411         params_str: str,
 412         version: str,
 413         atomic: bool,
 414         timeout: float,
 415         kubeconfig: str,
 416     ) -> str:
 417
 418         timeout_str = ""
 419         if timeout:
 420             timeout_str = "--timeout {}s".format(timeout)
 421
 422         # atomic
 423         atomic_str = ""
 424         if atomic:
 425             atomic_str = "--atomic"
 426         # namespace
 427         namespace_str = ""
 428         if namespace:
 429             namespace_str = "--namespace {}".format(namespace)
 430
 431         # version
 432         version_str = ""
 433         if version:
 434             version_str = "--version {}".format(version)
 435
 436         command = (
 437             "env KUBECONFIG={kubeconfig} {helm} install {name} {atomic} --output yaml  "
 438             "{params} {timeout} {ns} {model} {ver}".format(
 439                 kubeconfig=kubeconfig,
 440                 helm=self._helm_command,
 441                 name=kdu_instance,
 442                 atomic=atomic_str,
 443                 params=params_str,
 444                 timeout=timeout_str,
 445                 ns=namespace_str,
 446                 model=kdu_model,
 447                 ver=version_str,
 448             )
 449         )
 450         return command
 451
 452     def _get_upgrade_command(
 453         self,
 454         kdu_model: str,
 455         kdu_instance: str,
 456         namespace: str,
 457         params_str: str,
 458         version: str,
 459         atomic: bool,
 460         timeout: float,
 461         kubeconfig: str,
 462     ) -> str:
 463
 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
 510         return "env KUBECONFIG={} {} uninstall {} --namespace={}".format(
 511             kubeconfig, self._helm_command, kdu_instance, namespace
 512         )
 513
 514     def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
 515         repo_ids = []
 516         cluster_filter = {"_admin.helm-chart-v3.id": cluster_uuid}
 517         cluster = self.db.get_one("k8sclusters", cluster_filter)
 518         if cluster:
 519             repo_ids = cluster.get("_admin").get("helm_chart_repos") or []
 520             return repo_ids
 521         else:
 522             raise K8sException(
 523                 "k8cluster with helm-id : {} not found".format(cluster_uuid)
 524             )