X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Flibjuju.py;h=98e31d8518d448fac634165bc3fb52b0c73df8fa;hp=0bd917d4aa90898f7fbe923770fee7da9be7a40a;hb=a71d4a04c1e8ad3ffe1a129024e6dbc14d6d3bd5;hpb=2f66c4df28308b5b2747bb207e7797932e502436;ds=sidebyside diff --git a/n2vc/libjuju.py b/n2vc/libjuju.py index 0bd917d..98e31d8 100644 --- a/n2vc/libjuju.py +++ b/n2vc/libjuju.py @@ -14,27 +14,42 @@ import asyncio import logging -from juju.controller import Controller -from juju.client import client + import time from juju.errors import JujuAPIError from juju.model import Model from juju.machine import Machine from juju.application import Application -from juju.client._definitions import FullStatus +from juju.unit import Unit +from juju.client._definitions import ( + FullStatus, + QueryApplicationOffersResults, + Cloud, + CloudCredential, +) +from juju.controller import Controller +from juju.client import client +from juju import tag + +from n2vc.config import ModelConfig from n2vc.juju_watcher import JujuModelWatcher from n2vc.provisioner import AsyncSSHProvisioner from n2vc.n2vc_conn import N2VCConnector from n2vc.exceptions import ( JujuMachineNotFound, JujuApplicationNotFound, - JujuModelAlreadyExists, + JujuLeaderUnitNotFound, + JujuActionNotFound, JujuControllerFailedConnecting, JujuApplicationExists, + JujuInvalidK8sConfiguration, ) from n2vc.utils import DB_DATA from osm_common.dbbase import DbException +from kubernetes.client.configuration import Configuration + +RBAC_LABEL_KEY_NAME = "rbac-id" class Libjuju: @@ -49,8 +64,7 @@ class Libjuju: log: logging.Logger = None, db: dict = None, n2vc: N2VCConnector = None, - apt_mirror: str = None, - enable_os_upgrade: bool = True, + model_config: ModelConfig = {}, ): """ Constructor @@ -70,7 +84,13 @@ class Libjuju: self.log = log or logging.getLogger("Libjuju") self.db = db - self.endpoints = self._get_api_endpoints_db() or [endpoint] + 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 @@ -79,21 +99,19 @@ class Libjuju: self.n2vc = n2vc # 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.model_config = model_config 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.loop.create_task(self.health_check()) + self.health_check_task = self._create_health_check_task() - async def get_controller(self, timeout: float = 5.0) -> Controller: + def _create_health_check_task(self): + return self.loop.create_task(self.health_check()) + + async def get_controller(self, timeout: float = 15.0) -> Controller: """ Get controller @@ -146,44 +164,38 @@ class Libjuju: :param: controller: Controller that will be disconnected """ - await controller.disconnect() + if controller: + await controller.disconnect() - async def add_model(self, model_name: str, cloud_name: str): + async def add_model(self, model_name: str, cloud_name: str, credential_name=None): """ 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. """ # 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=cloud_name, + credential_name=credential_name or cloud_name, ) - self.models.add(model_name) finally: if model: await self.disconnect_model(model) @@ -227,6 +239,30 @@ class Libjuju: if need_to_disconnect: await self.disconnect_controller(controller) + async def models_exist(self, model_names: [str]) -> (bool, list): + """ + Check if models exists + + :param: model_names: List of strings with model names + + :return (bool, list[str]): (True if all models exists, List of model names that don't exist) + """ + if not model_names: + raise Exception( + "model_names must be a non-empty array. Given value: {}".format( + model_names + ) + ) + non_existing_models = [] + models = await self.list_models() + existing_models = list(set(models).intersection(model_names)) + non_existing_models = list(set(model_names) - set(existing_models)) + + return ( + len(non_existing_models) == 0, + non_existing_models, + ) + async def get_model_status(self, model_name: str) -> FullStatus: """ Get model status @@ -251,6 +287,7 @@ class Libjuju: progress_timeout: float = None, total_timeout: float = None, series: str = "xenial", + wait: bool = True, ) -> (Machine, bool): """ Create machine @@ -260,6 +297,8 @@ class Libjuju: :param: db_dict: Dictionary with data of the DB to write the updates :param: progress_timeout: Maximum time between two updates in the model :param: total_timeout: Timeout for the entity to be active + :param: series: Series of the machine (xenial, bionic, focal, ...) + :param: wait: Wait until machine is ready :return: (juju.machine.Machine, bool): Machine object and a boolean saying if the machine is new or it already existed @@ -292,7 +331,7 @@ class Libjuju: machine_id, model_name ) ) - machine = model.machines[machine_id] + machine = machines[machine_id] else: raise JujuMachineNotFound("Machine {} not found".format(machine_id)) @@ -311,14 +350,15 @@ class Libjuju: machine.entity_id, model_name ) ) - await JujuModelWatcher.wait_for( - model=model, - entity=machine, - progress_timeout=progress_timeout, - total_timeout=total_timeout, - db_dict=db_dict, - n2vc=self.n2vc, - ) + if wait: + await JujuModelWatcher.wait_for( + model=model, + entity=machine, + progress_timeout=progress_timeout, + total_timeout=total_timeout, + db_dict=db_dict, + n2vc=self.n2vc, + ) finally: await self.disconnect_model(model) await self.disconnect_controller(controller) @@ -399,7 +439,8 @@ class Libjuju: connection=connection, nonce=params.nonce, machine_id=machine_id, - api=self.api_proxy, + proxy=self.api_proxy, + series=params.series, ) ) @@ -442,6 +483,28 @@ class Libjuju: return machine_id + async def deploy( + self, uri: str, model_name: str, wait: bool = True, timeout: float = 3600 + ): + """ + 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 + """ + controller = await self.get_controller() + model = await self.get_model(controller, model_name) + try: + await model.deploy(uri) + if wait: + await JujuModelWatcher.wait_for_model(model, timeout=timeout) + self.log.debug("All units active in model {}".format(model_name)) + finally: + await self.disconnect_model(model) + await self.disconnect_controller(controller) + async def deploy_charm( self, application_name: str, @@ -453,6 +516,7 @@ class Libjuju: total_timeout: float = None, config: dict = None, series: str = None, + num_units: int = 1, ): """Deploy charm @@ -465,6 +529,7 @@ class Libjuju: :param: total_timeout: Timeout for the entity to be active :param: config: Config for the charm :param: series: Series of the charm + :param: num_units: Number of units :return: (juju.application.Application): Juju application """ @@ -508,6 +573,11 @@ class Libjuju: application_name, model_name ) ) + if num_units > 1: + for _ in range(num_units - 1): + m, _ = await self.create_machine(model_name, wait=False) + await application.add_unit(to=m.entity_id) + await JujuModelWatcher.wait_for( model=model, entity=application, @@ -556,7 +626,6 @@ class Libjuju: :param: application_name: Application name :param: model_name: Model name - :param: cloud_name: Cloud name :param: action_name: Name of the action :param: db_dict: Dictionary with data of the DB to write the updates :param: progress_timeout: Maximum time between two updates in the model @@ -576,23 +645,36 @@ class Libjuju: try: # Get application application = self._get_application( - model, application_name=application_name, + model, + application_name=application_name, ) if application is None: raise JujuApplicationNotFound("Cannot execute action") - # Get unit + # Get leader unit + # Racing condition: + # Ocassionally, self._get_leader_unit() will return None + # 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 u in application.units: - if await u.is_leader_from_status(): - unit = u + 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 Exception("Cannot execute action: leader unit not found") + raise JujuLeaderUnitNotFound( + "Cannot execute action: leader unit not found" + ) actions = await application.get_actions() if action_name not in actions: - raise Exception( + raise JujuActionNotFound( "Action {} not in available actions".format(action_name) ) @@ -623,8 +705,6 @@ class Libjuju: action_name, action.status, application_name, model_name ) ) - except Exception as e: - raise e finally: await self.disconnect_model(model) await self.disconnect_controller(controller) @@ -656,7 +736,8 @@ class Libjuju: try: # Get application application = self._get_application( - model, application_name=application_name, + model, + application_name=application_name, ) # Return list of actions @@ -667,24 +748,41 @@ class Libjuju: await self.disconnect_model(model) await self.disconnect_controller(controller) + async def get_metrics(self, model_name: str, application_name: str) -> dict: + """Get the metrics collected by the VCA. + + :param model_name The name or unique id of the network service + :param application_name The name of the application + """ + if not model_name or not application_name: + raise Exception("model_name and application_name must be non-empty strings") + metrics = {} + controller = await self.get_controller() + model = await self.get_model(controller, model_name) + try: + application = self._get_application(model, application_name) + if application is not None: + metrics = await application.get_metrics() + finally: + self.disconnect_model(model) + self.disconnect_controller(controller) + return metrics + async def add_relation( self, model_name: str, - application_name_1: str, - application_name_2: str, - relation_1: str, - relation_2: str, + endpoint_1: str, + endpoint_2: str, ): """Add relation - :param: model_name: Model name - :param: application_name_1 First application name - :param: application_name_2: Second application name - :param: relation_1: First relation name - :param: relation_2: Second relation name + :param: model_name: Model name + :param: endpoint_1 First endpoint name + ("app:endpoint" format or directly the saas name) + :param: endpoint_2: Second endpoint name (^ same format) """ - self.log.debug("Adding relation: {} -> {}".format(relation_1, relation_2)) + self.log.debug("Adding relation: {} -> {}".format(endpoint_1, endpoint_2)) # Get controller controller = await self.get_controller() @@ -692,13 +790,9 @@ class Libjuju: # Get model model = await self.get_model(controller, model_name) - # Build relation strings - r1 = "{}:{}".format(application_name_1, relation_1) - r2 = "{}:{}".format(application_name_2, relation_2) - # Add relation try: - await model.add_relation(relation1=r1, relation2=r2) + await model.add_relation(endpoint_1, endpoint_2) except JujuAPIError as e: if "not found" in e.message: self.log.warning("Relation not found: {}".format(e.message)) @@ -712,9 +806,31 @@ class Libjuju: await self.disconnect_model(model) await self.disconnect_controller(controller) - async def destroy_model( - self, model_name: str, total_timeout: float, + async def consume( + self, + offer_url: str, + model_name: str, ): + """ + Adds a remote offer to the model. Relations can be created later using "juju relate". + + :param: offer_url: Offer Url + :param: model_name: Model name + + :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 + """ + controller = await self.get_controller() + model = await controller.get_model(model_name) + + try: + await model.consume(offer_url) + finally: + await self.disconnect_model(model) + await self.disconnect_controller(controller) + + async def destroy_model(self, model_name: str, total_timeout: float): """ Destroy model @@ -723,70 +839,45 @@ 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 - # Destroy applications - for application_name in model.applications: - try: - await self.destroy_application( - model, application_name=application_name, - ) - except Exception as e: - self.log.error( - "Error destroying application {} in model {}: {}".format( - application_name, model_name, e - ) - ) - - # Destroy machines - machines = await model.get_machines() - for machine_id in machines: - try: - await self.destroy_machine( - model, machine_id=machine_id, total_timeout=total_timeout, - ) - except asyncio.CancelledError: - raise - except Exception: - pass + # Destroy machines that are manually provisioned + # and still are in pending state + await self._destroy_pending_machines(model, only_manual=True) # 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) + await controller.destroy_model(uuid, force=True, max_wait=0) # Wait until model is destroyed self.log.debug("Waiting for model {} to be destroyed...".format(model_name)) - last_exception = "" if total_timeout is None: total_timeout = 3600 end = time.time() + total_timeout while time.time() < end: - try: - models = await controller.list_models() - if model_name not in models: - self.log.debug( - "The model {} ({}) was destroyed".format(model_name, uuid) - ) - return - except asyncio.CancelledError: - raise - except Exception as e: - last_exception = e + models = await controller.list_models() + if model_name not in models: + self.log.debug( + "The model {} ({}) was destroyed".format(model_name, uuid) + ) + return await asyncio.sleep(5) raise Exception( - "Timeout waiting for model {} to be destroyed {}".format( - model_name, last_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) @@ -808,41 +899,23 @@ class Libjuju: else: self.log.warning("Application not found: {}".format(application_name)) - async def destroy_machine( - self, model: Model, machine_id: str, total_timeout: float = 3600 - ): + async def _destroy_pending_machines(self, model: Model, only_manual: bool = False): """ - Destroy machine + Destroy pending machines in a given model - :param: model: Model object - :param: machine_id: Machine id - :param: total_timeout: Timeout in seconds + :param: only_manual: Bool that indicates only manually provisioned + machines should be destroyed (if True), or that + all pending machines should be destroyed """ - machines = await model.get_machines() - if machine_id in machines: - machine = model.machines[machine_id] - # TODO: change this by machine.is_manual when this is upstreamed: - # https://github.com/juju/python-libjuju/pull/396 - if "instance-id" in machine.safe_data and machine.safe_data[ - "instance-id" - ].startswith("manual:"): + status = await model.get_status() + for machine_id in status.machines: + machine_status = status.machines[machine_id] + if machine_status.agent_status.status == "pending": + if only_manual and not machine_status.instance_id.startswith("manual:"): + break + machine = model.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 ): @@ -855,15 +928,18 @@ class Libjuju: self.log.debug("Configuring application {}".format(application_name)) if config: + controller = await self.get_controller() + model = None try: - controller = await self.get_controller() model = await self.get_model(controller, model_name) application = self._get_application( - model, application_name=application_name, + model, + application_name=application_name, ) await application.set_config(config) finally: - await self.disconnect_model(model) + if model: + await self.disconnect_model(model) await self.disconnect_controller(controller) def _get_api_endpoints_db(self) -> [str]: @@ -899,7 +975,8 @@ class Libjuju: if not juju_info: try: self.db.create( - DB_DATA.api_endpoints.table, DB_DATA.api_endpoints.filter, + DB_DATA.api_endpoints.table, + DB_DATA.api_endpoints.filter, ) except DbException as e: # Racing condition: check if another N2VC worker has created it @@ -926,6 +1003,7 @@ class Libjuju: :param: interval: Time in seconds between checks """ + controller = None while True: try: controller = await self.get_controller() @@ -935,3 +1013,191 @@ class Libjuju: finally: await self.disconnect_controller(controller) await asyncio.sleep(interval) + + async def list_models(self, contains: str = None) -> [str]: + """List models with certain names + + :param: contains: String that is contained in model name + + :retur: [models] Returns list of model names + """ + + controller = await self.get_controller() + try: + models = await controller.list_models() + if contains: + models = [model for model in models if contains in model] + return models + finally: + await self.disconnect_controller(controller) + + async def list_offers(self, model_name: str) -> QueryApplicationOffersResults: + """List models with certain names + + :param: model_name: Model name + + :return: Returns list of offers + """ + + controller = await self.get_controller() + try: + return await controller.list_offers(model_name) + finally: + await self.disconnect_controller(controller) + + async def add_k8s( + self, + name: str, + rbac_id: str, + token: str, + client_cert_data: str, + configuration: Configuration, + storage_class: str, + credential_name: str = None, + ): + """ + Add a Kubernetes cloud to the controller + + Similar to the `juju add-k8s` command in the CLI + + :param: name: Name for the K8s cloud + :param: configuration: Kubernetes configuration object + :param: storage_class: Storage Class to use in the cloud + :param: credential_name: Storage Class to use in the cloud + """ + + if not storage_class: + raise Exception("storage_class must be a non-empty string") + if not name: + raise Exception("name must be a non-empty string") + if not configuration: + raise Exception("configuration must be provided") + + endpoint = configuration.host + credential = self.get_k8s_cloud_credential( + configuration, + client_cert_data, + token, + ) + credential.attrs[RBAC_LABEL_KEY_NAME] = rbac_id + cloud = client.Cloud( + type_="kubernetes", + auth_types=[credential.auth_type], + endpoint=endpoint, + ca_certificates=[client_cert_data], + config={ + "operator-storage": storage_class, + "workload-storage": storage_class, + }, + ) + + return await self.add_cloud( + name, cloud, credential, credential_name=credential_name + ) + + def get_k8s_cloud_credential( + self, + configuration: Configuration, + client_cert_data: str, + token: str = None, + ) -> client.CloudCredential: + attrs = {} + # TODO: Test with AKS + key = None # open(configuration.key_file, "r").read() + username = configuration.username + password = configuration.password + + if client_cert_data: + attrs["ClientCertificateData"] = client_cert_data + if key: + attrs["ClientKeyData"] = key + if token: + if username or password: + raise JujuInvalidK8sConfiguration("Cannot set both token and user/pass") + attrs["Token"] = token + + auth_type = None + if key: + auth_type = "oauth2" + if client_cert_data: + auth_type = "oauth2withcert" + if not token: + raise JujuInvalidK8sConfiguration( + "missing token for auth type {}".format(auth_type) + ) + elif username: + if not password: + self.log.debug( + "credential for user {} has empty password".format(username) + ) + attrs["username"] = username + attrs["password"] = password + if client_cert_data: + auth_type = "userpasswithcert" + else: + auth_type = "userpass" + elif client_cert_data and token: + auth_type = "certificate" + else: + raise JujuInvalidK8sConfiguration("authentication method not supported") + return client.CloudCredential(auth_type=auth_type, attrs=attrs) + + async def add_cloud( + self, + name: str, + cloud: Cloud, + credential: CloudCredential = None, + credential_name: str = None, + ) -> Cloud: + """ + Add cloud to the controller + + :param: name: Name of the cloud to be added + :param: cloud: Cloud object + :param: credential: CloudCredentials object for the cloud + :param: credential_name: Credential name. + If not defined, cloud of the name will be used. + """ + controller = await self.get_controller() + try: + _ = await controller.add_cloud(name, cloud) + if credential: + await controller.add_credential( + credential_name or name, credential=credential, cloud=name + ) + # Need to return the object returned by the controller.add_cloud() function + # I'm returning the original value now until this bug is fixed: + # https://github.com/juju/python-libjuju/issues/443 + return cloud + finally: + await self.disconnect_controller(controller) + + async def remove_cloud(self, name: str): + """ + Remove cloud + + :param: name: Name of the cloud to be removed + """ + controller = await self.get_controller() + try: + await controller.remove_cloud(name) + finally: + await self.disconnect_controller(controller) + + 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 + return unit + + async def get_cloud_credentials(self, cloud_name: str, credential_name: str): + controller = await self.get_controller() + try: + facade = client.CloudFacade.from_connection(controller.connection()) + cloud_cred_tag = tag.credential(cloud_name, self.username, credential_name) + params = [client.Entity(cloud_cred_tag)] + return (await facade.Credential(params)).results + finally: + await self.disconnect_controller(controller)