osm_mon/plugins/OpenStack/Aodh/alarming.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 """Carry out alarming requests via Aodh API."""
  23
  24 import json
  25 import logging
  26
  27 import six
  28
  29 from osm_mon.core.database import DatabaseManager
  30 from osm_mon.core.message_bus.producer import KafkaProducer
  31 from osm_mon.plugins.OpenStack.common import Common
  32 from osm_mon.plugins.OpenStack.response import OpenStack_Response
  33 from osm_mon.plugins.OpenStack.settings import Config
  34
  35 log = logging.getLogger(__name__)
  36
  37 ALARM_NAMES = {
  38     "average_memory_usage_above_threshold": "average_memory_utilization",
  39     "disk_read_ops": "disk_read_ops",
  40     "disk_write_ops": "disk_write_ops",
  41     "disk_read_bytes": "disk_read_bytes",
  42     "disk_write_bytes": "disk_write_bytes",
  43     "net_packets_dropped": "packets_dropped",
  44     "packets_in_above_threshold": "packets_received",
  45     "packets_out_above_threshold": "packets_sent",
  46     "cpu_utilization_above_threshold": "cpu_utilization"}
  47
  48 METRIC_MAPPINGS = {
  49     "average_memory_utilization": "memory.percent",
  50     "disk_read_ops": "disk.read.requests",
  51     "disk_write_ops": "disk.write.requests",
  52     "disk_read_bytes": "disk.read.bytes",
  53     "disk_write_bytes": "disk.write.bytes",
  54     "packets_dropped": "interface.if_dropped",
  55     "packets_received": "interface.if_packets",
  56     "packets_sent": "interface.if_packets",
  57     "cpu_utilization": "cpu_util",
  58 }
  59
  60 SEVERITIES = {
  61     "warning": "low",
  62     "minor": "low",
  63     "major": "moderate",
  64     "critical": "critical",
  65     "indeterminate": "critical"}
  66
  67 STATISTICS = {
  68     "average": "mean",
  69     "minimum": "min",
  70     "maximum": "max",
  71     "count": "count",
  72     "sum": "sum"}
  73
  74
  75 class Alarming(object):
  76     """Carries out alarming requests and responses via Aodh API."""
  77
  78     def __init__(self):
  79         """Create the OpenStack alarming instance."""
  80         # Initialize configuration and notifications
  81         config = Config.instance()
  82         config.read_environ()
  83
  84         self._database_manager = DatabaseManager()
  85
  86         # Use the Response class to generate valid json response messages
  87         self._response = OpenStack_Response()
  88
  89         # Initializer a producer to send responses back to SO
  90         self._producer = KafkaProducer("alarm_response")
  91
  92     def alarming(self, message):
  93         """Consume info from the message bus to manage alarms."""
  94         values = json.loads(message.value)
  95
  96         log.info("OpenStack alarm action required.")
  97         vim_uuid = values['vim_uuid']
  98
  99         auth_token = Common.get_auth_token(vim_uuid)
 100
 101         alarm_endpoint = Common.get_endpoint("alarming", vim_uuid)
 102         metric_endpoint = Common.get_endpoint("metric", vim_uuid)
 103
 104         if message.key == "create_alarm_request":
 105             # Configure/Update an alarm
 106             alarm_details = values['alarm_create_request']
 107
 108             alarm_id, alarm_status = self.configure_alarm(
 109                 alarm_endpoint, metric_endpoint, auth_token, alarm_details)
 110
 111             # Generate a valid response message, send via producer
 112             try:
 113                 if alarm_status is True:
 114                     log.info("Alarm successfully created")
 115                     self._database_manager.save_alarm(alarm_id, vim_uuid)
 116
 117                 resp_message = self._response.generate_response(
 118                     'create_alarm_response', status=alarm_status,
 119                     alarm_id=alarm_id,
 120                     cor_id=alarm_details['correlation_id'])
 121                 log.info("Response Message: %s", resp_message)
 122                 self._producer.create_alarm_response(
 123                     'create_alarm_response', resp_message,
 124                     'alarm_response')
 125             except Exception as exc:
 126                 log.exception("Response creation failed:")
 127
 128         elif message.key == "list_alarm_request":
 129             # Check for a specified: alarm_name, resource_uuid, severity
 130             # and generate the appropriate list
 131             list_details = values['alarm_list_request']
 132
 133             alarm_list = self.list_alarms(
 134                 alarm_endpoint, auth_token, list_details)
 135
 136             try:
 137                 # Generate and send a list response back
 138                 resp_message = self._response.generate_response(
 139                     'list_alarm_response', alarm_list=alarm_list,
 140                     cor_id=list_details['correlation_id'])
 141                 log.info("Response Message: %s", resp_message)
 142                 self._producer.list_alarm_response(
 143                     'list_alarm_response', resp_message,
 144                     'alarm_response')
 145             except Exception as exc:
 146                 log.exception("Failed to send a valid response back.")
 147
 148         elif message.key == "delete_alarm_request":
 149             request_details = values['alarm_delete_request']
 150             alarm_id = request_details['alarm_uuid']
 151
 152             resp_status = self.delete_alarm(
 153                 alarm_endpoint, auth_token, alarm_id)
 154
 155             # Generate and send a response message
 156             try:
 157                 resp_message = self._response.generate_response(
 158                     'delete_alarm_response', alarm_id=alarm_id,
 159                     status=resp_status,
 160                     cor_id=request_details['correlation_id'])
 161                 log.info("Response message: %s", resp_message)
 162                 self._producer.delete_alarm_response(
 163                     'delete_alarm_response', resp_message,
 164                     'alarm_response')
 165             except Exception as exc:
 166                 log.warn("Failed to create delete response:%s", exc)
 167
 168         elif message.key == "acknowledge_alarm":
 169             # Acknowledge that an alarm has been dealt with by the SO
 170             alarm_id = values['ack_details']['alarm_uuid']
 171
 172             response = self.update_alarm_state(
 173                 alarm_endpoint, auth_token, alarm_id)
 174
 175             # Log if an alarm was reset
 176             if response is True:
 177                 log.info("Acknowledged the alarm and cleared it.")
 178             else:
 179                 log.warn("Failed to acknowledge/clear the alarm.")
 180
 181         elif message.key == "update_alarm_request":
 182             # Update alarm configurations
 183             alarm_details = values['alarm_update_request']
 184
 185             alarm_id, status = self.update_alarm(
 186                 alarm_endpoint, auth_token, alarm_details)
 187
 188             # Generate a response for an update request
 189             try:
 190                 resp_message = self._response.generate_response(
 191                     'update_alarm_response', alarm_id=alarm_id,
 192                     cor_id=alarm_details['correlation_id'],
 193                     status=status)
 194                 log.info("Response message: %s", resp_message)
 195                 self._producer.update_alarm_response(
 196                     'update_alarm_response', resp_message,
 197                     'alarm_response')
 198             except Exception as exc:
 199                 log.warn("Failed to send an update response:%s", exc)
 200
 201         else:
 202             log.debug("Unknown key, no action will be performed")
 203
 204         return
 205
 206     def configure_alarm(self, alarm_endpoint, metric_endpoint, auth_token, values):
 207         """Create requested alarm in Aodh."""
 208         url = "{}/v2/alarms/".format(alarm_endpoint)
 209
 210         # Check if the desired alarm is supported
 211         alarm_name = values['alarm_name'].lower()
 212         metric_name = values['metric_name'].lower()
 213         resource_id = values['resource_uuid']
 214
 215         if metric_name not in METRIC_MAPPINGS.keys():
 216             log.warn("This metric is not supported.")
 217             return None, False
 218
 219         # Check for the required metric
 220         metric_id = self.check_for_metric(auth_token, metric_endpoint, metric_name, resource_id)
 221
 222         try:
 223             if metric_id is not None:
 224                 # Create the alarm if metric is available
 225                 payload = self.check_payload(values, metric_name, resource_id,
 226                                              alarm_name)
 227                 new_alarm = Common.perform_request(
 228                     url, auth_token, req_type="post", payload=payload)
 229                 return json.loads(new_alarm.text)['alarm_id'], True
 230             else:
 231                 log.warn("The required Gnocchi metric does not exist.")
 232                 return None, False
 233
 234         except Exception as exc:
 235             log.warn("Failed to create the alarm: %s", exc)
 236         return None, False
 237
 238     def delete_alarm(self, endpoint, auth_token, alarm_id):
 239         """Delete alarm function."""
 240         url = "{}/v2/alarms/%s".format(endpoint) % alarm_id
 241
 242         try:
 243             result = Common.perform_request(
 244                 url, auth_token, req_type="delete")
 245             if str(result.status_code) == "404":
 246                 log.info("Alarm doesn't exist: %s", result.status_code)
 247                 # If status code is 404 alarm did not exist
 248                 return False
 249             else:
 250                 return True
 251
 252         except Exception as exc:
 253             log.warn("Failed to delete alarm: %s because %s.", alarm_id, exc)
 254         return False
 255
 256     def list_alarms(self, endpoint, auth_token, list_details):
 257         """Generate the requested list of alarms."""
 258         url = "{}/v2/alarms/".format(endpoint)
 259         a_list, name_list, sev_list, res_list = [], [], [], []
 260
 261         # TODO(mcgoughh): for now resource_id is a mandatory field
 262         # Check for a resource id
 263         try:
 264             resource = list_details['resource_uuid']
 265         except KeyError as exc:
 266             log.warn("Resource id not specified for list request: %s", exc)
 267             return None
 268
 269         # Checking what fields are specified for a list request
 270         try:
 271             name = list_details['alarm_name'].lower()
 272             if name not in ALARM_NAMES.keys():
 273                 log.warn("This alarm is not supported, won't be used!")
 274                 name = None
 275         except KeyError as exc:
 276             log.info("Alarm name isn't specified.")
 277             name = None
 278
 279         try:
 280             severity = list_details['severity'].lower()
 281             sev = SEVERITIES[severity]
 282         except KeyError as exc:
 283             log.info("Severity is unspecified/incorrectly configured")
 284             sev = None
 285
 286         # Perform the request to get the desired list
 287         try:
 288             result = Common.perform_request(
 289                 url, auth_token, req_type="get")
 290
 291             if result is not None:
 292                 # Get list based on resource id
 293                 for alarm in json.loads(result.text):
 294                     rule = alarm['gnocchi_resources_threshold_rule']
 295                     if resource == rule['resource_id']:
 296                         res_list.append(alarm)
 297                     if not res_list:
 298                         log.info("No alarms for this resource")
 299                         return a_list
 300
 301                 # Generate specified listed if requested
 302                 if name is not None and sev is not None:
 303                     log.info("Return a list of %s alarms with %s severity.",
 304                              name, sev)
 305                     for alarm in json.loads(result.text):
 306                         if name == alarm['name']:
 307                             name_list.append(alarm)
 308                     for alarm in json.loads(result.text):
 309                         if sev == alarm['severity']:
 310                             sev_list.append(alarm)
 311                     name_sev_list = list(set(name_list).intersection(sev_list))
 312                     a_list = list(set(name_sev_list).intersection(res_list))
 313                 elif name is not None:
 314                     log.info("Returning a %s list of alarms.", name)
 315                     for alarm in json.loads(result.text):
 316                         if name == alarm['name']:
 317                             name_list.append(alarm)
 318                     a_list = list(set(name_list).intersection(res_list))
 319                 elif sev is not None:
 320                     log.info("Returning %s severity alarm list.", sev)
 321                     for alarm in json.loads(result.text):
 322                         if sev == alarm['severity']:
 323                             sev_list.append(alarm)
 324                     a_list = list(set(sev_list).intersection(res_list))
 325                 else:
 326                     log.info("Returning an entire list of alarms.")
 327                     a_list = res_list
 328             else:
 329                 log.info("There are no alarms!")
 330
 331         except Exception as exc:
 332             log.info("Failed to generate required list: %s", exc)
 333             return None
 334
 335         return a_list
 336
 337     def update_alarm_state(self, endpoint, auth_token, alarm_id):
 338         """Set the state of an alarm to ok when ack message is received."""
 339         url = "{}/v2/alarms/%s/state".format(endpoint) % alarm_id
 340         payload = json.dumps("ok")
 341
 342         try:
 343             Common.perform_request(
 344                 url, auth_token, req_type="put", payload=payload)
 345             return True
 346         except Exception as exc:
 347             log.warn("Unable to update alarm state: %s", exc)
 348         return False
 349
 350     def update_alarm(self, endpoint, auth_token, values):
 351         """Get alarm name for an alarm configuration update."""
 352         # Get already existing alarm details
 353         url = "{}/v2/alarms/%s".format(endpoint) % values['alarm_uuid']
 354
 355         # Gets current configurations about the alarm
 356         try:
 357             result = Common.perform_request(
 358                 url, auth_token, req_type="get")
 359             alarm_name = json.loads(result.text)['name']
 360             rule = json.loads(result.text)['gnocchi_resources_threshold_rule']
 361             alarm_state = json.loads(result.text)['state']
 362             resource_id = rule['resource_id']
 363             metric_name = [key for key, value in six.iteritems(METRIC_MAPPINGS) if value == rule['metric']][0]
 364         except Exception as exc:
 365             log.warn("Failed to retrieve existing alarm info: %s.\
 366                      Can only update OSM alarms.", exc)
 367             return None, False
 368
 369         # Generates and check payload configuration for alarm update
 370         payload = self.check_payload(values, metric_name, resource_id,
 371                                      alarm_name, alarm_state=alarm_state)
 372
 373         # Updates the alarm configurations with the valid payload
 374         if payload is not None:
 375             try:
 376                 update_alarm = Common.perform_request(
 377                     url, auth_token, req_type="put", payload=payload)
 378
 379                 return json.loads(update_alarm.text)['alarm_id'], True
 380             except Exception as exc:
 381                 log.warn("Alarm update could not be performed: %s", exc)
 382                 return None, False
 383         return None, False
 384
 385     def check_payload(self, values, metric_name, resource_id,
 386                       alarm_name, alarm_state=None):
 387         """Check that the payload is configuration for update/create alarm."""
 388         try:
 389             cfg = Config.instance()
 390             # Check state and severity
 391
 392             severity = 'critical'
 393             if 'severity' in values:
 394                 severity = values['severity'].lower()
 395
 396             if severity == "indeterminate":
 397                 alarm_state = "insufficient data"
 398             if alarm_state is None:
 399                 alarm_state = "ok"
 400
 401             statistic = values['statistic'].lower()
 402
 403             granularity = '300'
 404             if 'granularity' in values:
 405                 granularity = values['granularity']
 406
 407             resource_type = 'generic'
 408             if 'resource_type' in values:
 409                 resource_type = values['resource_type'].lower()
 410
 411             # Try to configure the payload for the update/create request
 412             # Can only update: threshold, operation, statistic and
 413             # the severity of the alarm
 414             rule = {'threshold': values['threshold_value'],
 415                     'comparison_operator': values['operation'].lower(),
 416                     'metric': METRIC_MAPPINGS[metric_name],
 417                     'resource_id': resource_id,
 418                     'resource_type': resource_type,
 419                     'aggregation_method': STATISTICS[statistic],
 420                     'granularity': granularity, }
 421             payload = json.dumps({'state': alarm_state,
 422                                   'name': alarm_name,
 423                                   'severity': SEVERITIES[severity],
 424                                   'type': 'gnocchi_resources_threshold',
 425                                   'gnocchi_resources_threshold_rule': rule,
 426                                   'alarm_actions': [cfg.OS_NOTIFIER_URI], })
 427             return payload
 428         except KeyError as exc:
 429             log.warn("Alarm is not configured correctly: %s", exc)
 430         return None
 431
 432     def get_alarm_state(self, endpoint, auth_token, alarm_id):
 433         """Get the state of the alarm."""
 434         url = "{}/v2/alarms/%s/state".format(endpoint) % alarm_id
 435
 436         try:
 437             alarm_state = Common.perform_request(
 438                 url, auth_token, req_type="get")
 439             return json.loads(alarm_state.text)
 440         except Exception as exc:
 441             log.warn("Failed to get the state of the alarm:%s", exc)
 442         return None
 443
 444     def check_for_metric(self, auth_token, metric_endpoint, m_name, r_id):
 445         """Check for the alarm metric."""
 446         try:
 447             url = "{}/v1/metric?sort=name:asc".format(metric_endpoint)
 448             result = Common.perform_request(
 449                 url, auth_token, req_type="get")
 450             metric_list = []
 451             metrics_partial = json.loads(result.text)
 452             for metric in metrics_partial:
 453                 metric_list.append(metric)
 454
 455             while len(json.loads(result.text)) > 0:
 456                 last_metric_id = metrics_partial[-1]['id']
 457                 url = "{}/v1/metric?sort=name:asc&marker={}".format(metric_endpoint, last_metric_id)
 458                 result = Common.perform_request(
 459                     url, auth_token, req_type="get")
 460                 if len(json.loads(result.text)) > 0:
 461                     metrics_partial = json.loads(result.text)
 462                     for metric in metrics_partial:
 463                         metric_list.append(metric)
 464             metric_id = None
 465             for metric in metric_list:
 466                 name = metric['name']
 467                 resource = metric['resource_id']
 468                 if name == METRIC_MAPPINGS[m_name] and resource == r_id:
 469                     metric_id = metric['id']
 470             log.info("The required metric exists, an alarm will be created.")
 471             return metric_id
 472         except Exception as exc:
 473             log.info("Desired Gnocchi metric not found:%s", exc)
 474         return None