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, trust=True)
 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 = 1800):
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             self.log.debug("Destroying model {}".format(model_name))
1163
1164             model = await self.get_model(controller, model_name)
1165             # Destroy machines that are manually provisioned
1166             # and still are in pending state
1167             await self._destroy_pending_machines(model, only_manual=True)
1168             await self.disconnect_model(model)
1169
1170             await self._destroy_model(
1171                 model_name,
1172                 controller,
1173                 timeout=total_timeout,
1174             )
1175         finally:
1176             if model:
1177                 await self.disconnect_model(model)
1178             await self.disconnect_controller(controller)
1179
1180     async def _destroy_model(
1181         self, model_name: str, controller: Controller, timeout: float = 1800
1182     ):
1183         """
1184         Destroy model from controller
1185
1186         :param: model: Model name to be removed
1187         :param: controller: Controller object
1188         :param: timeout: Timeout in seconds
1189         """
1190
1191         async def _destroy_model_loop(model_name: str, controller: Controller):
1192             while await self.model_exists(model_name, controller=controller):
1193                 await controller.destroy_model(
1194                     model_name, destroy_storage=True, force=True, max_wait=0
1195                 )
1196                 await asyncio.sleep(5)
1197
1198         try:
1199             await asyncio.wait_for(
1200                 _destroy_model_loop(model_name, controller), timeout=timeout
1201             )
1202         except asyncio.TimeoutError:
1203             raise Exception(
1204                 "Timeout waiting for model {} to be destroyed".format(model_name)
1205             )
1206
1207     async def destroy_application(
1208         self, model_name: str, application_name: str, total_timeout: float
1209     ):
1210         """
1211         Destroy application
1212
1213         :param: model_name:         Model name
1214         :param: application_name:   Application name
1215         :param: total_timeout:      Timeout
1216         """
1217
1218         controller = await self.get_controller()
1219         model = None
1220
1221         try:
1222             model = await self.get_model(controller, model_name)
1223             self.log.debug(
1224                 "Destroying application {} in model {}".format(
1225                     application_name, model_name
1226                 )
1227             )
1228             application = self._get_application(model, application_name)
1229             if application:
1230                 await application.destroy()
1231             else:
1232                 self.log.warning("Application not found: {}".format(application_name))
1233
1234             self.log.debug(
1235                 "Waiting for application {} to be destroyed in model {}...".format(
1236                     application_name, model_name
1237                 )
1238             )
1239             if total_timeout is None:
1240                 total_timeout = 3600
1241             end = time.time() + total_timeout
1242             while time.time() < end:
1243                 if not self._get_application(model, application_name):
1244                     self.log.debug(
1245                         "The application {} was destroyed in model {} ".format(
1246                             application_name, model_name
1247                         )
1248                     )
1249                     return
1250                 await asyncio.sleep(5)
1251             raise Exception(
1252                 "Timeout waiting for application {} to be destroyed in model {}".format(
1253                     application_name, model_name
1254                 )
1255             )
1256         finally:
1257             if model is not None:
1258                 await self.disconnect_model(model)
1259             await self.disconnect_controller(controller)
1260
1261     async def _destroy_pending_machines(self, model: Model, only_manual: bool = False):
1262         """
1263         Destroy pending machines in a given model
1264
1265         :param: only_manual:    Bool that indicates only manually provisioned
1266                                 machines should be destroyed (if True), or that
1267                                 all pending machines should be destroyed
1268         """
1269         status = await model.get_status()
1270         for machine_id in status.machines:
1271             machine_status = status.machines[machine_id]
1272             if machine_status.agent_status.status == "pending":
1273                 if only_manual and not machine_status.instance_id.startswith("manual:"):
1274                     break
1275                 machine = model.machines[machine_id]
1276                 await machine.destroy(force=True)
1277
1278     async def configure_application(
1279         self, model_name: str, application_name: str, config: dict = None
1280     ):
1281         """Configure application
1282
1283         :param: model_name:         Model name
1284         :param: application_name:   Application name
1285         :param: config:             Config to apply to the charm
1286         """
1287         self.log.debug("Configuring application {}".format(application_name))
1288
1289         if config:
1290             controller = await self.get_controller()
1291             model = None
1292             try:
1293                 model = await self.get_model(controller, model_name)
1294                 application = self._get_application(
1295                     model,
1296                     application_name=application_name,
1297                 )
1298                 await application.set_config(config)
1299             finally:
1300                 if model:
1301                     await self.disconnect_model(model)
1302                 await self.disconnect_controller(controller)
1303
1304     def handle_exception(self, loop, context):
1305         # All unhandled exceptions by libjuju are handled here.
1306         pass
1307
1308     async def health_check(self, interval: float = 300.0):
1309         """
1310         Health check to make sure controller and controller_model connections are OK
1311
1312         :param: interval: Time in seconds between checks
1313         """
1314         controller = None
1315         while True:
1316             try:
1317                 controller = await self.get_controller()
1318                 # self.log.debug("VCA is alive")
1319             except Exception as e:
1320                 self.log.error("Health check to VCA failed: {}".format(e))
1321             finally:
1322                 await self.disconnect_controller(controller)
1323             await asyncio.sleep(interval)
1324
1325     async def list_models(self, contains: str = None) -> [str]:
1326         """List models with certain names
1327
1328         :param: contains:   String that is contained in model name
1329
1330         :retur: [models] Returns list of model names
1331         """
1332
1333         controller = await self.get_controller()
1334         try:
1335             models = await controller.list_models()
1336             if contains:
1337                 models = [model for model in models if contains in model]
1338             return models
1339         finally:
1340             await self.disconnect_controller(controller)
1341
1342     async def list_offers(self, model_name: str) -> QueryApplicationOffersResults:
1343         """List models with certain names
1344
1345         :param: model_name: Model name
1346
1347         :return:            Returns list of offers
1348         """
1349
1350         controller = await self.get_controller()
1351         try:
1352             return await controller.list_offers(model_name)
1353         finally:
1354             await self.disconnect_controller(controller)
1355
1356     async def add_k8s(
1357         self,
1358         name: str,
1359         rbac_id: str,
1360         token: str,
1361         client_cert_data: str,
1362         configuration: Configuration,
1363         storage_class: str,
1364         credential_name: str = None,
1365     ):
1366         """
1367         Add a Kubernetes cloud to the controller
1368
1369         Similar to the `juju add-k8s` command in the CLI
1370
1371         :param: name:               Name for the K8s cloud
1372         :param: configuration:      Kubernetes configuration object
1373         :param: storage_class:      Storage Class to use in the cloud
1374         :param: credential_name:    Storage Class to use in the cloud
1375         """
1376
1377         if not storage_class:
1378             raise Exception("storage_class must be a non-empty string")
1379         if not name:
1380             raise Exception("name must be a non-empty string")
1381         if not configuration:
1382             raise Exception("configuration must be provided")
1383
1384         endpoint = configuration.host
1385         credential = self.get_k8s_cloud_credential(
1386             configuration,
1387             client_cert_data,
1388             token,
1389         )
1390         credential.attrs[RBAC_LABEL_KEY_NAME] = rbac_id
1391         cloud = client.Cloud(
1392             type_="kubernetes",
1393             auth_types=[credential.auth_type],
1394             endpoint=endpoint,
1395             ca_certificates=[client_cert_data],
1396             config={
1397                 "operator-storage": storage_class,
1398                 "workload-storage": storage_class,
1399             },
1400         )
1401
1402         return await self.add_cloud(
1403             name, cloud, credential, credential_name=credential_name
1404         )
1405
1406     def get_k8s_cloud_credential(
1407         self,
1408         configuration: Configuration,
1409         client_cert_data: str,
1410         token: str = None,
1411     ) -> client.CloudCredential:
1412         attrs = {}
1413         # TODO: Test with AKS
1414         key = None  # open(configuration.key_file, "r").read()
1415         username = configuration.username
1416         password = configuration.password
1417
1418         if client_cert_data:
1419             attrs["ClientCertificateData"] = client_cert_data
1420         if key:
1421             attrs["ClientKeyData"] = key
1422         if token:
1423             if username or password:
1424                 raise JujuInvalidK8sConfiguration("Cannot set both token and user/pass")
1425             attrs["Token"] = token
1426
1427         auth_type = None
1428         if key:
1429             auth_type = "oauth2"
1430             if client_cert_data:
1431                 auth_type = "oauth2withcert"
1432             if not token:
1433                 raise JujuInvalidK8sConfiguration(
1434                     "missing token for auth type {}".format(auth_type)
1435                 )
1436         elif username:
1437             if not password:
1438                 self.log.debug(
1439                     "credential for user {} has empty password".format(username)
1440                 )
1441             attrs["username"] = username
1442             attrs["password"] = password
1443             if client_cert_data:
1444                 auth_type = "userpasswithcert"
1445             else:
1446                 auth_type = "userpass"
1447         elif client_cert_data and token:
1448             auth_type = "certificate"
1449         else:
1450             raise JujuInvalidK8sConfiguration("authentication method not supported")
1451         return client.CloudCredential(auth_type=auth_type, attrs=attrs)
1452
1453     async def add_cloud(
1454         self,
1455         name: str,
1456         cloud: Cloud,
1457         credential: CloudCredential = None,
1458         credential_name: str = None,
1459     ) -> Cloud:
1460         """
1461         Add cloud to the controller
1462
1463         :param: name:               Name of the cloud to be added
1464         :param: cloud:              Cloud object
1465         :param: credential:         CloudCredentials object for the cloud
1466         :param: credential_name:    Credential name.
1467                                     If not defined, cloud of the name will be used.
1468         """
1469         controller = await self.get_controller()
1470         try:
1471             _ = await controller.add_cloud(name, cloud)
1472             if credential:
1473                 await controller.add_credential(
1474                     credential_name or name, credential=credential, cloud=name
1475                 )
1476             # Need to return the object returned by the controller.add_cloud() function
1477             # I'm returning the original value now until this bug is fixed:
1478             #   https://github.com/juju/python-libjuju/issues/443
1479             return cloud
1480         finally:
1481             await self.disconnect_controller(controller)
1482
1483     async def remove_cloud(self, name: str):
1484         """
1485         Remove cloud
1486
1487         :param: name:   Name of the cloud to be removed
1488         """
1489         controller = await self.get_controller()
1490         try:
1491             await controller.remove_cloud(name)
1492         except juju.errors.JujuError as e:
1493             if len(e.errors) == 1 and f'cloud "{name}" not found' == e.errors[0]:
1494                 self.log.warning(f"Cloud {name} not found, so it could not be deleted.")
1495             else:
1496                 raise e
1497         finally:
1498             await self.disconnect_controller(controller)
1499
1500     @retry(attempts=20, delay=5, fallback=JujuLeaderUnitNotFound())
1501     async def _get_leader_unit(self, application: Application) -> Unit:
1502         unit = None
1503         for u in application.units:
1504             if await u.is_leader_from_status():
1505                 unit = u
1506                 break
1507         if not unit:
1508             raise Exception()
1509         return unit
1510
1511     async def get_cloud_credentials(self, cloud: Cloud) -> typing.List:
1512         """
1513         Get cloud credentials
1514
1515         :param: cloud: Cloud object. The returned credentials will be from this cloud.
1516
1517         :return: List of credentials object associated to the specified cloud
1518
1519         """
1520         controller = await self.get_controller()
1521         try:
1522             facade = client.CloudFacade.from_connection(controller.connection())
1523             cloud_cred_tag = tag.credential(
1524                 cloud.name, self.vca_connection.data.user, cloud.credential_name
1525             )
1526             params = [client.Entity(cloud_cred_tag)]
1527             return (await facade.Credential(params)).results
1528         finally:
1529             await self.disconnect_controller(controller)
1530
1531     async def check_application_exists(self, model_name, application_name) -> bool:
1532         """Check application exists
1533
1534         :param: model_name:         Model Name
1535         :param: application_name:   Application Name
1536
1537         :return: Boolean
1538         """
1539
1540         model = None
1541         controller = await self.get_controller()
1542         try:
1543             model = await self.get_model(controller, model_name)
1544             self.log.debug(
1545                 "Checking if application {} exists in model {}".format(
1546                     application_name, model_name
1547                 )
1548             )
1549             return self._get_application(model, application_name) is not None
1550         finally:
1551             if model:
1552                 await self.disconnect_model(model)
1553             await self.disconnect_controller(controller)