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 log
= logging
.getLogger(__name__
)
41 class plugin_alarms():
42 """Receives Alarm info from MetricAlarm and connects with the consumer/producer"""
44 self
.metricAlarm
= MetricAlarm()
45 self
.metric
= Metrics()
46 self
.producer
= KafkaProducer('')
47 #---------------------------------------------------------------------------------------------------------------------------
48 def configure_alarm(self
,alarm_info
):
49 alarm_id
= self
.metricAlarm
.config_alarm(self
.cloudwatch_conn
,alarm_info
)
51 #---------------------------------------------------------------------------------------------------------------------------
52 def update_alarm_configuration(self
,test
):
53 alarm_id
= self
.metricAlarm
.update_alarm(self
.cloudwatch_conn
,test
)
55 #---------------------------------------------------------------------------------------------------------------------------
56 def delete_alarm(self
,alarm_id
):
57 return self
.metricAlarm
.delete_Alarm(self
.cloudwatch_conn
,alarm_id
)
58 #---------------------------------------------------------------------------------------------------------------------------
59 def get_alarms_list(self
,instance_id
):
60 return self
.metricAlarm
.alarms_list(self
.cloudwatch_conn
,instance_id
)
61 #---------------------------------------------------------------------------------------------------------------------------
62 def get_ack_details(self
,ack_info
):
63 return self
.metricAlarm
.alarm_details(self
.cloudwatch_conn
,ack_info
)
64 #---------------------------------------------------------------------------------------------------------------------------
65 def get_metrics_data(self
,metric_name
,period
,instance_id
):
66 return self
.metric
.metricsData(self
.cloudwatch_conn
,metric_name
,period
,instance_id
)
67 #---------------------------------------------------------------------------------------------------------------------------
69 def alarm_calls(self
,message
,aws_conn
):
70 """Gets the message from the common consumer"""
72 self
.cloudwatch_conn
= aws_conn
['cloudwatch_connection']
73 self
.ec2_conn
= aws_conn
['ec2_connection']
75 log
.info("Action required against: %s" % (message
.topic
))
76 alarm_info
= json
.loads(message
.value
)
78 if message
.key
== "create_alarm_request":
79 alarm_inner_dict
= alarm_info
['alarm_create_request']
80 metric_status
= self
.check_metric(alarm_inner_dict
['metric_name'])
82 if self
.check_resource(alarm_inner_dict
['resource_uuid']) == True and metric_status
['status'] == True:
83 log
.debug ("Resource and Metrics exists")
85 alarm_info
['alarm_create_request']['metric_name'] = metric_status
['metric_name']
86 #Generate a valid response message, send via producer
87 config_resp
= self
.configure_alarm(alarm_info
) #alarm_info = message.value
89 if config_resp
== None:
90 log
.debug("Alarm Already exists")
91 payload
= json
.dumps(config_resp
)
92 file = open('../../core/models/create_alarm_resp.json','wb').write((payload
))
93 self
.producer
.create_alarm_response(key
='create_alarm_response',message
=payload
,topic
= 'alarm_response')
96 payload
= json
.dumps(config_resp
)
97 file = open('../../core/models/create_alarm_resp.json','wb').write((payload
))
98 self
.producer
.create_alarm_response(key
='create_alarm_response',message
=payload
,topic
= 'alarm_response')
99 log
.info("New alarm created with alarm info: %s", config_resp
)
102 log
.error("Resource ID doesn't exists")
104 elif message
.key
== "acknowledge_alarm":
105 alarm_inner_dict
= alarm_info
['ack_details']
107 if self
.check_resource(alarm_inner_dict
['resource_uuid']) == True:
108 alarm_info
= json
.loads(message
.value
)
109 #Generate a valid response message, send via producer
110 ack_details
= self
.get_ack_details(alarm_info
)
111 payload
= json
.dumps(ack_details
)
112 file = open('../../core/models/notify_alarm.json','wb').write((payload
))
113 self
.producer
.notify_alarm(key
='notify_alarm',message
=payload
,topic
= 'alarm_response')
114 log
.info("Acknowledge sent: %s", ack_details
)
117 log
.error("Resource ID is Incorrect")
120 elif message
.key
== "update_alarm_request":
121 alarm_inner_dict
= alarm_info
['alarm_update_request']
122 metric_status
= self
.check_metric(alarm_inner_dict
['metric_name'])
124 if metric_status
['status'] == True:
125 log
.debug ("Resource and Metrics exists")
126 alarm_info
['alarm_update_request']['metric_name'] = metric_status
['metric_name']
127 #Generate a valid response message, send via producer
128 update_resp
= self
.update_alarm_configuration(alarm_info
)
130 if update_resp
== None:
131 payload
= json
.dumps(update_resp
)
132 file = open('../../core/models/update_alarm_resp.json','wb').write((payload
))
133 self
.producer
.update_alarm_response(key
='update_alarm_response',message
=payload
,topic
= 'alarm_response')
134 log
.debug("Alarm Already exists")
137 payload
= json
.dumps(update_resp
)
138 file = open('../../core/models/update_alarm_resp.json','wb').write((payload
))
139 self
.producer
.update_alarm_response(key
='update_alarm_response',message
=payload
,topic
= 'alarm_response')
140 log
.info("Alarm Updated with alarm info: %s", update_resp
)
143 log
.info ("Metric Not Supported")
146 elif message
.key
== "delete_alarm_request":
147 del_info
= json
.loads(message
.value
)
148 #Generate a valid response message, send via producer
149 del_resp
= self
.delete_alarm(del_info
)
150 payload
= json
.dumps(del_resp
)
151 file = open('../../core/models/delete_alarm_resp.json','wb').write((payload
))
152 self
.producer
.delete_alarm_response(key
='delete_alarm_response',message
=payload
,topic
= 'alarm_response')
153 log
.info("Alarm Deleted with alarm info: %s", del_resp
)
156 elif message
.key
== "alarm_list_request":
157 alarm_inner_dict
= alarm_info
['alarm_list_request']
159 if self
.check_resource(alarm_inner_dict
['resource_uuid']) == True or alarm_inner_dict
['resource_uuid'] == "":
160 #Generate a valid response message, send via producer
161 list_resp
= self
.get_alarms_list(alarm_info
)#['alarm_names']
162 payload
= json
.dumps(list_resp
)
163 file = open('../../core/models/list_alarm_resp.json','wb').write((payload
))
164 self
.producer
.list_alarm_response(key
='list_alarm_response',message
=payload
,topic
= 'alarm_response')
167 log
.error("Resource ID is Incorrect")
170 log
.debug("Unknown key, no action will be performed")
172 except Exception as e
:
173 log
.error("Message retrieval exception: %s", str(e
))
174 #---------------------------------------------------------------------------------------------------------------------------
175 def check_resource(self
,resource_uuid
):
176 '''Finding Resource with the resource_uuid'''
179 instances
= self
.ec2_conn
.get_all_instance_status()
182 for instance_id
in instances
:
183 instance_id
= str(instance_id
).split(':')[1]
185 if instance_id
== resource_uuid
:
186 check_resp
['resource_uuid'] = resource_uuid
190 except Exception as e
:
191 log
.error("Error in Plugin Inputs %s",str(e
))
192 #---------------------------------------------------------------------------------------------------------------------------
193 def check_metric(self
,metric_name
):
194 ''' Checking whether the metric is supported by AWS '''
199 if metric_name
== 'CPU_UTILIZATION':
200 metric_name
= 'CPUUtilization'
203 elif metric_name
== 'DISK_READ_OPS':
204 metric_name
= 'DiskReadOps'
207 elif metric_name
== 'DISK_WRITE_OPS':
208 metric_name
= 'DiskWriteOps'
211 elif metric_name
== 'DISK_READ_BYTES':
212 metric_name
= 'DiskReadBytes'
215 elif metric_name
== 'DISK_WRITE_BYTES':
216 metric_name
= 'DiskWriteBytes'
219 elif metric_name
== 'PACKETS_RECEIVED':
220 metric_name
= 'NetworkPacketsIn'
223 elif metric_name
== 'PACKETS_SENT':
224 metric_name
= 'NetworkPacketsOut'
229 metric_status
= False
230 check_resp
['metric_name'] = metric_name
233 if metric_status
== True:
234 check_resp
['status'] = True
237 except Exception as e
:
238 log
.error("Error in Plugin Inputs %s",str(e
))
239 #---------------------------------------------------------------------------------------------------------------------------