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 json
  30 import logging
  31 import os
  32 import sys
  33 import traceback
  34
  35
  36 #Core producer
  37 from osm_mon.plugins.vRealiseOps.mon_plugin_vrops import MonPlugin
  38
  39 sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..', '..'))
  40 from osm_mon.core.message_bus.producer import KafkaProducer
  41 #from core.message_bus.producer import KafkaProducer
  42 from xml.etree import ElementTree as XmlElementTree
  43
  44 schema_version = "1.0"
  45 req_config_params = ('vrops_site', 'vrops_user', 'vrops_password',
  46                     'vcloud-site','admin_username','admin_password',
  47                     'vcenter_ip','vcenter_port','vcenter_user','vcenter_password',
  48                     'vim_tenant_name','orgname','tenant_id')
  49 MODULE_DIR = os.path.dirname(__file__)
  50 CONFIG_FILE_NAME = 'vrops_config.xml'
  51 CONFIG_FILE_PATH = os.path.join(MODULE_DIR, CONFIG_FILE_NAME)
  52
  53 def set_logger():
  54     """Set Logger
  55     """
  56     BASE_DIR = os.path.dirname(os.path.dirname(__file__))
  57     logger = logging.getLogger()
  58     formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  59     handler = logging.FileHandler(os.path.join(BASE_DIR,"mon_vrops_log.log"))
  60     handler.setFormatter(formatter)
  61     logger.addHandler(handler)
  62
  63
  64 class PluginReceiver():
  65     """MON Plugin receiver receiving request messages & responding using producer for vROPs
  66     telemetry plugin
  67     """
  68     def __init__(self):
  69         """Constructor of PluginReceiver
  70         """
  71
  72
  73         self.logger = logging.getLogger('PluginReceiver')
  74         self.logger.setLevel(logging.DEBUG)
  75         set_logger()
  76
  77         #Core producer
  78         self.producer_alarms = KafkaProducer('alarm_response')
  79         self.producer_metrics = KafkaProducer('metric_response')
  80         self.producer_access_credentials = KafkaProducer('vim_access_credentials_response')
  81
  82
  83     def consume(self, message):
  84         """Consume the message, act on it & respond
  85         """
  86         try:
  87             self.logger.info("Message received:\nTopic={}:{}:{}:\nKey={}\nValue={}"\
  88                 .format(message.topic, message.partition, message.offset, message.key, message.value))
  89             message_values = json.loads(message.value)
  90             self.logger.info("Action required for: {}".format(message.topic))
  91             if message.topic == 'alarm_request':
  92                 if message.key == "create_alarm_request":
  93                     config_alarm_info = json.loads(message.value)
  94                     alarm_uuid = self.create_alarm(config_alarm_info['alarm_create_request'])
  95                     self.logger.info("Alarm created with alarm uuid: {}".format(alarm_uuid))
  96                     #Publish message using producer
  97                     self.publish_create_alarm_status(alarm_uuid, config_alarm_info)
  98                 elif message.key == "update_alarm_request":
  99                     update_alarm_info = json.loads(message.value)
 100                     alarm_uuid = self.update_alarm(update_alarm_info['alarm_update_request'])
 101                     self.logger.info("Alarm defination updated : alarm uuid: {}".format(alarm_uuid))
 102                     #Publish message using producer
 103                     self.publish_update_alarm_status(alarm_uuid, update_alarm_info)
 104                 elif message.key == "delete_alarm_request":
 105                     delete_alarm_info = json.loads(message.value)
 106                     alarm_uuid = self.delete_alarm(delete_alarm_info['alarm_delete_request'])
 107                     self.logger.info("Alarm defination deleted : alarm uuid: {}".format(alarm_uuid))
 108                     #Publish message using producer
 109                     self.publish_delete_alarm_status(alarm_uuid, delete_alarm_info)
 110                 elif message.key == "list_alarm_request":
 111                     request_input = json.loads(message.value)
 112                     triggered_alarm_list = self.list_alarms(request_input['alarm_list_request'])
 113                     #Publish message using producer
 114                     self.publish_list_alarm_response(triggered_alarm_list, request_input)
 115             elif message.topic == 'metric_request':
 116                 if message.key == "read_metric_data_request":
 117                     metric_request_info = json.loads(message.value)
 118                     mon_plugin_obj = MonPlugin()
 119                     metrics_data = mon_plugin_obj.get_metrics_data(metric_request_info)
 120                     self.logger.info("Collected Metrics Data: {}".format(metrics_data))
 121                     #Publish message using producer
 122                     self.publish_metrics_data_status(metrics_data)
 123                 elif message.key == "create_metric_request":
 124                     metric_info = json.loads(message.value)
 125                     metric_status = self.verify_metric(metric_info['metric_create'])
 126                     #Publish message using producer
 127                     self.publish_create_metric_response(metric_info, metric_status)
 128                 elif message.key == "update_metric_request":
 129                     metric_info = json.loads(message.value)
 130                     metric_status = self.verify_metric(metric_info['metric_create'])
 131                     #Publish message using producer
 132                     self.publish_update_metric_response(metric_info, metric_status)
 133                 elif message.key == "delete_metric_request":
 134                     metric_info = json.loads(message.value)
 135                     #Deleting Metric Data is not allowed. Publish status as False
 136                     self.logger.warn("Deleting Metric is not allowed: {}".format(metric_info['metric_name']))
 137                     #Publish message using producer
 138                     self.publish_delete_metric_response(metric_info)
 139             elif message.topic == 'access_credentials':
 140                 if message.key == "vim_access_credentials":
 141                     access_info = json.loads(message.value)
 142                     access_update_status = self.update_access_credentials(access_info['access_config'])
 143                     self.publish_access_update_response(access_update_status, access_info)
 144
 145         except:
 146             self.logger.error("Exception in vROPs plugin receiver: {}".format(traceback.format_exc()))
 147
 148
 149     def create_alarm(self, config_alarm_info):
 150         """Create alarm using vROPs plugin
 151         """
 152         mon_plugin = MonPlugin()
 153         plugin_uuid = mon_plugin.configure_rest_plugin()
 154         alarm_uuid = mon_plugin.configure_alarm(config_alarm_info)
 155         return alarm_uuid
 156
 157     def publish_create_alarm_status(self, alarm_uuid, config_alarm_info):
 158         """Publish create alarm status using producer
 159         """
 160         topic = 'alarm_response'
 161         msg_key = 'create_alarm_response'
 162         response_msg = {"schema_version":schema_version,
 163                          "schema_type":"create_alarm_response",
 164                          "alarm_create_response":
 165                             {"correlation_id":config_alarm_info["alarm_create_request"]["correlation_id"],
 166                              "alarm_uuid":alarm_uuid,
 167                              "status": True if alarm_uuid else False
 168                             }
 169                        }
 170         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 171                 .format(topic, msg_key, response_msg))
 172         #Core producer
 173         self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 174
 175     def update_alarm(self, update_alarm_info):
 176         """Updare already created alarm
 177         """
 178         mon_plugin = MonPlugin()
 179         alarm_uuid = mon_plugin.update_alarm_configuration(update_alarm_info)
 180         return alarm_uuid
 181
 182     def publish_update_alarm_status(self, alarm_uuid, update_alarm_info):
 183         """Publish update alarm status requests using producer
 184         """
 185         topic = 'alarm_response'
 186         msg_key = 'update_alarm_response'
 187         response_msg = {"schema_version":schema_version,
 188                          "schema_type":"update_alarm_response",
 189                          "alarm_update_response":
 190                             {"correlation_id":update_alarm_info["alarm_update_request"]["correlation_id"],
 191                              "alarm_uuid":update_alarm_info["alarm_update_request"]["alarm_uuid"] \
 192                              if update_alarm_info["alarm_update_request"].get('alarm_uuid') is not None else None,
 193                              "status": True if alarm_uuid else False
 194                             }
 195                        }
 196         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 197                 .format(topic, msg_key, response_msg))
 198         #Core producer
 199         self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 200
 201     def delete_alarm(self, delete_alarm_info):
 202         """Delete alarm configuration
 203         """
 204         mon_plugin = MonPlugin()
 205         alarm_uuid = mon_plugin.delete_alarm_configuration(delete_alarm_info)
 206         return alarm_uuid
 207
 208     def publish_delete_alarm_status(self, alarm_uuid, delete_alarm_info):
 209         """Publish update alarm status requests using producer
 210         """
 211         topic = 'alarm_response'
 212         msg_key = 'delete_alarm_response'
 213         response_msg = {"schema_version":schema_version,
 214                          "schema_type":"delete_alarm_response",
 215                          "alarm_deletion_response":
 216                             {"correlation_id":delete_alarm_info["alarm_delete_request"]["correlation_id"],
 217                              "alarm_uuid":delete_alarm_info["alarm_delete_request"]["alarm_uuid"],
 218                              "status": True if alarm_uuid else False
 219                             }
 220                        }
 221         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 222                 .format(topic, msg_key, response_msg))
 223         #Core producer
 224         self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 225
 226
 227     def publish_metrics_data_status(self, metrics_data):
 228         """Publish the requested metric data using producer
 229         """
 230         topic = 'metric_response'
 231         msg_key = 'read_metric_data_response'
 232         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 233                 .format(topic, msg_key, metrics_data))
 234         #Core producer
 235         self.producer_metrics.publish(key=msg_key, value=json.dumps(metrics_data), topic=topic)
 236
 237
 238     def verify_metric(self, metric_info):
 239         """Verify if metric is supported or not
 240         """
 241         mon_plugin = MonPlugin()
 242         metric_key_status = mon_plugin.verify_metric_support(metric_info)
 243         return metric_key_status
 244
 245     def publish_create_metric_response(self, metric_info, metric_status):
 246         """Publish create metric response
 247         """
 248         topic = 'metric_response'
 249         msg_key = 'create_metric_response'
 250         response_msg = {"schema_version":schema_version,
 251                          "schema_type":"create_metric_response",
 252                          "correlation_id":metric_info['correlation_id'],
 253                          "metric_create_response":
 254                             {
 255                              "metric_uuid":'0',
 256                              "resource_uuid":metric_info['metric_create']['resource_uuid'],
 257                              "status":metric_status
 258                             }
 259                        }
 260         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 261                 .format(topic, msg_key, response_msg))
 262         #Core producer
 263         self.producer_metrics.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 264
 265     def publish_update_metric_response(self, metric_info, metric_status):
 266         """Publish update metric response
 267         """
 268         topic = 'metric_response'
 269         msg_key = 'update_metric_response'
 270         response_msg = {"schema_version":schema_version,
 271                         "schema_type":"metric_update_response",
 272                         "correlation_id":metric_info['correlation_id'],
 273                         "metric_update_response":
 274                             {
 275                              "metric_uuid":'0',
 276                              "resource_uuid":metric_info['metric_create']['resource_uuid'],
 277                              "status":metric_status
 278                             }
 279                        }
 280         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 281                 .format(topic, msg_key, response_msg))
 282         #Core producer
 283         self.producer_metrics.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 284
 285     def publish_delete_metric_response(self, metric_info):
 286         """
 287         """
 288         topic = 'metric_response'
 289         msg_key = 'delete_metric_response'
 290         if metric_info.has_key('tenant_uuid') and metric_info['tenant_uuid'] is not None:
 291             tenant_uuid = metric_info['tenant_uuid']
 292         else:
 293             tenant_uuid = None
 294
 295         response_msg = {"schema_version":schema_version,
 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":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":schema_version,
 323                         "schema_type":"list_alarm_response",
 324                         "correlation_id":list_alarm_input['alarm_list_request']['correlation_id'],
 325                         "list_alarm_resp":triggered_alarm_list
 326                        }
 327         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 328                 .format(topic, msg_key, response_msg))
 329         #Core producer
 330         self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 331
 332
 333     def update_access_credentials(self, access_info):
 334         """Verify if all the required access config params are provided and
 335            updates access config in default vrops config file
 336         """
 337         update_status = False
 338         wr_status = False
 339         #Check if all the required config params are passed in request
 340         if not all (keys in access_info for keys in req_config_params):
 341             self.logger.debug("All required Access Config Parameters not provided")
 342             self.logger.debug("List of required Access Config Parameters: {}".format(req_config_params))
 343             self.logger.debug("List of given Access Config Parameters: {}".format(access_info))
 344             return update_status
 345
 346         wr_status = self.write_access_config(access_info)
 347         return wr_status    #True/False
 348
 349     def write_access_config(self, access_info):
 350         """Write access configuration to vROPs config file.
 351         """
 352         wr_status = False
 353         try:
 354             tree = XmlElementTree.parse(CONFIG_FILE_PATH)
 355             root = tree.getroot()
 356             alarmParams = {}
 357             for config in root:
 358                 if config.tag == 'Access_Config':
 359                     for param in config:
 360                         for key,val in access_info.iteritems():
 361                             if param.tag == key:
 362                                 #print param.tag, val
 363                                 param.text = val
 364
 365             tree.write(CONFIG_FILE_PATH)
 366             wr_status = True
 367         except Exception as exp:
 368             self.logger.warn("Failed to update Access Config Parameters: {}".format(exp))
 369
 370         return wr_status
 371
 372
 373     def publish_access_update_response(self, access_update_status, access_info_req):
 374         """Publish access update response
 375         """
 376         topic = 'access_credentials'
 377         msg_key = 'vim_access_credentials_response'
 378         response_msg = {"schema_version":schema_version,
 379                         "schema_type":"vim_access_credentials_response",
 380                         "correlation_id":access_info_req['access_config']['correlation_id'],
 381                         "status":access_update_status
 382                        }
 383         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 384                 .format(topic, msg_key, response_msg))
 385         #Core Add producer
 386         self.producer_access_credentials.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 387
 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 """