n2vc/k8s_helm_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 asyncio
  23 from typing import Union
  24 import os
  25 import yaml
  26
  27 from n2vc.k8s_helm_base_conn import K8sHelmBaseConnector
  28 from n2vc.exceptions import K8sException
  29
  30
  31 class K8sHelmConnector(K8sHelmBaseConnector):
  32
  33     """
  34     ####################################################################################
  35     ################################### P U B L I C ####################################
  36     ####################################################################################
  37     """
  38
  39     def __init__(
  40         self,
  41         fs: object,
  42         db: object,
  43         kubectl_command: str = "/usr/bin/kubectl",
  44         helm_command: str = "/usr/bin/helm",
  45         log: object = None,
  46         on_update_db=None,
  47     ):
  48         """
  49         Initializes helm connector for helm v2
  50
  51         :param fs: file system for kubernetes and helm configuration
  52         :param db: database object to write current operation status
  53         :param kubectl_command: path to kubectl executable
  54         :param helm_command: path to helm executable
  55         :param log: logger
  56         :param on_update_db: callback called when k8s connector updates database
  57         """
  58
  59         # parent class
  60         K8sHelmBaseConnector.__init__(
  61             self,
  62             db=db,
  63             log=log,
  64             fs=fs,
  65             kubectl_command=kubectl_command,
  66             helm_command=helm_command,
  67             on_update_db=on_update_db,
  68         )
  69
  70         self.log.info("Initializing K8S Helm2 connector")
  71
  72         # initialize helm client-only
  73         self.log.debug("Initializing helm client-only...")
  74         command = "{} init --client-only {} ".format(
  75             self._helm_command,
  76             "--stable-repo-url {}".format(self._stable_repo_url)
  77             if self._stable_repo_url
  78             else "--skip-repos",
  79         )
  80         try:
  81             asyncio.create_task(
  82                 self._local_async_exec(command=command, raise_exception_on_error=False)
  83             )
  84         except Exception as e:
  85             self.warning(
  86                 msg="helm init failed (it was already initialized): {}".format(e)
  87             )
  88
  89         self.log.info("K8S Helm2 connector initialized")
  90
  91     async def install(
  92         self,
  93         cluster_uuid: str,
  94         kdu_model: str,
  95         kdu_instance: str,
  96         atomic: bool = True,
  97         timeout: float = 300,
  98         params: dict = None,
  99         db_dict: dict = None,
 100         kdu_name: str = None,
 101         namespace: str = None,
 102         **kwargs,
 103     ):
 104         """
 105         Deploys of a new KDU instance. It would implicitly rely on the `install` call
 106         to deploy the Chart/Bundle properly parametrized (in practice, this call would
 107         happen before any _initial-config-primitive_of the VNF is called).
 108
 109         :param cluster_uuid: UUID of a K8s cluster known by OSM
 110         :param kdu_model: chart/reference (string), which can be either
 111             of these options:
 112             - a name of chart available via the repos known by OSM
 113               (e.g. stable/openldap, stable/openldap:1.2.4)
 114             - a path to a packaged chart (e.g. mychart.tgz)
 115             - a path to an unpacked chart directory or a URL (e.g. mychart)
 116         :param kdu_instance: Kdu instance name
 117         :param atomic: If set, installation process purges chart/bundle on fail, also
 118             will wait until all the K8s objects are active
 119         :param timeout: Time in seconds to wait for the install of the chart/bundle
 120             (defaults to Helm default timeout: 300s)
 121         :param params: dictionary of key-value pairs for instantiation parameters
 122             (overriding default values)
 123         :param dict db_dict: where to write into database when the status changes.
 124                         It contains a dict with {collection: <str>, filter: {},
 125                         path: <str>},
 126                             e.g. {collection: "nsrs", filter:
 127                             {_id: <nsd-id>, path: "_admin.deployed.K8S.3"}
 128         :param kdu_name: Name of the KDU instance to be installed
 129         :param namespace: K8s namespace to use for the KDU instance
 130         :param kwargs: Additional parameters (None yet)
 131         :return: True if successful
 132         """
 133         self.log.debug("installing {} in cluster {}".format(kdu_model, cluster_uuid))
 134
 135         # sync local dir
 136         self.fs.sync(from_path=cluster_uuid)
 137
 138         # init env, paths
 139         paths, env = self._init_paths_env(
 140             cluster_name=cluster_uuid, create_if_not_exist=True
 141         )
 142
 143         await self._install_impl(
 144             cluster_uuid,
 145             kdu_model,
 146             paths,
 147             env,
 148             kdu_instance,
 149             atomic=atomic,
 150             timeout=timeout,
 151             params=params,
 152             db_dict=db_dict,
 153             kdu_name=kdu_name,
 154             namespace=namespace,
 155         )
 156
 157         # sync fs
 158         self.fs.reverse_sync(from_path=cluster_uuid)
 159
 160         self.log.debug("Returning kdu_instance {}".format(kdu_instance))
 161         return True
 162
 163     async def inspect_kdu(self, kdu_model: str, repo_url: str = None) -> str:
 164         self.log.debug(
 165             "inspect kdu_model {} from (optional) repo: {}".format(kdu_model, repo_url)
 166         )
 167
 168         return await self._exec_inspect_command(
 169             inspect_command="", kdu_model=kdu_model, repo_url=repo_url
 170         )
 171
 172     """
 173     ####################################################################################
 174     ################################### P R I V A T E ##################################
 175     ####################################################################################
 176     """
 177
 178     def _init_paths_env(self, cluster_name: str, create_if_not_exist: bool = True):
 179         """
 180         Creates and returns base cluster and kube dirs and returns them.
 181         Also created helm3 dirs according to new directory specification, paths are
 182         returned and also environment variables that must be provided to execute commands
 183
 184         Helm 2 directory specification uses helm_home dir:
 185
 186         The variables assigned for this paths are:
 187         - Helm hone: $HELM_HOME
 188         - helm kubeconfig: $KUBECONFIG
 189
 190         :param cluster_name:  cluster_name
 191         :return: Dictionary with config_paths and dictionary with helm environment variables
 192         """
 193         base = self.fs.path
 194         if base.endswith("/") or base.endswith("\\"):
 195             base = base[:-1]
 196
 197         # base dir for cluster
 198         cluster_dir = base + "/" + cluster_name
 199
 200         # kube dir
 201         kube_dir = cluster_dir + "/" + ".kube"
 202         if create_if_not_exist and not os.path.exists(kube_dir):
 203             self.log.debug("Creating dir {}".format(kube_dir))
 204             os.makedirs(kube_dir)
 205
 206         # helm home dir
 207         helm_dir = cluster_dir + "/" + ".helm"
 208         if create_if_not_exist and not os.path.exists(helm_dir):
 209             self.log.debug("Creating dir {}".format(helm_dir))
 210             os.makedirs(helm_dir)
 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             "helm_dir": helm_dir,
 220         }
 221
 222         for file_name, file in paths.items():
 223             if "dir" in file_name and not os.path.exists(file):
 224                 err_msg = "{} dir does not exist".format(file)
 225                 self.log.error(err_msg)
 226                 raise K8sException(err_msg)
 227
 228         # 3 - Prepare environment variables
 229         env = {"HELM_HOME": helm_dir, "KUBECONFIG": config_filename}
 230
 231         return paths, env
 232
 233     async def _get_services(self, cluster_id, kdu_instance, namespace, kubeconfig):
 234         # init config, env
 235         paths, env = self._init_paths_env(
 236             cluster_name=cluster_id, create_if_not_exist=True
 237         )
 238
 239         command1 = "env KUBECONFIG={} {} get manifest {} ".format(
 240             kubeconfig, self._helm_command, kdu_instance
 241         )
 242         command2 = "{} get --namespace={} -f -".format(self.kubectl_command, namespace)
 243         output, _rc = await self._local_async_exec_pipe(
 244             command1, command2, env=env, raise_exception_on_error=True
 245         )
 246         services = self._parse_services(output)
 247
 248         return services
 249
 250     async def _cluster_init(
 251         self, cluster_id: str, namespace: str, paths: dict, env: dict
 252     ):
 253         """
 254         Implements the helm version dependent cluster initialization:
 255         For helm2 it initialized tiller environment if needed
 256         """
 257
 258         # check if tiller pod is up in cluster
 259         command = "{} --kubeconfig={} --namespace={} get deployments".format(
 260             self.kubectl_command, paths["kube_config"], namespace
 261         )
 262         output, _rc = await self._local_async_exec(
 263             command=command, raise_exception_on_error=True, env=env
 264         )
 265
 266         output_table = self._output_to_table(output=output)
 267
 268         # find 'tiller' pod in all pods
 269         already_initialized = False
 270         try:
 271             for row in output_table:
 272                 if row[0].startswith("tiller-deploy"):
 273                     already_initialized = True
 274                     break
 275         except Exception:
 276             pass
 277
 278         # helm init
 279         n2vc_installed_sw = False
 280         if not already_initialized:
 281             self.log.info(
 282                 "Initializing helm in client and server: {}".format(cluster_id)
 283             )
 284             command = "{} --kubeconfig={} --namespace kube-system create serviceaccount {}".format(
 285                 self.kubectl_command, paths["kube_config"], self.service_account
 286             )
 287             _, _rc = await self._local_async_exec(
 288                 command=command, raise_exception_on_error=False, env=env
 289             )
 290
 291             command = (
 292                 "{} --kubeconfig={} create clusterrolebinding osm-tiller-cluster-rule "
 293                 "--clusterrole=cluster-admin --serviceaccount=kube-system:{}"
 294             ).format(self.kubectl_command, paths["kube_config"], self.service_account)
 295             _, _rc = await self._local_async_exec(
 296                 command=command, raise_exception_on_error=False, env=env
 297             )
 298
 299             command = (
 300                 "{} --kubeconfig={} --tiller-namespace={} --home={} --service-account {} "
 301                 " {} init"
 302             ).format(
 303                 self._helm_command,
 304                 paths["kube_config"],
 305                 namespace,
 306                 paths["helm_dir"],
 307                 self.service_account,
 308                 "--stable-repo-url {}".format(self._stable_repo_url)
 309                 if self._stable_repo_url
 310                 else "--skip-repos",
 311             )
 312             _, _rc = await self._local_async_exec(
 313                 command=command, raise_exception_on_error=True, env=env
 314             )
 315             n2vc_installed_sw = True
 316         else:
 317             # check client helm installation
 318             check_file = paths["helm_dir"] + "/repository/repositories.yaml"
 319             if not self._check_file_exists(
 320                 filename=check_file, exception_if_not_exists=False
 321             ):
 322                 self.log.info("Initializing helm in client: {}".format(cluster_id))
 323                 command = (
 324                     "{} --kubeconfig={} --tiller-namespace={} "
 325                     "--home={} init --client-only {} "
 326                 ).format(
 327                     self._helm_command,
 328                     paths["kube_config"],
 329                     namespace,
 330                     paths["helm_dir"],
 331                     "--stable-repo-url {}".format(self._stable_repo_url)
 332                     if self._stable_repo_url
 333                     else "--skip-repos",
 334                 )
 335                 output, _rc = await self._local_async_exec(
 336                     command=command, raise_exception_on_error=True, env=env
 337                 )
 338             else:
 339                 self.log.info("Helm client already initialized")
 340
 341         repo_list = await self.repo_list(cluster_id)
 342         for repo in repo_list:
 343             if repo["name"] == "stable" and repo["url"] != self._stable_repo_url:
 344                 self.log.debug("Add new stable repo url: {}")
 345                 await self.repo_remove(cluster_id, "stable")
 346                 if self._stable_repo_url:
 347                     await self.repo_add(cluster_id, "stable", self._stable_repo_url)
 348                 break
 349
 350         return n2vc_installed_sw
 351
 352     async def _uninstall_sw(self, cluster_id: str, namespace: str):
 353         # uninstall Tiller if necessary
 354
 355         self.log.debug("Uninstalling tiller from cluster {}".format(cluster_id))
 356
 357         # init paths, env
 358         paths, env = self._init_paths_env(
 359             cluster_name=cluster_id, create_if_not_exist=True
 360         )
 361
 362         if not namespace:
 363             # find namespace for tiller pod
 364             command = "{} --kubeconfig={} get deployments --all-namespaces".format(
 365                 self.kubectl_command, paths["kube_config"]
 366             )
 367             output, _rc = await self._local_async_exec(
 368                 command=command, raise_exception_on_error=False, env=env
 369             )
 370             output_table = self._output_to_table(output=output)
 371             namespace = None
 372             for r in output_table:
 373                 try:
 374                     if "tiller-deploy" in r[1]:
 375                         namespace = r[0]
 376                         break
 377                 except Exception:
 378                     pass
 379             else:
 380                 msg = "Tiller deployment not found in cluster {}".format(cluster_id)
 381                 self.log.error(msg)
 382
 383             self.log.debug("namespace for tiller: {}".format(namespace))
 384
 385         if namespace:
 386             # uninstall tiller from cluster
 387             self.log.debug("Uninstalling tiller from cluster {}".format(cluster_id))
 388             command = "{} --kubeconfig={} --home={} reset".format(
 389                 self._helm_command, paths["kube_config"], paths["helm_dir"]
 390             )
 391             self.log.debug("resetting: {}".format(command))
 392             output, _rc = await self._local_async_exec(
 393                 command=command, raise_exception_on_error=True, env=env
 394             )
 395             # Delete clusterrolebinding and serviceaccount.
 396             # Ignore if errors for backward compatibility
 397             command = (
 398                 "{} --kubeconfig={} delete clusterrolebinding.rbac.authorization.k8s."
 399                 "io/osm-tiller-cluster-rule"
 400             ).format(self.kubectl_command, paths["kube_config"])
 401             output, _rc = await self._local_async_exec(
 402                 command=command, raise_exception_on_error=False, env=env
 403             )
 404             command = (
 405                 "{} --kubeconfig={} --namespace {} delete serviceaccount/{}".format(
 406                     self.kubectl_command,
 407                     paths["kube_config"],
 408                     namespace,
 409                     self.service_account,
 410                 )
 411             )
 412             output, _rc = await self._local_async_exec(
 413                 command=command, raise_exception_on_error=False, env=env
 414             )
 415
 416         else:
 417             self.log.debug("namespace not found")
 418
 419     async def _instances_list(self, cluster_id):
 420         # init paths, env
 421         paths, env = self._init_paths_env(
 422             cluster_name=cluster_id, create_if_not_exist=True
 423         )
 424
 425         command = "{} list --output yaml".format(self._helm_command)
 426
 427         output, _rc = await self._local_async_exec(
 428             command=command, raise_exception_on_error=True, env=env
 429         )
 430
 431         if output and len(output) > 0:
 432             # parse yaml and update keys to lower case to unify with helm3
 433             instances = yaml.load(output, Loader=yaml.SafeLoader).get("Releases")
 434             new_instances = []
 435             for instance in instances:
 436                 new_instance = dict((k.lower(), v) for k, v in instance.items())
 437                 new_instances.append(new_instance)
 438             return new_instances
 439         else:
 440             return []
 441
 442     def _get_inspect_command(
 443         self, show_command: str, kdu_model: str, repo_str: str, version: str
 444     ):
 445         inspect_command = "{} inspect {} {}{} {}".format(
 446             self._helm_command, show_command, kdu_model, repo_str, version
 447         )
 448         return inspect_command
 449
 450     def _get_get_command(
 451         self, get_command: str, kdu_instance: str, namespace: str, kubeconfig: str
 452     ):
 453         get_command = "env KUBECONFIG={} {} get {} {} --output yaml".format(
 454             kubeconfig, self._helm_command, get_command, kdu_instance
 455         )
 456         return get_command
 457
 458     async def _status_kdu(
 459         self,
 460         cluster_id: str,
 461         kdu_instance: str,
 462         namespace: str = None,
 463         yaml_format: bool = False,
 464         show_error_log: bool = False,
 465     ) -> Union[str, dict]:
 466         self.log.debug(
 467             "status of kdu_instance: {}, namespace: {} ".format(kdu_instance, namespace)
 468         )
 469
 470         # init config, env
 471         paths, env = self._init_paths_env(
 472             cluster_name=cluster_id, create_if_not_exist=True
 473         )
 474         command = ("env KUBECONFIG={} {} status {} --output yaml").format(
 475             paths["kube_config"], self._helm_command, kdu_instance
 476         )
 477         output, rc = await self._local_async_exec(
 478             command=command,
 479             raise_exception_on_error=True,
 480             show_error_log=show_error_log,
 481             env=env,
 482         )
 483
 484         if yaml_format:
 485             return str(output)
 486
 487         if rc != 0:
 488             return None
 489
 490         data = yaml.load(output, Loader=yaml.SafeLoader)
 491
 492         # remove field 'notes'
 493         try:
 494             del data.get("info").get("status")["notes"]
 495         except KeyError:
 496             pass
 497
 498         # parse the manifest to a list of dictionaries
 499         if "manifest" in data:
 500             manifest_str = data.get("manifest")
 501             manifest_docs = yaml.load_all(manifest_str, Loader=yaml.SafeLoader)
 502
 503             data["manifest"] = []
 504             for doc in manifest_docs:
 505                 data["manifest"].append(doc)
 506
 507         # parse field 'resources'
 508         try:
 509             resources = str(data.get("info").get("status").get("resources"))
 510             resource_table = self._output_to_table(resources)
 511             data.get("info").get("status")["resources"] = resource_table
 512         except Exception:
 513             pass
 514
 515         # set description to lowercase (unify with helm3)
 516         try:
 517             data.get("info")["description"] = data.get("info").pop("Description")
 518         except KeyError:
 519             pass
 520
 521         return data
 522
 523     def _get_helm_chart_repos_ids(self, cluster_uuid) -> list:
 524         repo_ids = []
 525         cluster_filter = {"_admin.helm-chart.id": cluster_uuid}
 526         cluster = self.db.get_one("k8sclusters", cluster_filter)
 527         if cluster:
 528             repo_ids = cluster.get("_admin").get("helm_chart_repos") or []
 529             return repo_ids
 530         else:
 531             raise K8sException(
 532                 "k8cluster with helm-id : {} not found".format(cluster_uuid)
 533             )
 534
 535     async def _is_install_completed(self, cluster_id: str, kdu_instance: str) -> bool:
 536         # init config, env
 537         paths, env = self._init_paths_env(
 538             cluster_name=cluster_id, create_if_not_exist=True
 539         )
 540
 541         status = await self._status_kdu(
 542             cluster_id=cluster_id, kdu_instance=kdu_instance, yaml_format=False
 543         )
 544
 545         # extract info.status.resources-> str
 546         # format:
 547         #       ==> v1/Deployment
 548         #       NAME                    READY   UP-TO-DATE   AVAILABLE   AGE
 549         #       halting-horse-mongodb   0/1     1            0           0s
 550         #       halting-petit-mongodb   1/1     1            0           0s
 551         # blank line
 552         resources = K8sHelmBaseConnector._get_deep(
 553             status, ("info", "status", "resources")
 554         )
 555
 556         # convert to table
 557         resources = K8sHelmBaseConnector._output_to_table(resources)
 558
 559         num_lines = len(resources)
 560         index = 0
 561         ready = True
 562         while index < num_lines:
 563             try:
 564                 line1 = resources[index]
 565                 index += 1
 566                 # find '==>' in column 0
 567                 if line1[0] == "==>":
 568                     line2 = resources[index]
 569                     index += 1
 570                     # find READY in column 1
 571                     if line2[1] == "READY":
 572                         # read next lines
 573                         line3 = resources[index]
 574                         index += 1
 575                         while len(line3) > 1 and index < num_lines:
 576                             ready_value = line3[1]
 577                             parts = ready_value.split(sep="/")
 578                             current = int(parts[0])
 579                             total = int(parts[1])
 580                             if current < total:
 581                                 self.log.debug("NOT READY:\n    {}".format(line3))
 582                                 ready = False
 583                             line3 = resources[index]
 584                             index += 1
 585
 586             except Exception:
 587                 pass
 588
 589         return ready
 590
 591     def _get_install_command(
 592         self,
 593         kdu_model,
 594         kdu_instance,
 595         namespace,
 596         params_str,
 597         version,
 598         atomic,
 599         timeout,
 600         kubeconfig,
 601     ) -> str:
 602         timeout_str = ""
 603         if timeout:
 604             timeout_str = "--timeout {}".format(timeout)
 605
 606         # atomic
 607         atomic_str = ""
 608         if atomic:
 609             atomic_str = "--atomic"
 610         # namespace
 611         namespace_str = ""
 612         if namespace:
 613             namespace_str = "--namespace {}".format(namespace)
 614
 615         # version
 616         version_str = ""
 617         if version:
 618             version_str = "--version {}".format(version)
 619
 620         command = (
 621             "env KUBECONFIG={kubeconfig} {helm} install {atomic} --output yaml  "
 622             "{params} {timeout} --name={name} {ns} {model} {ver}".format(
 623                 kubeconfig=kubeconfig,
 624                 helm=self._helm_command,
 625                 atomic=atomic_str,
 626                 params=params_str,
 627                 timeout=timeout_str,
 628                 name=kdu_instance,
 629                 ns=namespace_str,
 630                 model=kdu_model,
 631                 ver=version_str,
 632             )
 633         )
 634         return command
 635
 636     def _get_upgrade_scale_command(
 637         self,
 638         kdu_model: str,
 639         kdu_instance: str,
 640         namespace: str,
 641         scale: int,
 642         version: str,
 643         atomic: bool,
 644         replica_str: str,
 645         timeout: float,
 646         resource_name: str,
 647         kubeconfig: str,
 648     ) -> str:
 649         """Generates the command to scale a Helm Chart release
 650
 651         Args:
 652             kdu_model (str): Kdu model name, corresponding to the Helm local location or repository
 653             kdu_instance (str): KDU instance, corresponding to the Helm Chart release in question
 654             namespace (str): Namespace where this KDU instance is deployed
 655             scale (int): Scale count
 656             version (str): Constraint with specific version of the Chart to use
 657             atomic (bool): If set, upgrade process rolls back changes made in case of failed upgrade.
 658                 The --wait flag will be set automatically if --atomic is used
 659             replica_str (str): The key under resource_name key where the scale count is stored
 660             timeout (float): The time, in seconds, to wait
 661             resource_name (str): The KDU's resource to scale
 662             kubeconfig (str): Kubeconfig file path
 663
 664         Returns:
 665             str: command to scale a Helm Chart release
 666         """
 667
 668         # scale
 669         if resource_name:
 670             scale_dict = {"{}.{}".format(resource_name, replica_str): scale}
 671         else:
 672             scale_dict = {replica_str: scale}
 673
 674         scale_str = self._params_to_set_option(scale_dict)
 675
 676         return self._get_upgrade_command(
 677             kdu_model=kdu_model,
 678             kdu_instance=kdu_instance,
 679             namespace=namespace,
 680             params_str=scale_str,
 681             version=version,
 682             atomic=atomic,
 683             timeout=timeout,
 684             kubeconfig=kubeconfig,
 685         )
 686
 687     def _get_upgrade_command(
 688         self,
 689         kdu_model,
 690         kdu_instance,
 691         namespace,
 692         params_str,
 693         version,
 694         atomic,
 695         timeout,
 696         kubeconfig,
 697         force: bool = False,
 698     ) -> str:
 699         """Generates the command to upgrade a Helm Chart release
 700
 701         Args:
 702             kdu_model (str): Kdu model name, corresponding to the Helm local location or repository
 703             kdu_instance (str): KDU instance, corresponding to the Helm Chart release in question
 704             namespace (str): Namespace where this KDU instance is deployed
 705             params_str (str): Params used to upgrade the Helm Chart release
 706             version (str): Constraint with specific version of the Chart to use
 707             atomic (bool): If set, upgrade process rolls back changes made in case of failed upgrade.
 708                 The --wait flag will be set automatically if --atomic is used
 709             timeout (float): The time, in seconds, to wait
 710             kubeconfig (str): Kubeconfig file path
 711             force (bool): If set, helm forces resource updates through a replacement strategy. This may recreate pods.
 712         Returns:
 713             str: command to upgrade a Helm Chart release
 714         """
 715
 716         timeout_str = ""
 717         if timeout:
 718             timeout_str = "--timeout {}".format(timeout)
 719
 720         # atomic
 721         atomic_str = ""
 722         if atomic:
 723             atomic_str = "--atomic"
 724
 725         # force
 726         force_str = ""
 727         if force:
 728             force_str = "--force "
 729
 730         # version
 731         version_str = ""
 732         if version:
 733             version_str = "--version {}".format(version)
 734
 735         # namespace
 736         namespace_str = ""
 737         if namespace:
 738             namespace_str = "--namespace {}".format(namespace)
 739
 740         command = (
 741             "env KUBECONFIG={kubeconfig} {helm} upgrade {namespace} {atomic} --output yaml {params} {timeout} {force}"
 742             "--reuse-values {name} {model} {ver}"
 743         ).format(
 744             kubeconfig=kubeconfig,
 745             helm=self._helm_command,
 746             namespace=namespace_str,
 747             atomic=atomic_str,
 748             force=force_str,
 749             params=params_str,
 750             timeout=timeout_str,
 751             name=kdu_instance,
 752             model=kdu_model,
 753             ver=version_str,
 754         )
 755         return command
 756
 757     def _get_rollback_command(
 758         self, kdu_instance, namespace, revision, kubeconfig
 759     ) -> str:
 760         return "env KUBECONFIG={} {} rollback {} {} --wait".format(
 761             kubeconfig, self._helm_command, kdu_instance, revision
 762         )
 763
 764     def _get_uninstall_command(
 765         self, kdu_instance: str, namespace: str, kubeconfig: str
 766     ) -> str:
 767         return "env KUBECONFIG={} {} delete --purge  {}".format(
 768             kubeconfig, self._helm_command, kdu_instance
 769         )