90b7480b89dcc15e1f9698021b856d687aa3b9db
[osm/MON.git] / osm_mon / plugins / vRealiseOps / plugin_receiver.py
1 # -*- coding: utf-8 -*-
2
3 ##
4 # Copyright 2016-2017 VMware Inc.
5 # This file is part of ETSI OSM
6 # All Rights Reserved.
7 #
8 # Licensed under the Apache License, Version 2.0 (the "License"); you may
9 # not use this file except in compliance with the License. You may obtain
10 # a copy of the License at
11 #
12 # http://www.apache.org/licenses/LICENSE-2.0
13 #
14 # Unless required by applicable law or agreed to in writing, software
15 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
16 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
17 # License for the specific language governing permissions and limitations
18 # under the License.
19 #
20 # For those usages not covered by the Apache License, Version 2.0 please
21 # contact: osslegalrouting@vmware.com
22 ##
23
24 """
25 Montoring plugin receiver that consumes the request messages &
26 responds using producer for vROPs
27 """
28
29 import sys
30 from mon_plugin_vrops import MonPlugin
31 from kafka_consumer_vrops import vROP_KafkaConsumer
32 #Core producer
33 sys.path.append("../../core/message_bus")
34 from producer import KafkaProducer
35 #from core.message_bus.producer import KafkaProducer
36 import json
37 import logging
38 import traceback
39 import os
40 from xml.etree import ElementTree as XmlElementTree
41
42 req_config_params = ('vrops_site', 'vrops_user', 'vrops_password',
43 'vcloud-site','admin_username','admin_password',
44 'vcenter_ip','vcenter_port','vcenter_user','vcenter_password',
45 'vim_tenant_name','orgname','tenant_id')
46 MODULE_DIR = os.path.dirname(__file__)
47 CONFIG_FILE_NAME = 'vrops_config.xml'
48 CONFIG_FILE_PATH = os.path.join(MODULE_DIR, CONFIG_FILE_NAME)
49
50 def set_logger():
51 """Set Logger
52 """
53 BASE_DIR = os.path.dirname(os.path.dirname(__file__))
54 logger = logging.getLogger()
55 formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
56 handler = logging.FileHandler(os.path.join(BASE_DIR,"mon_vrops_log.log"))
57 handler.setFormatter(formatter)
58 logger.addHandler(handler)
59
60
61 class PluginReceiver():
62 """MON Plugin receiver receiving request messages & responding using producer for vROPs
63 telemetry plugin
64 """
65 def __init__(self):
66 """Constructor of PluginReceiver
67 """
68
69 topics = ['alarm_request', 'metric_request', 'access_credentials']
70
71 self.logger = logging.getLogger('PluginReceiver')
72 self.logger.setLevel(logging.DEBUG)
73
74 #To Do - Add broker uri
75 broker_uri = None
76 #self.mon_plugin = MonPlugin()
77 self.consumer = vROP_KafkaConsumer(topics, broker_uri)
78 #Core producer
79 self.producer_alarms = KafkaProducer('alarm_response')
80 self.producer_metrics = KafkaProducer('metric_response')
81 self.producer_access_credentials = KafkaProducer('vim_access_credentials_response')
82
83
84 def consume(self):
85 """Consume the message, act on it & respond
86 """
87 try:
88 for message in self.consumer.vrops_consumer:
89 vim_type = None
90 self.logger.info("Message received:\nTopic={}:{}:{}:\nKey={}\nValue={}"\
91 .format(message.topic, message.partition, message.offset, message.key, message.value))
92 message_values = json.loads(message.value)
93 if message_values.has_key('vim_type') and message_values['vim_type'] is not None:
94 vim_type = message_values['vim_type'].lower()
95 if vim_type == 'vmware':
96 self.logger.info("Action required for: {}".format(message.topic))
97 if message.topic == 'alarm_request':
98 if message.key == "create_alarm_request":
99 config_alarm_info = json.loads(message.value)
100 alarm_uuid = self.create_alarm(config_alarm_info['alarm_create_request'])
101 self.logger.info("Alarm created with alarm uuid: {}".format(alarm_uuid))
102 #Publish message using producer
103 self.publish_create_alarm_status(alarm_uuid, config_alarm_info)
104 elif message.key == "update_alarm_request":
105 update_alarm_info = json.loads(message.value)
106 alarm_uuid = self.update_alarm(update_alarm_info['alarm_update_request'])
107 self.logger.info("Alarm defination updated : alarm uuid: {}".format(alarm_uuid))
108 #Publish message using producer
109 self.publish_update_alarm_status(alarm_uuid, update_alarm_info)
110 elif message.key == "delete_alarm_request":
111 delete_alarm_info = json.loads(message.value)
112 alarm_uuid = self.delete_alarm(delete_alarm_info['alarm_delete_request'])
113 self.logger.info("Alarm defination deleted : alarm uuid: {}".format(alarm_uuid))
114 #Publish message using producer
115 self.publish_delete_alarm_status(alarm_uuid, delete_alarm_info)
116 elif message.key == "list_alarm_request":
117 request_input = json.loads(message.value)
118 triggered_alarm_list = self.list_alarms(request_input['alarm_list_request'])
119 #Publish message using producer
120 self.publish_list_alarm_response(triggered_alarm_list, request_input)
121 elif message.topic == 'metric_request':
122 if message.key == "read_metric_data_request":
123 metric_request_info = json.loads(message.value)
124 mon_plugin_obj = MonPlugin()
125 metrics_data = mon_plugin_obj.get_metrics_data(metric_request_info)
126 self.logger.info("Collected Metrics Data: {}".format(metrics_data))
127 #Publish message using producer
128 self.publish_metrics_data_status(metrics_data)
129 elif message.key == "create_metric_request":
130 metric_info = json.loads(message.value)
131 metric_status = self.verify_metric(metric_info['metric_create'])
132 #Publish message using producer
133 self.publish_create_metric_response(metric_info, metric_status)
134 elif message.key == "update_metric_request":
135 metric_info = json.loads(message.value)
136 metric_status = self.verify_metric(metric_info['metric_create'])
137 #Publish message using producer
138 self.publish_update_metric_response(metric_info, metric_status)
139 elif message.key == "delete_metric_request":
140 metric_info = json.loads(message.value)
141 #Deleting Metric Data is not allowed. Publish status as False
142 self.logger.warn("Deleting Metric is not allowed: {}".format(metric_info['metric_name']))
143 #Publish message using producer
144 self.publish_delete_metric_response(metric_info)
145 elif message.topic == 'access_credentials':
146 if message.key == "vim_access_credentials":
147 access_info = json.loads(message.value)
148 access_update_status = self.update_access_credentials(access_info['access_config'])
149 self.publish_access_update_response(access_update_status, access_info)
150
151 except:
152 self.logger.error("Exception in receiver: {}".format(traceback.format_exc()))
153
154
155 def create_alarm(self, config_alarm_info):
156 """Create alarm using vROPs plugin
157 """
158 mon_plugin = MonPlugin()
159 plugin_uuid = mon_plugin.configure_rest_plugin()
160 alarm_uuid = mon_plugin.configure_alarm(config_alarm_info)
161 return alarm_uuid
162
163 def publish_create_alarm_status(self, alarm_uuid, config_alarm_info):
164 """Publish create alarm status using producer
165 """
166 topic = 'alarm_response'
167 msg_key = 'create_alarm_response'
168 response_msg = {"schema_version":1.0,
169 "schema_type":"create_alarm_response",
170 "alarm_create_response":
171 {"correlation_id":config_alarm_info["alarm_create_request"]["correlation_id"],
172 "alarm_uuid":alarm_uuid,
173 "status": True if alarm_uuid else False
174 }
175 }
176 self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
177 .format(topic, msg_key, response_msg))
178 #Core producer
179 self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
180
181 def update_alarm(self, update_alarm_info):
182 """Updare already created alarm
183 """
184 mon_plugin = MonPlugin()
185 alarm_uuid = mon_plugin.update_alarm_configuration(update_alarm_info)
186 return alarm_uuid
187
188 def publish_update_alarm_status(self, alarm_uuid, update_alarm_info):
189 """Publish update alarm status requests using producer
190 """
191 topic = 'alarm_response'
192 msg_key = 'update_alarm_response'
193 response_msg = {"schema_version":1.0,
194 "schema_type":"update_alarm_response",
195 "alarm_update_response":
196 {"correlation_id":update_alarm_info["alarm_update_request"]["correlation_id"],
197 "alarm_uuid":alarm_uuid,
198 "status": True if alarm_uuid else False
199 }
200 }
201 self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
202 .format(topic, msg_key, response_msg))
203 #Core producer
204 self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
205
206 def delete_alarm(self, delete_alarm_info):
207 """Delete alarm configuration
208 """
209 mon_plugin = MonPlugin()
210 alarm_uuid = mon_plugin.delete_alarm_configuration(delete_alarm_info)
211 return alarm_uuid
212
213 def publish_delete_alarm_status(self, alarm_uuid, delete_alarm_info):
214 """Publish update alarm status requests using producer
215 """
216 topic = 'alarm_response'
217 msg_key = 'delete_alarm_response'
218 response_msg = {"schema_version":1.0,
219 "schema_type":"delete_alarm_response",
220 "alarm_deletion_response":
221 {"correlation_id":delete_alarm_info["alarm_delete_request"]["correlation_id"],
222 "alarm_uuid":alarm_uuid,
223 "status": True if alarm_uuid else False
224 }
225 }
226 self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
227 .format(topic, msg_key, response_msg))
228 #Core producer
229 self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
230
231
232 def publish_metrics_data_status(self, metrics_data):
233 """Publish the requested metric data using producer
234 """
235 topic = 'metric_response'
236 msg_key = 'read_metric_data_response'
237 self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
238 .format(topic, msg_key, metrics_data))
239 #Core producer
240 self.producer_metrics.publish(key=msg_key, value=json.dumps(metrics_data), topic=topic)
241
242
243 def verify_metric(self, metric_info):
244 """Verify if metric is supported or not
245 """
246 mon_plugin = MonPlugin()
247 metric_key_status = mon_plugin.verify_metric_support(metric_info)
248 return metric_key_status
249
250 def publish_create_metric_response(self, metric_info, metric_status):
251 """Publish create metric response
252 """
253 topic = 'metric_response'
254 msg_key = 'create_metric_response'
255 response_msg = {"schema_version":1.0,
256 "schema_type":"create_metric_response",
257 "correlation_id":metric_info['correlation_id'],
258 "metric_create_response":
259 {
260 "metric_uuid":0,
261 "resource_uuid":metric_info['metric_create']['resource_uuid'],
262 "status":metric_status
263 }
264 }
265 self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
266 .format(topic, msg_key, response_msg))
267 #Core producer
268 self.producer_metrics.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
269
270 def publish_update_metric_response(self, metric_info, metric_status):
271 """Publish update metric response
272 """
273 topic = 'metric_response'
274 msg_key = 'update_metric_response'
275 response_msg = {"schema_version":1.0,
276 "schema_type":"metric_update_response",
277 "correlation_id":metric_info['correlation_id'],
278 "metric_update_response":
279 {
280 "metric_uuid":0,
281 "resource_uuid":metric_info['metric_create']['resource_uuid'],
282 "status":metric_status
283 }
284 }
285 self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
286 .format(topic, msg_key, response_msg))
287 #Core producer
288 self.producer_metrics.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
289
290 def publish_delete_metric_response(self, metric_info):
291 """
292 """
293 topic = 'metric_response'
294 msg_key = 'delete_metric_response'
295 response_msg = {"schema_version":1.0,
296 "schema_type":"delete_metric_response",
297 "correlation_id":metric_info['correlation_id'],
298 "metric_name":metric_info['metric_name'],
299 "metric_uuid":0,
300 "resource_uuid":metric_info['resource_uuid'],
301 "tenant_uuid":metric_info['tenant_uuid'],
302 "status":False
303 }
304 self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
305 .format(topic, msg_key, response_msg))
306 #Core producer
307 self.producer_metrics.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
308
309 def list_alarms(self, list_alarm_input):
310 """Collect list of triggered alarms based on input
311 """
312 mon_plugin = MonPlugin()
313 triggered_alarms = mon_plugin.get_triggered_alarms_list(list_alarm_input)
314 return triggered_alarms
315
316
317 def publish_list_alarm_response(self, triggered_alarm_list, list_alarm_input):
318 """Publish list of triggered alarms
319 """
320 topic = 'alarm_response'
321 msg_key = 'list_alarm_response'
322 response_msg = {"schema_version":1.0,
323 "schema_type":"list_alarm_response",
324 "correlation_id":list_alarm_input['alarm_list_request']['correlation_id'],
325 #"resource_uuid":list_alarm_input['alarm_list_request']['resource_uuid'],
326 "list_alarm_resp":triggered_alarm_list
327 }
328 self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
329 .format(topic, msg_key, response_msg))
330 #Core producer
331 self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
332
333
334 def update_access_credentials(self, access_info):
335 """Verify if all the required access config params are provided and
336 updates access config in default vrops config file
337 """
338 update_status = False
339 wr_status = False
340 #Check if all the required config params are passed in request
341 if not all (keys in access_info for keys in req_config_params):
342 self.logger.debug("All required Access Config Parameters not provided")
343 self.logger.debug("List of required Access Config Parameters: {}".format(req_config_params))
344 self.logger.debug("List of given Access Config Parameters: {}".format(access_info))
345 return update_status
346
347 wr_status = self.write_access_config(access_info)
348 return wr_status #True/False
349
350 def write_access_config(self, access_info):
351 """Write access configuration to vROPs config file.
352 """
353 wr_status = False
354 try:
355 tree = XmlElementTree.parse(CONFIG_FILE_PATH)
356 root = tree.getroot()
357 alarmParams = {}
358 for config in root:
359 if config.tag == 'Access_Config':
360 for param in config:
361 for key,val in access_info.iteritems():
362 if param.tag == key:
363 #print param.tag, val
364 param.text = val
365
366 tree.write(CONFIG_FILE_PATH)
367 wr_status = True
368 except Exception as exp:
369 self.logger.warn("Failed to update Access Config Parameters: {}".format(exp))
370
371 return wr_status
372
373
374 def publish_access_update_response(self, access_update_status, access_info_req):
375 """Publish access update response
376 """
377 topic = 'access_credentials'
378 msg_key = 'vim_access_credentials_response'
379 response_msg = {"schema_version":1.0,
380 "schema_type":"vim_access_credentials_response",
381 "correlation_id":access_info_req['access_config']['correlation_id'],
382 "status":access_update_status
383 }
384 self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
385 .format(topic, msg_key, response_msg))
386 #Core Add producer
387 self.producer_access_credentials.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
388
389 def main():
390 #log.basicConfig(filename='mon_vrops_log.log',level=log.DEBUG)
391 set_logger()
392 plugin_rcvr = PluginReceiver()
393 plugin_rcvr.consume()
394
395 if __name__ == "__main__":
396 main()
397