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