1 # Copyright 2017 Intel Research and Development Ireland Limited
2 # *************************************************************
4 # This file is part of OSM Monitoring module
5 # All Rights Reserved to Intel Corporation
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
11 # http://www.apache.org/licenses/LICENSE-2.0
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
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
22 """Carry out alarming requests via Aodh API."""
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
35 log
= logging
.getLogger(__name__
)
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"}
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",
64 "critical": "critical",
65 "indeterminate": "critical"}
75 class Alarming(object):
76 """Carries out alarming requests and responses via Aodh API."""
79 """Create the OpenStack alarming instance."""
80 # Initialize configuration and notifications
81 config
= Config
.instance()
84 self
._database
_manager
= DatabaseManager()
86 # Use the Response class to generate valid json response messages
87 self
._response
= OpenStack_Response()
89 # Initializer a producer to send responses back to SO
90 self
._producer
= KafkaProducer("alarm_response")
92 def alarming(self
, message
):
93 """Consume info from the message bus to manage alarms."""
94 values
= json
.loads(message
.value
)
96 log
.info("OpenStack alarm action required.")
97 vim_uuid
= values
['vim_uuid']
99 auth_token
= Common
.get_auth_token(vim_uuid
)
101 alarm_endpoint
= Common
.get_endpoint("alarming", vim_uuid
)
102 metric_endpoint
= Common
.get_endpoint("metric", vim_uuid
)
104 if message
.key
== "create_alarm_request":
105 # Configure/Update an alarm
106 alarm_details
= values
['alarm_create_request']
108 alarm_id
, alarm_status
= self
.configure_alarm(
109 alarm_endpoint
, metric_endpoint
, auth_token
, alarm_details
)
111 # Generate a valid response message, send via producer
113 if alarm_status
is True:
114 log
.info("Alarm successfully created")
115 self
._database
_manager
.save_alarm(alarm_id
, vim_uuid
)
117 resp_message
= self
._response
.generate_response(
118 'create_alarm_response', status
=alarm_status
,
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
,
125 except Exception as exc
:
126 log
.exception("Response creation failed:")
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']
133 alarm_list
= self
.list_alarms(
134 alarm_endpoint
, auth_token
, list_details
)
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
,
145 except Exception as exc
:
146 log
.exception("Failed to send a valid response back.")
148 elif message
.key
== "delete_alarm_request":
149 request_details
= values
['alarm_delete_request']
150 alarm_id
= request_details
['alarm_uuid']
152 resp_status
= self
.delete_alarm(
153 alarm_endpoint
, auth_token
, alarm_id
)
155 # Generate and send a response message
157 resp_message
= self
._response
.generate_response(
158 'delete_alarm_response', alarm_id
=alarm_id
,
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
,
165 except Exception as exc
:
166 log
.warn("Failed to create delete response:%s", exc
)
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']
172 response
= self
.update_alarm_state(
173 alarm_endpoint
, auth_token
, alarm_id
)
175 # Log if an alarm was reset
177 log
.info("Acknowledged the alarm and cleared it.")
179 log
.warn("Failed to acknowledge/clear the alarm.")
181 elif message
.key
== "update_alarm_request":
182 # Update alarm configurations
183 alarm_details
= values
['alarm_update_request']
185 alarm_id
, status
= self
.update_alarm(
186 alarm_endpoint
, auth_token
, alarm_details
)
188 # Generate a response for an update request
190 resp_message
= self
._response
.generate_response(
191 'update_alarm_response', alarm_id
=alarm_id
,
192 cor_id
=alarm_details
['correlation_id'],
194 log
.info("Response message: %s", resp_message
)
195 self
._producer
.update_alarm_response(
196 'update_alarm_response', resp_message
,
198 except Exception as exc
:
199 log
.warn("Failed to send an update response:%s", exc
)
202 log
.debug("Unknown key, no action will be performed")
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
)
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']
215 if metric_name
not in METRIC_MAPPINGS
.keys():
216 log
.warn("This metric is not supported.")
219 # Check for the required metric
220 metric_id
= self
.check_for_metric(auth_token
, metric_endpoint
, metric_name
, resource_id
)
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
,
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
231 log
.warn("The required Gnocchi metric does not exist.")
234 except Exception as exc
:
235 log
.warn("Failed to create the alarm: %s", exc
)
238 def delete_alarm(self
, endpoint
, auth_token
, alarm_id
):
239 """Delete alarm function."""
240 url
= "{}/v2/alarms/%s".format(endpoint
) % alarm_id
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
252 except Exception as exc
:
253 log
.warn("Failed to delete alarm: %s because %s.", alarm_id
, exc
)
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
= [], [], [], []
261 # TODO(mcgoughh): for now resource_id is a mandatory field
262 # Check for a resource id
264 resource
= list_details
['resource_uuid']
265 except KeyError as exc
:
266 log
.warn("Resource id not specified for list request: %s", exc
)
269 # Checking what fields are specified for a list request
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!")
275 except KeyError as exc
:
276 log
.info("Alarm name isn't specified.")
280 severity
= list_details
['severity'].lower()
281 sev
= SEVERITIES
[severity
]
282 except KeyError as exc
:
283 log
.info("Severity is unspecified/incorrectly configured")
286 # Perform the request to get the desired list
288 result
= Common
.perform_request(
289 url
, auth_token
, req_type
="get")
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
)
298 log
.info("No alarms for this resource")
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.",
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
))
326 log
.info("Returning an entire list of alarms.")
329 log
.info("There are no alarms!")
331 except Exception as exc
:
332 log
.info("Failed to generate required list: %s", exc
)
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")
343 Common
.perform_request(
344 url
, auth_token
, req_type
="put", payload
=payload
)
346 except Exception as exc
:
347 log
.warn("Unable to update alarm state: %s", exc
)
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']
355 # Gets current configurations about the alarm
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
)
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
)
373 # Updates the alarm configurations with the valid payload
374 if payload
is not None:
376 update_alarm
= Common
.perform_request(
377 url
, auth_token
, req_type
="put", payload
=payload
)
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
)
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."""
389 cfg
= Config
.instance()
390 # Check state and severity
392 severity
= 'critical'
393 if 'severity' in values
:
394 severity
= values
['severity'].lower()
396 if severity
== "indeterminate":
397 alarm_state
= "insufficient data"
398 if alarm_state
is None:
401 statistic
= values
['statistic'].lower()
404 if 'granularity' in values
:
405 granularity
= values
['granularity']
407 resource_type
= 'generic'
408 if 'resource_type' in values
:
409 resource_type
= values
['resource_type'].lower()
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
,
423 'severity': SEVERITIES
[severity
],
424 'type': 'gnocchi_resources_threshold',
425 'gnocchi_resources_threshold_rule': rule
,
426 'alarm_actions': [cfg
.OS_NOTIFIER_URI
], })
428 except KeyError as exc
:
429 log
.warn("Alarm is not configured correctly: %s", exc
)
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
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
)
444 def check_for_metric(self
, auth_token
, metric_endpoint
, m_name
, r_id
):
445 """Check for the alarm metric."""
447 url
= "{}/v1/metric?sort=name:asc".format(metric_endpoint
)
448 result
= Common
.perform_request(
449 url
, auth_token
, req_type
="get")
451 metrics_partial
= json
.loads(result
.text
)
452 for metric
in metrics_partial
:
453 metric_list
.append(metric
)
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
)
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.")
472 except Exception as exc
:
473 log
.info("Desired Gnocchi metric not found:%s", exc
)