723ca7ac2610b84c83b73b5fcad82b51ab2ab57b
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
.data_utils
.lcm_config
import LcmCfg
48 from osm_lcm
.lcm_hc
import get_health_check_file
50 from random
import choice
as random_choice
51 from n2vc
import version
as n2vc_version
54 if os
.getenv("OSMLCM_PDB_DEBUG", None) is not None:
58 __author__
= "Alfonso Tierno"
59 min_RO_version
= "6.0.2"
60 min_n2vc_version
= "0.0.2"
62 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
71 main_config
= LcmCfg()
73 def __init__(self
, config_file
, loop
=None):
75 Init, Connect to database, filesystem storage, and messaging
76 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
83 self
.pings_not_received
= 1
84 self
.consecutive_errors
= 0
85 self
.first_start
= False
88 self
.logger
= logging
.getLogger("lcm")
90 self
.worker_id
= self
.get_process_id()
92 config
= self
.read_config_file(config_file
)
93 self
.main_config
.set_from_dict(config
)
94 self
.main_config
.transform()
95 self
.main_config
.load_from_env()
96 self
.logger
.critical("Loaded configuration:" + str(self
.main_config
.to_dict()))
97 # TODO: check if lcm_hc.py is necessary
98 self
.health_check_file
= get_health_check_file(self
.main_config
.to_dict())
99 self
.loop
= loop
or asyncio
.get_event_loop()
104 ) = self
.wim
= self
.sdn
= self
.k8scluster
= self
.vca
= self
.k8srepo
= None
107 log_format_simple
= (
108 "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
110 log_formatter_simple
= logging
.Formatter(
111 log_format_simple
, datefmt
="%Y-%m-%dT%H:%M:%S"
113 if self
.main_config
.globalConfig
.logfile
:
114 file_handler
= logging
.handlers
.RotatingFileHandler(
115 self
.main_config
.globalConfig
.logfile
,
120 file_handler
.setFormatter(log_formatter_simple
)
121 self
.logger
.addHandler(file_handler
)
122 if not self
.main_config
.globalConfig
.to_dict()["nologging"]:
123 str_handler
= logging
.StreamHandler()
124 str_handler
.setFormatter(log_formatter_simple
)
125 self
.logger
.addHandler(str_handler
)
127 if self
.main_config
.globalConfig
.to_dict()["loglevel"]:
128 self
.logger
.setLevel(self
.main_config
.globalConfig
.loglevel
)
130 # logging other modules
131 for logger
in ("message", "database", "storage", "tsdb"):
132 logger_config
= self
.main_config
.to_dict()[logger
]
133 logger_module
= logging
.getLogger(logger_config
["logger_name"])
134 if logger_config
["logfile"]:
135 file_handler
= logging
.handlers
.RotatingFileHandler(
136 logger_config
["logfile"], maxBytes
=100e6
, backupCount
=9, delay
=0
138 file_handler
.setFormatter(log_formatter_simple
)
139 logger_module
.addHandler(file_handler
)
140 if logger_config
["loglevel"]:
141 logger_module
.setLevel(logger_config
["loglevel"])
142 self
.logger
.critical(
143 "starting osm/lcm version {} {}".format(lcm_version
, lcm_version_date
)
146 # check version of N2VC
147 # TODO enhance with int conversion or from distutils.version import LooseVersion
148 # or with list(map(int, version.split(".")))
149 if versiontuple(n2vc_version
) < versiontuple(min_n2vc_version
):
151 "Not compatible osm/N2VC version '{}'. Needed '{}' or higher".format(
152 n2vc_version
, min_n2vc_version
155 # check version of common
156 if versiontuple(common_version
) < versiontuple(min_common_version
):
158 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
159 common_version
, min_common_version
164 self
.db
= Database(self
.main_config
.to_dict()).instance
.db
166 self
.fs
= Filesystem(self
.main_config
.to_dict()).instance
.fs
169 # copy message configuration in order to remove 'group_id' for msg_admin
170 config_message
= self
.main_config
.message
.to_dict()
171 config_message
["loop"] = self
.loop
172 if config_message
["driver"] == "local":
173 self
.msg
= msglocal
.MsgLocal()
174 self
.msg
.connect(config_message
)
175 self
.msg_admin
= msglocal
.MsgLocal()
176 config_message
.pop("group_id", None)
177 self
.msg_admin
.connect(config_message
)
178 elif config_message
["driver"] == "kafka":
179 self
.msg
= msgkafka
.MsgKafka()
180 self
.msg
.connect(config_message
)
181 self
.msg_admin
= msgkafka
.MsgKafka()
182 config_message
.pop("group_id", None)
183 self
.msg_admin
.connect(config_message
)
186 "Invalid configuration param '{}' at '[message]':'driver'".format(
187 self
.main_config
.message
.driver
190 except (DbException
, FsException
, MsgException
) as e
:
191 self
.logger
.critical(str(e
), exc_info
=True)
192 raise LcmException(str(e
))
194 # contains created tasks/futures to be able to cancel
195 self
.lcm_tasks
= TaskRegistry(self
.worker_id
, self
.logger
)
197 async def check_RO_version(self
):
201 ro_uri
= self
.main_config
.RO
.uri
205 # try new RO, if fail old RO
207 self
.main_config
.RO
.uri
= ro_uri
+ "ro"
208 ro_server
= NgRoClient(self
.loop
, **self
.main_config
.RO
.to_dict())
209 ro_version
= await ro_server
.get_version()
210 self
.main_config
.RO
.ng
= True
212 self
.main_config
.RO
.uri
= ro_uri
+ "openmano"
213 ro_server
= ROClient(self
.loop
, **self
.main_config
.RO
.to_dict())
214 ro_version
= await ro_server
.get_version()
215 self
.main_config
.RO
.ng
= False
216 if versiontuple(ro_version
) < versiontuple(min_RO_version
):
218 "Not compatible osm/RO version '{}'. Needed '{}' or higher".format(
219 ro_version
, min_RO_version
223 "Connected to RO version {} new-generation version {}".format(
224 ro_version
, self
.main_config
.RO
.ng
228 except (ROClientException
, NgRoException
) as e
:
229 self
.main_config
.RO
.uri
= ro_uri
231 traceback
.print_tb(e
.__traceback
__)
232 error_text
= "Error while connecting to RO on {}: {}".format(
233 self
.main_config
.RO
.uri
, e
236 self
.logger
.critical(error_text
)
237 raise LcmException(error_text
)
238 if last_error
!= error_text
:
239 last_error
= error_text
241 error_text
+ ". Waiting until {} seconds".format(5 * tries
)
243 await asyncio
.sleep(5)
245 async def test(self
, param
=None):
246 self
.logger
.debug("Starting/Ending test task: {}".format(param
))
248 async def kafka_ping(self
):
249 self
.logger
.debug("Task kafka_ping Enter")
250 consecutive_errors
= 0
252 kafka_has_received
= False
253 self
.pings_not_received
= 1
256 await self
.msg_admin
.aiowrite(
262 "worker_id": self
.worker_id
,
263 "version": lcm_version
,
267 # time between pings are low when it is not received and at starting
269 self
.ping_interval_boot
270 if not kafka_has_received
271 else self
.ping_interval_pace
273 if not self
.pings_not_received
:
274 kafka_has_received
= True
275 self
.pings_not_received
+= 1
276 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
277 if self
.pings_not_received
> 10:
278 raise LcmException("It is not receiving pings from Kafka bus")
279 consecutive_errors
= 0
283 except Exception as e
:
284 # if not first_start is the first time after starting. So leave more time and wait
285 # to allow kafka starts
286 if consecutive_errors
== 8 if not first_start
else 30:
288 "Task kafka_read task exit error too many errors. Exception: {}".format(
293 consecutive_errors
+= 1
295 "Task kafka_read retrying after Exception {}".format(e
)
297 wait_time
= 2 if not first_start
else 5
298 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
300 def kafka_read_callback(self
, topic
, command
, params
):
303 if topic
!= "admin" and command
!= "ping":
305 "Task kafka_read receives {} {}: {}".format(topic
, command
, params
)
307 self
.consecutive_errors
= 0
308 self
.first_start
= False
310 if command
== "exit":
311 raise LcmExceptionExit
312 elif command
.startswith("#"):
314 elif command
== "echo":
319 elif command
== "test":
320 asyncio
.Task(self
.test(params
), loop
=self
.loop
)
324 if command
== "ping" and params
["to"] == "lcm" and params
["from"] == "lcm":
325 if params
.get("worker_id") != self
.worker_id
:
327 self
.pings_not_received
= 0
329 with
open(self
.health_check_file
, "w") as f
:
331 except Exception as e
:
333 "Cannot write into '{}' for healthcheck: {}".format(
334 self
.health_check_file
, e
339 if command
== "placement":
340 self
.ns
.update_nsrs_with_pla_result(params
)
342 elif topic
== "k8scluster":
343 if command
== "create" or command
== "created":
344 k8scluster_id
= params
.get("_id")
345 task
= asyncio
.ensure_future(self
.k8scluster
.create(params
, order_id
))
346 self
.lcm_tasks
.register(
347 "k8scluster", k8scluster_id
, order_id
, "k8scluster_create", task
350 elif command
== "edit" or command
== "edited":
351 k8scluster_id
= params
.get("_id")
352 task
= asyncio
.ensure_future(self
.k8scluster
.edit(params
, order_id
))
353 self
.lcm_tasks
.register(
354 "k8scluster", k8scluster_id
, order_id
, "k8scluster_edit", task
357 elif command
== "delete" or command
== "deleted":
358 k8scluster_id
= params
.get("_id")
359 task
= asyncio
.ensure_future(self
.k8scluster
.delete(params
, order_id
))
360 self
.lcm_tasks
.register(
361 "k8scluster", k8scluster_id
, order_id
, "k8scluster_delete", task
365 if command
== "create" or command
== "created":
366 vca_id
= params
.get("_id")
367 task
= asyncio
.ensure_future(self
.vca
.create(params
, order_id
))
368 self
.lcm_tasks
.register("vca", vca_id
, order_id
, "vca_create", task
)
370 elif command
== "edit" or command
== "edited":
371 vca_id
= params
.get("_id")
372 task
= asyncio
.ensure_future(self
.vca
.edit(params
, order_id
))
373 self
.lcm_tasks
.register("vca", vca_id
, order_id
, "vca_edit", task
)
375 elif command
== "delete" or command
== "deleted":
376 vca_id
= params
.get("_id")
377 task
= asyncio
.ensure_future(self
.vca
.delete(params
, order_id
))
378 self
.lcm_tasks
.register("vca", vca_id
, order_id
, "vca_delete", task
)
380 elif topic
== "k8srepo":
381 if command
== "create" or command
== "created":
382 k8srepo_id
= params
.get("_id")
383 self
.logger
.debug("k8srepo_id = {}".format(k8srepo_id
))
384 task
= asyncio
.ensure_future(self
.k8srepo
.create(params
, order_id
))
385 self
.lcm_tasks
.register(
386 "k8srepo", k8srepo_id
, order_id
, "k8srepo_create", task
389 elif command
== "delete" or command
== "deleted":
390 k8srepo_id
= params
.get("_id")
391 task
= asyncio
.ensure_future(self
.k8srepo
.delete(params
, order_id
))
392 self
.lcm_tasks
.register(
393 "k8srepo", k8srepo_id
, order_id
, "k8srepo_delete", task
397 if command
== "instantiate":
398 # self.logger.debug("Deploying NS {}".format(nsr_id))
400 nslcmop_id
= nslcmop
["_id"]
401 nsr_id
= nslcmop
["nsInstanceId"]
402 task
= asyncio
.ensure_future(self
.ns
.instantiate(nsr_id
, nslcmop_id
))
403 self
.lcm_tasks
.register(
404 "ns", nsr_id
, nslcmop_id
, "ns_instantiate", task
407 elif command
== "terminate":
408 # self.logger.debug("Deleting NS {}".format(nsr_id))
410 nslcmop_id
= nslcmop
["_id"]
411 nsr_id
= nslcmop
["nsInstanceId"]
412 self
.lcm_tasks
.cancel(topic
, nsr_id
)
413 task
= asyncio
.ensure_future(self
.ns
.terminate(nsr_id
, nslcmop_id
))
414 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_terminate", task
)
416 elif command
== "vca_status_refresh":
418 nslcmop_id
= nslcmop
["_id"]
419 nsr_id
= nslcmop
["nsInstanceId"]
420 task
= asyncio
.ensure_future(
421 self
.ns
.vca_status_refresh(nsr_id
, nslcmop_id
)
423 self
.lcm_tasks
.register(
424 "ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh", task
427 elif command
== "action":
428 # self.logger.debug("Update NS {}".format(nsr_id))
430 nslcmop_id
= nslcmop
["_id"]
431 nsr_id
= nslcmop
["nsInstanceId"]
432 task
= asyncio
.ensure_future(self
.ns
.action(nsr_id
, nslcmop_id
))
433 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_action", task
)
435 elif command
== "update":
436 # self.logger.debug("Update NS {}".format(nsr_id))
438 nslcmop_id
= nslcmop
["_id"]
439 nsr_id
= nslcmop
["nsInstanceId"]
440 task
= asyncio
.ensure_future(self
.ns
.update(nsr_id
, nslcmop_id
))
441 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_update", task
)
443 elif command
== "scale":
444 # self.logger.debug("Update NS {}".format(nsr_id))
446 nslcmop_id
= nslcmop
["_id"]
447 nsr_id
= nslcmop
["nsInstanceId"]
448 task
= asyncio
.ensure_future(self
.ns
.scale(nsr_id
, nslcmop_id
))
449 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_scale", task
)
451 elif command
== "heal":
452 # self.logger.debug("Healing NS {}".format(nsr_id))
454 nslcmop_id
= nslcmop
["_id"]
455 nsr_id
= nslcmop
["nsInstanceId"]
456 task
= asyncio
.ensure_future(self
.ns
.heal(nsr_id
, nslcmop_id
))
457 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_heal", task
)
459 elif command
== "migrate":
461 nslcmop_id
= nslcmop
["_id"]
462 nsr_id
= nslcmop
["nsInstanceId"]
463 task
= asyncio
.ensure_future(self
.ns
.migrate(nsr_id
, nslcmop_id
))
464 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "ns_migrate", task
)
466 elif command
== "verticalscale":
468 nslcmop_id
= nslcmop
["_id"]
469 nsr_id
= nslcmop
["nsInstanceId"]
470 task
= asyncio
.ensure_future(self
.ns
.vertical_scale(nsr_id
, nslcmop_id
))
472 "nsr_id,nslcmop_id,task {},{},{}".format(nsr_id
, nslcmop_id
, task
)
474 self
.lcm_tasks
.register(
475 "ns", nsr_id
, nslcmop_id
, "ns_verticalscale", task
478 "LCM task registered {},{},{} ".format(nsr_id
, nslcmop_id
, task
)
481 elif command
== "show":
484 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
486 "nsr:\n _id={}\n operational-status: {}\n config-status: {}"
487 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
490 db_nsr
["operational-status"],
491 db_nsr
["config-status"],
492 db_nsr
["detailed-status"],
493 db_nsr
["_admin"]["deployed"],
494 self
.lcm_tasks
.task_registry
["ns"].get(nsr_id
, ""),
497 except Exception as e
:
498 print("nsr {} not found: {}".format(nsr_id
, e
))
501 elif command
== "deleted":
502 return # TODO cleaning of task just in case should be done
514 ): # "scaled-cooldown-time"
517 elif topic
== "nsi": # netslice LCM processes (instantiate, terminate, etc)
518 if command
== "instantiate":
519 # self.logger.debug("Instantiating Network Slice {}".format(nsilcmop["netsliceInstanceId"]))
521 nsilcmop_id
= nsilcmop
["_id"] # slice operation id
522 nsir_id
= nsilcmop
["netsliceInstanceId"] # slice record id
523 task
= asyncio
.ensure_future(
524 self
.netslice
.instantiate(nsir_id
, nsilcmop_id
)
526 self
.lcm_tasks
.register(
527 "nsi", nsir_id
, nsilcmop_id
, "nsi_instantiate", task
530 elif command
== "terminate":
531 # self.logger.debug("Terminating Network Slice NS {}".format(nsilcmop["netsliceInstanceId"]))
533 nsilcmop_id
= nsilcmop
["_id"] # slice operation id
534 nsir_id
= nsilcmop
["netsliceInstanceId"] # slice record id
535 self
.lcm_tasks
.cancel(topic
, nsir_id
)
536 task
= asyncio
.ensure_future(
537 self
.netslice
.terminate(nsir_id
, nsilcmop_id
)
539 self
.lcm_tasks
.register(
540 "nsi", nsir_id
, nsilcmop_id
, "nsi_terminate", task
543 elif command
== "show":
546 db_nsir
= self
.db
.get_one("nsirs", {"_id": nsir_id
})
548 "nsir:\n _id={}\n operational-status: {}\n config-status: {}"
549 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
552 db_nsir
["operational-status"],
553 db_nsir
["config-status"],
554 db_nsir
["detailed-status"],
555 db_nsir
["_admin"]["deployed"],
556 self
.lcm_tasks
.task_registry
["nsi"].get(nsir_id
, ""),
559 except Exception as e
:
560 print("nsir {} not found: {}".format(nsir_id
, e
))
563 elif command
== "deleted":
564 return # TODO cleaning of task just in case should be done
571 ): # "scaled-cooldown-time"
573 elif topic
== "vim_account":
574 vim_id
= params
["_id"]
575 if command
in ("create", "created"):
576 if not self
.main_config
.RO
.ng
:
577 task
= asyncio
.ensure_future(self
.vim
.create(params
, order_id
))
578 self
.lcm_tasks
.register(
579 "vim_account", vim_id
, order_id
, "vim_create", task
582 elif command
== "delete" or command
== "deleted":
583 self
.lcm_tasks
.cancel(topic
, vim_id
)
584 task
= asyncio
.ensure_future(self
.vim
.delete(params
, order_id
))
585 self
.lcm_tasks
.register(
586 "vim_account", vim_id
, order_id
, "vim_delete", task
589 elif command
== "show":
590 print("not implemented show with vim_account")
593 elif command
in ("edit", "edited"):
594 if not self
.main_config
.RO
.ng
:
595 task
= asyncio
.ensure_future(self
.vim
.edit(params
, order_id
))
596 self
.lcm_tasks
.register(
597 "vim_account", vim_id
, order_id
, "vim_edit", task
600 elif command
== "deleted":
601 return # TODO cleaning of task just in case should be done
602 elif topic
== "wim_account":
603 wim_id
= params
["_id"]
604 if command
in ("create", "created"):
605 if not self
.main_config
.RO
.ng
:
606 task
= asyncio
.ensure_future(self
.wim
.create(params
, order_id
))
607 self
.lcm_tasks
.register(
608 "wim_account", wim_id
, order_id
, "wim_create", task
611 elif command
== "delete" or command
== "deleted":
612 self
.lcm_tasks
.cancel(topic
, wim_id
)
613 task
= asyncio
.ensure_future(self
.wim
.delete(params
, order_id
))
614 self
.lcm_tasks
.register(
615 "wim_account", wim_id
, order_id
, "wim_delete", task
618 elif command
== "show":
619 print("not implemented show with wim_account")
622 elif command
in ("edit", "edited"):
623 task
= asyncio
.ensure_future(self
.wim
.edit(params
, order_id
))
624 self
.lcm_tasks
.register(
625 "wim_account", wim_id
, order_id
, "wim_edit", task
628 elif command
== "deleted":
629 return # TODO cleaning of task just in case should be done
631 _sdn_id
= params
["_id"]
632 if command
in ("create", "created"):
633 if not self
.main_config
.RO
.ng
:
634 task
= asyncio
.ensure_future(self
.sdn
.create(params
, order_id
))
635 self
.lcm_tasks
.register(
636 "sdn", _sdn_id
, order_id
, "sdn_create", task
639 elif command
== "delete" or command
== "deleted":
640 self
.lcm_tasks
.cancel(topic
, _sdn_id
)
641 task
= asyncio
.ensure_future(self
.sdn
.delete(params
, order_id
))
642 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_delete", task
)
644 elif command
in ("edit", "edited"):
645 task
= asyncio
.ensure_future(self
.sdn
.edit(params
, order_id
))
646 self
.lcm_tasks
.register("sdn", _sdn_id
, order_id
, "sdn_edit", task
)
648 elif command
== "deleted":
649 return # TODO cleaning of task just in case should be done
650 self
.logger
.critical("unknown topic {} and command '{}'".format(topic
, command
))
652 async def kafka_read(self
):
654 "Task kafka_read Enter with worker_id={}".format(self
.worker_id
)
656 # future = asyncio.Future()
657 self
.consecutive_errors
= 0
658 self
.first_start
= True
659 while self
.consecutive_errors
< 10:
672 topics_admin
= ("admin",)
673 await asyncio
.gather(
675 topics
, self
.loop
, self
.kafka_read_callback
, from_beginning
=True
677 self
.msg_admin
.aioread(
680 self
.kafka_read_callback
,
685 except LcmExceptionExit
:
686 self
.logger
.debug("Bye!")
688 except Exception as e
:
689 # if not first_start is the first time after starting. So leave more time and wait
690 # to allow kafka starts
691 if self
.consecutive_errors
== 8 if not self
.first_start
else 30:
693 "Task kafka_read task exit error too many errors. Exception: {}".format(
698 self
.consecutive_errors
+= 1
700 "Task kafka_read retrying after Exception {}".format(e
)
702 wait_time
= 2 if not self
.first_start
else 5
703 await asyncio
.sleep(wait_time
, loop
=self
.loop
)
705 # self.logger.debug("Task kafka_read terminating")
706 self
.logger
.debug("Task kafka_read exit")
710 self
.loop
.run_until_complete(self
.check_RO_version())
712 self
.ns
= ns
.NsLcm(self
.msg
, self
.lcm_tasks
, self
.main_config
, self
.loop
)
713 # TODO: modify the rest of classes to use the LcmCfg object instead of dicts
714 self
.netslice
= netslice
.NetsliceLcm(
715 self
.msg
, self
.lcm_tasks
, self
.main_config
.to_dict(), self
.loop
, self
.ns
717 self
.vim
= vim_sdn
.VimLcm(
718 self
.msg
, self
.lcm_tasks
, self
.main_config
.to_dict(), self
.loop
720 self
.wim
= vim_sdn
.WimLcm(
721 self
.msg
, self
.lcm_tasks
, self
.main_config
.to_dict(), self
.loop
723 self
.sdn
= vim_sdn
.SdnLcm(
724 self
.msg
, self
.lcm_tasks
, self
.main_config
.to_dict(), self
.loop
726 self
.k8scluster
= vim_sdn
.K8sClusterLcm(
727 self
.msg
, self
.lcm_tasks
, self
.main_config
.to_dict(), self
.loop
729 self
.vca
= vim_sdn
.VcaLcm(
730 self
.msg
, self
.lcm_tasks
, self
.main_config
.to_dict(), self
.loop
732 self
.k8srepo
= vim_sdn
.K8sRepoLcm(
733 self
.msg
, self
.lcm_tasks
, self
.main_config
.to_dict(), self
.loop
736 self
.loop
.run_until_complete(
737 asyncio
.gather(self
.kafka_read(), self
.kafka_ping())
741 # self.logger.debug("Terminating cancelling creation tasks")
742 # self.lcm_tasks.cancel("ALL", "create")
744 # while self.is_pending_tasks():
745 # self.logger.debug("Task kafka_read terminating. Waiting for tasks termination")
746 # await asyncio.sleep(2, loop=self.loop)
749 # self.lcm_tasks.cancel("ALL", "ALL")
753 self
.db
.db_disconnect()
755 self
.msg
.disconnect()
757 self
.msg_admin
.disconnect()
759 self
.fs
.fs_disconnect()
761 def read_config_file(self
, config_file
):
763 with
open(config_file
) as f
:
764 return yaml
.safe_load(f
)
765 except Exception as e
:
766 self
.logger
.critical("At config file '{}': {}".format(config_file
, e
))
770 def get_process_id():
772 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
773 will provide a random one
776 # Try getting docker id. If fails, get pid
778 with
open("/proc/self/cgroup", "r") as f
:
779 text_id_
= f
.readline()
780 _
, _
, text_id
= text_id_
.rpartition("/")
781 text_id
= text_id
.replace("\n", "")[:12]
787 return "".join(random_choice("0123456789abcdef") for _
in range(12))
792 """Usage: {} [options]
793 -c|--config [configuration_file]: loads the configuration file (default: ./lcm.cfg)
794 --health-check: do not run lcm, but inspect kafka bus to determine if lcm is healthy
795 -h|--help: shows this help
800 # --log-socket-host HOST: send logs to this host")
801 # --log-socket-port PORT: send logs using this port (default: 9022)")
804 if __name__
== "__main__":
806 # print("SYS.PATH='{}'".format(sys.path))
807 # load parameters and configuration
813 opts
, args
= getopt
.getopt(
814 sys
.argv
[1:], "hc:", ["config=", "help", "health-check"]
816 # TODO add "log-socket-host=", "log-socket-port=", "log-file="
819 if o
in ("-h", "--help"):
822 elif o
in ("-c", "--config"):
824 elif o
== "--health-check":
825 from osm_lcm
.lcm_hc
import health_check
827 health_check(config_file
, Lcm
.ping_interval_pace
)
828 # elif o == "--log-socket-port":
829 # log_socket_port = a
830 # elif o == "--log-socket-host":
831 # log_socket_host = a
832 # elif o == "--log-file":
835 assert False, "Unhandled option"
838 if not path
.isfile(config_file
):
840 "configuration file '{}' does not exist".format(config_file
),
846 __file__
[: __file__
.rfind(".")] + ".cfg",
850 if path
.isfile(config_file
):
854 "No configuration file 'lcm.cfg' found neither at local folder nor at /etc/osm/",
858 lcm
= Lcm(config_file
)
860 except (LcmException
, getopt
.GetoptError
) as e
:
861 print(str(e
), file=sys
.stderr
)