--- /dev/null
+#!/usr/bin/env python
+#
+# Copyright 2016-2017 AT&T Intellectual Property, Inc
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# What this is: Monitor and closed-loop policy agent as part of the OPNFV VES
+# ves_onap_demo.
+#
+# Status: this is a work in progress, under test.
+
+from rest_dispatcher import PathDispatcher, set_404_content
+from wsgiref.simple_server import make_server
+import sys
+import os
+import platform
+import traceback
+import time
+from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
+import ConfigParser
+import logging.handlers
+from base64 import b64decode
+import string
+import json
+import jsonschema
+from functools import partial
+import requests
+
+monitor_mode = "f"
+vdu_id = ['','','','','','']
+summary_e = ['***** Summary of key stats *****','','','']
+summary_c = ['Collectd agents:']
+status = ['','Started','Started','Started']
+base_url = ''
+template_404 = b'''POST {0}'''
+columns = 0
+rows = 0
+
+class JSONObject:
+ def __init__(self, d):
+ self.__dict__ = d
+
+__all__ = []
+__version__ = 0.1
+__date__ = '2015-12-04'
+__updated__ = '2015-12-04'
+
+TESTRUN = False
+DEBUG = False
+PROFILE = False
+
+#------------------------------------------------------------------------------
+# Address of influxdb server.
+#------------------------------------------------------------------------------
+
+influxdb = '127.0.0.1'
+
+#------------------------------------------------------------------------------
+# Credentials we expect clients to authenticate themselves with.
+#------------------------------------------------------------------------------
+vel_username = ''
+vel_password = ''
+
+#------------------------------------------------------------------------------
+# The JSON schema which we will use to validate events.
+#------------------------------------------------------------------------------
+vel_schema = None
+
+#------------------------------------------------------------------------------
+# The JSON schema which we will use to validate client throttle state.
+#------------------------------------------------------------------------------
+throttle_schema = None
+
+#------------------------------------------------------------------------------
+# The JSON schema which we will use to provoke throttling commands for testing.
+#------------------------------------------------------------------------------
+test_control_schema = None
+
+#------------------------------------------------------------------------------
+# Pending command list from the testControl API
+# This is sent as a response commandList to the next received event.
+#------------------------------------------------------------------------------
+pending_command_list = None
+
+#------------------------------------------------------------------------------
+# Logger for this module.
+#------------------------------------------------------------------------------
+logger = None
+
+def listener(environ, start_response, schema):
+ '''
+ Handler for the Vendor Event Listener REST API.
+
+ Extract headers and the body and check that:
+
+ 1) The client authenticated themselves correctly.
+ 2) The body validates against the provided schema for the API.
+
+ '''
+ logger.info('Got a Vendor Event request')
+ print('==== ' + time.asctime() + ' ' + '=' * 49)
+
+ #--------------------------------------------------------------------------
+ # Extract the content from the request.
+ #--------------------------------------------------------------------------
+ length = int(environ.get('CONTENT_LENGTH', '0'))
+ logger.debug('Content Length: {0}'.format(length))
+ body = environ['wsgi.input'].read(length)
+ logger.debug('Content Body: {0}'.format(body))
+
+ mode, b64_credentials = string.split(environ.get('HTTP_AUTHORIZATION',
+ 'None None'))
+ # logger.debug('Auth. Mode: {0} Credentials: {1}'.format(mode,
+ # b64_credentials))
+ logger.debug('Auth. Mode: {0} Credentials: ****'.format(mode))
+ if (b64_credentials != 'None'):
+ credentials = b64decode(b64_credentials)
+ else:
+ credentials = None
+
+ # logger.debug('Credentials: {0}'.format(credentials))
+ logger.debug('Credentials: ****')
+
+ #--------------------------------------------------------------------------
+ # If we have a schema file then check that the event matches that expected.
+ #--------------------------------------------------------------------------
+ if (schema is not None):
+ logger.debug('Attempting to validate data: {0}\n'
+ 'Against schema: {1}'.format(body, schema))
+ try:
+ decoded_body = json.loads(body)
+ jsonschema.validate(decoded_body, schema)
+ logger.info('Event is valid!')
+ print('Valid body decoded & checked against schema OK:\n'
+ '{0}'.format(json.dumps(decoded_body,
+ sort_keys=True,
+ indent=4,
+ separators=(',', ': '))))
+
+ except jsonschema.SchemaError as e:
+ logger.error('Schema is not valid! {0}'.format(e))
+ print('Schema is not valid! {0}'.format(e))
+
+ except jsonschema.ValidationError as e:
+ logger.warn('Event is not valid against schema! {0}'.format(e))
+ print('Event is not valid against schema! {0}'.format(e))
+ print('Bad JSON body decoded:\n'
+ '{0}'.format(json.dumps(decoded_body,
+ sort_keys=True,
+ indent=4,
+ separators=(',', ': '))))
+
+ except Exception as e:
+ logger.error('Event invalid for unexpected reason! {0}'.format(e))
+ print('Schema is not valid for unexpected reason! {0}'.format(e))
+ else:
+ logger.debug('No schema so just decode JSON: {0}'.format(body))
+ try:
+ decoded_body = json.loads(body)
+ print('Valid JSON body (no schema checking) decoded:\n'
+ '{0}'.format(json.dumps(decoded_body,
+ sort_keys=True,
+ indent=4,
+ separators=(',', ': '))))
+ logger.info('Event is valid JSON but not checked against schema!')
+
+ except Exception as e:
+ logger.error('Event invalid for unexpected reason! {0}'.format(e))
+ print('JSON body not valid for unexpected reason! {0}'.format(e))
+
+ #--------------------------------------------------------------------------
+ # See whether the user authenticated themselves correctly.
+ #--------------------------------------------------------------------------
+ if (credentials == (vel_username + ':' + vel_password)):
+ logger.debug('Authenticated OK')
+# print('Authenticated OK')
+
+ #----------------------------------------------------------------------
+ # Respond to the caller. If we have a pending commandList from the
+ # testControl API, send it in response.
+ #----------------------------------------------------------------------
+ global pending_command_list
+ if pending_command_list is not None:
+ start_response('202 Accepted',
+ [('Content-type', 'application/json')])
+ response = pending_command_list
+ pending_command_list = None
+
+ print('\n'+ '='*80)
+ print('Sending pending commandList in the response:\n'
+ '{0}'.format(json.dumps(response,
+ sort_keys=True,
+ indent=4,
+ separators=(',', ': '))))
+ print('='*80 + '\n')
+ yield json.dumps(response)
+ else:
+ start_response('202 Accepted', [])
+ yield ''
+ else:
+ logger.warn('Failed to authenticate OK; creds: ' + credentials)
+ print('Failed to authenticate agent credentials: ', credentials,
+ 'against expected ', vel_username, ':', vel_password)
+
+ #----------------------------------------------------------------------
+ # Respond to the caller.
+ #----------------------------------------------------------------------
+ start_response('401 Unauthorized', [ ('Content-type',
+ 'application/json')])
+ req_error = { 'requestError': {
+ 'policyException': {
+ 'messageId': 'POL0001',
+ 'text': 'Failed to authenticate'
+ }
+ }
+ }
+ yield json.dumps(req_error)
+
+ save_event(body)
+
+#--------------------------------------------------------------------------
+# Send event to influxdb
+#--------------------------------------------------------------------------
+def send_to_influxdb(event,pdata):
+ url = 'http://{}:8086/write?db=veseventsdb'.format(influxdb)
+ print('Send {} to influxdb at {}: {}'.format(event,influxdb,pdata))
+ r = requests.post(url, data=pdata, headers={'Content-Type': 'text/plain'})
+ print('influxdb return code {}'.format(r.status_code))
+ if r.status_code != 204:
+ print('*** Influxdb save failed, return code {} ***'.format(r.status_code))
+
+#--------------------------------------------------------------------------
+# Save event data
+#--------------------------------------------------------------------------
+def save_event(body):
+ jobj = json.loads(body)
+ e = json.loads(body, object_hook=JSONObject)
+
+ domain = jobj['event']['commonEventHeader']['domain']
+ timestamp = jobj['event']['commonEventHeader']['lastEpochMicrosec']
+ agent = jobj['event']['commonEventHeader']['reportingEntityName'].upper( )
+ if "LOCALHOST" in agent:
+ agent = "computehost"
+ source = jobj['event']['commonEventHeader']['sourceId'].upper( )
+
+ if e.event.commonEventHeader.domain == "heartbeat":
+ print('Found Heartbeat')
+ send_to_influxdb(event,'heartbeat,system={},sequence={}'.format(agent,e.event.commonEventHeader.sequence))
+
+ if 'measurementsForVfScalingFields' in jobj['event']:
+ print('Found measurementsForVfScalingFields')
+
+# "measurementsForVfScalingFields": {
+# "additionalMeasurements": [
+# {
+# "arrayOfFields": [
+# {
+# "name": "load-longterm",
+# "value": "0.34"
+# },
+# {
+# "name": "load-shortterm",
+# "value": "0.32"
+# },
+# {
+# "name": "load-midterm",
+# "value": "0.34"
+# }
+# ],
+# "name": "load"
+# }
+# ],
+
+ if 'additionalMeasurements' in jobj['event']['measurementsForVfScalingFields']:
+ for meas in jobj['event']['measurementsForVfScalingFields']['additionalMeasurements']:
+ name = meas['name']
+ pdata = '{},system={}'.format(name,source)
+ for field in meas['arrayOfFields']:
+ pdata = pdata + ",{}={}".format(field['name'],field['value'])
+ i=pdata.find(',', pdata.find('system'))
+ pdata = pdata[:i] + ' ' + pdata[i+1:]
+ send_to_influxdb("systemLoad", pdata)
+
+# "memoryUsageArray": [
+# {
+# "memoryBuffered": 269056.0,
+# "memoryCached": 17636956.0,
+# "memoryFree": 244731658240,
+# "memorySlabRecl": 753160.0,
+# "memorySlabUnrecl": 210800.0,
+# "memoryUsed": 6240064.0,
+# "vmIdentifier": "opnfv01"
+# }
+# ],
+
+ if 'memoryUsageArray' in jobj['event']['measurementsForVfScalingFields']:
+ print('Found memoryUsageArray')
+ pdata = 'memoryUsage,system={}'.format(source)
+ vmid=e.event.measurementsForVfScalingFields.memoryUsageArray[0].vmIdentifier
+ d = jobj['event']['measurementsForVfScalingFields']['memoryUsageArray'][0].items()
+ for key,val in d:
+ if key != 'vmIdentifier':
+ pdata = pdata + ',{}={}'.format(key,val)
+ i=pdata.find(',', pdata.find('system'))
+ pdata = pdata[:i] + ' ' + pdata[i+1:]
+ send_to_influxdb("memoryUsage", pdata)
+
+# "vNicPerformanceArray": [
+# {
+# "receivedDiscardedPacketsAccumulated": 0,
+# "receivedErrorPacketsAccumulated": 0,
+# "receivedOctetsAccumulated": 476.801524578,
+# "receivedTotalPacketsAccumulated": 2.90000899705,
+# "transmittedDiscardedPacketsAccumulated": 0,
+# "transmittedErrorPacketsAccumulated": 0,
+# "transmittedOctetsAccumulated": 230.100735749,
+# "transmittedTotalPacketsAccumulated": 1.20000372292,
+# "vNicIdentifier": "eno4",
+# "valuesAreSuspect": "true"
+# },
+
+ if 'vNicPerformanceArray' in jobj['event']['measurementsForVfScalingFields']:
+ print('Found vNicPerformanceArray')
+ for vnic in jobj['event']['measurementsForVfScalingFields']['vNicPerformanceArray']:
+ vnid=vnic['vNicIdentifier']
+ pdata = 'vNicPerformance,system={},vnic={}'.format(vmid,vnid)
+ d = vnic.items()
+ for key,val in d:
+ if key != 'vNicIdentifier':
+ pdata = pdata + ',{}={}'.format(key,val)
+ i=pdata.find(',', pdata.find('vnic'))
+ pdata = pdata[:i] + ' ' + pdata[i+1:]
+ send_to_influxdb("vNicPerformance", pdata)
+
+def test_listener(environ, start_response, schema):
+ '''
+ Handler for the Test Collector Test Control API.
+
+ There is no authentication on this interface.
+
+ This simply stores a commandList which will be sent in response to the next
+ incoming event on the EVEL interface.
+ '''
+ global pending_command_list
+ logger.info('Got a Test Control input')
+ print('============================')
+ print('==== TEST CONTROL INPUT ====')
+
+ #--------------------------------------------------------------------------
+ # GET allows us to get the current pending request.
+ #--------------------------------------------------------------------------
+ if environ.get('REQUEST_METHOD') == 'GET':
+ start_response('200 OK', [('Content-type', 'application/json')])
+ yield json.dumps(pending_command_list)
+ return
+
+ #--------------------------------------------------------------------------
+ # Extract the content from the request.
+ #--------------------------------------------------------------------------
+ length = int(environ.get('CONTENT_LENGTH', '0'))
+ logger.debug('TestControl Content Length: {0}'.format(length))
+ body = environ['wsgi.input'].read(length)
+ logger.debug('TestControl Content Body: {0}'.format(body))
+
+ #--------------------------------------------------------------------------
+ # If we have a schema file then check that the event matches that expected.
+ #--------------------------------------------------------------------------
+ if (schema is not None):
+ logger.debug('Attempting to validate data: {0}\n'
+ 'Against schema: {1}'.format(body, schema))
+ try:
+ decoded_body = json.loads(body)
+ jsonschema.validate(decoded_body, schema)
+ logger.info('TestControl is valid!')
+ print('TestControl:\n'
+ '{0}'.format(json.dumps(decoded_body,
+ sort_keys=True,
+ indent=4,
+ separators=(',', ': '))))
+
+ except jsonschema.SchemaError as e:
+ logger.error('TestControl Schema is not valid: {0}'.format(e))
+ print('TestControl Schema is not valid: {0}'.format(e))
+
+ except jsonschema.ValidationError as e:
+ logger.warn('TestControl input not valid: {0}'.format(e))
+ print('TestControl input not valid: {0}'.format(e))
+ print('Bad JSON body decoded:\n'
+ '{0}'.format(json.dumps(decoded_body,
+ sort_keys=True,
+ indent=4,
+ separators=(',', ': '))))
+
+ except Exception as e:
+ logger.error('TestControl input not valid: {0}'.format(e))
+ print('TestControl Schema not valid: {0}'.format(e))
+ else:
+ logger.debug('Missing schema just decode JSON: {0}'.format(body))
+ try:
+ decoded_body = json.loads(body)
+ print('Valid JSON body (no schema checking) decoded:\n'
+ '{0}'.format(json.dumps(decoded_body,
+ sort_keys=True,
+ indent=4,
+ separators=(',', ': '))))
+ logger.info('TestControl input not checked against schema!')
+
+ except Exception as e:
+ logger.error('TestControl input not valid: {0}'.format(e))
+ print('TestControl input not valid: {0}'.format(e))
+
+ #--------------------------------------------------------------------------
+ # Respond to the caller. If we received otherField 'ThrottleRequest',
+ # generate the appropriate canned response.
+ #--------------------------------------------------------------------------
+ pending_command_list = decoded_body
+ print('===== TEST CONTROL END =====')
+ print('============================')
+ start_response('202 Accepted', [])
+ yield ''
+
+def main(argv=None):
+ '''
+ Main function for the collector start-up.
+
+ Called with command-line arguments:
+ * --config *<file>*
+ * --section *<section>*
+ * --verbose
+
+ Where:
+
+ *<file>* specifies the path to the configuration file.
+
+ *<section>* specifies the section within that config file.
+
+ *verbose* generates more information in the log files.
+
+ The process listens for REST API invocations and checks them. Errors are
+ displayed to stdout and logged.
+ '''
+
+ if argv is None:
+ argv = sys.argv
+ else:
+ sys.argv.extend(argv)
+
+ program_name = os.path.basename(sys.argv[0])
+ program_version = 'v{0}'.format(__version__)
+ program_build_date = str(__updated__)
+ program_version_message = '%%(prog)s {0} ({1})'.format(program_version,
+ program_build_date)
+ if (__import__('__main__').__doc__ is not None):
+ program_shortdesc = __import__('__main__').__doc__.split('\n')[1]
+ else:
+ program_shortdesc = 'Running in test harness'
+ program_license = '''{0}
+
+ Created on {1}.
+ Copyright 2015 Metaswitch Networks Ltd. All rights reserved.
+
+ Distributed on an "AS IS" basis without warranties
+ or conditions of any kind, either express or implied.
+
+USAGE
+'''.format(program_shortdesc, str(__date__))
+
+ try:
+ #----------------------------------------------------------------------
+ # Setup argument parser so we can parse the command-line.
+ #----------------------------------------------------------------------
+ parser = ArgumentParser(description=program_license,
+ formatter_class=ArgumentDefaultsHelpFormatter)
+ parser.add_argument('-i', '--influxdb',
+ dest='influxdb',
+ default='localhost',
+ help='InfluxDB server addresss')
+ parser.add_argument('-v', '--verbose',
+ dest='verbose',
+ action='count',
+ help='set verbosity level')
+ parser.add_argument('-V', '--version',
+ action='version',
+ version=program_version_message,
+ help='Display version information')
+ parser.add_argument('-a', '--api-version',
+ dest='api_version',
+ default='5',
+ help='set API version')
+ parser.add_argument('-c', '--config',
+ dest='config',
+ default='/etc/opt/att/collector.conf',
+ help='Use this config file.',
+ metavar='<file>')
+ parser.add_argument('-s', '--section',
+ dest='section',
+ default='default',
+ metavar='<section>',
+ help='section to use in the config file')
+
+ #----------------------------------------------------------------------
+ # Process arguments received.
+ #----------------------------------------------------------------------
+ args = parser.parse_args()
+ verbose = args.verbose
+ api_version = args.api_version
+ config_file = args.config
+ config_section = args.section
+
+ #----------------------------------------------------------------------
+ # Now read the config file, using command-line supplied values as
+ # overrides.
+ #----------------------------------------------------------------------
+ defaults = {'log_file': 'collector.log',
+ 'vel_port': '12233',
+ 'vel_path': '',
+ 'vel_topic_name': ''
+ }
+ overrides = {}
+ config = ConfigParser.SafeConfigParser(defaults)
+ config.read(config_file)
+
+ #----------------------------------------------------------------------
+ # extract the values we want.
+ #----------------------------------------------------------------------
+ global influxdb
+ global vel_username
+ global vel_password
+ global vel_topic_name
+ influxdb = config.get(config_section, 'influxdb', vars=overrides)
+ log_file = config.get(config_section, 'log_file', vars=overrides)
+ vel_port = config.get(config_section, 'vel_port', vars=overrides)
+ vel_path = config.get(config_section, 'vel_path', vars=overrides)
+
+ vel_topic_name = config.get(config_section,
+ 'vel_topic_name',
+ vars=overrides)
+ vel_username = config.get(config_section,
+ 'vel_username',
+ vars=overrides)
+ vel_password = config.get(config_section,
+ 'vel_password',
+ vars=overrides)
+ vel_schema_file = config.get(config_section,
+ 'schema_file',
+ vars=overrides)
+ base_schema_file = config.get(config_section,
+ 'base_schema_file',
+ vars=overrides)
+ throttle_schema_file = config.get(config_section,
+ 'throttle_schema_file',
+ vars=overrides)
+ test_control_schema_file = config.get(config_section,
+ 'test_control_schema_file',
+ vars=overrides)
+
+ #----------------------------------------------------------------------
+ # Finally we have enough info to start a proper flow trace.
+ #----------------------------------------------------------------------
+ global logger
+ print('Logfile: {0}'.format(log_file))
+ logger = logging.getLogger('monitor')
+ if verbose > 0:
+ print('Verbose mode on')
+ logger.setLevel(logging.DEBUG)
+ else:
+ logger.setLevel(logging.INFO)
+ handler = logging.handlers.RotatingFileHandler(log_file,
+ maxBytes=1000000,
+ backupCount=10)
+ if (platform.system() == 'Windows'):
+ date_format = '%Y-%m-%d %H:%M:%S'
+ else:
+ date_format = '%Y-%m-%d %H:%M:%S.%f %z'
+ formatter = logging.Formatter('%(asctime)s %(name)s - '
+ '%(levelname)s - %(message)s',
+ date_format)
+ handler.setFormatter(formatter)
+ logger.addHandler(handler)
+ logger.info('Started')
+
+ #----------------------------------------------------------------------
+ # Log the details of the configuration.
+ #----------------------------------------------------------------------
+ logger.debug('Log file = {0}'.format(log_file))
+ logger.debug('Influxdb server = {0}'.format(influxdb))
+ logger.debug('Event Listener Port = {0}'.format(vel_port))
+ logger.debug('Event Listener Path = {0}'.format(vel_path))
+ logger.debug('Event Listener Topic = {0}'.format(vel_topic_name))
+ logger.debug('Event Listener Username = {0}'.format(vel_username))
+ # logger.debug('Event Listener Password = {0}'.format(vel_password))
+ logger.debug('Event Listener JSON Schema File = {0}'.format(
+ vel_schema_file))
+ logger.debug('Base JSON Schema File = {0}'.format(base_schema_file))
+ logger.debug('Throttle JSON Schema File = {0}'.format(
+ throttle_schema_file))
+ logger.debug('Test Control JSON Schema File = {0}'.format(
+ test_control_schema_file))
+
+ #----------------------------------------------------------------------
+ # Perform some basic error checking on the config.
+ #----------------------------------------------------------------------
+ if (int(vel_port) < 1024 or int(vel_port) > 65535):
+ logger.error('Invalid Vendor Event Listener port ({0}) '
+ 'specified'.format(vel_port))
+ raise RuntimeError('Invalid Vendor Event Listener port ({0}) '
+ 'specified'.format(vel_port))
+
+ if (len(vel_path) > 0 and vel_path[-1] != '/'):
+ logger.warning('Event Listener Path ({0}) should have terminating '
+ '"/"! Adding one on to configured string.'.format(
+ vel_path))
+ vel_path += '/'
+
+ #----------------------------------------------------------------------
+ # Load up the vel_schema, if it exists.
+ #----------------------------------------------------------------------
+ if not os.path.exists(vel_schema_file):
+ logger.warning('Event Listener Schema File ({0}) not found. '
+ 'No validation will be undertaken.'.format(
+ vel_schema_file))
+ else:
+ global vel_schema
+ global throttle_schema
+ global test_control_schema
+ vel_schema = json.load(open(vel_schema_file, 'r'))
+ logger.debug('Loaded the JSON schema file')
+
+ #------------------------------------------------------------------
+ # Load up the throttle_schema, if it exists.
+ #------------------------------------------------------------------
+ if (os.path.exists(throttle_schema_file)):
+ logger.debug('Loading throttle schema')
+ throttle_fragment = json.load(open(throttle_schema_file, 'r'))
+ throttle_schema = {}
+ throttle_schema.update(vel_schema)
+ throttle_schema.update(throttle_fragment)
+ logger.debug('Loaded the throttle schema')
+
+ #------------------------------------------------------------------
+ # Load up the test control _schema, if it exists.
+ #------------------------------------------------------------------
+ if (os.path.exists(test_control_schema_file)):
+ logger.debug('Loading test control schema')
+ test_control_fragment = json.load(
+ open(test_control_schema_file, 'r'))
+ test_control_schema = {}
+ test_control_schema.update(vel_schema)
+ test_control_schema.update(test_control_fragment)
+ logger.debug('Loaded the test control schema')
+
+ #------------------------------------------------------------------
+ # Load up the base_schema, if it exists.
+ #------------------------------------------------------------------
+ if (os.path.exists(base_schema_file)):
+ logger.debug('Updating the schema with base definition')
+ base_schema = json.load(open(base_schema_file, 'r'))
+ vel_schema.update(base_schema)
+ logger.debug('Updated the JSON schema file')
+
+ #----------------------------------------------------------------------
+ # We are now ready to get started with processing. Start-up the various
+ # components of the system in order:
+ #
+ # 1) Create the dispatcher.
+ # 2) Register the functions for the URLs of interest.
+ # 3) Run the webserver.
+ #----------------------------------------------------------------------
+ root_url = '/{0}eventListener/v{1}{2}'.\
+ format(vel_path,
+ api_version,
+ '/' + vel_topic_name
+ if len(vel_topic_name) > 0
+ else '')
+ throttle_url = '/{0}eventListener/v{1}/clientThrottlingState'.\
+ format(vel_path, api_version)
+ set_404_content(root_url)
+ dispatcher = PathDispatcher()
+ vendor_event_listener = partial(listener, schema = vel_schema)
+ dispatcher.register('GET', root_url, vendor_event_listener)
+ dispatcher.register('POST', root_url, vendor_event_listener)
+ vendor_throttle_listener = partial(listener, schema = throttle_schema)
+ dispatcher.register('GET', throttle_url, vendor_throttle_listener)
+ dispatcher.register('POST', throttle_url, vendor_throttle_listener)
+
+ #----------------------------------------------------------------------
+ # We also add a POST-only mechanism for test control, so that we can
+ # send commands to a single attached client.
+ #----------------------------------------------------------------------
+ test_control_url = '/testControl/v{0}/commandList'.format(api_version)
+ test_control_listener = partial(test_listener,
+ schema = test_control_schema)
+ dispatcher.register('POST', test_control_url, test_control_listener)
+ dispatcher.register('GET', test_control_url, test_control_listener)
+
+ httpd = make_server('', int(vel_port), dispatcher)
+ print('Serving on port {0}...'.format(vel_port))
+ httpd.serve_forever()
+
+ logger.error('Main loop exited unexpectedly!')
+ return 0
+
+ except KeyboardInterrupt:
+ #----------------------------------------------------------------------
+ # handle keyboard interrupt
+ #----------------------------------------------------------------------
+ logger.info('Exiting on keyboard interrupt!')
+ return 0
+
+ except Exception as e:
+ #----------------------------------------------------------------------
+ # Handle unexpected exceptions.
+ #----------------------------------------------------------------------
+ if DEBUG or TESTRUN:
+ raise(e)
+ indent = len(program_name) * ' '
+ sys.stderr.write(program_name + ': ' + repr(e) + '\n')
+ sys.stderr.write(indent + ' for help use --help\n')
+ sys.stderr.write(traceback.format_exc())
+ logger.critical('Exiting because of exception: {0}'.format(e))
+ logger.critical(traceback.format_exc())
+ return 2
+
+#------------------------------------------------------------------------------
+# MAIN SCRIPT ENTRY POINT.
+#------------------------------------------------------------------------------
+if __name__ == '__main__':
+ if TESTRUN:
+ #----------------------------------------------------------------------
+ # Running tests - note that doctest comments haven't been included so
+ # this is a hook for future improvements.
+ #----------------------------------------------------------------------
+ import doctest
+ doctest.testmod()
+
+ if PROFILE:
+ #----------------------------------------------------------------------
+ # Profiling performance. Performance isn't expected to be a major
+ # issue, but this should all work as expected.
+ #----------------------------------------------------------------------
+ import cProfile
+ import pstats
+ profile_filename = 'collector_profile.txt'
+ cProfile.run('main()', profile_filename)
+ statsfile = open('collector_profile_stats.txt', 'wb')
+ p = pstats.Stats(profile_filename, stream=statsfile)
+ stats = p.strip_dirs().sort_stats('cumulative')
+ stats.print_stats()
+ statsfile.close()
+ sys.exit(0)
+
+ #--------------------------------------------------------------------------
+ # Normal operation - call through to the main function.
+ #--------------------------------------------------------------------------
+ sys.exit(main())
# See the License for the specific language governing permissions and
# limitations under the License.
#
-#. What this is: Setup script for VES agent framework.
+#. What this is: Setup script for the VES monitoring framework.
#. With this script VES support can be installed in one or more hosts, with:
#. - a dedicated or shared Kafka server for collection of events from collectd
#. - VES collectd agents running in host or guest mode
#. pre-installed VES collector e.g. from the ONAP project.
#. - Install Kafka server on one of the hosts, or use a pre-installed server
#. accessible from the agent hosts.
+#. - Install collectd on each host.
#. - Install the VES agent on each host.
#. - As needed, install the VES agent on each virtual host. This could include
#. pre-installed VES agents in VM or container images, which are configured
#. ves_kafka_host: kafka host IP or hostname (default: 127.0.0.1)
#.
#. Usage:
-#. wget https://raw.githubusercontent.com/opnfv/ves/master/tools/ves-setup.sh
-#. bash ves-setup.sh <collector|kafka|agent>
+#. git clone https://gerrit.opnfv.org/gerrit/ves /tmp/ves
+#. bash /tmp/ves/ves-setup.sh <collector|kafka|collectd|agent>
#. collector: setup VES collector (test collector)
#. kafka: setup kafka server for VES events from collect agent(s)
-#. agent: setup VES agent in host or guest mode
-#. Recommended sequence is:
-#. ssh into your collector host and run these commands:
-#. $ ves_host=$(ip route get 8.8.8.8 | awk '{print $NF; exit}')
-#. $ export ves_host
-#. $ bash ves-setup.sh collector
-#. ...then for each agent host:
-#. copy ~/ves_env.sh and ves-setup.sh to the host e.g. via scp
-#. ssh into the host and run, directly or via ssh -x
-#. $ bash ves-setup.sh kafka
-#. $ bash ves-setup.sh agent
+#. collectd: setup collectd with libvirt plugin, as a kafka publisher
+#. agent: setup VES agent in host or guest mode, as a kafka consumer
+#. See demo_deploy.sh in this repo for a recommended sequence of the above.
#.
#. Status: this is a work in progress, under test.
function common_prereqs() {
log "install common prerequisites"
- sudo apt-get install -y default-jre
- sudo apt-get install -y zookeeperd
- sudo apt-get install -y python-pip
- sudo pip install kafka-python
+ if [[ ! -f /.dockerenv ]]; then dosudo="sudo"; fi
+ $dosudo apt-get update
+ $dosudo apt-get install -y git
+ # Required for kafka
+ $dosudo apt-get install -y default-jre
+ $dosudo apt-get install -y zookeeperd
+ $dosudo apt-get install -y python-pip
+ $dosudo pip install kafka-python
+ # Required for building collectd
+ $dosudo apt-get install -y pkg-config
}
function setup_env() {
- if [[ ! -f ~/ves_env.sh ]]; then
- cat <<EOF >~/ves_env.sh
+ if [[ ! -d /tmp/ves ]]; then mkdir /tmp/ves; fi
+ cp $0 /tmp/ves
+ if [[ ! -f /tmp/ves/ves_env.sh ]]; then
+ cat <<EOF >/tmp/ves/ves_env.sh
#!/bin/bash
ves_mode="${ves_mode:=host}"
ves_host="${ves_host:=127.0.0.1}"
+ves_hostname="${ves_hostname:=localhost}"
ves_port="${ves_port:=30000}"
ves_path="${ves_path:=}"
ves_topic="${ves_topic:=}"
ves_kafka_port="${ves_kafka_port:=9092}"
export ves_mode
export ves_host
+export ves_hostname
export ves_port
export ves_path
export ves_topic
export ves_user
export ves_pass
export ves_interval
-export ves_kafka_host
-export ves_port
+export ves_kafka_hostame
export ves_kafka_port
EOF
fi
- source ~/ves_env.sh
+ source /tmp/ves/ves_env.sh
}
function setup_kafka() {
log "setup kafka server"
common_prereqs
+ setup_env
- log "get and unpack kafka_2.11-0.11.0.0.tgz"
- wget "http://www-eu.apache.org/dist/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz"
- tar -xvzf kafka_2.11-0.11.0.0.tgz
+ cd
+ ver="0.11.0.2"
+ log "get and unpack kafka_2.11-$ver.tgz"
+ wget "http://www-eu.apache.org/dist/kafka/$ver/kafka_2.11-$ver.tgz"
+ tar -xvzf kafka_2.11-$ver.tgz
log "set delete.topic.enable=true"
sed -i -- 's/#delete.topic.enable=true/delete.topic.enable=true/' \
- kafka_2.11-0.11.0.0/config/server.properties
- grep delete.topic.enable kafka_2.11-0.11.0.0/config/server.properties
+ kafka_2.11-$ver/config/server.properties
+ grep delete.topic.enable kafka_2.11-$ver/config/server.properties
# TODO: Barometer VES guide to clarify hostname must be in /etc/hosts
- sudo nohup kafka_2.11-0.11.0.0/bin/kafka-server-start.sh \
- kafka_2.11-0.11.0.0/config/server.properties \
- > kafka_2.11-0.11.0.0/kafka.log 2>&1 &
- # TODO: find a test that does not hang the script at
- # echo "Hello, World" | ~/kafka_2.11-0.11.0.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TopicTest > /dev/null
- # ~/kafka_2.11-0.11.0.0/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic TopicTest --from-beginning
+ sudo nohup kafka_2.11-$ver/bin/kafka-server-start.sh \
+ kafka_2.11-$ver/config/server.properties \
+ > kafka_2.11-$ver/kafka.log 2>&1 &
}
-function setup_agent() {
- log "setup agent"
- common_prereqs
-
- log "cleanup any previous failed install"
- sudo rm -rf ~/collectd-virt
- sudo rm -rf ~/librdkafka
- sudo rm -rf ~/collectd
-
+function setup_kafka_client() {
log "Install Apache Kafka C/C++ client library"
- sudo apt-get install -y build-essential
+ if [[ ! -f /.dockerenv ]]; then dosudo="sudo"; fi
+ $dosudo apt-get install -y build-essential
git clone https://github.com/edenhill/librdkafka.git ~/librdkafka
cd ~/librdkafka
git checkout -b v0.9.5 v0.9.5
# TODO: Barometer VES guide to clarify specific prerequisites for Ubuntu
- sudo apt-get install -y libpthread-stubs0-dev
- sudo apt-get install -y libssl-dev
- sudo apt-get install -y libsasl2-dev
- sudo apt-get install -y liblz4-dev
+ $dosudo apt-get install -y libpthread-stubs0-dev
+ $dosudo apt-get install -y libssl-dev
+ $dosudo apt-get install -y libsasl2-dev
+ $dosudo apt-get install -y liblz4-dev
./configure --prefix=/usr
make
- sudo make install
+ $dosudo make install
+}
+
+function setup_collectd() {
+ log "setup collectd"
+ common_prereqs
+ setup_env
+
+ log "cleanup any previous failed install"
+ sudo rm -rf ~/collectd-virt
+ sudo rm -rf ~/librdkafka
+ sudo rm -rf ~/collectd
+
+ setup_kafka_client
log "Build collectd with Kafka support"
git clone https://github.com/collectd/collectd.git ~/collectd
sudo systemctl daemon-reload
sudo systemctl start collectd.service
- log "install VES agent prerequisites"
- sudo pip install pyyaml
-
- log "clone OPNFV Barometer"
- git clone https://gerrit.opnfv.org/gerrit/barometer ~/barometer
-
- log "setup ves_app_config.conf"
- cd ~/barometer/3rd_party/collectd-ves-app/ves_app
- cat <<EOF >ves_app_config.conf
-[config]
-Domain = $ves_host
-Port = $ves_port
-Path = $ves_path
-Topic = $ves_topic
-UseHttps = $ves_https
-Username = $ves_user
-Password = $ves_pass
-SendEventInterval = $ves_interval
-ApiVersion = $ves_version
-KafkaPort = $ves_kafka_port
-KafkaBroker = $ves_kafka_host
-EOF
-
log "setup VES collectd config for VES $ves_mode mode"
if [[ "$ves_mode" == "host" ]]; then
# TODO: Barometer VES guide to clarify prerequisites install for Ubuntu
# http://docs.opnfv.org/en/latest/submodules/barometer/docs/release/userguide/feature.userguide.html#virt-plugin
sudo systemctl start libvirtd
- git clone https://github.com/maryamtahhan/collectd ~/collectd-virt
- cd ~/collectd-virt
+ rm -rf /tmp/ves/collectd-virt
+ git clone https://github.com/maryamtahhan/collectd /tmp/ves/collectd-virt
+ cd /tmp/ves/collectd-virt
./build.sh
./configure --enable-syslog --enable-logfile --enable-debug
make
sudo make install
-
- # TODO: Barometer VES guide refers to "At least one VM instance should be
- # up and running by hypervisor on the host." The process needs to accomodate
- # pre-installation of the VES agent *prior* to the first VM being created.
-
cat <<EOF | sudo tee -a /opt/collectd/etc/collectd.conf
+# for VES plugin
LoadPlugin logfile
<Plugin logfile>
LogLevel info
EOF
else
cat <<EOF | sudo tee -a /opt/collectd/etc/collectd.conf
+# for VES plugin
LoadPlugin logfile
<Plugin logfile>
LogLevel info
LoadPlugin cpu
-#LoadPlugin virt
-#<Plugin virt>
-# Connection "qemu:///system"
-# RefreshInterval 60
-# HostnameFormat uuid
-# PluginInstanceFormat name
-# ExtraStats "cpu_util"
-#</Plugin>
-
LoadPlugin write_kafka
<Plugin write_kafka>
Property "metadata.broker.list" "$ves_kafka_host:$ves_kafka_port"
EOF
fi
+ if [[ $(grep -c $ves_hostname /etc/hosts) -eq 0 ]]; then
+ log "add to /etc/hosts: $ves_kafka_host $ves_hostname"
+ echo "$ves_kafka_host $ves_hostname" | sudo tee -a /etc/hosts
+
log "restart collectd to apply updated config"
sudo systemctl restart collectd
+}
- log "start VES agent"
- cd ~/barometer/3rd_party/collectd-ves-app/ves_app
- nohup python ves_app.py \
- --events-schema=$ves_mode.yaml \
- --config=ves_app_config.conf > ~/ves_app.stdout.log 2>&1 &
+function setup_agent() {
+ log "setup VES agent"
+ if [[ ! -f /.dockerenv ]]; then
+ log "start the ves-agent container"
+ sudo docker run -it -d -v /tmp/ves:/opt/ves --name=ves-agent \
+ ubuntu:xenial /bin/bash
+ log "execute the agent setup script in the container"
+ sudo docker exec ves-agent /bin/bash /opt/ves/ves-setup.sh agent
+ else
+ common_prereqs
+ log "setup the VES environment"
+ source /opt/ves/ves_env.sh
+ log "install agent prerequisites"
+ pip install pyaml
+
+ setup_kafka_client
+
+ log "clone OPNFV Barometer"
+ rm -rf /opt/ves/barometer
+ git clone https://gerrit.opnfv.org/gerrit/barometer /opt/ves/barometer
+
+ log "setup ves_app_config.conf"
+ source /opt/ves/ves_env.sh
+ cd /opt/ves/barometer/3rd_party/collectd-ves-app/ves_app
+ cat <<EOF >ves_app_config.conf
+[config]
+Domain = $ves_host
+Port = $ves_port
+Path = $ves_path
+Topic = $ves_topic
+UseHttps = $ves_https
+Username = $ves_user
+Password = $ves_pass
+SendEventInterval = $ves_interval
+ApiVersion = $ves_version
+KafkaPort = $ves_kafka_port
+KafkaBroker = $ves_kafka_host
+EOF
+
+# log "add guest.yaml measurements to host.yaml (enables actual host data)"
+# tail --lines=+24 guest.yaml >>host.yaml
+
+ log "start VES agent"
+ echo "$ves_kafka_host $ves_hostname">>/etc/hosts
+ nohup python ves_app.py --events-schema=$ves_mode.yaml --loglevel ERROR \
+ --config=ves_app_config.conf > /opt/ves/ves_app.stdout.log 2>&1 &
+ fi
}
function setup_collector() {
log "install prerequistes"
sudo apt-get install -y jq
+ ves_hostname=$HOSTNAME
+ export ves_hostname
ves_host=$(ip route get 8.8.8.8 | awk '{print $NF; exit}')
export ves_host
setup_env
- echo "cleanup any earlier install attempts"
- sudo docker stop influxdb
- sudo docker rm influxdb
- sudo docker stop grafana
- sudo docker rm grafana
- sudo docker stop ves-collector
- sudo docker rm -v ves-collector
- sudo rm -rf /tmp/ves
-
- log "clone OPNFV VES"
- git clone https://gerrit.opnfv.org/gerrit/ves /tmp/ves
-
log "setup influxdb container"
- sudo docker run -d --name=influxdb -p 8086:8086 influxdb
- status=$(sudo docker inspect influxdb | jq -r '.[0].State.Status')
+ sudo docker run -d --name=ves-influxdb -p 8086:8086 influxdb
+ status=$(sudo docker inspect ves-influxdb | jq -r '.[0].State.Status')
while [[ "x$status" != "xrunning" ]]; do
log "InfluxDB container state is ($status)"
sleep 10
- status=$(sudo docker inspect influxdb | jq -r '.[0].State.Status')
+ status=$(sudo docker inspect ves-influxdb | jq -r '.[0].State.Status')
done
log "InfluxDB container state is $status"
--data-urlencode "q=CREATE DATABASE veseventsdb"
log "install Grafana container"
- sudo docker run -d --name grafana -p 3000:3000 grafana/grafana
- status=$(sudo docker inspect grafana | jq -r '.[0].State.Status')
+ sudo docker run -d --name ves-grafana -p 3001:3000 grafana/grafana
+ status=$(sudo docker inspect ves-grafana | jq -r '.[0].State.Status')
while [[ "x$status" != "xrunning" ]]; do
log "Grafana container state is ($status)"
sleep 10
- status=$(sudo docker inspect grafana | jq -r '.[0].State.Status')
+ status=$(sudo docker inspect ves-grafana | jq -r '.[0].State.Status')
done
log "Grafana container state is $status"
log "wait for Grafana API to be active"
- while ! curl http://$ves_host:3000 ; do
+ while ! curl http://$ves_host:3001 ; do
log "Grafana API is not yet responding... waiting 10 seconds"
sleep 10
done
log "add VESEvents datasource to Grafana"
- cat <<EOF >datasource.json
+ cat <<EOF >/tmp/ves/datasource.json
{ "name":"VESEvents",
"type":"influxdb",
"access":"direct",
EOF
curl -H "Accept: application/json" -H "Content-type: application/json" \
- -X POST -d @datasource.json \
- http://admin:admin@$ves_host:3000/api/datasources
+ -X POST -d @/tmp/ves/datasource.json \
+ http://admin:admin@$ves_host:3001/api/datasources
log "add VES dashboard to Grafana"
curl -H "Accept: application/json" -H "Content-type: application/json" \
-X POST \
- -d @/tmp/ves/tests/onap-demo/blueprints/tosca-vnfd-onap-demo/Dashboard.json\
- http://admin:admin@$ves_host:3000/api/dashboards/db
+ -d @/tmp/ves/tools/grafana/Dashboard.json\
+ http://admin:admin@$ves_host:3001/api/dashboards/db
log "setup collector container"
cd /tmp/ves
touch monitor.log
+ rm -rf /tmp/ves/evel-test-collector
git clone https://github.com/att/evel-test-collector.git
sed -i -- \
"s~log_file = /var/log/att/collector.log~log_file = /opt/ves/collector.log~" \
evel-test-collector/config/collector.conf
sed -i -- "s~vel_topic_name = example_vnf~vel_topic_name = $ves_topic~g" \
evel-test-collector/config/collector.conf
+ sed -i -- "/vel_topic_name = /a influxdb = $ves_host" \
+ evel-test-collector/config/collector.conf
- cp tests/onap-demo/blueprints/tosca-vnfd-onap-demo/monitor.py \
+ cp /tmp/ves/tools/monitor.py \
evel-test-collector/code/collector/monitor.py
# Note below: python (2.7) is required due to dependency on module 'ConfigParser'
python /opt/ves/evel-test-collector/code/collector/monitor.py \
--config /opt/ves/evel-test-collector/config/collector.conf \
--influxdb $ves_host \
---section default > /opt/ves/monitor.log
+--section default > /opt/ves/monitor.log 2>&1 &
EOF
sudo docker run -it -d -v /tmp/ves:/opt/ves --name=ves-collector \
-p 30000:30000 ubuntu:xenial /bin/bash
- sudo docker exec -it -d ves-collector bash /opt/ves/setup-collector.sh
+ sudo docker exec ves-collector /bin/bash /opt/ves/setup-collector.sh
# debug hints
# sudo docker exec -it ves-collector apt-get install -y tcpdump
# sudo docker exec -it ves-collector tcpdump -A -v -s 0 -i any port 30000
# curl http://$ves_host:30000
# sudo docker exec -it ves-collector /bin/bash
- # ~/kafka_2.11-0.11.0.0/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic collectd
+ # ~/kafka_2.11-0.11.0.2/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic collectd
+}
+
+function clean() {
+ log "clean installation for $1 at $2"
+ if [[ "$1" == "master" ]]; then
+ cs="ves-agent ves-collector ves-grafana ves-influxdb"
+ for c in $cs; do
+ log "stop and remove container $c"
+ sudo docker stop $c
+ sudo docker rm -v $c
+ done
+ fi
+ log "remove collectd config for VES"
+ sudo sed -i -- '/VES plugin/,$d' /opt/collectd/etc/collectd.conf
+ sudo systemctl restart collectd
+ sudo rm -rf /tmp/ves
}
dist=`grep DISTRIB_ID /etc/*-release | awk -F '=' '{print $2}'`
-setup_env
if [[ $(grep -c $HOSTNAME /etc/hosts) -eq 0 ]]; then
echo "$(ip route get 8.8.8.8 | awk '{print $NF; exit}') $HOSTNAME" |\
sudo tee -a /etc/hosts
fi
case "$1" in
+ "collectd")
+ setup_collectd
+ ;;
"agent")
setup_agent
;;
"kafka")
setup_kafka
;;
+ "clean")
+ clean $2 $3
+ ;;
*)
grep '#. ' $0
esac
Code Review / ves.git / commitdiff