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
70 topics
= ['alarm_request', 'metric_request', 'access_credentials']
72 self
.logger
= logging
.getLogger('PluginReceiver')
73 self
.logger
.setLevel(logging
.DEBUG
)
75 #To Do - Add broker uri
77 #self.mon_plugin = MonPlugin()
78 self
.consumer
= vROP_KafkaConsumer(topics
, broker_uri
)
80 self
.producer_alarms
= KafkaProducer('alarm_response')
81 self
.producer_metrics
= KafkaProducer('metric_response')
82 self
.producer_access_credentials
= KafkaProducer('vim_access_credentials_response')
86 """Consume the message, act on it & respond
89 for message
in self
.consumer
.vrops_consumer
:
91 self
.logger
.info("Message received:\nTopic={}:{}:{}:\nKey={}\nValue={}"\
92 .format(message
.topic
, message
.partition
, message
.offset
, message
.key
, message
.value
))
93 message_values
= json
.loads(message
.value
)
94 if message_values
.has_key('vim_type') and message_values
['vim_type'] is not None:
95 vim_type
= message_values
['vim_type'].lower()
96 if vim_type
== 'vmware':
97 self
.logger
.info("Action required for: {}".format(message
.topic
))
98 if message
.topic
== 'alarm_request':
99 if message
.key
== "create_alarm_request":
100 config_alarm_info
= json
.loads(message
.value
)
101 alarm_uuid
= self
.create_alarm(config_alarm_info
['alarm_create_request'])
102 self
.logger
.info("Alarm created with alarm uuid: {}".format(alarm_uuid
))
103 #Publish message using producer
104 self
.publish_create_alarm_status(alarm_uuid
, config_alarm_info
)
105 elif message
.key
== "update_alarm_request":
106 update_alarm_info
= json
.loads(message
.value
)
107 alarm_uuid
= self
.update_alarm(update_alarm_info
['alarm_update_request'])
108 self
.logger
.info("Alarm defination updated : alarm uuid: {}".format(alarm_uuid
))
109 #Publish message using producer
110 self
.publish_update_alarm_status(alarm_uuid
, update_alarm_info
)
111 elif message
.key
== "delete_alarm_request":
112 delete_alarm_info
= json
.loads(message
.value
)
113 alarm_uuid
= self
.delete_alarm(delete_alarm_info
['alarm_delete_request'])
114 self
.logger
.info("Alarm defination deleted : alarm uuid: {}".format(alarm_uuid
))
115 #Publish message using producer
116 self
.publish_delete_alarm_status(alarm_uuid
, delete_alarm_info
)
117 elif message
.key
== "list_alarm_request":
118 request_input
= json
.loads(message
.value
)
119 triggered_alarm_list
= self
.list_alarms(request_input
['alarm_list_request'])
120 #Publish message using producer
121 self
.publish_list_alarm_response(triggered_alarm_list
, request_input
)
122 elif message
.topic
== 'metric_request':
123 if message
.key
== "read_metric_data_request":
124 metric_request_info
= json
.loads(message
.value
)
125 mon_plugin_obj
= MonPlugin()
126 metrics_data
= mon_plugin_obj
.get_metrics_data(metric_request_info
)
127 self
.logger
.info("Collected Metrics Data: {}".format(metrics_data
))
128 #Publish message using producer
129 self
.publish_metrics_data_status(metrics_data
)
130 elif message
.key
== "create_metric_request":
131 metric_info
= json
.loads(message
.value
)
132 metric_status
= self
.verify_metric(metric_info
['metric_create'])
133 #Publish message using producer
134 self
.publish_create_metric_response(metric_info
, metric_status
)
135 elif message
.key
== "update_metric_request":
136 metric_info
= json
.loads(message
.value
)
137 metric_status
= self
.verify_metric(metric_info
['metric_create'])
138 #Publish message using producer
139 self
.publish_update_metric_response(metric_info
, metric_status
)
140 elif message
.key
== "delete_metric_request":
141 metric_info
= json
.loads(message
.value
)
142 #Deleting Metric Data is not allowed. Publish status as False
143 self
.logger
.warn("Deleting Metric is not allowed: {}".format(metric_info
['metric_name']))
144 #Publish message using producer
145 self
.publish_delete_metric_response(metric_info
)
146 elif message
.topic
== 'access_credentials':
147 if message
.key
== "vim_access_credentials":
148 access_info
= json
.loads(message
.value
)
149 access_update_status
= self
.update_access_credentials(access_info
['access_config'])
150 self
.publish_access_update_response(access_update_status
, access_info
)
153 self
.logger
.error("Exception in receiver: {}".format(traceback
.format_exc()))
156 def create_alarm(self
, config_alarm_info
):
157 """Create alarm using vROPs plugin
159 mon_plugin
= MonPlugin()
160 plugin_uuid
= mon_plugin
.configure_rest_plugin()
161 alarm_uuid
= mon_plugin
.configure_alarm(config_alarm_info
)
164 def publish_create_alarm_status(self
, alarm_uuid
, config_alarm_info
):
165 """Publish create alarm status using producer
167 topic
= 'alarm_response'
168 msg_key
= 'create_alarm_response'
169 response_msg
= {"schema_version":schema_version
,
170 "schema_type":"create_alarm_response",
171 "alarm_create_response":
172 {"correlation_id":config_alarm_info
["alarm_create_request"]["correlation_id"],
173 "alarm_uuid":alarm_uuid
,
174 "status": True if alarm_uuid
else False
177 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
178 .format(topic
, msg_key
, response_msg
))
180 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
182 def update_alarm(self
, update_alarm_info
):
183 """Updare already created alarm
185 mon_plugin
= MonPlugin()
186 alarm_uuid
= mon_plugin
.update_alarm_configuration(update_alarm_info
)
189 def publish_update_alarm_status(self
, alarm_uuid
, update_alarm_info
):
190 """Publish update alarm status requests using producer
192 topic
= 'alarm_response'
193 msg_key
= 'update_alarm_response'
194 response_msg
= {"schema_version":schema_version
,
195 "schema_type":"update_alarm_response",
196 "alarm_update_response":
197 {"correlation_id":update_alarm_info
["alarm_update_request"]["correlation_id"],
198 "alarm_uuid":update_alarm_info
["alarm_update_request"]["alarm_uuid"] \
199 if update_alarm_info
["alarm_update_request"].get('alarm_uuid') is not None else None,
200 "status": True if alarm_uuid
else False
203 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
204 .format(topic
, msg_key
, response_msg
))
206 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
208 def delete_alarm(self
, delete_alarm_info
):
209 """Delete alarm configuration
211 mon_plugin
= MonPlugin()
212 alarm_uuid
= mon_plugin
.delete_alarm_configuration(delete_alarm_info
)
215 def publish_delete_alarm_status(self
, alarm_uuid
, delete_alarm_info
):
216 """Publish update alarm status requests using producer
218 topic
= 'alarm_response'
219 msg_key
= 'delete_alarm_response'
220 response_msg
= {"schema_version":schema_version
,
221 "schema_type":"delete_alarm_response",
222 "alarm_deletion_response":
223 {"correlation_id":delete_alarm_info
["alarm_delete_request"]["correlation_id"],
224 "alarm_uuid":delete_alarm_info
["alarm_delete_request"]["alarm_uuid"],
225 "status": True if alarm_uuid
else False
228 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
229 .format(topic
, msg_key
, response_msg
))
231 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
234 def publish_metrics_data_status(self
, metrics_data
):
235 """Publish the requested metric data using producer
237 topic
= 'metric_response'
238 msg_key
= 'read_metric_data_response'
239 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
240 .format(topic
, msg_key
, metrics_data
))
242 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(metrics_data
), topic
=topic
)
245 def verify_metric(self
, metric_info
):
246 """Verify if metric is supported or not
248 mon_plugin
= MonPlugin()
249 metric_key_status
= mon_plugin
.verify_metric_support(metric_info
)
250 return metric_key_status
252 def publish_create_metric_response(self
, metric_info
, metric_status
):
253 """Publish create metric response
255 topic
= 'metric_response'
256 msg_key
= 'create_metric_response'
257 response_msg
= {"schema_version":schema_version
,
258 "schema_type":"create_metric_response",
259 "correlation_id":metric_info
['correlation_id'],
260 "metric_create_response":
263 "resource_uuid":metric_info
['metric_create']['resource_uuid'],
264 "status":metric_status
267 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
268 .format(topic
, msg_key
, response_msg
))
270 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
272 def publish_update_metric_response(self
, metric_info
, metric_status
):
273 """Publish update metric response
275 topic
= 'metric_response'
276 msg_key
= 'update_metric_response'
277 response_msg
= {"schema_version":schema_version
,
278 "schema_type":"metric_update_response",
279 "correlation_id":metric_info
['correlation_id'],
280 "metric_update_response":
283 "resource_uuid":metric_info
['metric_create']['resource_uuid'],
284 "status":metric_status
287 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
288 .format(topic
, msg_key
, response_msg
))
290 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
292 def publish_delete_metric_response(self
, metric_info
):
295 topic
= 'metric_response'
296 msg_key
= 'delete_metric_response'
297 if metric_info
.has_key('tenant_uuid') and metric_info
['tenant_uuid'] is not None:
298 tenant_uuid
= metric_info
['tenant_uuid']
302 response_msg
= {"schema_version":schema_version
,
303 "schema_type":"delete_metric_response",
304 "correlation_id":metric_info
['correlation_id'],
305 "metric_name":metric_info
['metric_name'],
307 "resource_uuid":metric_info
['resource_uuid'],
308 "tenant_uuid":tenant_uuid
,
311 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
312 .format(topic
, msg_key
, response_msg
))
314 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
316 def list_alarms(self
, list_alarm_input
):
317 """Collect list of triggered alarms based on input
319 mon_plugin
= MonPlugin()
320 triggered_alarms
= mon_plugin
.get_triggered_alarms_list(list_alarm_input
)
321 return triggered_alarms
324 def publish_list_alarm_response(self
, triggered_alarm_list
, list_alarm_input
):
325 """Publish list of triggered alarms
327 topic
= 'alarm_response'
328 msg_key
= 'list_alarm_response'
329 response_msg
= {"schema_version":schema_version
,
330 "schema_type":"list_alarm_response",
331 "correlation_id":list_alarm_input
['alarm_list_request']['correlation_id'],
332 "list_alarm_resp":triggered_alarm_list
334 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
335 .format(topic
, msg_key
, response_msg
))
337 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
340 def update_access_credentials(self
, access_info
):
341 """Verify if all the required access config params are provided and
342 updates access config in default vrops config file
344 update_status
= False
346 #Check if all the required config params are passed in request
347 if not all (keys
in access_info
for keys
in req_config_params
):
348 self
.logger
.debug("All required Access Config Parameters not provided")
349 self
.logger
.debug("List of required Access Config Parameters: {}".format(req_config_params
))
350 self
.logger
.debug("List of given Access Config Parameters: {}".format(access_info
))
353 wr_status
= self
.write_access_config(access_info
)
354 return wr_status
#True/False
356 def write_access_config(self
, access_info
):
357 """Write access configuration to vROPs config file.
361 tree
= XmlElementTree
.parse(CONFIG_FILE_PATH
)
362 root
= tree
.getroot()
365 if config
.tag
== 'Access_Config':
367 for key
,val
in access_info
.iteritems():
369 #print param.tag, val
372 tree
.write(CONFIG_FILE_PATH
)
374 except Exception as exp
:
375 self
.logger
.warn("Failed to update Access Config Parameters: {}".format(exp
))
380 def publish_access_update_response(self
, access_update_status
, access_info_req
):
381 """Publish access update response
383 topic
= 'access_credentials'
384 msg_key
= 'vim_access_credentials_response'
385 response_msg
= {"schema_version":schema_version
,
386 "schema_type":"vim_access_credentials_response",
387 "correlation_id":access_info_req
['access_config']['correlation_id'],
388 "status":access_update_status
390 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
391 .format(topic
, msg_key
, response_msg
))
393 self
.producer_access_credentials
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
396 #log.basicConfig(filename='mon_vrops_log.log',level=log.DEBUG)
398 plugin_rcvr
= PluginReceiver()
399 plugin_rcvr
.consume()
401 if __name__
== "__main__":