from juju.model import Model
from juju.machine import Machine
from juju.application import Application
-from juju.client._definitions import FullStatus, QueryApplicationOffersResults
+from juju.unit import Unit
+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
JujuModelAlreadyExists,
JujuControllerFailedConnecting,
JujuApplicationExists,
+ JujuInvalidK8sConfiguration,
)
from n2vc.utils import DB_DATA
from osm_common.dbbase import DbException
+from kubernetes.client.configuration import Configuration
class Libjuju:
"""
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
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 not model_names:
raise Exception(
- "model_names must be a non-empty array. Given value: {}".format(model_names)
+ "model_names must be a non-empty array. Given value: {}".format(
+ model_names
+ )
)
non_existing_models = []
models = await self.list_models()
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 JujuLeaderUnitNotFound("Cannot execute action: leader unit not found")
+ raise JujuLeaderUnitNotFound(
+ "Cannot execute action: leader unit not found"
+ )
actions = await application.get_actions()
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,
+ 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)
"""
- 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()
# 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))
await self.disconnect_model(model)
await self.disconnect_controller(controller)
+ 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
# 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)
- )
+ 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))
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,
+ 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)
+ 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, credential_name=credential_name
+ )
+
+ 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,
+ 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
projects / osm / N2VC.git / blobdiff