plugins/OpenStack/Aodh/alarming.py

   1 """Send alarm info from Aodh to SO via MON."""
   2
   3 import json
   4 import logging as log
   5
   6 from collections import OrderedDict
   7
   8 from kafka import KafkaConsumer
   9
  10 from plugins.OpenStack.Aodh.aodh_common import Aodh_Common
  11
  12
  13 class Alarming(object):
  14     """Receives alarm info from Aodh."""
  15
  16     def __init__(self):
  17         """Create the aodh_receiver instance."""
  18         self._aodh_common = Aodh_Common()
  19
  20         # Initialize a generic consumer object to consume message from the SO
  21         server = {'server': 'localhost:9092', 'topic': 'alarms'}
  22         self._consumer = KafkaConsumer(server['topic'],
  23                                        group_id='my-group',
  24                                        bootstrap_servers=server['server'])
  25
  26         # TODO(mcgoughh): Initialize a producer to send messages bask to the SO
  27
  28     def alarming(self):
  29         """Consume info from the message bus to manage alarms."""
  30         # Generate authentication credentials to access keystone;
  31         # auth_token, endpoint
  32         auth_token = self._aodh_common._authenticate()
  33         endpoint = self._aodh_common.get_endpoint()
  34
  35         # Check the alarming functionlity that needs to be performed
  36         for message in self._consumer:
  37             if message.topic == "alarms":
  38                 log.info("Alarm action required: %s" % (message.topic))
  39
  40                 if message.key == "configure_alarm":
  41                     # Configure/Update an alarm
  42                     alarm_details = json.loads(message.value)
  43                     alarm_id = self.configure_alarm(endpoint,
  44                                                     auth_token, alarm_details)
  45                     log.info("New alarm created with alarmID: %s", alarm_id)
  46
  47                     # TODO(mcgoughh): will send an acknowledge message back on
  48                     # the bus via the producer
  49
  50                 else:
  51                     # TODO(mcoughh): Key alternatives are "notify_alarm" and
  52                     # "acknowledge_alarm" will be accomodated later
  53                     log.debug("Unknown key, no action will be performed")
  54             else:
  55                 log.info("Message topic not relevant to this plugin: %s",
  56                          message.topic)
  57
  58         return
  59
  60     def alarm_check(self, endpoint, auth_token, alarm_name):
  61         """Get a list of alarms that exist in Aodh."""
  62         url = "{}/v2/alarms/".format(endpoint)
  63
  64         # TODO(mcgoughh): will query on resource_id once it has been
  65         # implemented need to create the query field when creating
  66         # the alarm
  67         query = OrderedDict([("q.field", 'name'), ("q.op", "eq"),
  68                             ("q.value", str(alarm_name))])
  69
  70         result = self._aodh_common._perform_request(
  71             url, auth_token, req_type="get", params=query)
  72
  73         try:
  74             alarm_id = json.loads(result.text)[0]['alarm_id']
  75             log.info("An existing alarm was found: %s", alarm_id)
  76             return alarm_id
  77         except Exception:
  78             log.debug("Alarm doesn't exist, needs to be created.")
  79             return None
  80
  81     def configure_alarm(self, endpoint, auth_token, values):
  82         """Get a list of alarms that exist in Aodh."""
  83         alarm_id = None
  84
  85         # TODO(mcgoughh): error check the values sent in the messag
  86         alarm_name = values['name']
  87
  88         # Check that this alarm doesn't exist already
  89         alarm_id = self.alarm_check(endpoint, auth_token, alarm_name)
  90
  91         if alarm_id is None:
  92             url = "{}/v2/alarms/".format(endpoint)
  93             severity = values['severity']
  94
  95             # Create a new threshold alarm with a resourceID
  96             # specified as a query
  97             rule = {'threshold': values['threshold'],
  98                     'comparison_operator': 'gt',
  99                     'metric': values['metric'],
 100                     'resource_id': values['resource_id'],
 101                     'resource_type': 'generic',
 102                     'aggregation_method': 'last', }
 103             payload = json.dumps({'state': 'alarm',
 104                                   'name': alarm_name,
 105                                   'severity': self.get_severity(severity),
 106                                   'type': 'gnocchi_resources_threshold',
 107                                   'gnocchi_resources_threshold_rule': rule, })
 108
 109             # Request performed to create alarm
 110             new_alarm = self._aodh_common._perform_request(
 111                 url, auth_token, req_type="post", payload=payload)
 112
 113             return json.loads(new_alarm.text)['alarm_id']
 114         else:
 115             return alarm_id
 116
 117     def delete_alarm(self, endpoint, auth_token, alarmID):
 118         """Delete alarm function."""
 119         url = "{}/v2/alarms/%s".format(endpoint) % (alarmID)
 120
 121         self._aodh_common._perform_request(url, auth_token, req_type="delete")
 122         return None
 123
 124     def get_severity(self, alarm_severity):
 125         """Get a normalized severity for Aodh."""
 126         # This logic can be changed, the other alternative was to have
 127         # MINOR and MAJOR = "moderate" instead.
 128         if alarm_severity == "WARNIING":
 129             aodh_severity = "low"
 130         elif alarm_severity == "MINOR":
 131             aodh_severity = "moderate"
 132         elif (alarm_severity == "MAJOR" or alarm_severity == "CRITICAL"):
 133             aodh_severity = "critical"
 134         else:
 135             aodh_severity = None
 136             log.warn("Invalid alarm severity configuration")
 137
 138         log.info("Severity has been normalized for Aodh to: %s", aodh_severity)
 139         return aodh_severity