Add n2vc relations after install configuration sw
[osm/LCM.git] / osm_lcm / lcm.py
1 #!/usr/bin/python3
2 # -*- coding: utf-8 -*-
3
4 ##
5 # Copyright 2018 Telefonica S.A.
6 #
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
10 #
11 # http://www.apache.org/licenses/LICENSE-2.0
12 #
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
17 # under the License.
18 ##
19
20
21 # DEBUG WITH PDB
22 import os
23 import pdb
24
25 import asyncio
26 import yaml
27 import logging
28 import logging.handlers
29 import getopt
30 import sys
31
32 from osm_lcm import ns
33 from osm_lcm import vim_sdn
34 from osm_lcm import netslice
35 from osm_lcm import ROclient
36
37 from time import time, sleep
38 from osm_lcm.lcm_utils import versiontuple, LcmException, TaskRegistry, LcmExceptionExit
39 from osm_lcm import version as lcm_version, version_date as lcm_version_date
40
41 from osm_common import dbmemory, dbmongo, fslocal, fsmongo, msglocal, msgkafka
42 from osm_common import version as common_version
43 from osm_common.dbbase import DbException
44 from osm_common.fsbase import FsException
45 from osm_common.msgbase import MsgException
46 from os import environ, path
47 from random import choice as random_choice
48 from n2vc import version as n2vc_version
49
50 if os.getenv('OSMLCM_PDB_DEBUG', None) is not None:
51 pdb.set_trace()
52
53
54 __author__ = "Alfonso Tierno"
55 min_RO_version = "6.0.2"
56 min_n2vc_version = "0.0.2"
57
58 min_common_version = "0.1.19"
59 health_check_file = path.expanduser("~") + "/time_last_ping" # TODO find better location for this file
60
61
62 class Lcm:
63
64 ping_interval_pace = 120 # how many time ping is send once is confirmed all is running
65 ping_interval_boot = 5 # how many time ping is sent when booting
66
67 def __init__(self, config_file, loop=None):
68 """
69 Init, Connect to database, filesystem storage, and messaging
70 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
71 :return: None
72 """
73 self.db = None
74 self.msg = None
75 self.msg_admin = None
76 self.fs = None
77 self.pings_not_received = 1
78 self.consecutive_errors = 0
79 self.first_start = False
80
81 # logging
82 self.logger = logging.getLogger('lcm')
83 # get id
84 self.worker_id = self.get_process_id()
85 # load configuration
86 config = self.read_config_file(config_file)
87 self.config = config
88 self.config["ro_config"] = {
89 "endpoint_url": "http://{}:{}/openmano".format(config["RO"]["host"], config["RO"]["port"]),
90 "tenant": config.get("tenant", "osm"),
91 "logger_name": "lcm.ROclient",
92 "loglevel": "ERROR",
93 }
94
95 self.loop = loop or asyncio.get_event_loop()
96
97 # logging
98 log_format_simple = "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)s %(message)s"
99 log_formatter_simple = logging.Formatter(log_format_simple, datefmt='%Y-%m-%dT%H:%M:%S')
100 config["database"]["logger_name"] = "lcm.db"
101 config["storage"]["logger_name"] = "lcm.fs"
102 config["message"]["logger_name"] = "lcm.msg"
103 if config["global"].get("logfile"):
104 file_handler = logging.handlers.RotatingFileHandler(config["global"]["logfile"],
105 maxBytes=100e6, backupCount=9, delay=0)
106 file_handler.setFormatter(log_formatter_simple)
107 self.logger.addHandler(file_handler)
108 if not config["global"].get("nologging"):
109 str_handler = logging.StreamHandler()
110 str_handler.setFormatter(log_formatter_simple)
111 self.logger.addHandler(str_handler)
112
113 if config["global"].get("loglevel"):
114 self.logger.setLevel(config["global"]["loglevel"])
115
116 # logging other modules
117 for k1, logname in {"message": "lcm.msg", "database": "lcm.db", "storage": "lcm.fs"}.items():
118 config[k1]["logger_name"] = logname
119 logger_module = logging.getLogger(logname)
120 if config[k1].get("logfile"):
121 file_handler = logging.handlers.RotatingFileHandler(config[k1]["logfile"],
122 maxBytes=100e6, backupCount=9, delay=0)
123 file_handler.setFormatter(log_formatter_simple)
124 logger_module.addHandler(file_handler)
125 if config[k1].get("loglevel"):
126 logger_module.setLevel(config[k1]["loglevel"])
127 self.logger.critical("starting osm/lcm version {} {}".format(lcm_version, lcm_version_date))
128
129 # check version of N2VC
130 # TODO enhance with int conversion or from distutils.version import LooseVersion
131 # or with list(map(int, version.split(".")))
132 if versiontuple(n2vc_version) < versiontuple(min_n2vc_version):
133 raise LcmException("Not compatible osm/N2VC version '{}'. Needed '{}' or higher".format(
134 n2vc_version, min_n2vc_version))
135 # check version of common
136 if versiontuple(common_version) < versiontuple(min_common_version):
137 raise LcmException("Not compatible osm/common version '{}'. Needed '{}' or higher".format(
138 common_version, min_common_version))
139
140 try:
141 # TODO check database version
142 if config["database"]["driver"] == "mongo":
143 self.db = dbmongo.DbMongo()
144 self.db.db_connect(config["database"])
145 elif config["database"]["driver"] == "memory":
146 self.db = dbmemory.DbMemory()
147 self.db.db_connect(config["database"])
148 else:
149 raise LcmException("Invalid configuration param '{}' at '[database]':'driver'".format(
150 config["database"]["driver"]))
151
152 if config["storage"]["driver"] == "local":
153 self.fs = fslocal.FsLocal()
154 self.fs.fs_connect(config["storage"])
155 elif config["storage"]["driver"] == "mongo":
156 self.fs = fsmongo.FsMongo()
157 self.fs.fs_connect(config["storage"])
158 else:
159 raise LcmException("Invalid configuration param '{}' at '[storage]':'driver'".format(
160 config["storage"]["driver"]))
161
162 # copy message configuration in order to remove 'group_id' for msg_admin
163 config_message = config["message"].copy()
164 config_message["loop"] = self.loop
165 if config_message["driver"] == "local":
166 self.msg = msglocal.MsgLocal()
167 self.msg.connect(config_message)
168 self.msg_admin = msglocal.MsgLocal()
169 config_message.pop("group_id", None)
170 self.msg_admin.connect(config_message)
171 elif config_message["driver"] == "kafka":
172 self.msg = msgkafka.MsgKafka()
173 self.msg.connect(config_message)
174 self.msg_admin = msgkafka.MsgKafka()
175 config_message.pop("group_id", None)
176 self.msg_admin.connect(config_message)
177 else:
178 raise LcmException("Invalid configuration param '{}' at '[message]':'driver'".format(
179 config["message"]["driver"]))
180 except (DbException, FsException, MsgException) as e:
181 self.logger.critical(str(e), exc_info=True)
182 raise LcmException(str(e))
183
184 # contains created tasks/futures to be able to cancel
185 self.lcm_tasks = TaskRegistry(self.worker_id, self.db, self.logger)
186
187 self.ns = ns.NsLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
188 self.netslice = netslice.NetsliceLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
189 self.vim = vim_sdn.VimLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
190 self.wim = vim_sdn.WimLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
191 self.sdn = vim_sdn.SdnLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
192 self.k8scluster = vim_sdn.K8sClusterLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
193 self.k8srepo = vim_sdn.K8sRepoLcm(self.db, self.msg, self.fs, self.lcm_tasks, self.config, self.loop)
194
195 async def check_RO_version(self):
196 tries = 14
197 last_error = None
198 while True:
199 try:
200 ro_server = ROclient.ROClient(self.loop, **self.config["ro_config"])
201 ro_version = await ro_server.get_version()
202 if versiontuple(ro_version) < versiontuple(min_RO_version):
203 raise LcmException("Not compatible osm/RO version '{}'. Needed '{}' or higher".format(
204 ro_version, min_RO_version))
205 self.logger.info("Connected to RO version {}".format(ro_version))
206 return
207 except ROclient.ROClientException as e:
208 tries -= 1
209 error_text = "Error while connecting to RO on {}: {}".format(self.config["ro_config"]["endpoint_url"],
210 e)
211 if tries <= 0:
212 self.logger.critical(error_text)
213 raise LcmException(error_text)
214 if last_error != error_text:
215 last_error = error_text
216 self.logger.error(error_text + ". Waiting until {} seconds".format(5*tries))
217 await asyncio.sleep(5)
218
219 async def test(self, param=None):
220 self.logger.debug("Starting/Ending test task: {}".format(param))
221
222 async def kafka_ping(self):
223 self.logger.debug("Task kafka_ping Enter")
224 consecutive_errors = 0
225 first_start = True
226 kafka_has_received = False
227 self.pings_not_received = 1
228 while True:
229 try:
230 await self.msg_admin.aiowrite(
231 "admin", "ping",
232 {"from": "lcm", "to": "lcm", "worker_id": self.worker_id, "version": lcm_version},
233 self.loop)
234 # time between pings are low when it is not received and at starting
235 wait_time = self.ping_interval_boot if not kafka_has_received else self.ping_interval_pace
236 if not self.pings_not_received:
237 kafka_has_received = True
238 self.pings_not_received += 1
239 await asyncio.sleep(wait_time, loop=self.loop)
240 if self.pings_not_received > 10:
241 raise LcmException("It is not receiving pings from Kafka bus")
242 consecutive_errors = 0
243 first_start = False
244 except LcmException:
245 raise
246 except Exception as e:
247 # if not first_start is the first time after starting. So leave more time and wait
248 # to allow kafka starts
249 if consecutive_errors == 8 if not first_start else 30:
250 self.logger.error("Task kafka_read task exit error too many errors. Exception: {}".format(e))
251 raise
252 consecutive_errors += 1
253 self.logger.error("Task kafka_read retrying after Exception {}".format(e))
254 wait_time = 2 if not first_start else 5
255 await asyncio.sleep(wait_time, loop=self.loop)
256
257 def kafka_read_callback(self, topic, command, params):
258 order_id = 1
259
260 if topic != "admin" and command != "ping":
261 self.logger.debug("Task kafka_read receives {} {}: {}".format(topic, command, params))
262 self.consecutive_errors = 0
263 self.first_start = False
264 order_id += 1
265 if command == "exit":
266 raise LcmExceptionExit
267 elif command.startswith("#"):
268 return
269 elif command == "echo":
270 # just for test
271 print(params)
272 sys.stdout.flush()
273 return
274 elif command == "test":
275 asyncio.Task(self.test(params), loop=self.loop)
276 return
277
278 if topic == "admin":
279 if command == "ping" and params["to"] == "lcm" and params["from"] == "lcm":
280 if params.get("worker_id") != self.worker_id:
281 return
282 self.pings_not_received = 0
283 try:
284 with open(health_check_file, "w") as f:
285 f.write(str(time()))
286 except Exception as e:
287 self.logger.error("Cannot write into '{}' for healthcheck: {}".format(health_check_file, e))
288 return
289 elif topic == "k8scluster":
290 if command == "create" or command == "created":
291 k8scluster_id = params.get("_id")
292 task = asyncio.ensure_future(self.k8scluster.create(params, order_id))
293 self.lcm_tasks.register("k8scluster", k8scluster_id, order_id, "k8scluster_create", task)
294 return
295 elif command == "delete" or command == "deleted":
296 k8scluster_id = params.get("_id")
297 task = asyncio.ensure_future(self.k8scluster.delete(params, order_id))
298 self.lcm_tasks.register("k8scluster", k8scluster_id, order_id, "k8scluster_delete", task)
299 return
300 elif topic == "k8srepo":
301 if command == "create" or command == "created":
302 k8srepo_id = params.get("_id")
303 self.logger.debug("k8srepo_id = {}".format(k8srepo_id))
304 task = asyncio.ensure_future(self.k8srepo.create(params, order_id))
305 self.lcm_tasks.register("k8srepo", k8srepo_id, order_id, "k8srepo_create", task)
306 return
307 elif command == "delete" or command == "deleted":
308 k8srepo_id = params.get("_id")
309 task = asyncio.ensure_future(self.k8srepo.delete(params, order_id))
310 self.lcm_tasks.register("k8srepo", k8srepo_id, order_id, "k8srepo_delete", task)
311 return
312 elif topic == "ns":
313 if command == "instantiate" or command == "instantiated":
314 # self.logger.debug("Deploying NS {}".format(nsr_id))
315 nslcmop = params
316 nslcmop_id = nslcmop["_id"]
317 nsr_id = nslcmop["nsInstanceId"]
318 task = asyncio.ensure_future(self.ns.instantiate(nsr_id, nslcmop_id))
319 self.lcm_tasks.register("ns", nsr_id, nslcmop_id, "ns_instantiate", task)
320 return
321 elif command == "terminate" or command == "terminated":
322 # self.logger.debug("Deleting NS {}".format(nsr_id))
323 nslcmop = params
324 nslcmop_id = nslcmop["_id"]
325 nsr_id = nslcmop["nsInstanceId"]
326 self.lcm_tasks.cancel(topic, nsr_id)
327 task = asyncio.ensure_future(self.ns.terminate(nsr_id, nslcmop_id))
328 self.lcm_tasks.register("ns", nsr_id, nslcmop_id, "ns_terminate", task)
329 return
330 elif command == "action":
331 # self.logger.debug("Update NS {}".format(nsr_id))
332 nslcmop = params
333 nslcmop_id = nslcmop["_id"]
334 nsr_id = nslcmop["nsInstanceId"]
335 task = asyncio.ensure_future(self.ns.action(nsr_id, nslcmop_id))
336 self.lcm_tasks.register("ns", nsr_id, nslcmop_id, "ns_action", task)
337 return
338 elif command == "scale":
339 # self.logger.debug("Update NS {}".format(nsr_id))
340 nslcmop = params
341 nslcmop_id = nslcmop["_id"]
342 nsr_id = nslcmop["nsInstanceId"]
343 task = asyncio.ensure_future(self.ns.scale(nsr_id, nslcmop_id))
344 self.lcm_tasks.register("ns", nsr_id, nslcmop_id, "ns_scale", task)
345 return
346 elif command == "show":
347 nsr_id = params
348 try:
349 db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
350 print("nsr:\n _id={}\n operational-status: {}\n config-status: {}"
351 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
352 "".format(nsr_id, db_nsr["operational-status"], db_nsr["config-status"],
353 db_nsr["detailed-status"],
354 db_nsr["_admin"]["deployed"], self.lcm_ns_tasks.get(nsr_id)))
355 except Exception as e:
356 print("nsr {} not found: {}".format(nsr_id, e))
357 sys.stdout.flush()
358 return
359 elif command == "deleted":
360 return # TODO cleaning of task just in case should be done
361 elif command in ("terminated", "instantiated", "scaled", "actioned"): # "scaled-cooldown-time"
362 return
363 elif topic == "nsi": # netslice LCM processes (instantiate, terminate, etc)
364 if command == "instantiate" or command == "instantiated":
365 # self.logger.debug("Instantiating Network Slice {}".format(nsilcmop["netsliceInstanceId"]))
366 nsilcmop = params
367 nsilcmop_id = nsilcmop["_id"] # slice operation id
368 nsir_id = nsilcmop["netsliceInstanceId"] # slice record id
369 task = asyncio.ensure_future(self.netslice.instantiate(nsir_id, nsilcmop_id))
370 self.lcm_tasks.register("nsi", nsir_id, nsilcmop_id, "nsi_instantiate", task)
371 return
372 elif command == "terminate" or command == "terminated":
373 # self.logger.debug("Terminating Network Slice NS {}".format(nsilcmop["netsliceInstanceId"]))
374 nsilcmop = params
375 nsilcmop_id = nsilcmop["_id"] # slice operation id
376 nsir_id = nsilcmop["netsliceInstanceId"] # slice record id
377 self.lcm_tasks.cancel(topic, nsir_id)
378 task = asyncio.ensure_future(self.netslice.terminate(nsir_id, nsilcmop_id))
379 self.lcm_tasks.register("nsi", nsir_id, nsilcmop_id, "nsi_terminate", task)
380 return
381 elif command == "show":
382 nsir_id = params
383 try:
384 db_nsir = self.db.get_one("nsirs", {"_id": nsir_id})
385 print("nsir:\n _id={}\n operational-status: {}\n config-status: {}"
386 "\n detailed-status: {}\n deploy: {}\n tasks: {}"
387 "".format(nsir_id, db_nsir["operational-status"], db_nsir["config-status"],
388 db_nsir["detailed-status"],
389 db_nsir["_admin"]["deployed"], self.lcm_netslice_tasks.get(nsir_id)))
390 except Exception as e:
391 print("nsir {} not found: {}".format(nsir_id, e))
392 sys.stdout.flush()
393 return
394 elif command == "deleted":
395 return # TODO cleaning of task just in case should be done
396 elif command in ("terminated", "instantiated", "scaled", "actioned"): # "scaled-cooldown-time"
397 return
398 elif topic == "vim_account":
399 vim_id = params["_id"]
400 if command in ("create", "created"):
401 task = asyncio.ensure_future(self.vim.create(params, order_id))
402 self.lcm_tasks.register("vim_account", vim_id, order_id, "vim_create", task)
403 return
404 elif command == "delete" or command == "deleted":
405 self.lcm_tasks.cancel(topic, vim_id)
406 task = asyncio.ensure_future(self.vim.delete(params, order_id))
407 self.lcm_tasks.register("vim_account", vim_id, order_id, "vim_delete", task)
408 return
409 elif command == "show":
410 print("not implemented show with vim_account")
411 sys.stdout.flush()
412 return
413 elif command in ("edit", "edited"):
414 task = asyncio.ensure_future(self.vim.edit(params, order_id))
415 self.lcm_tasks.register("vim_account", vim_id, order_id, "vim_edit", task)
416 return
417 elif command == "deleted":
418 return # TODO cleaning of task just in case should be done
419 elif topic == "wim_account":
420 wim_id = params["_id"]
421 if command in ("create", "created"):
422 task = asyncio.ensure_future(self.wim.create(params, order_id))
423 self.lcm_tasks.register("wim_account", wim_id, order_id, "wim_create", task)
424 return
425 elif command == "delete" or command == "deleted":
426 self.lcm_tasks.cancel(topic, wim_id)
427 task = asyncio.ensure_future(self.wim.delete(params, order_id))
428 self.lcm_tasks.register("wim_account", wim_id, order_id, "wim_delete", task)
429 return
430 elif command == "show":
431 print("not implemented show with wim_account")
432 sys.stdout.flush()
433 return
434 elif command in ("edit", "edited"):
435 task = asyncio.ensure_future(self.wim.edit(params, order_id))
436 self.lcm_tasks.register("wim_account", wim_id, order_id, "wim_edit", task)
437 return
438 elif command == "deleted":
439 return # TODO cleaning of task just in case should be done
440 elif topic == "sdn":
441 _sdn_id = params["_id"]
442 if command in ("create", "created"):
443 task = asyncio.ensure_future(self.sdn.create(params, order_id))
444 self.lcm_tasks.register("sdn", _sdn_id, order_id, "sdn_create", task)
445 return
446 elif command == "delete" or command == "deleted":
447 self.lcm_tasks.cancel(topic, _sdn_id)
448 task = asyncio.ensure_future(self.sdn.delete(params, order_id))
449 self.lcm_tasks.register("sdn", _sdn_id, order_id, "sdn_delete", task)
450 return
451 elif command in ("edit", "edited"):
452 task = asyncio.ensure_future(self.sdn.edit(params, order_id))
453 self.lcm_tasks.register("sdn", _sdn_id, order_id, "sdn_edit", task)
454 return
455 elif command == "deleted":
456 return # TODO cleaning of task just in case should be done
457 self.logger.critical("unknown topic {} and command '{}'".format(topic, command))
458
459 async def kafka_read(self):
460 self.logger.debug("Task kafka_read Enter with worker_id={}".format(self.worker_id))
461 # future = asyncio.Future()
462 self.consecutive_errors = 0
463 self.first_start = True
464 while self.consecutive_errors < 10:
465 try:
466 topics = ("ns", "vim_account", "wim_account", "sdn", "nsi", "k8scluster", "k8srepo")
467 topics_admin = ("admin", )
468 await asyncio.gather(
469 self.msg.aioread(topics, self.loop, self.kafka_read_callback),
470 self.msg_admin.aioread(topics_admin, self.loop, self.kafka_read_callback, group_id=False)
471 )
472
473 except LcmExceptionExit:
474 self.logger.debug("Bye!")
475 break
476 except Exception as e:
477 # if not first_start is the first time after starting. So leave more time and wait
478 # to allow kafka starts
479 if self.consecutive_errors == 8 if not self.first_start else 30:
480 self.logger.error("Task kafka_read task exit error too many errors. Exception: {}".format(e))
481 raise
482 self.consecutive_errors += 1
483 self.logger.error("Task kafka_read retrying after Exception {}".format(e))
484 wait_time = 2 if not self.first_start else 5
485 await asyncio.sleep(wait_time, loop=self.loop)
486
487 # self.logger.debug("Task kafka_read terminating")
488 self.logger.debug("Task kafka_read exit")
489
490 def start(self):
491
492 # check RO version
493 self.loop.run_until_complete(self.check_RO_version())
494
495 self.loop.run_until_complete(asyncio.gather(
496 self.kafka_read(),
497 self.kafka_ping()
498 ))
499 # TODO
500 # self.logger.debug("Terminating cancelling creation tasks")
501 # self.lcm_tasks.cancel("ALL", "create")
502 # timeout = 200
503 # while self.is_pending_tasks():
504 # self.logger.debug("Task kafka_read terminating. Waiting for tasks termination")
505 # await asyncio.sleep(2, loop=self.loop)
506 # timeout -= 2
507 # if not timeout:
508 # self.lcm_tasks.cancel("ALL", "ALL")
509 self.loop.close()
510 self.loop = None
511 if self.db:
512 self.db.db_disconnect()
513 if self.msg:
514 self.msg.disconnect()
515 if self.msg_admin:
516 self.msg_admin.disconnect()
517 if self.fs:
518 self.fs.fs_disconnect()
519
520 def read_config_file(self, config_file):
521 # TODO make a [ini] + yaml inside parser
522 # the configparser library is not suitable, because it does not admit comments at the end of line,
523 # and not parse integer or boolean
524 try:
525 # read file as yaml format
526 with open(config_file) as f:
527 conf = yaml.load(f, Loader=yaml.Loader)
528 # Ensure all sections are not empty
529 for k in ("global", "timeout", "RO", "VCA", "database", "storage", "message"):
530 if not conf.get(k):
531 conf[k] = {}
532
533 # read all environ that starts with OSMLCM_
534 for k, v in environ.items():
535 if not k.startswith("OSMLCM_"):
536 continue
537 subject, _, item = k[7:].lower().partition("_")
538 if not item:
539 continue
540 if subject in ("ro", "vca"):
541 # put in capital letter
542 subject = subject.upper()
543 try:
544 if item == "port" or subject == "timeout":
545 conf[subject][item] = int(v)
546 else:
547 conf[subject][item] = v
548 except Exception as e:
549 self.logger.warning("skipping environ '{}' on exception '{}'".format(k, e))
550
551 # backward compatibility of VCA parameters
552
553 if 'pubkey' in conf["VCA"]:
554 conf["VCA"]['public_key'] = conf["VCA"].pop('pubkey')
555 if 'cacert' in conf["VCA"]:
556 conf["VCA"]['ca_cert'] = conf["VCA"].pop('cacert')
557 if 'apiproxy' in conf["VCA"]:
558 conf["VCA"]['api_proxy'] = conf["VCA"].pop('apiproxy')
559
560 if 'enableosupgrade' in conf["VCA"]:
561 conf["VCA"]['enable_os_upgrade'] = conf["VCA"].pop('enableosupgrade')
562 if isinstance(conf["VCA"].get('enable_os_upgrade'), str):
563 if conf["VCA"]['enable_os_upgrade'].lower() == 'false':
564 conf["VCA"]['enable_os_upgrade'] = False
565 elif conf["VCA"]['enable_os_upgrade'].lower() == 'true':
566 conf["VCA"]['enable_os_upgrade'] = True
567
568 if 'aptmirror' in conf["VCA"]:
569 conf["VCA"]['apt_mirror'] = conf["VCA"].pop('aptmirror')
570
571 return conf
572 except Exception as e:
573 self.logger.critical("At config file '{}': {}".format(config_file, e))
574 exit(1)
575
576 @staticmethod
577 def get_process_id():
578 """
579 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
580 will provide a random one
581 :return: Obtained ID
582 """
583 # Try getting docker id. If fails, get pid
584 try:
585 with open("/proc/self/cgroup", "r") as f:
586 text_id_ = f.readline()
587 _, _, text_id = text_id_.rpartition("/")
588 text_id = text_id.replace('\n', '')[:12]
589 if text_id:
590 return text_id
591 except Exception:
592 pass
593 # Return a random id
594 return ''.join(random_choice("0123456789abcdef") for _ in range(12))
595
596
597 def usage():
598 print("""Usage: {} [options]
599 -c|--config [configuration_file]: loads the configuration file (default: ./lcm.cfg)
600 --health-check: do not run lcm, but inspect kafka bus to determine if lcm is healthy
601 -h|--help: shows this help
602 """.format(sys.argv[0]))
603 # --log-socket-host HOST: send logs to this host")
604 # --log-socket-port PORT: send logs using this port (default: 9022)")
605
606
607 def health_check():
608 retry = 2
609 while retry:
610 retry -= 1
611 try:
612 with open(health_check_file, "r") as f:
613 last_received_ping = f.read()
614
615 if time() - float(last_received_ping) < Lcm.ping_interval_pace + 10:
616 exit(0)
617 except Exception:
618 pass
619 if retry:
620 sleep(6)
621 exit(1)
622
623
624 if __name__ == '__main__':
625
626 try:
627 print("SYS.PATH='{}'".format(sys.path))
628 # load parameters and configuration
629 # -h
630 # -c value
631 # --config value
632 # --help
633 # --health-check
634 opts, args = getopt.getopt(sys.argv[1:], "hc:", ["config=", "help", "health-check"])
635 # TODO add "log-socket-host=", "log-socket-port=", "log-file="
636 config_file = None
637 for o, a in opts:
638 if o in ("-h", "--help"):
639 usage()
640 sys.exit()
641 elif o in ("-c", "--config"):
642 config_file = a
643 elif o == "--health-check":
644 health_check()
645 # elif o == "--log-socket-port":
646 # log_socket_port = a
647 # elif o == "--log-socket-host":
648 # log_socket_host = a
649 # elif o == "--log-file":
650 # log_file = a
651 else:
652 assert False, "Unhandled option"
653
654 if config_file:
655 if not path.isfile(config_file):
656 print("configuration file '{}' does not exist".format(config_file), file=sys.stderr)
657 exit(1)
658 else:
659 for config_file in (__file__[:__file__.rfind(".")] + ".cfg", "./lcm.cfg", "/etc/osm/lcm.cfg"):
660 if path.isfile(config_file):
661 break
662 else:
663 print("No configuration file 'lcm.cfg' found neither at local folder nor at /etc/osm/", file=sys.stderr)
664 exit(1)
665 lcm = Lcm(config_file)
666 lcm.start()
667 except (LcmException, getopt.GetoptError) as e:
668 print(str(e), file=sys.stderr)
669 # usage()
670 exit(1)