aboutsummaryrefslogtreecommitdiffstats
path: root/3rd_party/collectd-ves-app/ves_app/ves_app.py
diff options
context:
space:
mode:
Diffstat (limited to '3rd_party/collectd-ves-app/ves_app/ves_app.py')
-rw-r--r--3rd_party/collectd-ves-app/ves_app/ves_app.py939
1 files changed, 104 insertions, 835 deletions
diff --git a/3rd_party/collectd-ves-app/ves_app/ves_app.py b/3rd_party/collectd-ves-app/ves_app/ves_app.py
index 49e7635a..105c66e2 100644
--- a/3rd_party/collectd-ves-app/ves_app/ves_app.py
+++ b/3rd_party/collectd-ves-app/ves_app/ves_app.py
@@ -22,522 +22,23 @@ import argparse
from distutils.util import strtobool
from kafka import KafkaConsumer
+from normalizer import Normalizer
+from normalizer import CollectdValue
+
try:
# For Python 3.0 and later
import urllib.request as url
except ImportError:
# Fall back to Python 2's urllib2
import urllib2 as url
-import socket
-import time
-from threading import Timer
-from threading import Lock
-
-
-class Event(object):
- """Event header"""
-
- def __init__(self):
- """Construct the common header"""
- self.version = 2.0
- self.event_type = "Info" # use "Info" unless a notification is generated
- self.domain = ""
- self.event_id = ""
- self.source_id = ""
- self.source_name = ""
- self.functional_role = ""
- self.reporting_entity_id = ""
- self.reporting_entity_name = ""
- self.priority = "Normal" # will be derived from event if there is one
- self.start_epoch_microsec = 0
- self.last_epoch_micro_sec = 0
- self.sequence = 0
- self.event_name = ""
- self.internal_header_fields = {}
- self.nfc_naming_code = ""
- self.nf_naming_code = ""
-
- def get_json(self):
- """Get the object of the datatype"""
- obj = {}
- obj['version'] = self.version
- obj['eventType'] = self.event_type
- obj['domain'] = self.domain
- obj['eventId'] = self.event_id
- obj['sourceId'] = self.source_id
- obj['sourceName'] = self.source_name
- obj['reportingEntityId'] = self.reporting_entity_id
- obj['reportingEntityName'] = self.reporting_entity_name
- obj['priority'] = self.priority
- obj['startEpochMicrosec'] = self.start_epoch_microsec
- obj['lastEpochMicrosec'] = self.last_epoch_micro_sec
- obj['sequence'] = self.sequence
- obj['eventName'] = self.event_name
- obj['internalHeaderFields'] = self.internal_header_fields
- obj['nfcNamingCode'] = self.nfc_naming_code
- obj['nfNamingCode'] = self.nf_naming_code
- return json.dumps({
- 'event': {
- 'commonEventHeader': obj,
- self.get_name(): self.get_obj()
- }
- }).encode()
-
- def get_name():
- assert False, 'abstract method get_name() is not implemented'
-
- def get_obj():
- assert False, 'abstract method get_obj() is not implemented'
-
-
-class Field(object):
- """field datatype"""
-
- def __init__(self, name, value):
- self.name = name
- self.value = value
-
- def get_obj(self):
- return {
- 'name': self.name,
- 'value': self.value
- }
-
-
-class NamedArrayOfFields(object):
- """namedArrayOfFields datatype"""
-
- def __init__(self, name):
- self.name = name
- self.array_of_fields = []
-
- def add(self, field):
- self.array_of_fields.append(field.get_obj())
-
- def get_obj(self):
- return {
- 'name': self.name,
- 'arrayOfFields': self.array_of_fields
- }
-
-
-class VESDataType(object):
- """ Base VES datatype """
-
- def set_optional(self, obj, key, val):
- if val is not None:
- obj[key] = val
-
-
-class DiskUsage(VESDataType):
- """diskUsage datatype"""
- def __init__(self, identifier):
- self.disk_identifier = identifier
- self.disk_io_time_avg = None
- self.disk_io_time_last = None
- self.disk_io_time_max = None
- self.disk_io_time_min = None
- self.disk_merged_read_avg = None
- self.disk_merged_read_last = None
- self.disk_merged_read_max = None
- self.disk_merged_read_min = None
- self.disk_merged_write_avg = None
- self.disk_merged_write_last = None
- self.disk_merged_write_max = None
- self.disk_merged_write_min = None
- self.disk_octets_read_avg = None
- self.disk_octets_read_last = None
- self.disk_octets_read_max = None
- self.disk_octets_read_min = None
- self.disk_octets_write_avg = None
- self.disk_octets_write_last = None
- self.disk_octets_write_max = None
- self.disk_octets_write_min = None
- self.disk_ops_read_avg = None
- self.disk_ops_read_last = None
- self.disk_ops_read_max = None
- self.disk_ops_read_min = None
- self.disk_ops_write_avg = None
- self.disk_ops_write_last = None
- self.disk_ops_write_max = None
- self.disk_ops_write_min = None
- self.disk_pending_operations_avg = None
- self.disk_pending_operations_last = None
- self.disk_pending_operations_max = None
- self.disk_pending_operations_min = None
- self.disk_time_read_avg = None
- self.disk_time_read_last = None
- self.disk_time_read_max = None
- self.disk_time_read_min = None
- self.disk_time_write_avg = None
- self.disk_time_write_last = None
- self.disk_time_write_max = None
- self.disk_time_write_min = None
-
- def get_obj(self):
- obj = {
- # required
- 'diskIdentifier': self.disk_identifier
- }
- self.set_optional(obj, 'diskIoTimeAvg', self.disk_io_time_avg)
- self.set_optional(obj, 'diskIoTimeLast', self.disk_io_time_last)
- self.set_optional(obj, 'diskIoTimeMax', self.disk_io_time_max)
- self.set_optional(obj, 'diskIoTimeMin', self.disk_io_time_min)
- self.set_optional(obj, 'diskMergedReadAvg', self.disk_merged_read_avg)
- self.set_optional(obj, 'diskMergedReadLast', self.disk_merged_read_last)
- self.set_optional(obj, 'diskMergedReadMax', self.disk_merged_read_max)
- self.set_optional(obj, 'diskMergedReadMin', self.disk_merged_read_min)
- self.set_optional(obj, 'diskMergedWriteAvg', self.disk_merged_write_avg)
- self.set_optional(obj, 'diskMergedWriteLast', self.disk_merged_write_last)
- self.set_optional(obj, 'diskMergedWriteMax', self.disk_merged_write_max)
- self.set_optional(obj, 'diskMergedWriteMin', self.disk_merged_write_min)
- self.set_optional(obj, 'diskOctetsReadAvg', self.disk_octets_read_avg)
- self.set_optional(obj, 'diskOctetsReadLast', self.disk_octets_read_last)
- self.set_optional(obj, 'diskOctetsReadMax', self.disk_octets_read_max)
- self.set_optional(obj, 'diskOctetsReadMin', self.disk_octets_read_min)
- self.set_optional(obj, 'diskOctetsWriteAvg', self.disk_octets_write_avg)
- self.set_optional(obj, 'diskOctetsWriteLast', self.disk_octets_write_last)
- self.set_optional(obj, 'diskOctetsWriteMax', self.disk_octets_write_max)
- self.set_optional(obj, 'diskOctetsWriteMin', self.disk_octets_write_min)
- self.set_optional(obj, 'diskOpsReadAvg', self.disk_ops_read_avg)
- self.set_optional(obj, 'diskOpsReadLast', self.disk_ops_read_last)
- self.set_optional(obj, 'diskOpsReadMax', self.disk_ops_read_max)
- self.set_optional(obj, 'diskOpsReadMin', self.disk_ops_read_min)
- self.set_optional(obj, 'diskOpsWriteAvg', self.disk_ops_write_avg)
- self.set_optional(obj, 'diskOpsWriteLast', self.disk_ops_write_last)
- self.set_optional(obj, 'diskOpsWriteMax', self.disk_ops_write_max)
- self.set_optional(obj, 'diskOpsWriteMin', self.disk_ops_write_min)
- self.set_optional(obj, 'diskPendingOperationsAvg', self.disk_pending_operations_avg)
- self.set_optional(obj, 'diskPendingOperationsLast', self.disk_pending_operations_last)
- self.set_optional(obj, 'diskPendingOperationsMax', self.disk_pending_operations_max)
- self.set_optional(obj, 'diskPendingOperationsMin', self.disk_pending_operations_min)
- self.set_optional(obj, 'diskTimeReadAvg', self.disk_time_read_avg)
- self.set_optional(obj, 'diskTimeReadLast', self.disk_time_read_last)
- self.set_optional(obj, 'diskTimeReadMax', self.disk_time_read_max)
- self.set_optional(obj, 'diskTimeReadMin', self.disk_time_read_min)
- self.set_optional(obj, 'diskTimeWriteAvg', self.disk_time_write_avg)
- self.set_optional(obj, 'diskTimeWriteLast', self.disk_time_write_last)
- self.set_optional(obj, 'diskTimeWriteMax', self.disk_time_write_max)
- self.set_optional(obj, 'diskTimeWriteMin', self.disk_time_write_min)
- return obj
-
-
-class VNicPerformance(VESDataType):
- """vNicPerformance datatype"""
-
- def __init__(self, identifier):
- self.received_broadcast_packets_accumulated = None
- self.received_broadcast_packets_delta = None
- self.received_discarded_packets_accumulated = None
- self.received_discarded_packets_delta = None
- self.received_error_packets_accumulated = None
- self.received_error_packets_delta = None
- self.received_multicast_packets_accumulated = None
- self.received_multicast_packets_delta = None
- self.received_octets_accumulated = None
- self.received_octets_delta = None
- self.received_total_packets_accumulated = None
- self.received_total_packets_delta = None
- self.received_unicast_packets_accumulated = None
- self.received_unicast_packets_delta = None
- self.transmitted_broadcast_packets_accumulated = None
- self.transmitted_broadcast_packets_delta = None
- self.transmitted_discarded_packets_accumulated = None
- self.transmitted_discarded_packets_delta = None
- self.transmitted_error_packets_accumulated = None
- self.transmitted_error_packets_delta = None
- self.transmitted_multicast_packets_accumulated = None
- self.transmitted_multicast_packets_delta = None
- self.transmitted_octets_accumulated = None
- self.transmitted_octets_delta = None
- self.transmitted_total_packets_accumulated = None
- self.transmitted_total_packets_delta = None
- self.transmitted_unicast_packets_accumulated = None
- self.transmitted_unicast_packets_delta = None
- self.values_are_suspect = 'true'
- self.v_nic_identifier = identifier
-
- def get_obj(self):
- obj = {
- # required
- 'valuesAreSuspect': self.values_are_suspect,
- 'vNicIdentifier': self.v_nic_identifier
- }
- # optional
- self.set_optional(obj, 'receivedBroadcastPacketsAccumulated', self.received_broadcast_packets_accumulated)
- self.set_optional(obj, 'receivedBroadcastPacketsDelta', self.received_broadcast_packets_delta)
- self.set_optional(obj, 'receivedDiscardedPacketsAccumulated', self.received_discarded_packets_accumulated)
- self.set_optional(obj, 'receivedDiscardedPacketsDelta', self.received_discarded_packets_delta)
- self.set_optional(obj, 'receivedErrorPacketsAccumulated', self.received_error_packets_accumulated)
- self.set_optional(obj, 'receivedErrorPacketsDelta', self.received_error_packets_delta)
- self.set_optional(obj, 'receivedMulticastPacketsAccumulated', self.received_multicast_packets_accumulated)
- self.set_optional(obj, 'receivedMulticastPacketsDelta', self.received_multicast_packets_delta)
- self.set_optional(obj, 'receivedOctetsAccumulated', self.received_octets_accumulated)
- self.set_optional(obj, 'receivedOctetsDelta', self.received_octets_delta)
- self.set_optional(obj, 'receivedTotalPacketsAccumulated', self.received_total_packets_accumulated)
- self.set_optional(obj, 'receivedTotalPacketsDelta', self.received_total_packets_delta)
- self.set_optional(obj, 'receivedUnicastPacketsAccumulated', self.received_unicast_packets_accumulated)
- self.set_optional(obj, 'receivedUnicastPacketsDelta', self.received_unicast_packets_delta)
- self.set_optional(obj, 'transmittedBroadcastPacketsAccumulated', self.transmitted_broadcast_packets_accumulated)
- self.set_optional(obj, 'transmittedBroadcastPacketsDelta', self.transmitted_broadcast_packets_delta)
- self.set_optional(obj, 'transmittedDiscardedPacketsAccumulated', self.transmitted_discarded_packets_accumulated)
- self.set_optional(obj, 'transmittedDiscardedPacketsDelta', self.transmitted_discarded_packets_delta)
- self.set_optional(obj, 'transmittedErrorPacketsAccumulated', self.transmitted_error_packets_accumulated)
- self.set_optional(obj, 'transmittedErrorPacketsDelta', self.transmitted_error_packets_delta)
- self.set_optional(obj, 'transmittedMulticastPacketsAccumulated', self.transmitted_multicast_packets_accumulated)
- self.set_optional(obj, 'transmittedMulticastPacketsDelta', self.transmitted_multicast_packets_delta)
- self.set_optional(obj, 'transmittedOctetsAccumulated', self.transmitted_octets_accumulated)
- self.set_optional(obj, 'transmittedOctetsDelta', self.transmitted_octets_delta)
- self.set_optional(obj, 'transmittedTotalPacketsAccumulated', self.transmitted_total_packets_accumulated)
- self.set_optional(obj, 'transmittedTotalPacketsDelta', self.transmitted_total_packets_delta)
- self.set_optional(obj, 'transmittedUnicastPacketsAccumulated', self.transmitted_unicast_packets_accumulated)
- self.set_optional(obj, 'transmittedUnicastPacketsDelta', self.transmitted_unicast_packets_delta)
- return obj
-
-
-class CpuUsage(VESDataType):
- """cpuUsage datatype"""
-
- def __init__(self, identifier):
- self.cpu_identifier = identifier
- self.cpu_idle = None
- self.cpu_usage_interrupt = None
- self.cpu_usage_nice = None
- self.cpu_usage_soft_irq = None
- self.cpu_usage_steal = None
- self.cpu_usage_system = None
- self.cpu_usage_user = None
- self.cpu_wait = None
- self.percent_usage = 0
-
- def get_obj(self):
- obj = {
- # required
- 'cpuIdentifier': self.cpu_identifier,
- 'percentUsage': self.percent_usage
- }
- # optional
- self.set_optional(obj, 'cpuIdle', self.cpu_idle)
- self.set_optional(obj, 'cpuUsageInterrupt', self.cpu_usage_interrupt)
- self.set_optional(obj, 'cpuUsageNice', self.cpu_usage_nice)
- self.set_optional(obj, 'cpuUsageSoftIrq', self.cpu_usage_soft_irq)
- self.set_optional(obj, 'cpuUsageSteal', self.cpu_usage_steal)
- self.set_optional(obj, 'cpuUsageSystem', self.cpu_usage_system)
- self.set_optional(obj, 'cpuUsageUser', self.cpu_usage_user)
- self.set_optional(obj, 'cpuWait', self.cpu_wait)
- return obj
-
-
-class MemoryUsage(VESDataType):
- """memoryUsage datatype"""
-
- def __init__(self, identifier):
- self.memory_buffered = None
- self.memory_cached = None
- self.memory_configured = None
- self.memory_free = None
- self.memory_slab_recl = None
- self.memory_slab_unrecl = None
- self.memory_used = None
- self.vm_identifier = identifier
-
- def __str__(self):
- """ for debug purposes """
- return 'vm_identifier : {vm_identifier}\nbuffered : {buffered}\n' \
- 'cached : {cached}\nconfigured : {configured}\nfree : {free}\n' \
- 'slab_recl : {slab_recl}\nslab_unrecl : {slab_unrecl}\n' \
- 'used : {used}\n'.format(buffered=self.memory_buffered,
- cached=self.memory_cached, configured=self.memory_configured,
- free=self.memory_free, slab_recl=self.memory_slab_recl,
- slab_unrecl=self.memory_slab_unrecl, used=self.memory_used,
- vm_identifier=self.vm_identifier)
-
- def get_memory_free(self):
- if self.memory_free is None:
- # calculate the free memory
- if None not in (self.memory_configured, self.memory_used):
- return self.memory_configured - self.memory_used
- else:
- # required field, so return zero
- return 0
- else:
- return self.memory_free
-
- def get_memory_used(self):
- if self.memory_used is None:
- # calculate the memory used
- if None not in (self.memory_configured, self.memory_free, self.memory_buffered,
- self.memory_cached, self.memory_slab_recl, self.memory_slab_unrecl):
- return self.memory_configured - (self.memory_free +
- self.memory_buffered + self.memory_cached +
- self.memory_slab_recl + self.memory_slab_unrecl)
- else:
- # required field, so return zero
- return 0
- else:
- return self.memory_used
-
- def get_memory_total(self):
- if self.memory_configured is None:
- # calculate the total memory
- if None not in (self.memory_used, self.memory_free, self.memory_buffered,
- self.memory_cached, self.memory_slab_recl, self.memory_slab_unrecl):
- return (self.memory_used + self.memory_free +
- self.memory_buffered + self.memory_cached +
- self.memory_slab_recl + self.memory_slab_unrecl)
- else:
- return None
- else:
- return self.memory_configured
-
- def get_obj(self):
- obj = {
- # required fields
- 'memoryFree': self.get_memory_free(),
- 'memoryUsed': self.get_memory_used(),
- 'vmIdentifier': self.vm_identifier
- }
- # optional fields
- self.set_optional(obj, 'memoryBuffered', self.memory_buffered)
- self.set_optional(obj, 'memoryCached', self.memory_cached)
- self.set_optional(obj, 'memoryConfigured', self.memory_configured)
- self.set_optional(obj, 'memorySlabRecl', self.memory_slab_recl)
- self.set_optional(obj, 'memorySlabUnrecl', self.memory_slab_unrecl)
- return obj
-
-class MeasurementsForVfScaling(Event):
- """MeasurementsForVfScaling datatype"""
-
- def __init__(self, event_id):
- """Construct the header"""
- super(MeasurementsForVfScaling, self).__init__()
- # common attributes
- self.domain = "measurementsForVfScaling"
- self.event_type = 'hostOS'
- self.event_id = event_id
- # measurement attributes
- self.additional_measurements = []
- self.codec_usage_array = []
- self.concurrent_sessions = 0
- self.configured_entities = 0
- self.cpu_usage_array = []
- self.feature_usage_array = []
- self.filesystem_usage_array = []
- self.latency_distribution = []
- self.mean_request_latency = 0
- self.measurement_interval = 0
- self.number_of_media_ports_in_use = 0
- self.request_rate = 0
- self.vnfc_scaling_metric = 0
- self.additional_fields = []
- self.additional_objects = []
- self.disk_usage_array = []
- self.measurements_for_vf_scaling_version = 2.0
- self.memory_usage_array = []
- self.v_nic_performance_array = []
-
- def add_additional_measurement(self, named_array):
- self.additional_measurements.append(named_array.get_obj())
-
- def add_additional_fields(self, field):
- self.additional_fields.append(field.get_obj())
-
- def add_memory_usage(self, mem_usage):
- self.memory_usage_array.append(mem_usage.get_obj())
-
- def add_cpu_usage(self, cpu_usage):
- self.cpu_usage_array.append(cpu_usage.get_obj())
-
- def add_v_nic_performance(self, nic_performance):
- self.v_nic_performance_array.append(nic_performance.get_obj())
-
- def add_disk_usage(self, disk_usage):
- self.disk_usage_array.append(disk_usage.get_obj())
-
- def get_obj(self):
- """Get the object of the datatype"""
- obj = {}
- obj['additionalMeasurements'] = self.additional_measurements
- obj['codecUsageArray'] = self.codec_usage_array
- obj['concurrentSessions'] = self.concurrent_sessions
- obj['configuredEntities'] = self.configured_entities
- obj['cpuUsageArray'] = self.cpu_usage_array
- obj['featureUsageArray'] = self.feature_usage_array
- obj['filesystemUsageArray'] = self.filesystem_usage_array
- obj['latencyDistribution'] = self.latency_distribution
- obj['meanRequestLatency'] = self.mean_request_latency
- obj['measurementInterval'] = self.measurement_interval
- obj['numberOfMediaPortsInUse'] = self.number_of_media_ports_in_use
- obj['requestRate'] = self.request_rate
- obj['vnfcScalingMetric'] = self.vnfc_scaling_metric
- obj['additionalFields'] = self.additional_fields
- obj['additionalObjects'] = self.additional_objects
- obj['diskUsageArray'] = self.disk_usage_array
- obj['measurementsForVfScalingVersion'] = self.measurements_for_vf_scaling_version
- obj['memoryUsageArray'] = self.memory_usage_array
- obj['vNicPerformanceArray'] = self.v_nic_performance_array
- return obj
-
- def get_name(self):
- """Name of datatype"""
- return "measurementsForVfScalingFields"
-
-
-class Fault(Event):
- """Fault datatype"""
-
- def __init__(self, event_id):
- """Construct the header"""
- super(Fault, self).__init__()
- # common attributes
- self.domain = "fault"
- self.event_id = event_id
- self.event_type = "Fault"
- # fault attributes
- self.fault_fields_version = 1.1
- self.event_severity = 'NORMAL'
- self.event_source_type = 'other(0)'
- self.alarm_condition = ''
- self.specific_problem = ''
- self.vf_status = 'Active'
- self.alarm_interface_a = ''
- self.alarm_additional_information = []
- self.event_category = ""
-
- def get_name(self):
- """Name of datatype"""
- return 'faultFields'
-
- def get_obj(self):
- """Get the object of the datatype"""
- obj = {}
- obj['faultFieldsVersion'] = self.fault_fields_version
- obj['eventSeverity'] = self.event_severity
- obj['eventSourceType'] = self.event_source_type
- obj['alarmCondition'] = self.alarm_condition
- obj['specificProblem'] = self.specific_problem
- obj['vfStatus'] = self.vf_status
- obj['alarmInterfaceA'] = self.alarm_interface_a
- obj['alarmAdditionalInformation'] = self.alarm_additional_information
- obj['eventCategory'] = self.event_category
- return obj
-
-
-class VESApp(object):
+class VESApp(Normalizer):
"""VES Application"""
def __init__(self):
"""Application initialization"""
- self.__plugin_data_cache = {
- 'cpu': {'interval': 0.0, 'vls': []},
- 'virt': {'interval': 0.0, 'vls': []},
- 'disk': {'interval': 0.0, 'vls': []},
- 'interface': {'interval': 0.0, 'vls': []},
- 'memory': {'interval': 0.0, 'vls': []}
- }
- self.__app_config = {
+ self._app_config = {
'Domain': '127.0.0.1',
'Port': 30000,
'Path': '',
@@ -546,270 +47,56 @@ class VESApp(object):
'Topic': '',
'UseHttps': False,
'SendEventInterval': 20.0,
- 'FunctionalRole': 'Collectd VES Agent',
'ApiVersion': 5.1,
'KafkaPort': 9092,
'KafkaBroker': 'localhost'
}
- self.__host_name = None
- self.__ves_timer = None
- self.__lock = Lock()
- self.__event_id = 0
-
- def get_event_id(self):
- """get event id"""
- self.__event_id += 1
- return str(self.__event_id)
-
- def lock(self):
- """Lock the application"""
- self.__lock.acquire()
-
- def unlock(self):
- """Unlock the application"""
- self.__lock.release()
-
- def start_timer(self):
- """Start event timer"""
- self.__ves_timer = Timer(self.__app_config['SendEventInterval'], self.__on_time)
- self.__ves_timer.start()
-
- def stop_timer(self):
- """Stop event timer"""
- self.__ves_timer.cancel()
-
- def __on_time(self):
- """Timer thread"""
- self.event_timer()
- self.start_timer()
- def event_send(self, event):
+ def send_data(self, event):
"""Send event to VES"""
server_url = "http{}://{}:{}{}/eventListener/v{}{}".format(
- 's' if self.__app_config['UseHttps'] else '', self.__app_config['Domain'],
- int(self.__app_config['Port']), '{}'.format(
- '/{}'.format(self.__app_config['Path']) if (len(self.__app_config['Path']) > 0) else ''),
- int(self.__app_config['ApiVersion']), '{}'.format(
- '/{}'.format(self.__app_config['Topic']) if (len(self.__app_config['Topic']) > 0) else ''))
+ 's' if self._app_config['UseHttps'] else '',
+ self._app_config['Domain'], int(self._app_config['Port']),
+ '{}'.format('/{}'.format(self._app_config['Path']) if len(
+ self._app_config['Path']) > 0 else ''),
+ int(self._app_config['ApiVersion']), '{}'.format(
+ '/{}'.format(self._app_config['Topic']) if len(
+ self._app_config['Topic']) > 0 else ''))
logging.info('Vendor Event Listener is at: {}'.format(server_url))
credentials = base64.b64encode('{}:{}'.format(
- self.__app_config['Username'], self.__app_config['Password']).encode()).decode()
+ self._app_config['Username'],
+ self._app_config['Password']).encode()).decode()
logging.info('Authentication credentials are: {}'.format(credentials))
try:
request = url.Request(server_url)
request.add_header('Authorization', 'Basic {}'.format(credentials))
request.add_header('Content-Type', 'application/json')
- logging.debug("Sending {} to {}".format(event.get_json(), server_url))
- vel = url.urlopen(request, event.get_json(), timeout=1)
+ event_str = json.dumps(event).encode()
+ logging.debug("Sending {} to {}".format(event_str, server_url))
+ url.urlopen(request, event_str, timeout=1)
logging.debug("Sent data to {} successfully".format(server_url))
except url.HTTPError as e:
logging.error('Vendor Event Listener exception: {}'.format(e))
except url.URLError as e:
- logging.error('Vendor Event Listener is is not reachable: {}'.format(e))
+ logging.error(
+ 'Vendor Event Listener is is not reachable: {}'.format(e))
except Exception as e:
logging.error('Vendor Event Listener error: {}'.format(e))
- def bytes_to_kb(self, bytes):
- """Convert bytes to kibibytes"""
- return round((bytes / 1024.0), 3)
-
- def get_hostname(self):
- return socket.gethostname()
-
- def send_host_measurements(self):
- # get list of all VMs
- virt_vcpu_total = self.cache_get_value(plugin_name='virt', type_name='virt_cpu_total',
- mark_as_read=False)
- vm_names = [x['plugin_instance'] for x in virt_vcpu_total]
- for vm_name in vm_names:
- # make sure that 'virt' plugin cache is up-to-date
- vm_values = self.cache_get_value(plugin_name='virt', plugin_instance=vm_name,
- mark_as_read=False)
- us_up_to_date = True
- for vm_value in vm_values:
- if vm_value['updated'] == False:
- logging.warning("%s" % vm_value)
- us_up_to_date = False
- break
- if not us_up_to_date:
- # one of the cache value is not up-to-date, break
- logging.warning("virt collectd cache values are not up-to-date for {}".format(vm_name))
- continue
- # values are up-to-date, create an event message
- measurement = MeasurementsForVfScaling(self.get_event_id())
- measurement.functional_role = self.__app_config['FunctionalRole']
- # fill out reporting_entity
- measurement.reporting_entity_id = self.get_hostname()
- measurement.reporting_entity_name = measurement.reporting_entity_id
- # set source as a host value
- measurement.source_id = vm_name
- measurement.source_name = measurement.source_id
- # fill out EpochMicrosec (convert to us)
- measurement.start_epoch_microsec = (virt_vcpu_total[0]['time'] * 1000000)
- # plugin interval
- measurement.measurement_interval = self.__plugin_data_cache['virt']['interval']
- # memoryUsage
- mem_usage = MemoryUsage(vm_name)
- memory_total = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='memory', type_instance='total')
- memory_unused = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='memory', type_instance='unused')
- memory_rss = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='memory', type_instance='rss')
- if len(memory_total) > 0:
- mem_usage.memory_configured = self.bytes_to_kb(memory_total[0]['values'][0])
- if len(memory_unused) > 0:
- mem_usage.memory_free = self.bytes_to_kb(memory_unused[0]['values'][0])
- elif len(memory_rss) > 0:
- mem_usage.memory_free = self.bytes_to_kb(memory_rss[0]['values'][0])
- # since, "used" metric is not provided by virt plugn, set the rest of the memory stats
- # to zero to calculate used based on provided stats only
- mem_usage.memory_buffered = mem_usage.memory_cached = mem_usage.memory_slab_recl = \
- mem_usage.memory_slab_unrecl = 0
- measurement.add_memory_usage(mem_usage)
- # cpuUsage
- virt_vcpus = self.cache_get_value(plugin_instance=vm_name,
- plugin_name='virt', type_name='virt_vcpu')
- for virt_vcpu in virt_vcpus:
- cpu_usage = CpuUsage(virt_vcpu['type_instance'])
- cpu_usage.percent_usage = self.cpu_ns_to_percentage(virt_vcpu)
- measurement.add_cpu_usage(cpu_usage)
- # vNicPerformance
- if_packets = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='if_packets', mark_as_read=False)
- if_names = [x['type_instance'] for x in if_packets]
- for if_name in if_names:
- if_packets = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='if_packets', type_instance=if_name)
- if_octets = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='if_octets', type_instance=if_name)
- if_errors = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='if_errors', type_instance=if_name)
- if_dropped = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='if_dropped', type_instance=if_name)
- v_nic_performance = VNicPerformance(if_name)
- v_nic_performance.received_total_packets_accumulated = if_packets[0]['values'][0]
- v_nic_performance.transmitted_total_packets_accumulated = if_packets[0]['values'][1]
- v_nic_performance.received_octets_accumulated = if_octets[0]['values'][0]
- v_nic_performance.transmitted_octets_accumulated = if_octets[0]['values'][1]
- v_nic_performance.received_error_packets_accumulated = if_errors[0]['values'][0]
- v_nic_performance.transmitted_error_packets_accumulated = if_errors[0]['values'][1]
- v_nic_performance.received_discarded_packets_accumulated = if_dropped[0]['values'][0]
- v_nic_performance.transmitted_discarded_packets_accumulated = if_dropped[0]['values'][1]
- measurement.add_v_nic_performance(v_nic_performance)
- # diskUsage
- disk_octets = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='disk_octets', mark_as_read=False)
- disk_names = [x['type_instance'] for x in disk_octets]
- for disk_name in disk_names:
- disk_octets = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='disk_octets', type_instance=disk_name)
- disk_ops = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='disk_ops', type_instance=disk_name)
- disk_usage = DiskUsage(disk_name)
- disk_usage.disk_octets_read_last = disk_octets[0]['values'][0]
- disk_usage.disk_octets_write_last = disk_octets[0]['values'][1]
- disk_usage.disk_ops_read_last = disk_ops[0]['values'][0]
- disk_usage.disk_ops_write_last = disk_ops[0]['values'][1]
- measurement.add_disk_usage(disk_usage)
- # add additional measurements (perf)
- perf_values = self.cache_get_value(plugin_instance=vm_name, plugin_name='virt',
- type_name='perf')
- named_array = NamedArrayOfFields('perf')
- for perf in perf_values:
- named_array.add(Field(perf['type_instance'], str(perf['values'][0])))
- measurement.add_additional_measurement(named_array)
- # add host values as additional measurements
- self.set_additional_fields(measurement, exclude_plugins=['virt'])
- # send event to the VES
- self.event_send(measurement)
- if len(vm_names) > 0:
- # mark the additional measurements metrics as read
- self.mark_cache_values_as_read(exclude_plugins=['virt'])
-
- def event_timer(self):
- """Event timer thread"""
- self.lock()
- try:
- self.send_host_measurements()
- finally:
- self.unlock()
-
- def mark_cache_values_as_read(self, exclude_plugins=None):
- """mark the cache values as read"""
- for plugin_name in self.__plugin_data_cache.keys():
- if (exclude_plugins != None and plugin_name in exclude_plugins):
- # skip excluded plugins
- continue;
- for val in self.__plugin_data_cache[plugin_name]['vls']:
- val['updated'] = False
-
- def set_additional_measurements(self, measurement, exclude_plugins=None):
- """Set addition measurement filed with host/guets values"""
- # add host/guest values as additional measurements
- for plugin_name in self.__plugin_data_cache.keys():
- if (exclude_plugins != None and plugin_name in exclude_plugins):
- # skip excluded plugins
- continue;
- for val in self.__plugin_data_cache[plugin_name]['vls']:
- if val['updated']:
- array_name = self.make_dash_string(plugin_name, val['plugin_instance'],
- val['type_instance'])
- named_array = NamedArrayOfFields(array_name)
-
- for index in range(len(val['dsnames'])):
- mname = '{}-{}'.format(val['type'], val['dsnames'][index])
- named_array.add(Field(mname, str(val['values'][index])))
- measurement.add_additional_measurement(named_array);
- val['updated'] = False
-
- def set_additional_fields(self, measurement, exclude_plugins=None):
- # set host values as additional fields
- for plugin_name in self.__plugin_data_cache.keys():
- if (exclude_plugins != None and plugin_name in exclude_plugins):
- # skip excluded plugins
- continue;
- for val in self.__plugin_data_cache[plugin_name]['vls']:
- if val['updated']:
- name_prefix = self.make_dash_string(plugin_name, val['plugin_instance'],
- val['type_instance'])
-
- for index in range(len(val['dsnames'])):
- field_name = self.make_dash_string(name_prefix, val['type'], val['dsnames'][index])
- measurement.add_additional_fields(Field(field_name, str(val['values'][index])))
-
- def cpu_ns_to_percentage(self, vl):
- """Convert CPU usage ns to CPU %"""
- total = vl['values'][0]
- total_time = vl['time']
- pre_total = vl['pre_values'][0]
- pre_total_time = vl['pre_time']
- if (total_time - pre_total_time) == 0:
- # return zero usage if time diff is zero
- return 0.0
- percent = (100.0 * (total - pre_total)) / ((total_time - pre_total_time) * 1000000000.0)
- logging.debug("pre_time={}, pre_value={}, time={}, value={}, cpu={}%".format(
- pre_total_time, pre_total, total_time, total, round(percent, 2)))
- return round(percent, 2)
-
- def make_dash_string(self, *args):
- """Join non empty strings with dash symbol"""
- return '-'.join(filter(lambda x: len(x) > 0, args))
-
def config(self, config):
"""VES option configuration"""
for key, value in config.items('config'):
- if key in self.__app_config:
+ if key in self._app_config:
try:
- if type(self.__app_config[key]) == int:
+ if type(self._app_config[key]) == int:
value = int(value)
- elif type(self.__app_config[key]) == float:
+ elif type(self._app_config[key]) == float:
value = float(value)
- elif type(self.__app_config[key]) == bool:
- value = bool(distutils.util.strtobool(value))
+ elif type(self._app_config[key]) == bool:
+ value = bool(strtobool(value))
- if isinstance(value, type(self.__app_config[key])):
- self.__app_config[key] = value
+ if isinstance(value, type(self._app_config[key])):
+ self._app_config[key] = value
else:
logging.error("Type mismatch with %s" % key)
sys.exit()
@@ -820,125 +107,107 @@ class VESApp(object):
logging.error("Incorrect key configuration %s" % key)
sys.exit()
- def init(self):
- """Initialisation"""
- # start the VES timer
- self.start_timer()
-
- def update_cache_value(self, kafka_data):
- """Update value internal collectd cache values or create new one
- Please note, the cache should be locked before using this function"""
- found = False
- if kafka_data['plugin'] not in self.__plugin_data_cache:
- self.__plugin_data_cache[kafka_data['plugin']] = {'vls': []}
- plugin_vl = self.__plugin_data_cache[kafka_data['plugin']]['vls']
-
- for index in range(len(plugin_vl)):
- # record found, so just update time the values
- if (plugin_vl[index]['plugin_instance'] == kafka_data['plugin_instance']) and \
- (plugin_vl[index]['type_instance'] == kafka_data['type_instance']) and \
- (plugin_vl[index]['type'] == kafka_data['type']):
- plugin_vl[index]['pre_time'] = plugin_vl[index]['time']
- plugin_vl[index]['time'] = kafka_data['time']
- plugin_vl[index]['pre_values'] = plugin_vl[index]['values']
- plugin_vl[index]['values'] = kafka_data['values']
- plugin_vl[index]['dsnames'] = kafka_data['dsnames']
- plugin_vl[index]['updated'] = True
- found = True
- break
- if not found:
- value = {}
- # create new cache record
- value['plugin_instance'] = kafka_data['plugin_instance']
- value['type_instance'] = kafka_data['type_instance']
- value['values'] = kafka_data['values']
- value['pre_values'] = kafka_data['values']
- value['type'] = kafka_data['type']
- value['time'] = kafka_data['time']
- value['pre_time'] = kafka_data['time']
- value['host'] = kafka_data['host']
- value['dsnames'] = kafka_data['dsnames']
- value['updated'] = True
- self.__plugin_data_cache[kafka_data['plugin']]['vls'].append(value)
- # update plugin interval based on one received in the value
- self.__plugin_data_cache[kafka_data['plugin']]['interval'] = kafka_data['interval']
-
- def cache_get_value(self, plugin_name=None, plugin_instance=None,
- type_name=None, type_instance=None, type_names=None, mark_as_read=True):
- """Get cache value by given criteria"""
- ret_list = []
- if plugin_name in self.__plugin_data_cache:
- for val in self.__plugin_data_cache[plugin_name]['vls']:
- if (type_name == None or type_name == val['type']) and \
- (plugin_instance == None or plugin_instance == val['plugin_instance']) and \
- (type_instance == None or type_instance == val['type_instance']) and \
- (type_names == None or val['type'] in type_names):
- if mark_as_read:
- val['updated'] = False
- ret_list.append(val)
- return ret_list
-
- def shutdown(self):
- """Shutdown method for clean up"""
- self.stop_timer()
+ def init(self, configfile, schema_file):
+ if configfile is not None:
+ # read VES configuration file if provided
+ config = ConfigParser.ConfigParser()
+ config.optionxform = lambda option: option
+ config.read(configfile)
+ self.config(config)
+ # initialize normalizer
+ self.initialize(schema_file, self._app_config['SendEventInterval'])
def run(self):
"""Consumer JSON data from kafka broker"""
- kafka_server = '%s:%s' % (self.__app_config.get('KafkaBroker'), self.__app_config.get('KafkaPort'))
- consumer = KafkaConsumer('collectd', bootstrap_servers=kafka_server, auto_offset_reset='latest',
- enable_auto_commit=False,
- value_deserializer=lambda m: json.loads(m.decode('ascii')))
+ kafka_server = '{}:{}'.format(
+ self._app_config.get('KafkaBroker'),
+ self._app_config.get('KafkaPort'))
+ consumer = KafkaConsumer(
+ 'collectd', bootstrap_servers=kafka_server,
+ auto_offset_reset='latest', enable_auto_commit=False,
+ value_deserializer=lambda m: json.loads(m.decode('ascii')))
for message in consumer:
for kafka_data in message.value:
- self.lock()
- try:
- # Receive Kafka data and update cache values
- self.update_cache_value(kafka_data)
-
- finally:
- self.unlock()
+ # {
+ # u'dstypes': [u'derive'],
+ # u'plugin': u'cpu',
+ # u'dsnames': [u'value'],
+ # u'interval': 10.0,
+ # u'host': u'localhost',
+ # u'values': [99.9978996416267],
+ # u'time': 1502114956.244,
+ # u'plugin_instance': u'44',
+ # u'type_instance': u'idle',
+ # u'type': u'cpu'
+ # }
+ logging.debug('{}:run():data={}'.format(
+ self.__class__.__name__, kafka_data))
+ for ds_name in kafka_data['dsnames']:
+ index = kafka_data['dsnames'].index(ds_name)
+ val_hash = CollectdValue.hash_gen(
+ kafka_data['host'], kafka_data['plugin'],
+ kafka_data['plugin_instance'], kafka_data['type'],
+ kafka_data['type_instance'], ds_name)
+ collector = self.get_collector()
+ val = collector.get(val_hash)
+ if val:
+ # update the value
+ val.value = kafka_data['values'][index]
+ val.time = kafka_data['time']
+ del(val)
+ else:
+ # add new value into the collector
+ val = CollectdValue()
+ val.host = kafka_data['host']
+ val.plugin = kafka_data['plugin']
+ val.plugin_instance = kafka_data['plugin_instance']
+ val.type = kafka_data['type']
+ val.type_instance = kafka_data['type_instance']
+ val.value = kafka_data['values'][index]
+ val.interval = kafka_data['interval']
+ val.time = kafka_data['time']
+ val.ds_name = ds_name
+ collector.add(val)
def main():
# Parsing cmdline options
parser = argparse.ArgumentParser()
- parser.add_argument("--config", dest="configfile", default=None, help="Specify config file", metavar="FILE")
- parser.add_argument("--loglevel", dest="level", choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'], default='WARNING',
- help="Specify log level (default: %(default)s)", metavar="LEVEL")
+ parser.add_argument("--events-schema", dest="schema", required=True,
+ help="YAML events schema definition", metavar="FILE")
+ parser.add_argument("--config", dest="configfile", default=None,
+ help="Specify config file", metavar="FILE")
+ parser.add_argument("--loglevel", dest="level", default='INFO',
+ choices=['DEBUG', 'INFO', 'WARNING', 'ERROR'],
+ help="Specify log level (default: %(default)s)",
+ metavar="LEVEL")
parser.add_argument("--logfile", dest="logfile", default='ves_app.log',
- help="Specify log file (default: %(default)s)", metavar="FILE")
+ help="Specify log file (default: %(default)s)",
+ metavar="FILE")
args = parser.parse_args()
# Create log file
- logging.basicConfig(filename=args.logfile, format='%(asctime)s %(message)s', level=args.level)
+ logging.basicConfig(filename=args.logfile,
+ format='%(asctime)s %(message)s',
+ level=args.level)
+ if args.configfile is None:
+ logging.warning("No configfile specified, using default options")
# Create Application Instance
application_instance = VESApp()
+ application_instance.init(args.configfile, args.schema)
- # Read from config file
- if args.configfile:
- config = ConfigParser.ConfigParser()
- config.optionxform = lambda option: option
- config.read(args.configfile)
- # Write Config Values
- application_instance.config(config)
- else:
- logging.warning("No configfile specified, using default options")
-
- # Start timer for interval
- application_instance.init()
-
- # Read Data from Kakfa
try:
- # Kafka consumer & update cache
+ # Run the plugin
application_instance.run()
except KeyboardInterrupt:
logging.info(" - Ctrl-C handled, exiting gracefully")
- application_instance.shutdown()
+ except Exception as e:
+ logging.error('{}, {}'.format(type(e), e))
+ finally:
+ application_instance.destroy()
sys.exit()
- except:
- logging.error("Unknown Error")
if __name__ == '__main__':