1 # -*- coding: utf-8 -*-
4 # Copyright 2016-2017 VMware Inc.
5 # This file is part of ETSI OSM
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
12 # http://www.apache.org/licenses/LICENSE-2.0
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
20 # For those usages not covered by the Apache License, Version 2.0 please
21 # contact: osslegalrouting@vmware.com
25 Monitoring plugin receiver that consumes the request messages &
26 responds using producer for vROPs
37 from osm_mon
.plugins
.vRealiseOps
.mon_plugin_vrops
import MonPlugin
39 sys
.path
.append(os
.path
.join(os
.path
.dirname(os
.path
.abspath(__file__
)), '..', '..', '..'))
40 from osm_mon
.core
.message_bus
.producer
import KafkaProducer
42 from osm_mon
.core
.auth
import AuthManager
44 #from core.message_bus.producer import KafkaProducer
45 from xml
.etree
import ElementTree
as XmlElementTree
48 schema_version
= "1.0"
49 req_config_params
= ('vrops_site', 'vrops_user', 'vrops_password',
50 'vcloud-site','admin_username','admin_password',
51 'vcenter_ip','vcenter_port','vcenter_user','vcenter_password',
52 'vim_tenant_name','orgname')
53 MODULE_DIR
= os
.path
.dirname(__file__
)
54 CONFIG_FILE_NAME
= 'vrops_config.xml'
55 CONFIG_FILE_PATH
= os
.path
.join(MODULE_DIR
, CONFIG_FILE_NAME
)
60 BASE_DIR
= os
.path
.dirname(os
.path
.dirname(__file__
))
61 logger
= logging
.getLogger()
62 formatter
= logging
.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
63 handler
= logging
.FileHandler(os
.path
.join(BASE_DIR
,"mon_vrops_log.log"))
64 handler
.setFormatter(formatter
)
65 logger
.addHandler(handler
)
68 class PluginReceiver():
69 """MON Plugin receiver receiving request messages & responding using producer for vROPs
73 """Constructor of PluginReceiver
77 self
.logger
= logging
.getLogger('PluginReceiver')
78 self
.logger
.setLevel(logging
.DEBUG
)
82 self
.producer_alarms
= KafkaProducer('alarm_response')
83 self
.producer_metrics
= KafkaProducer('metric_response')
84 self
.producer_access_credentials
= KafkaProducer('vim_access_credentials_response')
87 def consume(self
, message
, vim_uuid
):
88 """Consume the message, act on it & respond
91 self
.logger
.info("Message received for VIM: {} :\nMessage Topic={}:{}:{}:\n"\
92 "Message Key={}\nMessage Value={}"\
93 .format(vim_uuid
, message
.topic
, message
.partition
, message
.offset
,\
94 message
.key
, message
.value
))
95 message_values
= json
.loads(message
.value
)
96 #Adding vim_uuid to message
97 message_values
['vim_uuid'] = vim_uuid
98 self
.logger
.info("Action required for: {}".format(message
.topic
))
99 if message
.topic
== 'alarm_request':
100 if message
.key
== "create_alarm_request":
101 config_alarm_info
= message_values
102 alarm_uuid
= self
.create_alarm(config_alarm_info
)
103 self
.logger
.info("Alarm created with alarm uuid: {}".format(alarm_uuid
))
104 #Publish message using producer
105 self
.publish_create_alarm_status(alarm_uuid
, config_alarm_info
)
106 elif message
.key
== "update_alarm_request":
107 update_alarm_info
= message_values
108 alarm_uuid
= self
.update_alarm(update_alarm_info
)
109 self
.logger
.info("Alarm definition updated : alarm uuid: {}".format(alarm_uuid
))
110 #Publish message using producer
111 self
.publish_update_alarm_status(alarm_uuid
, update_alarm_info
)
112 elif message
.key
== "delete_alarm_request":
113 delete_alarm_info
= message_values
114 alarm_uuid
= self
.delete_alarm(delete_alarm_info
)
115 self
.logger
.info("Alarm definition deleted : alarm uuid: {}".format(alarm_uuid
))
116 #Publish message using producer
117 self
.publish_delete_alarm_status(alarm_uuid
, delete_alarm_info
)
118 elif message
.key
== "list_alarm_request":
119 request_input
= message_values
120 triggered_alarm_list
= self
.list_alarms(request_input
)
121 #Publish message using producer
122 self
.publish_list_alarm_response(triggered_alarm_list
, request_input
)
123 elif message
.topic
== 'metric_request':
124 if message
.key
== "read_metric_data_request":
125 metric_request_info
= message_values
127 access_config
= self
.get_vim_access_config(metric_request_info
['vim_uuid'])
128 mon_plugin_obj
= MonPlugin(access_config
)
129 metrics_data
= mon_plugin_obj
.get_metrics_data(metric_request_info
)
130 self
.logger
.info("Collected Metrics Data: {}".format(metrics_data
))
131 #Publish message using producer
132 self
.publish_metrics_data_status(metrics_data
)
133 elif message
.key
== "create_metric_request":
134 metric_info
= message_values
135 metric_status
= self
.verify_metric(metric_info
)
136 #Publish message using producer
137 self
.publish_create_metric_response(metric_info
, metric_status
)
138 elif message
.key
== "update_metric_request":
139 metric_info
= message_values
140 metric_status
= self
.verify_metric(metric_info
)
141 #Publish message using producer
142 self
.publish_update_metric_response(metric_info
, metric_status
)
143 elif message
.key
== "delete_metric_request":
144 metric_info
= message_values
145 #Deleting Metric Data is not allowed. Publish status as False
146 self
.logger
.warning("Deleting Metric is not allowed by VMware vROPs plugin: {}"\
147 .format(metric_info
['metric_name']))
148 #Publish message using producer
149 self
.publish_delete_metric_response(metric_info
)
150 elif message
.topic
== 'access_credentials':
151 if message
.key
== "vim_access_credentials":
152 access_info
= message_values
153 access_update_status
= self
.update_access_credentials(access_info
['access_config'])
154 self
.publish_access_update_response(access_update_status
, access_info
)
157 self
.logger
.error("Exception in vROPs plugin receiver: {}".format(traceback
.format_exc()))
160 def create_alarm(self
, config_alarm_info
):
161 """Create alarm using vROPs plugin
164 access_config
= self
.get_vim_access_config(config_alarm_info
['vim_uuid'])
165 mon_plugin
= MonPlugin(access_config
)
166 plugin_uuid
= mon_plugin
.configure_rest_plugin()
167 alarm_uuid
= mon_plugin
.configure_alarm(config_alarm_info
['alarm_create_request'])
170 def publish_create_alarm_status(self
, alarm_uuid
, config_alarm_info
):
171 """Publish create alarm status using producer
173 topic
= 'alarm_response'
174 msg_key
= 'create_alarm_response'
175 response_msg
= {"schema_version":schema_version
,
176 "schema_type":"create_alarm_response",
177 "vim_uuid":config_alarm_info
["vim_uuid"],
178 "alarm_create_response":
179 {"correlation_id":config_alarm_info
["alarm_create_request"]["correlation_id"],
180 "alarm_uuid":alarm_uuid
,
181 "status": True if alarm_uuid
else False
184 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
185 .format(topic
, msg_key
, response_msg
))
187 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
189 def update_alarm(self
, update_alarm_info
):
190 """Update already created alarm
193 access_config
= self
.get_vim_access_config(update_alarm_info
['vim_uuid'])
194 mon_plugin
= MonPlugin(access_config
)
195 alarm_uuid
= mon_plugin
.update_alarm_configuration(update_alarm_info
['alarm_update_request'])
198 def publish_update_alarm_status(self
, alarm_uuid
, update_alarm_info
):
199 """Publish update alarm status requests using producer
201 topic
= 'alarm_response'
202 msg_key
= 'update_alarm_response'
203 response_msg
= {"schema_version":schema_version
,
204 "schema_type":"update_alarm_response",
205 "vim_uuid":update_alarm_info
["vim_uuid"],
206 "alarm_update_response":
207 {"correlation_id":update_alarm_info
["alarm_update_request"]["correlation_id"],
208 "alarm_uuid":update_alarm_info
["alarm_update_request"]["alarm_uuid"] \
209 if update_alarm_info
["alarm_update_request"].get('alarm_uuid') is not None else None,
210 "status": True if alarm_uuid
else False
213 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
214 .format(topic
, msg_key
, response_msg
))
216 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
218 def delete_alarm(self
, delete_alarm_info
):
219 """Delete alarm configuration
222 access_config
= self
.get_vim_access_config(delete_alarm_info
['vim_uuid'])
223 mon_plugin
= MonPlugin(access_config
)
224 alarm_uuid
= mon_plugin
.delete_alarm_configuration(delete_alarm_info
['alarm_delete_request'])
227 def publish_delete_alarm_status(self
, alarm_uuid
, delete_alarm_info
):
228 """Publish update alarm status requests using producer
230 topic
= 'alarm_response'
231 msg_key
= 'delete_alarm_response'
232 response_msg
= {"schema_version":schema_version
,
233 "schema_type":"delete_alarm_response",
234 "vim_uuid":delete_alarm_info
['vim_uuid'],
235 "alarm_deletion_response":
236 {"correlation_id":delete_alarm_info
["alarm_delete_request"]["correlation_id"],
237 "alarm_uuid":delete_alarm_info
["alarm_delete_request"]["alarm_uuid"],
238 "status": True if alarm_uuid
else False
241 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
242 .format(topic
, msg_key
, response_msg
))
244 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
247 def publish_metrics_data_status(self
, metrics_data
):
248 """Publish the requested metric data using producer
250 topic
= 'metric_response'
251 msg_key
= 'read_metric_data_response'
252 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
253 .format(topic
, msg_key
, metrics_data
))
255 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(metrics_data
), topic
=topic
)
258 def verify_metric(self
, metric_info
):
259 """Verify if metric is supported or not
262 access_config
= self
.get_vim_access_config(metric_info
['vim_uuid'])
263 mon_plugin
= MonPlugin(access_config
)
264 if 'metric_create_request' in metric_info
:
265 metric_key_status
= mon_plugin
.verify_metric_support(metric_info
['metric_create_request'])
267 metric_key_status
= mon_plugin
.verify_metric_support(metric_info
['metric_update_request'])
268 return metric_key_status
270 def publish_create_metric_response(self
, metric_info
, metric_status
):
271 """Publish create metric response
273 topic
= 'metric_response'
274 msg_key
= 'create_metric_response'
275 response_msg
= {"schema_version":schema_version
,
276 "schema_type":"create_metric_response",
277 ##"vim_uuid":metric_info['vim_uuid'],
278 ##"correlation_id":metric_info['correlation_id'],
279 "metric_create_response":
282 ##"resource_uuid":metric_info['metric_create']['resource_uuid'],
283 ##"vim_uuid":metric_info['vim_uuid'], #May be required. TODO - Confirm
284 "correlation_id":metric_info
['correlation_id'],
285 "status":metric_status
288 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
289 .format(topic
, msg_key
, response_msg
))
291 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
293 def publish_update_metric_response(self
, metric_info
, metric_status
):
294 """Publish update metric response
296 topic
= 'metric_response'
297 msg_key
= 'update_metric_response'
298 response_msg
= {"schema_version":schema_version
,
299 "schema_type":"metric_update_response",
300 "vim_uuid":metric_info
['vim_uuid'],
301 "metric_update_response":
304 "correlation_id":metric_info
['correlation_id'],
305 "resource_uuid":metric_info
['metric_create']['resource_uuid'],
306 "status":metric_status
309 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
310 .format(topic
, msg_key
, response_msg
))
312 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
314 def publish_delete_metric_response(self
, metric_info
):
315 """Publish delete metric response
317 topic
= 'metric_response'
318 msg_key
= 'delete_metric_response'
319 if 'tenant_uuid' in metric_info
and metric_info
['tenant_uuid'] is not None:
320 tenant_uuid
= metric_info
['tenant_uuid']
324 response_msg
= {"schema_version":schema_version
,
325 "schema_type":"delete_metric_response",
326 "vim_uuid":metric_info
['vim_uuid'],
327 "correlation_id":metric_info
['correlation_id'],
328 "metric_name":metric_info
['metric_name'],
330 "resource_uuid":metric_info
['resource_uuid'],
331 "tenant_uuid":tenant_uuid
,
334 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
335 .format(topic
, msg_key
, response_msg
))
337 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
339 def list_alarms(self
, list_alarm_input
):
340 """Collect list of triggered alarms based on input
343 access_config
= self
.get_vim_access_config(list_alarm_input
['vim_uuid'])
344 mon_plugin
= MonPlugin(access_config
)
345 triggered_alarms
= mon_plugin
.get_triggered_alarms_list(list_alarm_input
['alarm_list_request'])
346 return triggered_alarms
349 def publish_list_alarm_response(self
, triggered_alarm_list
, list_alarm_input
):
350 """Publish list of triggered alarms
352 topic
= 'alarm_response'
353 msg_key
= 'list_alarm_response'
354 response_msg
= {"schema_version":schema_version
,
355 "schema_type":"list_alarm_response",
357 "vim_uuid":list_alarm_input
['vim_uuid'],
358 "correlation_id":list_alarm_input
['alarm_list_request']['correlation_id'],
359 "list_alarm_response":triggered_alarm_list
361 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
362 .format(topic
, msg_key
, response_msg
))
364 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
367 def update_access_credentials(self
, access_info
):
368 """Verify if all the required access config params are provided and
369 updates access config in default vrops config file
371 update_status
= False
373 #Check if all the required config params are passed in request
374 if not all (keys
in access_info
for keys
in req_config_params
):
375 self
.logger
.debug("All required Access Config Parameters not provided")
376 self
.logger
.debug("List of required Access Config Parameters: {}".format(req_config_params
))
377 self
.logger
.debug("List of given Access Config Parameters: {}".format(access_info
))
380 wr_status
= self
.write_access_config(access_info
)
381 return wr_status
#True/False
383 def write_access_config(self
, access_info
):
384 """Write access configuration to vROPs config file.
388 tree
= XmlElementTree
.parse(CONFIG_FILE_PATH
)
389 root
= tree
.getroot()
392 if config
.tag
== 'Access_Config':
394 for key
,val
in six
.iteritems(access_info
):
396 #print param.tag, val
399 tree
.write(CONFIG_FILE_PATH
)
401 except Exception as exp
:
402 self
.logger
.warning("Failed to update Access Config Parameters: {}".format(exp
))
407 def publish_access_update_response(self
, access_update_status
, access_info_req
):
408 """Publish access update response
410 topic
= 'access_credentials'
411 msg_key
= 'vim_access_credentials_response'
412 response_msg
= {"schema_version":schema_version
,
413 "schema_type":"vim_access_credentials_response",
414 "correlation_id":access_info_req
['access_config']['correlation_id'],
415 "status":access_update_status
417 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
418 .format(topic
, msg_key
, response_msg
))
420 self
.producer_access_credentials
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
423 def get_vim_access_config(self
, vim_uuid
):
424 """Get VIM access configuration & account details from path: VIM_ACCOUNTS_FILE_PATH
426 vim_account_details
= None
428 auth_manager
= AuthManager()
429 vim_account_details
= auth_manager
.get_credentials(vim_uuid
)
432 if vim_account_details
is not None:
433 vim_account
['name'] = vim_account_details
.name
434 vim_account
['vim_tenant_name'] = vim_account_details
.tenant_name
435 vim_account
['vim_type'] = vim_account_details
.type
436 vim_account
['vim_url'] = vim_account_details
.url
437 vim_account
['org_user'] = vim_account_details
.user
438 vim_account
['org_password'] = vim_account_details
.password
439 vim_account
['vim_uuid'] = vim_account_details
.uuid
441 vim_config
= json
.loads(vim_account_details
.config
)
442 vim_account
['admin_username'] = vim_config
['admin_username']
443 vim_account
['admin_password'] = vim_config
['admin_password']
444 vim_account
['vrops_site'] = vim_config
['vrops_site']
445 vim_account
['vrops_user'] = vim_config
['vrops_user']
446 vim_account
['vrops_password'] = vim_config
['vrops_password']
447 vim_account
['vcenter_ip'] = vim_config
['vcenter_ip']
448 vim_account
['vcenter_port'] = vim_config
['vcenter_port']
449 vim_account
['vcenter_user'] = vim_config
['vcenter_user']
450 vim_account
['vcenter_password'] = vim_config
['vcenter_password']
452 if vim_config
['nsx_manager'] is not None:
453 vim_account
['nsx_manager'] = vim_config
['nsx_manager']
454 if vim_config
['nsx_user'] is not None:
455 vim_account
['nsx_user'] = vim_config
['nsx_user']
456 if vim_config
['nsx_password'] is not None:
457 vim_account
['nsx_password'] = vim_config
['nsx_password']
458 if vim_config
['orgname'] is not None:
459 vim_account
['orgname'] = vim_config
['orgname']
460 except Exception as exp
:
461 self
.logger
.error("VIM account details not sufficient: {}".format(exp
))
466 #log.basicConfig(filename='mon_vrops_log.log',level=log.DEBUG)
468 plugin_rcvr = PluginReceiver()
469 plugin_rcvr.consume()
471 if __name__ == "__main__":