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