kubectl_command: str = "/usr/bin/kubectl",
juju_command: str = "/usr/bin/juju",
log: object = None,
- loop: object = None,
on_update_db=None,
):
"""
:param kubectl_command: path to kubectl executable
:param helm_command: path to helm executable
:param log: logger
- :param: loop: Asyncio loop
"""
# parent class
- K8sConnector.__init__(
- self,
- db,
- log=log,
- on_update_db=on_update_db,
- )
+ K8sConnector.__init__(self, db, log=log, on_update_db=on_update_db)
self.fs = fs
- self.loop = loop or asyncio.get_event_loop()
self.log.debug("Initializing K8S Juju connector")
db_uri = EnvironConfig(prefixes=["OSMLCM_", "OSMMON_"]).get("database_uri")
self._store = MotorStore(db_uri)
- self.loading_libjuju = asyncio.Lock(loop=self.loop)
+ self.loading_libjuju = asyncio.Lock()
self.uninstall_locks = {}
self.log.debug("K8S Juju connector initialized")
cleanup_data = []
try:
self.log.debug("Initializing K8s cluster for juju")
- kubectl.create_cluster_role(
- name=metadata_name,
- labels=labels,
- )
+ kubectl.create_cluster_role(name=metadata_name, labels=labels)
self.log.debug("Cluster role created")
cleanup_data.append(
- {
- "delete": kubectl.delete_cluster_role,
- "args": (metadata_name,),
- }
+ {"delete": kubectl.delete_cluster_role, "args": (metadata_name,)}
)
- kubectl.create_service_account(
- name=metadata_name,
- labels=labels,
- )
+ kubectl.create_service_account(name=metadata_name, labels=labels)
self.log.debug("Service account created")
cleanup_data.append(
- {
- "delete": kubectl.delete_service_account,
- "args": (metadata_name,),
- }
+ {"delete": kubectl.delete_service_account, "args": (metadata_name,)}
)
- kubectl.create_cluster_role_binding(
- name=metadata_name,
- labels=labels,
- )
+ kubectl.create_cluster_role_binding(name=metadata_name, labels=labels)
self.log.debug("Role binding created")
cleanup_data.append(
{
- "delete": kubectl.delete_service_account,
+ "delete": kubectl.delete_cluster_role_binding,
"args": (metadata_name,),
}
)
- token, client_cert_data = await kubectl.get_secret_data(
- metadata_name,
- )
+ token, client_cert_data = await kubectl.get_secret_data(metadata_name)
default_storage_class = kubectl.get_default_storage_class()
self.log.debug("Default storage class: {}".format(default_storage_class))
async def repo_list(self):
raise MethodNotImplemented()
- async def repo_remove(
- self,
- name: str,
- ):
+ async def repo_remove(self, name: str):
raise MethodNotImplemented()
async def synchronize_repos(self, cluster_uuid: str, name: str):
previous_workdir = "/app/storage"
self.log.debug("[install] deploying {}".format(bundle))
- await libjuju.deploy(bundle, model_name=namespace, wait=atomic, timeout=timeout)
+ instantiation_params = params.get("overlay") if params else None
+ await libjuju.deploy(
+ bundle,
+ model_name=namespace,
+ wait=atomic,
+ timeout=timeout,
+ instantiation_params=instantiation_params,
+ )
os.chdir(previous_workdir)
# update information in the database (first, the VCA status, and then, the namespace)
return True
async def get_scale_count(
- self,
- resource_name: str,
- kdu_instance: str,
- namespace: str = None,
- **kwargs,
+ self, resource_name: str, kdu_instance: str, namespace: str = None, **kwargs
) -> int:
"""Get an application scale count
"""Rollback"""
async def rollback(
- self,
- cluster_uuid: str,
- kdu_instance: str,
- revision: int = 0,
+ self, cluster_uuid: str, kdu_instance: str, revision: int = 0
) -> str:
"""Rollback a model
"""Deletion"""
async def uninstall(
- self,
- cluster_uuid: str,
- kdu_instance: str,
- namespace: str = None,
- **kwargs,
+ self, cluster_uuid: str, kdu_instance: str, namespace: str = None, **kwargs
) -> bool:
"""Uninstall a KDU instance
will_not_delete = False
if model_name not in self.uninstall_locks:
- self.uninstall_locks[model_name] = asyncio.Lock(loop=self.loop)
+ self.uninstall_locks[model_name] = asyncio.Lock()
delete_lock = self.uninstall_locks[model_name]
while delete_lock.locked():
"""Introspection"""
- async def inspect_kdu(
- self,
- kdu_model: str,
- ) -> dict:
+ async def inspect_kdu(self, kdu_model: str) -> dict:
"""Inspect a KDU
Inspects a bundle and returns a dictionary of config parameters and
return kdu
- async def help_kdu(
- self,
- kdu_model: str,
- ) -> str:
+ async def help_kdu(self, kdu_model: str) -> str:
"""View the README
If available, returns the README of the bundle.
return status
async def add_relation(
- self,
- provider: RelationEndpoint,
- requirer: RelationEndpoint,
+ self, provider: RelationEndpoint, requirer: RelationEndpoint
):
"""
Add relation between two charmed endpoints
self.log.debug(f"adding new relation between {provider} and {requirer}")
cross_model_relation = (
provider.model_name != requirer.model_name
- or requirer.vca_id != requirer.vca_id
+ or provider.vca_id != requirer.vca_id
)
try:
if cross_model_relation:
requirer.model_name, offer, provider_libjuju
)
await requirer_libjuju.add_relation(
- requirer.model_name,
- requirer.endpoint,
- saas_name,
+ requirer.model_name, requirer.endpoint, saas_name
)
else:
# Standard relation
"""
return "cred-{}".format(cluster_uuid)
- def get_namespace(
- self,
- cluster_uuid: str,
- ) -> str:
+ def get_namespace(self, cluster_uuid: str) -> str:
"""Get the namespace UUID
Gets the namespace's unique name
if not self.libjuju:
async with self.loading_libjuju:
vca_connection = await get_connection(self._store)
- self.libjuju = Libjuju(vca_connection, loop=self.loop, log=self.log)
+ self.libjuju = Libjuju(vca_connection, log=self.log)
return self.libjuju
else:
vca_connection = await get_connection(self._store, vca_id)
- return Libjuju(
- vca_connection,
- loop=self.loop,
- log=self.log,
- n2vc=self,
- )
+ return Libjuju(vca_connection, log=self.log, n2vc=self)
def _get_kubectl(self, credentials: str) -> Kubectl:
"""
projects / osm / N2VC.git / blobdiff