2 # -*- coding: utf-8 -*-
7 import logging
.handlers
13 from lcm_utils
import versiontuple
, LcmException
, TaskRegistry
15 # from osm_lcm import version as lcm_version, version_date as lcm_version_date, ROclient
16 from osm_common
import dbmemory
, dbmongo
, fslocal
, msglocal
, msgkafka
17 from osm_common
import version
as common_version
18 from osm_common
.dbbase
import DbException
19 from osm_common
.fsbase
import FsException
20 from osm_common
.msgbase
import MsgException
21 from os
import environ
, path
22 from n2vc
import version
as n2vc_version
25 __author__
= "Alfonso Tierno"
26 min_RO_version
= [0, 5, 72]
27 min_n2vc_version
= "0.0.2"
28 min_common_version
= "0.1.7"
29 # uncomment if LCM is installed as library and installed, and get them from __init__.py
30 lcm_version
= '0.1.17'
31 lcm_version_date
= '2018-10-11'
36 def __init__(self
, config_file
, loop
=None):
38 Init, Connect to database, filesystem storage, and messaging
39 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
46 self
.pings_not_received
= 1
48 # contains created tasks/futures to be able to cancel
49 self
.lcm_tasks
= TaskRegistry()
51 self
.logger
= logging
.getLogger('lcm')
53 config
= self
.read_config_file(config_file
)
56 "endpoint_url": "http://{}:{}/openmano".format(config
["RO"]["host"], config
["RO"]["port"]),
57 "tenant": config
.get("tenant", "osm"),
58 "logger_name": "lcm.ROclient",
62 self
.vca_config
= config
["VCA"]
64 self
.loop
= loop
or asyncio
.get_event_loop()
67 log_format_simple
= "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
68 log_formatter_simple
= logging
.Formatter(log_format_simple
, datefmt
='%Y-%m-%dT%H:%M:%S')
69 config
["database"]["logger_name"] = "lcm.db"
70 config
["storage"]["logger_name"] = "lcm.fs"
71 config
["message"]["logger_name"] = "lcm.msg"
72 if config
["global"].get("logfile"):
73 file_handler
= logging
.handlers
.RotatingFileHandler(config
["global"]["logfile"],
74 maxBytes
=100e6
, backupCount
=9, delay
=0)
75 file_handler
.setFormatter(log_formatter_simple
)
76 self
.logger
.addHandler(file_handler
)
77 if not config
["global"].get("nologging"):
78 str_handler
= logging
.StreamHandler()
79 str_handler
.setFormatter(log_formatter_simple
)
80 self
.logger
.addHandler(str_handler
)
82 if config
["global"].get("loglevel"):
83 self
.logger
.setLevel(config
["global"]["loglevel"])
85 # logging other modules
86 for k1
, logname
in {"message": "lcm.msg", "database": "lcm.db", "storage": "lcm.fs"}.items():
87 config
[k1
]["logger_name"] = logname
88 logger_module
= logging
.getLogger(logname
)
89 if config
[k1
].get("logfile"):
90 file_handler
= logging
.handlers
.RotatingFileHandler(config
[k1
]["logfile"],
91 maxBytes
=100e6
, backupCount
=9, delay
=0)
92 file_handler
.setFormatter(log_formatter_simple
)
93 logger_module
.addHandler(file_handler
)
94 if config
[k1
].get("loglevel"):
95 logger_module
.setLevel(config
[k1
]["loglevel"])
96 self
.logger
.critical("starting osm/lcm version {} {}".format(lcm_version
, lcm_version_date
))
98 # check version of N2VC
99 # TODO enhance with int conversion or from distutils.version import LooseVersion
100 # or with list(map(int, version.split(".")))
101 if versiontuple(n2vc_version
) < versiontuple(min_n2vc_version
):
102 raise LcmException("Not compatible osm/N2VC version '{}'. Needed '{}' or higher".format(
103 n2vc_version
, min_n2vc_version
))
104 # check version of common
105 if versiontuple(common_version
) < versiontuple("0.1.7"):
106 raise LcmException("Not compatible osm/common version '{}'. Needed '{}' or higher".format(
107 common_version
, min_common_version
))
110 # TODO check database version
111 if config
["database"]["driver"] == "mongo":
112 self
.db
= dbmongo
.DbMongo()
113 self
.db
.db_connect(config
["database"])
114 elif config
["database"]["driver"] == "memory":
115 self
.db
= dbmemory
.DbMemory()
116 self
.db
.db_connect(config
["database"])
118 raise LcmException("Invalid configuration param '{}' at '[database]':'driver'".format(
119 config
["database"]["driver"]))
121 if config
["storage"]["driver"] == "local":
122 self
.fs
= fslocal
.FsLocal()
123 self
.fs
.fs_connect(config
["storage"])
125 raise LcmException("Invalid configuration param '{}' at '[storage]':'driver'".format(
126 config
["storage"]["driver"]))
128 if config
["message"]["driver"] == "local":
129 self
.msg
= msglocal
.MsgLocal()
130 self
.msg
.connect(config
["message"])
131 elif config
["message"]["driver"] == "kafka":
132 self
.msg
= msgkafka
.MsgKafka()
133 self
.msg
.connect(config
["message"])
135 raise LcmException("Invalid configuration param '{}' at '[message]':'driver'".format(
136 config
["storage"]["driver"]))
137 except (DbException
, FsException
, MsgException
) as e
:
138 self
.logger
.critical(str(e
), exc_info
=True)
139 raise LcmException(str(e
))
141 self
.ns
= ns
.NsLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.vca_config
, self
.loop
)
142 self
.vim
= vim_sdn
.VimLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.loop
)
143 self
.sdn
= vim_sdn
.SdnLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.loop
)
145 async def check_RO_version(self
):
147 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
148 RO_version
= await RO
.get_version()
149 if RO_version
< min_RO_version
:
150 raise LcmException("Not compatible osm/RO version '{}.{}.{}'. Needed '{}.{}.{}' or higher".format(
151 *RO_version
, *min_RO_version
153 except ROclient
.ROClientException
as e
:
154 error_text
= "Error while conneting to osm/RO " + str(e
)
155 self
.logger
.critical(error_text
, exc_info
=True)
156 raise LcmException(error_text
)
158 async def test(self
, param
=None):
159 self
.logger
.debug("Starting/Ending test task: {}".format(param
))
161 async def kafka_ping(self
):
162 self
.logger
.debug("Task kafka_ping Enter")
163 consecutive_errors
= 0
165 kafka_has_received
= False
166 self
.pings_not_received
= 1
169 await self
.msg
.aiowrite("admin", "ping", {"from": "lcm", "to": "lcm"}, self
.loop
)
170 # time between pings are low when it is not received and at starting
171 wait_time
= 5 if not kafka_has_received
else 120
172 if not self
.pings_not_received
:
173 kafka_has_received
= True
174 self
.pings_not_received
+= 1
175 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
176 if self
.pings_not_received
> 10:
177 raise LcmException("It is not receiving pings from Kafka bus")
178 consecutive_errors
= 0
182 except Exception as e
:
183 # if not first_start is the first time after starting. So leave more time and wait
184 # to allow kafka starts
185 if consecutive_errors
== 8 if not first_start
else 30:
186 self
.logger
.error("Task kafka_read task exit error too many errors. Exception: {}".format(e
))
188 consecutive_errors
+= 1
189 self
.logger
.error("Task kafka_read retrying after Exception {}".format(e
))
190 wait_time
= 1 if not first_start
else 5
191 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
193 async def kafka_read(self
):
194 self
.logger
.debug("Task kafka_read Enter")
196 # future = asyncio.Future()
197 consecutive_errors
= 0
199 while consecutive_errors
< 10:
201 topics
= ("admin", "ns", "vim_account", "sdn")
202 topic
, command
, params
= await self
.msg
.aioread(topics
, self
.loop
)
203 if topic
!= "admin" and command
!= "ping":
204 self
.logger
.debug("Task kafka_read receives {} {}: {}".format(topic
, command
, params
))
205 consecutive_errors
= 0
208 if command
== "exit":
211 elif command
.startswith("#"):
213 elif command
== "echo":
218 elif command
== "test":
219 asyncio
.Task(self
.test(params
), loop
=self
.loop
)
223 if command
== "ping" and params
["to"] == "lcm" and params
["from"] == "lcm":
224 self
.pings_not_received
= 0
227 if command
== "instantiate":
228 # self.logger.debug("Deploying NS {}".format(nsr_id))
230 nslcmop_id
= nslcmop
["_id"]
231 nsr_id
= nslcmop
["nsInstanceId"]
232 task
= asyncio
.ensure_future(self
.ns
.instantiate(nsr_id
, nslcmop_id
))
233 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_instantiate", task
)
235 elif command
== "terminate":
236 # self.logger.debug("Deleting NS {}".format(nsr_id))
238 nslcmop_id
= nslcmop
["_id"]
239 nsr_id
= nslcmop
["nsInstanceId"]
240 self
.lcm_tasks
.cancel(topic
, nsr_id
)
241 task
= asyncio
.ensure_future(self
.ns
.terminate(nsr_id
, nslcmop_id
))
242 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_terminate", task
)
244 elif command
== "action":
245 # self.logger.debug("Update NS {}".format(nsr_id))
247 nslcmop_id
= nslcmop
["_id"]
248 nsr_id
= nslcmop
["nsInstanceId"]
249 task
= asyncio
.ensure_future(self
.ns
.action(nsr_id
, nslcmop_id
))
250 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_action", task
)
252 elif command
== "scale":
253 # self.logger.debug("Update NS {}".format(nsr_id))
255 nslcmop_id
= nslcmop
["_id"]
256 nsr_id
= nslcmop
["nsInstanceId"]
257 task
= asyncio
.ensure_future(self
.ns
.scale(nsr_id
, nslcmop_id
))
258 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_scale", task
)
260 elif command
== "show":
262 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
263 print("nsr:\n _id={}\n operational-status: {}\n config-status: {}"
264 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
265 "".format(nsr_id
, db_nsr
["operational-status"], db_nsr
["config-status"],
266 db_nsr
["detailed-status"],
267 db_nsr
["_admin"]["deployed"], self
.lcm_ns_tasks
.get(nsr_id
)))
268 except Exception as e
:
269 print("nsr {} not found: {}".format(nsr_id
, e
))
272 elif command
== "deleted":
273 continue # TODO cleaning of task just in case should be done
274 elif command
in ("terminated", "instantiated", "scaled", "actioned"): # "scaled-cooldown-time"
276 elif topic
== "vim_account":
277 vim_id
= params
["_id"]
278 if command
== "create":
279 task
= asyncio
.ensure_future(self
.vim
.create(params
, order_id
))
280 self
.lcm_tasks
.register("vim_account", vim_id
, order_id
, "vim_create", task
)
282 elif command
== "delete":
283 self
.lcm_tasks
.cancel(topic
, vim_id
)
284 task
= asyncio
.ensure_future(self
.vim
.delete(vim_id
, order_id
))
285 self
.lcm_tasks
.register("vim_account", vim_id
, order_id
, "vim_delete", task
)
287 elif command
== "show":
288 print("not implemented show with vim_account")
291 elif command
== "edit":
292 task
= asyncio
.ensure_future(self
.vim
.edit(params
, order_id
))
293 self
.lcm_tasks
.register("vim_account", vim_id
, order_id
, "vim_edit", task
)
296 _sdn_id
= params
["_id"]
297 if command
== "create":
298 task
= asyncio
.ensure_future(self
.sdn
.create(params
, order_id
))
299 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_create", task
)
301 elif command
== "delete":
302 self
.lcm_tasks
.cancel(topic
, _sdn_id
)
303 task
= asyncio
.ensure_future(self
.sdn
.delete(_sdn_id
, order_id
))
304 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_delete", task
)
306 elif command
== "edit":
307 task
= asyncio
.ensure_future(self
.sdn
.edit(params
, order_id
))
308 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_edit", task
)
310 self
.logger
.critical("unknown topic {} and command '{}'".format(topic
, command
))
311 except Exception as e
:
312 # if not first_start is the first time after starting. So leave more time and wait
313 # to allow kafka starts
314 if consecutive_errors
== 8 if not first_start
else 30:
315 self
.logger
.error("Task kafka_read task exit error too many errors. Exception: {}".format(e
))
317 consecutive_errors
+= 1
318 self
.logger
.error("Task kafka_read retrying after Exception {}".format(e
))
319 wait_time
= 2 if not first_start
else 5
320 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
322 # self.logger.debug("Task kafka_read terminating")
323 self
.logger
.debug("Task kafka_read exit")
328 self
.loop
.run_until_complete(self
.check_RO_version())
330 self
.loop
.run_until_complete(asyncio
.gather(
335 # self.logger.debug("Terminating cancelling creation tasks")
336 # self.lcm_tasks.cancel("ALL", "create")
338 # while self.is_pending_tasks():
339 # self.logger.debug("Task kafka_read terminating. Waiting for tasks termination")
340 # await asyncio.sleep(2, loop=self.loop)
343 # self.lcm_tasks.cancel("ALL", "ALL")
347 self
.db
.db_disconnect()
349 self
.msg
.disconnect()
351 self
.fs
.fs_disconnect()
353 def read_config_file(self
, config_file
):
354 # TODO make a [ini] + yaml inside parser
355 # the configparser library is not suitable, because it does not admit comments at the end of line,
356 # and not parse integer or boolean
358 with
open(config_file
) as f
:
360 for k
, v
in environ
.items():
361 if not k
.startswith("OSMLCM_"):
363 k_items
= k
.lower().split("_")
366 for k_item
in k_items
[1:-1]:
367 if k_item
in ("ro", "vca"):
368 # put in capital letter
369 k_item
= k_item
.upper()
371 if k_items
[-1] == "port":
372 c
[k_items
[-1]] = int(v
)
375 except Exception as e
:
376 self
.logger
.warn("skipping environ '{}' on exception '{}'".format(k
, e
))
379 except Exception as e
:
380 self
.logger
.critical("At config file '{}': {}".format(config_file
, e
))
385 print("""Usage: {} [options]
386 -c|--config [configuration_file]: loads the configuration file (default: ./nbi.cfg)
387 -h|--help: shows this help
388 """.format(sys
.argv
[0]))
389 # --log-socket-host HOST: send logs to this host")
390 # --log-socket-port PORT: send logs using this port (default: 9022)")
393 if __name__
== '__main__':
395 # load parameters and configuration
396 opts
, args
= getopt
.getopt(sys
.argv
[1:], "hc:", ["config=", "help"])
397 # TODO add "log-socket-host=", "log-socket-port=", "log-file="
400 if o
in ("-h", "--help"):
403 elif o
in ("-c", "--config"):
405 # elif o == "--log-socket-port":
406 # log_socket_port = a
407 # elif o == "--log-socket-host":
408 # log_socket_host = a
409 # elif o == "--log-file":
412 assert False, "Unhandled option"
414 if not path
.isfile(config_file
):
415 print("configuration file '{}' that not exist".format(config_file
), file=sys
.stderr
)
418 for config_file
in (__file__
[:__file__
.rfind(".")] + ".cfg", "./lcm.cfg", "/etc/osm/lcm.cfg"):
419 if path
.isfile(config_file
):
422 print("No configuration file 'nbi.cfg' found neither at local folder nor at /etc/osm/", file=sys
.stderr
)
424 lcm
= Lcm(config_file
)
426 except (LcmException
, getopt
.GetoptError
) as e
:
427 print(str(e
), file=sys
.stderr
)