osm_mon/cmd/mon_utils.py

   1 # -*- coding: utf-8 -*-
   2
   3 # This file is part of OSM Monitoring module
   4
   5 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   6 # not use this file except in compliance with the License. You may obtain
   7 # a copy of the License at
   8
   9 #         http://www.apache.org/licenses/LICENSE-2.0
  10
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  13 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  14 # License for the specific language governing permissions and limitations
  15 # under the License.
  16
  17 import pymongo
  18 import time
  19 import socket
  20 import logging
  21 import kafka
  22
  23
  24 def wait_till_commondb_is_ready(config, process_name="osm-mon", commondb_wait_time=5):
  25
  26     logging.debug("wait_till_commondb_is_ready")
  27
  28     while(True):
  29         commondb_url = config.conf["database"].get("uri")
  30         try:
  31             commondb = pymongo.MongoClient(commondb_url)
  32             commondb.server_info()
  33             break
  34         except Exception:
  35             logging.info("{} process is waiting for commondb to come up...".format(process_name))
  36             time.sleep(commondb_wait_time)
  37
  38
  39 def wait_till_kafka_is_ready(config, process_name="osm-mon", kafka_wait_time=5):
  40
  41     logging.debug("wait_till_kafka_is_ready")
  42
  43     while(True):
  44         kafka_ready = False
  45         try:
  46             with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
  47                 # Verify is kafka port is up
  48                 if (
  49                     s.connect_ex(
  50                         (
  51                             config.conf.get("message", {}).get("host", "kafka"),
  52                             int(config.conf["message"].get("port")),
  53                         )
  54                     )
  55                     == 0
  56                 ):
  57                     # Get the list of topics. If kafka is not ready exception will be thrown.
  58                     consumer = kafka.KafkaConsumer(group_id=config.conf["message"].get("group_id"),
  59                                                    bootstrap_servers=[config.conf.get("message", {}).get("host",
  60                                                                       "kafka") + ":" + config.conf["message"]
  61                                                                       .get("port")])
  62                     topics = consumer.topics()
  63                     logging.debug("Number of topics found: %s", len(topics))
  64                     kafka_ready = True
  65         except Exception as e:
  66             logging.info("Error when trying to get kafka status.")
  67             logging.debug("Exception when trying to get kafka status: %s", str(e))
  68         finally:
  69             if kafka_ready:
  70                 break
  71             else:
  72                 logging.info("{} process is waiting for kafka to come up...".format(process_name))
  73                 time.sleep(kafka_wait_time)
  74
  75
  76 def wait_till_core_services_are_ready(config, process_name="osm-mon", commondb_wait_time=5, kafka_wait_time=5):
  77
  78     logging.debug("wait_till_core_services_are_ready")
  79
  80     if not config:
  81         logging.info("Config information is not available")
  82         return False
  83
  84     # Check if common-db is ready
  85     wait_till_commondb_is_ready(config, process_name, commondb_wait_time)
  86
  87     # Check if kafka is ready
  88     wait_till_kafka_is_ready(config, process_name, kafka_wait_time)
  89
  90     return True