)
from n2vc.n2vc_conn import N2VCConnector
from n2vc.n2vc_conn import obj_to_dict, obj_to_yaml
)
from n2vc.n2vc_conn import N2VCConnector
from n2vc.n2vc_conn import obj_to_dict, obj_to_yaml
from n2vc.store import MotorStore
from n2vc.utils import get_ee_id_components, generate_random_alfanum_string
from n2vc.vca.connection import get_connection
from n2vc.store import MotorStore
from n2vc.utils import get_ee_id_components, generate_random_alfanum_string
from n2vc.vca.connection import get_connection
:param: on_update_db: Callback function to be called for updating the database.
"""
# parent class constructor
:param: on_update_db: Callback function to be called for updating the database.
"""
# parent class constructor
- N2VCConnector.__init__(
- self,
- db=db,
- fs=fs,
- log=log,
- loop=loop,
- on_update_db=on_update_db,
- )
+ N2VCConnector.__init__(self, db=db, fs=fs, log=log, on_update_db=on_update_db)
db_uri = EnvironConfig(prefixes=["OSMLCM_", "OSMMON_"]).get("database_uri")
self._store = MotorStore(db_uri)
db_uri = EnvironConfig(prefixes=["OSMLCM_", "OSMMON_"]).get("database_uri")
self._store = MotorStore(db_uri)
# In case of native_charm is being deployed, if JujuApplicationExists error happens
# it will try to add_unit
# In case of native_charm is being deployed, if JujuApplicationExists error happens
# it will try to add_unit
- @retry(attempts=3, delay=5, retry_exceptions=(N2VCApplicationExists,), timeout=None)
+ @retry(
+ attempts=3,
+ delay=5,
+ retry_exceptions=(N2VCApplicationExists,),
+ timeout=None,
+ callback=retry_callback,
+ )
_, ns_id, _, _, _ = self._get_namespace_components(namespace=namespace)
model_name = "{}-k8s".format(ns_id)
if not await libjuju.model_exists(model_name):
_, ns_id, _, _, _ = self._get_namespace_components(namespace=namespace)
model_name = "{}-k8s".format(ns_id)
if not await libjuju.model_exists(model_name):
- self,
- provider: RelationEndpoint,
- requirer: RelationEndpoint,
+ self, provider: RelationEndpoint, requirer: RelationEndpoint
self.log.debug(f"adding new relation between {provider} and {requirer}")
cross_model_relation = (
provider.model_name != requirer.model_name
self.log.debug(f"adding new relation between {provider} and {requirer}")
cross_model_relation = (
provider.model_name != requirer.model_name
requirer.model_name, offer, provider_libjuju
)
await requirer_libjuju.add_relation(
requirer.model_name, offer, provider_libjuju
)
await requirer_libjuju.add_relation(
self.log.info("Deleting namespace={}".format(namespace))
will_not_delete = False
if namespace not in self.delete_namespace_locks:
self.log.info("Deleting namespace={}".format(namespace))
will_not_delete = False
if namespace not in self.delete_namespace_locks:
await libjuju.upgrade_charm(
application_name=application_name,
path=path,
await libjuju.upgrade_charm(
application_name=application_name,
path=path,
- libjuju = Libjuju(vca_connection, loop=self.loop, log=self.log, n2vc=self)
+ libjuju = Libjuju(vca_connection, log=self.log, n2vc=self)