X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Flibjuju.py;h=a6fd8fe48dd7f299f3a00a0ac30b1c6228c0fa64;hp=a79d00d8c8a46b56da6b1242a02a27612512f8c5;hb=eb8943a887e2fb8cce0240382811f9e504f3c7fb;hpb=c1cf1b9a33dd1929fb834424626661f9c9dce64d;ds=sidebyside diff --git a/n2vc/libjuju.py b/n2vc/libjuju.py index a79d00d..a6fd8fe 100644 --- a/n2vc/libjuju.py +++ b/n2vc/libjuju.py @@ -14,6 +14,7 @@ import asyncio import logging +import typing import time @@ -40,14 +41,16 @@ from n2vc.exceptions import ( JujuApplicationNotFound, JujuLeaderUnitNotFound, JujuActionNotFound, - JujuModelAlreadyExists, JujuControllerFailedConnecting, JujuApplicationExists, JujuInvalidK8sConfiguration, + JujuError, ) -from n2vc.utils import DB_DATA -from osm_common.dbbase import DbException +from n2vc.vca.cloud import Cloud as VcaCloud +from n2vc.vca.connection import Connection from kubernetes.client.configuration import Configuration +from retrying_async import retry + RBAC_LABEL_KEY_NAME = "rbac-id" @@ -55,69 +58,35 @@ RBAC_LABEL_KEY_NAME = "rbac-id" class Libjuju: def __init__( self, - endpoint: str, - api_proxy: str, - username: str, - password: str, - cacert: str, + vca_connection: Connection, loop: asyncio.AbstractEventLoop = None, log: logging.Logger = None, - db: dict = None, n2vc: N2VCConnector = None, - apt_mirror: str = None, - enable_os_upgrade: bool = True, ): """ Constructor - :param: endpoint: Endpoint of the juju controller (host:port) - :param: api_proxy: Endpoint of the juju controller - Reachable from the VNFs - :param: username: Juju username - :param: password: Juju password - :param: cacert: Juju CA Certificate + :param: vca_connection: n2vc.vca.connection object :param: loop: Asyncio loop :param: log: Logger - :param: db: DB object :param: n2vc: N2VC object - :param: apt_mirror: APT Mirror - :param: enable_os_upgrade: Enable OS Upgrade """ self.log = log or logging.getLogger("Libjuju") - self.db = db - db_endpoints = self._get_api_endpoints_db() - self.endpoints = None - if (db_endpoints and endpoint not in db_endpoints) or not db_endpoints: - self.endpoints = [endpoint] - self._update_api_endpoints_db(self.endpoints) - else: - self.endpoints = db_endpoints - self.api_proxy = api_proxy - self.username = username - self.password = password - self.cacert = cacert - self.loop = loop or asyncio.get_event_loop() self.n2vc = n2vc + self.vca_connection = vca_connection - # Generate config for models - self.model_config = {} - if apt_mirror: - self.model_config["apt-mirror"] = apt_mirror - self.model_config["enable-os-refresh-update"] = enable_os_upgrade - self.model_config["enable-os-upgrade"] = enable_os_upgrade - + 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.models = set() - self.log.debug("Libjuju initialized!") - - self.health_check_task = self._create_health_check_task() + 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()) - async def get_controller(self, timeout: float = 5.0) -> Controller: + async def get_controller(self, timeout: float = 60.0) -> Controller: """ Get controller @@ -128,23 +97,27 @@ class Libjuju: controller = Controller(loop=self.loop) await asyncio.wait_for( controller.connect( - endpoint=self.endpoints, - username=self.username, - password=self.password, - cacert=self.cacert, + endpoint=self.vca_connection.data.endpoints, + username=self.vca_connection.data.user, + password=self.vca_connection.data.secret, + cacert=self.vca_connection.data.cacert, ), timeout=timeout, ) - endpoints = await controller.api_endpoints - if self.endpoints != endpoints: - self.endpoints = endpoints - self._update_api_endpoints_db(self.endpoints) + if self.vca_connection.is_default: + endpoints = await controller.api_endpoints + if not all( + endpoint in self.vca_connection.endpoints for endpoint in endpoints + ): + await self.vca_connection.update_endpoints(endpoints) return controller except asyncio.CancelledError as e: raise e except Exception as e: self.log.error( - "Failed connecting to controller: {}...".format(self.endpoints) + "Failed connecting to controller: {}... {}".format( + self.vca_connection.data.endpoints, e + ) ) if controller: await self.disconnect_controller(controller) @@ -173,52 +146,123 @@ class Libjuju: if controller: await controller.disconnect() - async def add_model(self, model_name: str, cloud_name: str, credential_name=None): + @retry(attempts=3, delay=5, timeout=None) + async def add_model(self, model_name: str, cloud: VcaCloud): """ Create model :param: model_name: Model name - :param: cloud_name: Cloud name - :param: credential_name: Credential name to use for adding the model - If not specified, same name as the cloud will be used. + :param: cloud: Cloud object """ # Get controller controller = await self.get_controller() model = None try: - # Raise exception if model already exists - if await self.model_exists(model_name, controller=controller): - raise JujuModelAlreadyExists( - "Model {} already exists.".format(model_name) - ) - # Block until other workers have finished model creation while self.creating_model.locked(): await asyncio.sleep(0.1) - # If the model exists, return it from the controller - if model_name in self.models: - return - # Create the model async with self.creating_model: + if await self.model_exists(model_name, controller=controller): + return self.log.debug("Creating model {}".format(model_name)) model = await controller.add_model( model_name, - config=self.model_config, - cloud_name=cloud_name, - credential_name=credential_name or cloud_name, + config=self.vca_connection.data.model_config, + cloud_name=cloud.name, + credential_name=cloud.credential_name, ) - self.models.add(model_name) + except JujuAPIError as e: + if "already exists" in e.message: + pass + else: + raise e finally: if model: await self.disconnect_model(model) await self.disconnect_controller(controller) - async def get_model( - self, controller: Controller, model_name: str, id=None - ) -> Model: + async def get_executed_actions(self, model_name: str) -> list: + """ + Get executed/history of actions for a model. + + :param: model_name: Model name, str. + :return: List of executed actions for a model. + """ + model = None + executed_actions = [] + controller = await self.get_controller() + try: + model = await self.get_model(controller, model_name) + # Get all unique action names + actions = {} + for application in model.applications: + application_actions = await self.get_actions(application, model_name) + actions.update(application_actions) + # Get status of all actions + for application_action in actions: + app_action_status_list = await model.get_action_status( + name=application_action + ) + for action_id, action_status in app_action_status_list.items(): + executed_action = { + "id": action_id, + "action": application_action, + "status": action_status, + } + # Get action output by id + action_status = await model.get_action_output(executed_action["id"]) + for k, v in action_status.items(): + executed_action[k] = v + executed_actions.append(executed_action) + except Exception as e: + raise JujuError( + "Error in getting executed actions for model: {}. Error: {}".format( + model_name, str(e) + ) + ) + finally: + if model: + await self.disconnect_model(model) + await self.disconnect_controller(controller) + return executed_actions + + async def get_application_configs( + self, model_name: str, application_name: str + ) -> dict: + """ + Get available configs for an application. + + :param: model_name: Model name, str. + :param: application_name: Application name, str. + + :return: A dict which has key - action name, value - action description + """ + model = None + application_configs = {} + controller = await self.get_controller() + try: + model = await self.get_model(controller, model_name) + application = self._get_application( + model, application_name=application_name + ) + application_configs = await application.get_config() + except Exception as e: + raise JujuError( + "Error in getting configs for application: {} in model: {}. Error: {}".format( + application_name, model_name, str(e) + ) + ) + finally: + if model: + await self.disconnect_model(model) + await self.disconnect_controller(controller) + return application_configs + + @retry(attempts=3, delay=5) + async def get_model(self, controller: Controller, model_name: str) -> Model: """ Get model from controller @@ -229,9 +273,7 @@ class Libjuju: """ return await controller.get_model(model_name) - async def model_exists( - self, model_name: str, controller: Controller = None - ) -> bool: + async def model_exists(self, model_name: str, controller: Controller = None) -> bool: """ Check if model exists @@ -373,6 +415,7 @@ class Libjuju: total_timeout=total_timeout, db_dict=db_dict, n2vc=self.n2vc, + vca_id=self.vca_connection._vca_id, ) finally: await self.disconnect_model(model) @@ -454,7 +497,8 @@ class Libjuju: connection=connection, nonce=params.nonce, machine_id=machine_id, - proxy=self.api_proxy, + proxy=self.vca_connection.data.api_proxy, + series=params.series, ) ) @@ -484,6 +528,7 @@ class Libjuju: total_timeout=total_timeout, db_dict=db_dict, n2vc=self.n2vc, + vca_id=self.vca_connection._vca_id, ) except Exception as e: raise e @@ -599,6 +644,7 @@ class Libjuju: total_timeout=total_timeout, db_dict=db_dict, n2vc=self.n2vc, + vca_id=self.vca_connection._vca_id, ) self.log.debug( "Application {} is ready in model {}".format( @@ -671,19 +717,7 @@ class Libjuju: # because the leader elected hook has not been triggered yet. # Therefore, we are doing some retries. If it happens again, # re-open bug 1236 - attempts = 3 - time_between_retries = 10 - unit = None - for _ in range(attempts): - unit = await self._get_leader_unit(application) - if unit is None: - await asyncio.sleep(time_between_retries) - else: - break - if unit is None: - raise JujuLeaderUnitNotFound( - "Cannot execute action: leader unit not found" - ) + unit = await self._get_leader_unit(application) actions = await application.get_actions() @@ -706,6 +740,7 @@ class Libjuju: total_timeout=total_timeout, db_dict=db_dict, n2vc=self.n2vc, + vca_id=self.vca_connection._vca_id, ) output = await model.get_action_output(action_uuid=action.entity_id) @@ -853,8 +888,12 @@ class Libjuju: """ controller = await self.get_controller() - model = await self.get_model(controller, model_name) + model = None try: + if not await self.model_exists(model_name, controller=controller): + return + + model = await self.get_model(controller, model_name) self.log.debug("Destroying model {}".format(model_name)) uuid = model.info.uuid @@ -865,10 +904,6 @@ class Libjuju: # Disconnect model await self.disconnect_model(model) - # Destroy model - if model_name in self.models: - self.models.remove(model_name) - await controller.destroy_model(uuid, force=True, max_wait=0) # Wait until model is destroyed @@ -888,26 +923,66 @@ class Libjuju: raise Exception( "Timeout waiting for model {} to be destroyed".format(model_name) ) + except Exception as e: + if model: + await self.disconnect_model(model) + raise e finally: await self.disconnect_controller(controller) - async def destroy_application(self, model: Model, application_name: str): + async def destroy_application( + self, model_name: str, application_name: str, total_timeout: float + ): """ Destroy application - :param: model: Model object + :param: model_name: Model name :param: application_name: Application name + :param: total_timeout: Timeout """ - self.log.debug( - "Destroying application {} in model {}".format( - application_name, model.info.name + + controller = await self.get_controller() + model = None + + try: + model = await self.get_model(controller, model_name) + self.log.debug( + "Destroying application {} in model {}".format( + application_name, model_name + ) ) - ) - application = model.applications.get(application_name) - if application: - await application.destroy() - else: - self.log.warning("Application not found: {}".format(application_name)) + application = self._get_application(model, application_name) + if application: + await application.destroy() + else: + self.log.warning("Application not found: {}".format(application_name)) + + self.log.debug( + "Waiting for application {} to be destroyed in model {}...".format( + application_name, model_name + ) + ) + if total_timeout is None: + total_timeout = 3600 + end = time.time() + total_timeout + while time.time() < end: + if not self._get_application(model, application_name): + self.log.debug( + "The application {} was destroyed in model {} ".format( + application_name, model_name + ) + ) + return + await asyncio.sleep(5) + raise Exception( + "Timeout waiting for application {} to be destroyed in model {}".format( + application_name, model_name + ) + ) + finally: + if model is not None: + await self.disconnect_model(model) + await self.disconnect_controller(controller) async def _destroy_pending_machines(self, model: Model, only_manual: bool = False): """ @@ -926,33 +1001,6 @@ class Libjuju: machine = model.machines[machine_id] await machine.destroy(force=True) - # async def destroy_machine( - # self, model: Model, machine_id: str, total_timeout: float = 3600 - # ): - # """ - # Destroy machine - - # :param: model: Model object - # :param: machine_id: Machine id - # :param: total_timeout: Timeout in seconds - # """ - # machines = await model.get_machines() - # if machine_id in machines: - # machine = machines[machine_id] - # await machine.destroy(force=True) - # # max timeout - # end = time.time() + total_timeout - - # # wait for machine removal - # machines = await model.get_machines() - # while machine_id in machines and time.time() < end: - # self.log.debug("Waiting for machine {} is destroyed".format(machine_id)) - # await asyncio.sleep(0.5) - # machines = await model.get_machines() - # self.log.debug("Machine destroyed: {}".format(machine_id)) - # else: - # self.log.debug("Machine not found: {}".format(machine_id)) - async def configure_application( self, model_name: str, application_name: str, config: dict = None ): @@ -979,57 +1027,6 @@ class Libjuju: await self.disconnect_model(model) await self.disconnect_controller(controller) - def _get_api_endpoints_db(self) -> [str]: - """ - Get API Endpoints from DB - - :return: List of API endpoints - """ - self.log.debug("Getting endpoints from database") - - juju_info = self.db.get_one( - DB_DATA.api_endpoints.table, - q_filter=DB_DATA.api_endpoints.filter, - fail_on_empty=False, - ) - if juju_info and DB_DATA.api_endpoints.key in juju_info: - return juju_info[DB_DATA.api_endpoints.key] - - def _update_api_endpoints_db(self, endpoints: [str]): - """ - Update API endpoints in Database - - :param: List of endpoints - """ - self.log.debug("Saving endpoints {} in database".format(endpoints)) - - juju_info = self.db.get_one( - DB_DATA.api_endpoints.table, - q_filter=DB_DATA.api_endpoints.filter, - fail_on_empty=False, - ) - # If it doesn't, then create it - if not juju_info: - try: - self.db.create( - DB_DATA.api_endpoints.table, - DB_DATA.api_endpoints.filter, - ) - except DbException as e: - # Racing condition: check if another N2VC worker has created it - juju_info = self.db.get_one( - DB_DATA.api_endpoints.table, - q_filter=DB_DATA.api_endpoints.filter, - fail_on_empty=False, - ) - if not juju_info: - raise e - self.db.set_one( - DB_DATA.api_endpoints.table, - DB_DATA.api_endpoints.filter, - {DB_DATA.api_endpoints.key: endpoints}, - ) - def handle_exception(self, loop, context): # All unhandled exceptions by libjuju are handled here. pass @@ -1221,19 +1218,32 @@ class Libjuju: finally: await self.disconnect_controller(controller) + @retry(attempts=20, delay=5, fallback=JujuLeaderUnitNotFound()) async def _get_leader_unit(self, application: Application) -> Unit: unit = None for u in application.units: if await u.is_leader_from_status(): unit = u break + if not unit: + raise Exception() return unit - async def get_cloud_credentials(self, cloud_name: str, credential_name: str): + async def get_cloud_credentials(self, cloud: Cloud) -> typing.List: + """ + Get cloud credentials + + :param: cloud: Cloud object. The returned credentials will be from this cloud. + + :return: List of credentials object associated to the specified cloud + + """ controller = await self.get_controller() try: facade = client.CloudFacade.from_connection(controller.connection()) - cloud_cred_tag = tag.credential(cloud_name, self.username, credential_name) + cloud_cred_tag = tag.credential( + cloud.name, self.vca_connection.data.user, cloud.credential_name + ) params = [client.Entity(cloud_cred_tag)] return (await facade.Credential(params)).results finally: