osm_mon/plugins/OpenStack/Aodh/notifier.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: helena.mcgough@intel.com or adrian.hoban@intel.com
  21 ##
  22 # __author__ = Helena McGough
  23 #
  24 """A Webserver to send alarm notifications from Aodh to the SO."""
  25 import json
  26 import logging
  27 import os
  28 import re
  29 import sys
  30 import time
  31
  32 from six.moves.BaseHTTPServer import BaseHTTPRequestHandler
  33 from six.moves.BaseHTTPServer import HTTPServer
  34
  35 # Initialise a logger for alarm notifier
  36 from osm_mon.core.message_bus.producer import Producer
  37 from osm_mon.core.settings import Config
  38
  39 cfg = Config.instance()
  40
  41 logging.basicConfig(stream=sys.stdout,
  42                     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
  43                     datefmt='%m/%d/%Y %I:%M:%S %p',
  44                     level=logging.getLevelName(cfg.OSMMON_LOG_LEVEL))
  45 log = logging.getLogger(__name__)
  46
  47 kafka_logger = logging.getLogger('kafka')
  48 kafka_logger.setLevel(logging.getLevelName(cfg.OSMMON_KAFKA_LOG_LEVEL))
  49 kafka_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  50 kafka_handler = logging.StreamHandler(sys.stdout)
  51 kafka_handler.setFormatter(kafka_formatter)
  52 kafka_logger.addHandler(kafka_handler)
  53
  54 sys.path.append(os.path.abspath(os.path.join(os.path.realpath(__file__), '..', '..', '..', '..', '..')))
  55
  56 from osm_mon.core.database import DatabaseManager
  57
  58 from osm_mon.plugins.OpenStack.response import OpenStackResponseBuilder
  59
  60
  61 class NotifierHandler(BaseHTTPRequestHandler):
  62     """Handler class for alarm_actions triggered by OSM alarms."""
  63
  64     def _set_headers(self):
  65         """Set the headers for a request."""
  66         self.send_response(200)
  67         self.send_header('Content-type', 'text/html')
  68         self.end_headers()
  69
  70     def do_GET(self):
  71         """Get request functionality."""
  72         self._set_headers()
  73
  74     def do_POST(self):
  75         """POST request function."""
  76         # Gets header and data from the post request and records info
  77         self._set_headers()
  78         # Gets the size of data
  79         content_length = int(self.headers['Content-Length'])
  80         post_data = self.rfile.read(content_length)
  81         # Python 2/3 string compatibility
  82         try:
  83             post_data = post_data.decode()
  84         except AttributeError:
  85             pass
  86         log.info("This alarm was triggered: %s", post_data)
  87
  88         # Send alarm notification to message bus
  89         try:
  90             self.notify_alarm(json.loads(post_data))
  91         except Exception:
  92             log.exception("Error notifying alarm")
  93
  94     def notify_alarm(self, values):
  95         """Sends alarm notification message to bus."""
  96
  97         # Initialise configuration and authentication for response message
  98         response = OpenStackResponseBuilder()
  99
 100         database_manager = DatabaseManager()
 101
 102         alarm_id = values['alarm_id']
 103         alarm = database_manager.get_alarm(alarm_id, 'openstack')
 104         # Process an alarm notification if resource_id is valid
 105         # Get date and time for response message
 106         a_date = time.strftime("%d-%m-%Y") + " " + time.strftime("%X")
 107         # Generate and send response
 108         resp_message = response.generate_response(
 109             'notify_alarm',
 110             alarm_id=alarm_id,
 111             vdu_name=alarm.vdu_name,
 112             vnf_member_index=alarm.vnf_member_index,
 113             ns_id=alarm.ns_id,
 114             metric_name=alarm.metric_name,
 115             operation=alarm.operation,
 116             threshold_value=alarm.threshold,
 117             sev=values['severity'],
 118             date=a_date,
 119             state=values['current'])
 120         self._publish_response('notify_alarm', json.dumps(resp_message))
 121         log.info("Sent alarm notification: %s", resp_message)
 122
 123     def _publish_response(self, key: str, msg: str):
 124         producer = Producer()
 125         producer.send(topic='alarm_response', key=key, value=msg)
 126         producer.flush()
 127
 128
 129 def run(server_class=HTTPServer, handler_class=NotifierHandler, port=8662):
 130     """Run the webserver application to retrieve alarm notifications."""
 131     try:
 132         server_address = ('', port)
 133         httpd = server_class(server_address, handler_class)
 134         print('Starting alarm notifier...')
 135         log.info("Starting alarm notifier server on port: %s", port)
 136         httpd.serve_forever()
 137     except Exception as exc:
 138         log.warning("Failed to start webserver, %s", exc)
 139
 140
 141 if __name__ == "__main__":
 142     cfg = Config.instance()
 143     p = re.compile(':(\d+)', re.IGNORECASE)
 144     m = p.search(cfg.OS_NOTIFIER_URI)
 145     if m:
 146         port = m.group(1)
 147         run(port=int(port))
 148     else:
 149         run()