osm_lcm/nglcm.py

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

##
# Copyright 2018 Telefonica S.A.
#
# 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 sys
import yaml

from osm_lcm.lcm_utils import LcmException

from osm_common.dbbase import DbException
from osm_lcm.data_utils.database.database import Database
from osm_lcm.data_utils.lcm_config import LcmCfg
from os import path
from temporalio import workflow
from temporalio.client import Client
from temporalio.worker import Worker


class NGLcm:
    main_config = LcmCfg()

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

        # logging
        self.logger = logging.getLogger("lcm")
        # load configuration
        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()))

        # logging
        log_format_simple = (
            "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
        )
        log_formatter_simple = logging.Formatter(
            log_format_simple, datefmt="%Y-%m-%dT%H:%M:%S"
        )
        if self.main_config.globalConfig.logfile:
            file_handler = logging.handlers.RotatingFileHandler(
                self.main_config.globalConfig.logfile,
                maxBytes=100e6,
                backupCount=9,
                delay=0,
            )
            file_handler.setFormatter(log_formatter_simple)
            self.logger.addHandler(file_handler)
        if not self.main_config.globalConfig.to_dict()["nologging"]:
            str_handler = logging.StreamHandler()
            str_handler.setFormatter(log_formatter_simple)
            self.logger.addHandler(str_handler)

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

        # logging other modules
        for logger in ("message", "database", "storage", "tsdb"):
            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(log_formatter_simple)
                logger_module.addHandler(file_handler)
            if logger_config["loglevel"]:
                logger_module.setLevel(logger_config["loglevel"])
        self.logger.critical("starting osm/nglcm")

        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))

    async def start(self):
        # do some temporal stuff here
        temporal_api = (
            f"{self.main_config.temporal.host}:{str(self.main_config.temporal.port)}"
        )
        self.logger.info(f"Attempting to register with Temporal at {temporal_api}")
        client = await Client.connect(temporal_api)

        task_queue = "lcm-task-queue"
        workflows = [
            Heartbeat,
        ]
        activities = []

        worker = Worker(
            client, task_queue=task_queue, workflows=workflows, activities=activities
        )

        self.logger.info(f"Registered for queue {task_queue}")
        self.logger.info(f"Registered workflows {workflows}")
        self.logger.info(f"Registered activites {activities}")

        await worker.run()

    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)


@workflow.defn
class Heartbeat:
    @workflow.run
    async def run(self) -> str:
        return "alive"


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)