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
39 from osm_mon
.plugins
.vRealiseOps
.mon_plugin_vrops
import MonPlugin
41 sys
.path
.append(os
.path
.join(os
.path
.dirname(os
.path
.abspath(__file__
)), '..', '..', '..'))
42 from osm_mon
.core
.message_bus
.producer
import KafkaProducer
43 #from core.message_bus.producer import KafkaProducer
44 from xml
.etree
import ElementTree
as XmlElementTree
46 schema_version
= "1.0"
47 req_config_params
= ('vrops_site', 'vrops_user', 'vrops_password',
48 'vcloud-site','admin_username','admin_password',
49 'vcenter_ip','vcenter_port','vcenter_user','vcenter_password',
50 'vim_tenant_name','orgname','tenant_id')
51 MODULE_DIR
= os
.path
.dirname(__file__
)
52 CONFIG_FILE_NAME
= 'vrops_config.xml'
53 CONFIG_FILE_PATH
= os
.path
.join(MODULE_DIR
, CONFIG_FILE_NAME
)
58 BASE_DIR
= os
.path
.dirname(os
.path
.dirname(__file__
))
59 logger
= logging
.getLogger()
60 formatter
= logging
.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
61 handler
= logging
.FileHandler(os
.path
.join(BASE_DIR
,"mon_vrops_log.log"))
62 handler
.setFormatter(formatter
)
63 logger
.addHandler(handler
)
66 class PluginReceiver():
67 """MON Plugin receiver receiving request messages & responding using producer for vROPs
71 """Constructor of PluginReceiver
75 self
.logger
= logging
.getLogger('PluginReceiver')
76 self
.logger
.setLevel(logging
.DEBUG
)
80 self
.producer_alarms
= KafkaProducer('alarm_response')
81 self
.producer_metrics
= KafkaProducer('metric_response')
82 self
.producer_access_credentials
= KafkaProducer('vim_access_credentials_response')
85 def consume(self
, message
):
86 """Consume the message, act on it & respond
89 self
.logger
.info("Message received:\nTopic={}:{}:{}:\nKey={}\nValue={}"\
90 .format(message
.topic
, message
.partition
, message
.offset
, message
.key
, message
.value
))
91 message_values
= json
.loads(message
.value
)
92 self
.logger
.info("Action required for: {}".format(message
.topic
))
93 if message
.topic
== 'alarm_request':
94 if message
.key
== "create_alarm_request":
95 config_alarm_info
= json
.loads(message
.value
)
96 alarm_uuid
= self
.create_alarm(config_alarm_info
['alarm_create_request'])
97 self
.logger
.info("Alarm created with alarm uuid: {}".format(alarm_uuid
))
98 #Publish message using producer
99 self
.publish_create_alarm_status(alarm_uuid
, config_alarm_info
)
100 elif message
.key
== "update_alarm_request":
101 update_alarm_info
= json
.loads(message
.value
)
102 alarm_uuid
= self
.update_alarm(update_alarm_info
['alarm_update_request'])
103 self
.logger
.info("Alarm defination updated : alarm uuid: {}".format(alarm_uuid
))
104 #Publish message using producer
105 self
.publish_update_alarm_status(alarm_uuid
, update_alarm_info
)
106 elif message
.key
== "delete_alarm_request":
107 delete_alarm_info
= json
.loads(message
.value
)
108 alarm_uuid
= self
.delete_alarm(delete_alarm_info
['alarm_delete_request'])
109 self
.logger
.info("Alarm defination deleted : alarm uuid: {}".format(alarm_uuid
))
110 #Publish message using producer
111 self
.publish_delete_alarm_status(alarm_uuid
, delete_alarm_info
)
112 elif message
.key
== "list_alarm_request":
113 request_input
= json
.loads(message
.value
)
114 triggered_alarm_list
= self
.list_alarms(request_input
['alarm_list_request'])
115 #Publish message using producer
116 self
.publish_list_alarm_response(triggered_alarm_list
, request_input
)
117 elif message
.topic
== 'metric_request':
118 if message
.key
== "read_metric_data_request":
119 metric_request_info
= json
.loads(message
.value
)
120 mon_plugin_obj
= MonPlugin()
121 metrics_data
= mon_plugin_obj
.get_metrics_data(metric_request_info
)
122 self
.logger
.info("Collected Metrics Data: {}".format(metrics_data
))
123 #Publish message using producer
124 self
.publish_metrics_data_status(metrics_data
)
125 elif message
.key
== "create_metric_request":
126 metric_info
= json
.loads(message
.value
)
127 metric_status
= self
.verify_metric(metric_info
['metric_create_request'])
128 #Publish message using producer
129 self
.publish_create_metric_response(metric_info
, metric_status
)
130 elif message
.key
== "update_metric_request":
131 metric_info
= json
.loads(message
.value
)
132 metric_status
= self
.verify_metric(metric_info
['metric_create_request'])
133 #Publish message using producer
134 self
.publish_update_metric_response(metric_info
, metric_status
)
135 elif message
.key
== "delete_metric_request":
136 metric_info
= json
.loads(message
.value
)
137 #Deleting Metric Data is not allowed. Publish status as False
138 self
.logger
.warning("Deleting Metric is not allowed: {}".format(metric_info
['metric_name']))
139 #Publish message using producer
140 self
.publish_delete_metric_response(metric_info
)
141 elif message
.topic
== 'access_credentials':
142 if message
.key
== "vim_access_credentials":
143 access_info
= json
.loads(message
.value
)
144 access_update_status
= self
.update_access_credentials(access_info
['access_config'])
145 self
.publish_access_update_response(access_update_status
, access_info
)
148 self
.logger
.error("Exception in vROPs plugin receiver: {}".format(traceback
.format_exc()))
151 def create_alarm(self
, config_alarm_info
):
152 """Create alarm using vROPs plugin
154 mon_plugin
= MonPlugin()
155 plugin_uuid
= mon_plugin
.configure_rest_plugin()
156 alarm_uuid
= mon_plugin
.configure_alarm(config_alarm_info
)
159 def publish_create_alarm_status(self
, alarm_uuid
, config_alarm_info
):
160 """Publish create alarm status using producer
162 topic
= 'alarm_response'
163 msg_key
= 'create_alarm_response'
164 response_msg
= {"schema_version":schema_version
,
165 "schema_type":"create_alarm_response",
166 "alarm_create_response":
167 {"correlation_id":config_alarm_info
["alarm_create_request"]["correlation_id"],
168 "alarm_uuid":alarm_uuid
,
169 "status": True if alarm_uuid
else False
172 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
173 .format(topic
, msg_key
, response_msg
))
175 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
177 def update_alarm(self
, update_alarm_info
):
178 """Updare already created alarm
180 mon_plugin
= MonPlugin()
181 alarm_uuid
= mon_plugin
.update_alarm_configuration(update_alarm_info
)
184 def publish_update_alarm_status(self
, alarm_uuid
, update_alarm_info
):
185 """Publish update alarm status requests using producer
187 topic
= 'alarm_response'
188 msg_key
= 'update_alarm_response'
189 response_msg
= {"schema_version":schema_version
,
190 "schema_type":"update_alarm_response",
191 "alarm_update_response":
192 {"correlation_id":update_alarm_info
["alarm_update_request"]["correlation_id"],
193 "alarm_uuid":update_alarm_info
["alarm_update_request"]["alarm_uuid"] \
194 if update_alarm_info
["alarm_update_request"].get('alarm_uuid') is not None else None,
195 "status": True if alarm_uuid
else False
198 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
199 .format(topic
, msg_key
, response_msg
))
201 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
203 def delete_alarm(self
, delete_alarm_info
):
204 """Delete alarm configuration
206 mon_plugin
= MonPlugin()
207 alarm_uuid
= mon_plugin
.delete_alarm_configuration(delete_alarm_info
)
210 def publish_delete_alarm_status(self
, alarm_uuid
, delete_alarm_info
):
211 """Publish update alarm status requests using producer
213 topic
= 'alarm_response'
214 msg_key
= 'delete_alarm_response'
215 response_msg
= {"schema_version":schema_version
,
216 "schema_type":"delete_alarm_response",
217 "alarm_deletion_response":
218 {"correlation_id":delete_alarm_info
["alarm_delete_request"]["correlation_id"],
219 "alarm_uuid":delete_alarm_info
["alarm_delete_request"]["alarm_uuid"],
220 "status": True if alarm_uuid
else False
223 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
224 .format(topic
, msg_key
, response_msg
))
226 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
229 def publish_metrics_data_status(self
, metrics_data
):
230 """Publish the requested metric data using producer
232 topic
= 'metric_response'
233 msg_key
= 'read_metric_data_response'
234 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
235 .format(topic
, msg_key
, metrics_data
))
237 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(metrics_data
), topic
=topic
)
240 def verify_metric(self
, metric_info
):
241 """Verify if metric is supported or not
243 mon_plugin
= MonPlugin()
244 metric_key_status
= mon_plugin
.verify_metric_support(metric_info
)
245 return metric_key_status
247 def publish_create_metric_response(self
, metric_info
, metric_status
):
248 """Publish create metric response
250 topic
= 'metric_response'
251 msg_key
= 'create_metric_response'
252 response_msg
= {"schema_version":schema_version
,
253 "schema_type":"create_metric_response",
254 "correlation_id":metric_info
['correlation_id'],
255 "metric_create_response":
258 "resource_uuid":metric_info
['metric_create_request']['resource_uuid'],
259 "status":metric_status
262 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
263 .format(topic
, msg_key
, response_msg
))
265 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
267 def publish_update_metric_response(self
, metric_info
, metric_status
):
268 """Publish update metric response
270 topic
= 'metric_response'
271 msg_key
= 'update_metric_response'
272 response_msg
= {"schema_version":schema_version
,
273 "schema_type":"metric_update_response",
274 "correlation_id":metric_info
['correlation_id'],
275 "metric_update_response":
278 "resource_uuid":metric_info
['metric_create_request']['resource_uuid'],
279 "status":metric_status
282 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
283 .format(topic
, msg_key
, response_msg
))
285 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
287 def publish_delete_metric_response(self
, metric_info
):
290 topic
= 'metric_response'
291 msg_key
= 'delete_metric_response'
292 if 'tenant_uuid' in metric_info
and metric_info
['tenant_uuid'] is not None:
293 tenant_uuid
= metric_info
['tenant_uuid']
297 response_msg
= {"schema_version":schema_version
,
298 "schema_type":"delete_metric_response",
299 "correlation_id":metric_info
['correlation_id'],
300 "metric_name":metric_info
['metric_name'],
302 "resource_uuid":metric_info
['resource_uuid'],
303 "tenant_uuid":tenant_uuid
,
306 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
307 .format(topic
, msg_key
, response_msg
))
309 self
.producer_metrics
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
311 def list_alarms(self
, list_alarm_input
):
312 """Collect list of triggered alarms based on input
314 mon_plugin
= MonPlugin()
315 triggered_alarms
= mon_plugin
.get_triggered_alarms_list(list_alarm_input
)
316 return triggered_alarms
319 def publish_list_alarm_response(self
, triggered_alarm_list
, list_alarm_input
):
320 """Publish list of triggered alarms
322 topic
= 'alarm_response'
323 msg_key
= 'list_alarm_response'
324 response_msg
= {"schema_version":schema_version
,
325 "schema_type":"list_alarm_response",
326 "correlation_id":list_alarm_input
['alarm_list_request']['correlation_id'],
327 "list_alarm_response":triggered_alarm_list
329 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
330 .format(topic
, msg_key
, response_msg
))
332 self
.producer_alarms
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
335 def update_access_credentials(self
, access_info
):
336 """Verify if all the required access config params are provided and
337 updates access config in default vrops config file
339 update_status
= False
341 #Check if all the required config params are passed in request
342 if not all (keys
in access_info
for keys
in req_config_params
):
343 self
.logger
.debug("All required Access Config Parameters not provided")
344 self
.logger
.debug("List of required Access Config Parameters: {}".format(req_config_params
))
345 self
.logger
.debug("List of given Access Config Parameters: {}".format(access_info
))
348 wr_status
= self
.write_access_config(access_info
)
349 return wr_status
#True/False
351 def write_access_config(self
, access_info
):
352 """Write access configuration to vROPs config file.
356 tree
= XmlElementTree
.parse(CONFIG_FILE_PATH
)
357 root
= tree
.getroot()
360 if config
.tag
== 'Access_Config':
362 for key
,val
in six
.iteritems(access_info
):
364 #print param.tag, val
367 tree
.write(CONFIG_FILE_PATH
)
369 except Exception as exp
:
370 self
.logger
.warning("Failed to update Access Config Parameters: {}".format(exp
))
375 def publish_access_update_response(self
, access_update_status
, access_info_req
):
376 """Publish access update response
378 topic
= 'access_credentials'
379 msg_key
= 'vim_access_credentials_response'
380 response_msg
= {"schema_version":schema_version
,
381 "schema_type":"vim_access_credentials_response",
382 "correlation_id":access_info_req
['access_config']['correlation_id'],
383 "status":access_update_status
385 self
.logger
.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
386 .format(topic
, msg_key
, response_msg
))
388 self
.producer_access_credentials
.publish(key
=msg_key
, value
=json
.dumps(response_msg
), topic
=topic
)
392 #log.basicConfig(filename='mon_vrops_log.log',level=log.DEBUG)
394 plugin_rcvr = PluginReceiver()
395 plugin_rcvr.consume()
397 if __name__ == "__main__":