X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Flibjuju.py;h=6580067c1a3070d0fab992e45eca1118519b4be9;hp=b897bc6fd26d7b6c7b2d9eedc4a2835140c97e9a;hb=HEAD;hpb=856d33b9faace0c9c031b7ab3b377c4ad53888ab diff --git a/n2vc/libjuju.py b/n2vc/libjuju.py index b897bc6..f36ff39 100644 --- a/n2vc/libjuju.py +++ b/n2vc/libjuju.py @@ -14,15 +14,20 @@ import asyncio import logging +import os import typing +import yaml import time import juju.errors +from juju.bundle import BundleHandler from juju.model import Model from juju.machine import Machine from juju.application import Application from juju.unit import Unit +from juju.url import URL +from juju.version import DEFAULT_ARCHITECTURE from juju.client._definitions import ( FullStatus, QueryApplicationOffersResults, @@ -33,6 +38,7 @@ from juju.controller import Controller from juju.client import client from juju import tag +from n2vc.definitions import Offer, RelationEndpoint from n2vc.juju_watcher import JujuModelWatcher from n2vc.provisioner import AsyncSSHProvisioner from n2vc.n2vc_conn import N2VCConnector @@ -55,11 +61,18 @@ from retrying_async import retry RBAC_LABEL_KEY_NAME = "rbac-id" +@asyncio.coroutine +def retry_callback(attempt, exc, args, kwargs, delay=0.5, *, loop): + # Specifically overridden from upstream implementation so it can + # continue to work with Python 3.10 + yield from asyncio.sleep(attempt * delay) + return retry + + class Libjuju: def __init__( self, vca_connection: Connection, - loop: asyncio.AbstractEventLoop = None, log: logging.Logger = None, n2vc: N2VCConnector = None, ): @@ -67,7 +80,6 @@ class Libjuju: Constructor :param: vca_connection: n2vc.vca.connection object - :param: loop: Asyncio loop :param: log: Logger :param: n2vc: N2VC object """ @@ -76,15 +88,13 @@ class Libjuju: self.n2vc = n2vc self.vca_connection = vca_connection - self.loop = loop or asyncio.get_event_loop() - self.loop.set_exception_handler(self.handle_exception) - self.creating_model = asyncio.Lock(loop=self.loop) + self.creating_model = asyncio.Lock() if self.vca_connection.is_default: self.health_check_task = self._create_health_check_task() def _create_health_check_task(self): - return self.loop.create_task(self.health_check()) + return asyncio.get_event_loop().create_task(self.health_check()) async def get_controller(self, timeout: float = 60.0) -> Controller: """ @@ -94,7 +104,7 @@ class Libjuju: """ controller = None try: - controller = Controller(loop=self.loop) + controller = Controller() await asyncio.wait_for( controller.connect( endpoint=self.vca_connection.data.endpoints, @@ -121,7 +131,10 @@ class Libjuju: ) if controller: await self.disconnect_controller(controller) - raise JujuControllerFailedConnecting(e) + + raise JujuControllerFailedConnecting( + f"Error connecting to Juju controller: {e}" + ) async def disconnect(self): """Disconnect""" @@ -146,7 +159,7 @@ class Libjuju: if controller: await controller.disconnect() - @retry(attempts=3, delay=5, timeout=None) + @retry(attempts=3, delay=5, timeout=None, callback=retry_callback) async def add_model(self, model_name: str, cloud: VcaCloud): """ Create model @@ -261,7 +274,7 @@ class Libjuju: await self.disconnect_controller(controller) return application_configs - @retry(attempts=3, delay=5) + @retry(attempts=3, delay=5, callback=retry_callback) async def get_model(self, controller: Controller, model_name: str) -> Model: """ Get model from controller @@ -545,27 +558,122 @@ class Libjuju: return machine_id async def deploy( - self, uri: str, model_name: str, wait: bool = True, timeout: float = 3600 + self, + uri: str, + model_name: str, + wait: bool = True, + timeout: float = 3600, + instantiation_params: dict = None, ): """ Deploy bundle or charm: Similar to the juju CLI command `juju deploy` - :param: uri: Path or Charm Store uri in which the charm or bundle can be found - :param: model_name: Model name - :param: wait: Indicates whether to wait or not until all applications are active - :param: timeout: Time in seconds to wait until all applications are active + :param uri: Path or Charm Store uri in which the charm or bundle can be found + :param model_name: Model name + :param wait: Indicates whether to wait or not until all applications are active + :param timeout: Time in seconds to wait until all applications are active + :param instantiation_params: To be applied as overlay bundle over primary bundle. """ controller = await self.get_controller() model = await self.get_model(controller, model_name) + overlays = [] try: - await model.deploy(uri, trust=True) + await self._validate_instantiation_params(uri, model, instantiation_params) + overlays = self._get_overlays(model_name, instantiation_params) + await model.deploy(uri, trust=True, overlays=overlays) if wait: await JujuModelWatcher.wait_for_model(model, timeout=timeout) self.log.debug("All units active in model {}".format(model_name)) finally: + self._remove_overlay_file(overlays) await self.disconnect_model(model) await self.disconnect_controller(controller) + async def _validate_instantiation_params( + self, uri: str, model, instantiation_params: dict + ) -> None: + """Checks if all the applications in instantiation_params + exist ins the original bundle. + + Raises: + JujuApplicationNotFound if there is an invalid app in + the instantiation params. + """ + overlay_apps = self._get_apps_in_instantiation_params(instantiation_params) + if not overlay_apps: + return + original_apps = await self._get_apps_in_original_bundle(uri, model) + if not all(app in original_apps for app in overlay_apps): + raise JujuApplicationNotFound( + "Cannot find application {} in original bundle {}".format( + overlay_apps, original_apps + ) + ) + + async def _get_apps_in_original_bundle(self, uri: str, model) -> set: + """Bundle is downloaded in BundleHandler.fetch_plan. + That method takes care of opening and exception handling. + + Resolve method gets all the information regarding the channel, + track, revision, type, source. + + Returns: + Set with the names of the applications in original bundle. + """ + url = URL.parse(uri) + architecture = DEFAULT_ARCHITECTURE # only AMD64 is allowed + res = await model.deploy_types[str(url.schema)].resolve( + url, architecture, entity_url=uri + ) + handler = BundleHandler(model, trusted=True, forced=False) + await handler.fetch_plan(url, res.origin) + return handler.applications + + def _get_apps_in_instantiation_params(self, instantiation_params: dict) -> list: + """Extract applications key in instantiation params. + + Returns: + List with the names of the applications in instantiation params. + + Raises: + JujuError if applications key is not found. + """ + if not instantiation_params: + return [] + try: + return [key for key in instantiation_params.get("applications")] + except Exception as e: + raise JujuError("Invalid overlay format. {}".format(str(e))) + + def _get_overlays(self, model_name: str, instantiation_params: dict) -> list: + """Creates a temporary overlay file which includes the instantiation params. + Only one overlay file is created. + + Returns: + List with one overlay filename. Empty list if there are no instantiation params. + """ + if not instantiation_params: + return [] + file_name = model_name + "-overlay.yaml" + self._write_overlay_file(file_name, instantiation_params) + return [file_name] + + def _write_overlay_file(self, file_name: str, instantiation_params: dict) -> None: + with open(file_name, "w") as file: + yaml.dump(instantiation_params, file) + + def _remove_overlay_file(self, overlay: list) -> None: + """Overlay contains either one or zero file names.""" + if not overlay: + return + try: + filename = overlay[0] + os.remove(filename) + except OSError as e: + self.log.warning( + "Overlay file {} could not be removed: {}".format(filename, e) + ) + async def add_unit( self, application_name: str, @@ -594,7 +702,6 @@ class Libjuju: application = self._get_application(model, application_name) if application is not None: - # Checks if the given machine id in the model, # otherwise function raises an error _machine, _series = self._get_machine_info(model, machine_id) @@ -749,7 +856,6 @@ class Libjuju: try: if application_name not in model.applications: - if machine_id is not None: machine, series = self._get_machine_info(model, machine_id) @@ -791,12 +897,164 @@ class Libjuju: raise JujuApplicationExists( "Application {} exists".format(application_name) ) + except juju.errors.JujuError as e: + if "already exists" in e.message: + raise JujuApplicationExists( + "Application {} exists".format(application_name) + ) + else: + raise e + finally: + await self.disconnect_model(model) + await self.disconnect_controller(controller) + + return application + + async def upgrade_charm( + self, + application_name: str, + path: str, + model_name: str, + total_timeout: float = None, + **kwargs, + ): + """Upgrade Charm + + :param: application_name: Application name + :param: model_name: Model name + :param: path: Local path to the charm + :param: total_timeout: Timeout for the entity to be active + + :return: (str, str): (output and status) + """ + + self.log.debug( + "Upgrading charm {} in model {} from path {}".format( + application_name, model_name, path + ) + ) + + await self.resolve_application( + model_name=model_name, application_name=application_name + ) + + # Get controller + controller = await self.get_controller() + + # Get model + model = await self.get_model(controller, model_name) + + try: + # Get application + application = self._get_application( + model, + application_name=application_name, + ) + if application is None: + raise JujuApplicationNotFound( + "Cannot find application {} to upgrade".format(application_name) + ) + + await application.refresh(path=path) + + self.log.debug( + "Wait until charm upgrade is completed for application {} (model={})".format( + application_name, model_name + ) + ) + + await JujuModelWatcher.ensure_units_idle( + model=model, application=application + ) + + if application.status == "error": + error_message = "Unknown" + for unit in application.units: + if ( + unit.workload_status == "error" + and unit.workload_status_message != "" + ): + error_message = unit.workload_status_message + + message = "Application {} failed update in {}: {}".format( + application_name, model_name, error_message + ) + self.log.error(message) + raise JujuError(message=message) + + self.log.debug( + "Application {} is ready in model {}".format( + application_name, model_name + ) + ) + finally: await self.disconnect_model(model) await self.disconnect_controller(controller) return application + async def resolve_application(self, model_name: str, application_name: str): + controller = await self.get_controller() + model = await self.get_model(controller, model_name) + + try: + application = self._get_application( + model, + application_name=application_name, + ) + if application is None: + raise JujuApplicationNotFound( + "Cannot find application {} to resolve".format(application_name) + ) + + while application.status == "error": + for unit in application.units: + if unit.workload_status == "error": + self.log.debug( + "Model {}, Application {}, Unit {} in error state, resolving".format( + model_name, application_name, unit.entity_id + ) + ) + try: + await unit.resolved(retry=False) + except Exception: + pass + + await asyncio.sleep(1) + + finally: + await self.disconnect_model(model) + await self.disconnect_controller(controller) + + async def resolve(self, model_name: str): + controller = await self.get_controller() + model = await self.get_model(controller, model_name) + all_units_active = False + try: + while not all_units_active: + all_units_active = True + for application_name, application in model.applications.items(): + if application.status == "error": + for unit in application.units: + if unit.workload_status == "error": + self.log.debug( + "Model {}, Application {}, Unit {} in error state, resolving".format( + model_name, application_name, unit.entity_id + ) + ) + try: + await unit.resolved(retry=False) + all_units_active = False + except Exception: + pass + + if not all_units_active: + await asyncio.sleep(5) + finally: + await self.disconnect_model(model) + await self.disconnect_controller(controller) + async def scale_application( self, model_name: str, @@ -1109,10 +1367,10 @@ class Libjuju: try: await model.add_relation(endpoint_1, endpoint_2) except juju.errors.JujuAPIError as e: - if "not found" in e.message: + if self._relation_is_not_found(e): self.log.warning("Relation not found: {}".format(e.message)) return - if "already exists" in e.message: + if self._relation_already_exist(e): self.log.warning("Relation already exists: {}".format(e.message)) return # another exception, raise it @@ -1121,28 +1379,83 @@ class Libjuju: await self.disconnect_model(model) await self.disconnect_controller(controller) + def _relation_is_not_found(self, juju_error): + text = "not found" + return (text in juju_error.message) or ( + juju_error.error_code and text in juju_error.error_code + ) + + def _relation_already_exist(self, juju_error): + text = "already exists" + return (text in juju_error.message) or ( + juju_error.error_code and text in juju_error.error_code + ) + + async def offer(self, endpoint: RelationEndpoint) -> Offer: + """ + Create an offer from a RelationEndpoint + + :param: endpoint: Relation endpoint + + :return: Offer object + """ + model_name = endpoint.model_name + offer_name = f"{endpoint.application_name}-{endpoint.endpoint_name}" + controller = await self.get_controller() + model = None + try: + model = await self.get_model(controller, model_name) + await model.create_offer(endpoint.endpoint, offer_name=offer_name) + offer_list = await self._list_offers(model_name, offer_name=offer_name) + if offer_list: + return Offer(offer_list[0].offer_url) + else: + raise Exception("offer was not created") + except juju.errors.JujuError as e: + if "application offer already exists" not in e.message: + raise e + finally: + if model: + self.disconnect_model(model) + self.disconnect_controller(controller) + async def consume( self, - offer_url: str, model_name: str, - ): + offer: Offer, + provider_libjuju: "Libjuju", + ) -> str: """ - Adds a remote offer to the model. Relations can be created later using "juju relate". + Consumes a remote offer in the model. Relations can be created later using "juju relate". - :param: offer_url: Offer Url - :param: model_name: Model name + :param: model_name: Model name + :param: offer: Offer object to consume + :param: provider_libjuju: Libjuju object of the provider endpoint :raises ParseError if there's a problem parsing the offer_url :raises JujuError if remote offer includes and endpoint :raises JujuAPIError if the operation is not successful + + :returns: Saas name. It is the application name in the model that reference the remote application. """ + saas_name = f'{offer.name}-{offer.model_name.replace("-", "")}' + if offer.vca_id: + saas_name = f"{saas_name}-{offer.vca_id}" controller = await self.get_controller() - model = await controller.get_model(model_name) - + model = None + provider_controller = None try: - await model.consume(offer_url) + model = await controller.get_model(model_name) + provider_controller = await provider_libjuju.get_controller() + await model.consume( + offer.url, application_alias=saas_name, controller=provider_controller + ) + return saas_name finally: - await self.disconnect_model(model) + if model: + await self.disconnect_model(model) + if provider_controller: + await provider_libjuju.disconnect_controller(provider_controller) await self.disconnect_controller(controller) async def destroy_model(self, model_name: str, total_timeout: float = 1800): @@ -1157,28 +1470,38 @@ class Libjuju: model = None try: if not await self.model_exists(model_name, controller=controller): + self.log.warn(f"Model {model_name} doesn't exist") return - self.log.debug("Destroying model {}".format(model_name)) - + self.log.debug(f"Getting model {model_name} to be destroyed") model = await self.get_model(controller, model_name) + self.log.debug(f"Destroying manual machines in model {model_name}") # Destroy machines that are manually provisioned # and still are in pending state await self._destroy_pending_machines(model, only_manual=True) await self.disconnect_model(model) - await self._destroy_model( - model_name, - controller, + await asyncio.wait_for( + self._destroy_model(model_name, controller), timeout=total_timeout, ) + except Exception as e: + if not await self.model_exists(model_name, controller=controller): + self.log.warn( + f"Failed deleting model {model_name}: model doesn't exist" + ) + return + self.log.warn(f"Failed deleting model {model_name}: {e}") + raise e finally: if model: await self.disconnect_model(model) await self.disconnect_controller(controller) async def _destroy_model( - self, model_name: str, controller: Controller, timeout: float = 1800 + self, + model_name: str, + controller: Controller, ): """ Destroy model from controller @@ -1187,22 +1510,42 @@ class Libjuju: :param: controller: Controller object :param: timeout: Timeout in seconds """ + self.log.debug(f"Destroying model {model_name}") - async def _destroy_model_loop(model_name: str, controller: Controller): - while await self.model_exists(model_name, controller=controller): + async def _destroy_model_gracefully(model_name: str, controller: Controller): + self.log.info(f"Gracefully deleting model {model_name}") + resolved = False + while model_name in await controller.list_models(): + if not resolved: + await self.resolve(model_name) + resolved = True + await controller.destroy_model(model_name, destroy_storage=True) + + await asyncio.sleep(5) + self.log.info(f"Model {model_name} deleted gracefully") + + async def _destroy_model_forcefully(model_name: str, controller: Controller): + self.log.info(f"Forcefully deleting model {model_name}") + while model_name in await controller.list_models(): await controller.destroy_model( - model_name, destroy_storage=True, force=True, max_wait=0 + model_name, destroy_storage=True, force=True, max_wait=60 ) await asyncio.sleep(5) + self.log.info(f"Model {model_name} deleted forcefully") try: - await asyncio.wait_for( - _destroy_model_loop(model_name, controller), timeout=timeout - ) - except asyncio.TimeoutError: - raise Exception( - "Timeout waiting for model {} to be destroyed".format(model_name) - ) + try: + await asyncio.wait_for( + _destroy_model_gracefully(model_name, controller), timeout=120 + ) + except asyncio.TimeoutError: + await _destroy_model_forcefully(model_name, controller) + except juju.errors.JujuError as e: + if any("has been removed" in error for error in e.errors): + return + if any("model not found" in error for error in e.errors): + return + raise e async def destroy_application( self, model_name: str, application_name: str, total_timeout: float @@ -1301,10 +1644,6 @@ class Libjuju: await self.disconnect_model(model) await self.disconnect_controller(controller) - def handle_exception(self, loop, context): - # All unhandled exceptions by libjuju are handled here. - pass - async def health_check(self, interval: float = 300.0): """ Health check to make sure controller and controller_model connections are OK @@ -1339,17 +1678,29 @@ class Libjuju: finally: await self.disconnect_controller(controller) - async def list_offers(self, model_name: str) -> QueryApplicationOffersResults: - """List models with certain names + async def _list_offers( + self, model_name: str, offer_name: str = None + ) -> QueryApplicationOffersResults: + """ + List offers within a model :param: model_name: Model name + :param: offer_name: Offer name to filter. - :return: Returns list of offers + :return: Returns application offers results in the model """ controller = await self.get_controller() try: - return await controller.list_offers(model_name) + offers = (await controller.list_offers(model_name)).results + if offer_name: + matching_offer = [] + for offer in offers: + if offer.offer_name == offer_name: + matching_offer.append(offer) + break + offers = matching_offer + return offers finally: await self.disconnect_controller(controller) @@ -1497,7 +1848,9 @@ class Libjuju: finally: await self.disconnect_controller(controller) - @retry(attempts=20, delay=5, fallback=JujuLeaderUnitNotFound()) + @retry( + attempts=20, delay=5, fallback=JujuLeaderUnitNotFound(), callback=retry_callback + ) async def _get_leader_unit(self, application: Application) -> Unit: unit = None for u in application.units: