core/message_bus/consumer.py

   1 # Copyright 2017 Intel Research and Development Ireland Limited
   2 # *************************************************************
   3
   4 # This file is part of OSM Monitoring module
   5 # All Rights Reserved to Intel Corporation
   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 # For those usages not covered by the Apache License, Version 2.0 please
  20 # contact: prithiv.mohan@intel.com or adrian.hoban@intel.com
  21 ##
  22
  23 '''
  24 This is a kafka consumer app that reads the messages from the message bus for
  25 alarms and metrics responses.
  26
  27 '''
  28
  29 __author__ = "Prithiv Mohan"
  30 __date__   = "06/Sep/2017"
  31
  32
  33 from kafka import KafkaConsumer
  34 from kafka.errors import KafkaError
  35 import json
  36 import logging
  37 import logging.config
  38 import os
  39
  40 def logging_handler(filename, mode='a+', encoding=None):
  41     if not os.path.exists(filename):
  42         open(filename, 'a').close()
  43     return logging.FileHandler(filename, mode)
  44
  45 log_config = {
  46     'version': 1,
  47     'formatters': {
  48         'default': {
  49             'format': '%(asctime)s %(levelname)s %(name)s %(message)s'
  50         },
  51     },
  52     'handlers': {
  53         'file':{
  54             '()': logging_handler,
  55             'level':'DEBUG',
  56             'formatter': 'default',
  57             'filename': '/var/log/osm_mon.log',
  58             'mode': 'a+',
  59             'encoding': 'utf-8',
  60         },
  61     },
  62     'kafka': {
  63         'handlers': ['file'],
  64         'level': 'DEBUG',
  65     },
  66     'root': {
  67         'handlers': ['file'],
  68         'level': 'DEBUG',
  69     },
  70 }
  71
  72
  73 logging.config.dictConfig(log_config)
  74 logger = logging.getLogger('kafka')
  75
  76 if "BROKER_URI" in os.environ:
  77     broker = os.getenv("BROKER_URI")
  78 else:
  79     broker = "localhost:9092"
  80
  81 alarm_consumer = KafkaConsumer('alarm_response', 'osm_mon', bootstrap_servers = broker)
  82 metric_consumer = KafkaConsumer('metric_response', 'osm_mon', bootstrap_servers = broker)
  83 try:
  84     for message in alarm_consumer:
  85         logger.debug(message)
  86     for message in metric_consumer:
  87         logger.debug(message)
  88 except KafkaError:
  89     log.exception()
  90
  91 alarm_consumer.subscribe('alarm_response')
  92 metric_consumer.subscribe('metric_response')