X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=plugins%2FOpenStack%2FAodh%2Falarming.py;h=b44d624286590ed5309d6dd0d04532fe5d05c5a2;hb=5b0cadb9854d7d699d985e96c51269c721afdb3d;hp=3ae31ba974068e5775bbc620eaa8aac4afd7e8d6;hpb=f358b4fdd006b427e5b653d467c29ae37a47406e;p=osm%2FMON.git diff --git a/plugins/OpenStack/Aodh/alarming.py b/plugins/OpenStack/Aodh/alarming.py index 3ae31ba..b44d624 100644 --- a/plugins/OpenStack/Aodh/alarming.py +++ b/plugins/OpenStack/Aodh/alarming.py @@ -1,7 +1,13 @@ -"""Send alarm info from Aodh to SO via MON""" +"""Send alarm info from Aodh to SO via MON.""" import json -from plugins.Openstack.Aodh.aodh_common import Aodh_Common +import logging as log + +from collections import OrderedDict + +from kafka import KafkaConsumer + +from plugins.OpenStack.Aodh.aodh_common import Aodh_Common class Alarming(object): @@ -11,52 +17,123 @@ class Alarming(object): """Create the aodh_receiver instance.""" self._aodh_common = Aodh_Common() + # Initialize a generic consumer object to consume message from the SO + server = {'server': 'localhost:9092', 'topic': 'alarms'} + self._consumer = KafkaConsumer(server['topic'], + group_id='my-group', + bootstrap_servers=server['server']) + + # TODO(mcgoughh): Initialize a producer to send messages bask to the SO + def alarming(self): - """Receive payload from Aodh.""" + """Consume info from the message bus to manage alarms.""" + # Generate authentication credentials to access keystone; + # auth_token, endpoint auth_token = self._aodh_common._authenticate() endpoint = self._aodh_common.get_endpoint() - alarm_list = self._get_alarm_list(endpoint, auth_token) - # Confirm communication with Aodh by listing alarms - print("Alarm List ", alarm_list.text) - - alarm_id = self._create_alarm(endpoint, auth_token) - print(alarm_id) + # Check the alarming functionlity that needs to be performed + for message in self._consumer: + if message.topic == "alarms": + log.info("Alarm action required: %s" % (message.topic)) + + if message.key == "configure_alarm": + # Configure/Update an alarm + alarm_details = json.loads(message.value) + alarm_id = self.configure_alarm(endpoint, + auth_token, alarm_details) + log.info("New alarm created with alarmID: %s", alarm_id) + + # TODO(mcgoughh): will send an acknowledge message back on + # the bus via the producer + + else: + # TODO(mcoughh): Key alternatives are "notify_alarm" and + # "acknowledge_alarm" will be accomodated later + log.debug("Unknown key, no action will be performed") + else: + log.info("Message topic not relevant to this plugin: %s", + message.topic) -# alarm_info = self._get_alarm_info(endpoint, -# auth_token, "372af0e2-5c36-4e4d-8ce9-ca92d97d07d0") -# print("Alarm info", alarm_info.text) return - def _get_alarm_list(self, endpoint, auth_token): + def alarm_check(self, endpoint, auth_token, alarm_name): """Get a list of alarms that exist in Aodh.""" url = "{}/v2/alarms/".format(endpoint) - alarm_list = self._aodh_common._perform_request(url, auth_token, - req_type="get") - return alarm_list + # TODO(mcgoughh): will query on resource_id once it has been + # implemented need to create the query field when creating + # the alarm + query = OrderedDict([("q.field", 'name'), ("q.op", "eq"), + ("q.value", str(alarm_name))]) - def _get_alarm_info(self, endpoint, auth_token, alarmID): - """Get information about a specific alarm from Aodh.""" - url = "{}/v2/alarms/%s".format(endpoint) % (alarmID) + result = self._aodh_common._perform_request( + url, auth_token, req_type="get", params=query) - alarm_details = self._aodh_common._perform_request(url, auth_token, - req_type="get") - return alarm_details + try: + alarm_id = json.loads(result.text)[0]['alarm_id'] + log.info("An existing alarm was found: %s", alarm_id) + return alarm_id + except Exception: + log.debug("Alarm doesn't exist, needs to be created.") + return None - def _create_alarm(self, endpoint, auth_token): + def configure_alarm(self, endpoint, auth_token, values): """Get a list of alarms that exist in Aodh.""" - url = "{}/v2/alarms/".format(endpoint) - - rule = {'event_type': "threshold",} - payload = json.dumps({'state': 'alarm', - 'name': 'my_alarm', - 'severity': 'moderate', - 'type': 'event', - 'event_rule': rule,}) - - new_alarm = self._aodh_common._perform_request(url, auth_token, - req_type="post", payload=payload) - alarm_id =json.loads(new_alarm.text)['alarm_id'] - return alarm_id + alarm_id = None + + # TODO(mcgoughh): error check the values sent in the messag + alarm_name = values['name'] + + # Check that this alarm doesn't exist already + alarm_id = self.alarm_check(endpoint, auth_token, alarm_name) + + if alarm_id is None: + url = "{}/v2/alarms/".format(endpoint) + severity = values['severity'] + + # Create a new threshold alarm with a resourceID + # specified as a query + rule = {'threshold': values['threshold'], + 'comparison_operator': 'gt', + 'metric': values['metric'], + 'resource_id': values['resource_id'], + 'resource_type': 'generic', + 'aggregation_method': 'last', } + payload = json.dumps({'state': 'alarm', + 'name': alarm_name, + 'severity': self.get_severity(severity), + 'type': 'gnocchi_resources_threshold', + 'gnocchi_resources_threshold_rule': rule, }) + + # Request performed to create alarm + new_alarm = self._aodh_common._perform_request( + url, auth_token, req_type="post", payload=payload) + + return json.loads(new_alarm.text)['alarm_id'] + else: + return alarm_id + + def delete_alarm(self, endpoint, auth_token, alarmID): + """Delete alarm function.""" + url = "{}/v2/alarms/%s".format(endpoint) % (alarmID) + self._aodh_common._perform_request(url, auth_token, req_type="delete") + return None + + def get_severity(self, alarm_severity): + """Get a normalized severity for Aodh.""" + # This logic can be changed, the other alternative was to have + # MINOR and MAJOR = "moderate" instead. + if alarm_severity == "WARNIING": + aodh_severity = "low" + elif alarm_severity == "MINOR": + aodh_severity = "moderate" + elif (alarm_severity == "MAJOR" or alarm_severity == "CRITICAL"): + aodh_severity = "critical" + else: + aodh_severity = None + log.warn("Invalid alarm severity configuration") + + log.info("Severity has been normalized for Aodh to: %s", aodh_severity) + return aodh_severity