"""
Structures for holding data gathered by plugins. They enforce types so that a plugin cannot accidentally report
a number as a string.
"""
from enum import Enum
import typing
import ruxit.api.selectors
from ruxit.utils.firewall import Firewall
from ruxit.utils.firewall import BlockedException
[docs]class BaseMetric:
def __init__(self, key, value, dimensions: typing.Dict[str, str]=None, entity_selector=ruxit.api.selectors.FromPluginSelector()):
self.key = key
self.value = value
self.dimensions = dimensions
self.entity_selector = entity_selector
@property
def key(self):
"""
Key of the measurement. Must be a string.
"""
return self._key
@property
def value(self):
"""
Value of the measurement. Must be convertible to float.
"""
return self._value
@property
def dimensions(self):
"""
Dimensions of the measurement. A Dictionary with string keys and string values.
"""
return self._dimensions
@dimensions.setter
def dimensions(self, dimensions: typing.Dict[str, str]):
self._dimensions = {}
if dimensions:
for key, value in dimensions.items():
if isinstance(key, str) is False or isinstance(value, str) is False:
raise TypeError("Each dimensions key, and value must be unicode object, got %s and %s instead" % (type(key), type(value)))
self._dimensions = dimensions
def __repr__(self):
return "%s(key=%s,value=%s,dimensions=%s,selector=%s)" % (type(self).__name__, self.key, self.value, self.dimensions, self.entity_selector)
def __eq__(self, other):
if isinstance(other, type(self)):
return self.__dict__ == other.__dict__
return NotImplemented
[docs]class PluginMeasurement(BaseMetric):
"""
Single plugin measurement.
Use of this class is optional. :class:`~ruxit.api.results_builder.ResultsBuilder` exposes methods that handles this
type directly and those that abstract them away.
"""
def __init__(self, key: str, value: float, dimensions: typing.Dict[str, str]=None, entity_selector=ruxit.api.selectors.FromPluginSelector()):
BaseMetric.__init__(self, key, value, dimensions, entity_selector)
@BaseMetric.key.setter
def key(self, key: str):
if not isinstance(key, str):
raise TypeError("key must be an unicode object, got, %s instead" % type(key))
self._key = key
@BaseMetric.value.setter
def value(self, value: float):
if not isinstance(value, float):
try:
self._value = float(value)
except TypeError as ex:
raise TypeError("value must be convertable to float, got %s instead" % (type(value)))
else:
self._value = value
[docs]class PluginStateMetric(BaseMetric):
"""
State measurement.
States are string described in plugin json as statetimeseries
"""
def __init__(self, key: str, value: str, dimensions: typing.Dict[str, str]=None, entity_selector=ruxit.api.selectors.FromPluginSelector()):
BaseMetric.__init__(self, key, value, dimensions, entity_selector)
@BaseMetric.key.setter
def key(self, key: str):
if not isinstance(key, str):
raise TypeError("key must be an unicode object, got, %s instead" % type(key))
self._key = key
@BaseMetric.value.setter
def value(self, value):
if isinstance(value, Enum):
self._value = value.value
elif not isinstance(value, str):
try:
self._value = float(value)
except TypeError as ex:
raise TypeError("value must be convertable to float, got %s instead" % (type(value)))
else:
self._value = value
[docs]class PluginMeasurementV3(BaseMetric):
"""
Not allowed for Custom Plugin
"""
def __init__(self, key: int, value: float, dimensions: typing.Dict[str, str]=None, entity_selector=ruxit.api.selectors.FromPluginSelector()):
if Firewall.is_allowed_call():
BaseMetric.__init__(self, key, value, dimensions, entity_selector)
else:
raise BlockedException("PluginMeasurementV3 is not allowed for Custom Plugin")
@BaseMetric.key.setter
def key(self, key: int):
if not isinstance(key, int):
raise TypeError("key must be an integer, got, %s instead" % type(key))
self._key = key
@BaseMetric.value.setter
def value(self, value: float):
if not isinstance(value, float):
try:
self._value = float(value)
except TypeError as ex:
raise TypeError("value must be convertable to float, got %s instead" % (type(value)))
else:
self._value = value
[docs]class MEAttribute(Enum):
"""
Enumerates types of properties to report.
"""
ID_CALCULATION_INPUT = 1
INTERNAL_DEBUG_VALUE = 2
INTERNAL_DEBUG_PROPERTY = 3
PROCESS_GROUP_INSTANCE_MODULES = 4
DOCKER_CONTAINER_PROPS = 5
PROCESS_GROUP_TECHS = 6
AUTO_TAGS = 7
AGENT_SEC_SETTINGS = 8
LOG_PATH_SIZES = 9
AGENT_INSTRUMENTATION_INFO = 10
FILE_STATUS = 11
AUTO_TAGS_V2 = 12
LOGS_SAVED_IN_STORAGE = 13
CONTRIBUTED_NAME = 14
[docs]class PluginProperty:
"""
A text property a plugin can gather.
"""
def __init__(self, key=None, value=None, entity_selector=ruxit.api.selectors.FromPluginSelector(),
me_attribute=MEAttribute.INTERNAL_DEBUG_PROPERTY):
"""
Args:
key (string): property key
value (string): property value
entity_selector: plugin's entity selector :class:`~ruxit.api.selectors.FromPluginSelector` by default
me_attribute(MEAttribute): property type
"""
self.key = key
self.value = value
self.entity_selector = entity_selector
if isinstance(entity_selector, ruxit.api.selectors.DockerSelector):
self.me_attribute = MEAttribute.DOCKER_CONTAINER_PROPS
else:
self.me_attribute = me_attribute
@property
def key(self):
"""
Key of property.
"""
return self._key
@key.setter
def key(self, key):
if key is not None and not isinstance(key, str):
raise TypeError("key must be an unicode object or None type, got, %s instead" % type(key))
self._key = key
@property
def value(self):
"""
Value of property
"""
return self._value
@value.setter
def value(self, value):
if not isinstance(value, str):
raise TypeError("value must be an unicode object or None type, got, %s instead" % type(value))
self._value = value
@property
def me_attribute(self):
"""
Property type
"""
return self._me_attribute
@me_attribute.setter
def me_attribute(self, me_attribute):
if not isinstance(me_attribute, MEAttribute):
raise TypeError("value must be an MEAttribute object, got, %s instead" % type(me_attribute))
self._me_attribute = me_attribute
def __repr__(self):
return "PluginProperty(key=%s,value=%s,selector=%s,me_attribute=%s)" % (
self.key, self.value, self.entity_selector, self.me_attribute)
[docs]class PluginMeasurementList:
"""
Collection of PluginMeasurements. Used internally.
"""
def __init__(self, timestamp=None, measurements=None):
self.timestamp = timestamp
self.measurements = measurements
@property
def timestamp(self):
return self._timestamp
@timestamp.setter
def timestamp(self, timestamp):
self._timestamp = None
if timestamp:
self._timestamp = int(timestamp)
@property
def measurements(self):
return self._measurements
@measurements.setter
def measurements(self, measurements):
self._measurements = []
if measurements:
for m in measurements:
if not isinstance(m, BaseMetric):
raise TypeError("Each measurement should be an instance of BaseMetric, got %s instead" % type(m))
self._measurements = measurements
@property
def is_empty(self):
return len(self._measurements) == 0
@property
def length(self):
return len(self._measurements)
