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         except juju.errors.JujuError as e:
 795             if "already exists" in e.message:
 796                 raise JujuApplicationExists(
 797                     "Application {} exists".format(application_name)
 798                 )
 799             else:
 800                 raise e
 801         finally:
 802             await self.disconnect_model(model)
 803             await self.disconnect_controller(controller)
 804
 805         return application
 806
 807     async def scale_application(
 808         self,
 809         model_name: str,
 810         application_name: str,
 811         scale: int = 1,
 812         total_timeout: float = None,
 813     ):
 814         """
 815         Scale application (K8s)
 816
 817         :param: model_name:         Model name
 818         :param: application_name:   Application name
 819         :param: scale:              Scale to which to set this application
 820         :param: total_timeout:      Timeout for the entity to be active
 821         """
 822
 823         model = None
 824         controller = await self.get_controller()
 825         try:
 826             model = await self.get_model(controller, model_name)
 827
 828             self.log.debug(
 829                 "Scaling application {} in model {}".format(
 830                     application_name, model_name
 831                 )
 832             )
 833             application = self._get_application(model, application_name)
 834             if application is None:
 835                 raise JujuApplicationNotFound("Cannot scale application")
 836             await application.scale(scale=scale)
 837             # Wait until application is scaled in model
 838             self.log.debug(
 839                 "Waiting for application {} to be scaled in model {}...".format(
 840                     application_name, model_name
 841                 )
 842             )
 843             if total_timeout is None:
 844                 total_timeout = 1800
 845             end = time.time() + total_timeout
 846             while time.time() < end:
 847                 application_scale = self._get_application_count(model, application_name)
 848                 # Before calling wait_for_model function,
 849                 # wait until application unit count and scale count are equal.
 850                 # Because there is a delay before scaling triggers in Juju model.
 851                 if application_scale == scale:
 852                     await JujuModelWatcher.wait_for_model(
 853                         model=model, timeout=total_timeout
 854                     )
 855                     self.log.debug(
 856                         "Application {} is scaled in model {}".format(
 857                             application_name, model_name
 858                         )
 859                     )
 860                     return
 861                 await asyncio.sleep(5)
 862             raise Exception(
 863                 "Timeout waiting for application {} in model {} to be scaled".format(
 864                     application_name, model_name
 865                 )
 866             )
 867         finally:
 868             if model:
 869                 await self.disconnect_model(model)
 870             await self.disconnect_controller(controller)
 871
 872     def _get_application_count(self, model: Model, application_name: str) -> int:
 873         """Get number of units of the application
 874
 875         :param: model:              Model object
 876         :param: application_name:   Application name
 877
 878         :return: int (or None if application doesn't exist)
 879         """
 880         application = self._get_application(model, application_name)
 881         if application is not None:
 882             return len(application.units)
 883
 884     def _get_application(self, model: Model, application_name: str) -> Application:
 885         """Get application
 886
 887         :param: model:              Model object
 888         :param: application_name:   Application name
 889
 890         :return: juju.application.Application (or None if it doesn't exist)
 891         """
 892         if model.applications and application_name in model.applications:
 893             return model.applications[application_name]
 894
 895     def _get_unit(self, application: Application, machine_id: str) -> Unit:
 896         """Get unit
 897
 898         :param: application:        Application object
 899         :param: machine_id:         Machine id
 900
 901         :return: Unit
 902         """
 903         unit = None
 904         for u in application.units:
 905             if u.machine_id == machine_id:
 906                 unit = u
 907                 break
 908         return unit
 909
 910     def _get_machine_info(
 911         self,
 912         model,
 913         machine_id: str,
 914     ) -> (str, str):
 915         """Get machine info
 916
 917         :param: model:          Model object
 918         :param: machine_id:     Machine id
 919
 920         :return: (str, str): (machine, series)
 921         """
 922         if machine_id not in model.machines:
 923             msg = "Machine {} not found in model".format(machine_id)
 924             self.log.error(msg=msg)
 925             raise JujuMachineNotFound(msg)
 926         machine = model.machines[machine_id]
 927         return machine, machine.series
 928
 929     async def execute_action(
 930         self,
 931         application_name: str,
 932         model_name: str,
 933         action_name: str,
 934         db_dict: dict = None,
 935         machine_id: str = None,
 936         progress_timeout: float = None,
 937         total_timeout: float = None,
 938         **kwargs,
 939     ):
 940         """Execute action
 941
 942         :param: application_name:   Application name
 943         :param: model_name:         Model name
 944         :param: action_name:        Name of the action
 945         :param: db_dict:            Dictionary with data of the DB to write the updates
 946         :param: machine_id          Machine id
 947         :param: progress_timeout:   Maximum time between two updates in the model
 948         :param: total_timeout:      Timeout for the entity to be active
 949
 950         :return: (str, str): (output and status)
 951         """
 952         self.log.debug(
 953             "Executing action {} using params {}".format(action_name, kwargs)
 954         )
 955         # Get controller
 956         controller = await self.get_controller()
 957
 958         # Get model
 959         model = await self.get_model(controller, model_name)
 960
 961         try:
 962             # Get application
 963             application = self._get_application(
 964                 model,
 965                 application_name=application_name,
 966             )
 967             if application is None:
 968                 raise JujuApplicationNotFound("Cannot execute action")
 969             # Racing condition:
 970             #   Ocassionally, self._get_leader_unit() will return None
 971             #   because the leader elected hook has not been triggered yet.
 972             #   Therefore, we are doing some retries. If it happens again,
 973             #   re-open bug 1236
 974             if machine_id is None:
 975                 unit = await self._get_leader_unit(application)
 976                 self.log.debug(
 977                     "Action {} is being executed on the leader unit {}".format(
 978                         action_name, unit.name
 979                     )
 980                 )
 981             else:
 982                 unit = self._get_unit(application, machine_id)
 983                 if not unit:
 984                     raise JujuError(
 985                         "A unit with machine id {} not in available units".format(
 986                             machine_id
 987                         )
 988                     )
 989                 self.log.debug(
 990                     "Action {} is being executed on {} unit".format(
 991                         action_name, unit.name
 992                     )
 993                 )
 994
 995             actions = await application.get_actions()
 996
 997             if action_name not in actions:
 998                 raise JujuActionNotFound(
 999                     "Action {} not in available actions".format(action_name)
1000                 )
1001
1002             action = await unit.run_action(action_name, **kwargs)
1003
1004             self.log.debug(
1005                 "Wait until action {} is completed in application {} (model={})".format(
1006                     action_name, application_name, model_name
1007                 )
1008             )
1009             await JujuModelWatcher.wait_for(
1010                 model=model,
1011                 entity=action,
1012                 progress_timeout=progress_timeout,
1013                 total_timeout=total_timeout,
1014                 db_dict=db_dict,
1015                 n2vc=self.n2vc,
1016                 vca_id=self.vca_connection._vca_id,
1017             )
1018
1019             output = await model.get_action_output(action_uuid=action.entity_id)
1020             status = await model.get_action_status(uuid_or_prefix=action.entity_id)
1021             status = (
1022                 status[action.entity_id] if action.entity_id in status else "failed"
1023             )
1024
1025             self.log.debug(
1026                 "Action {} completed with status {} in application {} (model={})".format(
1027                     action_name, action.status, application_name, model_name
1028                 )
1029             )
1030         finally:
1031             await self.disconnect_model(model)
1032             await self.disconnect_controller(controller)
1033
1034         return output, status
1035
1036     async def get_actions(self, application_name: str, model_name: str) -> dict:
1037         """Get list of actions
1038
1039         :param: application_name: Application name
1040         :param: model_name: Model name
1041
1042         :return: Dict with this format
1043             {
1044                 "action_name": "Description of the action",
1045                 ...
1046             }
1047         """
1048         self.log.debug(
1049             "Getting list of actions for application {}".format(application_name)
1050         )
1051
1052         # Get controller
1053         controller = await self.get_controller()
1054
1055         # Get model
1056         model = await self.get_model(controller, model_name)
1057
1058         try:
1059             # Get application
1060             application = self._get_application(
1061                 model,
1062                 application_name=application_name,
1063             )
1064
1065             # Return list of actions
1066             return await application.get_actions()
1067
1068         finally:
1069             # Disconnect from model and controller
1070             await self.disconnect_model(model)
1071             await self.disconnect_controller(controller)
1072
1073     async def get_metrics(self, model_name: str, application_name: str) -> dict:
1074         """Get the metrics collected by the VCA.
1075
1076         :param model_name The name or unique id of the network service
1077         :param application_name The name of the application
1078         """
1079         if not model_name or not application_name:
1080             raise Exception("model_name and application_name must be non-empty strings")
1081         metrics = {}
1082         controller = await self.get_controller()
1083         model = await self.get_model(controller, model_name)
1084         try:
1085             application = self._get_application(model, application_name)
1086             if application is not None:
1087                 metrics = await application.get_metrics()
1088         finally:
1089             self.disconnect_model(model)
1090             self.disconnect_controller(controller)
1091         return metrics
1092
1093     async def add_relation(
1094         self,
1095         model_name: str,
1096         endpoint_1: str,
1097         endpoint_2: str,
1098     ):
1099         """Add relation
1100
1101         :param: model_name:     Model name
1102         :param: endpoint_1      First endpoint name
1103                                 ("app:endpoint" format or directly the saas name)
1104         :param: endpoint_2:     Second endpoint name (^ same format)
1105         """
1106
1107         self.log.debug("Adding relation: {} -> {}".format(endpoint_1, endpoint_2))
1108
1109         # Get controller
1110         controller = await self.get_controller()
1111
1112         # Get model
1113         model = await self.get_model(controller, model_name)
1114
1115         # Add relation
1116         try:
1117             await model.add_relation(endpoint_1, endpoint_2)
1118         except juju.errors.JujuAPIError as e:
1119             if "not found" in e.message:
1120                 self.log.warning("Relation not found: {}".format(e.message))
1121                 return
1122             if "already exists" in e.message:
1123                 self.log.warning("Relation already exists: {}".format(e.message))
1124                 return
1125             # another exception, raise it
1126             raise e
1127         finally:
1128             await self.disconnect_model(model)
1129             await self.disconnect_controller(controller)
1130
1131     async def consume(
1132         self,
1133         offer_url: str,
1134         model_name: str,
1135     ):
1136         """
1137         Adds a remote offer to the model. Relations can be created later using "juju relate".
1138
1139         :param: offer_url:      Offer Url
1140         :param: model_name:     Model name
1141
1142         :raises ParseError if there's a problem parsing the offer_url
1143         :raises JujuError if remote offer includes and endpoint
1144         :raises JujuAPIError if the operation is not successful
1145         """
1146         controller = await self.get_controller()
1147         model = await controller.get_model(model_name)
1148
1149         try:
1150             await model.consume(offer_url)
1151         finally:
1152             await self.disconnect_model(model)
1153             await self.disconnect_controller(controller)
1154
1155     async def destroy_model(self, model_name: str, total_timeout: float = 1800):
1156         """
1157         Destroy model
1158
1159         :param: model_name:     Model name
1160         :param: total_timeout:  Timeout
1161         """
1162
1163         controller = await self.get_controller()
1164         model = None
1165         try:
1166             if not await self.model_exists(model_name, controller=controller):
1167                 return
1168
1169             self.log.debug("Destroying model {}".format(model_name))
1170
1171             model = await self.get_model(controller, model_name)
1172             # Destroy machines that are manually provisioned
1173             # and still are in pending state
1174             await self._destroy_pending_machines(model, only_manual=True)
1175             await self.disconnect_model(model)
1176
1177             await self._destroy_model(
1178                 model_name,
1179                 controller,
1180                 timeout=total_timeout,
1181             )
1182         finally:
1183             if model:
1184                 await self.disconnect_model(model)
1185             await self.disconnect_controller(controller)
1186
1187     async def _destroy_model(
1188         self, model_name: str, controller: Controller, timeout: float = 1800
1189     ):
1190         """
1191         Destroy model from controller
1192
1193         :param: model: Model name to be removed
1194         :param: controller: Controller object
1195         :param: timeout: Timeout in seconds
1196         """
1197
1198         async def _destroy_model_loop(model_name: str, controller: Controller):
1199             while await self.model_exists(model_name, controller=controller):
1200                 await controller.destroy_model(
1201                     model_name, destroy_storage=True, force=True, max_wait=0
1202                 )
1203                 await asyncio.sleep(5)
1204
1205         try:
1206             await asyncio.wait_for(
1207                 _destroy_model_loop(model_name, controller), timeout=timeout
1208             )
1209         except asyncio.TimeoutError:
1210             raise Exception(
1211                 "Timeout waiting for model {} to be destroyed".format(model_name)
1212             )
1213
1214     async def destroy_application(
1215         self, model_name: str, application_name: str, total_timeout: float
1216     ):
1217         """
1218         Destroy application
1219
1220         :param: model_name:         Model name
1221         :param: application_name:   Application name
1222         :param: total_timeout:      Timeout
1223         """
1224
1225         controller = await self.get_controller()
1226         model = None
1227
1228         try:
1229             model = await self.get_model(controller, model_name)
1230             self.log.debug(
1231                 "Destroying application {} in model {}".format(
1232                     application_name, model_name
1233                 )
1234             )
1235             application = self._get_application(model, application_name)
1236             if application:
1237                 await application.destroy()
1238             else:
1239                 self.log.warning("Application not found: {}".format(application_name))
1240
1241             self.log.debug(
1242                 "Waiting for application {} to be destroyed in model {}...".format(
1243                     application_name, model_name
1244                 )
1245             )
1246             if total_timeout is None:
1247                 total_timeout = 3600
1248             end = time.time() + total_timeout
1249             while time.time() < end:
1250                 if not self._get_application(model, application_name):
1251                     self.log.debug(
1252                         "The application {} was destroyed in model {} ".format(
1253                             application_name, model_name
1254                         )
1255                     )
1256                     return
1257                 await asyncio.sleep(5)
1258             raise Exception(
1259                 "Timeout waiting for application {} to be destroyed in model {}".format(
1260                     application_name, model_name
1261                 )
1262             )
1263         finally:
1264             if model is not None:
1265                 await self.disconnect_model(model)
1266             await self.disconnect_controller(controller)
1267
1268     async def _destroy_pending_machines(self, model: Model, only_manual: bool = False):
1269         """
1270         Destroy pending machines in a given model
1271
1272         :param: only_manual:    Bool that indicates only manually provisioned
1273                                 machines should be destroyed (if True), or that
1274                                 all pending machines should be destroyed
1275         """
1276         status = await model.get_status()
1277         for machine_id in status.machines:
1278             machine_status = status.machines[machine_id]
1279             if machine_status.agent_status.status == "pending":
1280                 if only_manual and not machine_status.instance_id.startswith("manual:"):
1281                     break
1282                 machine = model.machines[machine_id]
1283                 await machine.destroy(force=True)
1284
1285     async def configure_application(
1286         self, model_name: str, application_name: str, config: dict = None
1287     ):
1288         """Configure application
1289
1290         :param: model_name:         Model name
1291         :param: application_name:   Application name
1292         :param: config:             Config to apply to the charm
1293         """
1294         self.log.debug("Configuring application {}".format(application_name))
1295
1296         if config:
1297             controller = await self.get_controller()
1298             model = None
1299             try:
1300                 model = await self.get_model(controller, model_name)
1301                 application = self._get_application(
1302                     model,
1303                     application_name=application_name,
1304                 )
1305                 await application.set_config(config)
1306             finally:
1307                 if model:
1308                     await self.disconnect_model(model)
1309                 await self.disconnect_controller(controller)
1310
1311     def handle_exception(self, loop, context):
1312         # All unhandled exceptions by libjuju are handled here.
1313         pass
1314
1315     async def health_check(self, interval: float = 300.0):
1316         """
1317         Health check to make sure controller and controller_model connections are OK
1318
1319         :param: interval: Time in seconds between checks
1320         """
1321         controller = None
1322         while True:
1323             try:
1324                 controller = await self.get_controller()
1325                 # self.log.debug("VCA is alive")
1326             except Exception as e:
1327                 self.log.error("Health check to VCA failed: {}".format(e))
1328             finally:
1329                 await self.disconnect_controller(controller)
1330             await asyncio.sleep(interval)
1331
1332     async def list_models(self, contains: str = None) -> [str]:
1333         """List models with certain names
1334
1335         :param: contains:   String that is contained in model name
1336
1337         :retur: [models] Returns list of model names
1338         """
1339
1340         controller = await self.get_controller()
1341         try:
1342             models = await controller.list_models()
1343             if contains:
1344                 models = [model for model in models if contains in model]
1345             return models
1346         finally:
1347             await self.disconnect_controller(controller)
1348
1349     async def list_offers(self, model_name: str) -> QueryApplicationOffersResults:
1350         """List models with certain names
1351
1352         :param: model_name: Model name
1353
1354         :return:            Returns list of offers
1355         """
1356
1357         controller = await self.get_controller()
1358         try:
1359             return await controller.list_offers(model_name)
1360         finally:
1361             await self.disconnect_controller(controller)
1362
1363     async def add_k8s(
1364         self,
1365         name: str,
1366         rbac_id: str,
1367         token: str,
1368         client_cert_data: str,
1369         configuration: Configuration,
1370         storage_class: str,
1371         credential_name: str = None,
1372     ):
1373         """
1374         Add a Kubernetes cloud to the controller
1375
1376         Similar to the `juju add-k8s` command in the CLI
1377
1378         :param: name:               Name for the K8s cloud
1379         :param: configuration:      Kubernetes configuration object
1380         :param: storage_class:      Storage Class to use in the cloud
1381         :param: credential_name:    Storage Class to use in the cloud
1382         """
1383
1384         if not storage_class:
1385             raise Exception("storage_class must be a non-empty string")
1386         if not name:
1387             raise Exception("name must be a non-empty string")
1388         if not configuration:
1389             raise Exception("configuration must be provided")
1390
1391         endpoint = configuration.host
1392         credential = self.get_k8s_cloud_credential(
1393             configuration,
1394             client_cert_data,
1395             token,
1396         )
1397         credential.attrs[RBAC_LABEL_KEY_NAME] = rbac_id
1398         cloud = client.Cloud(
1399             type_="kubernetes",
1400             auth_types=[credential.auth_type],
1401             endpoint=endpoint,
1402             ca_certificates=[client_cert_data],
1403             config={
1404                 "operator-storage": storage_class,
1405                 "workload-storage": storage_class,
1406             },
1407         )
1408
1409         return await self.add_cloud(
1410             name, cloud, credential, credential_name=credential_name
1411         )
1412
1413     def get_k8s_cloud_credential(
1414         self,
1415         configuration: Configuration,
1416         client_cert_data: str,
1417         token: str = None,
1418     ) -> client.CloudCredential:
1419         attrs = {}
1420         # TODO: Test with AKS
1421         key = None  # open(configuration.key_file, "r").read()
1422         username = configuration.username
1423         password = configuration.password
1424
1425         if client_cert_data:
1426             attrs["ClientCertificateData"] = client_cert_data
1427         if key:
1428             attrs["ClientKeyData"] = key
1429         if token:
1430             if username or password:
1431                 raise JujuInvalidK8sConfiguration("Cannot set both token and user/pass")
1432             attrs["Token"] = token
1433
1434         auth_type = None
1435         if key:
1436             auth_type = "oauth2"
1437             if client_cert_data:
1438                 auth_type = "oauth2withcert"
1439             if not token:
1440                 raise JujuInvalidK8sConfiguration(
1441                     "missing token for auth type {}".format(auth_type)
1442                 )
1443         elif username:
1444             if not password:
1445                 self.log.debug(
1446                     "credential for user {} has empty password".format(username)
1447                 )
1448             attrs["username"] = username
1449             attrs["password"] = password
1450             if client_cert_data:
1451                 auth_type = "userpasswithcert"
1452             else:
1453                 auth_type = "userpass"
1454         elif client_cert_data and token:
1455             auth_type = "certificate"
1456         else:
1457             raise JujuInvalidK8sConfiguration("authentication method not supported")
1458         return client.CloudCredential(auth_type=auth_type, attrs=attrs)
1459
1460     async def add_cloud(
1461         self,
1462         name: str,
1463         cloud: Cloud,
1464         credential: CloudCredential = None,
1465         credential_name: str = None,
1466     ) -> Cloud:
1467         """
1468         Add cloud to the controller
1469
1470         :param: name:               Name of the cloud to be added
1471         :param: cloud:              Cloud object
1472         :param: credential:         CloudCredentials object for the cloud
1473         :param: credential_name:    Credential name.
1474                                     If not defined, cloud of the name will be used.
1475         """
1476         controller = await self.get_controller()
1477         try:
1478             _ = await controller.add_cloud(name, cloud)
1479             if credential:
1480                 await controller.add_credential(
1481                     credential_name or name, credential=credential, cloud=name
1482                 )
1483             # Need to return the object returned by the controller.add_cloud() function
1484             # I'm returning the original value now until this bug is fixed:
1485             #   https://github.com/juju/python-libjuju/issues/443
1486             return cloud
1487         finally:
1488             await self.disconnect_controller(controller)
1489
1490     async def remove_cloud(self, name: str):
1491         """
1492         Remove cloud
1493
1494         :param: name:   Name of the cloud to be removed
1495         """
1496         controller = await self.get_controller()
1497         try:
1498             await controller.remove_cloud(name)
1499         except juju.errors.JujuError as e:
1500             if len(e.errors) == 1 and f'cloud "{name}" not found' == e.errors[0]:
1501                 self.log.warning(f"Cloud {name} not found, so it could not be deleted.")
1502             else:
1503                 raise e
1504         finally:
1505             await self.disconnect_controller(controller)
1506
1507     @retry(attempts=20, delay=5, fallback=JujuLeaderUnitNotFound())
1508     async def _get_leader_unit(self, application: Application) -> Unit:
1509         unit = None
1510         for u in application.units:
1511             if await u.is_leader_from_status():
1512                 unit = u
1513                 break
1514         if not unit:
1515             raise Exception()
1516         return unit
1517
1518     async def get_cloud_credentials(self, cloud: Cloud) -> typing.List:
1519         """
1520         Get cloud credentials
1521
1522         :param: cloud: Cloud object. The returned credentials will be from this cloud.
1523
1524         :return: List of credentials object associated to the specified cloud
1525
1526         """
1527         controller = await self.get_controller()
1528         try:
1529             facade = client.CloudFacade.from_connection(controller.connection())
1530             cloud_cred_tag = tag.credential(
1531                 cloud.name, self.vca_connection.data.user, cloud.credential_name
1532             )
1533             params = [client.Entity(cloud_cred_tag)]
1534             return (await facade.Credential(params)).results
1535         finally:
1536             await self.disconnect_controller(controller)
1537
1538     async def check_application_exists(self, model_name, application_name) -> bool:
1539         """Check application exists
1540
1541         :param: model_name:         Model Name
1542         :param: application_name:   Application Name
1543
1544         :return: Boolean
1545         """
1546
1547         model = None
1548         controller = await self.get_controller()
1549         try:
1550             model = await self.get_model(controller, model_name)
1551             self.log.debug(
1552                 "Checking if application {} exists in model {}".format(
1553                     application_name, model_name
1554                 )
1555             )
1556             return self._get_application(model, application_name) is not None
1557         finally:
1558             if model:
1559                 await self.disconnect_model(model)
1560             await self.disconnect_controller(controller)