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