"""
Data reporting API for plugins.
"""
import copy
from ruxit.api.data import BaseMetric, PluginMeasurementList, PluginMeasurement, PluginProperty
from ruxit.api.selectors import ExplicitPgiSelector, EntityType
from ruxit.utils.firewall import Firewall
import time
[docs]class ResultsBuilder:
"""
Class storing data gathered by plugins. An instance of this class is associated with each plugin instance.
The builder can store intermittent results and perform calculations on them when they need to be sent away. Those
include computing the difference between subsequent measurement, providing per second rates etc.
Measurements can be passed in the form of manually created :class:`~ruxit.api.data.PluginMeasurement` objects (via
add_*_result methods), or via absolute/relative/per_second methods which construct the measurement instances
themselves.
"""
_CUSTOM_PLUGIN_MEASUREMENTS_LIMIT = 1000
def __init__(self, logger):
self.old_relative_results = {}
self.old_per_second_results = {}
self.result = None
self.relative_results = {}
self.per_second_results = {}
self.absolute_results = []
self.events = []
self.plugin_properties = []
self._timestamp = None
self.logger = logger
@staticmethod
def _pm_to_key(pm):
return pm.key, tuple(pm.dimensions.items()), pm.entity_selector
@staticmethod
def _check_measurement_type(plugin_measurement):
if not isinstance(plugin_measurement, BaseMetric):
raise TypeError("Each measurement should be an instance of BaseMetric, got %s instead" % type(plugin_measurement))
@property
def CUSTOM_PLUGIN_MEASUREMENTS_LIMIT(self):
return self._CUSTOM_PLUGIN_MEASUREMENTS_LIMIT
def _createpm(self, **kwargs):
if 'entity_id' in kwargs:
if 'entity_selector' in kwargs:
raise ValueError("entity_id and entity_selector are mutually exclusive options")
entity_id = kwargs.pop('entity_id')
if 'entity_type' in kwargs:
if not isinstance(kwargs.get('entity_type'), EntityType):
raise ValueError("entity_type must be instance of EntityType")
entity_type = kwargs.pop('entity_type')
kwargs['entity_selector'] = ExplicitPgiSelector(pgi_id=entity_id, me_type=entity_type)
else:
kwargs['entity_selector'] = ExplicitPgiSelector(pgi_id=entity_id)
return PluginMeasurement(**kwargs)
[docs] def add_relative_result(self, plugin_measurement):
"""
Adds a relative result. Use for metrics that increase monotonically (counters etc.) After gathering 2nd result,
difference between current and previous one will be sent to the server, if it is non-negative. If there is a
negative difference between measurements - nothing is sent.
Args:
plugin_measurement(PluginMeasurement): measurement
"""
self._check_measurement_type(plugin_measurement)
self.relative_results[self._pm_to_key(plugin_measurement)] = plugin_measurement
[docs] def add_absolute_result(self, plugin_measurement):
"""
Adds an absolute results. Absolute results are sent unmodified - no additional calculations are performed.
Args:
plugin_measurement(PluginMeasurement): measurement
"""
self._check_measurement_type(plugin_measurement)
self.absolute_results.append(plugin_measurement)
[docs] def add_per_second_result(self, plugin_measurement):
"""
Adds per second result. Use for metrics that increase monotonically (counters etc.) After gathering 2nd result,
difference between current and previous one will be sent to the server, if it is non-negative. It will be
divided by the time difference between measurements in seconds - resulting in a per second rate. If there is a
negative difference between measurements - nothing is sent.
Args:
plugin_measurement(PluginMeasurement): measurement
"""
self._check_measurement_type(plugin_measurement)
self.per_second_results[self._pm_to_key(plugin_measurement)] = (plugin_measurement, time.monotonic())
[docs] def absolute(self, **kwargs):
"""
Adds absolute result. PluginMeasurement object is constructed implicitly in the method. Keyword arguments
correspond to :class:`~ruxit.api.data.PluginMeasurement` class keyword with the addition of:
Args:
entity_id: value of entity id.
entity_type (EntityType): type of entity. Default: :class:`ruxit.api.selectors.EntityType.PROCESS_GROUP_INSTANCE`
"""
_pm = self._createpm(**kwargs)
self.add_absolute_result(_pm)
[docs] def add_event(self, event):
"""
Adds a event
Args:
event(Event): event
"""
if Firewall.is_allowed_call():
self.events.append(event)
[docs] def relative(self, **kwargs):
"""
Adds relative result, with PluginMeasurement object constructed implicitly. Accepts arguments identical to
:meth:`~ruxit.api.results_builder.ResultsBuilder.absolute`
"""
_pm = self._createpm(**kwargs)
self.add_relative_result(_pm)
[docs] def per_second(self, **kwargs):
"""
Adds per second result, with PluginMeasurement object constructed implicitly. Accepts arguments identical to
:meth:`~ruxit.api.results_builder.ResultsBuilder.absolute`
"""
_pm = self._createpm(**kwargs)
self.add_per_second_result(_pm)
[docs] def add_property(self, plugin_property):
"""
Adds a property to the results
Args:
plugin_property (PluginProperty): the property.
"""
if not isinstance(plugin_property, PluginProperty):
raise TypeError("Each property should be an instance of PluginProperty, got %s instead" % type(plugin_property))
self.plugin_properties.append(plugin_property)
[docs] def set_timestamp(self, timestamp):
"""
Allows explicit setting of timestamp for all gathered measurements.
Args:
timestamp: timestamp
"""
self._timestamp = timestamp
[docs] def flush_result(self):
"""
Prepares gathered data to be sent out to the server. Computes values for relative and per_second results.
Returns:
measurements and properties gathered during the most recent data collection period.
"""
measurements = self.result
if not measurements:
measurements = PluginMeasurementList(measurements=[], timestamp=self._timestamp)
for relative_result_key, relative_metric in self.relative_results.items():
old_relative_metric = self.old_relative_results.get(relative_result_key, None)
if old_relative_metric is not None:
delta_metric = copy.copy (relative_metric)
value_delta = relative_metric.value - old_relative_metric.value
if value_delta < 0:
self.logger.info(
"Value decrease in relative result, skip reporting, new value %s, old value %s",
relative_metric.value,
old_relative_metric.value
)
continue
delta_metric.value = value_delta
measurements.measurements.append(delta_metric)
for per_second_result_key, per_second_result in self.per_second_results.items():
old_per_second_result = self.old_per_second_results.get(per_second_result_key, None)
if old_per_second_result is not None:
value_per_second = (per_second_result[0].value-old_per_second_result[0].value)/(per_second_result[1]-old_per_second_result[1])
if value_per_second < 0:
self.logger.info(
"Per second result less than 0, skip reporting, new value %s, old value %s",
per_second_result,
old_per_second_result
)
continue
per_second = copy.copy (per_second_result[0])
per_second.value = value_per_second
measurements.measurements.append(per_second)
measurements.measurements.extend(self.absolute_results)
self.old_relative_results = self.relative_results
self.old_per_second_results = self.per_second_results
self.relative_results = {}
self.per_second_results = {}
self.absolute_results = []
self.result = None
self._timestamp = None
properties = self.plugin_properties
self.plugin_properties = []
events = self.events
self.events = []
return measurements, properties, events
