X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=plugins%2FOpenStack%2FAodh%2Falarming.py;h=d409d71cb4b9103567836fbe412ccfcf49f87eb8;hb=95b92b3dacfd93fa1649a5f87dafd2fa6553a086;hp=0f4a2dac78816b190e08980dc54a2fca49d94b8c;hpb=c27631994b5cc72c2ce2ba0107b9178af6f73662;p=osm%2FMON.git diff --git a/plugins/OpenStack/Aodh/alarming.py b/plugins/OpenStack/Aodh/alarming.py index 0f4a2da..d409d71 100644 --- a/plugins/OpenStack/Aodh/alarming.py +++ b/plugins/OpenStack/Aodh/alarming.py @@ -1,32 +1,71 @@ -"""Send alarm info from Aodh to SO via MON.""" +# Copyright 2017 Intel Research and Development Ireland Limited +# ************************************************************* + +# This file is part of OSM Monitoring module +# All Rights Reserved to Intel Corporation + +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at + +# http://www.apache.org/licenses/LICENSE-2.0 + +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +# For those usages not covered by the Apache License, Version 2.0 please +# contact: helena.mcgough@intel.com or adrian.hoban@intel.com +## +"""Carry out alarming requests via Aodh API.""" import json -import logging as log +import logging +log = logging.getLogger(__name__) -from collections import OrderedDict +from core.message_bus.producer import KafkaProducer from kafka import KafkaConsumer from plugins.OpenStack.common import Common +from plugins.OpenStack.response import OpenStack_Response +__author__ = "Helena McGough" + +ALARM_NAMES = { + "average_memory_usage_above_threshold": "average_memory_utilization", + "disk_read_ops": "disk_read_ops", + "disk_write_ops": "disk_write_ops", + "disk_read_bytes": "disk_read_bytes", + "disk_write_bytes": "disk_write_bytes", + "net_packets_dropped": "packets_dropped", + "packets_in_above_threshold": "packets_received", + "packets_out_above_threshold": "packets_sent", + "cpu_utilization_above_threshold": "cpu_utilization"} SEVERITIES = { - "WARNING": "low", - "MINOR": "low", - "MAJOR": "moderate", - "CRITICAL": "critical", - "INDETERMINATE": "critical"} + "warning": "low", + "minor": "low", + "major": "moderate", + "critical": "critical", + "indeterminate": "critical"} + +STATISTICS = { + "average": "avg", + "minimum": "min", + "maximum": "max", + "count": "count", + "sum": "sum"} class Alarming(object): - """Receives alarm info from Aodh.""" + """Carries out alarming requests and responses via Aodh API.""" def __init__(self): - """Create the aodh_receiver instance.""" + """Create the OpenStack alarming instance.""" self._common = Common() - self.auth_token = None - self.endpoint = None - self.resp_status = None # TODO(mcgoughh): Remove hardcoded kafkaconsumer # Initialize a generic consumer object to consume message from the SO @@ -35,7 +74,11 @@ class Alarming(object): group_id='osm_mon', bootstrap_servers=server['server']) - # TODO(mcgoughh): Initialize a producer to send messages bask to the SO + # Use the Response class to generate valid json response messages + self._response = OpenStack_Response() + + # Initializer a producer to send responses back to SO + self._producer = KafkaProducer("alarm_response") def alarming(self): """Consume info from the message bus to manage alarms.""" @@ -48,94 +91,104 @@ class Alarming(object): if vim_type == "openstack": log.info("Alarm action required: %s" % (message.topic)) + # Generate and auth_token and endpoint for request + auth_token, endpoint = self.authenticate() + if message.key == "create_alarm_request": # Configure/Update an alarm alarm_details = values['alarm_create_request'] - # Generate an auth_token and endpoint - auth_token = self._common._authenticate( - tenant_id=alarm_details['tenant_uuid']) - endpoint = self._common.get_endpoint("alarming") - - alarm_id = self.configure_alarm( + alarm_id, alarm_status = self.configure_alarm( endpoint, auth_token, alarm_details) - # TODO(mcgoughh): will send an acknowledge message back on - # the bus via the producer - if alarm_id is not None: - self.resp_status = True - log.debug("A valid alarm was found/created: %s", - self.resp_status) - else: - self.resp_status = False - log.debug("Failed to create desired alarm: %s", - self.resp_status) + # Generate a valid response message, send via producer + try: + if alarm_status is True: + log.info("Alarm successfully created") + + resp_message = self._response.generate_response( + 'create_alarm_response', status=alarm_status, + alarm_id=alarm_id, + cor_id=alarm_details['correlation_id']) + log.info("Response Message: %s", resp_message) + self._producer.create_alarm_response( + 'create_alarm_resonse', resp_message, + 'alarm_response') + except Exception as exc: + log.warn("Response creation failed: %s", exc) elif message.key == "list_alarm_request": - auth_token = self._common._authenticate() - endpoint = self._common.get_endpoint("alarming") - - # List all of the alarms - alarm_list = self.list_alarms(endpoint, auth_token) - - # TODO(mcgoughh): send a repsonse back to SO - if alarm_list is not None: - self.resp_status = True - log.info("A list of alarms was generated: %s", - alarm_list) - else: - self.resp_status = False - log.warn("Failed to generae an alarm list") + # Check for a specifed: alarm_name, resource_uuid, severity + # and generate the appropriate list + list_details = values['alarm_list_request'] + + alarm_list = self.list_alarms( + endpoint, auth_token, list_details) + + try: + # Generate and send a list response back + resp_message = self._response.generate_response( + 'list_alarm_response', alarm_list=alarm_list, + cor_id=list_details['correlation_id']) + log.info("Response Message: %s", resp_message) + self._producer.list_alarm_response( + 'list_alarm_response', resp_message, + 'alarm_response') + except Exception as exc: + log.warn("Failed to send a valid response back.") elif message.key == "delete_alarm_request": - # Delete the specified alarm - auth_token = self._common._authenticate() - endpoint = self._common.get_endpoint("alarming") - - alarm_id = values['alarm_delete_request']['alarm_uuid'] + request_details = values['alarm_delete_request'] + alarm_id = request_details['alarm_uuid'] - response = self.delete_alarm( + resp_status = self.delete_alarm( endpoint, auth_token, alarm_id) - # TODO(mcgoughh): send a response back on the bus - if response is True: - log.info("Requested alarm has been deleted: %s", - alarm_id) - else: - log.warn("Failed to delete requested alarm.") + # Generate and send a response message + try: + resp_message = self._response.generate_response( + 'delete_alarm_response', alarm_id=alarm_id, + status=resp_status, + cor_id=request_details['correlation_id']) + log.info("Response message: %s", resp_message) + self._producer.delete_alarm_response( + 'delete_alarm_response', resp_message, + 'alarm_response') + except Exception as exc: + log.warn("Failed to create delete reponse:%s", exc) elif message.key == "acknowledge_alarm": # Acknowledge that an alarm has been dealt with by the SO - # Set its state to ok - auth_token = self._common._authenticate() - endpoint = self._common.get_endpoint("alarming") - alarm_id = values['ack_details']['alarm_uuid'] response = self.update_alarm_state( endpoint, auth_token, alarm_id) + # Log if an alarm was reset if response is True: - log.info("Status has been updated for alarm, %s.", - alarm_id) + log.info("Acknowledged the alarm and cleared it.") else: - log.warn("Failed update the state of requested alarm.") + log.warn("Failed to acknowledge/clear the alarm.") elif message.key == "update_alarm_request": # Update alarm configurations - auth_token = self._common._authenticate() - endpoint = self._common.get_endpoint("alarming") - alarm_details = values['alarm_update_request'] - alarm_id = self.update_alarm( + alarm_id, status = self.update_alarm( endpoint, auth_token, alarm_details) - # TODO(mcgoughh): send a response message to the SO - if alarm_id is not None: - log.info("Alarm configuration was update correctly.") - else: - log.warn("Unable to update the specified alarm") + # Generate a response for an update request + try: + resp_message = self._response.generate_response( + 'update_alarm_response', alarm_id=alarm_id, + cor_id=alarm_details['correlation_id'], + status=status) + log.info("Response message: %s", resp_message) + self._producer.update_alarm_response( + 'update_alarm_response', resp_message, + 'alarm_response') + except Exception as exc: + log.warn("Failed to send an update response:%s", exc) else: log.debug("Unknown key, no action will be performed") @@ -145,108 +198,154 @@ class Alarming(object): return - def get_alarm_id(self, endpoint, auth_token, alarm_name): - """Get a list of alarms that exist in Aodh.""" - alarm_id = None - url = "{}/v2/alarms/".format(endpoint) - - # 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", alarm_name)]) - - result = self._common._perform_request( - url, auth_token, req_type="get", params=query) - - 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 alarm_id - def configure_alarm(self, endpoint, auth_token, values): """Create requested alarm in Aodh.""" url = "{}/v2/alarms/".format(endpoint) - alarm_name = values['alarm_name'] + # Check if the desired alarm is supported + alarm_name = values['alarm_name'].lower() + metric_name = values['metric_name'].lower() + resource_id = values['resource_uuid'] - # Confirm alarm doesn't exist - alarm_id = self.get_alarm_id(endpoint, auth_token, alarm_name) - if alarm_id is None: - # Try to create the alarm - try: - metric_name = values['metric_name'] - resource_id = values['resource_uuid'] + if alarm_name not in ALARM_NAMES.keys(): + log.warn("This alarm is not supported, by a valid metric.") + return None, False + if ALARM_NAMES[alarm_name] != metric_name: + log.warn("This is not the correct metric for this alarm.") + return None, False + + # Check for the required metric + metric_id = self.check_for_metric(auth_token, metric_name, resource_id) + + try: + if metric_id is not None: + # Create the alarm if metric is available payload = self.check_payload(values, metric_name, resource_id, alarm_name) new_alarm = self._common._perform_request( url, auth_token, req_type="post", payload=payload) + return json.loads(new_alarm.text)['alarm_id'], True + else: + log.warn("The required Gnocchi metric does not exist.") + return None, False - return json.loads(new_alarm.text)['alarm_id'] - except Exception as exc: - log.warn("Alarm creation could not be performed: %s", exc) - return alarm_id - else: - log.warn("This alarm already exists. Try an update instead.") - return None + except Exception as exc: + log.warn("Failed to create the alarm: %s", exc) + return None, False def delete_alarm(self, endpoint, auth_token, alarm_id): """Delete alarm function.""" url = "{}/v2/alarms/%s".format(endpoint) % (alarm_id) - result = False try: - self._common._perform_request(url, auth_token, req_type="delete") - return True + result = self._common._perform_request( + url, auth_token, req_type="delete") + if str(result.status_code) == "404": + log.info("Alarm doesn't exist: %s", result.status_code) + # If status code is 404 alarm did not exist + return False + else: + return True + except Exception as exc: log.warn("Failed to delete alarm: %s because %s.", alarm_id, exc) - return result + return False - def list_alarms(self, endpoint, auth_token, - alarm_name=None, resource_id=None, severity=None): + def list_alarms(self, endpoint, auth_token, list_details): """Generate the requested list of alarms.""" - result = None - if (alarm_name and resource_id and severity) is None: - # List all alarms - url = "{}/v2/alarms/".format(endpoint) + url = "{}/v2/alarms/".format(endpoint) + a_list, name_list, sev_list, res_list = [], [], [], [] - try: - result = self._common._perform_request( - url, auth_token, req_type="get") - return json.loads(result.text) - except Exception as exc: - log.warn("Unable to generate alarm list: %s", exc) + # TODO(mcgoughh): for now resource_id is a mandatory field + resource = list_details['resource_uuid'] + + # Checking what fields are specified for a list request + try: + name = list_details['alarm_name'].lower() + if name not in ALARM_NAMES.keys(): + log.warn("This alarm is not supported, won't be used!") + name = None + except KeyError as exc: + log.info("Alarm name isn't specified.") + name = None + + try: + severity = list_details['severity'].lower() + sev = SEVERITIES[severity] + except KeyError as exc: + log.info("Severity is unspecified/incorrectly configured") + sev = None + + # Perform the request to get the desired list + try: + result = self._common._perform_request( + url, auth_token, req_type="get") + + if result is not None: + # Get list based on resource id + for alarm in json.loads(result.text): + rule = alarm['gnocchi_resources_threshold_rule'] + if resource == rule['resource_id']: + res_list.append(str(alarm)) + if not res_list: + log.info("No alarms for this resource") + return a_list + + # Generate specified listed if requested + if name is not None and sev is not None: + log.info("Return a list of %s alarms with %s severity.", + name, sev) + for alarm in json.loads(result.text): + if name == alarm['name']: + name_list.append(str(alarm)) + for alarm in json.loads(result.text): + if sev == alarm['severity']: + sev_list.append(str(alarm)) + name_sev_list = list(set(name_list).intersection(sev_list)) + a_list = list(set(name_sev_list).intersection(res_list)) + elif name is not None: + log.info("Returning a %s list of alarms.", name) + for alarm in json.loads(result.text): + if name == alarm['name']: + name_list.append(str(alarm)) + a_list = list(set(name_list).intersection(res_list)) + elif sev is not None: + log.info("Returning %s severity alarm list.", sev) + for alarm in json.loads(result.text): + if sev == alarm['severity']: + sev_list.append(str(alarm)) + a_list = list(set(sev_list).intersection(res_list)) + else: + log.info("Returning an entire list of alarms.") + a_list = res_list + else: + log.info("There are no alarms!") - return result - else: - # TODO(mcgoughh): support more specific lists - log.debug("Requested list is unavailable") + except Exception as exc: + log.info("Failed to generate required list: %s", exc) + return None - return result + return a_list def update_alarm_state(self, endpoint, auth_token, alarm_id): """Set the state of an alarm to ok when ack message is received.""" - result = False - url = "{}/v2/alarms/%s/state".format(endpoint) % alarm_id payload = json.dumps("ok") try: - result = self._common._perform_request( + self._common._perform_request( url, auth_token, req_type="put", payload=payload) return True except Exception as exc: log.warn("Unable to update alarm state: %s", exc) - return result + return False def update_alarm(self, endpoint, auth_token, values): """Get alarm name for an alarm configuration update.""" # Get already existing alarm details url = "{}/v2/alarms/%s".format(endpoint) % values['alarm_uuid'] + # Gets current configurations about the alarm try: result = self._common._perform_request( url, auth_token, req_type="get") @@ -257,43 +356,46 @@ class Alarming(object): metric_name = rule['metric'] except Exception as exc: log.warn("Failed to retreive existing alarm info: %s.\ - Can only update OSM created alarms.", exc) - return None + Can only update OSM alarms.", exc) + return None, False - # Genate and check payload configuration for alarm update + # Generates and check payload configuration for alarm update payload = self.check_payload(values, metric_name, resource_id, alarm_name, alarm_state=alarm_state) + # Updates the alarm configurations with the valid payload if payload is not None: try: update_alarm = self._common._perform_request( url, auth_token, req_type="put", payload=payload) - return json.loads(update_alarm.text)['alarm_id'] + return json.loads(update_alarm.text)['alarm_id'], True except Exception as exc: log.warn("Alarm update could not be performed: %s", exc) - return None - return None + return None, False + return None, False def check_payload(self, values, metric_name, resource_id, alarm_name, alarm_state=None): """Check that the payload is configuration for update/create alarm.""" try: # Check state and severity - severity = values['severity'] - if severity == "INDETERMINATE": + severity = values['severity'].lower() + if severity == "indeterminate": alarm_state = "insufficient data" - if alarm_state is None: alarm_state = "ok" + statistic = values['statistic'].lower() # Try to configure the payload for the update/create request + # Can only update: threshold, operation, statistic and + # the severity of the alarm rule = {'threshold': values['threshold_value'], 'comparison_operator': values['operation'].lower(), 'metric': metric_name, 'resource_id': resource_id, 'resource_type': 'generic', - 'aggregation_method': values['statistic'].lower()} + 'aggregation_method': STATISTICS[statistic]} payload = json.dumps({'state': alarm_state, 'name': alarm_name, 'severity': SEVERITIES[severity], @@ -303,3 +405,46 @@ class Alarming(object): except KeyError as exc: log.warn("Alarm is not configured correctly: %s", exc) return None + + def authenticate(self): + """Generate an authentication token and endpoint for alarm request.""" + try: + # Check for a tenant_id + auth_token = self._common._authenticate() + endpoint = self._common.get_endpoint("alarming") + return auth_token, endpoint + except Exception as exc: + log.warn("Authentication to Keystone failed:%s", exc) + return None, None + + def get_alarm_state(self, endpoint, auth_token, alarm_id): + """Get the state of the alarm.""" + url = "{}/v2/alarms/%s/state".format(endpoint) % alarm_id + + try: + alarm_state = self._common._perform_request( + url, auth_token, req_type="get") + return json.loads(alarm_state.text) + except Exception as exc: + log.warn("Failed to get the state of the alarm:%s", exc) + return None + + def check_for_metric(self, auth_token, m_name, r_id): + """Check for the alarm metric.""" + try: + endpoint = self._common.get_endpoint("metric") + + url = "{}/v1/metric/".format(endpoint) + metric_list = self._common._perform_request( + url, auth_token, req_type="get") + + for metric in json.loads(metric_list.text): + name = metric['name'] + resource = metric['resource_id'] + if (name == m_name and resource == r_id): + metric_id = metric['id'] + log.info("The required metric exists, an alarm will be created.") + return metric_id + except Exception as exc: + log.info("Desired Gnocchi metric not found:%s", exc) + return None