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 Monitoring 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 #Core producer
  35 import six
  36
  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
  42 from osm_mon.core.auth import AuthManager
  43
  44 #from core.message_bus.producer import KafkaProducer
  45 from xml.etree import ElementTree as XmlElementTree
  46
  47
  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)
  56
  57 def set_logger():
  58     """Set Logger
  59     """
  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)
  66
  67
  68 class PluginReceiver():
  69     """MON Plugin receiver receiving request messages & responding using producer for vROPs
  70     telemetry plugin
  71     """
  72     def __init__(self):
  73         """Constructor of PluginReceiver
  74         """
  75
  76
  77         self.logger = logging.getLogger('PluginReceiver')
  78         self.logger.setLevel(logging.DEBUG)
  79         set_logger()
  80
  81         #Core producer
  82         self.producer_alarms = KafkaProducer('alarm_response')
  83         self.producer_metrics = KafkaProducer('metric_response')
  84         self.producer_access_credentials = KafkaProducer('vim_access_credentials_response')
  85
  86
  87     def consume(self, message, vim_uuid):
  88         """Consume the message, act on it & respond
  89         """
  90         try:
  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
 126                     access_config = None
 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)
 155
 156         except:
 157             self.logger.error("Exception in vROPs plugin receiver: {}".format(traceback.format_exc()))
 158
 159
 160     def create_alarm(self, config_alarm_info):
 161         """Create alarm using vROPs plugin
 162         """
 163         access_config = None
 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'])
 168         return alarm_uuid
 169
 170     def publish_create_alarm_status(self, alarm_uuid, config_alarm_info):
 171         """Publish create alarm status using producer
 172         """
 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
 182                             }
 183                        }
 184         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 185                 .format(topic, msg_key, response_msg))
 186         #Core producer
 187         self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 188
 189     def update_alarm(self, update_alarm_info):
 190         """Update already created alarm
 191         """
 192         access_config = None
 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'])
 196         return alarm_uuid
 197
 198     def publish_update_alarm_status(self, alarm_uuid, update_alarm_info):
 199         """Publish update alarm status requests using producer
 200         """
 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
 211                             }
 212                        }
 213         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 214                 .format(topic, msg_key, response_msg))
 215         #Core producer
 216         self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 217
 218     def delete_alarm(self, delete_alarm_info):
 219         """Delete alarm configuration
 220         """
 221         access_config = None
 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'])
 225         return alarm_uuid
 226
 227     def publish_delete_alarm_status(self, alarm_uuid, delete_alarm_info):
 228         """Publish update alarm status requests using producer
 229         """
 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
 239                             }
 240                        }
 241         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 242                 .format(topic, msg_key, response_msg))
 243         #Core producer
 244         self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 245
 246
 247     def publish_metrics_data_status(self, metrics_data):
 248         """Publish the requested metric data using producer
 249         """
 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))
 254         #Core producer
 255         self.producer_metrics.publish(key=msg_key, value=json.dumps(metrics_data), topic=topic)
 256
 257
 258     def verify_metric(self, metric_info):
 259         """Verify if metric is supported or not
 260         """
 261         access_config = None
 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'])
 266         else:
 267             metric_key_status = mon_plugin.verify_metric_support(metric_info['metric_update_request'])
 268         return metric_key_status
 269
 270     def publish_create_metric_response(self, metric_info, metric_status):
 271         """Publish create metric response
 272         """
 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":
 280                             {
 281                              ##"metric_uuid":'0',
 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
 286                             }
 287                        }
 288         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 289                 .format(topic, msg_key, response_msg))
 290         #Core producer
 291         self.producer_metrics.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 292
 293     def publish_update_metric_response(self, metric_info, metric_status):
 294         """Publish update metric response
 295         """
 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":
 302                             {
 303                              "metric_uuid":'0',
 304                              "correlation_id":metric_info['correlation_id'],
 305                              "resource_uuid":metric_info['metric_create']['resource_uuid'],
 306                              "status":metric_status
 307                             }
 308                        }
 309         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 310                 .format(topic, msg_key, response_msg))
 311         #Core producer
 312         self.producer_metrics.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 313
 314     def publish_delete_metric_response(self, metric_info):
 315         """Publish delete metric response
 316         """
 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']
 321         else:
 322             tenant_uuid = None
 323
 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'],
 329                         "metric_uuid":'0',
 330                         "resource_uuid":metric_info['resource_uuid'],
 331                         "tenant_uuid":tenant_uuid,
 332                         "status":False
 333                        }
 334         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 335                 .format(topic, msg_key, response_msg))
 336         #Core producer
 337         self.producer_metrics.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 338
 339     def list_alarms(self, list_alarm_input):
 340         """Collect list of triggered alarms based on input
 341         """
 342         access_config = None
 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
 347
 348
 349     def publish_list_alarm_response(self, triggered_alarm_list, list_alarm_input):
 350         """Publish list of triggered alarms
 351         """
 352         topic = 'alarm_response'
 353         msg_key = 'list_alarm_response'
 354         response_msg = {"schema_version":schema_version,
 355                         "schema_type":"list_alarm_response",
 356                         "vim_type":"VMware",
 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
 360                        }
 361         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 362                 .format(topic, msg_key, response_msg))
 363         #Core producer
 364         self.producer_alarms.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 365
 366
 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
 370         """
 371         update_status = False
 372         wr_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))
 378             return update_status
 379
 380         wr_status = self.write_access_config(access_info)
 381         return wr_status    #True/False
 382
 383     def write_access_config(self, access_info):
 384         """Write access configuration to vROPs config file.
 385         """
 386         wr_status = False
 387         try:
 388             tree = XmlElementTree.parse(CONFIG_FILE_PATH)
 389             root = tree.getroot()
 390             alarmParams = {}
 391             for config in root:
 392                 if config.tag == 'Access_Config':
 393                     for param in config:
 394                         for key,val in six.iteritems(access_info):
 395                             if param.tag == key:
 396                                 #print param.tag, val
 397                                 param.text = val
 398
 399             tree.write(CONFIG_FILE_PATH)
 400             wr_status = True
 401         except Exception as exp:
 402             self.logger.warning("Failed to update Access Config Parameters: {}".format(exp))
 403
 404         return wr_status
 405
 406
 407     def publish_access_update_response(self, access_update_status, access_info_req):
 408         """Publish access update response
 409         """
 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
 416                        }
 417         self.logger.info("Publishing response:\nTopic={}\nKey={}\nValue={}"\
 418                 .format(topic, msg_key, response_msg))
 419         #Core Add producer
 420         self.producer_access_credentials.publish(key=msg_key, value=json.dumps(response_msg), topic=topic)
 421
 422
 423     def get_vim_access_config(self, vim_uuid):
 424         """Get VIM access configuration & account details from path: VIM_ACCOUNTS_FILE_PATH
 425         """
 426         vim_account_details = None
 427         vim_account = {}
 428         auth_manager = AuthManager()
 429         vim_account_details = auth_manager.get_credentials(vim_uuid)
 430
 431         try:
 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
 440
 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']
 451
 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))
 462         return vim_account
 463
 464 """
 465 def main():
 466     #log.basicConfig(filename='mon_vrops_log.log',level=log.DEBUG)
 467     set_logger()
 468     plugin_rcvr = PluginReceiver()
 469     plugin_rcvr.consume()
 470
 471 if __name__ == "__main__":
 472     main()
 473 """
 474