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                     series=params.series,
 459                 )
 460             )
 461
 462             machine = None
 463             for _ in range(10):
 464                 machine_list = await model.get_machines()
 465                 if machine_id in machine_list:
 466                     self.log.debug("Machine {} found in model!".format(machine_id))
 467                     machine = model.machines.get(machine_id)
 468                     break
 469                 await asyncio.sleep(2)
 470
 471             if machine is None:
 472                 msg = "Machine {} not found in model".format(machine_id)
 473                 self.log.error(msg=msg)
 474                 raise JujuMachineNotFound(msg)
 475
 476             self.log.debug(
 477                 "Wait until machine {} is ready in model {}".format(
 478                     machine.entity_id, model_name
 479                 )
 480             )
 481             await JujuModelWatcher.wait_for(
 482                 model=model,
 483                 entity=machine,
 484                 progress_timeout=progress_timeout,
 485                 total_timeout=total_timeout,
 486                 db_dict=db_dict,
 487                 n2vc=self.n2vc,
 488             )
 489         except Exception as e:
 490             raise e
 491         finally:
 492             await self.disconnect_model(model)
 493             await self.disconnect_controller(controller)
 494
 495         self.log.debug(
 496             "Machine provisioned {} in model {}".format(machine_id, model_name)
 497         )
 498
 499         return machine_id
 500
 501     async def deploy(
 502         self, uri: str, model_name: str, wait: bool = True, timeout: float = 3600
 503     ):
 504         """
 505         Deploy bundle or charm: Similar to the juju CLI command `juju deploy`
 506
 507         :param: uri:            Path or Charm Store uri in which the charm or bundle can be found
 508         :param: model_name:     Model name
 509         :param: wait:           Indicates whether to wait or not until all applications are active
 510         :param: timeout:        Time in seconds to wait until all applications are active
 511         """
 512         controller = await self.get_controller()
 513         model = await self.get_model(controller, model_name)
 514         try:
 515             await model.deploy(uri)
 516             if wait:
 517                 await JujuModelWatcher.wait_for_model(model, timeout=timeout)
 518                 self.log.debug("All units active in model {}".format(model_name))
 519         finally:
 520             await self.disconnect_model(model)
 521             await self.disconnect_controller(controller)
 522
 523     async def deploy_charm(
 524         self,
 525         application_name: str,
 526         path: str,
 527         model_name: str,
 528         machine_id: str,
 529         db_dict: dict = None,
 530         progress_timeout: float = None,
 531         total_timeout: float = None,
 532         config: dict = None,
 533         series: str = None,
 534         num_units: int = 1,
 535     ):
 536         """Deploy charm
 537
 538         :param: application_name:   Application name
 539         :param: path:               Local path to the charm
 540         :param: model_name:         Model name
 541         :param: machine_id          ID of the machine
 542         :param: db_dict:            Dictionary with data of the DB to write the updates
 543         :param: progress_timeout:   Maximum time between two updates in the model
 544         :param: total_timeout:      Timeout for the entity to be active
 545         :param: config:             Config for the charm
 546         :param: series:             Series of the charm
 547         :param: num_units:          Number of units
 548
 549         :return: (juju.application.Application): Juju application
 550         """
 551         self.log.debug(
 552             "Deploying charm {} to machine {} in model ~{}".format(
 553                 application_name, machine_id, model_name
 554             )
 555         )
 556         self.log.debug("charm: {}".format(path))
 557
 558         # Get controller
 559         controller = await self.get_controller()
 560
 561         # Get model
 562         model = await self.get_model(controller, model_name)
 563
 564         try:
 565             application = None
 566             if application_name not in model.applications:
 567
 568                 if machine_id is not None:
 569                     if machine_id not in model.machines:
 570                         msg = "Machine {} not found in model".format(machine_id)
 571                         self.log.error(msg=msg)
 572                         raise JujuMachineNotFound(msg)
 573                     machine = model.machines[machine_id]
 574                     series = machine.series
 575
 576                 application = await model.deploy(
 577                     entity_url=path,
 578                     application_name=application_name,
 579                     channel="stable",
 580                     num_units=1,
 581                     series=series,
 582                     to=machine_id,
 583                     config=config,
 584                 )
 585
 586                 self.log.debug(
 587                     "Wait until application {} is ready in model {}".format(
 588                         application_name, model_name
 589                     )
 590                 )
 591                 if num_units > 1:
 592                     for _ in range(num_units - 1):
 593                         m, _ = await self.create_machine(model_name, wait=False)
 594                         await application.add_unit(to=m.entity_id)
 595
 596                 await JujuModelWatcher.wait_for(
 597                     model=model,
 598                     entity=application,
 599                     progress_timeout=progress_timeout,
 600                     total_timeout=total_timeout,
 601                     db_dict=db_dict,
 602                     n2vc=self.n2vc,
 603                 )
 604                 self.log.debug(
 605                     "Application {} is ready in model {}".format(
 606                         application_name, model_name
 607                     )
 608                 )
 609             else:
 610                 raise JujuApplicationExists(
 611                     "Application {} exists".format(application_name)
 612                 )
 613         finally:
 614             await self.disconnect_model(model)
 615             await self.disconnect_controller(controller)
 616
 617         return application
 618
 619     def _get_application(self, model: Model, application_name: str) -> Application:
 620         """Get application
 621
 622         :param: model:              Model object
 623         :param: application_name:   Application name
 624
 625         :return: juju.application.Application (or None if it doesn't exist)
 626         """
 627         if model.applications and application_name in model.applications:
 628             return model.applications[application_name]
 629
 630     async def execute_action(
 631         self,
 632         application_name: str,
 633         model_name: str,
 634         action_name: str,
 635         db_dict: dict = None,
 636         progress_timeout: float = None,
 637         total_timeout: float = None,
 638         **kwargs
 639     ):
 640         """Execute action
 641
 642         :param: application_name:   Application name
 643         :param: model_name:         Model name
 644         :param: action_name:        Name of the action
 645         :param: db_dict:            Dictionary with data of the DB to write the updates
 646         :param: progress_timeout:   Maximum time between two updates in the model
 647         :param: total_timeout:      Timeout for the entity to be active
 648
 649         :return: (str, str): (output and status)
 650         """
 651         self.log.debug(
 652             "Executing action {} using params {}".format(action_name, kwargs)
 653         )
 654         # Get controller
 655         controller = await self.get_controller()
 656
 657         # Get model
 658         model = await self.get_model(controller, model_name)
 659
 660         try:
 661             # Get application
 662             application = self._get_application(
 663                 model,
 664                 application_name=application_name,
 665             )
 666             if application is None:
 667                 raise JujuApplicationNotFound("Cannot execute action")
 668
 669             # Get leader unit
 670             # Racing condition:
 671             #   Ocassionally, self._get_leader_unit() will return None
 672             #   because the leader elected hook has not been triggered yet.
 673             #   Therefore, we are doing some retries. If it happens again,
 674             #   re-open bug 1236
 675             attempts = 3
 676             time_between_retries = 10
 677             unit = None
 678             for _ in range(attempts):
 679                 unit = await self._get_leader_unit(application)
 680                 if unit is None:
 681                     await asyncio.sleep(time_between_retries)
 682                 else:
 683                     break
 684             if unit is None:
 685                 raise JujuLeaderUnitNotFound(
 686                     "Cannot execute action: leader unit not found"
 687                 )
 688
 689             actions = await application.get_actions()
 690
 691             if action_name not in actions:
 692                 raise JujuActionNotFound(
 693                     "Action {} not in available actions".format(action_name)
 694                 )
 695
 696             action = await unit.run_action(action_name, **kwargs)
 697
 698             self.log.debug(
 699                 "Wait until action {} is completed in application {} (model={})".format(
 700                     action_name, application_name, model_name
 701                 )
 702             )
 703             await JujuModelWatcher.wait_for(
 704                 model=model,
 705                 entity=action,
 706                 progress_timeout=progress_timeout,
 707                 total_timeout=total_timeout,
 708                 db_dict=db_dict,
 709                 n2vc=self.n2vc,
 710             )
 711
 712             output = await model.get_action_output(action_uuid=action.entity_id)
 713             status = await model.get_action_status(uuid_or_prefix=action.entity_id)
 714             status = (
 715                 status[action.entity_id] if action.entity_id in status else "failed"
 716             )
 717
 718             self.log.debug(
 719                 "Action {} completed with status {} in application {} (model={})".format(
 720                     action_name, action.status, application_name, model_name
 721                 )
 722             )
 723         finally:
 724             await self.disconnect_model(model)
 725             await self.disconnect_controller(controller)
 726
 727         return output, status
 728
 729     async def get_actions(self, application_name: str, model_name: str) -> dict:
 730         """Get list of actions
 731
 732         :param: application_name: Application name
 733         :param: model_name: Model name
 734
 735         :return: Dict with this format
 736             {
 737                 "action_name": "Description of the action",
 738                 ...
 739             }
 740         """
 741         self.log.debug(
 742             "Getting list of actions for application {}".format(application_name)
 743         )
 744
 745         # Get controller
 746         controller = await self.get_controller()
 747
 748         # Get model
 749         model = await self.get_model(controller, model_name)
 750
 751         try:
 752             # Get application
 753             application = self._get_application(
 754                 model,
 755                 application_name=application_name,
 756             )
 757
 758             # Return list of actions
 759             return await application.get_actions()
 760
 761         finally:
 762             # Disconnect from model and controller
 763             await self.disconnect_model(model)
 764             await self.disconnect_controller(controller)
 765
 766     async def get_metrics(self, model_name: str, application_name: str) -> dict:
 767         """Get the metrics collected by the VCA.
 768
 769         :param model_name The name or unique id of the network service
 770         :param application_name The name of the application
 771         """
 772         if not model_name or not application_name:
 773             raise Exception("model_name and application_name must be non-empty strings")
 774         metrics = {}
 775         controller = await self.get_controller()
 776         model = await self.get_model(controller, model_name)
 777         try:
 778             application = self._get_application(model, application_name)
 779             if application is not None:
 780                 metrics = await application.get_metrics()
 781         finally:
 782             self.disconnect_model(model)
 783             self.disconnect_controller(controller)
 784         return metrics
 785
 786     async def add_relation(
 787         self,
 788         model_name: str,
 789         endpoint_1: str,
 790         endpoint_2: str,
 791     ):
 792         """Add relation
 793
 794         :param: model_name:     Model name
 795         :param: endpoint_1      First endpoint name
 796                                 ("app:endpoint" format or directly the saas name)
 797         :param: endpoint_2:     Second endpoint name (^ same format)
 798         """
 799
 800         self.log.debug("Adding relation: {} -> {}".format(endpoint_1, endpoint_2))
 801
 802         # Get controller
 803         controller = await self.get_controller()
 804
 805         # Get model
 806         model = await self.get_model(controller, model_name)
 807
 808         # Add relation
 809         try:
 810             await model.add_relation(endpoint_1, endpoint_2)
 811         except JujuAPIError as e:
 812             if "not found" in e.message:
 813                 self.log.warning("Relation not found: {}".format(e.message))
 814                 return
 815             if "already exists" in e.message:
 816                 self.log.warning("Relation already exists: {}".format(e.message))
 817                 return
 818             # another exception, raise it
 819             raise e
 820         finally:
 821             await self.disconnect_model(model)
 822             await self.disconnect_controller(controller)
 823
 824     async def consume(
 825         self,
 826         offer_url: str,
 827         model_name: str,
 828     ):
 829         """
 830         Adds a remote offer to the model. Relations can be created later using "juju relate".
 831
 832         :param: offer_url:      Offer Url
 833         :param: model_name:     Model name
 834
 835         :raises ParseError if there's a problem parsing the offer_url
 836         :raises JujuError if remote offer includes and endpoint
 837         :raises JujuAPIError if the operation is not successful
 838         """
 839         controller = await self.get_controller()
 840         model = await controller.get_model(model_name)
 841
 842         try:
 843             await model.consume(offer_url)
 844         finally:
 845             await self.disconnect_model(model)
 846             await self.disconnect_controller(controller)
 847
 848     async def destroy_model(self, model_name: str, total_timeout: float):
 849         """
 850         Destroy model
 851
 852         :param: model_name:     Model name
 853         :param: total_timeout:  Timeout
 854         """
 855
 856         controller = await self.get_controller()
 857         model = await self.get_model(controller, model_name)
 858         try:
 859             self.log.debug("Destroying model {}".format(model_name))
 860             uuid = model.info.uuid
 861
 862             # Destroy machines that are manually provisioned
 863             # and still are in pending state
 864             await self._destroy_pending_machines(model, only_manual=True)
 865
 866             # Disconnect model
 867             await self.disconnect_model(model)
 868
 869             # Destroy model
 870             if model_name in self.models:
 871                 self.models.remove(model_name)
 872
 873             await controller.destroy_model(uuid, force=True, max_wait=0)
 874
 875             # Wait until model is destroyed
 876             self.log.debug("Waiting for model {} to be destroyed...".format(model_name))
 877
 878             if total_timeout is None:
 879                 total_timeout = 3600
 880             end = time.time() + total_timeout
 881             while time.time() < end:
 882                 models = await controller.list_models()
 883                 if model_name not in models:
 884                     self.log.debug(
 885                         "The model {} ({}) was destroyed".format(model_name, uuid)
 886                     )
 887                     return
 888                 await asyncio.sleep(5)
 889             raise Exception(
 890                 "Timeout waiting for model {} to be destroyed".format(model_name)
 891             )
 892         finally:
 893             await self.disconnect_controller(controller)
 894
 895     async def destroy_application(self, model: Model, application_name: str):
 896         """
 897         Destroy application
 898
 899         :param: model:              Model object
 900         :param: application_name:   Application name
 901         """
 902         self.log.debug(
 903             "Destroying application {} in model {}".format(
 904                 application_name, model.info.name
 905             )
 906         )
 907         application = model.applications.get(application_name)
 908         if application:
 909             await application.destroy()
 910         else:
 911             self.log.warning("Application not found: {}".format(application_name))
 912
 913     async def _destroy_pending_machines(self, model: Model, only_manual: bool = False):
 914         """
 915         Destroy pending machines in a given model
 916
 917         :param: only_manual:    Bool that indicates only manually provisioned
 918                                 machines should be destroyed (if True), or that
 919                                 all pending machines should be destroyed
 920         """
 921         status = await model.get_status()
 922         for machine_id in status.machines:
 923             machine_status = status.machines[machine_id]
 924             if machine_status.agent_status.status == "pending":
 925                 if only_manual and not machine_status.instance_id.startswith("manual:"):
 926                     break
 927                 machine = model.machines[machine_id]
 928                 await machine.destroy(force=True)
 929
 930     async def configure_application(
 931         self, model_name: str, application_name: str, config: dict = None
 932     ):
 933         """Configure application
 934
 935         :param: model_name:         Model name
 936         :param: application_name:   Application name
 937         :param: config:             Config to apply to the charm
 938         """
 939         self.log.debug("Configuring application {}".format(application_name))
 940
 941         if config:
 942             controller = await self.get_controller()
 943             model = None
 944             try:
 945                 model = await self.get_model(controller, model_name)
 946                 application = self._get_application(
 947                     model,
 948                     application_name=application_name,
 949                 )
 950                 await application.set_config(config)
 951             finally:
 952                 if model:
 953                     await self.disconnect_model(model)
 954                 await self.disconnect_controller(controller)
 955
 956     def _get_api_endpoints_db(self) -> [str]:
 957         """
 958         Get API Endpoints from DB
 959
 960         :return: List of API endpoints
 961         """
 962         self.log.debug("Getting endpoints from database")
 963
 964         juju_info = self.db.get_one(
 965             DB_DATA.api_endpoints.table,
 966             q_filter=DB_DATA.api_endpoints.filter,
 967             fail_on_empty=False,
 968         )
 969         if juju_info and DB_DATA.api_endpoints.key in juju_info:
 970             return juju_info[DB_DATA.api_endpoints.key]
 971
 972     def _update_api_endpoints_db(self, endpoints: [str]):
 973         """
 974         Update API endpoints in Database
 975
 976         :param: List of endpoints
 977         """
 978         self.log.debug("Saving endpoints {} in database".format(endpoints))
 979
 980         juju_info = self.db.get_one(
 981             DB_DATA.api_endpoints.table,
 982             q_filter=DB_DATA.api_endpoints.filter,
 983             fail_on_empty=False,
 984         )
 985         # If it doesn't, then create it
 986         if not juju_info:
 987             try:
 988                 self.db.create(
 989                     DB_DATA.api_endpoints.table,
 990                     DB_DATA.api_endpoints.filter,
 991                 )
 992             except DbException as e:
 993                 # Racing condition: check if another N2VC worker has created it
 994                 juju_info = self.db.get_one(
 995                     DB_DATA.api_endpoints.table,
 996                     q_filter=DB_DATA.api_endpoints.filter,
 997                     fail_on_empty=False,
 998                 )
 999                 if not juju_info:
1000                     raise e
1001         self.db.set_one(
1002             DB_DATA.api_endpoints.table,
1003             DB_DATA.api_endpoints.filter,
1004             {DB_DATA.api_endpoints.key: endpoints},
1005         )
1006
1007     def handle_exception(self, loop, context):
1008         # All unhandled exceptions by libjuju are handled here.
1009         pass
1010
1011     async def health_check(self, interval: float = 300.0):
1012         """
1013         Health check to make sure controller and controller_model connections are OK
1014
1015         :param: interval: Time in seconds between checks
1016         """
1017         controller = None
1018         while True:
1019             try:
1020                 controller = await self.get_controller()
1021                 # self.log.debug("VCA is alive")
1022             except Exception as e:
1023                 self.log.error("Health check to VCA failed: {}".format(e))
1024             finally:
1025                 await self.disconnect_controller(controller)
1026             await asyncio.sleep(interval)
1027
1028     async def list_models(self, contains: str = None) -> [str]:
1029         """List models with certain names
1030
1031         :param: contains:   String that is contained in model name
1032
1033         :retur: [models] Returns list of model names
1034         """
1035
1036         controller = await self.get_controller()
1037         try:
1038             models = await controller.list_models()
1039             if contains:
1040                 models = [model for model in models if contains in model]
1041             return models
1042         finally:
1043             await self.disconnect_controller(controller)
1044
1045     async def list_offers(self, model_name: str) -> QueryApplicationOffersResults:
1046         """List models with certain names
1047
1048         :param: model_name: Model name
1049
1050         :return:            Returns list of offers
1051         """
1052
1053         controller = await self.get_controller()
1054         try:
1055             return await controller.list_offers(model_name)
1056         finally:
1057             await self.disconnect_controller(controller)
1058
1059     async def add_k8s(
1060         self,
1061         name: str,
1062         rbac_id: str,
1063         token: str,
1064         client_cert_data: str,
1065         configuration: Configuration,
1066         storage_class: str,
1067         credential_name: str = None,
1068     ):
1069         """
1070         Add a Kubernetes cloud to the controller
1071
1072         Similar to the `juju add-k8s` command in the CLI
1073
1074         :param: name:               Name for the K8s cloud
1075         :param: configuration:      Kubernetes configuration object
1076         :param: storage_class:      Storage Class to use in the cloud
1077         :param: credential_name:    Storage Class to use in the cloud
1078         """
1079
1080         if not storage_class:
1081             raise Exception("storage_class must be a non-empty string")
1082         if not name:
1083             raise Exception("name must be a non-empty string")
1084         if not configuration:
1085             raise Exception("configuration must be provided")
1086
1087         endpoint = configuration.host
1088         credential = self.get_k8s_cloud_credential(
1089             configuration,
1090             client_cert_data,
1091             token,
1092         )
1093         credential.attrs[RBAC_LABEL_KEY_NAME] = rbac_id
1094         cloud = client.Cloud(
1095             type_="kubernetes",
1096             auth_types=[credential.auth_type],
1097             endpoint=endpoint,
1098             ca_certificates=[client_cert_data],
1099             config={
1100                 "operator-storage": storage_class,
1101                 "workload-storage": storage_class,
1102             },
1103         )
1104
1105         return await self.add_cloud(
1106             name, cloud, credential, credential_name=credential_name
1107         )
1108
1109     def get_k8s_cloud_credential(
1110         self,
1111         configuration: Configuration,
1112         client_cert_data: str,
1113         token: str = None,
1114     ) -> client.CloudCredential:
1115         attrs = {}
1116         # TODO: Test with AKS
1117         key = None  # open(configuration.key_file, "r").read()
1118         username = configuration.username
1119         password = configuration.password
1120
1121         if client_cert_data:
1122             attrs["ClientCertificateData"] = client_cert_data
1123         if key:
1124             attrs["ClientKeyData"] = key
1125         if token:
1126             if username or password:
1127                 raise JujuInvalidK8sConfiguration("Cannot set both token and user/pass")
1128             attrs["Token"] = token
1129
1130         auth_type = None
1131         if key:
1132             auth_type = "oauth2"
1133             if client_cert_data:
1134                 auth_type = "oauth2withcert"
1135             if not token:
1136                 raise JujuInvalidK8sConfiguration(
1137                     "missing token for auth type {}".format(auth_type)
1138                 )
1139         elif username:
1140             if not password:
1141                 self.log.debug(
1142                     "credential for user {} has empty password".format(username)
1143                 )
1144             attrs["username"] = username
1145             attrs["password"] = password
1146             if client_cert_data:
1147                 auth_type = "userpasswithcert"
1148             else:
1149                 auth_type = "userpass"
1150         elif client_cert_data and token:
1151             auth_type = "certificate"
1152         else:
1153             raise JujuInvalidK8sConfiguration("authentication method not supported")
1154         return client.CloudCredential(auth_type=auth_type, attrs=attrs)
1155
1156     async def add_cloud(
1157         self,
1158         name: str,
1159         cloud: Cloud,
1160         credential: CloudCredential = None,
1161         credential_name: str = None,
1162     ) -> Cloud:
1163         """
1164         Add cloud to the controller
1165
1166         :param: name:               Name of the cloud to be added
1167         :param: cloud:              Cloud object
1168         :param: credential:         CloudCredentials object for the cloud
1169         :param: credential_name:    Credential name.
1170                                     If not defined, cloud of the name will be used.
1171         """
1172         controller = await self.get_controller()
1173         try:
1174             _ = await controller.add_cloud(name, cloud)
1175             if credential:
1176                 await controller.add_credential(
1177                     credential_name or name, credential=credential, cloud=name
1178                 )
1179             # Need to return the object returned by the controller.add_cloud() function
1180             # I'm returning the original value now until this bug is fixed:
1181             #   https://github.com/juju/python-libjuju/issues/443
1182             return cloud
1183         finally:
1184             await self.disconnect_controller(controller)
1185
1186     async def remove_cloud(self, name: str):
1187         """
1188         Remove cloud
1189
1190         :param: name:   Name of the cloud to be removed
1191         """
1192         controller = await self.get_controller()
1193         try:
1194             await controller.remove_cloud(name)
1195         finally:
1196             await self.disconnect_controller(controller)
1197
1198     async def _get_leader_unit(self, application: Application) -> Unit:
1199         unit = None
1200         for u in application.units:
1201             if await u.is_leader_from_status():
1202                 unit = u
1203                 break
1204         return unit
1205
1206     async def get_cloud_credentials(self, cloud_name: str, credential_name: str):
1207         controller = await self.get_controller()
1208         try:
1209             facade = client.CloudFacade.from_connection(controller.connection())
1210             cloud_cred_tag = tag.credential(cloud_name, self.username, credential_name)
1211             params = [client.Entity(cloud_cred_tag)]
1212             return (await facade.Credential(params)).results
1213         finally:
1214             await self.disconnect_controller(controller)