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 Montoring plugin receiver that consumes the request messages &
26 responds using producer for vROPs
30 from mon_plugin_vrops
import MonPlugin
31 from kafka_consumer_vrops
import vROP_KafkaConsumer
33 sys
.path
.append("../../core/message_bus")
34 from producer
import KafkaProducer
35 #from core.message_bus.producer import KafkaProducer
40 from xml
.etree
import ElementTree
as XmlElementTree
42 schema_version
= "1.0"
43 req_config_params
= ('vrops_site', 'vrops_user', 'vrops_password',
44 'vcloud-site','admin_username','admin_password',
45 'vcenter_ip','vcenter_port','vcenter_user','vcenter_password',
46 'vim_tenant_name','orgname','tenant_id')
47 MODULE_DIR
= os
.path
.dirname(__file__
)
48 CONFIG_FILE_NAME
= 'vrops_config.xml'
49 CONFIG_FILE_PATH
= os
.path
.join(MODULE_DIR
, CONFIG_FILE_NAME
)
54 BASE_DIR
= os
.path
.dirname(os
.path
.dirname(__file__
))
55 logger
= logging
.getLogger()
56 formatter
= logging
.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
57 handler
= logging
.FileHandler(os
.path
.join(BASE_DIR
,"mon_vrops_log.log"))
58 handler
.setFormatter(formatter
)
59 logger
.addHandler(handler
)
62 class PluginReceiver():
63 """MON Plugin receiver receiving request messages & responding using producer for vROPs
67 """Constructor of PluginReceiver
71 self
.logger
= logging
.getLogger('PluginReceiver')
72 self
.logger
.setLevel(logging
.DEBUG
)
76 self
.producer_alarms
= KafkaProducer('alarm_response')
77 self
.producer_metrics
= KafkaProducer('metric_response')
78 self
.producer_access_credentials
= KafkaProducer('vim_access_credentials_response')
81 def consume(self
, message
):
82 """Consume the message, act on it & respond
85 self
.logger
.info("Message received:\nTopic={}:{}:{}:\nKey={}\nValue={}"\
86 .format(message
.topic
, message
.partition
, message
.offset
, message
.key
, message
.value
))
87 message_values
= json
.loads(message
.value
)
88 self
.logger
.info("Action required for: {}".format(message
.topic
))
89 if message
.topic
== 'alarm_request':
90 if message
.key
== "create_alarm_request":
91 config_alarm_info
= json
.loads(message
.value
)
92 alarm_uuid
= self
.create_alarm(config_alarm_info
['alarm_create_request'])
93 self
.logger
.info("Alarm created with alarm uuid: {}".format(alarm_uuid
))
94 #Publish message using producer
95 self
.publish_create_alarm_status(alarm_uuid
, config_alarm_info
)
96 elif message
.key
== "update_alarm_request":
97 update_alarm_info
= json
.loads(message
.value
)
98 alarm_uuid
= self
.update_alarm(update_alarm_info
['alarm_update_request'])
99 self
.logger
.info("Alarm defination updated : alarm uuid: {}".format(alarm_uuid
))
100 #Publish message using producer
101 self
.publish_update_alarm_status(alarm_uuid
, update_alarm_info
)
102 elif message
.key
== "delete_alarm_request":
103 delete_alarm_info
= json
.loads(message
.value
)
104 alarm_uuid
= self
.delete_alarm(delete_alarm_info
['alarm_delete_request'])
105 self
.logger
.info("Alarm defination deleted : alarm uuid: {}".format(alarm_uuid
))
106 #Publish message using producer
107 self
.publish_delete_alarm_status(alarm_uuid
, delete_alarm_info
)
108 elif message
.key
== "list_alarm_request":
109 request_input
= json
.loads(message
.value
)
110 triggered_alarm_list
= self
.list_alarms(request_input
['alarm_list_request'])
111 #Publish message using producer
112 self
.publish_list_alarm_response(triggered_alarm_list
, request_input
)
113 elif message
.topic
== 'metric_request':
114 if message
.key
== "read_metric_data_request":
115 metric_request_info
= json
.loads(message
.value
)
116 mon_plugin_obj
= MonPlugin()
117 metrics_data
= mon_plugin_obj
.get_metrics_data(metric_request_info
)
118 self
.logger
.info("Collected Metrics Data: {}".format(metrics_data
))
119 #Publish message using producer
120 self
.publish_metrics_data_status(metrics_data
)
121 elif message
.key
== "create_metric_request":
122 metric_info
= json
.loads(message
.value
)
123 metric_status
= self
.verify_metric(metric_info
['metric_create'])
124 #Publish message using producer
125 self
.publish_create_metric_response(metric_info
, metric_status
)
126 elif message
.key
== "update_metric_request":
127 metric_info
= json
.loads(message
.value
)
128 metric_status
= self
.verify_metric(metric_info
['metric_create'])
129 #Publish message using producer
130 self
.publish_update_metric_response(metric_info
, metric_status
)
131 elif message
.key
== "delete_metric_request":
132 metric_info
= json
.loads(message
.value
)
133 #Deleting Metric Data is not allowed. Publish status as False
134 self
.logger
.warn("Deleting Metric is not allowed: {}".format(metric_info
['metric_name']))
135 #Publish message using producer
136 self
.publish_delete_metric_response(metric_info
)
137 elif message
.topic
== 'access_credentials':
138 if message
.key
== "vim_access_credentials":
139 access_info
= json
.loads(message
.value
)
140 access_update_status
= self
.update_access_credentials(access_info
['access_config'])
141 self
.publish_access_update_response(access_update_status
, access_info
)
144 self
.logger
.error("Exception in vROPs plugin receiver: {}".format(traceback
.format_exc()))
147 def create_alarm(self
, config_alarm_info
):
148 """Create alarm using vROPs plugin
150 mon_plugin
= MonPlugin()
151 plugin_uuid
= mon_plugin
.configure_rest_plugin()
152 alarm_uuid
= mon_plugin
.configure_alarm(config_alarm_info
)
155 def publish_create_alarm_status(self
, alarm_uuid
, config_alarm_info
):
156 """Publish create alarm status using producer
158 topic
= 'alarm_response'
159 msg_key
= 'create_alarm_response'
160 response_msg
= {"schema_version":schema_version
,
161 "schema_type":"create_alarm_response",
162 "alarm_create_response":
163 {"correlation_id":config_alarm_info
["alarm_create_request"]["correlation_id"],
164 "alarm_uuid":alarm_uuid
,
165 "status": True if alarm_uuid
else False
168 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
169 .format(topic
, msg_key
, response_msg
))
171 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
173 def update_alarm(self
, update_alarm_info
):
174 """Updare already created alarm
176 mon_plugin
= MonPlugin()
177 alarm_uuid
= mon_plugin
.update_alarm_configuration(update_alarm_info
)
180 def publish_update_alarm_status(self
, alarm_uuid
, update_alarm_info
):
181 """Publish update alarm status requests using producer
183 topic
= 'alarm_response'
184 msg_key
= 'update_alarm_response'
185 response_msg
= {"schema_version":schema_version
,
186 "schema_type":"update_alarm_response",
187 "alarm_update_response":
188 {"correlation_id":update_alarm_info
["alarm_update_request"]["correlation_id"],
189 "alarm_uuid":update_alarm_info
["alarm_update_request"]["alarm_uuid"] \
190 if update_alarm_info
["alarm_update_request"].get('alarm_uuid') is not None else None,
191 "status": True if alarm_uuid
else False
194 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
195 .format(topic
, msg_key
, response_msg
))
197 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
199 def delete_alarm(self
, delete_alarm_info
):
200 """Delete alarm configuration
202 mon_plugin
= MonPlugin()
203 alarm_uuid
= mon_plugin
.delete_alarm_configuration(delete_alarm_info
)
206 def publish_delete_alarm_status(self
, alarm_uuid
, delete_alarm_info
):
207 """Publish update alarm status requests using producer
209 topic
= 'alarm_response'
210 msg_key
= 'delete_alarm_response'
211 response_msg
= {"schema_version":schema_version
,
212 "schema_type":"delete_alarm_response",
213 "alarm_deletion_response":
214 {"correlation_id":delete_alarm_info
["alarm_delete_request"]["correlation_id"],
215 "alarm_uuid":delete_alarm_info
["alarm_delete_request"]["alarm_uuid"],
216 "status": True if alarm_uuid
else False
219 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
220 .format(topic
, msg_key
, response_msg
))
222 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
225 def publish_metrics_data_status(self
, metrics_data
):
226 """Publish the requested metric data using producer
228 topic
= 'metric_response'
229 msg_key
= 'read_metric_data_response'
230 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
231 .format(topic
, msg_key
, metrics_data
))
233 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(metrics_data
), topic
=topic
)
236 def verify_metric(self
, metric_info
):
237 """Verify if metric is supported or not
239 mon_plugin
= MonPlugin()
240 metric_key_status
= mon_plugin
.verify_metric_support(metric_info
)
241 return metric_key_status
243 def publish_create_metric_response(self
, metric_info
, metric_status
):
244 """Publish create metric response
246 topic
= 'metric_response'
247 msg_key
= 'create_metric_response'
248 response_msg
= {"schema_version":schema_version
,
249 "schema_type":"create_metric_response",
250 "correlation_id":metric_info
['correlation_id'],
251 "metric_create_response":
254 "resource_uuid":metric_info
['metric_create']['resource_uuid'],
255 "status":metric_status
258 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
259 .format(topic
, msg_key
, response_msg
))
261 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
263 def publish_update_metric_response(self
, metric_info
, metric_status
):
264 """Publish update metric response
266 topic
= 'metric_response'
267 msg_key
= 'update_metric_response'
268 response_msg
= {"schema_version":schema_version
,
269 "schema_type":"metric_update_response",
270 "correlation_id":metric_info
['correlation_id'],
271 "metric_update_response":
274 "resource_uuid":metric_info
['metric_create']['resource_uuid'],
275 "status":metric_status
278 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
279 .format(topic
, msg_key
, response_msg
))
281 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
283 def publish_delete_metric_response(self
, metric_info
):
286 topic
= 'metric_response'
287 msg_key
= 'delete_metric_response'
288 if metric_info
.has_key('tenant_uuid') and metric_info
['tenant_uuid'] is not None:
289 tenant_uuid
= metric_info
['tenant_uuid']
293 response_msg
= {"schema_version":schema_version
,
294 "schema_type":"delete_metric_response",
295 "correlation_id":metric_info
['correlation_id'],
296 "metric_name":metric_info
['metric_name'],
298 "resource_uuid":metric_info
['resource_uuid'],
299 "tenant_uuid":tenant_uuid
,
302 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
303 .format(topic
, msg_key
, response_msg
))
305 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
307 def list_alarms(self
, list_alarm_input
):
308 """Collect list of triggered alarms based on input
310 mon_plugin
= MonPlugin()
311 triggered_alarms
= mon_plugin
.get_triggered_alarms_list(list_alarm_input
)
312 return triggered_alarms
315 def publish_list_alarm_response(self
, triggered_alarm_list
, list_alarm_input
):
316 """Publish list of triggered alarms
318 topic
= 'alarm_response'
319 msg_key
= 'list_alarm_response'
320 response_msg
= {"schema_version":schema_version
,
321 "schema_type":"list_alarm_response",
322 "correlation_id":list_alarm_input
['alarm_list_request']['correlation_id'],
323 "list_alarm_resp":triggered_alarm_list
325 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
326 .format(topic
, msg_key
, response_msg
))
328 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
331 def update_access_credentials(self
, access_info
):
332 """Verify if all the required access config params are provided and
333 updates access config in default vrops config file
335 update_status
= False
337 #Check if all the required config params are passed in request
338 if not all (keys
in access_info
for keys
in req_config_params
):
339 self
.logger
.debug("All required Access Config Parameters not provided")
340 self
.logger
.debug("List of required Access Config Parameters: {}".format(req_config_params
))
341 self
.logger
.debug("List of given Access Config Parameters: {}".format(access_info
))
344 wr_status
= self
.write_access_config(access_info
)
345 return wr_status
#True/False
347 def write_access_config(self
, access_info
):
348 """Write access configuration to vROPs config file.
352 tree
= XmlElementTree
.parse(CONFIG_FILE_PATH
)
353 root
= tree
.getroot()
356 if config
.tag
== 'Access_Config':
358 for key
,val
in access_info
.iteritems():
360 #print param.tag, val
363 tree
.write(CONFIG_FILE_PATH
)
365 except Exception as exp
:
366 self
.logger
.warn("Failed to update Access Config Parameters: {}".format(exp
))
371 def publish_access_update_response(self
, access_update_status
, access_info_req
):
372 """Publish access update response
374 topic
= 'access_credentials'
375 msg_key
= 'vim_access_credentials_response'
376 response_msg
= {"schema_version":schema_version
,
377 "schema_type":"vim_access_credentials_response",
378 "correlation_id":access_info_req
['access_config']['correlation_id'],
379 "status":access_update_status
381 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
382 .format(topic
, msg_key
, response_msg
))
384 self
.producer_access_credentials
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
388 #log.basicConfig(filename='mon_vrops_log.log',level=log.DEBUG)
390 plugin_rcvr = PluginReceiver()
391 plugin_rcvr.consume()
393 if __name__ == "__main__":