X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Flibjuju.py;h=7a730335553392c5041706ce28e6303155bff94f;hp=a457309cf61de5c4235cb5760e77f237ca5478f8;hb=b0a8f409e149715bf37d30c414474888c8a499f3;hpb=5ef42a14f766aa00a5917845ba1b3e04cc3c15f7 diff --git a/n2vc/libjuju.py b/n2vc/libjuju.py index a457309..7a73033 100644 --- a/n2vc/libjuju.py +++ b/n2vc/libjuju.py @@ -14,8 +14,7 @@ import asyncio import logging -from juju.controller import Controller -from juju.client import client + import time from juju.errors import JujuAPIError @@ -29,6 +28,10 @@ from juju.client._definitions import ( Cloud, CloudCredential, ) +from juju.controller import Controller +from juju.client import client +from juju import tag + from n2vc.juju_watcher import JujuModelWatcher from n2vc.provisioner import AsyncSSHProvisioner from n2vc.n2vc_conn import N2VCConnector @@ -37,7 +40,6 @@ from n2vc.exceptions import ( JujuApplicationNotFound, JujuLeaderUnitNotFound, JujuActionNotFound, - JujuModelAlreadyExists, JujuControllerFailedConnecting, JujuApplicationExists, JujuInvalidK8sConfiguration, @@ -46,6 +48,8 @@ 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: def __init__( @@ -81,9 +85,12 @@ class Libjuju: self.log = log or logging.getLogger("Libjuju") self.db = db db_endpoints = self._get_api_endpoints_db() - self.endpoints = db_endpoints or [endpoint] - if db_endpoints is None: + 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 @@ -101,12 +108,14 @@ class Libjuju: 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 @@ -159,7 +168,8 @@ 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, credential_name=None): """ @@ -175,22 +185,14 @@ class Libjuju: 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, @@ -198,7 +200,6 @@ class Libjuju: cloud_name=cloud_name, credential_name=credential_name or cloud_name, ) - self.models.add(model_name) finally: if model: await self.disconnect_model(model) @@ -443,6 +444,7 @@ class Libjuju: nonce=params.nonce, machine_id=machine_id, proxy=self.api_proxy, + series=params.series, ) ) @@ -485,6 +487,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, @@ -625,7 +649,8 @@ 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") @@ -715,7 +740,8 @@ class Libjuju: try: # Get application application = self._get_application( - model, application_name=application_name, + model, + application_name=application_name, ) # Return list of actions @@ -747,7 +773,10 @@ class Libjuju: return metrics async def add_relation( - self, model_name: str, endpoint_1: str, endpoint_2: str, + self, + model_name: str, + endpoint_1: str, + endpoint_2: str, ): """Add relation @@ -782,7 +811,9 @@ class Libjuju: await self.disconnect_controller(controller) async def consume( - self, offer_url: str, model_name: str, + self, + offer_url: str, + model_name: str, ): """ Adds a remote offer to the model. Relations can be created later using "juju relate". @@ -812,18 +843,22 @@ 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 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, force=True, max_wait=0) # Wait until model is destroyed @@ -843,6 +878,10 @@ 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) @@ -864,32 +903,22 @@ 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 - # ): - # """ - # 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 _destroy_pending_machines(self, model: Model, only_manual: bool = False): + """ + Destroy pending machines in a given model + + :param: only_manual: Bool that indicates only manually provisioned + machines should be destroyed (if True), or that + all pending machines should be destroyed + """ + 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) async def configure_application( self, model_name: str, application_name: str, config: dict = None @@ -903,15 +932,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]: @@ -947,7 +979,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 @@ -974,6 +1007,7 @@ class Libjuju: :param: interval: Time in seconds between checks """ + controller = None while True: try: controller = await self.get_controller() @@ -1018,6 +1052,9 @@ class Libjuju: 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, @@ -1041,17 +1078,17 @@ class Libjuju: raise Exception("configuration must be provided") endpoint = configuration.host - credential = self.get_k8s_cloud_credential(configuration) - ca_certificates = ( - [credential.attrs["ClientCertificateData"]] - if "ClientCertificateData" in credential.attrs - else [] + 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=ca_certificates, + ca_certificates=[client_cert_data], config={ "operator-storage": storage_class, "workload-storage": storage_class, @@ -1063,30 +1100,21 @@ class Libjuju: ) def get_k8s_cloud_credential( - self, configuration: Configuration, + self, + configuration: Configuration, + client_cert_data: str, + token: str = None, ) -> client.CloudCredential: attrs = {} - ca_cert = configuration.ssl_ca_cert or configuration.cert_file - key = configuration.key_file - api_key = configuration.api_key - token = None + # TODO: Test with AKS + key = None # open(configuration.key_file, "r").read() username = configuration.username password = configuration.password - if "authorization" in api_key: - authorization = api_key["authorization"] - if "Bearer " in authorization: - bearer_list = authorization.split(" ") - if len(bearer_list) == 2: - [_, token] = bearer_list - else: - raise JujuInvalidK8sConfiguration("unknown format of api_key") - else: - token = authorization - if ca_cert: - attrs["ClientCertificateData"] = open(ca_cert, "r").read() + if client_cert_data: + attrs["ClientCertificateData"] = client_cert_data if key: - attrs["ClientKeyData"] = open(key, "r").read() + attrs["ClientKeyData"] = key if token: if username or password: raise JujuInvalidK8sConfiguration("Cannot set both token and user/pass") @@ -1095,6 +1123,8 @@ class Libjuju: 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) @@ -1106,15 +1136,15 @@ class Libjuju: ) attrs["username"] = username attrs["password"] = password - if ca_cert: + if client_cert_data: auth_type = "userpasswithcert" else: auth_type = "userpass" - elif ca_cert and token: + 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,) + return client.CloudCredential(auth_type=auth_type, attrs=attrs) async def add_cloud( self, @@ -1165,3 +1195,13 @@ class Libjuju: 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)