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__(self,
  59                                       db=db,
  60                                       log=log,
  61                                       fs=fs,
  62                                       kubectl_command=kubectl_command,
  63                                       helm_command=helm_command,
  64                                       on_update_db=on_update_db)
  65
  66         self.log.info("K8S Helm3 connector initialized")
  67
  68     async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
  69
  70         self.log.debug(
  71             "inspect kdu_model {} from (optional) repo: {}".format(kdu_model, repo_url)
  72         )
  73
  74         return await self._exec_inspect_comand(
  75             inspect_command="all", kdu_model=kdu_model, repo_url=repo_url
  76         )
  77
  78     """
  79     ####################################################################################
  80     ################################### P R I V A T E ##################################
  81     ####################################################################################
  82     """
  83
  84     def _init_paths_env(self, cluster_name: str, create_if_not_exist: bool = True):
  85         """
  86         Creates and returns base cluster and kube dirs and returns them.
  87         Also created helm3 dirs according to new directory specification, paths are
  88         returned and also environment variables that must be provided to execute commands
  89
  90         Helm 3 directory specification uses XDG categories for variable support:
  91         - Cache: $XDG_CACHE_HOME, for example, ${HOME}/.cache/helm/
  92         - Configuration: $XDG_CONFIG_HOME, for example, ${HOME}/.config/helm/
  93         - Data: $XDG_DATA_HOME, for example ${HOME}/.local/share/helm
  94
  95         The variables assigned for this paths are:
  96         (In the documentation the variables names are $HELM_PATH_CACHE, $HELM_PATH_CONFIG,
  97         $HELM_PATH_DATA but looking and helm env the variable names are different)
  98         - Cache: $HELM_CACHE_HOME
  99         - Config: $HELM_CONFIG_HOME
 100         - Data: $HELM_DATA_HOME
 101         - helm kubeconfig: $KUBECONFIG
 102
 103         :param cluster_name:  cluster_name
 104         :return: Dictionary with config_paths and dictionary with helm environment variables
 105         """
 106
 107         base = self.fs.path
 108         if base.endswith("/") or base.endswith("\\"):
 109             base = base[:-1]
 110
 111         # base dir for cluster
 112         cluster_dir = base + "/" + cluster_name
 113
 114         # kube dir
 115         kube_dir = cluster_dir + "/" + ".kube"
 116         if create_if_not_exist and not os.path.exists(kube_dir):
 117             self.log.debug("Creating dir {}".format(kube_dir))
 118             os.makedirs(kube_dir)
 119
 120         helm_path_cache = cluster_dir + "/.cache/helm"
 121         if create_if_not_exist and not os.path.exists(helm_path_cache):
 122             self.log.debug("Creating dir {}".format(helm_path_cache))
 123             os.makedirs(helm_path_cache)
 124
 125         helm_path_config = cluster_dir + "/.config/helm"
 126         if create_if_not_exist and not os.path.exists(helm_path_config):
 127             self.log.debug("Creating dir {}".format(helm_path_config))
 128             os.makedirs(helm_path_config)
 129
 130         helm_path_data = cluster_dir + "/.local/share/helm"
 131         if create_if_not_exist and not os.path.exists(helm_path_data):
 132             self.log.debug("Creating dir {}".format(helm_path_data))
 133             os.makedirs(helm_path_data)
 134
 135         config_filename = kube_dir + "/config"
 136
 137         # 2 - Prepare dictionary with paths
 138         paths = {
 139             "kube_dir": kube_dir,
 140             "kube_config": config_filename,
 141             "cluster_dir": cluster_dir
 142         }
 143
 144         # 3 - Prepare environment variables
 145         env = {
 146             "HELM_CACHE_HOME": helm_path_cache,
 147             "HELM_CONFIG_HOME": helm_path_config,
 148             "HELM_DATA_HOME": helm_path_data,
 149             "KUBECONFIG": config_filename
 150         }
 151
 152         for file_name, file in paths.items():
 153             if "dir" in file_name and not os.path.exists(file):
 154                 err_msg = "{} dir does not exist".format(file)
 155                 self.log.error(err_msg)
 156                 raise K8sException(err_msg)
 157
 158         return paths, env
 159
 160     async def _get_namespaces(self,
 161                               cluster_id: str):
 162
 163         self.log.debug("get namespaces cluster_id {}".format(cluster_id))
 164
 165         # init config, env
 166         paths, env = self._init_paths_env(
 167             cluster_name=cluster_id, create_if_not_exist=True
 168         )
 169
 170         command = "{} --kubeconfig={} get namespaces -o=yaml".format(
 171             self.kubectl_command, paths["kube_config"]
 172         )
 173         output, _rc = await self._local_async_exec(
 174             command=command, raise_exception_on_error=True, env=env
 175         )
 176
 177         data = yaml.load(output, Loader=yaml.SafeLoader)
 178         namespaces = [item["metadata"]["name"] for item in data["items"]]
 179         self.log.debug(f"namespaces {namespaces}")
 180
 181         return namespaces
 182
 183     async def _create_namespace(self,
 184                                 cluster_id: str,
 185                                 namespace: str):
 186
 187         self.log.debug(f"create namespace: {cluster_id} for cluster_id: {namespace}")
 188
 189         # init config, env
 190         paths, env = self._init_paths_env(
 191             cluster_name=cluster_id, create_if_not_exist=True
 192         )
 193
 194         command = "{} --kubeconfig={} create namespace {}".format(
 195             self.kubectl_command, paths["kube_config"], namespace
 196         )
 197         _, _rc = await self._local_async_exec(
 198             command=command, raise_exception_on_error=True, env=env
 199         )
 200         self.log.debug(f"namespace {namespace} created")
 201
 202         return _rc
 203
 204     async def _get_services(self, cluster_id: str, kdu_instance: str, namespace: str):
 205
 206         # init config, env
 207         paths, env = self._init_paths_env(
 208             cluster_name=cluster_id, create_if_not_exist=True
 209         )
 210
 211         command1 = "{} get manifest {} --namespace={}".format(
 212             self._helm_command, kdu_instance, namespace
 213         )
 214         command2 = "{} get --namespace={} -f -".format(
 215             self.kubectl_command, namespace
 216         )
 217         output, _rc = await self._local_async_exec_pipe(
 218             command1, command2, env=env, raise_exception_on_error=True
 219         )
 220         services = self._parse_services(output)
 221
 222         return services
 223
 224     async def _cluster_init(self, cluster_id, namespace, paths, env):
 225         """
 226         Implements the helm version dependent cluster initialization:
 227         For helm3 it creates the namespace if it is not created
 228         """
 229         if namespace != "kube-system":
 230             namespaces = await self._get_namespaces(cluster_id)
 231             if namespace not in namespaces:
 232                 await self._create_namespace(cluster_id, namespace)
 233
 234         # If default repo is not included add
 235         cluster_uuid = "{}:{}".format(namespace, cluster_id)
 236         repo_list = await self.repo_list(cluster_uuid)
 237         for repo in repo_list:
 238             self.log.debug("repo")
 239             if repo["name"] == "stable":
 240                 self.log.debug("Default repo already present")
 241                 break
 242         else:
 243             await self.repo_add(cluster_uuid,
 244                                 "stable",
 245                                 "https://kubernetes-charts.storage.googleapis.com/")
 246
 247         # Returns False as no software needs to be uninstalled
 248         return False
 249
 250     async def _uninstall_sw(self, cluster_id: str, namespace: str):
 251         # nothing to do to uninstall sw
 252         pass
 253
 254     async def _instances_list(self, cluster_id: str):
 255
 256         # init paths, env
 257         paths, env = self._init_paths_env(
 258             cluster_name=cluster_id, create_if_not_exist=True
 259         )
 260
 261         command = "{} list --all-namespaces  --output yaml".format(
 262             self._helm_command
 263         )
 264         output, _rc = await self._local_async_exec(
 265             command=command, raise_exception_on_error=True, env=env
 266         )
 267
 268         if output and len(output) > 0:
 269             self.log.debug("instances list output: {}".format(output))
 270             return yaml.load(output, Loader=yaml.SafeLoader)
 271         else:
 272             return []
 273
 274     def _get_inspect_command(self, inspect_command: str, kdu_model: str, repo_str: str,
 275                              version: str):
 276         inspect_command = "{} show {} {}{} {}".format(
 277             self._helm_command, inspect_command, kdu_model, repo_str, version
 278         )
 279         return inspect_command
 280
 281     async def _status_kdu(
 282         self,
 283         cluster_id: str,
 284         kdu_instance: str,
 285         namespace: str = None,
 286         show_error_log: bool = False,
 287         return_text: bool = False,
 288     ):
 289
 290         self.log.debug("status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace))
 291
 292         if not namespace:
 293             namespace = "kube-system"
 294
 295         # init config, env
 296         paths, env = self._init_paths_env(
 297             cluster_name=cluster_id, create_if_not_exist=True
 298         )
 299         command = "{} status {} --namespace={} --output yaml".format(
 300             self._helm_command, kdu_instance, namespace
 301         )
 302
 303         output, rc = await self._local_async_exec(
 304             command=command,
 305             raise_exception_on_error=True,
 306             show_error_log=show_error_log,
 307             env=env
 308         )
 309
 310         if return_text:
 311             return str(output)
 312
 313         if rc != 0:
 314             return None
 315
 316         data = yaml.load(output, Loader=yaml.SafeLoader)
 317
 318         # remove field 'notes' and manifest
 319         try:
 320             del data.get("info")["notes"]
 321             del data["manifest"]
 322         except KeyError:
 323             pass
 324
 325         # unable to parse 'resources' as currently it is not included in helm3
 326         return data
 327
 328     def _get_install_command(self, kdu_model: str, kdu_instance: str, namespace: str,
 329                              params_str: str, version: str, atomic: bool, timeout: float) -> str:
 330
 331         timeout_str = ""
 332         if timeout:
 333             timeout_str = "--timeout {}s".format(timeout)
 334
 335         # atomic
 336         atomic_str = ""
 337         if atomic:
 338             atomic_str = "--atomic"
 339         # namespace
 340         namespace_str = ""
 341         if namespace:
 342             namespace_str = "--namespace {}".format(namespace)
 343
 344         # version
 345         version_str = ""
 346         if version:
 347             version_str = "--version {}".format(version)
 348
 349         command = (
 350             "{helm} install {name} {atomic} --output yaml  "
 351             "{params} {timeout} {ns} {model} {ver}".format(
 352                 helm=self._helm_command,
 353                 name=kdu_instance,
 354                 atomic=atomic_str,
 355                 params=params_str,
 356                 timeout=timeout_str,
 357                 ns=namespace_str,
 358                 model=kdu_model,
 359                 ver=version_str,
 360             )
 361         )
 362         return command
 363
 364     def _get_upgrade_command(self, kdu_model: str, kdu_instance: str, namespace: str,
 365                              params_str: str, version: str, atomic: bool, timeout: float) -> str:
 366
 367         timeout_str = ""
 368         if timeout:
 369             timeout_str = "--timeout {}s".format(timeout)
 370
 371         # atomic
 372         atomic_str = ""
 373         if atomic:
 374             atomic_str = "--atomic"
 375
 376         # version
 377         version_str = ""
 378         if version:
 379             version_str = "--version {}".format(version)
 380
 381         # namespace
 382         namespace_str = ""
 383         if namespace:
 384             namespace_str = "--namespace {}".format(namespace)
 385
 386         command = (
 387             "{helm} upgrade {name} {model} {namespace} {atomic} --output yaml {params} "
 388             "{timeout}  {ver}".format(
 389                 helm=self._helm_command,
 390                 name=kdu_instance,
 391                 namespace=namespace_str,
 392                 atomic=atomic_str,
 393                 params=params_str,
 394                 timeout=timeout_str,
 395                 model=kdu_model,
 396                 ver=version_str,
 397             )
 398         )
 399         return command
 400
 401     def _get_rollback_command(self, kdu_instance: str, namespace: str, revision: float) -> str:
 402         return "{} rollback {} {} --namespace={} --wait".format(
 403             self._helm_command, kdu_instance, revision, namespace
 404         )
 405
 406     def _get_uninstall_command(self, kdu_instance: str, namespace: str) -> str:
 407
 408         return "{} uninstall {} --namespace={}".format(
 409             self._helm_command, kdu_instance, namespace)