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 import typing
  18
  19 import time
  20
  21 import juju.errors
  22 from juju.model import Model
  23 from juju.machine import Machine
  24 from juju.application import Application
  25 from juju.unit import Unit
  26 from juju.client._definitions import (
  27     FullStatus,
  28     QueryApplicationOffersResults,
  29     Cloud,
  30     CloudCredential,
  31 )
  32 from juju.controller import Controller
  33 from juju.client import client
  34 from juju import tag
  35
  36 from n2vc.juju_watcher import JujuModelWatcher
  37 from n2vc.provisioner import AsyncSSHProvisioner
  38 from n2vc.n2vc_conn import N2VCConnector
  39 from n2vc.exceptions import (
  40     JujuMachineNotFound,
  41     JujuApplicationNotFound,
  42     JujuLeaderUnitNotFound,
  43     JujuActionNotFound,
  44     JujuControllerFailedConnecting,
  45     JujuApplicationExists,
  46     JujuInvalidK8sConfiguration,
  47     JujuError,
  48 )
  49 from n2vc.vca.cloud import Cloud as VcaCloud
  50 from n2vc.vca.connection import Connection
  51 from kubernetes.client.configuration import Configuration
  52 from retrying_async import retry
  53
  54
  55 RBAC_LABEL_KEY_NAME = "rbac-id"
  56
  57
  58 class Libjuju:
  59     def __init__(
  60         self,
  61         vca_connection: Connection,
  62         loop: asyncio.AbstractEventLoop = None,
  63         log: logging.Logger = None,
  64         n2vc: N2VCConnector = None,
  65     ):
  66         """
  67         Constructor
  68
  69         :param: vca_connection:         n2vc.vca.connection object
  70         :param: loop:                   Asyncio loop
  71         :param: log:                    Logger
  72         :param: n2vc:                   N2VC object
  73         """
  74
  75         self.log = log or logging.getLogger("Libjuju")
  76         self.n2vc = n2vc
  77         self.vca_connection = vca_connection
  78
  79         self.loop = loop or asyncio.get_event_loop()
  80         self.loop.set_exception_handler(self.handle_exception)
  81         self.creating_model = asyncio.Lock(loop=self.loop)
  82
  83         if self.vca_connection.is_default:
  84             self.health_check_task = self._create_health_check_task()
  85
  86     def _create_health_check_task(self):
  87         return self.loop.create_task(self.health_check())
  88
  89     async def get_controller(self, timeout: float = 60.0) -> Controller:
  90         """
  91         Get controller
  92
  93         :param: timeout: Time in seconds to wait for controller to connect
  94         """
  95         controller = None
  96         try:
  97             controller = Controller()
  98             await asyncio.wait_for(
  99                 controller.connect(
 100                     endpoint=self.vca_connection.data.endpoints,
 101                     username=self.vca_connection.data.user,
 102                     password=self.vca_connection.data.secret,
 103                     cacert=self.vca_connection.data.cacert,
 104                 ),
 105                 timeout=timeout,
 106             )
 107             if self.vca_connection.is_default:
 108                 endpoints = await controller.api_endpoints
 109                 if not all(
 110                     endpoint in self.vca_connection.endpoints for endpoint in endpoints
 111                 ):
 112                     await self.vca_connection.update_endpoints(endpoints)
 113             return controller
 114         except asyncio.CancelledError as e:
 115             raise e
 116         except Exception as e:
 117             self.log.error(
 118                 "Failed connecting to controller: {}... {}".format(
 119                     self.vca_connection.data.endpoints, e
 120                 )
 121             )
 122             if controller:
 123                 await self.disconnect_controller(controller)
 124
 125             raise JujuControllerFailedConnecting(
 126                 f"Error connecting to Juju controller: {e}"
 127             )
 128
 129     async def disconnect(self):
 130         """Disconnect"""
 131         # Cancel health check task
 132         self.health_check_task.cancel()
 133         self.log.debug("Libjuju disconnected!")
 134
 135     async def disconnect_model(self, model: Model):
 136         """
 137         Disconnect model
 138
 139         :param: model: Model that will be disconnected
 140         """
 141         await model.disconnect()
 142
 143     async def disconnect_controller(self, controller: Controller):
 144         """
 145         Disconnect controller
 146
 147         :param: controller: Controller that will be disconnected
 148         """
 149         if controller:
 150             await controller.disconnect()
 151
 152     @retry(attempts=3, delay=5, timeout=None)
 153     async def add_model(self, model_name: str, cloud: VcaCloud):
 154         """
 155         Create model
 156
 157         :param: model_name: Model name
 158         :param: cloud: Cloud object
 159         """
 160
 161         # Get controller
 162         controller = await self.get_controller()
 163         model = None
 164         try:
 165             # Block until other workers have finished model creation
 166             while self.creating_model.locked():
 167                 await asyncio.sleep(0.1)
 168
 169             # Create the model
 170             async with self.creating_model:
 171                 if await self.model_exists(model_name, controller=controller):
 172                     return
 173                 self.log.debug("Creating model {}".format(model_name))
 174                 model = await controller.add_model(
 175                     model_name,
 176                     config=self.vca_connection.data.model_config,
 177                     cloud_name=cloud.name,
 178                     credential_name=cloud.credential_name,
 179                 )
 180         except juju.errors.JujuAPIError as e:
 181             if "already exists" in e.message:
 182                 pass
 183             else:
 184                 raise e
 185         finally:
 186             if model:
 187                 await self.disconnect_model(model)
 188             await self.disconnect_controller(controller)
 189
 190     async def get_executed_actions(self, model_name: str) -> list:
 191         """
 192         Get executed/history of actions for a model.
 193
 194         :param: model_name: Model name, str.
 195         :return: List of executed actions for a model.
 196         """
 197         model = None
 198         executed_actions = []
 199         controller = await self.get_controller()
 200         try:
 201             model = await self.get_model(controller, model_name)
 202             # Get all unique action names
 203             actions = {}
 204             for application in model.applications:
 205                 application_actions = await self.get_actions(application, model_name)
 206                 actions.update(application_actions)
 207             # Get status of all actions
 208             for application_action in actions:
 209                 app_action_status_list = await model.get_action_status(
 210                     name=application_action
 211                 )
 212                 for action_id, action_status in app_action_status_list.items():
 213                     executed_action = {
 214                         "id": action_id,
 215                         "action": application_action,
 216                         "status": action_status,
 217                     }
 218                     # Get action output by id
 219                     action_status = await model.get_action_output(executed_action["id"])
 220                     for k, v in action_status.items():
 221                         executed_action[k] = v
 222                     executed_actions.append(executed_action)
 223         except Exception as e:
 224             raise JujuError(
 225                 "Error in getting executed actions for model: {}. Error: {}".format(
 226                     model_name, str(e)
 227                 )
 228             )
 229         finally:
 230             if model:
 231                 await self.disconnect_model(model)
 232             await self.disconnect_controller(controller)
 233         return executed_actions
 234
 235     async def get_application_configs(
 236         self, model_name: str, application_name: str
 237     ) -> dict:
 238         """
 239         Get available configs for an application.
 240
 241         :param: model_name: Model name, str.
 242         :param: application_name: Application name, str.
 243
 244         :return: A dict which has key - action name, value - action description
 245         """
 246         model = None
 247         application_configs = {}
 248         controller = await self.get_controller()
 249         try:
 250             model = await self.get_model(controller, model_name)
 251             application = self._get_application(
 252                 model, application_name=application_name
 253             )
 254             application_configs = await application.get_config()
 255         except Exception as e:
 256             raise JujuError(
 257                 "Error in getting configs for application: {} in model: {}. Error: {}".format(
 258                     application_name, model_name, str(e)
 259                 )
 260             )
 261         finally:
 262             if model:
 263                 await self.disconnect_model(model)
 264             await self.disconnect_controller(controller)
 265         return application_configs
 266
 267     @retry(attempts=3, delay=5)
 268     async def get_model(self, controller: Controller, model_name: str) -> Model:
 269         """
 270         Get model from controller
 271
 272         :param: controller: Controller
 273         :param: model_name: Model name
 274
 275         :return: Model: The created Juju model object
 276         """
 277         return await controller.get_model(model_name)
 278
 279     async def model_exists(
 280         self, model_name: str, controller: Controller = None
 281     ) -> bool:
 282         """
 283         Check if model exists
 284
 285         :param: controller: Controller
 286         :param: model_name: Model name
 287
 288         :return bool
 289         """
 290         need_to_disconnect = False
 291
 292         # Get controller if not passed
 293         if not controller:
 294             controller = await self.get_controller()
 295             need_to_disconnect = True
 296
 297         # Check if model exists
 298         try:
 299             return model_name in await controller.list_models()
 300         finally:
 301             if need_to_disconnect:
 302                 await self.disconnect_controller(controller)
 303
 304     async def models_exist(self, model_names: [str]) -> (bool, list):
 305         """
 306         Check if models exists
 307
 308         :param: model_names: List of strings with model names
 309
 310         :return (bool, list[str]): (True if all models exists, List of model names that don't exist)
 311         """
 312         if not model_names:
 313             raise Exception(
 314                 "model_names must be a non-empty array. Given value: {}".format(
 315                     model_names
 316                 )
 317             )
 318         non_existing_models = []
 319         models = await self.list_models()
 320         existing_models = list(set(models).intersection(model_names))
 321         non_existing_models = list(set(model_names) - set(existing_models))
 322
 323         return (
 324             len(non_existing_models) == 0,
 325             non_existing_models,
 326         )
 327
 328     async def get_model_status(self, model_name: str) -> FullStatus:
 329         """
 330         Get model status
 331
 332         :param: model_name: Model name
 333
 334         :return: Full status object
 335         """
 336         controller = await self.get_controller()
 337         model = await self.get_model(controller, model_name)
 338         try:
 339             return await model.get_status()
 340         finally:
 341             await self.disconnect_model(model)
 342             await self.disconnect_controller(controller)
 343
 344     async def create_machine(
 345         self,
 346         model_name: str,
 347         machine_id: str = None,
 348         db_dict: dict = None,
 349         progress_timeout: float = None,
 350         total_timeout: float = None,
 351         series: str = "bionic",
 352         wait: bool = True,
 353     ) -> (Machine, bool):
 354         """
 355         Create machine
 356
 357         :param: model_name:         Model name
 358         :param: machine_id:         Machine id
 359         :param: db_dict:            Dictionary with data of the DB to write the updates
 360         :param: progress_timeout:   Maximum time between two updates in the model
 361         :param: total_timeout:      Timeout for the entity to be active
 362         :param: series:             Series of the machine (xenial, bionic, focal, ...)
 363         :param: wait:               Wait until machine is ready
 364
 365         :return: (juju.machine.Machine, bool):  Machine object and a boolean saying
 366                                                 if the machine is new or it already existed
 367         """
 368         new = False
 369         machine = None
 370
 371         self.log.debug(
 372             "Creating machine (id={}) in model: {}".format(machine_id, model_name)
 373         )
 374
 375         # Get controller
 376         controller = await self.get_controller()
 377
 378         # Get model
 379         model = await self.get_model(controller, model_name)
 380         try:
 381             if machine_id is not None:
 382                 self.log.debug(
 383                     "Searching machine (id={}) in model {}".format(
 384                         machine_id, model_name
 385                     )
 386                 )
 387
 388                 # Get machines from model and get the machine with machine_id if exists
 389                 machines = await model.get_machines()
 390                 if machine_id in machines:
 391                     self.log.debug(
 392                         "Machine (id={}) found in model {}".format(
 393                             machine_id, model_name
 394                         )
 395                     )
 396                     machine = machines[machine_id]
 397                 else:
 398                     raise JujuMachineNotFound("Machine {} not found".format(machine_id))
 399
 400             if machine is None:
 401                 self.log.debug("Creating a new machine in model {}".format(model_name))
 402
 403                 # Create machine
 404                 machine = await model.add_machine(
 405                     spec=None, constraints=None, disks=None, series=series
 406                 )
 407                 new = True
 408
 409                 # Wait until the machine is ready
 410                 self.log.debug(
 411                     "Wait until machine {} is ready in model {}".format(
 412                         machine.entity_id, model_name
 413                     )
 414                 )
 415                 if wait:
 416                     await JujuModelWatcher.wait_for(
 417                         model=model,
 418                         entity=machine,
 419                         progress_timeout=progress_timeout,
 420                         total_timeout=total_timeout,
 421                         db_dict=db_dict,
 422                         n2vc=self.n2vc,
 423                         vca_id=self.vca_connection._vca_id,
 424                     )
 425         finally:
 426             await self.disconnect_model(model)
 427             await self.disconnect_controller(controller)
 428
 429         self.log.debug(
 430             "Machine {} ready at {} in model {}".format(
 431                 machine.entity_id, machine.dns_name, model_name
 432             )
 433         )
 434         return machine, new
 435
 436     async def provision_machine(
 437         self,
 438         model_name: str,
 439         hostname: str,
 440         username: str,
 441         private_key_path: str,
 442         db_dict: dict = None,
 443         progress_timeout: float = None,
 444         total_timeout: float = None,
 445     ) -> str:
 446         """
 447         Manually provisioning of a machine
 448
 449         :param: model_name:         Model name
 450         :param: hostname:           IP to access the machine
 451         :param: username:           Username to login to the machine
 452         :param: private_key_path:   Local path for the private key
 453         :param: db_dict:            Dictionary with data of the DB to write the updates
 454         :param: progress_timeout:   Maximum time between two updates in the model
 455         :param: total_timeout:      Timeout for the entity to be active
 456
 457         :return: (Entity): Machine id
 458         """
 459         self.log.debug(
 460             "Provisioning machine. model: {}, hostname: {}, username: {}".format(
 461                 model_name, hostname, username
 462             )
 463         )
 464
 465         # Get controller
 466         controller = await self.get_controller()
 467
 468         # Get model
 469         model = await self.get_model(controller, model_name)
 470
 471         try:
 472             # Get provisioner
 473             provisioner = AsyncSSHProvisioner(
 474                 host=hostname,
 475                 user=username,
 476                 private_key_path=private_key_path,
 477                 log=self.log,
 478             )
 479
 480             # Provision machine
 481             params = await provisioner.provision_machine()
 482
 483             params.jobs = ["JobHostUnits"]
 484
 485             self.log.debug("Adding machine to model")
 486             connection = model.connection()
 487             client_facade = client.ClientFacade.from_connection(connection)
 488
 489             results = await client_facade.AddMachines(params=[params])
 490             error = results.machines[0].error
 491
 492             if error:
 493                 msg = "Error adding machine: {}".format(error.message)
 494                 self.log.error(msg=msg)
 495                 raise ValueError(msg)
 496
 497             machine_id = results.machines[0].machine
 498
 499             self.log.debug("Installing Juju agent into machine {}".format(machine_id))
 500             asyncio.ensure_future(
 501                 provisioner.install_agent(
 502                     connection=connection,
 503                     nonce=params.nonce,
 504                     machine_id=machine_id,
 505                     proxy=self.vca_connection.data.api_proxy,
 506                     series=params.series,
 507                 )
 508             )
 509
 510             machine = None
 511             for _ in range(10):
 512                 machine_list = await model.get_machines()
 513                 if machine_id in machine_list:
 514                     self.log.debug("Machine {} found in model!".format(machine_id))
 515                     machine = model.machines.get(machine_id)
 516                     break
 517                 await asyncio.sleep(2)
 518
 519             if machine is None:
 520                 msg = "Machine {} not found in model".format(machine_id)
 521                 self.log.error(msg=msg)
 522                 raise JujuMachineNotFound(msg)
 523
 524             self.log.debug(
 525                 "Wait until machine {} is ready in model {}".format(
 526                     machine.entity_id, model_name
 527                 )
 528             )
 529             await JujuModelWatcher.wait_for(
 530                 model=model,
 531                 entity=machine,
 532                 progress_timeout=progress_timeout,
 533                 total_timeout=total_timeout,
 534                 db_dict=db_dict,
 535                 n2vc=self.n2vc,
 536                 vca_id=self.vca_connection._vca_id,
 537             )
 538         except Exception as e:
 539             raise e
 540         finally:
 541             await self.disconnect_model(model)
 542             await self.disconnect_controller(controller)
 543
 544         self.log.debug(
 545             "Machine provisioned {} in model {}".format(machine_id, model_name)
 546         )
 547
 548         return machine_id
 549
 550     async def deploy(
 551         self, uri: str, model_name: str, wait: bool = True, timeout: float = 3600
 552     ):
 553         """
 554         Deploy bundle or charm: Similar to the juju CLI command `juju deploy`
 555
 556         :param: uri:            Path or Charm Store uri in which the charm or bundle can be found
 557         :param: model_name:     Model name
 558         :param: wait:           Indicates whether to wait or not until all applications are active
 559         :param: timeout:        Time in seconds to wait until all applications are active
 560         """
 561         controller = await self.get_controller()
 562         model = await self.get_model(controller, model_name)
 563         try:
 564             await model.deploy(uri, trust=True)
 565             if wait:
 566                 await JujuModelWatcher.wait_for_model(model, timeout=timeout)
 567                 self.log.debug("All units active in model {}".format(model_name))
 568         finally:
 569             await self.disconnect_model(model)
 570             await self.disconnect_controller(controller)
 571
 572     async def add_unit(
 573         self,
 574         application_name: str,
 575         model_name: str,
 576         machine_id: str,
 577         db_dict: dict = None,
 578         progress_timeout: float = None,
 579         total_timeout: float = None,
 580     ):
 581         """Add unit
 582
 583         :param: application_name:   Application name
 584         :param: model_name:         Model name
 585         :param: machine_id          Machine id
 586         :param: db_dict:            Dictionary with data of the DB to write the updates
 587         :param: progress_timeout:   Maximum time between two updates in the model
 588         :param: total_timeout:      Timeout for the entity to be active
 589
 590         :return: None
 591         """
 592
 593         model = None
 594         controller = await self.get_controller()
 595         try:
 596             model = await self.get_model(controller, model_name)
 597             application = self._get_application(model, application_name)
 598
 599             if application is not None:
 600
 601                 # Checks if the given machine id in the model,
 602                 # otherwise function raises an error
 603                 _machine, _series = self._get_machine_info(model, machine_id)
 604
 605                 self.log.debug(
 606                     "Adding unit (machine {}) to application {} in model ~{}".format(
 607                         machine_id, application_name, model_name
 608                     )
 609                 )
 610
 611                 await application.add_unit(to=machine_id)
 612
 613                 await JujuModelWatcher.wait_for(
 614                     model=model,
 615                     entity=application,
 616                     progress_timeout=progress_timeout,
 617                     total_timeout=total_timeout,
 618                     db_dict=db_dict,
 619                     n2vc=self.n2vc,
 620                     vca_id=self.vca_connection._vca_id,
 621                 )
 622                 self.log.debug(
 623                     "Unit is added to application {} in model {}".format(
 624                         application_name, model_name
 625                     )
 626                 )
 627             else:
 628                 raise JujuApplicationNotFound(
 629                     "Application {} not exists".format(application_name)
 630                 )
 631         finally:
 632             if model:
 633                 await self.disconnect_model(model)
 634             await self.disconnect_controller(controller)
 635
 636     async def destroy_unit(
 637         self,
 638         application_name: str,
 639         model_name: str,
 640         machine_id: str,
 641         total_timeout: float = None,
 642     ):
 643         """Destroy unit
 644
 645         :param: application_name:   Application name
 646         :param: model_name:         Model name
 647         :param: machine_id          Machine id
 648         :param: total_timeout:      Timeout for the entity to be active
 649
 650         :return: None
 651         """
 652
 653         model = None
 654         controller = await self.get_controller()
 655         try:
 656             model = await self.get_model(controller, model_name)
 657             application = self._get_application(model, application_name)
 658
 659             if application is None:
 660                 raise JujuApplicationNotFound(
 661                     "Application not found: {} (model={})".format(
 662                         application_name, model_name
 663                     )
 664                 )
 665
 666             unit = self._get_unit(application, machine_id)
 667             if not unit:
 668                 raise JujuError(
 669                     "A unit with machine id {} not in available units".format(
 670                         machine_id
 671                     )
 672                 )
 673
 674             unit_name = unit.name
 675
 676             self.log.debug(
 677                 "Destroying unit {} from application {} in model {}".format(
 678                     unit_name, application_name, model_name
 679                 )
 680             )
 681             await application.destroy_unit(unit_name)
 682
 683             self.log.debug(
 684                 "Waiting for unit {} to be destroyed in application {} (model={})...".format(
 685                     unit_name, application_name, model_name
 686                 )
 687             )
 688
 689             # TODO: Add functionality in the Juju watcher to replace this kind of blocks
 690             if total_timeout is None:
 691                 total_timeout = 3600
 692             end = time.time() + total_timeout
 693             while time.time() < end:
 694                 if not self._get_unit(application, machine_id):
 695                     self.log.debug(
 696                         "The unit {} was destroyed in application {} (model={}) ".format(
 697                             unit_name, application_name, model_name
 698                         )
 699                     )
 700                     return
 701                 await asyncio.sleep(5)
 702             self.log.debug(
 703                 "Unit {} is destroyed from application {} in model {}".format(
 704                     unit_name, application_name, model_name
 705                 )
 706             )
 707         finally:
 708             if model:
 709                 await self.disconnect_model(model)
 710             await self.disconnect_controller(controller)
 711
 712     async def deploy_charm(
 713         self,
 714         application_name: str,
 715         path: str,
 716         model_name: str,
 717         machine_id: str,
 718         db_dict: dict = None,
 719         progress_timeout: float = None,
 720         total_timeout: float = None,
 721         config: dict = None,
 722         series: str = None,
 723         num_units: int = 1,
 724     ):
 725         """Deploy charm
 726
 727         :param: application_name:   Application name
 728         :param: path:               Local path to the charm
 729         :param: model_name:         Model name
 730         :param: machine_id          ID of the machine
 731         :param: db_dict:            Dictionary with data of the DB to write the updates
 732         :param: progress_timeout:   Maximum time between two updates in the model
 733         :param: total_timeout:      Timeout for the entity to be active
 734         :param: config:             Config for the charm
 735         :param: series:             Series of the charm
 736         :param: num_units:          Number of units
 737
 738         :return: (juju.application.Application): Juju application
 739         """
 740         self.log.debug(
 741             "Deploying charm {} to machine {} in model ~{}".format(
 742                 application_name, machine_id, model_name
 743             )
 744         )
 745         self.log.debug("charm: {}".format(path))
 746
 747         # Get controller
 748         controller = await self.get_controller()
 749
 750         # Get model
 751         model = await self.get_model(controller, model_name)
 752
 753         try:
 754             if application_name not in model.applications:
 755
 756                 if machine_id is not None:
 757                     machine, series = self._get_machine_info(model, machine_id)
 758
 759                 application = await model.deploy(
 760                     entity_url=path,
 761                     application_name=application_name,
 762                     channel="stable",
 763                     num_units=1,
 764                     series=series,
 765                     to=machine_id,
 766                     config=config,
 767                 )
 768
 769                 self.log.debug(
 770                     "Wait until application {} is ready in model {}".format(
 771                         application_name, model_name
 772                     )
 773                 )
 774                 if num_units > 1:
 775                     for _ in range(num_units - 1):
 776                         m, _ = await self.create_machine(model_name, wait=False)
 777                         await application.add_unit(to=m.entity_id)
 778
 779                 await JujuModelWatcher.wait_for(
 780                     model=model,
 781                     entity=application,
 782                     progress_timeout=progress_timeout,
 783                     total_timeout=total_timeout,
 784                     db_dict=db_dict,
 785                     n2vc=self.n2vc,
 786                     vca_id=self.vca_connection._vca_id,
 787                 )
 788                 self.log.debug(
 789                     "Application {} is ready in model {}".format(
 790                         application_name, model_name
 791                     )
 792                 )
 793             else:
 794                 raise JujuApplicationExists(
 795                     "Application {} exists".format(application_name)
 796                 )
 797         except juju.errors.JujuError as e:
 798             if "already exists" in e.message:
 799                 raise JujuApplicationExists(
 800                     "Application {} exists".format(application_name)
 801                 )
 802             else:
 803                 raise e
 804         finally:
 805             await self.disconnect_model(model)
 806             await self.disconnect_controller(controller)
 807
 808         return application
 809
 810     async def scale_application(
 811         self,
 812         model_name: str,
 813         application_name: str,
 814         scale: int = 1,
 815         total_timeout: float = None,
 816     ):
 817         """
 818         Scale application (K8s)
 819
 820         :param: model_name:         Model name
 821         :param: application_name:   Application name
 822         :param: scale:              Scale to which to set this application
 823         :param: total_timeout:      Timeout for the entity to be active
 824         """
 825
 826         model = None
 827         controller = await self.get_controller()
 828         try:
 829             model = await self.get_model(controller, model_name)
 830
 831             self.log.debug(
 832                 "Scaling application {} in model {}".format(
 833                     application_name, model_name
 834                 )
 835             )
 836             application = self._get_application(model, application_name)
 837             if application is None:
 838                 raise JujuApplicationNotFound("Cannot scale application")
 839             await application.scale(scale=scale)
 840             # Wait until application is scaled in model
 841             self.log.debug(
 842                 "Waiting for application {} to be scaled in model {}...".format(
 843                     application_name, model_name
 844                 )
 845             )
 846             if total_timeout is None:
 847                 total_timeout = 1800
 848             end = time.time() + total_timeout
 849             while time.time() < end:
 850                 application_scale = self._get_application_count(model, application_name)
 851                 # Before calling wait_for_model function,
 852                 # wait until application unit count and scale count are equal.
 853                 # Because there is a delay before scaling triggers in Juju model.
 854                 if application_scale == scale:
 855                     await JujuModelWatcher.wait_for_model(
 856                         model=model, timeout=total_timeout
 857                     )
 858                     self.log.debug(
 859                         "Application {} is scaled in model {}".format(
 860                             application_name, model_name
 861                         )
 862                     )
 863                     return
 864                 await asyncio.sleep(5)
 865             raise Exception(
 866                 "Timeout waiting for application {} in model {} to be scaled".format(
 867                     application_name, model_name
 868                 )
 869             )
 870         finally:
 871             if model:
 872                 await self.disconnect_model(model)
 873             await self.disconnect_controller(controller)
 874
 875     def _get_application_count(self, model: Model, application_name: str) -> int:
 876         """Get number of units of the application
 877
 878         :param: model:              Model object
 879         :param: application_name:   Application name
 880
 881         :return: int (or None if application doesn't exist)
 882         """
 883         application = self._get_application(model, application_name)
 884         if application is not None:
 885             return len(application.units)
 886
 887     def _get_application(self, model: Model, application_name: str) -> Application:
 888         """Get application
 889
 890         :param: model:              Model object
 891         :param: application_name:   Application name
 892
 893         :return: juju.application.Application (or None if it doesn't exist)
 894         """
 895         if model.applications and application_name in model.applications:
 896             return model.applications[application_name]
 897
 898     def _get_unit(self, application: Application, machine_id: str) -> Unit:
 899         """Get unit
 900
 901         :param: application:        Application object
 902         :param: machine_id:         Machine id
 903
 904         :return: Unit
 905         """
 906         unit = None
 907         for u in application.units:
 908             if u.machine_id == machine_id:
 909                 unit = u
 910                 break
 911         return unit
 912
 913     def _get_machine_info(
 914         self,
 915         model,
 916         machine_id: str,
 917     ) -> (str, str):
 918         """Get machine info
 919
 920         :param: model:          Model object
 921         :param: machine_id:     Machine id
 922
 923         :return: (str, str): (machine, series)
 924         """
 925         if machine_id not in model.machines:
 926             msg = "Machine {} not found in model".format(machine_id)
 927             self.log.error(msg=msg)
 928             raise JujuMachineNotFound(msg)
 929         machine = model.machines[machine_id]
 930         return machine, machine.series
 931
 932     async def execute_action(
 933         self,
 934         application_name: str,
 935         model_name: str,
 936         action_name: str,
 937         db_dict: dict = None,
 938         machine_id: str = None,
 939         progress_timeout: float = None,
 940         total_timeout: float = None,
 941         **kwargs,
 942     ):
 943         """Execute action
 944
 945         :param: application_name:   Application name
 946         :param: model_name:         Model name
 947         :param: action_name:        Name of the action
 948         :param: db_dict:            Dictionary with data of the DB to write the updates
 949         :param: machine_id          Machine id
 950         :param: progress_timeout:   Maximum time between two updates in the model
 951         :param: total_timeout:      Timeout for the entity to be active
 952
 953         :return: (str, str): (output and status)
 954         """
 955         self.log.debug(
 956             "Executing action {} using params {}".format(action_name, kwargs)
 957         )
 958         # Get controller
 959         controller = await self.get_controller()
 960
 961         # Get model
 962         model = await self.get_model(controller, model_name)
 963
 964         try:
 965             # Get application
 966             application = self._get_application(
 967                 model,
 968                 application_name=application_name,
 969             )
 970             if application is None:
 971                 raise JujuApplicationNotFound("Cannot execute action")
 972             # Racing condition:
 973             #   Ocassionally, self._get_leader_unit() will return None
 974             #   because the leader elected hook has not been triggered yet.
 975             #   Therefore, we are doing some retries. If it happens again,
 976             #   re-open bug 1236
 977             if machine_id is None:
 978                 unit = await self._get_leader_unit(application)
 979                 self.log.debug(
 980                     "Action {} is being executed on the leader unit {}".format(
 981                         action_name, unit.name
 982                     )
 983                 )
 984             else:
 985                 unit = self._get_unit(application, machine_id)
 986                 if not unit:
 987                     raise JujuError(
 988                         "A unit with machine id {} not in available units".format(
 989                             machine_id
 990                         )
 991                     )
 992                 self.log.debug(
 993                     "Action {} is being executed on {} unit".format(
 994                         action_name, unit.name
 995                     )
 996                 )
 997
 998             actions = await application.get_actions()
 999
1000             if action_name not in actions:
1001                 raise JujuActionNotFound(
1002                     "Action {} not in available actions".format(action_name)
1003                 )
1004
1005             action = await unit.run_action(action_name, **kwargs)
1006
1007             self.log.debug(
1008                 "Wait until action {} is completed in application {} (model={})".format(
1009                     action_name, application_name, model_name
1010                 )
1011             )
1012             await JujuModelWatcher.wait_for(
1013                 model=model,
1014                 entity=action,
1015                 progress_timeout=progress_timeout,
1016                 total_timeout=total_timeout,
1017                 db_dict=db_dict,
1018                 n2vc=self.n2vc,
1019                 vca_id=self.vca_connection._vca_id,
1020             )
1021
1022             output = await model.get_action_output(action_uuid=action.entity_id)
1023             status = await model.get_action_status(uuid_or_prefix=action.entity_id)
1024             status = (
1025                 status[action.entity_id] if action.entity_id in status else "failed"
1026             )
1027
1028             self.log.debug(
1029                 "Action {} completed with status {} in application {} (model={})".format(
1030                     action_name, action.status, application_name, model_name
1031                 )
1032             )
1033         finally:
1034             await self.disconnect_model(model)
1035             await self.disconnect_controller(controller)
1036
1037         return output, status
1038
1039     async def get_actions(self, application_name: str, model_name: str) -> dict:
1040         """Get list of actions
1041
1042         :param: application_name: Application name
1043         :param: model_name: Model name
1044
1045         :return: Dict with this format
1046             {
1047                 "action_name": "Description of the action",
1048                 ...
1049             }
1050         """
1051         self.log.debug(
1052             "Getting list of actions for application {}".format(application_name)
1053         )
1054
1055         # Get controller
1056         controller = await self.get_controller()
1057
1058         # Get model
1059         model = await self.get_model(controller, model_name)
1060
1061         try:
1062             # Get application
1063             application = self._get_application(
1064                 model,
1065                 application_name=application_name,
1066             )
1067
1068             # Return list of actions
1069             return await application.get_actions()
1070
1071         finally:
1072             # Disconnect from model and controller
1073             await self.disconnect_model(model)
1074             await self.disconnect_controller(controller)
1075
1076     async def get_metrics(self, model_name: str, application_name: str) -> dict:
1077         """Get the metrics collected by the VCA.
1078
1079         :param model_name The name or unique id of the network service
1080         :param application_name The name of the application
1081         """
1082         if not model_name or not application_name:
1083             raise Exception("model_name and application_name must be non-empty strings")
1084         metrics = {}
1085         controller = await self.get_controller()
1086         model = await self.get_model(controller, model_name)
1087         try:
1088             application = self._get_application(model, application_name)
1089             if application is not None:
1090                 metrics = await application.get_metrics()
1091         finally:
1092             self.disconnect_model(model)
1093             self.disconnect_controller(controller)
1094         return metrics
1095
1096     async def add_relation(
1097         self,
1098         model_name: str,
1099         endpoint_1: str,
1100         endpoint_2: str,
1101     ):
1102         """Add relation
1103
1104         :param: model_name:     Model name
1105         :param: endpoint_1      First endpoint name
1106                                 ("app:endpoint" format or directly the saas name)
1107         :param: endpoint_2:     Second endpoint name (^ same format)
1108         """
1109
1110         self.log.debug("Adding relation: {} -> {}".format(endpoint_1, endpoint_2))
1111
1112         # Get controller
1113         controller = await self.get_controller()
1114
1115         # Get model
1116         model = await self.get_model(controller, model_name)
1117
1118         # Add relation
1119         try:
1120             await model.add_relation(endpoint_1, endpoint_2)
1121         except juju.errors.JujuAPIError as e:
1122             if "not found" in e.message:
1123                 self.log.warning("Relation not found: {}".format(e.message))
1124                 return
1125             if "already exists" in e.message:
1126                 self.log.warning("Relation already exists: {}".format(e.message))
1127                 return
1128             # another exception, raise it
1129             raise e
1130         finally:
1131             await self.disconnect_model(model)
1132             await self.disconnect_controller(controller)
1133
1134     async def consume(
1135         self,
1136         offer_url: str,
1137         model_name: str,
1138     ):
1139         """
1140         Adds a remote offer to the model. Relations can be created later using "juju relate".
1141
1142         :param: offer_url:      Offer Url
1143         :param: model_name:     Model name
1144
1145         :raises ParseError if there's a problem parsing the offer_url
1146         :raises JujuError if remote offer includes and endpoint
1147         :raises JujuAPIError if the operation is not successful
1148         """
1149         controller = await self.get_controller()
1150         model = await controller.get_model(model_name)
1151
1152         try:
1153             await model.consume(offer_url)
1154         finally:
1155             await self.disconnect_model(model)
1156             await self.disconnect_controller(controller)
1157
1158     async def destroy_model(self, model_name: str, total_timeout: float = 1800):
1159         """
1160         Destroy model
1161
1162         :param: model_name:     Model name
1163         :param: total_timeout:  Timeout
1164         """
1165
1166         controller = await self.get_controller()
1167         model = None
1168         try:
1169             if not await self.model_exists(model_name, controller=controller):
1170                 return
1171
1172             self.log.debug("Destroying model {}".format(model_name))
1173
1174             model = await self.get_model(controller, model_name)
1175             # Destroy machines that are manually provisioned
1176             # and still are in pending state
1177             await self._destroy_pending_machines(model, only_manual=True)
1178             await self.disconnect_model(model)
1179
1180             await self._destroy_model(
1181                 model_name,
1182                 controller,
1183                 timeout=total_timeout,
1184             )
1185         except Exception as e:
1186             if not await self.model_exists(model_name, controller=controller):
1187                 return
1188             raise e
1189         finally:
1190             if model:
1191                 await self.disconnect_model(model)
1192             await self.disconnect_controller(controller)
1193
1194     async def _destroy_model(
1195         self, model_name: str, controller: Controller, timeout: float = 1800
1196     ):
1197         """
1198         Destroy model from controller
1199
1200         :param: model: Model name to be removed
1201         :param: controller: Controller object
1202         :param: timeout: Timeout in seconds
1203         """
1204
1205         async def _destroy_model_loop(model_name: str, controller: Controller):
1206             while await self.model_exists(model_name, controller=controller):
1207                 await controller.destroy_model(
1208                     model_name, destroy_storage=True, force=True, max_wait=0
1209                 )
1210                 await asyncio.sleep(5)
1211
1212         try:
1213             await asyncio.wait_for(
1214                 _destroy_model_loop(model_name, controller), timeout=timeout
1215             )
1216         except asyncio.TimeoutError:
1217             raise Exception(
1218                 "Timeout waiting for model {} to be destroyed".format(model_name)
1219             )
1220         except juju.errors.JujuError as e:
1221             if any("has been removed" in error for error in e.errors):
1222                 return
1223             raise e
1224
1225     async def destroy_application(
1226         self, model_name: str, application_name: str, total_timeout: float
1227     ):
1228         """
1229         Destroy application
1230
1231         :param: model_name:         Model name
1232         :param: application_name:   Application name
1233         :param: total_timeout:      Timeout
1234         """
1235
1236         controller = await self.get_controller()
1237         model = None
1238
1239         try:
1240             model = await self.get_model(controller, model_name)
1241             self.log.debug(
1242                 "Destroying application {} in model {}".format(
1243                     application_name, model_name
1244                 )
1245             )
1246             application = self._get_application(model, application_name)
1247             if application:
1248                 await application.destroy()
1249             else:
1250                 self.log.warning("Application not found: {}".format(application_name))
1251
1252             self.log.debug(
1253                 "Waiting for application {} to be destroyed in model {}...".format(
1254                     application_name, model_name
1255                 )
1256             )
1257             if total_timeout is None:
1258                 total_timeout = 3600
1259             end = time.time() + total_timeout
1260             while time.time() < end:
1261                 if not self._get_application(model, application_name):
1262                     self.log.debug(
1263                         "The application {} was destroyed in model {} ".format(
1264                             application_name, model_name
1265                         )
1266                     )
1267                     return
1268                 await asyncio.sleep(5)
1269             raise Exception(
1270                 "Timeout waiting for application {} to be destroyed in model {}".format(
1271                     application_name, model_name
1272                 )
1273             )
1274         finally:
1275             if model is not None:
1276                 await self.disconnect_model(model)
1277             await self.disconnect_controller(controller)
1278
1279     async def _destroy_pending_machines(self, model: Model, only_manual: bool = False):
1280         """
1281         Destroy pending machines in a given model
1282
1283         :param: only_manual:    Bool that indicates only manually provisioned
1284                                 machines should be destroyed (if True), or that
1285                                 all pending machines should be destroyed
1286         """
1287         status = await model.get_status()
1288         for machine_id in status.machines:
1289             machine_status = status.machines[machine_id]
1290             if machine_status.agent_status.status == "pending":
1291                 if only_manual and not machine_status.instance_id.startswith("manual:"):
1292                     break
1293                 machine = model.machines[machine_id]
1294                 await machine.destroy(force=True)
1295
1296     async def configure_application(
1297         self, model_name: str, application_name: str, config: dict = None
1298     ):
1299         """Configure application
1300
1301         :param: model_name:         Model name
1302         :param: application_name:   Application name
1303         :param: config:             Config to apply to the charm
1304         """
1305         self.log.debug("Configuring application {}".format(application_name))
1306
1307         if config:
1308             controller = await self.get_controller()
1309             model = None
1310             try:
1311                 model = await self.get_model(controller, model_name)
1312                 application = self._get_application(
1313                     model,
1314                     application_name=application_name,
1315                 )
1316                 await application.set_config(config)
1317             finally:
1318                 if model:
1319                     await self.disconnect_model(model)
1320                 await self.disconnect_controller(controller)
1321
1322     def handle_exception(self, loop, context):
1323         # All unhandled exceptions by libjuju are handled here.
1324         pass
1325
1326     async def health_check(self, interval: float = 300.0):
1327         """
1328         Health check to make sure controller and controller_model connections are OK
1329
1330         :param: interval: Time in seconds between checks
1331         """
1332         controller = None
1333         while True:
1334             try:
1335                 controller = await self.get_controller()
1336                 # self.log.debug("VCA is alive")
1337             except Exception as e:
1338                 self.log.error("Health check to VCA failed: {}".format(e))
1339             finally:
1340                 await self.disconnect_controller(controller)
1341             await asyncio.sleep(interval)
1342
1343     async def list_models(self, contains: str = None) -> [str]:
1344         """List models with certain names
1345
1346         :param: contains:   String that is contained in model name
1347
1348         :retur: [models] Returns list of model names
1349         """
1350
1351         controller = await self.get_controller()
1352         try:
1353             models = await controller.list_models()
1354             if contains:
1355                 models = [model for model in models if contains in model]
1356             return models
1357         finally:
1358             await self.disconnect_controller(controller)
1359
1360     async def list_offers(self, model_name: str) -> QueryApplicationOffersResults:
1361         """List models with certain names
1362
1363         :param: model_name: Model name
1364
1365         :return:            Returns list of offers
1366         """
1367
1368         controller = await self.get_controller()
1369         try:
1370             return await controller.list_offers(model_name)
1371         finally:
1372             await self.disconnect_controller(controller)
1373
1374     async def add_k8s(
1375         self,
1376         name: str,
1377         rbac_id: str,
1378         token: str,
1379         client_cert_data: str,
1380         configuration: Configuration,
1381         storage_class: str,
1382         credential_name: str = None,
1383     ):
1384         """
1385         Add a Kubernetes cloud to the controller
1386
1387         Similar to the `juju add-k8s` command in the CLI
1388
1389         :param: name:               Name for the K8s cloud
1390         :param: configuration:      Kubernetes configuration object
1391         :param: storage_class:      Storage Class to use in the cloud
1392         :param: credential_name:    Storage Class to use in the cloud
1393         """
1394
1395         if not storage_class:
1396             raise Exception("storage_class must be a non-empty string")
1397         if not name:
1398             raise Exception("name must be a non-empty string")
1399         if not configuration:
1400             raise Exception("configuration must be provided")
1401
1402         endpoint = configuration.host
1403         credential = self.get_k8s_cloud_credential(
1404             configuration,
1405             client_cert_data,
1406             token,
1407         )
1408         credential.attrs[RBAC_LABEL_KEY_NAME] = rbac_id
1409         cloud = client.Cloud(
1410             type_="kubernetes",
1411             auth_types=[credential.auth_type],
1412             endpoint=endpoint,
1413             ca_certificates=[client_cert_data],
1414             config={
1415                 "operator-storage": storage_class,
1416                 "workload-storage": storage_class,
1417             },
1418         )
1419
1420         return await self.add_cloud(
1421             name, cloud, credential, credential_name=credential_name
1422         )
1423
1424     def get_k8s_cloud_credential(
1425         self,
1426         configuration: Configuration,
1427         client_cert_data: str,
1428         token: str = None,
1429     ) -> client.CloudCredential:
1430         attrs = {}
1431         # TODO: Test with AKS
1432         key = None  # open(configuration.key_file, "r").read()
1433         username = configuration.username
1434         password = configuration.password
1435
1436         if client_cert_data:
1437             attrs["ClientCertificateData"] = client_cert_data
1438         if key:
1439             attrs["ClientKeyData"] = key
1440         if token:
1441             if username or password:
1442                 raise JujuInvalidK8sConfiguration("Cannot set both token and user/pass")
1443             attrs["Token"] = token
1444
1445         auth_type = None
1446         if key:
1447             auth_type = "oauth2"
1448             if client_cert_data:
1449                 auth_type = "oauth2withcert"
1450             if not token:
1451                 raise JujuInvalidK8sConfiguration(
1452                     "missing token for auth type {}".format(auth_type)
1453                 )
1454         elif username:
1455             if not password:
1456                 self.log.debug(
1457                     "credential for user {} has empty password".format(username)
1458                 )
1459             attrs["username"] = username
1460             attrs["password"] = password
1461             if client_cert_data:
1462                 auth_type = "userpasswithcert"
1463             else:
1464                 auth_type = "userpass"
1465         elif client_cert_data and token:
1466             auth_type = "certificate"
1467         else:
1468             raise JujuInvalidK8sConfiguration("authentication method not supported")
1469         return client.CloudCredential(auth_type=auth_type, attrs=attrs)
1470
1471     async def add_cloud(
1472         self,
1473         name: str,
1474         cloud: Cloud,
1475         credential: CloudCredential = None,
1476         credential_name: str = None,
1477     ) -> Cloud:
1478         """
1479         Add cloud to the controller
1480
1481         :param: name:               Name of the cloud to be added
1482         :param: cloud:              Cloud object
1483         :param: credential:         CloudCredentials object for the cloud
1484         :param: credential_name:    Credential name.
1485                                     If not defined, cloud of the name will be used.
1486         """
1487         controller = await self.get_controller()
1488         try:
1489             _ = await controller.add_cloud(name, cloud)
1490             if credential:
1491                 await controller.add_credential(
1492                     credential_name or name, credential=credential, cloud=name
1493                 )
1494             # Need to return the object returned by the controller.add_cloud() function
1495             # I'm returning the original value now until this bug is fixed:
1496             #   https://github.com/juju/python-libjuju/issues/443
1497             return cloud
1498         finally:
1499             await self.disconnect_controller(controller)
1500
1501     async def remove_cloud(self, name: str):
1502         """
1503         Remove cloud
1504
1505         :param: name:   Name of the cloud to be removed
1506         """
1507         controller = await self.get_controller()
1508         try:
1509             await controller.remove_cloud(name)
1510         except juju.errors.JujuError as e:
1511             if len(e.errors) == 1 and f'cloud "{name}" not found' == e.errors[0]:
1512                 self.log.warning(f"Cloud {name} not found, so it could not be deleted.")
1513             else:
1514                 raise e
1515         finally:
1516             await self.disconnect_controller(controller)
1517
1518     @retry(attempts=20, delay=5, fallback=JujuLeaderUnitNotFound())
1519     async def _get_leader_unit(self, application: Application) -> Unit:
1520         unit = None
1521         for u in application.units:
1522             if await u.is_leader_from_status():
1523                 unit = u
1524                 break
1525         if not unit:
1526             raise Exception()
1527         return unit
1528
1529     async def get_cloud_credentials(self, cloud: Cloud) -> typing.List:
1530         """
1531         Get cloud credentials
1532
1533         :param: cloud: Cloud object. The returned credentials will be from this cloud.
1534
1535         :return: List of credentials object associated to the specified cloud
1536
1537         """
1538         controller = await self.get_controller()
1539         try:
1540             facade = client.CloudFacade.from_connection(controller.connection())
1541             cloud_cred_tag = tag.credential(
1542                 cloud.name, self.vca_connection.data.user, cloud.credential_name
1543             )
1544             params = [client.Entity(cloud_cred_tag)]
1545             return (await facade.Credential(params)).results
1546         finally:
1547             await self.disconnect_controller(controller)
1548
1549     async def check_application_exists(self, model_name, application_name) -> bool:
1550         """Check application exists
1551
1552         :param: model_name:         Model Name
1553         :param: application_name:   Application Name
1554
1555         :return: Boolean
1556         """
1557
1558         model = None
1559         controller = await self.get_controller()
1560         try:
1561             model = await self.get_model(controller, model_name)
1562             self.log.debug(
1563                 "Checking if application {} exists in model {}".format(
1564                     application_name, model_name
1565                 )
1566             )
1567             return self._get_application(model, application_name) is not None
1568         finally:
1569             if model:
1570                 await self.disconnect_model(model)
1571             await self.disconnect_controller(controller)