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     async def synchronize_repos(
 252         self,
 253         cluster_uuid: str,
 254         name: str
 255     ):
 256         """
 257         Returns None as currently add_repo is not implemented
 258         """
 259         return None
 260
 261     """Reset"""
 262     async def reset(
 263             self,
 264             cluster_uuid: str,
 265             force: bool = False,
 266             uninstall_sw: bool = False
 267     ) -> bool:
 268         """Reset a cluster
 269
 270         Resets the Kubernetes cluster by removing the model that represents it.
 271
 272         :param cluster_uuid str: The UUID of the cluster to reset
 273         :return: Returns True if successful or raises an exception.
 274         """
 275
 276         try:
 277             if not self.authenticated:
 278                 await self.login(cluster_uuid)
 279
 280             if self.controller.is_connected():
 281                 # Destroy the model
 282                 namespace = self.get_namespace(cluster_uuid)
 283                 if await self.has_model(namespace):
 284                     print("[reset] Destroying model")
 285                     await self.controller.destroy_model(
 286                         namespace,
 287                         destroy_storage=True
 288                     )
 289
 290                 # Disconnect from the controller
 291                 print("[reset] Disconnecting controller")
 292                 await self.logout()
 293
 294                 # Destroy the controller (via CLI)
 295                 print("[reset] Destroying controller")
 296                 await self.destroy_controller(cluster_uuid)
 297
 298                 print("[reset] Removing k8s cloud")
 299                 k8s_cloud = "k8s-{}".format(cluster_uuid)
 300                 await self.remove_cloud(k8s_cloud)
 301
 302         except Exception as ex:
 303             print("Caught exception during reset: {}".format(ex))
 304
 305         return True
 306
 307     """Deployment"""
 308
 309     async def install(
 310         self,
 311         cluster_uuid: str,
 312         kdu_model: str,
 313         atomic: bool = True,
 314         timeout: float = 300,
 315         params: dict = None,
 316         db_dict: dict = None,
 317         kdu_name: str = None
 318     ) -> bool:
 319         """Install a bundle
 320
 321         :param cluster_uuid str: The UUID of the cluster to install to
 322         :param kdu_model str: The name or path of a bundle to install
 323         :param atomic bool: If set, waits until the model is active and resets
 324                             the cluster on failure.
 325         :param timeout int: The time, in seconds, to wait for the install
 326                             to finish
 327         :param params dict: Key-value pairs of instantiation parameters
 328         :param kdu_name: Name of the KDU instance to be installed
 329
 330         :return: If successful, returns ?
 331         """
 332
 333         if not self.authenticated:
 334             print("[install] Logging in to the controller")
 335             await self.login(cluster_uuid)
 336
 337         ##
 338         # Get or create the model, based on the NS
 339         # uuid.
 340         if kdu_name:
 341             kdu_instance = "{}-{}".format(kdu_name, db_dict["filter"]["_id"])
 342         else:
 343             kdu_instance = db_dict["filter"]["_id"]
 344
 345         self.log.debug("Checking for model named {}".format(kdu_instance))
 346
 347         # Create the new model
 348         self.log.debug("Adding model: {}".format(kdu_instance))
 349         model = await self.add_model(kdu_instance, cluster_uuid=cluster_uuid)
 350
 351         if model:
 352             # TODO: Instantiation parameters
 353
 354             """
 355             "Juju bundle that models the KDU, in any of the following ways:
 356                 - <juju-repo>/<juju-bundle>
 357                 - <juju-bundle folder under k8s_models folder in the package>
 358                 - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder in the package>
 359                 - <URL_where_to_fetch_juju_bundle>
 360             """
 361
 362             bundle = kdu_model
 363             if kdu_model.startswith("cs:"):
 364                 bundle = kdu_model
 365             elif kdu_model.startswith("http"):
 366                 # Download the file
 367                 pass
 368             else:
 369                 # Local file
 370
 371                 # if kdu_model.endswith(".tar.gz") or kdu_model.endswith(".tgz")
 372                 # Uncompress temporarily
 373                 # bundle = <uncompressed file>
 374                 pass
 375
 376             if not bundle:
 377                 # Raise named exception that the bundle could not be found
 378                 raise Exception()
 379
 380             print("[install] deploying {}".format(bundle))
 381             await model.deploy(bundle)
 382
 383             # Get the application
 384             if atomic:
 385                 # applications = model.applications
 386                 print("[install] Applications: {}".format(model.applications))
 387                 for name in model.applications:
 388                     print("[install] Waiting for {} to settle".format(name))
 389                     application = model.applications[name]
 390                     try:
 391                         # It's not enough to wait for all units to be active;
 392                         # the application status needs to be active as well.
 393                         print("Waiting for all units to be active...")
 394                         await model.block_until(
 395                             lambda: all(
 396                                 unit.agent_status == 'idle'
 397                                 and application.status in ['active', 'unknown']
 398                                 and unit.workload_status in [
 399                                     'active', 'unknown'
 400                                 ] for unit in application.units
 401                             ),
 402                             timeout=timeout
 403                         )
 404                         print("All units active.")
 405
 406                     except concurrent.futures._base.TimeoutError:
 407                         print("[install] Timeout exceeded; resetting cluster")
 408                         await self.reset(cluster_uuid)
 409                         return False
 410
 411             # Wait for the application to be active
 412             if model.is_connected():
 413                 print("[install] Disconnecting model")
 414                 await model.disconnect()
 415
 416             return kdu_instance
 417         raise Exception("Unable to install")
 418
 419     async def instances_list(
 420             self,
 421             cluster_uuid: str
 422     ) -> list:
 423         """
 424         returns a list of deployed releases in a cluster
 425
 426         :param cluster_uuid: the cluster
 427         :return:
 428         """
 429         return []
 430
 431     async def upgrade(
 432         self,
 433         cluster_uuid: str,
 434         kdu_instance: str,
 435         kdu_model: str = None,
 436         params: dict = None,
 437     ) -> str:
 438         """Upgrade a model
 439
 440         :param cluster_uuid str: The UUID of the cluster to upgrade
 441         :param kdu_instance str: The unique name of the KDU instance
 442         :param kdu_model str: The name or path of the bundle to upgrade to
 443         :param params dict: Key-value pairs of instantiation parameters
 444
 445         :return: If successful, reference to the new revision number of the
 446                  KDU instance.
 447         """
 448
 449         # TODO: Loop through the bundle and upgrade each charm individually
 450
 451         """
 452         The API doesn't have a concept of bundle upgrades, because there are
 453         many possible changes: charm revision, disk, number of units, etc.
 454
 455         As such, we are only supporting a limited subset of upgrades. We'll
 456         upgrade the charm revision but leave storage and scale untouched.
 457
 458         Scale changes should happen through OSM constructs, and changes to
 459         storage would require a redeployment of the service, at least in this
 460         initial release.
 461         """
 462         namespace = self.get_namespace(cluster_uuid)
 463         model = await self.get_model(namespace, cluster_uuid=cluster_uuid)
 464
 465         with open(kdu_model, 'r') as f:
 466             bundle = yaml.safe_load(f)
 467
 468             """
 469             {
 470                 'description': 'Test bundle',
 471                 'bundle': 'kubernetes',
 472                 'applications': {
 473                     'mariadb-k8s': {
 474                         'charm': 'cs:~charmed-osm/mariadb-k8s-20',
 475                         'scale': 1,
 476                         'options': {
 477                             'password': 'manopw',
 478                             'root_password': 'osm4u',
 479                             'user': 'mano'
 480                         },
 481                         'series': 'kubernetes'
 482                     }
 483                 }
 484             }
 485             """
 486             # TODO: This should be returned in an agreed-upon format
 487             for name in bundle['applications']:
 488                 print(model.applications)
 489                 application = model.applications[name]
 490                 print(application)
 491
 492                 path = bundle['applications'][name]['charm']
 493
 494                 try:
 495                     await application.upgrade_charm(switch=path)
 496                 except juju.errors.JujuError as ex:
 497                     if 'already running charm' in str(ex):
 498                         # We're already running this version
 499                         pass
 500
 501         await model.disconnect()
 502
 503         return True
 504         raise NotImplemented()
 505
 506     """Rollback"""
 507     async def rollback(
 508         self,
 509         cluster_uuid: str,
 510         kdu_instance: str,
 511         revision: int = 0,
 512     ) -> str:
 513         """Rollback a model
 514
 515         :param cluster_uuid str: The UUID of the cluster to rollback
 516         :param kdu_instance str: The unique name of the KDU instance
 517         :param revision int: The revision to revert to. If omitted, rolls back
 518                              the previous upgrade.
 519
 520         :return: If successful, returns the revision of active KDU instance,
 521                  or raises an exception
 522         """
 523         raise NotImplemented()
 524
 525     """Deletion"""
 526     async def uninstall(
 527         self,
 528         cluster_uuid: str,
 529         kdu_instance: str
 530     ) -> bool:
 531         """Uninstall a KDU instance
 532
 533         :param cluster_uuid str: The UUID of the cluster
 534         :param kdu_instance str: The unique name of the KDU instance
 535
 536         :return: Returns True if successful, or raises an exception
 537         """
 538         if not self.authenticated:
 539             self.log.debug("[uninstall] Connecting to controller")
 540             await self.login(cluster_uuid)
 541
 542         self.log.debug("[uninstall] Destroying model")
 543
 544         await self.controller.destroy_models(kdu_instance)
 545
 546         self.log.debug("[uninstall] Model destroyed and disconnecting")
 547         await self.logout()
 548
 549         return True
 550
 551     """Introspection"""
 552     async def inspect_kdu(
 553         self,
 554         kdu_model: str,
 555     ) -> dict:
 556         """Inspect a KDU
 557
 558         Inspects a bundle and returns a dictionary of config parameters and
 559         their default values.
 560
 561         :param kdu_model str: The name or path of the bundle to inspect.
 562
 563         :return: If successful, returns a dictionary of available parameters
 564                  and their default values.
 565         """
 566
 567         kdu = {}
 568         with open(kdu_model, 'r') as f:
 569             bundle = yaml.safe_load(f)
 570
 571             """
 572             {
 573                 'description': 'Test bundle',
 574                 'bundle': 'kubernetes',
 575                 'applications': {
 576                     'mariadb-k8s': {
 577                         'charm': 'cs:~charmed-osm/mariadb-k8s-20',
 578                         'scale': 1,
 579                         'options': {
 580                             'password': 'manopw',
 581                             'root_password': 'osm4u',
 582                             'user': 'mano'
 583                         },
 584                         'series': 'kubernetes'
 585                     }
 586                 }
 587             }
 588             """
 589             # TODO: This should be returned in an agreed-upon format
 590             kdu = bundle['applications']
 591
 592         return kdu
 593
 594     async def help_kdu(
 595         self,
 596         kdu_model: str,
 597     ) -> str:
 598         """View the README
 599
 600         If available, returns the README of the bundle.
 601
 602         :param kdu_model str: The name or path of a bundle
 603
 604         :return: If found, returns the contents of the README.
 605         """
 606         readme = None
 607
 608         files = ['README', 'README.txt', 'README.md']
 609         path = os.path.dirname(kdu_model)
 610         for file in os.listdir(path):
 611             if file in files:
 612                 with open(file, 'r') as f:
 613                     readme = f.read()
 614                     break
 615
 616         return readme
 617
 618     async def status_kdu(
 619         self,
 620         cluster_uuid: str,
 621         kdu_instance: str,
 622     ) -> dict:
 623         """Get the status of the KDU
 624
 625         Get the current status of the KDU instance.
 626
 627         :param cluster_uuid str: The UUID of the cluster
 628         :param kdu_instance str: The unique id of the KDU instance
 629
 630         :return: Returns a dictionary containing namespace, state, resources,
 631                  and deployment_time.
 632         """
 633         status = {}
 634
 635         model = await self.get_model(self.get_namespace(cluster_uuid), cluster_uuid=cluster_uuid)
 636
 637         # model = await self.get_model_by_uuid(cluster_uuid)
 638         if model:
 639             model_status = await model.get_status()
 640             status = model_status.applications
 641
 642             for name in model_status.applications:
 643                 application = model_status.applications[name]
 644                 status[name] = {
 645                     'status': application['status']['status']
 646                 }
 647
 648             if model.is_connected():
 649                 await model.disconnect()
 650
 651         return status
 652
 653     # Private methods
 654     async def add_k8s(
 655         self,
 656         cloud_name: str,
 657         credentials: str,
 658     ) -> bool:
 659         """Add a k8s cloud to Juju
 660
 661         Adds a Kubernetes cloud to Juju, so it can be bootstrapped with a
 662         Juju Controller.
 663
 664         :param cloud_name str: The name of the cloud to add.
 665         :param credentials dict: A dictionary representing the output of
 666             `kubectl config view --raw`.
 667
 668         :returns: True if successful, otherwise raises an exception.
 669         """
 670
 671         cmd = [self.juju_command, "add-k8s", "--local", cloud_name]
 672         print(cmd)
 673
 674         process = await asyncio.create_subprocess_exec(
 675             *cmd,
 676             stdout=asyncio.subprocess.PIPE,
 677             stderr=asyncio.subprocess.PIPE,
 678             stdin=asyncio.subprocess.PIPE,
 679         )
 680
 681         # Feed the process the credentials
 682         process.stdin.write(credentials.encode("utf-8"))
 683         await process.stdin.drain()
 684         process.stdin.close()
 685
 686         stdout, stderr = await process.communicate()
 687
 688         return_code = process.returncode
 689
 690         print("add-k8s return code: {}".format(return_code))
 691
 692         if return_code > 0:
 693             raise Exception(stderr)
 694
 695         return True
 696
 697     async def add_model(
 698         self,
 699         model_name: str,
 700         cluster_uuid: str,
 701     ) -> juju.model.Model:
 702         """Adds a model to the controller
 703
 704         Adds a new model to the Juju controller
 705
 706         :param model_name str: The name of the model to add.
 707         :returns: The juju.model.Model object of the new model upon success or
 708                   raises an exception.
 709         """
 710         if not self.authenticated:
 711             await self.login(cluster_uuid)
 712
 713         self.log.debug("Adding model '{}' to cluster_uuid '{}'".format(model_name, cluster_uuid))
 714         try:
 715             model = await self.controller.add_model(
 716                 model_name,
 717                 config={'authorized-keys': self.juju_public_key}
 718             )
 719         except Exception as ex:
 720             self.log.debug(ex)
 721             self.log.debug("Caught exception: {}".format(ex))
 722             pass
 723
 724         return model
 725
 726     async def bootstrap(
 727         self,
 728         cloud_name: str,
 729         cluster_uuid: str,
 730         loadbalancer: bool
 731     ) -> bool:
 732         """Bootstrap a Kubernetes controller
 733
 734         Bootstrap a Juju controller inside the Kubernetes cluster
 735
 736         :param cloud_name str: The name of the cloud.
 737         :param cluster_uuid str: The UUID of the cluster to bootstrap.
 738         :param loadbalancer bool: If the controller should use loadbalancer or not.
 739         :returns: True upon success or raises an exception.
 740         """
 741
 742         if not loadbalancer:
 743             cmd = [self.juju_command, "bootstrap", cloud_name, cluster_uuid]
 744         else:
 745             """
 746             For public clusters, specify that the controller service is using a LoadBalancer.
 747             """
 748             cmd = [self.juju_command, "bootstrap", cloud_name, cluster_uuid, "--config", "controller-service-type=loadbalancer"]
 749
 750         print("Bootstrapping controller {} in cloud {}".format(
 751             cluster_uuid, cloud_name
 752         ))
 753
 754         process = await asyncio.create_subprocess_exec(
 755             *cmd,
 756             stdout=asyncio.subprocess.PIPE,
 757             stderr=asyncio.subprocess.PIPE,
 758         )
 759
 760         stdout, stderr = await process.communicate()
 761
 762         return_code = process.returncode
 763
 764         if return_code > 0:
 765             #
 766             if b'already exists' not in stderr:
 767                 raise Exception(stderr)
 768
 769         return True
 770
 771     async def destroy_controller(
 772         self,
 773         cluster_uuid: str
 774     ) -> bool:
 775         """Destroy a Kubernetes controller
 776
 777         Destroy an existing Kubernetes controller.
 778
 779         :param cluster_uuid str: The UUID of the cluster to bootstrap.
 780         :returns: True upon success or raises an exception.
 781         """
 782         cmd = [
 783             self.juju_command,
 784             "destroy-controller",
 785             "--destroy-all-models",
 786             "--destroy-storage",
 787             "-y",
 788             cluster_uuid
 789         ]
 790
 791         process = await asyncio.create_subprocess_exec(
 792             *cmd,
 793             stdout=asyncio.subprocess.PIPE,
 794             stderr=asyncio.subprocess.PIPE,
 795         )
 796
 797         stdout, stderr = await process.communicate()
 798
 799         return_code = process.returncode
 800
 801         if return_code > 0:
 802             #
 803             if 'already exists' not in stderr:
 804                 raise Exception(stderr)
 805
 806     def get_config(
 807         self,
 808         cluster_uuid: str,
 809     ) -> dict:
 810         """Get the cluster configuration
 811
 812         Gets the configuration of the cluster
 813
 814         :param cluster_uuid str: The UUID of the cluster.
 815         :return: A dict upon success, or raises an exception.
 816         """
 817         cluster_config = "{}/{}.yaml".format(self.fs.path, cluster_uuid)
 818         if os.path.exists(cluster_config):
 819             with open(cluster_config, 'r') as f:
 820                 config = yaml.safe_load(f.read())
 821                 return config
 822         else:
 823             raise Exception(
 824                 "Unable to locate configuration for cluster {}".format(
 825                     cluster_uuid
 826                 )
 827             )
 828
 829     async def get_model(
 830         self,
 831         model_name: str,
 832         cluster_uuid: str,
 833     ) -> juju.model.Model:
 834         """Get a model from the Juju Controller.
 835
 836         Note: Model objects returned must call disconnected() before it goes
 837         out of scope.
 838
 839         :param model_name str: The name of the model to get
 840         :return The juju.model.Model object if found, or None.
 841         """
 842         if not self.authenticated:
 843             await self.login(cluster_uuid)
 844
 845         model = None
 846         models = await self.controller.list_models()
 847         self.log.debug(models)
 848         if model_name in models:
 849             self.log.debug("Found model: {}".format(model_name))
 850             model = await self.controller.get_model(
 851                 model_name
 852             )
 853         return model
 854
 855     def get_namespace(
 856         self,
 857         cluster_uuid: str,
 858     ) -> str:
 859         """Get the namespace UUID
 860         Gets the namespace's unique name
 861
 862         :param cluster_uuid str: The UUID of the cluster
 863         :returns: The namespace UUID, or raises an exception
 864         """
 865         config = self.get_config(cluster_uuid)
 866
 867         # Make sure the name is in the config
 868         if 'namespace' not in config:
 869             raise Exception("Namespace not found.")
 870
 871         # TODO: We want to make sure this is unique to the cluster, in case
 872         # the cluster is being reused.
 873         # Consider pre/appending the cluster id to the namespace string
 874         return config['namespace']
 875
 876     async def has_model(
 877         self,
 878         model_name: str
 879     ) -> bool:
 880         """Check if a model exists in the controller
 881
 882         Checks to see if a model exists in the connected Juju controller.
 883
 884         :param model_name str: The name of the model
 885         :return: A boolean indicating if the model exists
 886         """
 887         models = await self.controller.list_models()
 888
 889         if model_name in models:
 890             return True
 891         return False
 892
 893     def is_local_k8s(
 894         self,
 895         credentials: str,
 896     ) -> bool:
 897         """Check if a cluster is local
 898
 899         Checks if a cluster is running in the local host
 900
 901         :param credentials dict: A dictionary containing the k8s credentials
 902         :returns: A boolean if the cluster is running locally
 903         """
 904         creds = yaml.safe_load(credentials)
 905         if os.getenv("OSMLCM_VCA_APIPROXY"):
 906             host_ip = os.getenv("OSMLCM_VCA_APIPROXY")
 907
 908         if creds and host_ip:
 909             for cluster in creds['clusters']:
 910                 if 'server' in cluster['cluster']:
 911                     if host_ip in cluster['cluster']['server']:
 912                         return True
 913
 914         return False
 915
 916     async def login(self, cluster_uuid):
 917         """Login to the Juju controller."""
 918
 919         if self.authenticated:
 920             return
 921
 922         self.connecting = True
 923
 924         # Test: Make sure we have the credentials loaded
 925         config = self.get_config(cluster_uuid)
 926
 927         self.juju_endpoint = config['endpoint']
 928         self.juju_user = config['username']
 929         self.juju_secret = config['secret']
 930         self.juju_ca_cert = config['cacert']
 931         self.juju_public_key = None
 932
 933         self.controller = Controller()
 934
 935         if self.juju_secret:
 936             self.log.debug(
 937                 "Connecting to controller... ws://{} as {}/{}".format(
 938                     self.juju_endpoint,
 939                     self.juju_user,
 940                     self.juju_secret,
 941                 )
 942             )
 943             try:
 944                 await self.controller.connect(
 945                     endpoint=self.juju_endpoint,
 946                     username=self.juju_user,
 947                     password=self.juju_secret,
 948                     cacert=self.juju_ca_cert,
 949                 )
 950                 self.authenticated = True
 951                 self.log.debug("JujuApi: Logged into controller")
 952             except Exception as ex:
 953                 print(ex)
 954                 self.log.debug("Caught exception: {}".format(ex))
 955                 pass
 956         else:
 957             self.log.fatal("VCA credentials not configured.")
 958             self.authenticated = False
 959
 960     async def logout(self):
 961         """Logout of the Juju controller."""
 962         print("[logout]")
 963         if not self.authenticated:
 964             return False
 965
 966         for model in self.models:
 967             print("Logging out of model {}".format(model))
 968             await self.models[model].disconnect()
 969
 970         if self.controller:
 971             self.log.debug("Disconnecting controller {}".format(
 972                 self.controller
 973             ))
 974             await self.controller.disconnect()
 975             self.controller = None
 976
 977         self.authenticated = False
 978
 979     async def remove_cloud(
 980         self,
 981         cloud_name: str,
 982     ) -> bool:
 983         """Remove a k8s cloud from Juju
 984
 985         Removes a Kubernetes cloud from Juju.
 986
 987         :param cloud_name str: The name of the cloud to add.
 988
 989         :returns: True if successful, otherwise raises an exception.
 990         """
 991
 992         # Remove the bootstrapped controller
 993         cmd = [self.juju_command, "remove-k8s", "--client", cloud_name]
 994         process = await asyncio.create_subprocess_exec(
 995             *cmd,
 996             stdout=asyncio.subprocess.PIPE,
 997             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,
1011             stdout=asyncio.subprocess.PIPE,
1012             stderr=asyncio.subprocess.PIPE,
1013         )
1014
1015         stdout, stderr = await process.communicate()
1016
1017         return_code = process.returncode
1018
1019         if return_code > 0:
1020             raise Exception(stderr)
1021
1022         return True
1023
1024     async def set_config(
1025         self,
1026         cluster_uuid: str,
1027         config: dict,
1028     ) -> bool:
1029         """Save the cluster configuration
1030
1031         Saves the cluster information to the file store
1032
1033         :param cluster_uuid str: The UUID of the cluster
1034         :param config dict: A dictionary containing the cluster configuration
1035         :returns: Boolean upon success or raises an exception.
1036         """
1037
1038         cluster_config = "{}/{}.yaml".format(self.fs.path, cluster_uuid)
1039         if not os.path.exists(cluster_config):
1040             print("Writing config to {}".format(cluster_config))
1041             with open(cluster_config, 'w') as f:
1042                 f.write(yaml.dump(config, Dumper=yaml.Dumper))
1043
1044         return True