n2vc/k8s_juju_conn.py

   1 # Copyright 2019 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 import asyncio
  16 import concurrent
  17 import os
  18 import uuid
  19 import yaml
  20
  21 import juju
  22 from juju.controller import Controller
  23 from juju.model import Model
  24 from n2vc.exceptions import K8sException
  25 from n2vc.k8s_conn import K8sConnector
  26
  27 from .exceptions import MethodNotImplemented
  28
  29
  30 # from juju.bundle import BundleHandler
  31 # import re
  32 # import ssl
  33 # from .vnf import N2VC
  34 class K8sJujuConnector(K8sConnector):
  35     def __init__(
  36         self,
  37         fs: object,
  38         db: object,
  39         kubectl_command: str = "/usr/bin/kubectl",
  40         juju_command: str = "/usr/bin/juju",
  41         log: object = None,
  42         on_update_db=None,
  43     ):
  44         """
  45
  46         :param kubectl_command: path to kubectl executable
  47         :param helm_command: path to helm executable
  48         :param fs: file system for kubernetes and helm configuration
  49         :param log: logger
  50         """
  51
  52         # parent class
  53         K8sConnector.__init__(
  54             self, db, log=log, on_update_db=on_update_db,
  55         )
  56
  57         self.fs = fs
  58         self.log.debug("Initializing K8S Juju connector")
  59
  60         self.authenticated = False
  61         self.models = {}
  62
  63         self.juju_command = juju_command
  64         self.juju_secret = ""
  65
  66         self.log.debug("K8S Juju connector initialized")
  67
  68     """Initialization"""
  69
  70     async def init_env(
  71         self,
  72         k8s_creds: str,
  73         namespace: str = "kube-system",
  74         reuse_cluster_uuid: str = None,
  75     ) -> (str, bool):
  76         """
  77         It prepares a given K8s cluster environment to run Juju bundles.
  78
  79         :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid
  80             '.kube/config'
  81         :param namespace: optional namespace to be used for juju. By default,
  82             'kube-system' will be used
  83         :param reuse_cluster_uuid: existing cluster uuid for reuse
  84         :return: uuid of the K8s cluster and True if connector has installed some
  85             software in the cluster
  86             (on error, an exception will be raised)
  87         """
  88
  89         """Bootstrapping
  90
  91         Bootstrapping cannot be done, by design, through the API. We need to
  92         use the CLI tools.
  93         """
  94
  95         """
  96         WIP: Workflow
  97
  98         1. Has the environment already been bootstrapped?
  99         - Check the database to see if we have a record for this env
 100
 101         2. If this is a new env, create it
 102         - Add the k8s cloud to Juju
 103         - Bootstrap
 104         - Record it in the database
 105
 106         3. Connect to the Juju controller for this cloud
 107
 108         """
 109         # cluster_uuid = reuse_cluster_uuid
 110         # if not cluster_uuid:
 111         #     cluster_uuid = str(uuid4())
 112
 113         ##################################################
 114         # TODO: Pull info from db based on the namespace #
 115         ##################################################
 116
 117         ###################################################
 118         # TODO: Make it idempotent, calling add-k8s and   #
 119         # bootstrap whenever reuse_cluster_uuid is passed #
 120         # as parameter                                    #
 121         # `init_env` is called to initialize the K8s      #
 122         # cluster for juju. If this initialization fails, #
 123         # it can be called again by LCM with the param    #
 124         # reuse_cluster_uuid, e.g. to try to fix it.       #
 125         ###################################################
 126
 127         if not reuse_cluster_uuid:
 128             # This is a new cluster, so bootstrap it
 129
 130             cluster_uuid = str(uuid.uuid4())
 131
 132             # Is a local k8s cluster?
 133             localk8s = self.is_local_k8s(k8s_creds)
 134
 135             # If the k8s is external, the juju controller needs a loadbalancer
 136             loadbalancer = False if localk8s else True
 137
 138             # Name the new k8s cloud
 139             k8s_cloud = "k8s-{}".format(cluster_uuid)
 140
 141             self.log.debug("Adding k8s cloud {}".format(k8s_cloud))
 142             await self.add_k8s(k8s_cloud, k8s_creds)
 143
 144             # Bootstrap Juju controller
 145             self.log.debug("Bootstrapping...")
 146             await self.bootstrap(k8s_cloud, cluster_uuid, loadbalancer)
 147             self.log.debug("Bootstrap done.")
 148
 149             # Get the controller information
 150
 151             # Parse ~/.local/share/juju/controllers.yaml
 152             # controllers.testing.api-endpoints|ca-cert|uuid
 153             self.log.debug("Getting controller endpoints")
 154             with open(os.path.expanduser("~/.local/share/juju/controllers.yaml")) as f:
 155                 controllers = yaml.load(f, Loader=yaml.Loader)
 156                 controller = controllers["controllers"][cluster_uuid]
 157                 endpoints = controller["api-endpoints"]
 158                 self.juju_endpoint = endpoints[0]
 159                 self.juju_ca_cert = controller["ca-cert"]
 160
 161             # Parse ~/.local/share/juju/accounts
 162             # controllers.testing.user|password
 163             self.log.debug("Getting accounts")
 164             with open(os.path.expanduser("~/.local/share/juju/accounts.yaml")) as f:
 165                 controllers = yaml.load(f, Loader=yaml.Loader)
 166                 controller = controllers["controllers"][cluster_uuid]
 167
 168                 self.juju_user = controller["user"]
 169                 self.juju_secret = controller["password"]
 170
 171             # raise Exception("EOL")
 172
 173             self.juju_public_key = None
 174
 175             config = {
 176                 "endpoint": self.juju_endpoint,
 177                 "username": self.juju_user,
 178                 "secret": self.juju_secret,
 179                 "cacert": self.juju_ca_cert,
 180                 "namespace": namespace,
 181                 "loadbalancer": loadbalancer,
 182             }
 183
 184             # Store the cluster configuration so it
 185             # can be used for subsequent calls
 186             self.log.debug("Setting config")
 187             await self.set_config(cluster_uuid, config)
 188
 189         else:
 190             # This is an existing cluster, so get its config
 191             cluster_uuid = reuse_cluster_uuid
 192
 193             config = self.get_config(cluster_uuid)
 194
 195             self.juju_endpoint = config["endpoint"]
 196             self.juju_user = config["username"]
 197             self.juju_secret = config["secret"]
 198             self.juju_ca_cert = config["cacert"]
 199             self.juju_public_key = None
 200
 201         # Login to the k8s cluster
 202         if not self.authenticated:
 203             await self.login(cluster_uuid)
 204
 205         # We're creating a new cluster
 206         # print("Getting model {}".format(self.get_namespace(cluster_uuid),
 207         #    cluster_uuid=cluster_uuid))
 208         # model = await self.get_model(
 209         #    self.get_namespace(cluster_uuid),
 210         #    cluster_uuid=cluster_uuid
 211         # )
 212
 213         # Disconnect from the model
 214         # if model and model.is_connected():
 215         #    await model.disconnect()
 216
 217         return cluster_uuid, True
 218
 219     """Repo Management"""
 220
 221     async def repo_add(
 222         self, name: str, url: str, _type: str = "charm",
 223     ):
 224         raise MethodNotImplemented()
 225
 226     async def repo_list(self):
 227         raise MethodNotImplemented()
 228
 229     async def repo_remove(
 230         self, name: str,
 231     ):
 232         raise MethodNotImplemented()
 233
 234     async def synchronize_repos(self, cluster_uuid: str, name: str):
 235         """
 236         Returns None as currently add_repo is not implemented
 237         """
 238         return None
 239
 240     """Reset"""
 241
 242     async def reset(
 243         self, cluster_uuid: str, force: bool = False, uninstall_sw: bool = False
 244     ) -> bool:
 245         """Reset a cluster
 246
 247         Resets the Kubernetes cluster by removing the model that represents it.
 248
 249         :param cluster_uuid str: The UUID of the cluster to reset
 250         :return: Returns True if successful or raises an exception.
 251         """
 252
 253         try:
 254             if not self.authenticated:
 255                 await self.login(cluster_uuid)
 256
 257             if self.controller.is_connected():
 258                 # Destroy the model
 259                 namespace = self.get_namespace(cluster_uuid)
 260                 if await self.has_model(namespace):
 261                     self.log.debug("[reset] Destroying model")
 262                     await self.controller.destroy_model(namespace, destroy_storage=True)
 263
 264                 # Disconnect from the controller
 265                 self.log.debug("[reset] Disconnecting controller")
 266                 await self.logout()
 267
 268                 # Destroy the controller (via CLI)
 269                 self.log.debug("[reset] Destroying controller")
 270                 await self.destroy_controller(cluster_uuid)
 271
 272                 self.log.debug("[reset] Removing k8s cloud")
 273                 k8s_cloud = "k8s-{}".format(cluster_uuid)
 274                 await self.remove_cloud(k8s_cloud)
 275
 276         except Exception as ex:
 277             self.log.debug("Caught exception during reset: {}".format(ex))
 278
 279         return True
 280
 281     """Deployment"""
 282
 283     async def install(
 284         self,
 285         cluster_uuid: str,
 286         kdu_model: str,
 287         atomic: bool = True,
 288         timeout: float = 300,
 289         params: dict = None,
 290         db_dict: dict = None,
 291         kdu_name: str = None,
 292         namespace: str = None,
 293     ) -> bool:
 294         """Install a bundle
 295
 296         :param cluster_uuid str: The UUID of the cluster to install to
 297         :param kdu_model str: The name or path of a bundle to install
 298         :param atomic bool: If set, waits until the model is active and resets
 299                             the cluster on failure.
 300         :param timeout int: The time, in seconds, to wait for the install
 301                             to finish
 302         :param params dict: Key-value pairs of instantiation parameters
 303         :param kdu_name: Name of the KDU instance to be installed
 304         :param namespace: K8s namespace to use for the KDU instance
 305
 306         :return: If successful, returns ?
 307         """
 308
 309         if not self.authenticated:
 310             self.log.debug("[install] Logging in to the controller")
 311             await self.login(cluster_uuid)
 312
 313         ##
 314         # Get or create the model, based on the NS
 315         # uuid.
 316         if kdu_name:
 317             kdu_instance = "{}-{}".format(kdu_name, db_dict["filter"]["_id"])
 318         else:
 319             kdu_instance = db_dict["filter"]["_id"]
 320
 321         self.log.debug("Checking for model named {}".format(kdu_instance))
 322
 323         # Create the new model
 324         self.log.debug("Adding model: {}".format(kdu_instance))
 325         model = await self.add_model(kdu_instance, cluster_uuid=cluster_uuid)
 326
 327         if model:
 328             # TODO: Instantiation parameters
 329
 330             """
 331             "Juju bundle that models the KDU, in any of the following ways:
 332                 - <juju-repo>/<juju-bundle>
 333                 - <juju-bundle folder under k8s_models folder in the package>
 334                 - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder
 335                     in the package>
 336                 - <URL_where_to_fetch_juju_bundle>
 337             """
 338
 339             previous_workdir = os.getcwd()
 340
 341             bundle = kdu_model
 342             if kdu_model.startswith("cs:"):
 343                 bundle = kdu_model
 344             elif kdu_model.startswith("http"):
 345                 # Download the file
 346                 pass
 347             else:
 348                 new_workdir = kdu_model.strip(kdu_model.split("/")[-1])
 349
 350                 os.chdir(new_workdir)
 351
 352                 bundle = "local:{}".format(kdu_model)
 353
 354             if not bundle:
 355                 # Raise named exception that the bundle could not be found
 356                 raise Exception()
 357
 358             self.log.debug("[install] deploying {}".format(bundle))
 359             await model.deploy(bundle)
 360
 361             # Get the application
 362             if atomic:
 363                 # applications = model.applications
 364                 self.log.debug("[install] Applications: {}".format(model.applications))
 365                 for name in model.applications:
 366                     self.log.debug("[install] Waiting for {} to settle".format(name))
 367                     application = model.applications[name]
 368                     try:
 369                         # It's not enough to wait for all units to be active;
 370                         # the application status needs to be active as well.
 371                         self.log.debug("Waiting for all units to be active...")
 372                         await model.block_until(
 373                             lambda: all(
 374                                 unit.agent_status == "idle"
 375                                 and application.status in ["active", "unknown"]
 376                                 and unit.workload_status in ["active", "unknown"]
 377                                 for unit in application.units
 378                             ),
 379                             timeout=timeout,
 380                         )
 381                         self.log.debug("All units active.")
 382
 383                     # TODO use asyncio.TimeoutError
 384                     except concurrent.futures._base.TimeoutError:
 385                         os.chdir(previous_workdir)
 386                         self.log.debug("[install] Timeout exceeded; resetting cluster")
 387                         await self.reset(cluster_uuid)
 388                         return False
 389
 390             # Wait for the application to be active
 391             if model.is_connected():
 392                 self.log.debug("[install] Disconnecting model")
 393                 await model.disconnect()
 394
 395             os.chdir(previous_workdir)
 396
 397             return kdu_instance
 398         raise Exception("Unable to install")
 399
 400     async def instances_list(self, cluster_uuid: str) -> list:
 401         """
 402         returns a list of deployed releases in a cluster
 403
 404         :param cluster_uuid: the cluster
 405         :return:
 406         """
 407         return []
 408
 409     async def upgrade(
 410         self,
 411         cluster_uuid: str,
 412         kdu_instance: str,
 413         kdu_model: str = None,
 414         params: dict = None,
 415     ) -> str:
 416         """Upgrade a model
 417
 418         :param cluster_uuid str: The UUID of the cluster to upgrade
 419         :param kdu_instance str: The unique name of the KDU instance
 420         :param kdu_model str: The name or path of the bundle to upgrade to
 421         :param params dict: Key-value pairs of instantiation parameters
 422
 423         :return: If successful, reference to the new revision number of the
 424                  KDU instance.
 425         """
 426
 427         # TODO: Loop through the bundle and upgrade each charm individually
 428
 429         """
 430         The API doesn't have a concept of bundle upgrades, because there are
 431         many possible changes: charm revision, disk, number of units, etc.
 432
 433         As such, we are only supporting a limited subset of upgrades. We'll
 434         upgrade the charm revision but leave storage and scale untouched.
 435
 436         Scale changes should happen through OSM constructs, and changes to
 437         storage would require a redeployment of the service, at least in this
 438         initial release.
 439         """
 440         namespace = self.get_namespace(cluster_uuid)
 441         model = await self.get_model(namespace, cluster_uuid=cluster_uuid)
 442
 443         with open(kdu_model, "r") as f:
 444             bundle = yaml.safe_load(f)
 445
 446             """
 447             {
 448                 'description': 'Test bundle',
 449                 'bundle': 'kubernetes',
 450                 'applications': {
 451                     'mariadb-k8s': {
 452                         'charm': 'cs:~charmed-osm/mariadb-k8s-20',
 453                         'scale': 1,
 454                         'options': {
 455                             'password': 'manopw',
 456                             'root_password': 'osm4u',
 457                             'user': 'mano'
 458                         },
 459                         'series': 'kubernetes'
 460                     }
 461                 }
 462             }
 463             """
 464             # TODO: This should be returned in an agreed-upon format
 465             for name in bundle["applications"]:
 466                 self.log.debug(model.applications)
 467                 application = model.applications[name]
 468                 self.log.debug(application)
 469
 470                 path = bundle["applications"][name]["charm"]
 471
 472                 try:
 473                     await application.upgrade_charm(switch=path)
 474                 except juju.errors.JujuError as ex:
 475                     if "already running charm" in str(ex):
 476                         # We're already running this version
 477                         pass
 478
 479         await model.disconnect()
 480
 481         return True
 482         raise MethodNotImplemented()
 483
 484     """Rollback"""
 485
 486     async def rollback(
 487         self, cluster_uuid: str, kdu_instance: str, revision: int = 0,
 488     ) -> str:
 489         """Rollback a model
 490
 491         :param cluster_uuid str: The UUID of the cluster to rollback
 492         :param kdu_instance str: The unique name of the KDU instance
 493         :param revision int: The revision to revert to. If omitted, rolls back
 494                              the previous upgrade.
 495
 496         :return: If successful, returns the revision of active KDU instance,
 497                  or raises an exception
 498         """
 499         raise MethodNotImplemented()
 500
 501     """Deletion"""
 502
 503     async def uninstall(self, cluster_uuid: str, kdu_instance: str) -> bool:
 504         """Uninstall a KDU instance
 505
 506         :param cluster_uuid str: The UUID of the cluster
 507         :param kdu_instance str: The unique name of the KDU instance
 508
 509         :return: Returns True if successful, or raises an exception
 510         """
 511         if not self.authenticated:
 512             self.log.debug("[uninstall] Connecting to controller")
 513             await self.login(cluster_uuid)
 514
 515         self.log.debug("[uninstall] Destroying model")
 516
 517         await self.controller.destroy_models(kdu_instance)
 518
 519         self.log.debug("[uninstall] Model destroyed and disconnecting")
 520         await self.logout()
 521
 522         return True
 523
 524     async def exec_primitive(
 525         self,
 526         cluster_uuid: str = None,
 527         kdu_instance: str = None,
 528         primitive_name: str = None,
 529         timeout: float = 300,
 530         params: dict = None,
 531         db_dict: dict = None,
 532     ) -> str:
 533         """Exec primitive (Juju action)
 534
 535         :param cluster_uuid str: The UUID of the cluster
 536         :param kdu_instance str: The unique name of the KDU instance
 537         :param primitive_name: Name of action that will be executed
 538         :param timeout: Timeout for action execution
 539         :param params: Dictionary of all the parameters needed for the action
 540         :db_dict: Dictionary for any additional data
 541
 542         :return: Returns the output of the action
 543         """
 544         if not self.authenticated:
 545             self.log.debug("[exec_primitive] Connecting to controller")
 546             await self.login(cluster_uuid)
 547
 548         if not params or "application-name" not in params:
 549             raise K8sException(
 550                 "Missing application-name argument, \
 551                                 argument needed for K8s actions"
 552             )
 553         try:
 554             self.log.debug(
 555                 "[exec_primitive] Getting model "
 556                 "kdu_instance: {}".format(kdu_instance)
 557             )
 558
 559             model = await self.get_model(kdu_instance, cluster_uuid)
 560
 561             application_name = params["application-name"]
 562             application = model.applications[application_name]
 563
 564             actions = await application.get_actions()
 565             if primitive_name not in actions:
 566                 raise K8sException("Primitive {} not found".format(primitive_name))
 567
 568             unit = None
 569             for u in application.units:
 570                 if await u.is_leader_from_status():
 571                     unit = u
 572                     break
 573
 574             if unit is None:
 575                 raise K8sException("No leader unit found to execute action")
 576
 577             self.log.debug("[exec_primitive] Running action: {}".format(primitive_name))
 578             action = await unit.run_action(primitive_name, **params)
 579
 580             output = await model.get_action_output(action_uuid=action.entity_id)
 581             status = await model.get_action_status(uuid_or_prefix=action.entity_id)
 582
 583             status = (
 584                 status[action.entity_id] if action.entity_id in status else "failed"
 585             )
 586
 587             if status != "completed":
 588                 raise K8sException(
 589                     "status is not completed: {} output: {}".format(status, output)
 590                 )
 591
 592             return output
 593
 594         except Exception as e:
 595             error_msg = "Error executing primitive {}: {}".format(primitive_name, e)
 596             self.log.error(error_msg)
 597             raise K8sException(message=error_msg)
 598
 599     """Introspection"""
 600
 601     async def inspect_kdu(self, kdu_model: str,) -> dict:
 602         """Inspect a KDU
 603
 604         Inspects a bundle and returns a dictionary of config parameters and
 605         their default values.
 606
 607         :param kdu_model str: The name or path of the bundle to inspect.
 608
 609         :return: If successful, returns a dictionary of available parameters
 610                  and their default values.
 611         """
 612
 613         kdu = {}
 614         with open(kdu_model, "r") as f:
 615             bundle = yaml.safe_load(f)
 616
 617             """
 618             {
 619                 'description': 'Test bundle',
 620                 'bundle': 'kubernetes',
 621                 'applications': {
 622                     'mariadb-k8s': {
 623                         'charm': 'cs:~charmed-osm/mariadb-k8s-20',
 624                         'scale': 1,
 625                         'options': {
 626                             'password': 'manopw',
 627                             'root_password': 'osm4u',
 628                             'user': 'mano'
 629                         },
 630                         'series': 'kubernetes'
 631                     }
 632                 }
 633             }
 634             """
 635             # TODO: This should be returned in an agreed-upon format
 636             kdu = bundle["applications"]
 637
 638         return kdu
 639
 640     async def help_kdu(self, kdu_model: str,) -> str:
 641         """View the README
 642
 643         If available, returns the README of the bundle.
 644
 645         :param kdu_model str: The name or path of a bundle
 646
 647         :return: If found, returns the contents of the README.
 648         """
 649         readme = None
 650
 651         files = ["README", "README.txt", "README.md"]
 652         path = os.path.dirname(kdu_model)
 653         for file in os.listdir(path):
 654             if file in files:
 655                 with open(file, "r") as f:
 656                     readme = f.read()
 657                     break
 658
 659         return readme
 660
 661     async def status_kdu(self, cluster_uuid: str, kdu_instance: str,) -> dict:
 662         """Get the status of the KDU
 663
 664         Get the current status of the KDU instance.
 665
 666         :param cluster_uuid str: The UUID of the cluster
 667         :param kdu_instance str: The unique id of the KDU instance
 668
 669         :return: Returns a dictionary containing namespace, state, resources,
 670                  and deployment_time.
 671         """
 672         status = {}
 673
 674         model = await self.get_model(
 675             self.get_namespace(cluster_uuid), cluster_uuid=cluster_uuid
 676         )
 677
 678         # model = await self.get_model_by_uuid(cluster_uuid)
 679         if model:
 680             model_status = await model.get_status()
 681             status = model_status.applications
 682
 683             for name in model_status.applications:
 684                 application = model_status.applications[name]
 685                 status[name] = {"status": application["status"]["status"]}
 686
 687             if model.is_connected():
 688                 await model.disconnect()
 689
 690         return status
 691
 692     # Private methods
 693     async def add_k8s(self, cloud_name: str, credentials: str,) -> bool:
 694         """Add a k8s cloud to Juju
 695
 696         Adds a Kubernetes cloud to Juju, so it can be bootstrapped with a
 697         Juju Controller.
 698
 699         :param cloud_name str: The name of the cloud to add.
 700         :param credentials dict: A dictionary representing the output of
 701             `kubectl config view --raw`.
 702
 703         :returns: True if successful, otherwise raises an exception.
 704         """
 705
 706         cmd = [self.juju_command, "add-k8s", "--local", cloud_name]
 707         self.log.debug(cmd)
 708
 709         process = await asyncio.create_subprocess_exec(
 710             *cmd,
 711             stdout=asyncio.subprocess.PIPE,
 712             stderr=asyncio.subprocess.PIPE,
 713             stdin=asyncio.subprocess.PIPE,
 714         )
 715
 716         # Feed the process the credentials
 717         process.stdin.write(credentials.encode("utf-8"))
 718         await process.stdin.drain()
 719         process.stdin.close()
 720
 721         _stdout, stderr = await process.communicate()
 722
 723         return_code = process.returncode
 724
 725         self.log.debug("add-k8s return code: {}".format(return_code))
 726
 727         if return_code > 0:
 728             raise Exception(stderr)
 729
 730         return True
 731
 732     async def add_model(self, model_name: str, cluster_uuid: str,) -> Model:
 733         """Adds a model to the controller
 734
 735         Adds a new model to the Juju controller
 736
 737         :param model_name str: The name of the model to add.
 738         :returns: The juju.model.Model object of the new model upon success or
 739                   raises an exception.
 740         """
 741         if not self.authenticated:
 742             await self.login(cluster_uuid)
 743
 744         self.log.debug(
 745             "Adding model '{}' to cluster_uuid '{}'".format(model_name, cluster_uuid)
 746         )
 747         try:
 748             model = await self.controller.add_model(
 749                 model_name, config={"authorized-keys": self.juju_public_key}
 750             )
 751         except Exception as ex:
 752             self.log.debug(ex)
 753             self.log.debug("Caught exception: {}".format(ex))
 754             pass
 755
 756         return model
 757
 758     async def bootstrap(
 759         self, cloud_name: str, cluster_uuid: str, loadbalancer: bool
 760     ) -> bool:
 761         """Bootstrap a Kubernetes controller
 762
 763         Bootstrap a Juju controller inside the Kubernetes cluster
 764
 765         :param cloud_name str: The name of the cloud.
 766         :param cluster_uuid str: The UUID of the cluster to bootstrap.
 767         :param loadbalancer bool: If the controller should use loadbalancer or not.
 768         :returns: True upon success or raises an exception.
 769         """
 770
 771         if not loadbalancer:
 772             cmd = [self.juju_command, "bootstrap", cloud_name, cluster_uuid]
 773         else:
 774             """
 775             For public clusters, specify that the controller service is using a
 776             LoadBalancer.
 777             """
 778             cmd = [
 779                 self.juju_command,
 780                 "bootstrap",
 781                 cloud_name,
 782                 cluster_uuid,
 783                 "--config",
 784                 "controller-service-type=loadbalancer",
 785             ]
 786
 787         self.log.debug(
 788             "Bootstrapping controller {} in cloud {}".format(cluster_uuid, cloud_name)
 789         )
 790
 791         process = await asyncio.create_subprocess_exec(
 792             *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
 793         )
 794
 795         _stdout, stderr = await process.communicate()
 796
 797         return_code = process.returncode
 798
 799         if return_code > 0:
 800             #
 801             if b"already exists" not in stderr:
 802                 raise Exception(stderr)
 803
 804         return True
 805
 806     async def destroy_controller(self, cluster_uuid: str) -> bool:
 807         """Destroy a Kubernetes controller
 808
 809         Destroy an existing Kubernetes controller.
 810
 811         :param cluster_uuid str: The UUID of the cluster to bootstrap.
 812         :returns: True upon success or raises an exception.
 813         """
 814         cmd = [
 815             self.juju_command,
 816             "destroy-controller",
 817             "--destroy-all-models",
 818             "--destroy-storage",
 819             "-y",
 820             cluster_uuid,
 821         ]
 822
 823         process = await asyncio.create_subprocess_exec(
 824             *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
 825         )
 826
 827         _stdout, stderr = await process.communicate()
 828
 829         return_code = process.returncode
 830
 831         if return_code > 0:
 832             #
 833             if "already exists" not in stderr:
 834                 raise Exception(stderr)
 835
 836     def get_config(self, cluster_uuid: str,) -> dict:
 837         """Get the cluster configuration
 838
 839         Gets the configuration of the cluster
 840
 841         :param cluster_uuid str: The UUID of the cluster.
 842         :return: A dict upon success, or raises an exception.
 843         """
 844         cluster_config = "{}/{}.yaml".format(self.fs.path, cluster_uuid)
 845         if os.path.exists(cluster_config):
 846             with open(cluster_config, "r") as f:
 847                 config = yaml.safe_load(f.read())
 848                 return config
 849         else:
 850             raise Exception(
 851                 "Unable to locate configuration for cluster {}".format(cluster_uuid)
 852             )
 853
 854     async def get_model(self, model_name: str, cluster_uuid: str,) -> Model:
 855         """Get a model from the Juju Controller.
 856
 857         Note: Model objects returned must call disconnected() before it goes
 858         out of scope.
 859
 860         :param model_name str: The name of the model to get
 861         :return The juju.model.Model object if found, or None.
 862         """
 863         if not self.authenticated:
 864             await self.login(cluster_uuid)
 865
 866         model = None
 867         models = await self.controller.list_models()
 868         if model_name in models:
 869             self.log.debug("Found model: {}".format(model_name))
 870             model = await self.controller.get_model(model_name)
 871         return model
 872
 873     def get_namespace(self, cluster_uuid: str,) -> str:
 874         """Get the namespace UUID
 875         Gets the namespace's unique name
 876
 877         :param cluster_uuid str: The UUID of the cluster
 878         :returns: The namespace UUID, or raises an exception
 879         """
 880         config = self.get_config(cluster_uuid)
 881
 882         # Make sure the name is in the config
 883         if "namespace" not in config:
 884             raise Exception("Namespace not found.")
 885
 886         # TODO: We want to make sure this is unique to the cluster, in case
 887         # the cluster is being reused.
 888         # Consider pre/appending the cluster id to the namespace string
 889         return config["namespace"]
 890
 891     async def has_model(self, model_name: str) -> bool:
 892         """Check if a model exists in the controller
 893
 894         Checks to see if a model exists in the connected Juju controller.
 895
 896         :param model_name str: The name of the model
 897         :return: A boolean indicating if the model exists
 898         """
 899         models = await self.controller.list_models()
 900
 901         if model_name in models:
 902             return True
 903         return False
 904
 905     def is_local_k8s(self, credentials: str,) -> bool:
 906         """Check if a cluster is local
 907
 908         Checks if a cluster is running in the local host
 909
 910         :param credentials dict: A dictionary containing the k8s credentials
 911         :returns: A boolean if the cluster is running locally
 912         """
 913         creds = yaml.safe_load(credentials)
 914         if os.getenv("OSMLCM_VCA_APIPROXY"):
 915             host_ip = os.getenv("OSMLCM_VCA_APIPROXY")
 916
 917         if creds and host_ip:
 918             for cluster in creds["clusters"]:
 919                 if "server" in cluster["cluster"]:
 920                     if host_ip in cluster["cluster"]["server"]:
 921                         return True
 922
 923         return False
 924
 925     async def login(self, cluster_uuid):
 926         """Login to the Juju controller."""
 927
 928         if self.authenticated:
 929             return
 930
 931         self.connecting = True
 932
 933         # Test: Make sure we have the credentials loaded
 934         config = self.get_config(cluster_uuid)
 935
 936         self.juju_endpoint = config["endpoint"]
 937         self.juju_user = config["username"]
 938         self.juju_secret = config["secret"]
 939         self.juju_ca_cert = config["cacert"]
 940         self.juju_public_key = None
 941
 942         self.controller = Controller()
 943
 944         if self.juju_secret:
 945             self.log.debug(
 946                 "Connecting to controller... ws://{} as {}/{}".format(
 947                     self.juju_endpoint, self.juju_user, self.juju_secret,
 948                 )
 949             )
 950             try:
 951                 await self.controller.connect(
 952                     endpoint=self.juju_endpoint,
 953                     username=self.juju_user,
 954                     password=self.juju_secret,
 955                     cacert=self.juju_ca_cert,
 956                 )
 957                 self.authenticated = True
 958                 self.log.debug("JujuApi: Logged into controller")
 959             except Exception as ex:
 960                 self.log.debug(ex)
 961                 self.log.debug("Caught exception: {}".format(ex))
 962                 pass
 963         else:
 964             self.log.fatal("VCA credentials not configured.")
 965             self.authenticated = False
 966
 967     async def logout(self):
 968         """Logout of the Juju controller."""
 969         self.log.debug("[logout]")
 970         if not self.authenticated:
 971             return False
 972
 973         for model in self.models:
 974             self.log.debug("Logging out of model {}".format(model))
 975             await self.models[model].disconnect()
 976
 977         if self.controller:
 978             self.log.debug("Disconnecting controller {}".format(self.controller))
 979             await self.controller.disconnect()
 980             self.controller = None
 981
 982         self.authenticated = False
 983
 984     async def remove_cloud(self, cloud_name: str,) -> bool:
 985         """Remove a k8s cloud from Juju
 986
 987         Removes a Kubernetes cloud from Juju.
 988
 989         :param cloud_name str: The name of the cloud to add.
 990
 991         :returns: True if successful, otherwise raises an exception.
 992         """
 993
 994         # Remove the bootstrapped controller
 995         cmd = [self.juju_command, "remove-k8s", "--client", cloud_name]
 996         process = await asyncio.create_subprocess_exec(
 997             *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
 998         )
 999
1000         _stdout, stderr = await process.communicate()
1001
1002         return_code = process.returncode
1003
1004         if return_code > 0:
1005             raise Exception(stderr)
1006
1007         # Remove the cloud from the local config
1008         cmd = [self.juju_command, "remove-cloud", "--client", cloud_name]
1009         process = await asyncio.create_subprocess_exec(
1010             *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
1011         )
1012
1013         _stdout, stderr = await process.communicate()
1014
1015         return_code = process.returncode
1016
1017         if return_code > 0:
1018             raise Exception(stderr)
1019
1020         return True
1021
1022     async def set_config(self, cluster_uuid: str, config: dict,) -> bool:
1023         """Save the cluster configuration
1024
1025         Saves the cluster information to the file store
1026
1027         :param cluster_uuid str: The UUID of the cluster
1028         :param config dict: A dictionary containing the cluster configuration
1029         :returns: Boolean upon success or raises an exception.
1030         """
1031
1032         cluster_config = "{}/{}.yaml".format(self.fs.path, cluster_uuid)
1033         if not os.path.exists(cluster_config):
1034             self.log.debug("Writing config to {}".format(cluster_config))
1035             with open(cluster_config, "w") as f:
1036                 f.write(yaml.dump(config, Dumper=yaml.Dumper))
1037
1038         return True