osm_lcm/temporal/vnf_activities.py

#########################################################################
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import logging
from typing import List, Any

from osm_common.temporal_constants import (
    ACTIVITY_CHANGE_VNF_STATE,
    ACTIVITY_CHANGE_VNF_INSTANTIATION_STATE,
    ACTIVITY_GET_TASK_QUEUE,
    ACTIVITY_GET_VIM_CLOUD,
    ACTIVITY_GET_VNF_DESCRIPTOR,
    ACTIVITY_GET_VNF_RECORD,
    ACTIVITY_SEND_NOTIFICATION_FOR_VNF,
    ACTIVITY_SET_VNF_MODEL,
    VIM_TYPE_TASK_QUEUE_MAPPINGS,
)
from osm_common.dataclasses.temporal_dataclasses import (
    ChangeVnfInstantiationStateInput,
    ChangeVnfStateInput,
    GetTaskQueueInput,
    GetTaskQueueOutput,
    GetVimCloudInput,
    GetVimCloudOutput,
    GetVnfDescriptorInput,
    GetVnfDescriptorOutput,
    GetVnfRecordInput,
    GetVnfRecordOutput,
    SetVnfModelInput,
    VduComputeConstraints,
)
from osm_lcm.data_utils.database.database import Database
from temporalio import activity


CONFIG_IDENTIFIER = "config::"


class VnfOperations:
    def __init__(self, db: Database):
        self.db: Database = db
        self.logger = logging.getLogger(f"lcm.act.{self.__class__.__name__}")

    @activity.defn(name=ACTIVITY_GET_TASK_QUEUE)
    async def get_task_queue(
        self, get_task_queue_input: GetTaskQueueInput
    ) -> GetTaskQueueOutput:
        """Finds the appropriate task queue according to VIM type of VNF.

        Collaborators:
            DB read:           vim_accounts, vnfrs

        Raises (retryable):
            DbException: If DB read operations fail, the collection or DB record ID does not exist.

        Activity Lifecycle:
            This activity should complete relatively quickly (less than a
            second). However, it would be reasonable to wait up to 10
            seconds.

            This activity will not report a heartbeat due to its
            short-running nature.

            It is not necessary to implement a back-off strategy for this
            activity, the operation is idempotent.

        """
        vnfr = self.db.get_one("vnfrs", {"_id": get_task_queue_input.vnfr_uuid})
        vim_record = self.db.get_one("vim_accounts", {"_id": vnfr["vim-account-id"]})
        task_queue = VIM_TYPE_TASK_QUEUE_MAPPINGS[vim_record["vim_type"]]
        self.logger.debug(f"Got the task queue {task_queue} for VNF operations.")
        return GetTaskQueueOutput(task_queue)

    @activity.defn(name=ACTIVITY_GET_VIM_CLOUD)
    async def get_vim_cloud(
        self, get_vim_cloud_input: GetVimCloudInput
    ) -> GetVimCloudOutput:
        """Finds the cloud by checking the VIM account of VNF.

        Collaborators:
            DB Read:  vnfrs, vim_accounts

        Raises (retryable):
            DbException: If DB read operations fail, the collection or DB record ID does not exist.

        Activity Lifecycle:
            This activity should complete relatively quickly (less than a
            second). However, it would be reasonable to wait up to 10
            seconds.

            This activity will not report a heartbeat due to its
            short-running nature.

            It is not necessary to implement a back-off strategy for this
            activity, the operation is idempotent.

        """
        vnfr = self.db.get_one("vnfrs", {"_id": get_vim_cloud_input.vnfr_uuid})
        vim_record = self.db.get_one("vim_accounts", {"_id": vnfr["vim-account-id"]})
        cloud = vim_record["config"].get("cloud", "")
        self.logger.debug(f"Got the cloud type {cloud} for VNF operations.")
        return GetVimCloudOutput(cloud=cloud)

    @activity.defn(name=ACTIVITY_GET_VNF_RECORD)
    async def get_vnf_record(
        self, get_vnf_record_input: GetVnfRecordInput
    ) -> GetVnfRecordOutput:
        """Gets the VNF record and VNF descriptor from Database.

        Collaborators:
            DB read:           vnfrs

        Raises (retryable):
            DbException: If DB read operations fail, the collection or DB record ID does not exist.

        Activity Lifecycle:
            This activity should complete relatively quickly (less than 10
            second).

            This activity will not report a heartbeat due to its
            short-running nature.

            This is an idempotent activity.

        """
        vnfr = self.db.get_one("vnfrs", {"_id": get_vnf_record_input.vnfr_uuid})
        self.logger.debug("Got the vnfr from Database for VNF operations.")
        return GetVnfRecordOutput(vnfr=vnfr)

    @activity.defn(name=ACTIVITY_GET_VNF_DESCRIPTOR)
    async def get_vnf_descriptor(
        self, get_vnf_descriptor_input: GetVnfDescriptorInput
    ) -> GetVnfDescriptorOutput:
        """Gets the VNF record and VNF descriptor from Database.

        Collaborators:
            DB read:           vnfds

        Raises (retryable):
            DbException: If DB read operations fail, the collection or DB record ID does not exist.

        Activity Lifecycle:
            This activity should complete relatively quickly (less than 10
            second).

            This activity will not report a heartbeat due to its
            short-running nature.

            This is an idempotent activity.

        """
        vnfd = self.db.get_one("vnfds", {"_id": get_vnf_descriptor_input.vnfd_uuid})
        self.logger.debug("Got the vnfr and vnfd from Database for VNF operations.")
        return GetVnfDescriptorOutput(vnfd=vnfd)

    @staticmethod
    def get_vdu_instantiation_params(
        vdu_id: str, vnf_instantiation_config: dict
    ) -> dict:
        for vdu in vnf_instantiation_config.get("vdu", []):
            if vdu.get("id") == vdu_id:
                return vdu.get("configurable-properties", {})
        return {}

    @staticmethod
    def get_compute_constraints(vdu: dict, vnfd: dict) -> VduComputeConstraints:
        compute_desc_id = vdu.get("virtual-compute-desc")
        if not compute_desc_id:
            return VduComputeConstraints(cores=0, mem=0)
        flavor_details = VnfOperations._get_flavor_details(compute_desc_id, vnfd)
        if not flavor_details:
            return VduComputeConstraints(cores=0, mem=0)

        cpu_cores = flavor_details.get("virtual-cpu", {}).get("num-virtual-cpu", 0)
        memory_gb = flavor_details.get("virtual-memory", {}).get("size", 0)
        return VduComputeConstraints(cores=cpu_cores, mem=int(memory_gb))

    @staticmethod
    def _get_flavor_details(compute_desc_id: str, vnfd: dict) -> Any:
        for flavor in vnfd.get("virtual-compute-desc", []):
            if flavor.get("id") == compute_desc_id:
                return flavor
        return None

    @staticmethod
    def get_application_config(vdu: dict, vdu_instantiation_config: dict) -> dict:
        configurable_properties = vdu.get("configurable-properties", [])

        config_from_descriptor = VnfOperations._get_only_config_items(
            VnfOperations._list_to_dict(configurable_properties)
        )

        config_from_instantiation = VnfOperations._get_only_config_items(
            vdu_instantiation_config
        )
        return {**config_from_descriptor, **config_from_instantiation}

    @staticmethod
    def _get_only_config_items(config: dict) -> dict:
        return {
            key[len(CONFIG_IDENTIFIER) :]: value
            for key, value in config.items()
            if key.startswith(CONFIG_IDENTIFIER)
        }

    @staticmethod
    def _list_to_dict(indata: List[dict]) -> dict:
        return {
            item["key"]: item["value"]
            for item in indata
            if item.get("key") and item.get("value")
        }


class VnfDbActivity:
    """Perform Database operations for VNF accounts.

    Args:
        db  (Database):       Data Access Object
    """

    def __init__(self, db: Database):
        self.db: Database = db
        self.logger = logging.getLogger(f"lcm.act.{self.__class__.__name__}")

    @activity.defn(name=ACTIVITY_CHANGE_VNF_STATE)
    async def change_vnf_state(self, vnf_state_input: ChangeVnfStateInput) -> None:
        """Updates the VNF State in VNFR.

        Collaborators:
            DB Write:           vnfrs

        Raises (retryable):
            DbException: If DB access/update fails, the collection or DB record ID does not exist.

        Activity Lifecycle:
            This activity should complete relatively quickly (less than a
            second). However, it would be reasonable to wait up to 10
            seconds.

            This activity will not report a heartbeat due to its
            short-running nature.

            It is not necessary to implement a back-off strategy for this
            activity, the operation is idempotent.

        """
        update_vnf_state = {"vnfState": vnf_state_input.state.name}
        self.db.set_one("vnfrs", {"_id": vnf_state_input.vnfr_uuid}, update_vnf_state)
        self.logger.debug(
            f"VNF {vnf_state_input.vnfr_uuid} state is updated to {vnf_state_input.state.name}."
        )

    @activity.defn(name=ACTIVITY_CHANGE_VNF_INSTANTIATION_STATE)
    async def change_vnf_instantiation_state(
        self, vnf_instantiation_state_input: ChangeVnfInstantiationStateInput
    ) -> None:
        """Updates the VNF Instantiation State in VNFR.

        Collaborators:
            DB Write:           vnfrs

        Raises (retryable):
            DbException: If DB access or update fails, the collection or DB record ID does not exist.

        Activity Lifecycle:
            This activity should complete relatively quickly (less than a
            second). However, it would be reasonable to wait up to 10
            seconds.

            This activity will not report a heartbeat due to its
            short-running nature.

            It is not necessary to implement a back-off strategy for this
            activity, the operation is idempotent.

        """
        update_vnf_instantiation_state = {
            "instantiationState": vnf_instantiation_state_input.state.name
        }
        self.db.set_one(
            "vnfrs",
            {"_id": vnf_instantiation_state_input.vnfr_uuid},
            update_vnf_instantiation_state,
        )
        self.logger.debug(
            f"VNF {vnf_instantiation_state_input.vnfr_uuid} state is updated to {vnf_instantiation_state_input.state.name}."
        )

    @activity.defn(name=ACTIVITY_SET_VNF_MODEL)
    async def set_vnf_model(self, set_vnf_model_input: SetVnfModelInput) -> None:
        """Updates the model name of VNF in VNFR.

        Collaborators:
            DB Write:           vnfrs

        Raises (retryable):
            DbException: If DB access or update fails, the collection or DB record ID does not exist.

        Activity Lifecycle:
            This activity should complete relatively quickly (less than a
            second). However, it would be reasonable to wait up to 10
            seconds.

            This activity will not report a heartbeat due to its
            short-running nature.

            It is not necessary to implement a back-off strategy for this
            activity, the operation is idempotent.

        """
        update_namespace = {"namespace": set_vnf_model_input.model_name}
        self.db.set_one(
            "vnfrs", {"_id": set_vnf_model_input.vnfr_uuid}, update_namespace
        )
        self.logger.debug(
            f"VNF {set_vnf_model_input.vnfr_uuid} model name is updated to {set_vnf_model_input.model_name}."
        )


class VnfSendNotifications:
    """Perform Notification operations."""

    def __init__(self):
        self.logger = logging.getLogger(f"lcm.act.{self.__class__.__name__}")

    @activity.defn(name=ACTIVITY_SEND_NOTIFICATION_FOR_VNF)
    async def send_notification_for_vnf(
        self, input: ChangeVnfInstantiationStateInput
    ) -> None:
        """If VNF LCM operation state changes, send notification updates.

        This activity does nothing.

        """
        self.logger.debug("Send notification for VNF not implemented.")