d967c03854436ea3b39770330a99d529ed74867d
2 # -*- coding: utf-8 -*-
5 # Copyright 2018 Telefonica S.A.
7 # Licensed under the Apache License, Version 2.0 (the "License"); you may
8 # not use this file except in compliance with the License. You may obtain
9 # a copy of the License at
11 # http://www.apache.org/licenses/LICENSE-2.0
13 # Unless required by applicable law or agreed to in writing, software
14 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
15 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16 # License for the specific language governing permissions and limitations
23 import logging
.handlers
30 from lcm_utils
import versiontuple
, LcmException
, TaskRegistry
, LcmExceptionExit
32 # from osm_lcm import version as lcm_version, version_date as lcm_version_date, ROclient
33 from osm_common
import dbmemory
, dbmongo
, fslocal
, msglocal
, msgkafka
34 from osm_common
import version
as common_version
35 from osm_common
.dbbase
import DbException
36 from osm_common
.fsbase
import FsException
37 from osm_common
.msgbase
import MsgException
38 from os
import environ
, path
39 from n2vc
import version
as n2vc_version
42 __author__
= "Alfonso Tierno"
43 min_RO_version
= [0, 6, 0]
44 min_n2vc_version
= "0.0.2"
45 min_common_version
= "0.1.11"
46 # uncomment if LCM is installed as library and installed, and get them from __init__.py
47 lcm_version
= '0.1.31'
48 lcm_version_date
= '2019-01-16'
53 ping_interval_pace
= 120 # how many time ping is send once is confirmed all is running
54 ping_interval_boot
= 5 # how many time ping is sent when booting
56 def __init__(self
, config_file
, loop
=None):
58 Init, Connect to database, filesystem storage, and messaging
59 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
66 self
.pings_not_received
= 1
68 # contains created tasks/futures to be able to cancel
69 self
.lcm_tasks
= TaskRegistry()
71 self
.logger
= logging
.getLogger('lcm')
73 config
= self
.read_config_file(config_file
)
76 "endpoint_url": "http://{}:{}/openmano".format(config
["RO"]["host"], config
["RO"]["port"]),
77 "tenant": config
.get("tenant", "osm"),
78 "logger_name": "lcm.ROclient",
82 self
.vca_config
= config
["VCA"]
84 self
.loop
= loop
or asyncio
.get_event_loop()
87 log_format_simple
= "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
88 log_formatter_simple
= logging
.Formatter(log_format_simple
, datefmt
='%Y-%m-%dT%H:%M:%S')
89 config
["database"]["logger_name"] = "lcm.db"
90 config
["storage"]["logger_name"] = "lcm.fs"
91 config
["message"]["logger_name"] = "lcm.msg"
92 if config
["global"].get("logfile"):
93 file_handler
= logging
.handlers
.RotatingFileHandler(config
["global"]["logfile"],
94 maxBytes
=100e6
, backupCount
=9, delay
=0)
95 file_handler
.setFormatter(log_formatter_simple
)
96 self
.logger
.addHandler(file_handler
)
97 if not config
["global"].get("nologging"):
98 str_handler
= logging
.StreamHandler()
99 str_handler
.setFormatter(log_formatter_simple
)
100 self
.logger
.addHandler(str_handler
)
102 if config
["global"].get("loglevel"):
103 self
.logger
.setLevel(config
["global"]["loglevel"])
105 # logging other modules
106 for k1
, logname
in {"message": "lcm.msg", "database": "lcm.db", "storage": "lcm.fs"}.items():
107 config
[k1
]["logger_name"] = logname
108 logger_module
= logging
.getLogger(logname
)
109 if config
[k1
].get("logfile"):
110 file_handler
= logging
.handlers
.RotatingFileHandler(config
[k1
]["logfile"],
111 maxBytes
=100e6
, backupCount
=9, delay
=0)
112 file_handler
.setFormatter(log_formatter_simple
)
113 logger_module
.addHandler(file_handler
)
114 if config
[k1
].get("loglevel"):
115 logger_module
.setLevel(config
[k1
]["loglevel"])
116 self
.logger
.critical("starting osm/lcm version {} {}".format(lcm_version
, lcm_version_date
))
118 # check version of N2VC
119 # TODO enhance with int conversion or from distutils.version import LooseVersion
120 # or with list(map(int, version.split(".")))
121 if versiontuple(n2vc_version
) < versiontuple(min_n2vc_version
):
122 raise LcmException("Not compatible osm/N2VC version '{}'. Needed '{}' or higher".format(
123 n2vc_version
, min_n2vc_version
))
124 # check version of common
125 if versiontuple(common_version
) < versiontuple(min_common_version
):
126 raise LcmException("Not compatible osm/common version '{}'. Needed '{}' or higher".format(
127 common_version
, min_common_version
))
130 # TODO check database version
131 if config
["database"]["driver"] == "mongo":
132 self
.db
= dbmongo
.DbMongo()
133 self
.db
.db_connect(config
["database"])
134 elif config
["database"]["driver"] == "memory":
135 self
.db
= dbmemory
.DbMemory()
136 self
.db
.db_connect(config
["database"])
138 raise LcmException("Invalid configuration param '{}' at '[database]':'driver'".format(
139 config
["database"]["driver"]))
141 if config
["storage"]["driver"] == "local":
142 self
.fs
= fslocal
.FsLocal()
143 self
.fs
.fs_connect(config
["storage"])
145 raise LcmException("Invalid configuration param '{}' at '[storage]':'driver'".format(
146 config
["storage"]["driver"]))
148 if config
["message"]["driver"] == "local":
149 self
.msg
= msglocal
.MsgLocal()
150 self
.msg
.connect(config
["message"])
151 elif config
["message"]["driver"] == "kafka":
152 self
.msg
= msgkafka
.MsgKafka()
153 self
.msg
.connect(config
["message"])
155 raise LcmException("Invalid configuration param '{}' at '[message]':'driver'".format(
156 config
["storage"]["driver"]))
157 except (DbException
, FsException
, MsgException
) as e
:
158 self
.logger
.critical(str(e
), exc_info
=True)
159 raise LcmException(str(e
))
161 self
.ns
= ns
.NsLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.vca_config
, self
.loop
)
162 self
.netslice
= netslice
.NetsliceLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
,
163 self
.vca_config
, self
.loop
)
164 self
.vim
= vim_sdn
.VimLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.loop
)
165 self
.wim
= vim_sdn
.WimLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.loop
)
166 self
.sdn
= vim_sdn
.SdnLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.loop
)
168 async def check_RO_version(self
):
170 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
171 RO_version
= await RO
.get_version()
172 if RO_version
< min_RO_version
:
173 raise LcmException("Not compatible osm/RO version '{}.{}.{}'. Needed '{}.{}.{}' or higher".format(
174 *RO_version
, *min_RO_version
176 except ROclient
.ROClientException
as e
:
177 error_text
= "Error while conneting to osm/RO " + str(e
)
178 self
.logger
.critical(error_text
, exc_info
=True)
179 raise LcmException(error_text
)
181 async def test(self
, param
=None):
182 self
.logger
.debug("Starting/Ending test task: {}".format(param
))
184 async def kafka_ping(self
):
185 self
.logger
.debug("Task kafka_ping Enter")
186 consecutive_errors
= 0
188 kafka_has_received
= False
189 self
.pings_not_received
= 1
192 await self
.msg
.aiowrite("admin", "ping", {"from": "lcm", "to": "lcm"}, self
.loop
)
193 # time between pings are low when it is not received and at starting
194 wait_time
= self
.ping_interval_boot
if not kafka_has_received
else self
.ping_interval_pace
195 if not self
.pings_not_received
:
196 kafka_has_received
= True
197 self
.pings_not_received
+= 1
198 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
199 if self
.pings_not_received
> 10:
200 raise LcmException("It is not receiving pings from Kafka bus")
201 consecutive_errors
= 0
205 except Exception as e
:
206 # if not first_start is the first time after starting. So leave more time and wait
207 # to allow kafka starts
208 if consecutive_errors
== 8 if not first_start
else 30:
209 self
.logger
.error("Task kafka_read task exit error too many errors. Exception: {}".format(e
))
211 consecutive_errors
+= 1
212 self
.logger
.error("Task kafka_read retrying after Exception {}".format(e
))
213 wait_time
= 1 if not first_start
else 5
214 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
216 def kafka_read_callback(self
, topic
, command
, params
):
219 if topic
!= "admin" and command
!= "ping":
220 self
.logger
.debug("Task kafka_read receives {} {}: {}".format(topic
, command
, params
))
221 self
.consecutive_errors
= 0
222 self
.first_start
= False
224 if command
== "exit":
225 raise LcmExceptionExit
226 elif command
.startswith("#"):
228 elif command
== "echo":
233 elif command
== "test":
234 asyncio
.Task(self
.test(params
), loop
=self
.loop
)
238 if command
== "ping" and params
["to"] == "lcm" and params
["from"] == "lcm":
239 self
.pings_not_received
= 0
242 if command
== "instantiate":
243 # self.logger.debug("Deploying NS {}".format(nsr_id))
245 nslcmop_id
= nslcmop
["_id"]
246 nsr_id
= nslcmop
["nsInstanceId"]
247 task
= asyncio
.ensure_future(self
.ns
.instantiate(nsr_id
, nslcmop_id
))
248 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_instantiate", task
)
250 elif command
== "terminate":
251 # self.logger.debug("Deleting NS {}".format(nsr_id))
253 nslcmop_id
= nslcmop
["_id"]
254 nsr_id
= nslcmop
["nsInstanceId"]
255 self
.lcm_tasks
.cancel(topic
, nsr_id
)
256 task
= asyncio
.ensure_future(self
.ns
.terminate(nsr_id
, nslcmop_id
))
257 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_terminate", task
)
259 elif command
== "action":
260 # self.logger.debug("Update NS {}".format(nsr_id))
262 nslcmop_id
= nslcmop
["_id"]
263 nsr_id
= nslcmop
["nsInstanceId"]
264 task
= asyncio
.ensure_future(self
.ns
.action(nsr_id
, nslcmop_id
))
265 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_action", task
)
267 elif command
== "scale":
268 # self.logger.debug("Update NS {}".format(nsr_id))
270 nslcmop_id
= nslcmop
["_id"]
271 nsr_id
= nslcmop
["nsInstanceId"]
272 task
= asyncio
.ensure_future(self
.ns
.scale(nsr_id
, nslcmop_id
))
273 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_scale", task
)
275 elif command
== "show":
277 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
278 print("nsr:\n _id={}\n operational-status: {}\n config-status: {}"
279 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
280 "".format(nsr_id
, db_nsr
["operational-status"], db_nsr
["config-status"],
281 db_nsr
["detailed-status"],
282 db_nsr
["_admin"]["deployed"], self
.lcm_ns_tasks
.get(nsr_id
)))
283 except Exception as e
:
284 print("nsr {} not found: {}".format(nsr_id
, e
))
287 elif command
== "deleted":
288 return # TODO cleaning of task just in case should be done
289 elif command
in ("terminated", "instantiated", "scaled", "actioned"): # "scaled-cooldown-time"
291 elif topic
== "nsi": # netslice LCM processes (instantiate, terminate, etc)
292 if command
== "instantiate":
293 # self.logger.debug("Instantiating Network Slice {}".format(nsilcmop["netsliceInstanceId"]))
295 nsilcmop_id
= nsilcmop
["_id"] # slice operation id
296 nsir_id
= nsilcmop
["netsliceInstanceId"] # slice record id
297 task
= asyncio
.ensure_future(self
.netslice
.instantiate(nsir_id
, nsilcmop_id
))
298 self
.lcm_tasks
.register("nsi", nsir_id
, nsilcmop_id
, "nsi_instantiate", task
)
300 elif command
== "terminate":
301 # self.logger.debug("Terminating Network Slice NS {}".format(nsilcmop["netsliceInstanceId"]))
303 nsilcmop_id
= nsilcmop
["_id"] # slice operation id
304 nsir_id
= nsilcmop
["netsliceInstanceId"] # slice record id
305 self
.lcm_tasks
.cancel(topic
, nsir_id
)
306 task
= asyncio
.ensure_future(self
.netslice
.terminate(nsir_id
, nsilcmop_id
))
307 self
.lcm_tasks
.register("nsi", nsir_id
, nsilcmop_id
, "nsi_terminate", task
)
309 elif command
== "show":
311 db_nsir
= self
.db
.get_one("nsirs", {"_id": nsir_id
})
312 print("nsir:\n _id={}\n operational-status: {}\n config-status: {}"
313 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
314 "".format(nsir_id
, db_nsir
["operational-status"], db_nsir
["config-status"],
315 db_nsir
["detailed-status"],
316 db_nsir
["_admin"]["deployed"], self
.lcm_netslice_tasks
.get(nsir_id
)))
317 except Exception as e
:
318 print("nsir {} not found: {}".format(nsir_id
, e
))
321 elif command
== "deleted":
322 return # TODO cleaning of task just in case should be done
323 elif command
in ("terminated", "instantiated", "scaled", "actioned"): # "scaled-cooldown-time"
325 elif topic
== "vim_account":
326 vim_id
= params
["_id"]
327 if command
== "create":
328 task
= asyncio
.ensure_future(self
.vim
.create(params
, order_id
))
329 self
.lcm_tasks
.register("vim_account", vim_id
, order_id
, "vim_create", task
)
331 elif command
== "delete":
332 self
.lcm_tasks
.cancel(topic
, vim_id
)
333 task
= asyncio
.ensure_future(self
.vim
.delete(vim_id
, order_id
))
334 self
.lcm_tasks
.register("vim_account", vim_id
, order_id
, "vim_delete", task
)
336 elif command
== "show":
337 print("not implemented show with vim_account")
340 elif command
== "edit":
341 task
= asyncio
.ensure_future(self
.vim
.edit(params
, order_id
))
342 self
.lcm_tasks
.register("vim_account", vim_id
, order_id
, "vim_edit", task
)
344 elif topic
== "wim_account":
345 wim_id
= params
["_id"]
346 if command
== "create":
347 task
= asyncio
.ensure_future(self
.wim
.create(params
, order_id
))
348 self
.lcm_tasks
.register("wim_account", wim_id
, order_id
, "wim_create", task
)
350 elif command
== "delete":
351 self
.lcm_tasks
.cancel(topic
, wim_id
)
352 task
= asyncio
.ensure_future(self
.wim
.delete(wim_id
, order_id
))
353 self
.lcm_tasks
.register("wim_account", wim_id
, order_id
, "wim_delete", task
)
355 elif command
== "show":
356 print("not implemented show with wim_account")
359 elif command
== "edit":
360 task
= asyncio
.ensure_future(self
.wim
.edit(params
, order_id
))
361 self
.lcm_tasks
.register("wim_account", wim_id
, order_id
, "wim_edit", task
)
364 _sdn_id
= params
["_id"]
365 if command
== "create":
366 task
= asyncio
.ensure_future(self
.sdn
.create(params
, order_id
))
367 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_create", task
)
369 elif command
== "delete":
370 self
.lcm_tasks
.cancel(topic
, _sdn_id
)
371 task
= asyncio
.ensure_future(self
.sdn
.delete(_sdn_id
, order_id
))
372 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_delete", task
)
374 elif command
== "edit":
375 task
= asyncio
.ensure_future(self
.sdn
.edit(params
, order_id
))
376 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_edit", task
)
378 self
.logger
.critical("unknown topic {} and command '{}'".format(topic
, command
))
380 async def kafka_read(self
):
381 self
.logger
.debug("Task kafka_read Enter")
382 # future = asyncio.Future()
383 self
.consecutive_errors
= 0
384 self
.first_start
= True
385 while self
.consecutive_errors
< 10:
387 topics
= ("admin", "ns", "vim_account", "wim_account", "sdn", "nsi")
388 await self
.msg
.aioread(topics
, self
.loop
, self
.kafka_read_callback
)
390 except LcmExceptionExit
:
391 self
.logger
.debug("Bye!")
393 except Exception as e
:
394 # if not first_start is the first time after starting. So leave more time and wait
395 # to allow kafka starts
396 if self
.consecutive_errors
== 8 if not self
.first_start
else 30:
397 self
.logger
.error("Task kafka_read task exit error too many errors. Exception: {}".format(e
))
399 self
.consecutive_errors
+= 1
400 self
.logger
.error("Task kafka_read retrying after Exception {}".format(e
))
401 wait_time
= 2 if not self
.first_start
else 5
402 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
404 # self.logger.debug("Task kafka_read terminating")
405 self
.logger
.debug("Task kafka_read exit")
407 def health_check(self
):
413 def health_check_callback(topic
, command
, params
):
415 print("receiving callback {} {} {}".format(topic
, command
, params
))
416 if topic
== "admin" and command
== "ping" and params
["to"] == "lcm" and params
["from"] == "lcm":
417 # print("received LCM ping")
422 task
= asyncio
.ensure_future(self
.msg
.aioread(("admin",), self
.loop
, health_check_callback
))
423 self
.loop
.run_until_complete(task
)
431 self
.loop
.run_until_complete(self
.check_RO_version())
433 self
.loop
.run_until_complete(asyncio
.gather(
438 # self.logger.debug("Terminating cancelling creation tasks")
439 # self.lcm_tasks.cancel("ALL", "create")
441 # while self.is_pending_tasks():
442 # self.logger.debug("Task kafka_read terminating. Waiting for tasks termination")
443 # await asyncio.sleep(2, loop=self.loop)
446 # self.lcm_tasks.cancel("ALL", "ALL")
450 self
.db
.db_disconnect()
452 self
.msg
.disconnect()
454 self
.fs
.fs_disconnect()
456 def read_config_file(self
, config_file
):
457 # TODO make a [ini] + yaml inside parser
458 # the configparser library is not suitable, because it does not admit comments at the end of line,
459 # and not parse integer or boolean
461 with
open(config_file
) as f
:
463 for k
, v
in environ
.items():
464 if not k
.startswith("OSMLCM_"):
466 k_items
= k
.lower().split("_")
469 if k_items
[1] in ("ro", "vca"):
470 # put in capital letter
471 k_items
[1] = k_items
[1].upper()
474 for k_item
in k_items
[1:-1]:
476 if k_items
[-1] == "port":
477 c
[k_items
[-1]] = int(v
)
480 except Exception as e
:
481 self
.logger
.warn("skipping environ '{}' on exception '{}'".format(k
, e
))
484 except Exception as e
:
485 self
.logger
.critical("At config file '{}': {}".format(config_file
, e
))
490 print("""Usage: {} [options]
491 -c|--config [configuration_file]: loads the configuration file (default: ./nbi.cfg)
492 --health-check: do not run lcm, but inspect kafka bus to determine if lcm is healthy
493 -h|--help: shows this help
494 """.format(sys
.argv
[0]))
495 # --log-socket-host HOST: send logs to this host")
496 # --log-socket-port PORT: send logs using this port (default: 9022)")
499 if __name__
== '__main__':
501 # load parameters and configuration
502 opts
, args
= getopt
.getopt(sys
.argv
[1:], "hc:", ["config=", "help", "health-check"])
503 # TODO add "log-socket-host=", "log-socket-port=", "log-file="
507 if o
in ("-h", "--help"):
510 elif o
in ("-c", "--config"):
512 elif o
== "--health-check":
514 # elif o == "--log-socket-port":
515 # log_socket_port = a
516 # elif o == "--log-socket-host":
517 # log_socket_host = a
518 # elif o == "--log-file":
521 assert False, "Unhandled option"
523 if not path
.isfile(config_file
):
524 print("configuration file '{}' not exist".format(config_file
), file=sys
.stderr
)
527 for config_file
in (__file__
[:__file__
.rfind(".")] + ".cfg", "./lcm.cfg", "/etc/osm/lcm.cfg"):
528 if path
.isfile(config_file
):
531 print("No configuration file 'lcm.cfg' found neither at local folder nor at /etc/osm/", file=sys
.stderr
)
533 lcm
= Lcm(config_file
)
538 except (LcmException
, getopt
.GetoptError
) as e
:
539 print(str(e
), file=sys
.stderr
)