osm_mon/core/message_bus/common_consumer

   1 # Copyright 2017 Intel Research and Development Ireland Limited
   2 # *************************************************************
   3 # This file is part of OSM Monitoring module
   4 # All Rights Reserved to Intel Corporation
   5 #
   6 # Licensed under the Apache License, Version 2.0 (the "License"); you
   7 # may not use this file except in compliance with the License. You may
   8 # obtain a copy of the License at
   9 #
  10 #         http://www.apache.org/licenses/LICENSE-2.0
  11 #
  12 # Unless required by applicable law or agreed to in writing, software
  13 # distributed under the License is distributed on an "AS IS" BASIS,
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  15 # implied. See the License for the specific language governing
  16 # permissions and limitations under the License.
  17 #
  18 # For those usages not covered by the Apache License, Version 2.0 please
  19 # contact: helena.mcgough@intel.com or adrian.hoban@intel.com
  20 """A common KafkaConsumer for all MON plugins."""
  21
  22 import json
  23 import logging
  24 import sys
  25 import os
  26
  27 sys.path.append("/root/MON")
  28
  29 logging.basicConfig(filename='MON_plugins.log',
  30                     format='%(asctime)s %(message)s',
  31                     datefmt='%m/%d/%Y %I:%M:%S %p', filemode='a',
  32                     level=logging.INFO)
  33 log = logging.getLogger(__name__)
  34
  35 from kafka import KafkaConsumer
  36 from kafka.errors import KafkaError
  37
  38 from osm_mon.plugins.OpenStack.Aodh import alarming
  39 from osm_mon.plugins.OpenStack.common import Common
  40 from osm_mon.plugins.OpenStack.Gnocchi import metrics
  41
  42 from osm_mon.plugins.CloudWatch.plugin_alarm import plugin_alarms
  43 from osm_mon.plugins.CloudWatch.plugin_metric import plugin_metrics
  44 from osm_mon.plugins.CloudWatch.connection import Connection
  45 from osm_mon.plugins.CloudWatch.access_credentials import AccessCredentials
  46
  47 from osm_mon.plugins.vRealiseOps import plugin_receiver
  48
  49 # Initialize servers
  50 server = {'server': 'localhost:9092'}
  51
  52 # Initialize consumers for alarms and metrics
  53 common_consumer = KafkaConsumer(bootstrap_servers=server['server'])
  54
  55 # Create OpenStack alarming and metric instances
  56 auth_token = None
  57 openstack_auth = Common()
  58 openstack_metrics = metrics.Metrics()
  59 openstack_alarms = alarming.Alarming()
  60
  61 # Create CloudWatch alarm and metric instances
  62 cloudwatch_alarms = plugin_alarms()
  63 cloudwatch_metrics = plugin_metrics()
  64 aws_connection = Connection()
  65 aws_access_credentials = AccessCredentials()
  66
  67 #Create vROps plugin_receiver class instance
  68 vrops_rcvr = plugin_receiver.PluginReceiver()
  69
  70 def get_vim_type(message):
  71     """Get the vim type that is required by the message."""
  72     try:
  73         return json.loads(message.value)["vim_type"].lower()
  74     except Exception as exc:
  75         log.warn("vim_type is not configured correctly; %s", exc)
  76     return None
  77
  78 # Define subscribe the consumer for the plugins
  79 topics = ['metric_request', 'alarm_request', 'access_credentials']
  80 common_consumer.subscribe(topics)
  81
  82 try:
  83     log.info("Listening for alarm_request and metric_request messages")
  84     for message in common_consumer:
  85         # Check the message topic
  86         if message.topic == "metric_request":
  87             # Check the vim desired by the message
  88             vim_type = get_vim_type(message)
  89
  90             if vim_type == "openstack":
  91                 log.info("This message is for the OpenStack plugin.")
  92                 openstack_metrics.metric_calls(
  93                     message, openstack_auth, auth_token)
  94
  95             elif vim_type == "aws":
  96                 log.info("This message is for the CloudWatch plugin.")
  97                 aws_conn = aws_connection.setEnvironment()
  98                 cloudwatch_metrics.metric_calls(message,aws_conn)
  99
 100             elif vim_type == "vmware":
 101                 log.info("This metric_request message is for the vROPs plugin.")
 102                 vrops_rcvr.consume(message)
 103
 104             else:
 105                 log.debug("vim_type is misconfigured or unsupported; %s",
 106                           vim_type)
 107
 108         elif message.topic == "alarm_request":
 109             # Check the vim desired by the message
 110             vim_type = get_vim_type(message)
 111             if vim_type == "openstack":
 112                 log.info("This message is for the OpenStack plugin.")
 113                 openstack_alarms.alarming(message, openstack_auth, auth_token)
 114
 115             elif vim_type == "aws":
 116                 log.info("This message is for the CloudWatch plugin.")
 117                 aws_conn = aws_connection.setEnvironment()
 118                 cloudwatch_alarms.alarm_calls(message, aws_conn)
 119
 120             elif vim_type == "vmware":
 121                 log.info("This alarm_request message is for the vROPs plugin.")
 122                 vrops_rcvr.consume(message)
 123
 124             else:
 125                 log.debug("vim_type is misconfigured or unsupported; %s",
 126                           vim_type)
 127
 128         elif message.topic == "access_credentials":
 129             # Check the vim desired by the message
 130             vim_type = get_vim_type(message)
 131             if vim_type == "openstack":
 132                 log.info("This message is for the OpenStack plugin.")
 133                 auth_token = openstack_auth._authenticate(message=message)
 134
 135             elif vim_type == "aws":
 136                 log.info("This message is for the CloudWatch plugin.")
 137                 aws_access_credentials.access_credential_calls(message)
 138
 139             elif vim_type == "vmware":
 140                 log.info("This access_credentials message is for the vROPs plugin.")
 141                 vrops_rcvr.consume(message)
 142
 143             else:
 144                 log.debug("vim_type is misconfigured or unsupported; %s",
 145                           vim_type)
 146
 147         else:
 148             log.info("This topic is not relevant to any of the MON plugins.")
 149
 150
 151 except KafkaError as exc:
 152     log.warn("Exception: %s", exc)