from n2vc.k8s_helm_conn import K8sHelmConnector
from n2vc.k8s_helm3_conn import K8sHelm3Connector
from n2vc.k8s_juju_conn import K8sJujuConnector
+from n2vc.n2vc_juju_conn import N2VCJujuConnector
from n2vc.exceptions import K8sException, N2VCException
from osm_common.dbbase import DbException
from copy import deepcopy
vim_config_encrypted = {"1.1": ("admin_password", "nsx_password", "vcenter_password"),
"default": ("admin_password", "nsx_password", "vcenter_password", "vrops_password")}
- def __init__(self, db, msg, fs, lcm_tasks, config, loop):
+ def __init__(self, msg, lcm_tasks, config, loop):
"""
Init, Connect to database, filesystem storage, and messaging
:param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
self.lcm_tasks = lcm_tasks
self.ro_config = config["ro_config"]
- super().__init__(db, msg, fs, self.logger)
+ super().__init__(msg, self.logger)
async def create(self, vim_content, order_id):
# values that are encrypted at wim config because they are passwords
wim_config_encrypted = ()
- def __init__(self, db, msg, fs, lcm_tasks, config, loop):
+ def __init__(self, msg, lcm_tasks, config, loop):
"""
Init, Connect to database, filesystem storage, and messaging
:param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
self.lcm_tasks = lcm_tasks
self.ro_config = config["ro_config"]
- super().__init__(db, msg, fs, self.logger)
+ super().__init__(msg, self.logger)
async def create(self, wim_content, order_id):
class SdnLcm(LcmBase):
- def __init__(self, db, msg, fs, lcm_tasks, config, loop):
+ def __init__(self, msg, lcm_tasks, config, loop):
"""
Init, Connect to database, filesystem storage, and messaging
:param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
self.lcm_tasks = lcm_tasks
self.ro_config = config["ro_config"]
- super().__init__(db, msg, fs, self.logger)
+ super().__init__(msg, self.logger)
async def create(self, sdn_content, order_id):
class K8sClusterLcm(LcmBase):
timeout_create = 300
- def __init__(self, db, msg, fs, lcm_tasks, config, loop):
+ def __init__(self, msg, lcm_tasks, config, loop):
"""
Init, Connect to database, filesystem storage, and messaging
:param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
self.loop = loop
self.lcm_tasks = lcm_tasks
self.vca_config = config["VCA"]
- self.fs = fs
- self.db = db
+
+ super().__init__(msg, self.logger)
self.helm2_k8scluster = K8sHelmConnector(
kubectl_command=self.vca_config.get("kubectlpath"),
helm_command=self.vca_config.get("helmpath"),
- fs=self.fs,
log=self.logger,
+ on_update_db=None,
db=self.db,
- on_update_db=None
+ fs=self.fs
)
self.helm3_k8scluster = K8sHelm3Connector(
self.juju_k8scluster = K8sJujuConnector(
kubectl_command=self.vca_config.get("kubectlpath"),
juju_command=self.vca_config.get("jujupath"),
- fs=self.fs,
log=self.logger,
- db=self.db,
loop=self.loop,
on_update_db=None,
- vca_config=self.vca_config,
+ db=self.db,
+ fs=self.fs
)
+
self.k8s_map = {
"helm-chart": self.helm2_k8scluster,
"helm-chart-v3": self.helm3_k8scluster,
"juju-bundle": self.juju_k8scluster,
}
- super().__init__(db, msg, fs, self.logger)
-
async def create(self, k8scluster_content, order_id):
op_id = k8scluster_content.pop('op_id', None)
for task_name in ("helm-chart", "juju-bundle", "helm-chart-v3"):
if init_target and task_name not in init_target:
continue
- task = asyncio.ensure_future(self.k8s_map[task_name].init_env(k8s_credentials,
- reuse_cluster_uuid=k8scluster_id))
+ task = asyncio.ensure_future(
+ self.k8s_map[task_name].init_env(
+ k8s_credentials,
+ reuse_cluster_uuid=k8scluster_id,
+ vca_id=db_k8scluster.get("vca_id"),
+ )
+ )
pending_tasks.append(task)
task2name[task] = task_name
if k8s_jb_id: # delete in reverse order of creation
step = "Removing juju-bundle '{}'".format(k8s_jb_id)
uninstall_sw = deep_get(db_k8scluster, ("_admin", "juju-bundle", "created")) or False
- cluster_removed = await self.juju_k8scluster.reset(cluster_uuid=k8s_jb_id, uninstall_sw=uninstall_sw)
+ cluster_removed = await self.juju_k8scluster.reset(
+ cluster_uuid=k8s_jb_id,
+ uninstall_sw=uninstall_sw,
+ vca_id=db_k8scluster.get("vca_id"),
+ )
db_k8scluster_update["_admin.juju-bundle.id"] = None
db_k8scluster_update["_admin.juju-bundle.operationalState"] = "DISABLED"
if k8s_h3c_id:
step = "Removing helm-chart-v3 '{}'".format(k8s_hc_id)
uninstall_sw = deep_get(db_k8scluster, ("_admin", "helm-chart-v3", "created")) or False
- cluster_removed = await self.helm3_k8scluster.reset(cluster_uuid=k8s_hc_id, uninstall_sw=uninstall_sw)
+ cluster_removed = await self.helm3_k8scluster.reset(cluster_uuid=k8s_h3c_id, uninstall_sw=uninstall_sw)
db_k8scluster_update["_admin.helm-chart-v3.id"] = None
db_k8scluster_update["_admin.helm-chart-v3.operationalState"] = "DISABLED"
self.lcm_tasks.remove("k8scluster", k8scluster_id, order_id)
+class VcaLcm(LcmBase):
+ timeout_create = 30
+
+ def __init__(self, msg, lcm_tasks, config, loop):
+ """
+ Init, Connect to database, filesystem storage, and messaging
+ :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
+ :return: None
+ """
+
+ self.logger = logging.getLogger("lcm.vca")
+ self.loop = loop
+ self.lcm_tasks = lcm_tasks
+
+ super().__init__(msg, self.logger)
+
+ # create N2VC connector
+ self.n2vc = N2VCJujuConnector(
+ log=self.logger,
+ loop=self.loop,
+ fs=self.fs,
+ db=self.db
+ )
+
+ def _get_vca_by_id(self, vca_id: str) -> dict:
+ db_vca = self.db.get_one("vca", {"_id": vca_id})
+ self.db.encrypt_decrypt_fields(
+ db_vca,
+ "decrypt",
+ ["secret", "cacert"],
+ schema_version=db_vca["schema_version"], salt=db_vca["_id"]
+ )
+ return db_vca
+
+ async def create(self, vca_content, order_id):
+ op_id = vca_content.pop("op_id", None)
+ if not self.lcm_tasks.lock_HA("vca", "create", op_id):
+ return
+
+ vca_id = vca_content["_id"]
+ self.logger.debug("Task vca_create={} {}".format(vca_id, "Enter"))
+
+ db_vca = None
+ db_vca_update = {}
+
+ try:
+ self.logger.debug("Task vca_create={} {}".format(vca_id, "Getting vca from db"))
+ db_vca = self._get_vca_by_id(vca_id)
+
+ task = asyncio.ensure_future(
+ asyncio.wait_for(
+ self.n2vc.validate_vca(db_vca["_id"]),
+ timeout=self.timeout_create,
+ )
+ )
+
+ await asyncio.wait([task], return_when=asyncio.FIRST_COMPLETED)
+ if task.exception():
+ raise task.exception()
+ self.logger.debug("Task vca_create={} {}".format(vca_id, "vca registered and validated successfully"))
+ db_vca_update["_admin.operationalState"] = "ENABLED"
+ db_vca_update["_admin.detailed-status"] = "Connectivity: ok"
+ operation_details = "VCA validated"
+ operation_state = "COMPLETED"
+
+ self.logger.debug("Task vca_create={} {}".format(vca_id, "Done. Result: {}".format(operation_state)))
+
+ except Exception as e:
+ error_msg = "Failed with exception: {}".format(e)
+ self.logger.error("Task vca_create={} {}".format(vca_id, error_msg))
+ db_vca_update["_admin.operationalState"] = "ERROR"
+ db_vca_update["_admin.detailed-status"] = error_msg
+ operation_state = "FAILED"
+ operation_details = error_msg
+ finally:
+ try:
+ self.update_db_2("vca", vca_id, db_vca_update)
+
+ # Register the operation and unlock
+ self.lcm_tasks.unlock_HA(
+ "vca",
+ "create",
+ op_id,
+ operationState=operation_state,
+ detailed_status=operation_details
+ )
+ except DbException as e:
+ self.logger.error("Task vca_create={} {}".format(vca_id, "Cannot update database: {}".format(e)))
+ self.lcm_tasks.remove("vca", vca_id, order_id)
+
+ async def delete(self, vca_content, order_id):
+
+ # HA tasks and backward compatibility:
+ # If "vim_content" does not include "op_id", we a running a legacy NBI version.
+ # In such a case, HA is not supported by NBI, "op_id" is None, and lock_HA() will do nothing.
+ # Register "delete" task here for related future HA operations
+ op_id = vca_content.pop("op_id", None)
+ if not self.lcm_tasks.lock_HA("vca", "delete", op_id):
+ return
+
+ db_vca_update = {}
+ vca_id = vca_content["_id"]
+
+ try:
+ self.logger.debug("Task vca_delete={} {}".format(vca_id, "Deleting vca from db"))
+ self.db.del_one("vca", {"_id": vca_id})
+ db_vca_update = None
+ operation_details = "deleted"
+ operation_state = "COMPLETED"
+
+ self.logger.debug("Task vca_delete={} {}".format(vca_id, "Done. Result: {}".format(operation_state)))
+ except Exception as e:
+ error_msg = "Failed with exception: {}".format(e)
+ self.logger.error("Task vca_delete={} {}".format(vca_id, error_msg))
+ db_vca_update["_admin.operationalState"] = "ERROR"
+ db_vca_update["_admin.detailed-status"] = error_msg
+ operation_state = "FAILED"
+ operation_details = error_msg
+ finally:
+ try:
+ self.update_db_2("vca", vca_id, db_vca_update)
+ self.lcm_tasks.unlock_HA(
+ "vca",
+ "delete",
+ op_id,
+ operationState=operation_state,
+ detailed_status=operation_details,
+ )
+ except DbException as e:
+ self.logger.error("Task vca_delete={} {}".format(vca_id, "Cannot update database: {}".format(e)))
+ self.lcm_tasks.remove("vca", vca_id, order_id)
+
+
class K8sRepoLcm(LcmBase):
- def __init__(self, db, msg, fs, lcm_tasks, config, loop):
+ def __init__(self, msg, lcm_tasks, config, loop):
"""
Init, Connect to database, filesystem storage, and messaging
:param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
self.loop = loop
self.lcm_tasks = lcm_tasks
self.vca_config = config["VCA"]
- self.fs = fs
- self.db = db
- super().__init__(db, msg, fs, self.logger)
+ super().__init__(msg, self.logger)
+
+ self.k8srepo = K8sHelmConnector(
+ kubectl_command=self.vca_config.get("kubectlpath"),
+ helm_command=self.vca_config.get("helmpath"),
+ fs=self.fs,
+ log=self.logger,
+ db=self.db,
+ on_update_db=None
+ )
async def create(self, k8srepo_content, order_id):