n2vc/libjuju.py

   1 # Copyright 2020 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 logging
  17
  18 import time
  19
  20 from juju.errors import JujuAPIError
  21 from juju.model import Model
  22 from juju.machine import Machine
  23 from juju.application import Application
  24 from juju.unit import Unit
  25 from juju.client._definitions import (
  26     FullStatus,
  27     QueryApplicationOffersResults,
  28     Cloud,
  29     CloudCredential,
  30 )
  31 from juju.controller import Controller
  32 from juju.client import client
  33 from juju import tag
  34
  35 from n2vc.juju_watcher import JujuModelWatcher
  36 from n2vc.provisioner import AsyncSSHProvisioner
  37 from n2vc.n2vc_conn import N2VCConnector
  38 from n2vc.exceptions import (
  39     JujuMachineNotFound,
  40     JujuApplicationNotFound,
  41     JujuLeaderUnitNotFound,
  42     JujuActionNotFound,
  43     JujuModelAlreadyExists,
  44     JujuControllerFailedConnecting,
  45     JujuApplicationExists,
  46     JujuInvalidK8sConfiguration,
  47 )
  48 from n2vc.utils import DB_DATA
  49 from osm_common.dbbase import DbException
  50 from kubernetes.client.configuration import Configuration
  51
  52 RBAC_LABEL_KEY_NAME = "rbac-id"
  53
  54
  55 class Libjuju:
  56     def __init__(
  57         self,
  58         endpoint: str,
  59         api_proxy: str,
  60         username: str,
  61         password: str,
  62         cacert: str,
  63         loop: asyncio.AbstractEventLoop = None,
  64         log: logging.Logger = None,
  65         db: dict = None,
  66         n2vc: N2VCConnector = None,
  67         apt_mirror: str = None,
  68         enable_os_upgrade: bool = True,
  69     ):
  70         """
  71         Constructor
  72
  73         :param: endpoint:               Endpoint of the juju controller (host:port)
  74         :param: api_proxy:              Endpoint of the juju controller - Reachable from the VNFs
  75         :param: username:               Juju username
  76         :param: password:               Juju password
  77         :param: cacert:                 Juju CA Certificate
  78         :param: loop:                   Asyncio loop
  79         :param: log:                    Logger
  80         :param: db:                     DB object
  81         :param: n2vc:                   N2VC object
  82         :param: apt_mirror:             APT Mirror
  83         :param: enable_os_upgrade:      Enable OS Upgrade
  84         """
  85
  86         self.log = log or logging.getLogger("Libjuju")
  87         self.db = db
  88         db_endpoints = self._get_api_endpoints_db()
  89         self.endpoints = None
  90         if (db_endpoints and endpoint not in db_endpoints) or not db_endpoints:
  91             self.endpoints = [endpoint]
  92             self._update_api_endpoints_db(self.endpoints)
  93         else:
  94             self.endpoints = db_endpoints
  95         self.api_proxy = api_proxy
  96         self.username = username
  97         self.password = password
  98         self.cacert = cacert
  99         self.loop = loop or asyncio.get_event_loop()
 100         self.n2vc = n2vc
 101
 102         # Generate config for models
 103         self.model_config = {}
 104         if apt_mirror:
 105             self.model_config["apt-mirror"] = apt_mirror
 106         self.model_config["enable-os-refresh-update"] = enable_os_upgrade
 107         self.model_config["enable-os-upgrade"] = enable_os_upgrade
 108
 109         self.loop.set_exception_handler(self.handle_exception)
 110         self.creating_model = asyncio.Lock(loop=self.loop)
 111
 112         self.models = set()
 113         self.log.debug("Libjuju initialized!")
 114
 115         self.health_check_task = self._create_health_check_task()
 116
 117     def _create_health_check_task(self):
 118         return self.loop.create_task(self.health_check())
 119
 120     async def get_controller(self, timeout: float = 5.0) -> Controller:
 121         """
 122         Get controller
 123
 124         :param: timeout: Time in seconds to wait for controller to connect
 125         """
 126         controller = None
 127         try:
 128             controller = Controller(loop=self.loop)
 129             await asyncio.wait_for(
 130                 controller.connect(
 131                     endpoint=self.endpoints,
 132                     username=self.username,
 133                     password=self.password,
 134                     cacert=self.cacert,
 135                 ),
 136                 timeout=timeout,
 137             )
 138             endpoints = await controller.api_endpoints
 139             if self.endpoints != endpoints:
 140                 self.endpoints = endpoints
 141                 self._update_api_endpoints_db(self.endpoints)
 142             return controller
 143         except asyncio.CancelledError as e:
 144             raise e
 145         except Exception as e:
 146             self.log.error(
 147                 "Failed connecting to controller: {}...".format(self.endpoints)
 148             )
 149             if controller:
 150                 await self.disconnect_controller(controller)
 151             raise JujuControllerFailedConnecting(e)
 152
 153     async def disconnect(self):
 154         """Disconnect"""
 155         # Cancel health check task
 156         self.health_check_task.cancel()
 157         self.log.debug("Libjuju disconnected!")
 158
 159     async def disconnect_model(self, model: Model):
 160         """
 161         Disconnect model
 162
 163         :param: model: Model that will be disconnected
 164         """
 165         await model.disconnect()
 166
 167     async def disconnect_controller(self, controller: Controller):
 168         """
 169         Disconnect controller
 170
 171         :param: controller: Controller that will be disconnected
 172         """
 173         if controller:
 174             await controller.disconnect()
 175
 176     async def add_model(self, model_name: str, cloud_name: str, credential_name=None):
 177         """
 178         Create model
 179
 180         :param: model_name: Model name
 181         :param: cloud_name: Cloud name
 182         :param: credential_name: Credential name to use for adding the model
 183                                  If not specified, same name as the cloud will be used.
 184         """
 185
 186         # Get controller
 187         controller = await self.get_controller()
 188         model = None
 189         try:
 190             # Raise exception if model already exists
 191             if await self.model_exists(model_name, controller=controller):
 192                 raise JujuModelAlreadyExists(
 193                     "Model {} already exists.".format(model_name)
 194                 )
 195
 196             # Block until other workers have finished model creation
 197             while self.creating_model.locked():
 198                 await asyncio.sleep(0.1)
 199
 200             # If the model exists, return it from the controller
 201             if model_name in self.models:
 202                 return
 203
 204             # Create the model
 205             async with self.creating_model:
 206                 self.log.debug("Creating model {}".format(model_name))
 207                 model = await controller.add_model(
 208                     model_name,
 209                     config=self.model_config,
 210                     cloud_name=cloud_name,
 211                     credential_name=credential_name or cloud_name,
 212                 )
 213                 self.models.add(model_name)
 214         finally:
 215             if model:
 216                 await self.disconnect_model(model)
 217             await self.disconnect_controller(controller)
 218
 219     async def get_model(
 220         self, controller: Controller, model_name: str, id=None
 221     ) -> Model:
 222         """
 223         Get model from controller
 224
 225         :param: controller: Controller
 226         :param: model_name: Model name
 227
 228         :return: Model: The created Juju model object
 229         """
 230         return await controller.get_model(model_name)
 231
 232     async def model_exists(
 233         self, model_name: str, controller: Controller = None
 234     ) -> bool:
 235         """
 236         Check if model exists
 237
 238         :param: controller: Controller
 239         :param: model_name: Model name
 240
 241         :return bool
 242         """
 243         need_to_disconnect = False
 244
 245         # Get controller if not passed
 246         if not controller:
 247             controller = await self.get_controller()
 248             need_to_disconnect = True
 249
 250         # Check if model exists
 251         try:
 252             return model_name in await controller.list_models()
 253         finally:
 254             if need_to_disconnect:
 255                 await self.disconnect_controller(controller)
 256
 257     async def models_exist(self, model_names: [str]) -> (bool, list):
 258         """
 259         Check if models exists
 260
 261         :param: model_names: List of strings with model names
 262
 263         :return (bool, list[str]): (True if all models exists, List of model names that don't exist)
 264         """
 265         if not model_names:
 266             raise Exception(
 267                 "model_names must be a non-empty array. Given value: {}".format(
 268                     model_names
 269                 )
 270             )
 271         non_existing_models = []
 272         models = await self.list_models()
 273         existing_models = list(set(models).intersection(model_names))
 274         non_existing_models = list(set(model_names) - set(existing_models))
 275
 276         return (
 277             len(non_existing_models) == 0,
 278             non_existing_models,
 279         )
 280
 281     async def get_model_status(self, model_name: str) -> FullStatus:
 282         """
 283         Get model status
 284
 285         :param: model_name: Model name
 286
 287         :return: Full status object
 288         """
 289         controller = await self.get_controller()
 290         model = await self.get_model(controller, model_name)
 291         try:
 292             return await model.get_status()
 293         finally:
 294             await self.disconnect_model(model)
 295             await self.disconnect_controller(controller)
 296
 297     async def create_machine(
 298         self,
 299         model_name: str,
 300         machine_id: str = None,
 301         db_dict: dict = None,
 302         progress_timeout: float = None,
 303         total_timeout: float = None,
 304         series: str = "xenial",
 305         wait: bool = True,
 306     ) -> (Machine, bool):
 307         """
 308         Create machine
 309
 310         :param: model_name:         Model name
 311         :param: machine_id:         Machine id
 312         :param: db_dict:            Dictionary with data of the DB to write the updates
 313         :param: progress_timeout:   Maximum time between two updates in the model
 314         :param: total_timeout:      Timeout for the entity to be active
 315         :param: series:             Series of the machine (xenial, bionic, focal, ...)
 316         :param: wait:               Wait until machine is ready
 317
 318         :return: (juju.machine.Machine, bool):  Machine object and a boolean saying
 319                                                 if the machine is new or it already existed
 320         """
 321         new = False
 322         machine = None
 323
 324         self.log.debug(
 325             "Creating machine (id={}) in model: {}".format(machine_id, model_name)
 326         )
 327
 328         # Get controller
 329         controller = await self.get_controller()
 330
 331         # Get model
 332         model = await self.get_model(controller, model_name)
 333         try:
 334             if machine_id is not None:
 335                 self.log.debug(
 336                     "Searching machine (id={}) in model {}".format(
 337                         machine_id, model_name
 338                     )
 339                 )
 340
 341                 # Get machines from model and get the machine with machine_id if exists
 342                 machines = await model.get_machines()
 343                 if machine_id in machines:
 344                     self.log.debug(
 345                         "Machine (id={}) found in model {}".format(
 346                             machine_id, model_name
 347                         )
 348                     )
 349                     machine = machines[machine_id]
 350                 else:
 351                     raise JujuMachineNotFound("Machine {} not found".format(machine_id))
 352
 353             if machine is None:
 354                 self.log.debug("Creating a new machine in model {}".format(model_name))
 355
 356                 # Create machine
 357                 machine = await model.add_machine(
 358                     spec=None, constraints=None, disks=None, series=series
 359                 )
 360                 new = True
 361
 362                 # Wait until the machine is ready
 363                 self.log.debug(
 364                     "Wait until machine {} is ready in model {}".format(
 365                         machine.entity_id, model_name
 366                     )
 367                 )
 368                 if wait:
 369                     await JujuModelWatcher.wait_for(
 370                         model=model,
 371                         entity=machine,
 372                         progress_timeout=progress_timeout,
 373                         total_timeout=total_timeout,
 374                         db_dict=db_dict,
 375                         n2vc=self.n2vc,
 376                     )
 377         finally:
 378             await self.disconnect_model(model)
 379             await self.disconnect_controller(controller)
 380
 381         self.log.debug(
 382             "Machine {} ready at {} in model {}".format(
 383                 machine.entity_id, machine.dns_name, model_name
 384             )
 385         )
 386         return machine, new
 387
 388     async def provision_machine(
 389         self,
 390         model_name: str,
 391         hostname: str,
 392         username: str,
 393         private_key_path: str,
 394         db_dict: dict = None,
 395         progress_timeout: float = None,
 396         total_timeout: float = None,
 397     ) -> str:
 398         """
 399         Manually provisioning of a machine
 400
 401         :param: model_name:         Model name
 402         :param: hostname:           IP to access the machine
 403         :param: username:           Username to login to the machine
 404         :param: private_key_path:   Local path for the private key
 405         :param: db_dict:            Dictionary with data of the DB to write the updates
 406         :param: progress_timeout:   Maximum time between two updates in the model
 407         :param: total_timeout:      Timeout for the entity to be active
 408
 409         :return: (Entity): Machine id
 410         """
 411         self.log.debug(
 412             "Provisioning machine. model: {}, hostname: {}, username: {}".format(
 413                 model_name, hostname, username
 414             )
 415         )
 416
 417         # Get controller
 418         controller = await self.get_controller()
 419
 420         # Get model
 421         model = await self.get_model(controller, model_name)
 422
 423         try:
 424             # Get provisioner
 425             provisioner = AsyncSSHProvisioner(
 426                 host=hostname,
 427                 user=username,
 428                 private_key_path=private_key_path,
 429                 log=self.log,
 430             )
 431
 432             # Provision machine
 433             params = await provisioner.provision_machine()
 434
 435             params.jobs = ["JobHostUnits"]
 436
 437             self.log.debug("Adding machine to model")
 438             connection = model.connection()
 439             client_facade = client.ClientFacade.from_connection(connection)
 440
 441             results = await client_facade.AddMachines(params=[params])
 442             error = results.machines[0].error
 443
 444             if error:
 445                 msg = "Error adding machine: {}".format(error.message)
 446                 self.log.error(msg=msg)
 447                 raise ValueError(msg)
 448
 449             machine_id = results.machines[0].machine
 450
 451             self.log.debug("Installing Juju agent into machine {}".format(machine_id))
 452             asyncio.ensure_future(
 453                 provisioner.install_agent(
 454                     connection=connection,
 455                     nonce=params.nonce,
 456                     machine_id=machine_id,
 457                     proxy=self.api_proxy,
 458                 )
 459             )
 460
 461             machine = None
 462             for _ in range(10):
 463                 machine_list = await model.get_machines()
 464                 if machine_id in machine_list:
 465                     self.log.debug("Machine {} found in model!".format(machine_id))
 466                     machine = model.machines.get(machine_id)
 467                     break
 468                 await asyncio.sleep(2)
 469
 470             if machine is None:
 471                 msg = "Machine {} not found in model".format(machine_id)
 472                 self.log.error(msg=msg)
 473                 raise JujuMachineNotFound(msg)
 474
 475             self.log.debug(
 476                 "Wait until machine {} is ready in model {}".format(
 477                     machine.entity_id, model_name
 478                 )
 479             )
 480             await JujuModelWatcher.wait_for(
 481                 model=model,
 482                 entity=machine,
 483                 progress_timeout=progress_timeout,
 484                 total_timeout=total_timeout,
 485                 db_dict=db_dict,
 486                 n2vc=self.n2vc,
 487             )
 488         except Exception as e:
 489             raise e
 490         finally:
 491             await self.disconnect_model(model)
 492             await self.disconnect_controller(controller)
 493
 494         self.log.debug(
 495             "Machine provisioned {} in model {}".format(machine_id, model_name)
 496         )
 497
 498         return machine_id
 499
 500     async def deploy(
 501         self, uri: str, model_name: str, wait: bool = True, timeout: float = 3600
 502     ):
 503         """
 504         Deploy bundle or charm: Similar to the juju CLI command `juju deploy`
 505
 506         :param: uri:            Path or Charm Store uri in which the charm or bundle can be found
 507         :param: model_name:     Model name
 508         :param: wait:           Indicates whether to wait or not until all applications are active
 509         :param: timeout:        Time in seconds to wait until all applications are active
 510         """
 511         controller = await self.get_controller()
 512         model = await self.get_model(controller, model_name)
 513         try:
 514             await model.deploy(uri)
 515             if wait:
 516                 await JujuModelWatcher.wait_for_model(model, timeout=timeout)
 517                 self.log.debug("All units active in model {}".format(model_name))
 518         finally:
 519             await self.disconnect_model(model)
 520             await self.disconnect_controller(controller)
 521
 522     async def deploy_charm(
 523         self,
 524         application_name: str,
 525         path: str,
 526         model_name: str,
 527         machine_id: str,
 528         db_dict: dict = None,
 529         progress_timeout: float = None,
 530         total_timeout: float = None,
 531         config: dict = None,
 532         series: str = None,
 533         num_units: int = 1,
 534     ):
 535         """Deploy charm
 536
 537         :param: application_name:   Application name
 538         :param: path:               Local path to the charm
 539         :param: model_name:         Model name
 540         :param: machine_id          ID of the machine
 541         :param: db_dict:            Dictionary with data of the DB to write the updates
 542         :param: progress_timeout:   Maximum time between two updates in the model
 543         :param: total_timeout:      Timeout for the entity to be active
 544         :param: config:             Config for the charm
 545         :param: series:             Series of the charm
 546         :param: num_units:          Number of units
 547
 548         :return: (juju.application.Application): Juju application
 549         """
 550         self.log.debug(
 551             "Deploying charm {} to machine {} in model ~{}".format(
 552                 application_name, machine_id, model_name
 553             )
 554         )
 555         self.log.debug("charm: {}".format(path))
 556
 557         # Get controller
 558         controller = await self.get_controller()
 559
 560         # Get model
 561         model = await self.get_model(controller, model_name)
 562
 563         try:
 564             application = None
 565             if application_name not in model.applications:
 566
 567                 if machine_id is not None:
 568                     if machine_id not in model.machines:
 569                         msg = "Machine {} not found in model".format(machine_id)
 570                         self.log.error(msg=msg)
 571                         raise JujuMachineNotFound(msg)
 572                     machine = model.machines[machine_id]
 573                     series = machine.series
 574
 575                 application = await model.deploy(
 576                     entity_url=path,
 577                     application_name=application_name,
 578                     channel="stable",
 579                     num_units=1,
 580                     series=series,
 581                     to=machine_id,
 582                     config=config,
 583                 )
 584
 585                 self.log.debug(
 586                     "Wait until application {} is ready in model {}".format(
 587                         application_name, model_name
 588                     )
 589                 )
 590                 if num_units > 1:
 591                     for _ in range(num_units - 1):
 592                         m, _ = await self.create_machine(model_name, wait=False)
 593                         await application.add_unit(to=m.entity_id)
 594
 595                 await JujuModelWatcher.wait_for(
 596                     model=model,
 597                     entity=application,
 598                     progress_timeout=progress_timeout,
 599                     total_timeout=total_timeout,
 600                     db_dict=db_dict,
 601                     n2vc=self.n2vc,
 602                 )
 603                 self.log.debug(
 604                     "Application {} is ready in model {}".format(
 605                         application_name, model_name
 606                     )
 607                 )
 608             else:
 609                 raise JujuApplicationExists(
 610                     "Application {} exists".format(application_name)
 611                 )
 612         finally:
 613             await self.disconnect_model(model)
 614             await self.disconnect_controller(controller)
 615
 616         return application
 617
 618     def _get_application(self, model: Model, application_name: str) -> Application:
 619         """Get application
 620
 621         :param: model:              Model object
 622         :param: application_name:   Application name
 623
 624         :return: juju.application.Application (or None if it doesn't exist)
 625         """
 626         if model.applications and application_name in model.applications:
 627             return model.applications[application_name]
 628
 629     async def execute_action(
 630         self,
 631         application_name: str,
 632         model_name: str,
 633         action_name: str,
 634         db_dict: dict = None,
 635         progress_timeout: float = None,
 636         total_timeout: float = None,
 637         **kwargs
 638     ):
 639         """Execute action
 640
 641         :param: application_name:   Application name
 642         :param: model_name:         Model name
 643         :param: action_name:        Name of the action
 644         :param: db_dict:            Dictionary with data of the DB to write the updates
 645         :param: progress_timeout:   Maximum time between two updates in the model
 646         :param: total_timeout:      Timeout for the entity to be active
 647
 648         :return: (str, str): (output and status)
 649         """
 650         self.log.debug(
 651             "Executing action {} using params {}".format(action_name, kwargs)
 652         )
 653         # Get controller
 654         controller = await self.get_controller()
 655
 656         # Get model
 657         model = await self.get_model(controller, model_name)
 658
 659         try:
 660             # Get application
 661             application = self._get_application(
 662                 model,
 663                 application_name=application_name,
 664             )
 665             if application is None:
 666                 raise JujuApplicationNotFound("Cannot execute action")
 667
 668             # Get leader unit
 669             # Racing condition:
 670             #   Ocassionally, self._get_leader_unit() will return None
 671             #   because the leader elected hook has not been triggered yet.
 672             #   Therefore, we are doing some retries. If it happens again,
 673             #   re-open bug 1236
 674             attempts = 3
 675             time_between_retries = 10
 676             unit = None
 677             for _ in range(attempts):
 678                 unit = await self._get_leader_unit(application)
 679                 if unit is None:
 680                     await asyncio.sleep(time_between_retries)
 681                 else:
 682                     break
 683             if unit is None:
 684                 raise JujuLeaderUnitNotFound(
 685                     "Cannot execute action: leader unit not found"
 686                 )
 687
 688             actions = await application.get_actions()
 689
 690             if action_name not in actions:
 691                 raise JujuActionNotFound(
 692                     "Action {} not in available actions".format(action_name)
 693                 )
 694
 695             action = await unit.run_action(action_name, **kwargs)
 696
 697             self.log.debug(
 698                 "Wait until action {} is completed in application {} (model={})".format(
 699                     action_name, application_name, model_name
 700                 )
 701             )
 702             await JujuModelWatcher.wait_for(
 703                 model=model,
 704                 entity=action,
 705                 progress_timeout=progress_timeout,
 706                 total_timeout=total_timeout,
 707                 db_dict=db_dict,
 708                 n2vc=self.n2vc,
 709             )
 710
 711             output = await model.get_action_output(action_uuid=action.entity_id)
 712             status = await model.get_action_status(uuid_or_prefix=action.entity_id)
 713             status = (
 714                 status[action.entity_id] if action.entity_id in status else "failed"
 715             )
 716
 717             self.log.debug(
 718                 "Action {} completed with status {} in application {} (model={})".format(
 719                     action_name, action.status, application_name, model_name
 720                 )
 721             )
 722         finally:
 723             await self.disconnect_model(model)
 724             await self.disconnect_controller(controller)
 725
 726         return output, status
 727
 728     async def get_actions(self, application_name: str, model_name: str) -> dict:
 729         """Get list of actions
 730
 731         :param: application_name: Application name
 732         :param: model_name: Model name
 733
 734         :return: Dict with this format
 735             {
 736                 "action_name": "Description of the action",
 737                 ...
 738             }
 739         """
 740         self.log.debug(
 741             "Getting list of actions for application {}".format(application_name)
 742         )
 743
 744         # Get controller
 745         controller = await self.get_controller()
 746
 747         # Get model
 748         model = await self.get_model(controller, model_name)
 749
 750         try:
 751             # Get application
 752             application = self._get_application(
 753                 model,
 754                 application_name=application_name,
 755             )
 756
 757             # Return list of actions
 758             return await application.get_actions()
 759
 760         finally:
 761             # Disconnect from model and controller
 762             await self.disconnect_model(model)
 763             await self.disconnect_controller(controller)
 764
 765     async def get_metrics(self, model_name: str, application_name: str) -> dict:
 766         """Get the metrics collected by the VCA.
 767
 768         :param model_name The name or unique id of the network service
 769         :param application_name The name of the application
 770         """
 771         if not model_name or not application_name:
 772             raise Exception("model_name and application_name must be non-empty strings")
 773         metrics = {}
 774         controller = await self.get_controller()
 775         model = await self.get_model(controller, model_name)
 776         try:
 777             application = self._get_application(model, application_name)
 778             if application is not None:
 779                 metrics = await application.get_metrics()
 780         finally:
 781             self.disconnect_model(model)
 782             self.disconnect_controller(controller)
 783         return metrics
 784
 785     async def add_relation(
 786         self,
 787         model_name: str,
 788         endpoint_1: str,
 789         endpoint_2: str,
 790     ):
 791         """Add relation
 792
 793         :param: model_name:     Model name
 794         :param: endpoint_1      First endpoint name
 795                                 ("app:endpoint" format or directly the saas name)
 796         :param: endpoint_2:     Second endpoint name (^ same format)
 797         """
 798
 799         self.log.debug("Adding relation: {} -> {}".format(endpoint_1, endpoint_2))
 800
 801         # Get controller
 802         controller = await self.get_controller()
 803
 804         # Get model
 805         model = await self.get_model(controller, model_name)
 806
 807         # Add relation
 808         try:
 809             await model.add_relation(endpoint_1, endpoint_2)
 810         except JujuAPIError as e:
 811             if "not found" in e.message:
 812                 self.log.warning("Relation not found: {}".format(e.message))
 813                 return
 814             if "already exists" in e.message:
 815                 self.log.warning("Relation already exists: {}".format(e.message))
 816                 return
 817             # another exception, raise it
 818             raise e
 819         finally:
 820             await self.disconnect_model(model)
 821             await self.disconnect_controller(controller)
 822
 823     async def consume(
 824         self,
 825         offer_url: str,
 826         model_name: str,
 827     ):
 828         """
 829         Adds a remote offer to the model. Relations can be created later using "juju relate".
 830
 831         :param: offer_url:      Offer Url
 832         :param: model_name:     Model name
 833
 834         :raises ParseError if there's a problem parsing the offer_url
 835         :raises JujuError if remote offer includes and endpoint
 836         :raises JujuAPIError if the operation is not successful
 837         """
 838         controller = await self.get_controller()
 839         model = await controller.get_model(model_name)
 840
 841         try:
 842             await model.consume(offer_url)
 843         finally:
 844             await self.disconnect_model(model)
 845             await self.disconnect_controller(controller)
 846
 847     async def destroy_model(self, model_name: str, total_timeout: float):
 848         """
 849         Destroy model
 850
 851         :param: model_name:     Model name
 852         :param: total_timeout:  Timeout
 853         """
 854
 855         controller = await self.get_controller()
 856         model = await self.get_model(controller, model_name)
 857         try:
 858             self.log.debug("Destroying model {}".format(model_name))
 859             uuid = model.info.uuid
 860
 861             # Destroy machines that are manually provisioned
 862             # and still are in pending state
 863             await self._destroy_pending_machines(model, only_manual=True)
 864
 865             # Disconnect model
 866             await self.disconnect_model(model)
 867
 868             # Destroy model
 869             if model_name in self.models:
 870                 self.models.remove(model_name)
 871
 872             await controller.destroy_model(uuid, force=True, max_wait=0)
 873
 874             # Wait until model is destroyed
 875             self.log.debug("Waiting for model {} to be destroyed...".format(model_name))
 876
 877             if total_timeout is None:
 878                 total_timeout = 3600
 879             end = time.time() + total_timeout
 880             while time.time() < end:
 881                 models = await controller.list_models()
 882                 if model_name not in models:
 883                     self.log.debug(
 884                         "The model {} ({}) was destroyed".format(model_name, uuid)
 885                     )
 886                     return
 887                 await asyncio.sleep(5)
 888             raise Exception(
 889                 "Timeout waiting for model {} to be destroyed".format(model_name)
 890             )
 891         finally:
 892             await self.disconnect_controller(controller)
 893
 894     async def destroy_application(self, model: Model, application_name: str):
 895         """
 896         Destroy application
 897
 898         :param: model:              Model object
 899         :param: application_name:   Application name
 900         """
 901         self.log.debug(
 902             "Destroying application {} in model {}".format(
 903                 application_name, model.info.name
 904             )
 905         )
 906         application = model.applications.get(application_name)
 907         if application:
 908             await application.destroy()
 909         else:
 910             self.log.warning("Application not found: {}".format(application_name))
 911
 912     async def _destroy_pending_machines(self, model: Model, only_manual: bool = False):
 913         """
 914         Destroy pending machines in a given model
 915
 916         :param: only_manual:    Bool that indicates only manually provisioned
 917                                 machines should be destroyed (if True), or that
 918                                 all pending machines should be destroyed
 919         """
 920         status = await model.get_status()
 921         for machine_id in status.machines:
 922             machine_status = status.machines[machine_id]
 923             if machine_status.agent_status.status == "pending":
 924                 if only_manual and not machine_status.instance_id.startswith("manual:"):
 925                     break
 926                 machine = model.machines[machine_id]
 927                 await machine.destroy(force=True)
 928
 929     async def configure_application(
 930         self, model_name: str, application_name: str, config: dict = None
 931     ):
 932         """Configure application
 933
 934         :param: model_name:         Model name
 935         :param: application_name:   Application name
 936         :param: config:             Config to apply to the charm
 937         """
 938         self.log.debug("Configuring application {}".format(application_name))
 939
 940         if config:
 941             controller = await self.get_controller()
 942             model = None
 943             try:
 944                 model = await self.get_model(controller, model_name)
 945                 application = self._get_application(
 946                     model,
 947                     application_name=application_name,
 948                 )
 949                 await application.set_config(config)
 950             finally:
 951                 if model:
 952                     await self.disconnect_model(model)
 953                 await self.disconnect_controller(controller)
 954
 955     def _get_api_endpoints_db(self) -> [str]:
 956         """
 957         Get API Endpoints from DB
 958
 959         :return: List of API endpoints
 960         """
 961         self.log.debug("Getting endpoints from database")
 962
 963         juju_info = self.db.get_one(
 964             DB_DATA.api_endpoints.table,
 965             q_filter=DB_DATA.api_endpoints.filter,
 966             fail_on_empty=False,
 967         )
 968         if juju_info and DB_DATA.api_endpoints.key in juju_info:
 969             return juju_info[DB_DATA.api_endpoints.key]
 970
 971     def _update_api_endpoints_db(self, endpoints: [str]):
 972         """
 973         Update API endpoints in Database
 974
 975         :param: List of endpoints
 976         """
 977         self.log.debug("Saving endpoints {} in database".format(endpoints))
 978
 979         juju_info = self.db.get_one(
 980             DB_DATA.api_endpoints.table,
 981             q_filter=DB_DATA.api_endpoints.filter,
 982             fail_on_empty=False,
 983         )
 984         # If it doesn't, then create it
 985         if not juju_info:
 986             try:
 987                 self.db.create(
 988                     DB_DATA.api_endpoints.table,
 989                     DB_DATA.api_endpoints.filter,
 990                 )
 991             except DbException as e:
 992                 # Racing condition: check if another N2VC worker has created it
 993                 juju_info = self.db.get_one(
 994                     DB_DATA.api_endpoints.table,
 995                     q_filter=DB_DATA.api_endpoints.filter,
 996                     fail_on_empty=False,
 997                 )
 998                 if not juju_info:
 999                     raise e
1000         self.db.set_one(
1001             DB_DATA.api_endpoints.table,
1002             DB_DATA.api_endpoints.filter,
1003             {DB_DATA.api_endpoints.key: endpoints},
1004         )
1005
1006     def handle_exception(self, loop, context):
1007         # All unhandled exceptions by libjuju are handled here.
1008         pass
1009
1010     async def health_check(self, interval: float = 300.0):
1011         """
1012         Health check to make sure controller and controller_model connections are OK
1013
1014         :param: interval: Time in seconds between checks
1015         """
1016         controller = None
1017         while True:
1018             try:
1019                 controller = await self.get_controller()
1020                 # self.log.debug("VCA is alive")
1021             except Exception as e:
1022                 self.log.error("Health check to VCA failed: {}".format(e))
1023             finally:
1024                 await self.disconnect_controller(controller)
1025             await asyncio.sleep(interval)
1026
1027     async def list_models(self, contains: str = None) -> [str]:
1028         """List models with certain names
1029
1030         :param: contains:   String that is contained in model name
1031
1032         :retur: [models] Returns list of model names
1033         """
1034
1035         controller = await self.get_controller()
1036         try:
1037             models = await controller.list_models()
1038             if contains:
1039                 models = [model for model in models if contains in model]
1040             return models
1041         finally:
1042             await self.disconnect_controller(controller)
1043
1044     async def list_offers(self, model_name: str) -> QueryApplicationOffersResults:
1045         """List models with certain names
1046
1047         :param: model_name: Model name
1048
1049         :return:            Returns list of offers
1050         """
1051
1052         controller = await self.get_controller()
1053         try:
1054             return await controller.list_offers(model_name)
1055         finally:
1056             await self.disconnect_controller(controller)
1057
1058     async def add_k8s(
1059         self,
1060         name: str,
1061         rbac_id: str,
1062         token: str,
1063         client_cert_data: str,
1064         configuration: Configuration,
1065         storage_class: str,
1066         credential_name: str = None,
1067     ):
1068         """
1069         Add a Kubernetes cloud to the controller
1070
1071         Similar to the `juju add-k8s` command in the CLI
1072
1073         :param: name:               Name for the K8s cloud
1074         :param: configuration:      Kubernetes configuration object
1075         :param: storage_class:      Storage Class to use in the cloud
1076         :param: credential_name:    Storage Class to use in the cloud
1077         """
1078
1079         if not storage_class:
1080             raise Exception("storage_class must be a non-empty string")
1081         if not name:
1082             raise Exception("name must be a non-empty string")
1083         if not configuration:
1084             raise Exception("configuration must be provided")
1085
1086         endpoint = configuration.host
1087         credential = self.get_k8s_cloud_credential(
1088             configuration,
1089             client_cert_data,
1090             token,
1091         )
1092         credential.attrs[RBAC_LABEL_KEY_NAME] = rbac_id
1093         cloud = client.Cloud(
1094             type_="kubernetes",
1095             auth_types=[credential.auth_type],
1096             endpoint=endpoint,
1097             ca_certificates=[client_cert_data],
1098             config={
1099                 "operator-storage": storage_class,
1100                 "workload-storage": storage_class,
1101             },
1102         )
1103
1104         return await self.add_cloud(
1105             name, cloud, credential, credential_name=credential_name
1106         )
1107
1108     def get_k8s_cloud_credential(
1109         self,
1110         configuration: Configuration,
1111         client_cert_data: str,
1112         token: str = None,
1113     ) -> client.CloudCredential:
1114         attrs = {}
1115         # TODO: Test with AKS
1116         key = None  # open(configuration.key_file, "r").read()
1117         username = configuration.username
1118         password = configuration.password
1119
1120         if client_cert_data:
1121             attrs["ClientCertificateData"] = client_cert_data
1122         if key:
1123             attrs["ClientKeyData"] = key
1124         if token:
1125             if username or password:
1126                 raise JujuInvalidK8sConfiguration("Cannot set both token and user/pass")
1127             attrs["Token"] = token
1128
1129         auth_type = None
1130         if key:
1131             auth_type = "oauth2"
1132             if client_cert_data:
1133                 auth_type = "oauth2withcert"
1134             if not token:
1135                 raise JujuInvalidK8sConfiguration(
1136                     "missing token for auth type {}".format(auth_type)
1137                 )
1138         elif username:
1139             if not password:
1140                 self.log.debug(
1141                     "credential for user {} has empty password".format(username)
1142                 )
1143             attrs["username"] = username
1144             attrs["password"] = password
1145             if client_cert_data:
1146                 auth_type = "userpasswithcert"
1147             else:
1148                 auth_type = "userpass"
1149         elif client_cert_data and token:
1150             auth_type = "certificate"
1151         else:
1152             raise JujuInvalidK8sConfiguration("authentication method not supported")
1153         return client.CloudCredential(auth_type=auth_type, attrs=attrs)
1154
1155     async def add_cloud(
1156         self,
1157         name: str,
1158         cloud: Cloud,
1159         credential: CloudCredential = None,
1160         credential_name: str = None,
1161     ) -> Cloud:
1162         """
1163         Add cloud to the controller
1164
1165         :param: name:               Name of the cloud to be added
1166         :param: cloud:              Cloud object
1167         :param: credential:         CloudCredentials object for the cloud
1168         :param: credential_name:    Credential name.
1169                                     If not defined, cloud of the name will be used.
1170         """
1171         controller = await self.get_controller()
1172         try:
1173             _ = await controller.add_cloud(name, cloud)
1174             if credential:
1175                 await controller.add_credential(
1176                     credential_name or name, credential=credential, cloud=name
1177                 )
1178             # Need to return the object returned by the controller.add_cloud() function
1179             # I'm returning the original value now until this bug is fixed:
1180             #   https://github.com/juju/python-libjuju/issues/443
1181             return cloud
1182         finally:
1183             await self.disconnect_controller(controller)
1184
1185     async def remove_cloud(self, name: str):
1186         """
1187         Remove cloud
1188
1189         :param: name:   Name of the cloud to be removed
1190         """
1191         controller = await self.get_controller()
1192         try:
1193             await controller.remove_cloud(name)
1194         finally:
1195             await self.disconnect_controller(controller)
1196
1197     async def _get_leader_unit(self, application: Application) -> Unit:
1198         unit = None
1199         for u in application.units:
1200             if await u.is_leader_from_status():
1201                 unit = u
1202                 break
1203         return unit
1204
1205     async def get_cloud_credentials(self, cloud_name: str, credential_name: str):
1206         controller = await self.get_controller()
1207         try:
1208             facade = client.CloudFacade.from_connection(controller.connection())
1209             cloud_cred_tag = tag.credential(cloud_name, self.username, credential_name)
1210             params = [client.Entity(cloud_cred_tag)]
1211             return (await facade.Credential(params)).results
1212         finally:
1213             await self.disconnect_controller(controller)