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