import asyncio
import logging
from juju.controller import Controller
-from juju.client.connector import NoConnectionException
from juju.client import client
import time
from juju.model import Model
from juju.machine import Machine
from juju.application import Application
-from juju.client._definitions import FullStatus
+from juju.client._definitions import (
+ FullStatus,
+ QueryApplicationOffersResults,
+ Cloud,
+ CloudCredential,
+)
from n2vc.juju_watcher import JujuModelWatcher
from n2vc.provisioner import AsyncSSHProvisioner
from n2vc.n2vc_conn import N2VCConnector
from n2vc.exceptions import (
JujuMachineNotFound,
JujuApplicationNotFound,
+ JujuLeaderUnitNotFound,
+ JujuActionNotFound,
JujuModelAlreadyExists,
JujuControllerFailedConnecting,
JujuApplicationExists,
+ JujuInvalidK8sConfiguration,
)
+from n2vc.utils import DB_DATA
+from osm_common.dbbase import DbException
+from kubernetes.client.configuration import Configuration
class Libjuju:
:param: enable_os_upgrade: Enable OS Upgrade
"""
- self.endpoints = [endpoint] # TODO: Store and get endpoints from DB
+ 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._update_api_endpoints_db(self.endpoints)
self.api_proxy = api_proxy
self.username = username
self.password = password
self.cacert = cacert
self.loop = loop or asyncio.get_event_loop()
- self.log = log or logging.getLogger("Libjuju")
- self.db = db
self.n2vc = n2vc
# Generate config for models
self.model_config["enable-os-refresh-update"] = enable_os_upgrade
self.model_config["enable-os-upgrade"] = enable_os_upgrade
- self.reconnecting = asyncio.Lock(loop=self.loop)
+ self.loop.set_exception_handler(self.handle_exception)
self.creating_model = asyncio.Lock(loop=self.loop)
self.models = set()
- self.controller = Controller(loop=self.loop)
-
- self.loop.run_until_complete(self.connect())
+ self.log.debug("Libjuju initialized!")
- async def connect(self):
- """Connect to the controller"""
-
- self.log.debug("Connecting from controller")
- await self.controller.connect(
- endpoint=self.endpoints,
- username=self.username,
- password=self.password,
- cacert=self.cacert,
- )
- e = self.controller.connection().endpoint
- self.log.info("Connected to controller: {}".format(e))
+ self.health_check_task = self.loop.create_task(self.health_check())
- async def disconnect(self):
- """Disconnect from controller"""
-
- self.log.debug("Disconnecting from controller")
- await self.controller.disconnect()
- self.log.info("Disconnected from controller")
-
- def controller_connected(self) -> bool:
- """Check if the controller connection is open
-
- :return: bool: True if connected, False if not connected
+ async def get_controller(self, timeout: float = 5.0) -> Controller:
"""
+ Get controller
- is_connected = False
+ :param: timeout: Time in seconds to wait for controller to connect
+ """
+ controller = None
try:
- is_connected = self.controller.connection().is_open
- except NoConnectionException:
- self.log.warning("VCA not connected")
- return is_connected
+ controller = Controller(loop=self.loop)
+ await asyncio.wait_for(
+ controller.connect(
+ endpoint=self.endpoints,
+ username=self.username,
+ password=self.password,
+ cacert=self.cacert,
+ ),
+ timeout=timeout,
+ )
+ endpoints = await controller.api_endpoints
+ if self.endpoints != endpoints:
+ self.endpoints = endpoints
+ self._update_api_endpoints_db(self.endpoints)
+ return controller
+ except asyncio.CancelledError as e:
+ raise e
+ except Exception as e:
+ self.log.error(
+ "Failed connecting to controller: {}...".format(self.endpoints)
+ )
+ if controller:
+ await self.disconnect_controller(controller)
+ raise JujuControllerFailedConnecting(e)
+
+ async def disconnect(self):
+ """Disconnect"""
+ # Cancel health check task
+ self.health_check_task.cancel()
+ self.log.debug("Libjuju disconnected!")
async def disconnect_model(self, model: Model):
"""
:param: model: Model that will be disconnected
"""
- try:
- await model.disconnect()
- except Exception:
- pass
+ await model.disconnect()
- async def _reconnect(
- self,
- retry: bool = False,
- timeout: int = 5,
- time_between_retries: int = 3,
- maximum_retries: int = 0,
- ):
+ async def disconnect_controller(self, controller: Controller):
"""
- Reconnect to the controller
-
- :param: retry: Set it to True to retry if the connection fails
- :param: time_between_retries: Time in seconds between retries
- :param: maximum_retries Maximum retries. If not set, it will retry forever
+ Disconnect controller
- :raises: Exception if cannot connect to the controller
+ :param: controller: Controller that will be disconnected
"""
-
- if self.reconnecting.locked():
- # Return if another function is trying to reconnect
- return
- async with self.reconnecting:
- attempt = 0
- while True:
- try:
- await asyncio.wait_for(self.connect(), timeout=timeout)
- break
- except asyncio.TimeoutError:
- self.log.error("Error reconnecting to controller: Timeout")
- except Exception as e:
- self.log.error("Error reconnecting to controller: {}".format(e))
-
- attempt += 1
- maximum_retries_reached = attempt == maximum_retries
-
- if not retry or maximum_retries_reached:
- raise JujuControllerFailedConnecting("Controller is not connected")
- else:
- await asyncio.sleep(time_between_retries)
+ await controller.disconnect()
async def add_model(self, model_name: str, cloud_name: str):
"""
:param: cloud_name: Cloud name
"""
- # Reconnect to the controller if not connected
- if not self.controller_connected():
- await self._reconnect()
-
- # Raise exception if model already exists
- if await self.model_exists(model_name):
- raise JujuModelAlreadyExists("Model {} already exists.".format(model_name))
+ # 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)
+ # 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 await self.get_model(model_name)
+ # If the model exists, return it from the controller
+ if model_name in self.models:
+ return
- # Create the model
- self.log.debug("Creating model {}".format(model_name))
- async with self.creating_model:
- model = await self.controller.add_model(
- model_name,
- config=self.model_config,
- cloud_name=cloud_name,
- credential_name=cloud_name,
- )
- await self.disconnect_model(model)
- self.models.add(model_name)
+ # Create the model
+ async with self.creating_model:
+ 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,
+ )
+ self.models.add(model_name)
+ finally:
+ if model:
+ await self.disconnect_model(model)
+ await self.disconnect_controller(controller)
- async def get_model(self, model_name: str) -> Model:
+ async def get_model(
+ self, controller: Controller, model_name: str, id=None
+ ) -> Model:
"""
Get model from controller
+ :param: controller: Controller
:param: model_name: Model name
:return: Model: The created Juju model object
"""
+ return await controller.get_model(model_name)
- # Check if controller is connected
- if not self.controller_connected():
- await self._reconnect()
- return await self.controller.get_model(model_name)
-
- async def model_exists(self, model_name: str) -> bool:
+ async def model_exists(
+ self, model_name: str, controller: Controller = None
+ ) -> bool:
"""
Check if model exists
+ :param: controller: Controller
:param: model_name: Model name
:return bool
"""
+ need_to_disconnect = False
+
+ # Get controller if not passed
+ if not controller:
+ controller = await self.get_controller()
+ need_to_disconnect = True
+
+ # Check if model exists
+ try:
+ return model_name in await controller.list_models()
+ finally:
+ if need_to_disconnect:
+ await self.disconnect_controller(controller)
+
+ async def models_exist(self, model_names: [str]) -> (bool, list):
+ """
+ Check if models exists
- # Check if controller is connected
- if not self.controller_connected():
- await self._reconnect()
+ :param: model_names: List of strings with model names
- return model_name in await self.controller.list_models()
+ :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:
"""
:return: Full status object
"""
- model = await self.get_model(model_name)
- status = await model.get_status()
- await self.disconnect_model(model)
- return status
+ controller = await self.get_controller()
+ model = await self.get_model(controller, model_name)
+ try:
+ return await model.get_status()
+ finally:
+ await self.disconnect_model(model)
+ await self.disconnect_controller(controller)
async def create_machine(
self,
progress_timeout: float = None,
total_timeout: float = None,
series: str = "xenial",
+ wait: bool = True,
) -> (Machine, bool):
"""
Create machine
: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
"Creating machine (id={}) in model: {}".format(machine_id, model_name)
)
+ # Get controller
+ controller = await self.get_controller()
+
# Get model
- model = await self.get_model(model_name)
+ model = await self.get_model(controller, model_name)
try:
if machine_id is not None:
self.log.debug(
machine_id, model_name
)
)
- machine = model.machines[machine_id]
+ machine = machines[machine_id]
else:
raise JujuMachineNotFound("Machine {} not found".format(machine_id))
new = True
# Wait until the machine is ready
- await JujuModelWatcher.wait_for(
- model=model,
- entity=machine,
- progress_timeout=progress_timeout,
- total_timeout=total_timeout,
- db_dict=db_dict,
- n2vc=self.n2vc,
+ self.log.debug(
+ "Wait until machine {} is ready in model {}".format(
+ machine.entity_id, model_name
+ )
)
- except Exception as e:
- raise e
+ 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)
- self.log.debug("Machine ready at {}".format(machine.dns_name))
+ self.log.debug(
+ "Machine {} ready at {} in model {}".format(
+ machine.entity_id, machine.dns_name, model_name
+ )
+ )
return machine, new
async def provision_machine(
)
)
+ # Get controller
+ controller = await self.get_controller()
+
# Get model
- model = await self.get_model(model_name)
+ model = await self.get_model(controller, model_name)
try:
# Get provisioner
connection=connection,
nonce=params.nonce,
machine_id=machine_id,
- api=self.api_proxy,
+ proxy=self.api_proxy,
)
)
self.log.error(msg=msg)
raise JujuMachineNotFound(msg)
+ self.log.debug(
+ "Wait until machine {} is ready in model {}".format(
+ machine.entity_id, model_name
+ )
+ )
await JujuModelWatcher.wait_for(
model=model,
entity=machine,
raise e
finally:
await self.disconnect_model(model)
+ await self.disconnect_controller(controller)
- self.log.debug("Machine provisioned {}".format(machine_id))
+ self.log.debug(
+ "Machine provisioned {} in model {}".format(machine_id, model_name)
+ )
return machine_id
total_timeout: float = None,
config: dict = None,
series: str = None,
+ num_units: int = 1,
):
"""Deploy charm
: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
"""
+ self.log.debug(
+ "Deploying charm {} to machine {} in model ~{}".format(
+ application_name, machine_id, model_name
+ )
+ )
+ self.log.debug("charm: {}".format(path))
+
+ # Get controller
+ controller = await self.get_controller()
# Get model
- model = await self.get_model(model_name)
+ model = await self.get_model(controller, model_name)
try:
application = None
if application_name not in model.applications:
- self.log.debug(
- "Deploying charm {} to machine {} in model ~{}".format(
- application_name, machine_id, model_name
- )
- )
- self.log.debug("charm: {}".format(path))
+
if machine_id is not None:
if machine_id not in model.machines:
msg = "Machine {} not found in model".format(machine_id)
config=config,
)
+ self.log.debug(
+ "Wait until application {} is ready in model {}".format(
+ 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,
db_dict=db_dict,
n2vc=self.n2vc,
)
+ self.log.debug(
+ "Application {} is ready in model {}".format(
+ application_name, model_name
+ )
+ )
else:
- raise JujuApplicationExists("Application {} exists".format(application_name))
-
- except Exception as e:
- raise e
+ raise JujuApplicationExists(
+ "Application {} exists".format(application_name)
+ )
finally:
await self.disconnect_model(model)
-
- self.log.debug("application deployed")
+ await self.disconnect_controller(controller)
return application
- async def _get_application(
- self, model: Model, application_name: str
- ) -> Application:
+ def _get_application(self, model: Model, application_name: str) -> Application:
"""Get application
:param: model: Model object
: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
:return: (str, str): (output and status)
"""
- # Get model and observer
- model = await self.get_model(model_name)
+ self.log.debug(
+ "Executing action {} using params {}".format(action_name, kwargs)
+ )
+ # Get controller
+ controller = await self.get_controller()
+
+ # Get model
+ model = await self.get_model(controller, model_name)
try:
# Get application
- application = await self._get_application(
+ application = self._get_application(
model, application_name=application_name,
)
if application is None:
if await u.is_leader_from_status():
unit = u
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)
)
- self.log.debug(
- "Executing action {} using params {}".format(action_name, kwargs)
- )
action = await unit.run_action(action_name, **kwargs)
- # Register action with observer and wait for it to finish
+ self.log.debug(
+ "Wait until action {} is completed in application {} (model={})".format(
+ action_name, application_name, model_name
+ )
+ )
await JujuModelWatcher.wait_for(
model=model,
entity=action,
db_dict=db_dict,
n2vc=self.n2vc,
)
+
output = await model.get_action_output(action_uuid=action.entity_id)
status = await model.get_action_status(uuid_or_prefix=action.entity_id)
status = (
status[action.entity_id] if action.entity_id in status else "failed"
)
- self.log.debug("action completed with status: {}".format(action.status))
- except Exception as e:
- raise e
+ self.log.debug(
+ "Action {} completed with status {} in application {} (model={})".format(
+ action_name, action.status, application_name, model_name
+ )
+ )
finally:
await self.disconnect_model(model)
+ await self.disconnect_controller(controller)
return output, status
...
}
"""
+ self.log.debug(
+ "Getting list of actions for application {}".format(application_name)
+ )
+
+ # Get controller
+ controller = await self.get_controller()
# Get model
- model = await self.get_model(model_name)
+ model = await self.get_model(controller, model_name)
- # Get application
- application = await self._get_application(
- model, application_name=application_name,
- )
+ try:
+ # Get application
+ application = self._get_application(
+ model, application_name=application_name,
+ )
+
+ # Return list of actions
+ return await application.get_actions()
- # Get list of actions
- actions = await application.get_actions()
+ finally:
+ # Disconnect from model and controller
+ await self.disconnect_model(model)
+ await self.disconnect_controller(controller)
- # Disconnect from model
- await self.disconnect_model(model)
+ async def get_metrics(self, model_name: str, application_name: str) -> dict:
+ """Get the metrics collected by the VCA.
- return actions
+ :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,
+ self, model_name: 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)
"""
- # Get model
- model = await self.get_model(model_name)
+ self.log.debug("Adding relation: {} -> {}".format(endpoint_1, endpoint_2))
- # Build relation strings
- r1 = "{}:{}".format(application_name_1, relation_1)
- r2 = "{}:{}".format(application_name_2, relation_2)
+ # Get controller
+ controller = await self.get_controller()
+
+ # Get model
+ model = await self.get_model(controller, model_name)
# Add relation
- self.log.debug("Adding relation: {} -> {}".format(r1, r2))
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))
raise e
finally:
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
:param: model_name: Model name
:param: total_timeout: Timeout
"""
- model = await self.get_model(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
+ controller = await self.get_controller()
+ model = await self.get_model(controller, model_name)
+ try:
+ self.log.debug("Destroying model {}".format(model_name))
+ uuid = model.info.uuid
+
+ # 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
- machines = await model.get_machines()
- for machine_id in machines:
- try:
- await self.destroy_machine(
- model, machine_id=machine_id, total_timeout=total_timeout,
+ # 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)
+
+ # 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
+ await asyncio.sleep(5)
+ raise Exception(
+ "Timeout waiting for model {} to be destroyed {}".format(
+ model_name, last_exception
)
- except asyncio.CancelledError:
- raise
- except Exception:
- pass
-
- # Disconnect model
- await self.disconnect_model(model)
-
- # Destroy model
- self.models.remove(model_name)
- await self.controller.destroy_model(uuid)
-
- # 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 self.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
- await asyncio.sleep(5)
- raise Exception(
- "Timeout waiting for model {} to be destroyed {}".format(
- model_name, last_exception
)
- )
+ finally:
+ await self.disconnect_controller(controller)
async def destroy_application(self, model: Model, application_name: str):
"""
"""
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:"):
- await machine.destroy(force=True)
-
- # max timeout
- end = time.time() + total_timeout
-
- # wait for machine removal
+ 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()
- 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))
+ self.log.debug("Machine destroyed: {}".format(machine_id))
else:
self.log.debug("Machine not found: {}".format(machine_id))
:param: application_name: Application name
:param: config: Config to apply to the charm
"""
+ self.log.debug("Configuring application {}".format(application_name))
+
if config:
- model = await self.get_model(model_name)
- application = await self._get_application(
- model, application_name=application_name,
- )
- await application.set_config(config)
- await self.disconnect_model(model)
+ try:
+ controller = await self.get_controller()
+ model = await self.get_model(controller, model_name)
+ application = self._get_application(
+ model, application_name=application_name,
+ )
+ await application.set_config(config)
+ finally:
+ 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
+
+ async def health_check(self, interval: float = 300.0):
+ """
+ Health check to make sure controller and controller_model connections are OK
+
+ :param: interval: Time in seconds between checks
+ """
+ while True:
+ try:
+ controller = await self.get_controller()
+ # self.log.debug("VCA is alive")
+ except Exception as e:
+ self.log.error("Health check to VCA failed: {}".format(e))
+ 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, configuration: Configuration, storage_class: str
+ ):
+ """
+ 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
+ """
+
+ 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)
+ ca_certificates = (
+ [credential.attrs["ClientCertificateData"]]
+ if "ClientCertificateData" in credential.attrs
+ else []
+ )
+ cloud = client.Cloud(
+ type_="kubernetes",
+ auth_types=[credential.auth_type],
+ endpoint=endpoint,
+ ca_certificates=ca_certificates,
+ config={
+ "operator-storage": storage_class,
+ "workload-storage": storage_class,
+ },
+ )
+
+ return await self.add_cloud(name, cloud, credential)
+
+ def get_k8s_cloud_credential(
+ self, configuration: Configuration,
+ ) -> client.CloudCredential:
+ attrs = {}
+ ca_cert = configuration.ssl_ca_cert or configuration.cert_file
+ key = configuration.key_file
+ api_key = configuration.api_key
+ token = None
+ 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 key:
+ attrs["ClientKeyData"] = open(key, "r").read()
+ 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 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 ca_cert:
+ auth_type = "userpasswithcert"
+ else:
+ auth_type = "userpass"
+ elif ca_cert 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
+ ) -> 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
+ """
+ controller = await self.get_controller()
+ try:
+ _ = await controller.add_cloud(name, cloud)
+ if credential:
+ await controller.add_credential(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)
projects / osm / N2VC.git / blobdiff