osm_mon/core/message_bus/common_consumer.py

   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 os
  25 import sys
  26 import yaml
  27
  28 logging.basicConfig(stream=sys.stdout,
  29                     format='%(asctime)s %(message)s',
  30                     datefmt='%m/%d/%Y %I:%M:%S %p',
  31                     level=logging.INFO)
  32 log = logging.getLogger(__name__)
  33
  34 sys.path.append(os.path.abspath(os.path.join(os.path.realpath(__file__), '..', '..', '..', '..')))
  35
  36 from kafka import KafkaConsumer
  37
  38 from osm_mon.plugins.OpenStack.Aodh import alarming
  39 from osm_mon.plugins.OpenStack.Gnocchi import metrics
  40
  41 from osm_mon.plugins.CloudWatch.plugin_alarm import plugin_alarms
  42 from osm_mon.plugins.CloudWatch.plugin_metric import plugin_metrics
  43 from osm_mon.plugins.CloudWatch.connection import Connection
  44 from osm_mon.plugins.CloudWatch.access_credentials import AccessCredentials
  45
  46 from osm_mon.plugins.vRealiseOps import plugin_receiver
  47
  48 from osm_mon.core.auth import AuthManager
  49 from osm_mon.core.database import DatabaseManager
  50
  51 # Initialize servers
  52 if "BROKER_URI" in os.environ:
  53     server = {'server': os.getenv("BROKER_URI")}
  54 else:
  55     server = {'server': 'localhost:9092'}
  56
  57 # Initialize consumers for alarms and metrics
  58 common_consumer = KafkaConsumer(bootstrap_servers=server['server'],
  59                                 key_deserializer=bytes.decode,
  60                                 value_deserializer=bytes.decode,
  61                                 group_id="mon-consumer")
  62
  63 auth_manager = AuthManager()
  64 database_manager = DatabaseManager()
  65 database_manager.create_tables()
  66
  67 # Create OpenStack alarming and metric instances
  68 openstack_metrics = metrics.Metrics()
  69 openstack_alarms = alarming.Alarming()
  70
  71 # Create CloudWatch alarm and metric instances
  72 cloudwatch_alarms = plugin_alarms()
  73 cloudwatch_metrics = plugin_metrics()
  74 aws_connection = Connection()
  75 aws_access_credentials = AccessCredentials()
  76
  77 # Create vROps plugin_receiver class instance
  78 vrops_rcvr = plugin_receiver.PluginReceiver()
  79
  80
  81 def get_vim_type(msg):
  82     """Get the vim type that is required by the message."""
  83     try:
  84         vim_uuid = json.loads(msg.value)["vim_uuid"].lower()
  85         credentials = database_manager.get_credentials(vim_uuid)
  86         return credentials.type
  87     except Exception as exc:
  88         log.warn("vim_type is not configured correctly; %s", exc)
  89     return None
  90
  91
  92 # Define subscribe the consumer for the plugins
  93 topics = ['metric_request', 'alarm_request', 'access_credentials', 'vim_account']
  94 # TODO: Remove access_credentials
  95 common_consumer.subscribe(topics)
  96
  97 log.info("Listening for alarm_request and metric_request messages")
  98 for message in common_consumer:
  99     log.info("Message arrived: %s", message)
 100     try:
 101         try:
 102             values = json.loads(message.value)
 103         except:
 104             values = yaml.safe_load(message.value)
 105         # Check the message topic
 106         if message.topic == "metric_request":
 107             # Check the vim desired by the message
 108             vim_type = get_vim_type(message)
 109
 110             if vim_type == "openstack":
 111                 log.info("This message is for the OpenStack plugin.")
 112                 openstack_metrics.metric_calls(message)
 113             elif vim_type == "aws":
 114                 log.info("This message is for the CloudWatch plugin.")
 115                 aws_conn = aws_connection.setEnvironment()
 116                 cloudwatch_metrics.metric_calls(message, aws_conn)
 117
 118             elif vim_type == "vmware":
 119                 log.info("This metric_request message is for the vROPs plugin.")
 120                 vrops_rcvr.consume(message)
 121
 122             else:
 123                 log.debug("vim_type is misconfigured or unsupported; %s",
 124                           vim_type)
 125
 126         elif message.topic == "alarm_request":
 127             # Check the vim desired by the message
 128             vim_type = get_vim_type(message)
 129             if vim_type == "openstack":
 130                 log.info("This message is for the OpenStack plugin.")
 131                 openstack_alarms.alarming(message)
 132
 133             elif vim_type == "aws":
 134                 log.info("This message is for the CloudWatch plugin.")
 135                 aws_conn = aws_connection.setEnvironment()
 136                 cloudwatch_alarms.alarm_calls(message, aws_conn)
 137
 138             elif vim_type == "vmware":
 139                 log.info("This alarm_request message is for the vROPs plugin.")
 140                 vrops_rcvr.consume(message)
 141
 142             else:
 143                 log.debug("vim_type is misconfigured or unsupported; %s",
 144                           vim_type)
 145
 146         elif message.topic == "vim_account":
 147             if message.key == "create" or message.key == "edit":
 148                 auth_manager.store_auth_credentials(values)
 149             if message.key == "delete":
 150                 auth_manager.delete_auth_credentials(values)
 151
 152         # TODO: Remove in the near future when all plugins support vim_uuid. Modify tests accordingly.
 153         elif message.topic == "access_credentials":
 154             # Check the vim desired by the message
 155             vim_type = get_vim_type(message)
 156
 157             if vim_type == "aws":
 158                 log.info("This message is for the CloudWatch plugin.")
 159                 aws_access_credentials.access_credential_calls(message)
 160
 161             elif vim_type == "vmware":
 162                 log.info("This access_credentials message is for the vROPs plugin.")
 163                 vrops_rcvr.consume(message)
 164
 165             else:
 166                 log.debug("vim_type is misconfigured or unsupported; %s",
 167                           vim_type)
 168
 169         else:
 170             log.info("This topic is not relevant to any of the MON plugins.")
 171
 172
 173     except Exception as exc:
 174         log.exception("Exception: %s")