X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Flcm.py;h=dd7c4ec213e22f34ea82a0eba4be7e86ed1a6d7c;hb=843adbc77237767f62b198e9cbb5ffb4c7caf17d;hp=b306720246b9e4fa72c90ae522c8a0fb6a191f12;hpb=94f0611a61f3f115bed849887b7c538f40f11c6d;p=osm%2FLCM.git

diff --git a/osm_lcm/lcm.py b/osm_lcm/lcm.py
index b306720..dd7c4ec 100644
--- a/osm_lcm/lcm.py
+++ b/osm_lcm/lcm.py
@@ -29,10 +29,9 @@ import logging.handlers
 import getopt
 import sys
 
-from osm_lcm import ns
-from osm_lcm import vim_sdn
-from osm_lcm import netslice
-from osm_lcm import ROclient
+from osm_lcm import ns, prometheus, vim_sdn, netslice
+from osm_lcm.ng_ro import NgRoException, NgRoClient
+from osm_lcm.ROclient import ROClient, ROClientException
 
 from time import time
 from osm_lcm.lcm_utils import versiontuple, LcmException, TaskRegistry, LcmExceptionExit
@@ -86,11 +85,18 @@ class Lcm:
         config = self.read_config_file(config_file)
         self.config = config
         self.config["ro_config"] = {
-            "endpoint_url": "http://{}:{}/openmano".format(config["RO"]["host"], config["RO"]["port"]),
+            "ng": config["RO"].get("ng", False),
+            "uri": config["RO"].get("uri"),
             "tenant": config.get("tenant", "osm"),
-            "logger_name": "lcm.ROclient",
-            "loglevel": "ERROR",
+            "logger_name": "lcm.roclient",
+            "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.loop = loop or asyncio.get_event_loop()
 
@@ -184,8 +190,13 @@ class Lcm:
         # 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.config, self.loop)
-        self.netslice = netslice.NetsliceLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
+        if self.config.get("prometheus"):
+            self.prometheus = prometheus.Prometheus(self.config["prometheus"], self.worker_id, self.db, self.loop)
+        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)
@@ -197,17 +208,19 @@ class Lcm:
         last_error = None
         while True:
             try:
-                ro_server = ROclient.ROClient(self.loop, **self.config["ro_config"])
+                if self.config["ro_config"].get("ng"):
+                    ro_server = NgRoClient(self.loop, **self.config["ro_config"])
+                else:
+                    ro_server = 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(
                         ro_version, min_RO_version))
                 self.logger.info("Connected to RO version {}".format(ro_version))
                 return
-            except ROclient.ROClientException as e:
+            except (ROClientException, NgRoException) as e:
                 tries -= 1
-                error_text = "Error while connecting to RO on {}: {}".format(self.config["ro_config"]["endpoint_url"],
-                                                                             e)
+                error_text = "Error while connecting to RO on {}: {}".format(self.config["ro_config"]["uri"], e)
                 if tries <= 0:
                     self.logger.critical(error_text)
                     raise LcmException(error_text)
@@ -470,7 +483,7 @@ class Lcm:
                 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)
                 )
 
@@ -496,6 +509,10 @@ class Lcm:
         # check RO version
         self.loop.run_until_complete(self.check_RO_version())
 
+        # configure Prometheus
+        if self.prometheus:
+            self.loop.run_until_complete(self.prometheus.start())
+
         self.loop.run_until_complete(asyncio.gather(
             self.kafka_read(),
             self.kafka_ping()