blob: a3c57b3dd9e5736b7bdb38baae0a3ba667b8f0e6 [file] [log] [blame]
#!/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, VnfPrepareWorkflow
from osm_lcm.temporal.vnf_activities import (
VnfDbActivity,
VnfOperations,
VnfSendNotifications,
)
from osm_lcm.temporal.ns_workflows import NsInstantiateWorkflow
from osm_lcm.temporal.ns_activities import NsOperations, NsDbActivity
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"])
logging.getLogger("juju.client.connection").setLevel(logging.CRITICAL)
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)
ns_operation_instance = NsOperations(self.db)
ns_data_activity_instance = NsDbActivity(self.db)
nslcm_activity_instance = NsLcmActivity(self.db)
paas_connector_instance = JujuPaasConnector(self.db)
vim_data_activity_instance = VimDbActivity(self.db)
vnf_data_activity_instance = VnfDbActivity(self.db)
vnf_operation_instance = VnfOperations(self.db)
vnf_send_notifications_instance = VnfSendNotifications()
workflows = [
NsInstantiateWorkflow,
NsNoOpWorkflow,
VimCreateWorkflow,
VimDeleteWorkflow,
VimUpdateWorkflow,
VduInstantiateWorkflow,
VnfInstantiateWorkflow,
VnfPrepareWorkflow,
]
activities = [
ns_data_activity_instance.get_model_info,
ns_data_activity_instance.update_ns_state,
ns_operation_instance.check_ns_instantiate_finished,
ns_operation_instance.deploy_ns,
nslcm_activity_instance.update_ns_lcm_operation_state,
nslcm_activity_instance.no_op,
paas_connector_instance.create_model,
paas_connector_instance.deploy_charm,
paas_connector_instance.check_charm_status,
paas_connector_instance.test_vim_connectivity,
vim_data_activity_instance.update_vim_operation_state,
vim_data_activity_instance.update_vim_state,
vim_data_activity_instance.delete_vim_record,
vnf_data_activity_instance.change_vnf_state,
vnf_data_activity_instance.change_vnf_instantiation_state,
vnf_operation_instance.get_task_queue,
vnf_operation_instance.get_vnf_details,
vnf_send_notifications_instance.send_notification_for_vnf,
vnf_data_activity_instance.set_vnf_model,
]
# 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)