ping_interval_pace = 120 # how many time ping is send once is confirmed all is running
ping_interval_boot = 5 # how many time ping is sent when booting
+ cfg_logger_name = {"message": "lcm.msg", "database": "lcm.db", "storage": "lcm.fs", "tsdb": "lcm.prometheus"}
+ # ^ contains for each section at lcm.cfg the used logger name
def __init__(self, config_file, loop=None):
"""
"loglevel": config["RO"].get("loglevel", "ERROR"),
}
if not self.config["ro_config"]["uri"]:
- if not self.config["ro_config"]["ng"]:
- self.config["ro_config"]["uri"] = "http://{}:{}/openmano".format(config["RO"]["host"],
- config["RO"]["port"])
- else:
- self.config["ro_config"]["uri"] = "http://{}:{}/ro".format(config["RO"]["host"], config["RO"]["port"])
+ self.config["ro_config"]["uri"] = "http://{}:{}/".format(config["RO"]["host"], config["RO"]["port"])
+ elif "/ro" in self.config["ro_config"]["uri"][-4:] or "/openmano" in self.config["ro_config"]["uri"][-10:]:
+ # uri ends with '/ro', '/ro/', '/openmano', '/openmano/'
+ index = self.config["ro_config"]["uri"][-1].rfind("/")
+ self.config["ro_config"]["uri"] = self.config["ro_config"]["uri"][index+1]
self.loop = loop or asyncio.get_event_loop()
+ self.ns = self.netslice = self.vim = self.wim = self.sdn = self.k8scluster = self.k8srepo = None
# logging
log_format_simple = "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
self.logger.setLevel(config["global"]["loglevel"])
# logging other modules
- for k1, logname in {"message": "lcm.msg", "database": "lcm.db", "storage": "lcm.fs"}.items():
+ for k1, logname in self.cfg_logger_name.items():
config[k1]["logger_name"] = logname
logger_module = logging.getLogger(logname)
if config[k1].get("logfile"):
# contains created tasks/futures to be able to cancel
self.lcm_tasks = TaskRegistry(self.worker_id, self.db, self.logger)
- if self.config.get("prometheus"):
- self.prometheus = prometheus.Prometheus(self.config["prometheus"], self.worker_id, self.db, self.loop)
+ if self.config.get("tsdb") and self.config["tsdb"].get("driver"):
+ if self.config["tsdb"]["driver"] == "prometheus":
+ self.prometheus = prometheus.Prometheus(self.config["tsdb"], self.worker_id, self.db, self.loop)
+ else:
+ raise LcmException("Invalid configuration param '{}' at '[tsdb]':'driver'".format(
+ config["tsdb"]["driver"]))
else:
self.prometheus = None
- self.ns = ns.NsLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop, self.prometheus)
- self.netslice = netslice.NetsliceLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop,
- self.ns)
- 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:
+ ro_uri = self.config["ro_config"]["uri"]
try:
- if self.config["ro_config"].get("ng"):
+ # try new RO, if fail old RO
+ try:
+ self.config["ro_config"]["uri"] = ro_uri + "ro"
ro_server = NgRoClient(self.loop, **self.config["ro_config"])
- else:
+ ro_version = await ro_server.get_version()
+ self.config["ro_config"]["ng"] = True
+ except Exception:
+ self.config["ro_config"]["uri"] = ro_uri + "openmano"
ro_server = ROClient(self.loop, **self.config["ro_config"])
- ro_version = await ro_server.get_version()
+ ro_version = await ro_server.get_version()
+ self.config["ro_config"]["ng"] = False
if versiontuple(ro_version) < versiontuple(min_RO_version):
raise LcmException("Not compatible osm/RO version '{}'. Needed '{}' or higher".format(
ro_version, min_RO_version))
- self.logger.info("Connected to RO version {}".format(ro_version))
+ self.logger.info("Connected to RO version {} new-generation version {}".
+ format(ro_version, self.config["ro_config"]["ng"]))
return
except (ROClientException, NgRoException) as e:
+ self.config["ro_config"]["uri"] = ro_uri
tries -= 1
error_text = "Error while connecting to RO on {}: {}".format(self.config["ro_config"]["uri"], e)
if tries <= 0:
elif topic == "vim_account":
vim_id = params["_id"]
if command in ("create", "created"):
- task = asyncio.ensure_future(self.vim.create(params, order_id))
- self.lcm_tasks.register("vim_account", vim_id, order_id, "vim_create", task)
+ if not self.config["ro_config"].get("ng"):
+ task = asyncio.ensure_future(self.vim.create(params, order_id))
+ self.lcm_tasks.register("vim_account", vim_id, order_id, "vim_create", task)
return
elif command == "delete" or command == "deleted":
self.lcm_tasks.cancel(topic, vim_id)
sys.stdout.flush()
return
elif command in ("edit", "edited"):
- task = asyncio.ensure_future(self.vim.edit(params, order_id))
- self.lcm_tasks.register("vim_account", vim_id, order_id, "vim_edit", task)
+ if not self.config["ro_config"].get("ng"):
+ task = asyncio.ensure_future(self.vim.edit(params, order_id))
+ self.lcm_tasks.register("vim_account", vim_id, order_id, "vim_edit", task)
return
elif command == "deleted":
return # TODO cleaning of task just in case should be done
elif topic == "wim_account":
wim_id = params["_id"]
if command in ("create", "created"):
- task = asyncio.ensure_future(self.wim.create(params, order_id))
- self.lcm_tasks.register("wim_account", wim_id, order_id, "wim_create", task)
+ if not self.config["ro_config"].get("ng"):
+ task = asyncio.ensure_future(self.wim.create(params, order_id))
+ self.lcm_tasks.register("wim_account", wim_id, order_id, "wim_create", task)
return
elif command == "delete" or command == "deleted":
self.lcm_tasks.cancel(topic, wim_id)
elif topic == "sdn":
_sdn_id = params["_id"]
if command in ("create", "created"):
- task = asyncio.ensure_future(self.sdn.create(params, order_id))
- self.lcm_tasks.register("sdn", _sdn_id, order_id, "sdn_create", task)
+ if not self.config["ro_config"].get("ng"):
+ task = asyncio.ensure_future(self.sdn.create(params, order_id))
+ self.lcm_tasks.register("sdn", _sdn_id, order_id, "sdn_create", task)
return
elif command == "delete" or command == "deleted":
self.lcm_tasks.cancel(topic, _sdn_id)
topics = ("ns", "vim_account", "wim_account", "sdn", "nsi", "k8scluster", "k8srepo", "pla")
topics_admin = ("admin", )
await asyncio.gather(
- self.msg.aioread(topics, self.loop, self.kafka_read_callback),
+ self.msg.aioread(topics, self.loop, self.kafka_read_callback, from_beginning=True),
self.msg_admin.aioread(topics_admin, self.loop, self.kafka_read_callback, group_id=False)
)
# check RO version
self.loop.run_until_complete(self.check_RO_version())
- # configure Prometheus
+ self.ns = ns.NsLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop, self.prometheus)
+ self.netslice = netslice.NetsliceLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop,
+ self.ns)
+ 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)
+
+ # configure tsdb prometheus
if self.prometheus:
self.loop.run_until_complete(self.prometheus.start())
projects / osm / LCM.git / blobdiff