7f949f3227ebcb972bb5d3b8e396fc9fb70563d1
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
28 import logging
.handlers
32 from osm_lcm
import ns
, vim_sdn
, netslice
33 from osm_lcm
.ng_ro
import NgRoException
, NgRoClient
34 from osm_lcm
.ROclient
import ROClient
, ROClientException
37 from osm_lcm
.lcm_utils
import versiontuple
, LcmException
, TaskRegistry
, LcmExceptionExit
38 from osm_lcm
import version
as lcm_version
, version_date
as lcm_version_date
40 from osm_common
import msglocal
, msgkafka
41 from osm_common
import version
as common_version
42 from osm_common
.dbbase
import DbException
43 from osm_common
.fsbase
import FsException
44 from osm_common
.msgbase
import MsgException
45 from osm_lcm
.data_utils
.database
.database
import Database
46 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
47 from osm_lcm
.lcm_hc
import get_health_check_file
48 from os
import environ
, path
49 from random
import choice
as random_choice
50 from n2vc
import version
as n2vc_version
53 if os
.getenv("OSMLCM_PDB_DEBUG", None) is not None:
57 __author__
= "Alfonso Tierno"
58 min_RO_version
= "6.0.2"
59 min_n2vc_version
= "0.0.2"
61 min_common_version
= "0.1.19"
66 ping_interval_pace
= (
67 120 # how many time ping is send once is confirmed all is running
69 ping_interval_boot
= 5 # how many time ping is sent when booting
74 "tsdb": "lcm.prometheus",
76 # ^ contains for each section at lcm.cfg the used logger name
78 def __init__(self
, config_file
, loop
=None):
80 Init, Connect to database, filesystem storage, and messaging
81 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
88 self
.pings_not_received
= 1
89 self
.consecutive_errors
= 0
90 self
.first_start
= False
93 self
.logger
= logging
.getLogger("lcm")
95 self
.worker_id
= self
.get_process_id()
97 config
= self
.read_config_file(config_file
)
99 self
.health_check_file
= get_health_check_file(self
.config
)
100 self
.config
["ro_config"] = {
101 "ng": config
["RO"].get("ng", False),
102 "uri": config
["RO"].get("uri"),
103 "tenant": config
.get("tenant", "osm"),
104 "logger_name": "lcm.roclient",
105 "loglevel": config
["RO"].get("loglevel", "ERROR"),
107 if not self
.config
["ro_config"]["uri"]:
108 self
.config
["ro_config"]["uri"] = "http://{}:{}/".format(
109 config
["RO"]["host"], config
["RO"]["port"]
112 "/ro" in self
.config
["ro_config"]["uri"][-4:]
113 or "/openmano" in self
.config
["ro_config"]["uri"][-10:]
115 # uri ends with '/ro', '/ro/', '/openmano', '/openmano/'
116 index
= self
.config
["ro_config"]["uri"][-1].rfind("/")
117 self
.config
["ro_config"]["uri"] = self
.config
["ro_config"]["uri"][index
+ 1]
119 self
.loop
= loop
or asyncio
.get_event_loop()
124 ) = self
.wim
= self
.sdn
= self
.k8scluster
= self
.vca
= self
.k8srepo
= None
127 log_format_simple
= (
128 "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
130 log_formatter_simple
= logging
.Formatter(
131 log_format_simple
, datefmt
="%Y-%m-%dT%H:%M:%S"
133 config
["database"]["logger_name"] = "lcm.db"
134 config
["storage"]["logger_name"] = "lcm.fs"
135 config
["message"]["logger_name"] = "lcm.msg"
136 if config
["global"].get("logfile"):
137 file_handler
= logging
.handlers
.RotatingFileHandler(
138 config
["global"]["logfile"], maxBytes
=100e6
, backupCount
=9, delay
=0
140 file_handler
.setFormatter(log_formatter_simple
)
141 self
.logger
.addHandler(file_handler
)
142 if not config
["global"].get("nologging"):
143 str_handler
= logging
.StreamHandler()
144 str_handler
.setFormatter(log_formatter_simple
)
145 self
.logger
.addHandler(str_handler
)
147 if config
["global"].get("loglevel"):
148 self
.logger
.setLevel(config
["global"]["loglevel"])
150 # logging other modules
151 for k1
, logname
in self
.cfg_logger_name
.items():
152 config
[k1
]["logger_name"] = logname
153 logger_module
= logging
.getLogger(logname
)
154 if config
[k1
].get("logfile"):
155 file_handler
= logging
.handlers
.RotatingFileHandler(
156 config
[k1
]["logfile"], maxBytes
=100e6
, backupCount
=9, delay
=0
158 file_handler
.setFormatter(log_formatter_simple
)
159 logger_module
.addHandler(file_handler
)
160 if config
[k1
].get("loglevel"):
161 logger_module
.setLevel(config
[k1
]["loglevel"])
162 self
.logger
.critical(
163 "starting osm/lcm version {} {}".format(lcm_version
, lcm_version_date
)
166 # check version of N2VC
167 # TODO enhance with int conversion or from distutils.version import LooseVersion
168 # or with list(map(int, version.split(".")))
169 if versiontuple(n2vc_version
) < versiontuple(min_n2vc_version
):
171 "Not compatible osm/N2VC version '{}'. Needed '{}' or higher".format(
172 n2vc_version
, min_n2vc_version
175 # check version of common
176 if versiontuple(common_version
) < versiontuple(min_common_version
):
178 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
179 common_version
, min_common_version
184 self
.db
= Database(config
).instance
.db
186 self
.fs
= Filesystem(config
).instance
.fs
189 # copy message configuration in order to remove 'group_id' for msg_admin
190 config_message
= config
["message"].copy()
191 config_message
["loop"] = self
.loop
192 if config_message
["driver"] == "local":
193 self
.msg
= msglocal
.MsgLocal()
194 self
.msg
.connect(config_message
)
195 self
.msg_admin
= msglocal
.MsgLocal()
196 config_message
.pop("group_id", None)
197 self
.msg_admin
.connect(config_message
)
198 elif config_message
["driver"] == "kafka":
199 self
.msg
= msgkafka
.MsgKafka()
200 self
.msg
.connect(config_message
)
201 self
.msg_admin
= msgkafka
.MsgKafka()
202 config_message
.pop("group_id", None)
203 self
.msg_admin
.connect(config_message
)
206 "Invalid configuration param '{}' at '[message]':'driver'".format(
207 config
["message"]["driver"]
210 except (DbException
, FsException
, MsgException
) as e
:
211 self
.logger
.critical(str(e
), exc_info
=True)
212 raise LcmException(str(e
))
214 # contains created tasks/futures to be able to cancel
215 self
.lcm_tasks
= TaskRegistry(self
.worker_id
, self
.logger
)
217 async def check_RO_version(self
):
221 ro_uri
= self
.config
["ro_config"]["uri"]
223 # try new RO, if fail old RO
225 self
.config
["ro_config"]["uri"] = ro_uri
+ "ro"
226 ro_server
= NgRoClient(self
.loop
, **self
.config
["ro_config"])
227 ro_version
= await ro_server
.get_version()
228 self
.config
["ro_config"]["ng"] = True
230 self
.config
["ro_config"]["uri"] = ro_uri
+ "openmano"
231 ro_server
= ROClient(self
.loop
, **self
.config
["ro_config"])
232 ro_version
= await ro_server
.get_version()
233 self
.config
["ro_config"]["ng"] = False
234 if versiontuple(ro_version
) < versiontuple(min_RO_version
):
236 "Not compatible osm/RO version '{}'. Needed '{}' or higher".format(
237 ro_version
, min_RO_version
241 "Connected to RO version {} new-generation version {}".format(
242 ro_version
, self
.config
["ro_config"]["ng"]
246 except (ROClientException
, NgRoException
) as e
:
247 self
.config
["ro_config"]["uri"] = ro_uri
249 traceback
.print_tb(e
.__traceback
__)
250 error_text
= "Error while connecting to RO on {}: {}".format(
251 self
.config
["ro_config"]["uri"], e
254 self
.logger
.critical(error_text
)
255 raise LcmException(error_text
)
256 if last_error
!= error_text
:
257 last_error
= error_text
259 error_text
+ ". Waiting until {} seconds".format(5 * tries
)
261 await asyncio
.sleep(5)
263 async def test(self
, param
=None):
264 self
.logger
.debug("Starting/Ending test task: {}".format(param
))
266 async def kafka_ping(self
):
267 self
.logger
.debug("Task kafka_ping Enter")
268 consecutive_errors
= 0
270 kafka_has_received
= False
271 self
.pings_not_received
= 1
274 await self
.msg_admin
.aiowrite(
280 "worker_id": self
.worker_id
,
281 "version": lcm_version
,
285 # time between pings are low when it is not received and at starting
287 self
.ping_interval_boot
288 if not kafka_has_received
289 else self
.ping_interval_pace
291 if not self
.pings_not_received
:
292 kafka_has_received
= True
293 self
.pings_not_received
+= 1
294 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
295 if self
.pings_not_received
> 10:
296 raise LcmException("It is not receiving pings from Kafka bus")
297 consecutive_errors
= 0
301 except Exception as e
:
302 # if not first_start is the first time after starting. So leave more time and wait
303 # to allow kafka starts
304 if consecutive_errors
== 8 if not first_start
else 30:
306 "Task kafka_read task exit error too many errors. Exception: {}".format(
311 consecutive_errors
+= 1
313 "Task kafka_read retrying after Exception {}".format(e
)
315 wait_time
= 2 if not first_start
else 5
316 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
318 def kafka_read_callback(self
, topic
, command
, params
):
321 if topic
!= "admin" and command
!= "ping":
323 "Task kafka_read receives {} {}: {}".format(topic
, command
, params
)
325 self
.consecutive_errors
= 0
326 self
.first_start
= False
328 if command
== "exit":
329 raise LcmExceptionExit
330 elif command
.startswith("#"):
332 elif command
== "echo":
337 elif command
== "test":
338 asyncio
.Task(self
.test(params
), loop
=self
.loop
)
342 if command
== "ping" and params
["to"] == "lcm" and params
["from"] == "lcm":
343 if params
.get("worker_id") != self
.worker_id
:
345 self
.pings_not_received
= 0
347 with
open(self
.health_check_file
, "w") as f
:
349 except Exception as e
:
351 "Cannot write into '{}' for healthcheck: {}".format(
352 self
.health_check_file
, e
357 if command
== "placement":
358 self
.ns
.update_nsrs_with_pla_result(params
)
360 elif topic
== "k8scluster":
361 if command
== "create" or command
== "created":
362 k8scluster_id
= params
.get("_id")
363 task
= asyncio
.ensure_future(self
.k8scluster
.create(params
, order_id
))
364 self
.lcm_tasks
.register(
365 "k8scluster", k8scluster_id
, order_id
, "k8scluster_create", task
368 elif command
== "edit" or command
== "edited":
369 k8scluster_id
= params
.get("_id")
370 task
= asyncio
.ensure_future(self
.k8scluster
.edit(params
, order_id
))
371 self
.lcm_tasks
.register(
372 "k8scluster", k8scluster_id
, order_id
, "k8scluster_edit", task
375 elif command
== "delete" or command
== "deleted":
376 k8scluster_id
= params
.get("_id")
377 task
= asyncio
.ensure_future(self
.k8scluster
.delete(params
, order_id
))
378 self
.lcm_tasks
.register(
379 "k8scluster", k8scluster_id
, order_id
, "k8scluster_delete", task
383 if command
== "create" or command
== "created":
384 vca_id
= params
.get("_id")
385 task
= asyncio
.ensure_future(self
.vca
.create(params
, order_id
))
386 self
.lcm_tasks
.register("vca", vca_id
, order_id
, "vca_create", task
)
388 elif command
== "delete" or command
== "deleted":
389 vca_id
= params
.get("_id")
390 task
= asyncio
.ensure_future(self
.vca
.delete(params
, order_id
))
391 self
.lcm_tasks
.register("vca", vca_id
, order_id
, "vca_delete", task
)
393 elif topic
== "k8srepo":
394 if command
== "create" or command
== "created":
395 k8srepo_id
= params
.get("_id")
396 self
.logger
.debug("k8srepo_id = {}".format(k8srepo_id
))
397 task
= asyncio
.ensure_future(self
.k8srepo
.create(params
, order_id
))
398 self
.lcm_tasks
.register(
399 "k8srepo", k8srepo_id
, order_id
, "k8srepo_create", task
402 elif command
== "delete" or command
== "deleted":
403 k8srepo_id
= params
.get("_id")
404 task
= asyncio
.ensure_future(self
.k8srepo
.delete(params
, order_id
))
405 self
.lcm_tasks
.register(
406 "k8srepo", k8srepo_id
, order_id
, "k8srepo_delete", task
410 if command
== "instantiate":
411 # self.logger.debug("Deploying NS {}".format(nsr_id))
413 nslcmop_id
= nslcmop
["_id"]
414 nsr_id
= nslcmop
["nsInstanceId"]
415 task
= asyncio
.ensure_future(self
.ns
.instantiate(nsr_id
, nslcmop_id
))
416 self
.lcm_tasks
.register(
417 "ns", nsr_id
, nslcmop_id
, "ns_instantiate", task
420 elif command
== "terminate":
421 # self.logger.debug("Deleting NS {}".format(nsr_id))
423 nslcmop_id
= nslcmop
["_id"]
424 nsr_id
= nslcmop
["nsInstanceId"]
425 self
.lcm_tasks
.cancel(topic
, nsr_id
)
426 task
= asyncio
.ensure_future(self
.ns
.terminate(nsr_id
, nslcmop_id
))
427 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_terminate", task
)
429 elif command
== "vca_status_refresh":
431 nslcmop_id
= nslcmop
["_id"]
432 nsr_id
= nslcmop
["nsInstanceId"]
433 task
= asyncio
.ensure_future(
434 self
.ns
.vca_status_refresh(nsr_id
, nslcmop_id
)
436 self
.lcm_tasks
.register(
437 "ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh", task
440 elif command
== "action":
441 # self.logger.debug("Update NS {}".format(nsr_id))
443 nslcmop_id
= nslcmop
["_id"]
444 nsr_id
= nslcmop
["nsInstanceId"]
445 task
= asyncio
.ensure_future(self
.ns
.action(nsr_id
, nslcmop_id
))
446 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_action", task
)
448 elif command
== "update":
449 # self.logger.debug("Update NS {}".format(nsr_id))
451 nslcmop_id
= nslcmop
["_id"]
452 nsr_id
= nslcmop
["nsInstanceId"]
453 task
= asyncio
.ensure_future(self
.ns
.update(nsr_id
, nslcmop_id
))
454 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_update", task
)
456 elif command
== "scale":
457 # self.logger.debug("Update NS {}".format(nsr_id))
459 nslcmop_id
= nslcmop
["_id"]
460 nsr_id
= nslcmop
["nsInstanceId"]
461 task
= asyncio
.ensure_future(self
.ns
.scale(nsr_id
, nslcmop_id
))
462 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_scale", task
)
464 elif command
== "heal":
465 # self.logger.debug("Healing NS {}".format(nsr_id))
467 nslcmop_id
= nslcmop
["_id"]
468 nsr_id
= nslcmop
["nsInstanceId"]
469 task
= asyncio
.ensure_future(self
.ns
.heal(nsr_id
, nslcmop_id
))
470 self
.lcm_tasks
.register(
471 "ns", nsr_id
, nslcmop_id
, "ns_heal", task
474 elif command
== "migrate":
476 nslcmop_id
= nslcmop
["_id"]
477 nsr_id
= nslcmop
["nsInstanceId"]
478 task
= asyncio
.ensure_future(self
.ns
.migrate(nsr_id
, nslcmop_id
))
479 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_migrate", task
)
481 elif command
== "verticalscale":
483 nslcmop_id
= nslcmop
["_id"]
484 nsr_id
= nslcmop
["nsInstanceId"]
485 task
= asyncio
.ensure_future(self
.ns
.vertical_scale(nsr_id
, nslcmop_id
))
486 self
.logger
.debug("nsr_id,nslcmop_id,task {},{},{}".format(nsr_id
, nslcmop_id
, task
))
487 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_verticalscale", task
)
488 self
.logger
.debug("LCM task registered {},{},{} ".format(nsr_id
, nslcmop_id
, task
))
490 elif command
== "show":
493 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
495 "nsr:\n _id={}\n operational-status: {}\n config-status: {}"
496 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
499 db_nsr
["operational-status"],
500 db_nsr
["config-status"],
501 db_nsr
["detailed-status"],
502 db_nsr
["_admin"]["deployed"],
503 self
.lcm_ns_tasks
.get(nsr_id
),
506 except Exception as e
:
507 print("nsr {} not found: {}".format(nsr_id
, e
))
510 elif command
== "deleted":
511 return # TODO cleaning of task just in case should be done
523 ): # "scaled-cooldown-time"
526 elif topic
== "nsi": # netslice LCM processes (instantiate, terminate, etc)
527 if command
== "instantiate":
528 # self.logger.debug("Instantiating Network Slice {}".format(nsilcmop["netsliceInstanceId"]))
530 nsilcmop_id
= nsilcmop
["_id"] # slice operation id
531 nsir_id
= nsilcmop
["netsliceInstanceId"] # slice record id
532 task
= asyncio
.ensure_future(
533 self
.netslice
.instantiate(nsir_id
, nsilcmop_id
)
535 self
.lcm_tasks
.register(
536 "nsi", nsir_id
, nsilcmop_id
, "nsi_instantiate", task
539 elif command
== "terminate":
540 # self.logger.debug("Terminating Network Slice NS {}".format(nsilcmop["netsliceInstanceId"]))
542 nsilcmop_id
= nsilcmop
["_id"] # slice operation id
543 nsir_id
= nsilcmop
["netsliceInstanceId"] # slice record id
544 self
.lcm_tasks
.cancel(topic
, nsir_id
)
545 task
= asyncio
.ensure_future(
546 self
.netslice
.terminate(nsir_id
, nsilcmop_id
)
548 self
.lcm_tasks
.register(
549 "nsi", nsir_id
, nsilcmop_id
, "nsi_terminate", task
552 elif command
== "show":
555 db_nsir
= self
.db
.get_one("nsirs", {"_id": nsir_id
})
557 "nsir:\n _id={}\n operational-status: {}\n config-status: {}"
558 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
561 db_nsir
["operational-status"],
562 db_nsir
["config-status"],
563 db_nsir
["detailed-status"],
564 db_nsir
["_admin"]["deployed"],
565 self
.lcm_netslice_tasks
.get(nsir_id
),
568 except Exception as e
:
569 print("nsir {} not found: {}".format(nsir_id
, e
))
572 elif command
== "deleted":
573 return # TODO cleaning of task just in case should be done
580 ): # "scaled-cooldown-time"
582 elif topic
== "vim_account":
583 vim_id
= params
["_id"]
584 if command
in ("create", "created"):
585 if not self
.config
["ro_config"].get("ng"):
586 task
= asyncio
.ensure_future(self
.vim
.create(params
, order_id
))
587 self
.lcm_tasks
.register(
588 "vim_account", vim_id
, order_id
, "vim_create", task
591 elif command
== "delete" or command
== "deleted":
592 self
.lcm_tasks
.cancel(topic
, vim_id
)
593 task
= asyncio
.ensure_future(self
.vim
.delete(params
, order_id
))
594 self
.lcm_tasks
.register(
595 "vim_account", vim_id
, order_id
, "vim_delete", task
598 elif command
== "show":
599 print("not implemented show with vim_account")
602 elif command
in ("edit", "edited"):
603 if not self
.config
["ro_config"].get("ng"):
604 task
= asyncio
.ensure_future(self
.vim
.edit(params
, order_id
))
605 self
.lcm_tasks
.register(
606 "vim_account", vim_id
, order_id
, "vim_edit", task
609 elif command
== "deleted":
610 return # TODO cleaning of task just in case should be done
611 elif topic
== "wim_account":
612 wim_id
= params
["_id"]
613 if command
in ("create", "created"):
614 if not self
.config
["ro_config"].get("ng"):
615 task
= asyncio
.ensure_future(self
.wim
.create(params
, order_id
))
616 self
.lcm_tasks
.register(
617 "wim_account", wim_id
, order_id
, "wim_create", task
620 elif command
== "delete" or command
== "deleted":
621 self
.lcm_tasks
.cancel(topic
, wim_id
)
622 task
= asyncio
.ensure_future(self
.wim
.delete(params
, order_id
))
623 self
.lcm_tasks
.register(
624 "wim_account", wim_id
, order_id
, "wim_delete", task
627 elif command
== "show":
628 print("not implemented show with wim_account")
631 elif command
in ("edit", "edited"):
632 task
= asyncio
.ensure_future(self
.wim
.edit(params
, order_id
))
633 self
.lcm_tasks
.register(
634 "wim_account", wim_id
, order_id
, "wim_edit", task
637 elif command
== "deleted":
638 return # TODO cleaning of task just in case should be done
640 _sdn_id
= params
["_id"]
641 if command
in ("create", "created"):
642 if not self
.config
["ro_config"].get("ng"):
643 task
= asyncio
.ensure_future(self
.sdn
.create(params
, order_id
))
644 self
.lcm_tasks
.register(
645 "sdn", _sdn_id
, order_id
, "sdn_create", task
648 elif command
== "delete" or command
== "deleted":
649 self
.lcm_tasks
.cancel(topic
, _sdn_id
)
650 task
= asyncio
.ensure_future(self
.sdn
.delete(params
, order_id
))
651 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_delete", task
)
653 elif command
in ("edit", "edited"):
654 task
= asyncio
.ensure_future(self
.sdn
.edit(params
, order_id
))
655 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_edit", task
)
657 elif command
== "deleted":
658 return # TODO cleaning of task just in case should be done
659 self
.logger
.critical("unknown topic {} and command '{}'".format(topic
, command
))
661 async def kafka_read(self
):
663 "Task kafka_read Enter with worker_id={}".format(self
.worker_id
)
665 # future = asyncio.Future()
666 self
.consecutive_errors
= 0
667 self
.first_start
= True
668 while self
.consecutive_errors
< 10:
681 topics_admin
= ("admin",)
682 await asyncio
.gather(
684 topics
, self
.loop
, self
.kafka_read_callback
, from_beginning
=True
686 self
.msg_admin
.aioread(
689 self
.kafka_read_callback
,
694 except LcmExceptionExit
:
695 self
.logger
.debug("Bye!")
697 except Exception as e
:
698 # if not first_start is the first time after starting. So leave more time and wait
699 # to allow kafka starts
700 if self
.consecutive_errors
== 8 if not self
.first_start
else 30:
702 "Task kafka_read task exit error too many errors. Exception: {}".format(
707 self
.consecutive_errors
+= 1
709 "Task kafka_read retrying after Exception {}".format(e
)
711 wait_time
= 2 if not self
.first_start
else 5
712 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
714 # self.logger.debug("Task kafka_read terminating")
715 self
.logger
.debug("Task kafka_read exit")
720 self
.loop
.run_until_complete(self
.check_RO_version())
722 self
.ns
= ns
.NsLcm(self
.msg
, self
.lcm_tasks
, self
.config
, self
.loop
)
723 self
.netslice
= netslice
.NetsliceLcm(
724 self
.msg
, self
.lcm_tasks
, self
.config
, self
.loop
, self
.ns
726 self
.vim
= vim_sdn
.VimLcm(self
.msg
, self
.lcm_tasks
, self
.config
, self
.loop
)
727 self
.wim
= vim_sdn
.WimLcm(self
.msg
, self
.lcm_tasks
, self
.config
, self
.loop
)
728 self
.sdn
= vim_sdn
.SdnLcm(self
.msg
, self
.lcm_tasks
, self
.config
, self
.loop
)
729 self
.k8scluster
= vim_sdn
.K8sClusterLcm(
730 self
.msg
, self
.lcm_tasks
, self
.config
, self
.loop
732 self
.vca
= vim_sdn
.VcaLcm(self
.msg
, self
.lcm_tasks
, self
.config
, self
.loop
)
733 self
.k8srepo
= vim_sdn
.K8sRepoLcm(
734 self
.msg
, self
.lcm_tasks
, self
.config
, self
.loop
737 self
.loop
.run_until_complete(
738 asyncio
.gather(self
.kafka_read(), self
.kafka_ping())
742 # self.logger.debug("Terminating cancelling creation tasks")
743 # self.lcm_tasks.cancel("ALL", "create")
745 # while self.is_pending_tasks():
746 # self.logger.debug("Task kafka_read terminating. Waiting for tasks termination")
747 # await asyncio.sleep(2, loop=self.loop)
750 # self.lcm_tasks.cancel("ALL", "ALL")
754 self
.db
.db_disconnect()
756 self
.msg
.disconnect()
758 self
.msg_admin
.disconnect()
760 self
.fs
.fs_disconnect()
762 def read_config_file(self
, config_file
):
763 # TODO make a [ini] + yaml inside parser
764 # the configparser library is not suitable, because it does not admit comments at the end of line,
765 # and not parse integer or boolean
767 # read file as yaml format
768 with
open(config_file
) as f
:
769 conf
= yaml
.safe_load(f
)
770 # Ensure all sections are not empty
783 # read all environ that starts with OSMLCM_
784 for k
, v
in environ
.items():
785 if not k
.startswith("OSMLCM_"):
787 subject
, _
, item
= k
[7:].lower().partition("_")
790 if subject
in ("ro", "vca"):
791 # put in capital letter
792 subject
= subject
.upper()
794 if item
== "port" or subject
== "timeout":
795 conf
[subject
][item
] = int(v
)
797 conf
[subject
][item
] = v
798 except Exception as e
:
800 "skipping environ '{}' on exception '{}'".format(k
, e
)
803 # backward compatibility of VCA parameters
805 if "pubkey" in conf
["VCA"]:
806 conf
["VCA"]["public_key"] = conf
["VCA"].pop("pubkey")
807 if "cacert" in conf
["VCA"]:
808 conf
["VCA"]["ca_cert"] = conf
["VCA"].pop("cacert")
809 if "apiproxy" in conf
["VCA"]:
810 conf
["VCA"]["api_proxy"] = conf
["VCA"].pop("apiproxy")
812 if "enableosupgrade" in conf
["VCA"]:
813 conf
["VCA"]["enable_os_upgrade"] = conf
["VCA"].pop("enableosupgrade")
814 if isinstance(conf
["VCA"].get("enable_os_upgrade"), str):
815 if conf
["VCA"]["enable_os_upgrade"].lower() == "false":
816 conf
["VCA"]["enable_os_upgrade"] = False
817 elif conf
["VCA"]["enable_os_upgrade"].lower() == "true":
818 conf
["VCA"]["enable_os_upgrade"] = True
820 if "aptmirror" in conf
["VCA"]:
821 conf
["VCA"]["apt_mirror"] = conf
["VCA"].pop("aptmirror")
824 except Exception as e
:
825 self
.logger
.critical("At config file '{}': {}".format(config_file
, e
))
829 def get_process_id():
831 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
832 will provide a random one
835 # Try getting docker id. If fails, get pid
837 with
open("/proc/self/cgroup", "r") as f
:
838 text_id_
= f
.readline()
839 _
, _
, text_id
= text_id_
.rpartition("/")
840 text_id
= text_id
.replace("\n", "")[:12]
846 return "".join(random_choice("0123456789abcdef") for _
in range(12))
851 """Usage: {} [options]
852 -c|--config [configuration_file]: loads the configuration file (default: ./lcm.cfg)
853 --health-check: do not run lcm, but inspect kafka bus to determine if lcm is healthy
854 -h|--help: shows this help
859 # --log-socket-host HOST: send logs to this host")
860 # --log-socket-port PORT: send logs using this port (default: 9022)")
863 if __name__
== "__main__":
866 # print("SYS.PATH='{}'".format(sys.path))
867 # load parameters and configuration
873 opts
, args
= getopt
.getopt(
874 sys
.argv
[1:], "hc:", ["config=", "help", "health-check"]
876 # TODO add "log-socket-host=", "log-socket-port=", "log-file="
879 if o
in ("-h", "--help"):
882 elif o
in ("-c", "--config"):
884 elif o
== "--health-check":
885 from osm_lcm
.lcm_hc
import health_check
887 health_check(config_file
, Lcm
.ping_interval_pace
)
888 # elif o == "--log-socket-port":
889 # log_socket_port = a
890 # elif o == "--log-socket-host":
891 # log_socket_host = a
892 # elif o == "--log-file":
895 assert False, "Unhandled option"
898 if not path
.isfile(config_file
):
900 "configuration file '{}' does not exist".format(config_file
),
906 __file__
[: __file__
.rfind(".")] + ".cfg",
910 if path
.isfile(config_file
):
914 "No configuration file 'lcm.cfg' found neither at local folder nor at /etc/osm/",
918 lcm
= Lcm(config_file
)
920 except (LcmException
, getopt
.GetoptError
) as e
:
921 print(str(e
), file=sys
.stderr
)