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