X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=plugins%2FOpenStack%2FAodh%2Falarming.py;h=f530a35033f0378b5cdf8beb7b7b873912414d2c;hb=f0434f4d3dab5f0a4fb417560de4b6bd6d6d45bb;hp=3ae31ba974068e5775bbc620eaa8aac4afd7e8d6;hpb=f358b4fdd006b427e5b653d467c29ae37a47406e;p=osm%2FMON.git diff --git a/plugins/OpenStack/Aodh/alarming.py b/plugins/OpenStack/Aodh/alarming.py index 3ae31ba..f530a35 100644 --- a/plugins/OpenStack/Aodh/alarming.py +++ b/plugins/OpenStack/Aodh/alarming.py @@ -1,62 +1,380 @@ -"""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 -from plugins.Openstack.Aodh.aodh_common import Aodh_Common +import logging as log + +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", + "Read_Latency_Above_Threshold", + "Write_Latency_Above_Threshold", + "DISK_READ_OPS", + "DISK_WRITE_OPS", + "DISK_READ_BYTES", + "DISK_WRITE_BYTES", + "Net_Packets_Dropped", + "Packets_in_Above_Threshold", + "Packets_out_Above_Threshold", + "CPU_Utilization_Above_Threshold"] + +SEVERITIES = { + "WARNING": "low", + "MINOR": "low", + "MAJOR": "moderate", + "CRITICAL": "critical", + "INDETERMINATE": "critical"} 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.""" - self._aodh_common = Aodh_Common() + """Create the OpenStack alarming instance.""" + self._common = Common() + + # TODO(mcgoughh): Remove hardcoded kafkaconsumer + # Initialize a generic consumer object to consume message from the SO + server = {'server': 'localhost:9092', 'topic': 'alarm_request'} + self._consumer = KafkaConsumer(server['topic'], + group_id='osm_mon', + bootstrap_servers=server['server']) + + # 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): - """Receive payload from Aodh.""" - auth_token = self._aodh_common._authenticate() - endpoint = self._aodh_common.get_endpoint() + """Consume info from the message bus to manage alarms.""" + # Check the alarming functionlity that needs to be performed + for message in self._consumer: + + values = json.loads(message.value) + vim_type = values['vim_type'].lower() + + 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(values) + + if message.key == "create_alarm_request": + # Configure/Update an alarm + alarm_details = values['alarm_create_request'] + + alarm_id, alarm_status = self.configure_alarm( + endpoint, auth_token, alarm_details) + + # Generate a valid response message, send via producer + try: + resp_message = self._response.generate_response( + 'create_alarm_response', status=alarm_status, + alarm_id=alarm_id, + cor_id=alarm_details['correlation_id']) + 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": + # Check for a specifed: alarm_name, resource_uuid, severity + # and generate the appropriate list + list_details = values['alarm_list_request'] + try: + name = list_details['alarm_name'] + alarm_list = self.list_alarms( + endpoint, auth_token, alarm_name=name) + except Exception as a_name: + log.debug("No name specified for list:%s", a_name) + try: + resource = list_details['resource_uuid'] + alarm_list = self.list_alarms( + endpoint, auth_token, resource_id=resource) + except Exception as r_id: + log.debug("No resource id specified for this list:\ + %s", r_id) + try: + severe = list_details['severity'] + alarm_list = self.list_alarms( + endpoint, auth_token, severity=severe) + except Exception as exc: + log.warn("No severity specified for list: %s.\ + will return full list.", exc) + alarm_list = self.list_alarms( + endpoint, auth_token) + + 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']) + 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": + request_details = values['alarm_delete_request'] + alarm_id = request_details['alarm_uuid'] - alarm_list = self._get_alarm_list(endpoint, auth_token) - # Confirm communication with Aodh by listing alarms - print("Alarm List ", alarm_list.text) + resp_status = self.delete_alarm( + endpoint, auth_token, alarm_id) - alarm_id = self._create_alarm(endpoint, auth_token) - print(alarm_id) + # 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']) + 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 + 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("Acknowledged the alarm and cleared it.") + else: + log.warn("Failed to acknowledge/clear the alarm.") + + elif message.key == "update_alarm_request": + # Update alarm configurations + alarm_details = values['alarm_update_request'] + + alarm_id, status = self.update_alarm( + endpoint, auth_token, alarm_details) + + # 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) + 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") + 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): - """Get a list of alarms that exist in Aodh.""" + def configure_alarm(self, endpoint, auth_token, values): + """Create requested alarm in Aodh.""" url = "{}/v2/alarms/".format(endpoint) - alarm_list = self._aodh_common._perform_request(url, auth_token, - req_type="get") - return alarm_list + # Check if the desired alarm is supported + alarm_name = values['alarm_name'] + if alarm_name not in ALARM_NAMES: + log.warn("This alarm is not supported, by a valid metric.") + return None, False + + try: + metric_name = values['metric_name'] + resource_id = values['resource_uuid'] + # Check the payload for the desired alarm + 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) - def _get_alarm_info(self, endpoint, auth_token, alarmID): - """Get information about a specific alarm from Aodh.""" - url = "{}/v2/alarms/%s".format(endpoint) % (alarmID) + return json.loads(new_alarm.text)['alarm_id'], True + except Exception as exc: + log.warn("Alarm creation could not be performed: %s", exc) + return None, False - alarm_details = self._aodh_common._perform_request(url, auth_token, - req_type="get") - return alarm_details + def delete_alarm(self, endpoint, auth_token, alarm_id): + """Delete alarm function.""" + url = "{}/v2/alarms/%s".format(endpoint) % (alarm_id) - def _create_alarm(self, endpoint, auth_token): - """Get a list of alarms that exist in Aodh.""" + try: + result = self._common._perform_request( + url, auth_token, req_type="delete") + if str(result.status_code) == "404": + # 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 False + + def list_alarms(self, endpoint, auth_token, + alarm_name=None, resource_id=None, severity=None): + """Generate the requested list of alarms.""" url = "{}/v2/alarms/".format(endpoint) + alarm_list = [] + + result = self._common._perform_request( + url, auth_token, req_type="get") + if result is not None: + # Check for a specified list based on: + # alarm_name, severity, resource_id + if alarm_name is not None: + for alarm in json.loads(result.text): + if alarm_name in str(alarm): + alarm_list.append(str(alarm)) + elif resource_id is not None: + for alarm in json.loads(result.text): + if resource_id in str(alarm): + alarm_list.append(str(alarm)) + elif severity is not None: + for alarm in json.loads(result.text): + if severity in str(alarm): + alarm_list.append(str(alarm)) + else: + alarm_list = result.text + else: + return None + return alarm_list + + def update_alarm_state(self, endpoint, auth_token, alarm_id): + """Set the state of an alarm to ok when ack message is received.""" + url = "{}/v2/alarms/%s/state".format(endpoint) % alarm_id + payload = json.dumps("ok") + + try: + 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 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") + alarm_name = json.loads(result.text)['name'] + rule = json.loads(result.text)['gnocchi_resources_threshold_rule'] + alarm_state = json.loads(result.text)['state'] + resource_id = rule['resource_id'] + 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, False + + # 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'], True + except Exception as exc: + log.warn("Alarm update could not be performed: %s", exc) + 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": + alarm_state = "insufficient data" + if alarm_state is None: + alarm_state = "ok" + + # 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()} + payload = json.dumps({'state': alarm_state, + 'name': alarm_name, + 'severity': SEVERITIES[severity], + 'type': 'gnocchi_resources_threshold', + 'gnocchi_resources_threshold_rule': rule, }) + return payload + except KeyError as exc: + log.warn("Alarm is not configured correctly: %s", exc) + return None + + def authenticate(self, values): + """Generate an authentication token and endpoint for alarm request.""" + try: + # Check for a tenant_id + auth_token = self._common._authenticate( + tenant_id=values['tenant_uuid']) + endpoint = self._common.get_endpoint("alarming") + except Exception as exc: + log.warn("Tenant ID is not specified. Will use a generic\ + authentication: %s", exc) + auth_token = self._common._authenticate() + endpoint = self._common.get_endpoint("alarming") - rule = {'event_type': "threshold",} - payload = json.dumps({'state': 'alarm', - 'name': 'my_alarm', - 'severity': 'moderate', - 'type': 'event', - 'event_rule': rule,}) + return auth_token, endpoint - 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 + 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