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
27 from osm_lcm
import ROclient
, ns
, vim_sdn
, netslice
28 from time
import time
, sleep
29 from osm_lcm
.lcm_utils
import versiontuple
, LcmException
, TaskRegistry
, LcmExceptionExit
30 from osm_lcm
import version
as lcm_version
, version_date
as lcm_version_date
32 from osm_common
import dbmemory
, dbmongo
, fslocal
, msglocal
, msgkafka
33 from osm_common
import version
as common_version
34 from osm_common
.dbbase
import DbException
35 from osm_common
.fsbase
import FsException
36 from osm_common
.msgbase
import MsgException
37 from os
import environ
, path
38 from random
import choice
as random_choice
39 from n2vc
import version
as n2vc_version
42 __author__
= "Alfonso Tierno"
43 min_RO_version
= "6.0.2"
44 min_n2vc_version
= "0.0.2"
45 min_common_version
= "0.1.19"
46 # uncomment if LCM is installed as library and installed, and get them from __init__.py
47 # lcm_version = '0.1.41'
48 # lcm_version_date = '2019-06-19'
49 health_check_file
= path
.expanduser("~") + "/time_last_ping" # TODO find better location for this file
54 ping_interval_pace
= 120 # how many time ping is send once is confirmed all is running
55 ping_interval_boot
= 5 # how many time ping is sent when booting
57 def __init__(self
, config_file
, loop
=None):
59 Init, Connect to database, filesystem storage, and messaging
60 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
68 self
.pings_not_received
= 1
69 self
.consecutive_errors
= 0
70 self
.first_start
= False
73 self
.logger
= logging
.getLogger('lcm')
75 self
.worker_id
= self
.get_process_id()
77 config
= self
.read_config_file(config_file
)
80 "endpoint_url": "http://{}:{}/openmano".format(config
["RO"]["host"], config
["RO"]["port"]),
81 "tenant": config
.get("tenant", "osm"),
82 "logger_name": "lcm.ROclient",
86 self
.vca_config
= config
["VCA"]
88 self
.loop
= loop
or asyncio
.get_event_loop()
91 log_format_simple
= "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
92 log_formatter_simple
= logging
.Formatter(log_format_simple
, datefmt
='%Y-%m-%dT%H:%M:%S')
93 config
["database"]["logger_name"] = "lcm.db"
94 config
["storage"]["logger_name"] = "lcm.fs"
95 config
["message"]["logger_name"] = "lcm.msg"
96 if config
["global"].get("logfile"):
97 file_handler
= logging
.handlers
.RotatingFileHandler(config
["global"]["logfile"],
98 maxBytes
=100e6
, backupCount
=9, delay
=0)
99 file_handler
.setFormatter(log_formatter_simple
)
100 self
.logger
.addHandler(file_handler
)
101 if not config
["global"].get("nologging"):
102 str_handler
= logging
.StreamHandler()
103 str_handler
.setFormatter(log_formatter_simple
)
104 self
.logger
.addHandler(str_handler
)
106 if config
["global"].get("loglevel"):
107 self
.logger
.setLevel(config
["global"]["loglevel"])
109 # logging other modules
110 for k1
, logname
in {"message": "lcm.msg", "database": "lcm.db", "storage": "lcm.fs"}.items():
111 config
[k1
]["logger_name"] = logname
112 logger_module
= logging
.getLogger(logname
)
113 if config
[k1
].get("logfile"):
114 file_handler
= logging
.handlers
.RotatingFileHandler(config
[k1
]["logfile"],
115 maxBytes
=100e6
, backupCount
=9, delay
=0)
116 file_handler
.setFormatter(log_formatter_simple
)
117 logger_module
.addHandler(file_handler
)
118 if config
[k1
].get("loglevel"):
119 logger_module
.setLevel(config
[k1
]["loglevel"])
120 self
.logger
.critical("starting osm/lcm version {} {}".format(lcm_version
, lcm_version_date
))
122 # check version of N2VC
123 # TODO enhance with int conversion or from distutils.version import LooseVersion
124 # or with list(map(int, version.split(".")))
125 if versiontuple(n2vc_version
) < versiontuple(min_n2vc_version
):
126 raise LcmException("Not compatible osm/N2VC version '{}'. Needed '{}' or higher".format(
127 n2vc_version
, min_n2vc_version
))
128 # check version of common
129 if versiontuple(common_version
) < versiontuple(min_common_version
):
130 raise LcmException("Not compatible osm/common version '{}'. Needed '{}' or higher".format(
131 common_version
, min_common_version
))
134 # TODO check database version
135 if config
["database"]["driver"] == "mongo":
136 self
.db
= dbmongo
.DbMongo()
137 self
.db
.db_connect(config
["database"])
138 elif config
["database"]["driver"] == "memory":
139 self
.db
= dbmemory
.DbMemory()
140 self
.db
.db_connect(config
["database"])
142 raise LcmException("Invalid configuration param '{}' at '[database]':'driver'".format(
143 config
["database"]["driver"]))
145 if config
["storage"]["driver"] == "local":
146 self
.fs
= fslocal
.FsLocal()
147 self
.fs
.fs_connect(config
["storage"])
149 raise LcmException("Invalid configuration param '{}' at '[storage]':'driver'".format(
150 config
["storage"]["driver"]))
152 config_message
= config
["message"].copy()
153 config_message
["loop"] = self
.loop
154 if config_message
["driver"] == "local":
155 self
.msg
= msglocal
.MsgLocal()
156 self
.msg
.connect(config_message
)
157 self
.msg_admin
= msglocal
.MsgLocal()
158 config_message
.pop("group_id", None)
159 self
.msg_admin
.connect(config_message
)
160 elif config_message
["driver"] == "kafka":
161 self
.msg
= msgkafka
.MsgKafka()
162 self
.msg
.connect(config_message
)
163 self
.msg_admin
= msgkafka
.MsgKafka()
164 config_message
.pop("group_id", None)
165 self
.msg_admin
.connect(config_message
)
167 raise LcmException("Invalid configuration param '{}' at '[message]':'driver'".format(
168 config
["message"]["driver"]))
169 except (DbException
, FsException
, MsgException
) as e
:
170 self
.logger
.critical(str(e
), exc_info
=True)
171 raise LcmException(str(e
))
173 # contains created tasks/futures to be able to cancel
174 self
.lcm_tasks
= TaskRegistry(self
.worker_id
, self
.db
, self
.logger
)
176 self
.ns
= ns
.NsLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.vca_config
, self
.loop
)
177 self
.netslice
= netslice
.NetsliceLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
,
178 self
.vca_config
, self
.loop
)
179 self
.vim
= vim_sdn
.VimLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.loop
)
180 self
.wim
= vim_sdn
.WimLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.loop
)
181 self
.sdn
= vim_sdn
.SdnLcm(self
.db
, self
.msg
, self
.fs
, self
.lcm_tasks
, self
.ro_config
, self
.loop
)
183 async def check_RO_version(self
):
188 ro_server
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
189 ro_version
= await ro_server
.get_version()
190 if versiontuple(ro_version
) < versiontuple(min_RO_version
):
191 raise LcmException("Not compatible osm/RO version '{}'. Needed '{}' or higher".format(
192 ro_version
, min_RO_version
))
193 self
.logger
.info("Connected to RO version {}".format(ro_version
))
195 except ROclient
.ROClientException
as e
:
197 error_text
= "Error while connecting to RO on {}: {}".format(self
.ro_config
["endpoint_url"], e
)
199 self
.logger
.critical(error_text
)
200 raise LcmException(error_text
)
201 if last_error
!= error_text
:
202 last_error
= error_text
203 self
.logger
.error(error_text
+ ". Waiting until {} seconds".format(5*tries
))
204 await asyncio
.sleep(5)
206 async def test(self
, param
=None):
207 self
.logger
.debug("Starting/Ending test task: {}".format(param
))
209 async def kafka_ping(self
):
210 self
.logger
.debug("Task kafka_ping Enter")
211 consecutive_errors
= 0
213 kafka_has_received
= False
214 self
.pings_not_received
= 1
217 await self
.msg_admin
.aiowrite(
219 {"from": "lcm", "to": "lcm", "worker_id": self
.worker_id
, "version": lcm_version
},
221 # time between pings are low when it is not received and at starting
222 wait_time
= self
.ping_interval_boot
if not kafka_has_received
else self
.ping_interval_pace
223 if not self
.pings_not_received
:
224 kafka_has_received
= True
225 self
.pings_not_received
+= 1
226 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
227 if self
.pings_not_received
> 10:
228 raise LcmException("It is not receiving pings from Kafka bus")
229 consecutive_errors
= 0
233 except Exception as e
:
234 # if not first_start is the first time after starting. So leave more time and wait
235 # to allow kafka starts
236 if consecutive_errors
== 8 if not first_start
else 30:
237 self
.logger
.error("Task kafka_read task exit error too many errors. Exception: {}".format(e
))
239 consecutive_errors
+= 1
240 self
.logger
.error("Task kafka_read retrying after Exception {}".format(e
))
241 wait_time
= 2 if not first_start
else 5
242 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
244 def kafka_read_callback(self
, topic
, command
, params
):
247 if topic
!= "admin" and command
!= "ping":
248 self
.logger
.debug("Task kafka_read receives {} {}: {}".format(topic
, command
, params
))
249 self
.consecutive_errors
= 0
250 self
.first_start
= False
252 if command
== "exit":
253 raise LcmExceptionExit
254 elif command
.startswith("#"):
256 elif command
== "echo":
261 elif command
== "test":
262 asyncio
.Task(self
.test(params
), loop
=self
.loop
)
266 if command
== "ping" and params
["to"] == "lcm" and params
["from"] == "lcm":
267 if params
.get("worker_id") != self
.worker_id
:
269 self
.pings_not_received
= 0
271 with
open(health_check_file
, "w") as f
:
273 except Exception as e
:
274 self
.logger
.error("Cannot write into '{}' for healthcheck: {}".format(health_check_file
, e
))
277 if command
== "instantiate":
278 # self.logger.debug("Deploying NS {}".format(nsr_id))
280 nslcmop_id
= nslcmop
["_id"]
281 nsr_id
= nslcmop
["nsInstanceId"]
282 task
= asyncio
.ensure_future(self
.ns
.instantiate(nsr_id
, nslcmop_id
))
283 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_instantiate", task
)
285 elif command
== "terminate":
286 # self.logger.debug("Deleting NS {}".format(nsr_id))
288 nslcmop_id
= nslcmop
["_id"]
289 nsr_id
= nslcmop
["nsInstanceId"]
290 self
.lcm_tasks
.cancel(topic
, nsr_id
)
291 task
= asyncio
.ensure_future(self
.ns
.terminate(nsr_id
, nslcmop_id
))
292 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_terminate", task
)
294 elif command
== "action":
295 # self.logger.debug("Update NS {}".format(nsr_id))
297 nslcmop_id
= nslcmop
["_id"]
298 nsr_id
= nslcmop
["nsInstanceId"]
299 task
= asyncio
.ensure_future(self
.ns
.action(nsr_id
, nslcmop_id
))
300 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_action", task
)
302 elif command
== "scale":
303 # self.logger.debug("Update NS {}".format(nsr_id))
305 nslcmop_id
= nslcmop
["_id"]
306 nsr_id
= nslcmop
["nsInstanceId"]
307 task
= asyncio
.ensure_future(self
.ns
.scale(nsr_id
, nslcmop_id
))
308 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_scale", task
)
310 elif command
== "show":
313 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
314 print("nsr:\n _id={}\n operational-status: {}\n config-status: {}"
315 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
316 "".format(nsr_id
, db_nsr
["operational-status"], db_nsr
["config-status"],
317 db_nsr
["detailed-status"],
318 db_nsr
["_admin"]["deployed"], self
.lcm_ns_tasks
.get(nsr_id
)))
319 except Exception as e
:
320 print("nsr {} not found: {}".format(nsr_id
, e
))
323 elif command
== "deleted":
324 return # TODO cleaning of task just in case should be done
325 elif command
in ("terminated", "instantiated", "scaled", "actioned"): # "scaled-cooldown-time"
327 elif topic
== "nsi": # netslice LCM processes (instantiate, terminate, etc)
328 if command
== "instantiate":
329 # self.logger.debug("Instantiating Network Slice {}".format(nsilcmop["netsliceInstanceId"]))
331 nsilcmop_id
= nsilcmop
["_id"] # slice operation id
332 nsir_id
= nsilcmop
["netsliceInstanceId"] # slice record id
333 task
= asyncio
.ensure_future(self
.netslice
.instantiate(nsir_id
, nsilcmop_id
))
334 self
.lcm_tasks
.register("nsi", nsir_id
, nsilcmop_id
, "nsi_instantiate", task
)
336 elif command
== "terminate":
337 # self.logger.debug("Terminating Network Slice NS {}".format(nsilcmop["netsliceInstanceId"]))
339 nsilcmop_id
= nsilcmop
["_id"] # slice operation id
340 nsir_id
= nsilcmop
["netsliceInstanceId"] # slice record id
341 self
.lcm_tasks
.cancel(topic
, nsir_id
)
342 task
= asyncio
.ensure_future(self
.netslice
.terminate(nsir_id
, nsilcmop_id
))
343 self
.lcm_tasks
.register("nsi", nsir_id
, nsilcmop_id
, "nsi_terminate", task
)
345 elif command
== "show":
348 db_nsir
= self
.db
.get_one("nsirs", {"_id": nsir_id
})
349 print("nsir:\n _id={}\n operational-status: {}\n config-status: {}"
350 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
351 "".format(nsir_id
, db_nsir
["operational-status"], db_nsir
["config-status"],
352 db_nsir
["detailed-status"],
353 db_nsir
["_admin"]["deployed"], self
.lcm_netslice_tasks
.get(nsir_id
)))
354 except Exception as e
:
355 print("nsir {} not found: {}".format(nsir_id
, e
))
358 elif command
== "deleted":
359 return # TODO cleaning of task just in case should be done
360 elif command
in ("terminated", "instantiated", "scaled", "actioned"): # "scaled-cooldown-time"
362 elif topic
== "vim_account":
363 vim_id
= params
["_id"]
364 if command
== "create":
365 task
= asyncio
.ensure_future(self
.vim
.create(params
, order_id
))
366 self
.lcm_tasks
.register("vim_account", vim_id
, order_id
, "vim_create", task
)
368 elif command
== "delete":
369 self
.lcm_tasks
.cancel(topic
, vim_id
)
370 task
= asyncio
.ensure_future(self
.vim
.delete(params
, order_id
))
371 self
.lcm_tasks
.register("vim_account", vim_id
, order_id
, "vim_delete", task
)
373 elif command
== "show":
374 print("not implemented show with vim_account")
377 elif command
== "edit":
378 task
= asyncio
.ensure_future(self
.vim
.edit(params
, order_id
))
379 self
.lcm_tasks
.register("vim_account", vim_id
, order_id
, "vim_edit", task
)
381 elif topic
== "wim_account":
382 wim_id
= params
["_id"]
383 if command
== "create":
384 task
= asyncio
.ensure_future(self
.wim
.create(params
, order_id
))
385 self
.lcm_tasks
.register("wim_account", wim_id
, order_id
, "wim_create", task
)
387 elif command
== "delete":
388 self
.lcm_tasks
.cancel(topic
, wim_id
)
389 task
= asyncio
.ensure_future(self
.wim
.delete(params
, order_id
))
390 self
.lcm_tasks
.register("wim_account", wim_id
, order_id
, "wim_delete", task
)
392 elif command
== "show":
393 print("not implemented show with wim_account")
396 elif command
== "edit":
397 task
= asyncio
.ensure_future(self
.wim
.edit(params
, order_id
))
398 self
.lcm_tasks
.register("wim_account", wim_id
, order_id
, "wim_edit", task
)
401 _sdn_id
= params
["_id"]
402 if command
== "create":
403 task
= asyncio
.ensure_future(self
.sdn
.create(params
, order_id
))
404 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_create", task
)
406 elif command
== "delete":
407 self
.lcm_tasks
.cancel(topic
, _sdn_id
)
408 task
= asyncio
.ensure_future(self
.sdn
.delete(params
, order_id
))
409 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_delete", task
)
411 elif command
== "edit":
412 task
= asyncio
.ensure_future(self
.sdn
.edit(params
, order_id
))
413 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_edit", task
)
415 self
.logger
.critical("unknown topic {} and command '{}'".format(topic
, command
))
417 async def kafka_read(self
):
418 self
.logger
.debug("Task kafka_read Enter with worker_id={}".format(self
.worker_id
))
419 # future = asyncio.Future()
420 self
.consecutive_errors
= 0
421 self
.first_start
= True
422 while self
.consecutive_errors
< 10:
424 topics
= ("ns", "vim_account", "wim_account", "sdn", "nsi")
425 topics_admin
= ("admin", )
426 await asyncio
.gather(
427 self
.msg
.aioread(topics
, self
.loop
, self
.kafka_read_callback
),
428 self
.msg_admin
.aioread(topics_admin
, self
.loop
, self
.kafka_read_callback
, group_id
=False)
431 except LcmExceptionExit
:
432 self
.logger
.debug("Bye!")
434 except Exception as e
:
435 # if not first_start is the first time after starting. So leave more time and wait
436 # to allow kafka starts
437 if self
.consecutive_errors
== 8 if not self
.first_start
else 30:
438 self
.logger
.error("Task kafka_read task exit error too many errors. Exception: {}".format(e
))
440 self
.consecutive_errors
+= 1
441 self
.logger
.error("Task kafka_read retrying after Exception {}".format(e
))
442 wait_time
= 2 if not self
.first_start
else 5
443 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
445 # self.logger.debug("Task kafka_read terminating")
446 self
.logger
.debug("Task kafka_read exit")
451 self
.loop
.run_until_complete(self
.check_RO_version())
453 self
.loop
.run_until_complete(asyncio
.gather(
458 # self.logger.debug("Terminating cancelling creation tasks")
459 # self.lcm_tasks.cancel("ALL", "create")
461 # while self.is_pending_tasks():
462 # self.logger.debug("Task kafka_read terminating. Waiting for tasks termination")
463 # await asyncio.sleep(2, loop=self.loop)
466 # self.lcm_tasks.cancel("ALL", "ALL")
470 self
.db
.db_disconnect()
472 self
.msg
.disconnect()
474 self
.msg_admin
.disconnect()
476 self
.fs
.fs_disconnect()
478 def read_config_file(self
, config_file
):
479 # TODO make a [ini] + yaml inside parser
480 # the configparser library is not suitable, because it does not admit comments at the end of line,
481 # and not parse integer or boolean
483 with
open(config_file
) as f
:
485 for k
, v
in environ
.items():
486 if not k
.startswith("OSMLCM_"):
488 k_items
= k
.lower().split("_")
491 if k_items
[1] in ("ro", "vca"):
492 # put in capital letter
493 k_items
[1] = k_items
[1].upper()
496 for k_item
in k_items
[1:-1]:
498 if k_items
[-1] == "port":
499 c
[k_items
[-1]] = int(v
)
502 except Exception as e
:
503 self
.logger
.warn("skipping environ '{}' on exception '{}'".format(k
, e
))
506 except Exception as e
:
507 self
.logger
.critical("At config file '{}': {}".format(config_file
, e
))
511 def get_process_id():
513 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
514 will provide a random one
517 # Try getting docker id. If fails, get pid
519 with
open("/proc/self/cgroup", "r") as f
:
520 text_id_
= f
.readline()
521 _
, _
, text_id
= text_id_
.rpartition("/")
522 text_id
= text_id
.replace('\n', '')[:12]
528 return ''.join(random_choice("0123456789abcdef") for _
in range(12))
532 print("""Usage: {} [options]
533 -c|--config [configuration_file]: loads the configuration file (default: ./nbi.cfg)
534 --health-check: do not run lcm, but inspect kafka bus to determine if lcm is healthy
535 -h|--help: shows this help
536 """.format(sys
.argv
[0]))
537 # --log-socket-host HOST: send logs to this host")
538 # --log-socket-port PORT: send logs using this port (default: 9022)")
546 with
open(health_check_file
, "r") as f
:
547 last_received_ping
= f
.read()
549 if time() - float(last_received_ping
) < Lcm
.ping_interval_pace
+ 10:
558 if __name__
== '__main__':
560 # load parameters and configuration
561 opts
, args
= getopt
.getopt(sys
.argv
[1:], "hc:", ["config=", "help", "health-check"])
562 # TODO add "log-socket-host=", "log-socket-port=", "log-file="
565 if o
in ("-h", "--help"):
568 elif o
in ("-c", "--config"):
570 elif o
== "--health-check":
572 # elif o == "--log-socket-port":
573 # log_socket_port = a
574 # elif o == "--log-socket-host":
575 # log_socket_host = a
576 # elif o == "--log-file":
579 assert False, "Unhandled option"
581 if not path
.isfile(config_file
):
582 print("configuration file '{}' not exist".format(config_file
), file=sys
.stderr
)
585 for config_file
in (__file__
[:__file__
.rfind(".")] + ".cfg", "./lcm.cfg", "/etc/osm/lcm.cfg"):
586 if path
.isfile(config_file
):
589 print("No configuration file 'lcm.cfg' found neither at local folder nor at /etc/osm/", file=sys
.stderr
)
591 lcm
= Lcm(config_file
)
593 except (LcmException
, getopt
.GetoptError
) as e
:
594 print(str(e
), file=sys
.stderr
)