osm_lcm/nglcm.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-

##
#
# 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 asyncio
import getopt
import logging
import logging.handlers
import os
import sys
import yaml

from osm_common.dbbase import DbException
from osm_common.temporal_constants import LCM_TASK_QUEUE
from osm_lcm.data_utils.database.database import Database
from osm_lcm.data_utils.lcm_config import LcmCfg
from osm_lcm.lcm_utils import LcmException
from os import path
from osm_lcm.temporal.lcm_activities import NsLcmActivity
from osm_lcm.temporal.lcm_workflows import NsNoOpWorkflow
from osm_lcm.temporal.vim_activities import VimDbActivity
from osm_lcm.temporal.juju_paas_activities import JujuPaasConnector
from osm_lcm.temporal.vim_workflows import (
    VimCreateWorkflow,
    VimDeleteWorkflow,
    VimUpdateWorkflow,
)
from osm_lcm.temporal.vdu_workflows import VduInstantiateWorkflow
from osm_lcm.temporal.vnf_workflows import VnfInstantiateWorkflow
from osm_lcm.temporal.vnf_activities import VnfDbActivity, VnfOperations
from temporalio.client import Client
from temporalio.worker import Worker


class NGLcm:
    main_config = LcmCfg()

    def __init__(self, config_file):
        """
        Init, Connect to database, filesystem storage, and messaging
        :param config_file: two level dictionary with configuration. Top level should contain 'database', 'storage',
        :return: None
        """
        self.db = None
        self.logger = logging.getLogger("lcm")
        self._load_configuration(config_file)
        self._configure_logging()

        try:
            self.db = Database(self.main_config.to_dict()).instance.db
        except DbException as e:
            self.logger.critical(str(e), exc_info=True)
            raise LcmException(str(e))

    def _load_configuration(self, config_file):
        config = self._read_config_file(config_file)
        self.main_config.set_from_dict(config)
        self.main_config.transform()
        self.main_config.load_from_env()
        self.logger.critical("Loaded configuration:" + str(self.main_config.to_dict()))

    def _read_config_file(self, config_file):
        try:
            with open(config_file) as f:
                return yaml.safe_load(f)
        except Exception as e:
            self.logger.critical("At config file '{}': {}".format(config_file, e))
            exit(1)

    @staticmethod
    def _get_log_formatter_simple():
        log_format_simple = (
            "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
        )
        return logging.Formatter(log_format_simple, datefmt="%Y-%m-%dT%H:%M:%S")

    def _create_file_handler(self):
        return logging.handlers.RotatingFileHandler(
            self.main_config.globalConfig.logfile,
            maxBytes=100e6,
            backupCount=9,
            delay=0,
        )

    def _log_other_modules(self):
        for logger in ("message", "database", "storage", "tsdb", "temporal"):
            logger_config = self.main_config.to_dict()[logger]
            logger_module = logging.getLogger(logger_config["logger_name"])
            if logger_config["logfile"]:
                file_handler = logging.handlers.RotatingFileHandler(
                    logger_config["logfile"], maxBytes=100e6, backupCount=9, delay=0
                )
                file_handler.setFormatter(self._get_log_formatter_simple())
                logger_module.addHandler(file_handler)
            if logger_config["loglevel"]:
                logger_module.setLevel(logger_config["loglevel"])

    def _configure_logging(self):
        if self.main_config.globalConfig.logfile:
            file_handler = self._create_file_handler()
            file_handler.setFormatter(self._get_log_formatter_simple())
            self.logger.addHandler(file_handler)

        if not self.main_config.globalConfig.to_dict()["nologging"]:
            str_handler = logging.StreamHandler()
            str_handler.setFormatter(self._get_log_formatter_simple())
            self.logger.addHandler(str_handler)

        if self.main_config.globalConfig.to_dict()["loglevel"]:
            self.logger.setLevel(self.main_config.globalConfig.loglevel)

        self._log_other_modules()
        self.logger.critical("starting osm/nglcm")

    async def start(self):
        temporal_api = (
            f"{self.main_config.temporal.host}:{str(self.main_config.temporal.port)}"
        )
        client = await Client.connect(temporal_api)
        vim_data_activity_instance = VimDbActivity(self.db)
        paas_connector_instance = JujuPaasConnector(self.db)
        nslcm_activity_instance = NsLcmActivity(self.db)
        vnf_operation_instance = VnfOperations(self.db)
        vnf_data_activity_instance = VnfDbActivity(self.db)

        workflows = [
            NsNoOpWorkflow,
            VimCreateWorkflow,
            VimDeleteWorkflow,
            VimUpdateWorkflow,
            VduInstantiateWorkflow,
            VnfInstantiateWorkflow,
        ]
        activities = [
            vim_data_activity_instance.update_vim_operation_state,
            vim_data_activity_instance.update_vim_state,
            vim_data_activity_instance.delete_vim_record,
            nslcm_activity_instance.update_ns_lcm_operation_state,
            nslcm_activity_instance.no_op,
            paas_connector_instance.test_vim_connectivity,
            paas_connector_instance.create_model_if_doesnt_exist,
            paas_connector_instance.deploy_charm,
            paas_connector_instance.check_charm_status,
            vnf_operation_instance.get_task_queue,
            vnf_data_activity_instance.change_nf_state,
            vnf_data_activity_instance.change_nf_instantiation_state,
            vnf_data_activity_instance.change_nf_notification_state,
        ]

        # Check if we are running under a debugger
        debug = os.getenv("VSCODE_IPC_HOOK_CLI") is not None

        worker = Worker(
            client,
            task_queue=LCM_TASK_QUEUE,
            workflows=workflows,
            activities=activities,
            debug_mode=debug,
        )

        self.logger.info("Starting LCM temporal worker")
        await worker.run()


if __name__ == "__main__":
    try:
        opts, args = getopt.getopt(
            sys.argv[1:], "hc:", ["config=", "help", "health-check"]
        )
        # TODO add  "log-socket-host=", "log-socket-port=", "log-file="
        config_file = None
        for o, a in opts:
            if o in ("-c", "--config"):
                config_file = a
            else:
                assert False, "Unhandled option"

        if config_file:
            if not path.isfile(config_file):
                print(
                    "configuration file '{}' does not exist".format(config_file),
                    file=sys.stderr,
                )
                exit(1)
        else:
            for config_file in (
                __file__[: __file__.rfind(".")] + ".cfg",
                "./lcm.cfg",
                "/etc/osm/lcm.cfg",
            ):
                print(f"{config_file}")
                if path.isfile(config_file):
                    break
            else:
                print(
                    "No configuration file 'lcm.cfg' found neither at local folder nor at /etc/osm/",
                    file=sys.stderr,
                )
                exit(1)
        lcm = NGLcm(config_file)
        asyncio.run(lcm.start())
    except (LcmException, getopt.GetoptError) as e:
        print(str(e), file=sys.stderr)
        # usage()
        exit(1)