# ENV OSMLCM_VCA_PUBKEY pubkey
# ENV OSMLCM_VCA_CACERT cacert
+# VCA - k8s
+ENV OSMLCM_VCA_HELMPATH /usr/local/bin/helm
+ENV OSMLCM_VCA_KUBECTLPATH /usr/bin/kubectl
+ENV OSMLCM_VCA_JUJUPATH /usr/local/bin/juju
+
# database
ENV OSMLCM_DATABASE_DRIVER mongo
ENV OSMLCM_DATABASE_URI mongodb://mongo:27017
ENV OSMLCM_MESSAGE_HOST kafka
ENV OSMLCM_MESSAGE_PORT 9092
-# k8s
-ENV OSMLCM_VCA_HELMPATH /usr/local/bin/helm
-ENV OSMLCM_VCA_KUBECTLPATH /usr/bin/kubectl
-ENV OSMLCM_VCA_JUJUPATH /usr/local/bin/juju
# logs
# ENV OSMLCM_GLOBAL_LOGFILE /app/log/lcm.log
ENV OSMLCM_GLOBAL_LOGLEVEL DEBUG
+# timeouts
+# ENV OSMLCM_TIMEOUT_NS_DEPLOY 7200
+# ENV OSMLCM_TIMEOUT_NSI_DEPLOY 7200
+
# Copy the current directory contents into the container at /app/LCM
ADD . /app/LCM
# logfile: /app/log # or /var/log/osm/lcm.log
# nologging: True # do no log to stdout/stderr
+#[timeout]
+timeout:
+ # ns_deploy: 7200 # total deploy timeout for a ns 2 hours
+ # nsi_deploy: 7200 # total deploy timeout for a nsi 2 hours
+
#[RO]
RO:
host: ro # hostname or IP
# load configuration
config = self.read_config_file(config_file)
self.config = config
- self.ro_config = {
+ self.config["ro_config"] = {
"endpoint_url": "http://{}:{}/openmano".format(config["RO"]["host"], config["RO"]["port"]),
"tenant": config.get("tenant", "osm"),
"logger_name": "lcm.ROclient",
"loglevel": "ERROR",
}
- self.vca_config = config["VCA"]
-
self.loop = loop or asyncio.get_event_loop()
# logging
# contains created tasks/futures to be able to cancel
self.lcm_tasks = TaskRegistry(self.worker_id, self.db, self.logger)
- self.ns = ns.NsLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.ro_config, self.vca_config, self.loop)
- self.netslice = netslice.NetsliceLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.ro_config,
- self.vca_config, self.loop)
- self.vim = vim_sdn.VimLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.ro_config, self.loop)
- self.wim = vim_sdn.WimLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.ro_config, self.loop)
- self.sdn = vim_sdn.SdnLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.ro_config, self.loop)
- self.k8scluster = vim_sdn.K8sClusterLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.vca_config, self.loop)
- self.k8srepo = vim_sdn.K8sRepoLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.vca_config, self.loop)
+ self.ns = ns.NsLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
+ self.netslice = netslice.NetsliceLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
+ self.vim = vim_sdn.VimLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
+ self.wim = vim_sdn.WimLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
+ self.sdn = vim_sdn.SdnLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
+ self.k8scluster = vim_sdn.K8sClusterLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
+ self.k8srepo = vim_sdn.K8sRepoLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
async def check_RO_version(self):
tries = 14
last_error = None
while True:
try:
- ro_server = ROclient.ROClient(self.loop, **self.ro_config)
+ ro_server = ROclient.ROClient(self.loop, **self.config["ro_config"])
ro_version = await ro_server.get_version()
if versiontuple(ro_version) < versiontuple(min_RO_version):
raise LcmException("Not compatible osm/RO version '{}'. Needed '{}' or higher".format(
return
except ROclient.ROClientException as e:
tries -= 1
- error_text = "Error while connecting to RO on {}: {}".format(self.ro_config["endpoint_url"], e)
+ error_text = "Error while connecting to RO on {}: {}".format(self.config["ro_config"]["endpoint_url"],
+ e)
if tries <= 0:
self.logger.critical(error_text)
raise LcmException(error_text)
# the configparser library is not suitable, because it does not admit comments at the end of line,
# and not parse integer or boolean
try:
+ # read file as yaml format
with open(config_file) as f:
conf = yaml.load(f, Loader=yaml.Loader)
+ # Ensure all sections are not empty
+ for k in ("global", "timeout", "RO", "VCA", "database", "storage", "message"):
+ if not conf.get(k):
+ conf[k] = {}
+
+ # read all environ that starts with OSMLCM_
for k, v in environ.items():
if not k.startswith("OSMLCM_"):
continue
- k_items = k.lower().split("_")
- if len(k_items) < 3:
+ subject, _, item = k[7:].lower().partition("_")
+ if not item:
continue
- if k_items[1] in ("ro", "vca"):
+ if subject in ("ro", "vca"):
# put in capital letter
- k_items[1] = k_items[1].upper()
- c = conf
+ subject = subject.upper()
try:
- for k_item in k_items[1:-1]:
- c = c[k_item]
- if k_items[-1] == "port":
- c[k_items[-1]] = int(v)
+ if item == "port" or subject == "timeout":
+ conf[subject][item] = int(v)
else:
- c[k_items[-1]] = v
+ conf[subject][item] = v
except Exception as e:
- self.logger.warn("skipping environ '{}' on exception '{}'".format(k, e))
+ self.logger.warning("skipping environ '{}' on exception '{}'".format(k, e))
+
+ # backward compatibility of VCA parameters
+
+ if 'pubkey' in conf["VCA"]:
+ conf["VCA"]['public_key'] = conf["VCA"].pop('pubkey')
+ if 'cacert' in conf["VCA"]:
+ conf["VCA"]['ca_cert'] = conf["VCA"].pop('cacert')
+ if 'apiproxy' in conf["VCA"]:
+ conf["VCA"]['api_proxy'] = conf["VCA"].pop('apiproxy')
+
+ if 'enableosupgrade' in conf["VCA"]:
+ conf["VCA"]['enable_os_upgrade'] = conf["VCA"].pop('enableosupgrade')
+ if isinstance(conf["VCA"].get('enable_os_upgrade'), str):
+ if conf["VCA"]['enable_os_upgrade'].lower() == 'false':
+ conf["VCA"]['enable_os_upgrade'] = False
+ elif conf["VCA"]['enable_os_upgrade'].lower() == 'true':
+ conf["VCA"]['enable_os_upgrade'] = True
+
+ if 'aptmirror' in conf["VCA"]:
+ conf["VCA"]['apt_mirror'] = conf["VCA"].pop('aptmirror')
return conf
except Exception as e:
return tuple(filled)
-def deep_get(target_dict, key_list):
+def deep_get(target_dict, key_list, default_value=None):
"""
Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
Example target_dict={a: {b: 5}}; key_list=[a,b] returns 5; both key_list=[a,b,c] and key_list=[f,h] return None
:param target_dict: dictionary to be read
:param key_list: list of keys to read from target_dict
+ :param default_value: value to return if key is not present in the nested dictionary
:return: The wanted value if exist, None otherwise
"""
for key in key_list:
if not isinstance(target_dict, dict) or key not in target_dict:
- return None
+ return default_value
target_dict = target_dict[key]
return target_dict
-# LcmBase must be listed before TaskRegistry, as it is a dependency.
+def get_iterable(in_dict, in_key):
+ """
+ Similar to <dict>.get(), but if value is None, False, ..., An empty tuple is returned instead
+ :param in_dict: a dictionary
+ :param in_key: the key to look for at in_dict
+ :return: in_dict[in_var] or () if it is None or not present
+ """
+ if not in_dict.get(in_key):
+ return ()
+ return in_dict[in_key]
+
+
+def populate_dict(target_dict, key_list, value):
+ """
+ Update target_dict creating nested dictionaries with the key_list. Last key_list item is asigned the value.
+ Example target_dict={K: J}; key_list=[a,b,c]; target_dict will be {K: J, a: {b: {c: value}}}
+ :param target_dict: dictionary to be changed
+ :param key_list: list of keys to insert at target_dict
+ :param value:
+ :return: None
+ """
+ for key in key_list[0:-1]:
+ if key not in target_dict:
+ target_dict[key] = {}
+ target_dict = target_dict[key]
+ target_dict[key_list[-1]] = value
+
+
class LcmBase:
def __init__(self, db, msg, fs, logger):
import logging
import logging.handlers
import traceback
-from osm_lcm.ns import populate_dict as populate_dict
from osm_lcm import ROclient, ns
-from osm_lcm.lcm_utils import LcmException, LcmBase
+from osm_lcm.lcm_utils import LcmException, LcmBase, populate_dict, get_iterable, deep_get
from osm_common.dbbase import DbException
from time import time
from copy import deepcopy
__author__ = "Felipe Vicens, Pol Alemany, Alfonso Tierno"
-def get_iterable(in_dict, in_key):
- """
- Similar to <dict>.get(), but if value is None, False, ..., An empty tuple is returned instead
- :param in_dict: a dictionary
- :param in_key: the key to look for at in_dict
- :return: in_dict[in_var] or () if it is None or not present
- """
- if not in_dict.get(in_key):
- return ()
- return in_dict[in_key]
-
-
class NetsliceLcm(LcmBase):
- total_deploy_timeout = 2 * 3600 # global timeout for deployment
+ timeout_nsi_deploy = 2 * 3600 # default global timeout for deployment a nsi
- def __init__(self, db, msg, fs, lcm_tasks, ro_config, vca_config, loop):
+ def __init__(self, db, msg, fs, 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',
self.logger = logging.getLogger('lcm.netslice')
self.loop = loop
self.lcm_tasks = lcm_tasks
- self.ns = ns.NsLcm(db, msg, fs, lcm_tasks, ro_config, vca_config, loop)
- self.ro_config = ro_config
+ self.ns = ns.NsLcm(db, msg, fs, lcm_tasks, config, loop)
+ self.ro_config = config["ro_config"]
+ self.timeout = config["timeout"]
super().__init__(db, msg, fs, self.logger)
break
# Creating netslice-vld at RO
- RO_nsir = db_nsir["_admin"].get("deployed", {}).get("RO", [])
+ RO_nsir = deep_get(db_nsir, ("_admin", "deployed", "RO"), [])
if vld_id in RO_nsir:
db_nsir_update["_admin.deployed.RO"] = RO_nsir
step = "Getting nsilcmop={} from db".format(nsilcmop_id)
db_nsilcmop = self.db.get_one("nsilcmops", {"_id": nsilcmop_id})
+ start_deploy = time()
+ nsi_params = db_nsilcmop.get("operationParams")
+ if nsi_params and nsi_params.get("timeout_nsi_deploy"):
+ timeout_nsi_deploy = nsi_params["timeout_nsi_deploy"]
+ else:
+ timeout_nsi_deploy = self.timeout.get("nsi_deploy", self.timeout_nsi_deploy)
+
# Empty list to keep track of network service records status in the netslice
nsir_admin = db_nsir_admin = db_nsir.get("_admin")
self.logger.debug(logging_text + step)
# TODO: substitute while for await (all task to be done or not)
- deployment_timeout = 2 * 3600 # Two hours
- while deployment_timeout > 0:
+ while time() <= start_deploy + timeout_nsi_deploy:
# Check ns instantiation status
nsi_ready = True
nsir = self.db.get_one("nsis", {"_id": nsir_id})
# TODO: future improvement due to synchronism -> await asyncio.wait(vca_task_list, timeout=300)
await asyncio.sleep(5, loop=self.loop)
- deployment_timeout -= 5
- if deployment_timeout <= 0:
+ else: # timeout_nsi_deploy reached:
raise LcmException("Timeout waiting nsi to be ready. nsi_id={}".format(nsir_id))
db_nsir_update["operational-status"] = "running"
from jinja2 import Environment, Template, meta, TemplateError, TemplateNotFound, TemplateSyntaxError
from osm_lcm import ROclient
-from osm_lcm.lcm_utils import LcmException, LcmExceptionNoMgmtIP, LcmBase, deep_get
+from osm_lcm.lcm_utils import LcmException, LcmExceptionNoMgmtIP, LcmBase, deep_get, get_iterable, populate_dict
from n2vc.k8s_helm_conn import K8sHelmConnector
from n2vc.k8s_juju_conn import K8sJujuConnector
__author__ = "Alfonso Tierno"
-def get_iterable(in_dict, in_key):
- """
- Similar to <dict>.get(), but if value is None, False, ..., An empty tuple is returned instead
- :param in_dict: a dictionary
- :param in_key: the key to look for at in_dict
- :return: in_dict[in_var] or () if it is None or not present
- """
- if not in_dict.get(in_key):
- return ()
- return in_dict[in_key]
-
-
-def populate_dict(target_dict, key_list, value):
- """
- Update target_dict creating nested dictionaries with the key_list. Last key_list item is asigned the value.
- Example target_dict={K: J}; key_list=[a,b,c]; target_dict will be {K: J, a: {b: {c: value}}}
- :param target_dict: dictionary to be changed
- :param key_list: list of keys to insert at target_dict
- :param value:
- :return: None
- """
- for key in key_list[0:-1]:
- if key not in target_dict:
- target_dict[key] = {}
- target_dict = target_dict[key]
- target_dict[key_list[-1]] = value
-
-
class NsLcm(LcmBase):
timeout_vca_on_error = 5 * 60 # Time for charm from first time at blocked,error status to mark as failed
- total_deploy_timeout = 2 * 3600 # global timeout for deployment
+ timeout_ns_deploy = 2 * 3600 # default global timeout for deployment a ns
timeout_charm_delete = 10 * 60
timeout_primitive = 10 * 60 # timeout for primitive execution
SUBOPERATION_STATUS_NEW = -2
SUBOPERATION_STATUS_SKIP = -3
- def __init__(self, db, msg, fs, lcm_tasks, ro_config, vca_config, loop):
+ def __init__(self, db, msg, fs, 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',
self.loop = loop
self.lcm_tasks = lcm_tasks
- self.ro_config = ro_config
- self.vca_config = vca_config
- if 'pubkey' in self.vca_config:
- self.vca_config['public_key'] = self.vca_config['pubkey']
- if 'cacert' in self.vca_config:
- self.vca_config['ca_cert'] = self.vca_config['cacert']
- if 'apiproxy' in self.vca_config:
- self.vca_config['api_proxy'] = self.vca_config['apiproxy']
- if 'enableosupgrade' in self.vca_config:
- if self.vca_config['enableosupgrade'].lower() == 'false':
- self.vca_config['enable_os_upgrade'] = False
- elif self.vca_config['enableosupgrade'].lower() == 'true':
- self.vca_config['enable_os_upgrade'] = True
- if 'aptmirror' in self.vca_config:
- self.vca_config['apt_mirror'] = self.vca_config['aptmirror']
+ self.timeout = config["timeout"]
+ self.ro_config = config["ro_config"]
+ self.vca_config = config["VCA"].copy()
# create N2VC connector
self.n2vc = N2VCJujuConnector(
start_deploy = time()
vdu_flag = False # If any of the VNFDs has VDUs
ns_params = db_nslcmop.get("operationParams")
+ if ns_params and ns_params.get("timeout_ns_deploy"):
+ timeout_ns_deploy = ns_params["timeout_ns_deploy"]
+ else:
+ timeout_ns_deploy = self.timeout.get("ns_deploy", self.timeout_ns_deploy)
# deploy RO
self.logger.debug(logging_text + step)
old_desc = None
- while time() <= start_deploy + self.total_deploy_timeout:
+ while time() <= start_deploy + timeout_ns_deploy:
desc = await self.RO.show("ns", RO_nsr_id)
# deploymentStatus
detailed_status_old = db_nsr_update["_admin.deployed.RO.detailed-status"] = detailed_status
self.update_db_2("nsrs", nsr_id, db_nsr_update)
await asyncio.sleep(5, loop=self.loop)
- else: # total_deploy_timeout
+ else: # timeout_ns_deploy
raise ROclient.ROClientException("Timeout waiting ns to be ready")
step = "Updating NSR"
# read from db: operation
step = "Getting nslcmop={} from db".format(nslcmop_id)
db_nslcmop = self.db.get_one("nslcmops", {"_id": nslcmop_id})
+ ns_params = db_nslcmop.get("operationParams")
+ if ns_params and ns_params.get("timeout_ns_deploy"):
+ timeout_ns_deploy = ns_params["timeout_ns_deploy"]
+ else:
+ timeout_ns_deploy = self.timeout.get("ns_deploy", self.timeout_ns_deploy)
# read from db: ns
step = "Getting nsr={} from db".format(nsr_id)
# Wait until all tasks of "task_instantiation_list" have been finished
- # while time() <= start_deploy + self.total_deploy_timeout:
error_text_list = []
- timeout = 3600
# let's begin with all OK
instantiated_ok = True
if task_instantiation_list:
# wait for all tasks completion
- done, pending = await asyncio.wait(task_instantiation_list, timeout=timeout)
+ done, pending = await asyncio.wait(task_instantiation_list, timeout=timeout_ns_deploy)
for task in pending:
instantiated_ok = False
OSMLCM_RO_XXX: configuration of RO
"""
-
-vca_config = { # TODO replace with os.get_env to get other configurations
- "host": getenv("OSMLCM_VCA_HOST", "vca"),
- "port": getenv("OSMLCM_VCA_PORT", 17070),
- "user": getenv("OSMLCM_VCA_USER", "admin"),
- "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
- "pubkey": getenv("OSMLCM_VCA_PUBKEY", None),
- 'cacert': getenv("OSMLCM_VCA_CACERT", None)
-}
-
-ro_config = {
- "endpoint_url": "http://{}:{}/openmano".format(getenv("OSMLCM_RO_HOST", "ro"), getenv("OSMLCM_RO_PORT", "9090")),
- "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
- "logger_name": "lcm.ROclient",
- "loglevel": "DEBUG",
+lcm_config = {
+ "timeout": {},
+ "VCA": { # TODO replace with os.get_env to get other configurations
+ "host": getenv("OSMLCM_VCA_HOST", "vca"),
+ "port": getenv("OSMLCM_VCA_PORT", 17070),
+ "user": getenv("OSMLCM_VCA_USER", "admin"),
+ "secret": getenv("OSMLCM_VCA_SECRET", "vca"),
+ "public_key": getenv("OSMLCM_VCA_PUBKEY", None),
+ 'ca_cert': getenv("OSMLCM_VCA_CACERT", None)
+ },
+ "ro_config": {
+ "endpoint_url": "http://{}:{}/openmano".format(getenv("OSMLCM_RO_HOST", "ro"),
+ getenv("OSMLCM_RO_PORT", "9090")),
+ "tenant": getenv("OSMLCM_RO_TENANT", "osm"),
+ "logger_name": "lcm.ROclient",
+ "loglevel": "DEBUG",
+ }
}
self.lcm_tasks.lookfor_related.return_value = ("", [])
# Create NsLCM class
- self.my_ns = NsLcm(self.db, self.msg, self.fs, self.lcm_tasks, ro_config, vca_config, self.loop)
+ self.my_ns = NsLcm(self.db, self.msg, self.fs, self.lcm_tasks, lcm_config, self.loop)
self.my_ns._wait_dependent_n2vc = asynctest.CoroutineMock()
# Mock logging
# Mock RO
if not getenv("OSMLCMTEST_RO_NOMOCK"):
- # self.my_ns.RO = asynctest.Mock(ROclient.ROClient(self.loop, **ro_config))
+ # self.my_ns.RO = asynctest.Mock(ROclient.ROClient(self.loop, **lcm_config["ro_config"]))
# TODO first time should be empty list, following should return a dict
self.my_ns.RO.get_list = asynctest.CoroutineMock(self.my_ns.RO.get_list, return_value=[])
self.my_ns.RO.create = asynctest.CoroutineMock(self.my_ns.RO.create, side_effect=self._ro_create())
vim_config_encrypted = {"1.1": ("admin_password", "nsx_password", "vcenter_password"),
"default": ("admin_password", "nsx_password", "vcenter_password", "vrops_password")}
- def __init__(self, db, msg, fs, lcm_tasks, ro_config, loop):
+ def __init__(self, db, msg, fs, 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',
self.logger = logging.getLogger('lcm.vim')
self.loop = loop
self.lcm_tasks = lcm_tasks
- self.ro_config = ro_config
+ self.ro_config = config["ro_config"]
super().__init__(db, msg, fs, self.logger)
# values that are encrypted at wim config because they are passwords
wim_config_encrypted = ()
- def __init__(self, db, msg, fs, lcm_tasks, ro_config, loop):
+ def __init__(self, db, msg, fs, 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',
self.logger = logging.getLogger('lcm.vim')
self.loop = loop
self.lcm_tasks = lcm_tasks
- self.ro_config = ro_config
+ self.ro_config = config["ro_config"]
super().__init__(db, msg, fs, self.logger)
class SdnLcm(LcmBase):
- def __init__(self, db, msg, fs, lcm_tasks, ro_config, loop):
+ def __init__(self, db, msg, fs, 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',
self.logger = logging.getLogger('lcm.sdn')
self.loop = loop
self.lcm_tasks = lcm_tasks
- self.ro_config = ro_config
+ self.ro_config = config["ro_config"]
super().__init__(db, msg, fs, self.logger)
class K8sClusterLcm(LcmBase):
- def __init__(self, db, msg, fs, lcm_tasks, vca_config, loop):
+ def __init__(self, db, msg, fs, 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',
self.logger = logging.getLogger('lcm.k8scluster')
self.loop = loop
self.lcm_tasks = lcm_tasks
- self.vca_config = vca_config
+ self.vca_config = config["VCA"]
self.fs = fs
self.db = db
class K8sRepoLcm(LcmBase):
- def __init__(self, db, msg, fs, lcm_tasks, vca_config, loop):
+ def __init__(self, db, msg, fs, 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',
self.logger = logging.getLogger('lcm.k8srepo')
self.loop = loop
self.lcm_tasks = lcm_tasks
- self.vca_config = vca_config
+ self.vca_config = config["VCA"]
self.fs = fs
self.db = db
projects / osm / LCM.git / commitdiff