1 """Send alarm info from Aodh to SO via MON."""
6 from collections
import OrderedDict
8 from kafka
import KafkaConsumer
10 from plugins
.OpenStack
.Aodh
.aodh_common
import Aodh_Common
13 class Alarming(object):
14 """Receives alarm info from Aodh."""
17 """Create the aodh_receiver instance."""
18 self
._aodh
_common
= Aodh_Common()
20 # Initialize a generic consumer object to consume message from the SO
21 server
= {'server': 'localhost:9092', 'topic': 'alarms'}
22 self
._consumer
= KafkaConsumer(server
['topic'],
24 bootstrap_servers
=server
['server'])
26 # TODO(mcgoughh): Initialize a producer to send messages bask to the SO
29 """Consume info from the message bus to manage alarms."""
30 # Generate authentication credentials to access keystone;
31 # auth_token, endpoint
32 auth_token
= self
._aodh
_common
._authenticate
()
33 endpoint
= self
._aodh
_common
.get_endpoint()
35 # Check the alarming functionlity that needs to be performed
36 for message
in self
._consumer
:
37 if message
.topic
== "alarms":
38 log
.info("Alarm action required: %s" % (message
.topic
))
40 if message
.key
== "configure_alarm":
41 # Configure/Update an alarm
42 alarm_details
= json
.loads(message
.value
)
43 alarm_id
= self
.configure_alarm(endpoint
,
44 auth_token
, alarm_details
)
45 log
.info("New alarm created with alarmID: %s", alarm_id
)
47 # TODO(mcgoughh): will send an acknowledge message back on
48 # the bus via the producer
51 # TODO(mcoughh): Key alternatives are "notify_alarm" and
52 # "acknowledge_alarm" will be accomodated later
53 log
.debug("Unknown key, no action will be performed")
55 log
.info("Message topic not relevant to this plugin: %s",
60 def alarm_check(self
, endpoint
, auth_token
, alarm_name
):
61 """Get a list of alarms that exist in Aodh."""
62 url
= "{}/v2/alarms/".format(endpoint
)
64 # TODO(mcgoughh): will query on resource_id once it has been
65 # implemented need to create the query field when creating
67 query
= OrderedDict([("q.field", 'name'), ("q.op", "eq"),
68 ("q.value", str(alarm_name
))])
70 result
= self
._aodh
_common
._perform
_request
(
71 url
, auth_token
, req_type
="get", params
=query
)
74 alarm_id
= json
.loads(result
.text
)[0]['alarm_id']
75 log
.info("An existing alarm was found: %s", alarm_id
)
78 log
.debug("Alarm doesn't exist, needs to be created.")
81 def configure_alarm(self
, endpoint
, auth_token
, values
):
82 """Get a list of alarms that exist in Aodh."""
85 # TODO(mcgoughh): error check the values sent in the messag
86 alarm_name
= values
['name']
88 # Check that this alarm doesn't exist already
89 alarm_id
= self
.alarm_check(endpoint
, auth_token
, alarm_name
)
92 url
= "{}/v2/alarms/".format(endpoint
)
93 severity
= values
['severity']
95 # Create a new threshold alarm with a resourceID
96 # specified as a query
97 rule
= {'threshold': values
['threshold'],
98 'comparison_operator': 'gt',
99 'metric': values
['metric'],
100 'resource_id': values
['resource_id'],
101 'resource_type': 'generic',
102 'aggregation_method': 'last', }
103 payload
= json
.dumps({'state': 'alarm',
105 'severity': self
.get_severity(severity
),
106 'type': 'gnocchi_resources_threshold',
107 'gnocchi_resources_threshold_rule': rule
, })
109 # Request performed to create alarm
110 new_alarm
= self
._aodh
_common
._perform
_request
(
111 url
, auth_token
, req_type
="post", payload
=payload
)
113 return json
.loads(new_alarm
.text
)['alarm_id']
117 def delete_alarm(self
, endpoint
, auth_token
, alarmID
):
118 """Delete alarm function."""
119 url
= "{}/v2/alarms/%s".format(endpoint
) % (alarmID
)
121 self
._aodh
_common
._perform
_request
(url
, auth_token
, req_type
="delete")
124 def get_severity(self
, alarm_severity
):
125 """Get a normalized severity for Aodh."""
126 # This logic can be changed, the other alternative was to have
127 # MINOR and MAJOR = "moderate" instead.
128 if alarm_severity
== "WARNIING":
129 aodh_severity
= "low"
130 elif alarm_severity
== "MINOR":
131 aodh_severity
= "moderate"
132 elif (alarm_severity
== "MAJOR" or alarm_severity
== "CRITICAL"):
133 aodh_severity
= "critical"
136 log
.warn("Invalid alarm severity configuration")
138 log
.info("Severity has been normalized for Aodh to: %s", aodh_severity
)