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
log=self.logger,
loop=self.loop,
on_update_db=None,
- vca_config=self.vca_config,
db=self.db,
fs=self.fs
)
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"
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, msg, lcm_tasks, config, loop):
projects / osm / LCM.git / blobdiff