2 # Copyright 2017 xFlow Research Pvt. Ltd
3 # This file is part of MON module
6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
7 # not use this file except in compliance with the License. You may obtain
8 # a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 # License for the specific language governing permissions and limitations
18 # For those usages not covered by the Apache License, Version 2.0 please
19 # contact with: wajeeha.hamid@xflowresearch.com
23 AWS-Plugin implements all the methods of MON to interact with AWS using the BOTO client
26 __author__
= "Wajeeha Hamid"
27 __date__
= "18-September-2017"
32 from jsmin
import jsmin
33 from connection
import Connection
34 from metric_alarms
import MetricAlarm
35 from metrics
import Metrics
36 sys
.path
.append("../../core/message_bus")
37 from producer
import KafkaProducer
39 class plugin_alarms():
40 """Receives Alarm info from MetricAlarm and connects with the consumer/producer"""
42 self
.metricAlarm
= MetricAlarm()
43 self
.metric
= Metrics()
44 self
.producer
= KafkaProducer('')
45 #---------------------------------------------------------------------------------------------------------------------------
46 def configure_alarm(self
,alarm_info
):
47 alarm_id
= self
.metricAlarm
.config_alarm(self
.cloudwatch_conn
,alarm_info
)
49 #---------------------------------------------------------------------------------------------------------------------------
50 def update_alarm_configuration(self
,test
):
51 alarm_id
= self
.metricAlarm
.update_alarm(self
.cloudwatch_conn
,test
)
53 #---------------------------------------------------------------------------------------------------------------------------
54 def delete_alarm(self
,alarm_id
):
55 return self
.metricAlarm
.delete_Alarm(self
.cloudwatch_conn
,alarm_id
)
56 #---------------------------------------------------------------------------------------------------------------------------
57 def get_alarms_list(self
,instance_id
):
58 return self
.metricAlarm
.alarms_list(self
.cloudwatch_conn
,instance_id
)
59 #---------------------------------------------------------------------------------------------------------------------------
60 def get_ack_details(self
,ack_info
):
61 return self
.metricAlarm
.alarm_details(self
.cloudwatch_conn
,ack_info
)
62 #---------------------------------------------------------------------------------------------------------------------------
63 def get_metrics_data(self
,metric_name
,period
,instance_id
):
64 return self
.metric
.metricsData(self
.cloudwatch_conn
,metric_name
,period
,instance_id
)
65 #---------------------------------------------------------------------------------------------------------------------------
67 def alarm_calls(self
,message
,aws_conn
):
68 """Gets the message from the common consumer"""
70 self
.cloudwatch_conn
= aws_conn
['cloudwatch_connection']
71 self
.ec2_conn
= aws_conn
['ec2_connection']
73 log
.info("Action required against: %s" % (message
.topic
))
74 alarm_info
= json
.loads(message
.value
)
76 if message
.key
== "create_alarm_request":
77 alarm_inner_dict
= alarm_info
['alarm_create_request']
78 metric_status
= self
.check_metric(alarm_inner_dict
['metric_name'])
80 if self
.check_resource(alarm_inner_dict
['resource_uuid']) == True and metric_status
['status'] == True:
81 log
.debug ("Resource and Metrics exists")
83 alarm_info
['alarm_create_request']['metric_name'] = metric_status
['metric_name']
84 #Generate a valid response message, send via producer
85 config_resp
= self
.configure_alarm(alarm_info
) #alarm_info = message.value
87 if config_resp
== None:
88 log
.debug("Alarm Already exists")
89 payload
= json
.dumps(config_resp
)
90 file = open('../../core/models/create_alarm_resp.json','wb').write((payload
))
91 self
.producer
.create_alarm_response(key
='create_alarm_response',message
=payload
,topic
= 'alarm_response')
94 payload
= json
.dumps(config_resp
)
95 file = open('../../core/models/create_alarm_resp.json','wb').write((payload
))
96 self
.producer
.create_alarm_response(key
='create_alarm_response',message
=payload
,topic
= 'alarm_response')
97 log
.info("New alarm created with alarm info: %s", config_resp
)
100 log
.error("Resource ID doesn't exists")
102 elif message
.key
== "acknowledge_alarm":
103 alarm_inner_dict
= alarm_info
['ack_details']
105 if self
.check_resource(alarm_inner_dict
['resource_uuid']) == True:
106 alarm_info
= json
.loads(message
.value
)
107 #Generate a valid response message, send via producer
108 ack_details
= self
.get_ack_details(alarm_info
)
109 payload
= json
.dumps(ack_details
)
110 file = open('../../core/models/notify_alarm.json','wb').write((payload
))
111 self
.producer
.notify_alarm(key
='notify_alarm',message
=payload
,topic
= 'alarm_response')
112 log
.info("Acknowledge sent: %s", ack_details
)
115 log
.error("Resource ID is Incorrect")
118 elif message
.key
== "update_alarm_request":
119 alarm_inner_dict
= alarm_info
['alarm_update_request']
120 metric_status
= self
.check_metric(alarm_inner_dict
['metric_name'])
122 if metric_status
['status'] == True:
123 log
.debug ("Resource and Metrics exists")
124 alarm_info
['alarm_update_request']['metric_name'] = metric_status
['metric_name']
125 #Generate a valid response message, send via producer
126 update_resp
= self
.update_alarm_configuration(alarm_info
)
128 if update_resp
== None:
129 payload
= json
.dumps(update_resp
)
130 file = open('../../core/models/update_alarm_resp.json','wb').write((payload
))
131 self
.producer
.update_alarm_response(key
='update_alarm_response',message
=payload
,topic
= 'alarm_response')
132 log
.debug("Alarm Already exists")
135 payload
= json
.dumps(update_resp
)
136 file = open('../../core/models/update_alarm_resp.json','wb').write((payload
))
137 self
.producer
.update_alarm_response(key
='update_alarm_response',message
=payload
,topic
= 'alarm_response')
138 log
.info("Alarm Updated with alarm info: %s", update_resp
)
141 log
.info ("Metric Not Supported")
144 elif message
.key
== "delete_alarm_request":
145 del_info
= json
.loads(message
.value
)
146 #Generate a valid response message, send via producer
147 del_resp
= self
.delete_alarm(del_info
)
148 payload
= json
.dumps(del_resp
)
149 file = open('../../core/models/delete_alarm_resp.json','wb').write((payload
))
150 self
.producer
.delete_alarm_response(key
='delete_alarm_response',message
=payload
,topic
= 'alarm_response')
151 log
.info("Alarm Deleted with alarm info: %s", del_resp
)
154 elif message
.key
== "alarm_list_request":
155 alarm_inner_dict
= alarm_info
['alarm_list_request']
157 if self
.check_resource(alarm_inner_dict
['resource_uuid']) == True or alarm_inner_dict
['resource_uuid'] == "":
158 #Generate a valid response message, send via producer
159 list_resp
= self
.get_alarms_list(alarm_info
)#['alarm_names']
160 payload
= json
.dumps(list_resp
)
161 file = open('../../core/models/list_alarm_resp.json','wb').write((payload
))
162 self
.producer
.list_alarm_response(key
='list_alarm_response',message
=payload
,topic
= 'alarm_response')
165 log
.error("Resource ID is Incorrect")
168 log
.debug("Unknown key, no action will be performed")
170 except Exception as e
:
171 log
.error("Message retrieval exception: %s", str(e
))
172 #---------------------------------------------------------------------------------------------------------------------------
173 def check_resource(self
,resource_uuid
):
174 '''Finding Resource with the resource_uuid'''
177 instances
= self
.ec2_conn
.get_all_instance_status()
180 for instance_id
in instances
:
181 instance_id
= str(instance_id
).split(':')[1]
183 if instance_id
== resource_uuid
:
184 check_resp
['resource_uuid'] = resource_uuid
188 except Exception as e
:
189 log
.error("Error in Plugin Inputs %s",str(e
))
190 #---------------------------------------------------------------------------------------------------------------------------
191 def check_metric(self
,metric_name
):
192 ''' Checking whether the metric is supported by AWS '''
197 if metric_name
== 'CPU_UTILIZATION':
198 metric_name
= 'CPUUtilization'
201 elif metric_name
== 'DISK_READ_OPS':
202 metric_name
= 'DiskReadOps'
205 elif metric_name
== 'DISK_WRITE_OPS':
206 metric_name
= 'DiskWriteOps'
209 elif metric_name
== 'DISK_READ_BYTES':
210 metric_name
= 'DiskReadBytes'
213 elif metric_name
== 'DISK_WRITE_BYTES':
214 metric_name
= 'DiskWriteBytes'
217 elif metric_name
== 'PACKETS_RECEIVED':
218 metric_name
= 'NetworkPacketsIn'
221 elif metric_name
== 'PACKETS_SENT':
222 metric_name
= 'NetworkPacketsOut'
227 metric_status
= False
228 check_resp
['metric_name'] = metric_name
231 if metric_status
== True:
232 check_resp
['status'] = True
235 except Exception as e
:
236 log
.error("Error in Plugin Inputs %s",str(e
))
237 #---------------------------------------------------------------------------------------------------------------------------