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