plumd.ResultSet.add_list() - Code Metrics - Inspection of "fixup influxdb writer (readability), misc bug fixe..." - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 8d751b...7e5eed )

by Kenny

created 2016-02-27 15:41 UTC

plumd.ResultSet.add_list() A

↳ Parent: plumd.ResultSet

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	2
dl	0
loc	8
rs	9.4285

# -*- coding: utf-8 -*-
"""There are many differences in the various time series databases. Plumd is
designed to support multiple backends using the classes defined here to
represent metrics in a uniform way:

    ResultSet - a collection of one or more Results from a call to poll()
    Result - a measurement of a single thing eg. disk free=0.0,used=100.0
    Meta - metadata associated with a Result eg. dev=sda1
    Int/Float/String/Boolean/Gauge/Rate/Counter/Timer - primitives for above

The ResultSet class is used to encapsulate all of the measurements taken
during a plugins poll() call and uses a single timestamp for that poll.

The Result class is used to encapsulate a single measurement - for example
measuring cpu utilization may result in idle, user, system etc values. These
would be recorded in a single Result object. In addition the Result class
allows for arbitrary metadata to be associated with a Result - eg. when
recording disk space there may be used, free, total however also metadata such
as dev=sda1 to associate the measurements with a specific block device.

The Meta class is used to encapsulate arbitrary metadata. The Result class
uses it to associate metadata with a measurement however it can also be used
for example to describe host level metadata (eg. hostname=<hostname>). The
Meta class allows writer plugins such as graphite to render the metadata as
part of the metric path (eg. servers.host.<hostname>.disk.used.dev.<device>) -
if there was no distiction between metadata key=value and metric key=value this
would not be possible. Also, the Meta class ensures that metadata keys are
unique.

The _Metrics classes are used to allow plumd to render metrics in a specific
format. For example influxdb wants integer values to end with an i when sending
via their line protocol. The various _Metric classes force the plugin to
explicitly say what each metric is (eg. Int, Float, String, Boolean, Counter,
Gauge, Timer, etc). Metrics are defined as namedtuples for reduced memory
utilization.

The Render class is an abstract base class that writer plugins can subclass. The
render() function needs to be defined so that it returns a formatted metric. A
writer plugin would then instantiate the subclassed Render and use it to record
formatted metrics an each push() call. It can then step through the passed
metrics in chunks and send them to its backend. Having the metric types defined
explicitly allows the writers renderer to format metrics for arbitrary backends.


.. moduleauthor:: Kenny Freeman <[email protected]>

"""

__author__ = 'Kenny Freeman'
__email__ = '[email protected]'
__version__ = '0.3.3'
__license__     = "ISCL"
__docformat__   = 'reStructuredText'

import abc
import logging
import collections

import arrow # pip install arrow


DEFAULT_CONFIG = {
    'log.level':        "warn",               # crit, error, warn, info, debug
    'config.file':      "/etc/plumd.yaml",    # path to main configuration file
    'config.plugins':   "/etc/plumd.conf.d/", # path to the plugin confs
    'delay.startup':    1,                    # random delay for plugin start
    'delay.poll':       1,                    # random delay for first poll()
    'poll.interval':    15,                   # interval of reader poll() calls
    'max.queue':        512,                  # maximum size of internal queues
    'meta':             {},                   # key=value's for all metrics
    'meta.hostname':    True                  # set and use hostname in meta
}


LOG_LEVELS = {                                # map config level to python level
    'crit':  logging.CRITICAL,
    'error': logging.ERROR,
    'warn':  logging.WARN,
    'info':  logging.INFO,
    'debug': logging.DEBUG
}


class DuplicatePlugin(Exception):
    """Two or more plugins with the same id were defined."""
    pass


class PluginNotFoundError(Exception):
    """A plugin file, module or class was not found."""
    pass


class PluginLoadError(Exception):
    """Exception(s) were raised while loading a plugin."""
    pass


class ConfigError(Exception):
    """Invalid configuration encountered."""
    pass


#_Metric : the base class of all metric primitives is a named tuple.
_Metric = collections.namedtuple('metric', ['name', 'value'])


class Int(_Metric):
    """An integer metric - ensures the value passed is an int.

    raises:
        ValueError if the passed value doesn't cast properly to int

    :param name: The name of the metric
    :type name: str
    :param value: The recorded value of the metric
    :type value: int
    :raises: ValueError
    """

    def __new__(self, name, value):
        value = int(value)
        self = super(Int, self).__new__(self, name, value)
        return self


class Float(_Metric):
    """A float metric.

    raises:
        ValueError if the passed value doesn't cast properly to float

    :param name: The name of the metric
    :type name: str
    :param value: The recorded value of the metric
    :type value: float
    :raises: ValueError
    """

    def __new__(self, name, value):
        value = float(value)
        self = super(Float, self).__new__(self, name, value)
        return self


class String(_Metric):
    """A string metric.

    raises:
        ValueError if the passed value doesn't cast properly to str

    :param name: The name of the metric
    :type name: str
    :param value: The recorded value of the metric
    :type value: str
    :raises: ValueError
    """

    def __new__(self, name, value):
        value = str(value)
        self = super(String, self).__new__(self, name, value)
        return self


class Boolean(_Metric):
    """A boolean metric.

    raises:
        ValueError if the passed value doesn't cast properly to bool

    :param name: The name of the metric
    :type name: str
    :param value: The recorded value of the metric
    :type value: bool
    :raises: ValueError
    """

    def __new__(self, name, value):
        value = bool(value)
        self = super(Boolean, self).__new__(self, name, value)
        return self


class Gauge(Float):
    """A gauge value - this is always a Float value.


    raises:
        ValueError if the passed value doesn't cast properly to int

    :param name: The name of the metric
    :type name: str
    :param value: The recorded value of the metric
    :type value: float
    :raises: ValueError
    """
    pass


class Counter(Int):
    """A metric that counts things - this is always an Integer value.


    raises:
        ValueError if the passed value doesn't cast properly to int

    :param name: The name of the metric
    :type name: str
    :param value: The recorded value of the metric
    :type value: int
    :raises: ValueError
    """
    pass


class Rate(Float):
    """A metric that describes a rate - this is always a Float.

    raises:
        ValueError if the passed value doesn't cast properly to float

    :param name: The name of the metric
    :type name: str
    :param value: The recorded value of the metric
    :type value: float
    :raises: ValueError
    """
    pass


class Timer(Float):
    """A metric that describes a timer value - this is always a Float.

    raises:
        ValueError if the passed value doesn't cast properly to float

    :param name: The name of the metric
    :type name: str
    :param value: The recorded value of the metric
    :type value: float
    :raises: ValueError
    """
    pass


class Meta(object):
    """Encapsulates generic key=value metadata however ensures that the
    values recorded are one of the four supported base types of Int, Float,
    String and Boolean. Also ensures keys are unique.

    :param metas: Variable list of values.
    :type metas: Int or Float or String or Boolean
    :raises: ValueError if metas is not one of the expected types or if any of
        the keys already exist.
    """
    __slots__ = [ '_meta', 'types' ]
    types = [Int, Float, String, Boolean]

    def __init__(self):
        self._meta = {}


    def __str__(self):
        """Return a human readable str.
        :rtype: str"""
        s = "Meta(items={0})"
        return s.format(sorted(self._meta.keys()))


    def __repr__(self):
        """Return a human readable str.
        :rtype: str"""
        s = "Meta(items={0})"
        return s.format(sorted(self._meta.keys()))


    @property
    def nkeys(self):
        """Return the number of key=value pairs."""
        return len(self._meta.keys())


    @property
    def keys(self):
        """Return a (by default) sorted list of metadata keys.

        :rtype: :class:`list`"""
        return sorted(self._meta.keys())

    @property
    def items(self):
        """Return a (by default) sorted list of (key, value) tuples.

        :rtype: :class:`list`"""
        mlist = [(k, self._meta[k]) for k in sorted(self._meta.keys())]
        return mlist


    def add(self, val):
        """Record a value to our metadata.

        :param val: The value to add.
        :type val: Int or Float or String or Boolean
        :raises: ValueError if val is an unexpected type or the
            keys already exist.
        """
        if val.__class__ not in self.types:
            cname = val.__class__.__name__
            err = "Class not supported: class: {0} : value: {1}"
            raise ValueError(err.format(cname, val))
        # using indexes: 0 => name, 1 => value
        if val[0] not in self._meta:
            self._meta[val[0]] = val
            return
        raise ValueError("key exists: {0}".format(val[0]))


class Result(object):
    """Encapsulates a set of metrics and an optional Meta metadata object. Does
    not check for duplicate metrics in the same Result.

    :param name: The name of the result eg. cpu
    :type name: str
    :param metrics: optional list of metrics
    :type metrics: Int or Float or String or Boolean or Gauge or Counter
        or Rate or Timer
    """
    __slots__ = [ '_name', '_metrics', '_meta', 'types' ]
    types = [Int, Float, String, Boolean, Gauge, Counter, Rate, Timer]

    def __init__(self, name, metrics=None):
        """Create a new Result."""
        self._name = name
        self._metrics = list()
        if metrics:
            for metric in metrics:
                mclass = metric.__class__
                if mclass not in self.types:
                    err = "Class not supported: class: {0} : metric: {1}"
                    raise ValueError(err.format(mclass.__name__, metric))
                self._metrics.append(metric)
        self._meta = Meta()


    def __str__(self):
        """Return a human readable str.
        :rtype: str"""
        s = "Result(name={0}, metrics={1})"
        return s.format(len(self._name), self._metrics)


    def __repr__(self):
        """Return a human readable str.
        :rtype: str"""
        s = "Result(name={0}, metrics={1})"
        return s.format(len(self._name), self._metrics)


    @property
    def name(self):
        """Return the Result name."""
        return self._name


    @property
    def metrics(self):
        """Return the Results list of metrics."""
        return self._metrics


    @property
    def meta(self):
        """Return the results metadata object."""
        return self._meta


    def add(self, metric):
        """Add a metric to our list of metrics.

        raises:
            ValueError if the metric is not a supported type

        :param metric: One of the metric types (Int, Float, String, etc)
        :type metric: Int or Float or String or Boolean or Counter or Gauge
        :raises: ValueError
        """
        mclass = metric.__class__
        if mclass not in self.types:
            err = "Class not supported: class: {0} : metric: {1}"
            raise ValueError(err.format(mclass.__name__, metric))
        self._metrics.append(metric)


    def add_list(self, metrics):
        """Add a list of metrics to our list of metrics.

        raises:
            ValueError if any metrics are not a supported type

        :param metrics: list of metrics (Int, Float, String, etc)
        :type metrics: list
        :raises: ValueError
        """
        for metric in metrics:
            mclass = metric.__class__
            if mclass not in self.types:
                err = "Class not supported: class: {0} : metric: {1}"
                raise ValueError(err.format(mclass.__name__, metric))
            self._metrics.append(metric)


class ResultSet(object):
    """A class to encapsulate a series of measurement Results during a plugins
    poll() call. Each poll() must return a ResultSet object.

    The class has a list of Result objects and a timestamp. The timestamp
    is set to the current UTC time when the object is created. This is done
    since each poll() call should normally complete within ms or us and the
    target audience for the recorded metrics are eg. grafana graphs and
    system alerts with resolutions down to 1 second typically. The difference
    in time between the first recorded metric and last is generally going to
    be much less than 1 second.

    :Example:

    >>>import plumd
    >>>metrics = [ plumd.Float("idle", 0.0), plumd.Float("user", 100.0) ]
    >>>res = Result("cpu", metrics)
    >>>rset = ResultSet(results=[res])

    >>>metrics = [ plumd.Float("free", 80.0), plumd.Float("used", 20.0) ]
    >>>res = Result("disk", metrics)
    >>>res.meta.add(plumd.String("dev", "sda1"))
    >>>rset = ResultSet(results=[res])
    """
    __slots__ = [ '_results', '_time' ] # do this to save memory

    def __init__(self, results=None):
        """Class to encapsulate a collection of metrics eg. from poll()."""
        self._results = list() if results is None else list(results)
        self._time = arrow.utcnow()


    @property
    def time(self):
        """Return our timestamp.
        :rtype: float
        """
        return self._time


    @property
    def nresults(self):
        """Return the number of results recorded so far.
        :rtype: int
        """
        return len(self._results)


    @property
    def results(self):
        """Yields a tuple for each Result recorded in the format:

        ( time, result_name, result_meta, [metric, metric, metric] )

        :rtype: generator
        """
        for robj in self._results:
            yield (self._time, robj.name, robj.meta, robj.metrics)


    def __str__(self):
        """Return a human readable str.
        :rtype: str"""
        s = "ResultSet(results={0}, time={1})"
        return s.format(len(self._results), self._time.timestamp)


    def __repr__(self):
        """Return a human readable str.
        :rtype: str"""
        s = "ResultSet(results={0}, time={1})"
        return s.format(len(self._results), self._time.timestamp)


    def add(self, result):
        """Record a result in this ResultSet.

        :param result: a Result object
        :type results: Result
        """
        self._results.append(result)


    def add_list(self, results):
        """Record a list of results in this ResultSet.

        :param results: a list of Result objects
        :type results: list
        """
        for result in results:
            self._results.append(result)


class Render(object):
    """A helper class to render metrics, buffer them and return chunks of
    metrics. Used as a superclass for the various metric writers (eg.
    influxdb, graphite, etc).

    The idea is to feed this object a ResultSet on each call to a writer
    plugins push() and then consume chunks of metrics from it until a full
    chunk cannot be returned.

    To use this, a writer plugin subclasses Render and defines the process
    method.The plugins implementation of process() should format the metrics it
    gets passed in a format suitable for sending to its backend and store them
    in the metrics dequeue.

    The plugin can then create an instance of it's render object and in push()
    simply call instance.process(results) and instance.get_chunk() to get the
    next chunk of metrics.

    On shutdown it can also call instance.get_chunk(partial=True) to get the
    remaining metrics. Normally it would want to fetch full batches of metrics.

    :param rconfig: config.Conf object passed from a writer plugin instance
    :type rconfig: :class:`config.Conf`
    """
    __metaclass__ = abc.ABCMeta

    def __init__(self, rconfig):
        # Use a double ended queue with a maximum size
        maxlen = rconfig.get('maxqueue', default=8192)
        self.metrics = collections.deque(maxlen=maxlen)
        # define a metadata object and copy in any configured host metadata
        self.meta = Meta()
        clookup = {
            int: Int,
            float: Float,
            str: String,
            bool: Boolean
        }
        for key, val in rconfig.get('meta', default={}).items():
            if val.__class__ not in clookup:
                args = ( key, val.__class__.__name__ )
                err = "host metadata: unsupported type: {0}={1}".format(*args)
                raise ValueError(err)
            self.meta.add(clookup[val.__class__](key, val))


    @abc.abstractmethod
    def process(self, rset):
        """Record a result set in our backends format. Subclasses for specific
        backends must define this method.

        :param rset: A :class:`ResultSet` object
        :type rset: :class:`ResultSet`
        """
        raise NotImplementedError("process must be defined by a subclass")


    def next_batch(self, bsize, partial=None):
        """Return the next full batch of metrics - if partial is True and
        self.metrics contains less than nmetrics metrics then return
        the remaining metrics.

        Returns a tuple of (queue size, the number of metrics returned,
        list of metrics).

        todo: also support number of bytes to return.

        :param bsize: the batch size of metrics to return
        :type bsize: int
        :param partial: return less than chunk_size metrics if not enough
        :type partial: bool
        :rtype: tuple(int, int, list)
        """
        qlen = len(self.metrics)
        mlist = []
        if qlen >= bsize:
            mlist = [ self.metrics.popleft() for i in range(0, bsize) ]
            qlen = len(self.metrics)
        elif partial:
            mlist = [ self.metrics.popleft() for i in range(0, qlen) ]
            qlen = len(self.metrics)
        return ( qlen, len(mlist), mlist )


1			# -- coding: utf-8 --
2			"""There are many differences in the various time series databases. Plumd is
3			designed to support multiple backends using the classes defined here to
4			represent metrics in a uniform way:
5
6			ResultSet - a collection of one or more Results from a call to poll()
7			Result - a measurement of a single thing eg. disk free=0.0,used=100.0
8			Meta - metadata associated with a Result eg. dev=sda1
9			Int/Float/String/Boolean/Gauge/Rate/Counter/Timer - primitives for above
10
11			The ResultSet class is used to encapsulate all of the measurements taken
12			during a plugins poll() call and uses a single timestamp for that poll.
13
14			The Result class is used to encapsulate a single measurement - for example
15			measuring cpu utilization may result in idle, user, system etc values. These
16			would be recorded in a single Result object. In addition the Result class
17			allows for arbitrary metadata to be associated with a Result - eg. when
18			recording disk space there may be used, free, total however also metadata such
19			as dev=sda1 to associate the measurements with a specific block device.
20
21			The Meta class is used to encapsulate arbitrary metadata. The Result class
22			uses it to associate metadata with a measurement however it can also be used
23			for example to describe host level metadata (eg. hostname=<hostname>). The
24			Meta class allows writer plugins such as graphite to render the metadata as
25			part of the metric path (eg. servers.host.<hostname>.disk.used.dev.<device>) -
26			if there was no distiction between metadata key=value and metric key=value this
27			would not be possible. Also, the Meta class ensures that metadata keys are
28			unique.
29
30			The _Metrics classes are used to allow plumd to render metrics in a specific
31			format. For example influxdb wants integer values to end with an i when sending
32			via their line protocol. The various _Metric classes force the plugin to
33			explicitly say what each metric is (eg. Int, Float, String, Boolean, Counter,
34			Gauge, Timer, etc). Metrics are defined as namedtuples for reduced memory
35			utilization.
36
37			The Render class is an abstract base class that writer plugins can subclass. The
38			render() function needs to be defined so that it returns a formatted metric. A
39			writer plugin would then instantiate the subclassed Render and use it to record
40			formatted metrics an each push() call. It can then step through the passed
41			metrics in chunks and send them to its backend. Having the metric types defined
42			explicitly allows the writers renderer to format metrics for arbitrary backends.
43
44
45			.. moduleauthor:: Kenny Freeman <[email protected]>
46
47			"""
48
49			__author__ = 'Kenny Freeman'
50			__email__ = '[email protected]'
51			__version__ = '0.3.3'
52			__license__ = "ISCL"
53			__docformat__ = 'reStructuredText'
54
55			import abc
56			import logging
57			import collections
58
59			import arrow # pip install arrow
60
61
62			DEFAULT_CONFIG = {
63			'log.level': "warn", # crit, error, warn, info, debug
64			'config.file': "/etc/plumd.yaml", # path to main configuration file
65			'config.plugins': "/etc/plumd.conf.d/", # path to the plugin confs
66			'delay.startup': 1, # random delay for plugin start
67			'delay.poll': 1, # random delay for first poll()
68			'poll.interval': 15, # interval of reader poll() calls
69			'max.queue': 512, # maximum size of internal queues
70			'meta': {}, # key=value's for all metrics
71			'meta.hostname': True # set and use hostname in meta
72			}
73
74
75			LOG_LEVELS = { # map config level to python level
76			'crit': logging.CRITICAL,
77			'error': logging.ERROR,
78			'warn': logging.WARN,
79			'info': logging.INFO,
80			'debug': logging.DEBUG
81			}
82
83
84			class DuplicatePlugin(Exception):
85			"""Two or more plugins with the same id were defined."""
86			pass
87
88
89			class PluginNotFoundError(Exception):
90			"""A plugin file, module or class was not found."""
91			pass
92
93
94			class PluginLoadError(Exception):
95			"""Exception(s) were raised while loading a plugin."""
96			pass
97
98
99			class ConfigError(Exception):
100			"""Invalid configuration encountered."""
101			pass
102
103
104			#_Metric : the base class of all metric primitives is a named tuple.
105			_Metric = collections.namedtuple('metric', ['name', 'value'])
106
107
108			class Int(_Metric):
109			"""An integer metric - ensures the value passed is an int.
110
111			raises:
112			ValueError if the passed value doesn't cast properly to int
113
114			:param name: The name of the metric
115			:type name: str
116			:param value: The recorded value of the metric
117			:type value: int
118			:raises: ValueError
119			"""
120
121			def __new__(self, name, value):
122			value = int(value)
123			self = super(Int, self).__new__(self, name, value)
124			return self
125
126
127			class Float(_Metric):
128			"""A float metric.
129
130			raises:
131			ValueError if the passed value doesn't cast properly to float
132
133			:param name: The name of the metric
134			:type name: str
135			:param value: The recorded value of the metric
136			:type value: float
137			:raises: ValueError
138			"""
139
140			def __new__(self, name, value):
141			value = float(value)
142			self = super(Float, self).__new__(self, name, value)
143			return self
144
145
146			class String(_Metric):
147			"""A string metric.
148
149			raises:
150			ValueError if the passed value doesn't cast properly to str
151
152			:param name: The name of the metric
153			:type name: str
154			:param value: The recorded value of the metric
155			:type value: str
156			:raises: ValueError
157			"""
158
159			def __new__(self, name, value):
160			value = str(value)
161			self = super(String, self).__new__(self, name, value)
162			return self
163
164
165			class Boolean(_Metric):
166			"""A boolean metric.
167
168			raises:
169			ValueError if the passed value doesn't cast properly to bool
170
171			:param name: The name of the metric
172			:type name: str
173			:param value: The recorded value of the metric
174			:type value: bool
175			:raises: ValueError
176			"""
177
178			def __new__(self, name, value):
179			value = bool(value)
180			self = super(Boolean, self).__new__(self, name, value)
181			return self
182
183
184			class Gauge(Float):
185			"""A gauge value - this is always a Float value.
186
187
188			raises:
189			ValueError if the passed value doesn't cast properly to int
190
191			:param name: The name of the metric
192			:type name: str
193			:param value: The recorded value of the metric
194			:type value: float
195			:raises: ValueError
196			"""
197			pass
198
199
200			class Counter(Int):
201			"""A metric that counts things - this is always an Integer value.
202
203
204			raises:
205			ValueError if the passed value doesn't cast properly to int
206
207			:param name: The name of the metric
208			:type name: str
209			:param value: The recorded value of the metric
210			:type value: int
211			:raises: ValueError
212			"""
213			pass
214
215
216			class Rate(Float):
217			"""A metric that describes a rate - this is always a Float.
218
219			raises:
220			ValueError if the passed value doesn't cast properly to float
221
222			:param name: The name of the metric
223			:type name: str
224			:param value: The recorded value of the metric
225			:type value: float
226			:raises: ValueError
227			"""
228			pass
229
230
231			class Timer(Float):
232			"""A metric that describes a timer value - this is always a Float.
233
234			raises:
235			ValueError if the passed value doesn't cast properly to float
236
237			:param name: The name of the metric
238			:type name: str
239			:param value: The recorded value of the metric
240			:type value: float
241			:raises: ValueError
242			"""
243			pass
244
245
246			class Meta(object):
247			"""Encapsulates generic key=value metadata however ensures that the
248			values recorded are one of the four supported base types of Int, Float,
249			String and Boolean. Also ensures keys are unique.
250
251			:param metas: Variable list of values.
252			:type metas: Int or Float or String or Boolean
253			:raises: ValueError if metas is not one of the expected types or if any of
254			the keys already exist.
255			"""
256			__slots__ = [ '_meta', 'types' ]
257			types = [Int, Float, String, Boolean]
258
259			def __init__(self):
260			self._meta = {}
261
262
263			def __str__(self):
264			"""Return a human readable str.
265			:rtype: str"""
266			s = "Meta(items={0})"
267			return s.format(sorted(self._meta.keys()))
268
269
270			def __repr__(self):
271			"""Return a human readable str.
272			:rtype: str"""
273			s = "Meta(items={0})"
274			return s.format(sorted(self._meta.keys()))
275
276
277			@property
278			def nkeys(self):
279			"""Return the number of key=value pairs."""
280			return len(self._meta.keys())
281
282
283			@property
284			def keys(self):
285			"""Return a (by default) sorted list of metadata keys.
286
287			:rtype: :class:`list`"""
288			return sorted(self._meta.keys())
289
290			@property
291			def items(self):
292			"""Return a (by default) sorted list of (key, value) tuples.
293
294			:rtype: :class:`list`"""
295			mlist = [(k, self._meta[k]) for k in sorted(self._meta.keys())]
296			return mlist
297
298
299			def add(self, val):
300			"""Record a value to our metadata.
301
302			:param val: The value to add.
303			:type val: Int or Float or String or Boolean
304			:raises: ValueError if val is an unexpected type or the
305			keys already exist.
306			"""
307			if val.__class__ not in self.types:
308			cname = val.__class__.__name__
309			err = "Class not supported: class: {0} : value: {1}"
310			raise ValueError(err.format(cname, val))
311			# using indexes: 0 => name, 1 => value
312			if val[0] not in self._meta:
313			self._meta[val[0]] = val
314			return
315			raise ValueError("key exists: {0}".format(val[0]))
316
317
318			class Result(object):
319			"""Encapsulates a set of metrics and an optional Meta metadata object. Does
320			not check for duplicate metrics in the same Result.
321
322			:param name: The name of the result eg. cpu
323			:type name: str
324			:param metrics: optional list of metrics
325			:type metrics: Int or Float or String or Boolean or Gauge or Counter
326			or Rate or Timer
327			"""
328			__slots__ = [ '_name', '_metrics', '_meta', 'types' ]
329			types = [Int, Float, String, Boolean, Gauge, Counter, Rate, Timer]
330
331			def __init__(self, name, metrics=None):
332			"""Create a new Result."""
333			self._name = name
334			self._metrics = list()
335			if metrics:
336			for metric in metrics:
337			mclass = metric.__class__
338			if mclass not in self.types:
339			err = "Class not supported: class: {0} : metric: {1}"
340			raise ValueError(err.format(mclass.__name__, metric))
341			self._metrics.append(metric)
342			self._meta = Meta()
343
344
345			def __str__(self):
346			"""Return a human readable str.
347			:rtype: str"""
348			s = "Result(name={0}, metrics={1})"
349			return s.format(len(self._name), self._metrics)
350
351
352			def __repr__(self):
353			"""Return a human readable str.
354			:rtype: str"""
355			s = "Result(name={0}, metrics={1})"
356			return s.format(len(self._name), self._metrics)
357
358
359			@property
360			def name(self):
361			"""Return the Result name."""
362			return self._name
363
364
365			@property
366			def metrics(self):
367			"""Return the Results list of metrics."""
368			return self._metrics
369
370
371			@property
372			def meta(self):
373			"""Return the results metadata object."""
374			return self._meta
375
376
377			def add(self, metric):
378			"""Add a metric to our list of metrics.
379
380			raises:
381			ValueError if the metric is not a supported type
382
383			:param metric: One of the metric types (Int, Float, String, etc)
384			:type metric: Int or Float or String or Boolean or Counter or Gauge
385			:raises: ValueError
386			"""
387			mclass = metric.__class__
388			if mclass not in self.types:
389			err = "Class not supported: class: {0} : metric: {1}"
390			raise ValueError(err.format(mclass.__name__, metric))
391			self._metrics.append(metric)
392
393
394			def add_list(self, metrics):
395			"""Add a list of metrics to our list of metrics.
396
397			raises:
398			ValueError if any metrics are not a supported type
399
400			:param metrics: list of metrics (Int, Float, String, etc)
401			:type metrics: list
402			:raises: ValueError
403			"""
404			for metric in metrics:
405			mclass = metric.__class__
406			if mclass not in self.types:
407			err = "Class not supported: class: {0} : metric: {1}"
408			raise ValueError(err.format(mclass.__name__, metric))
409			self._metrics.append(metric)
410
411
412			class ResultSet(object):
413			"""A class to encapsulate a series of measurement Results during a plugins
414			poll() call. Each poll() must return a ResultSet object.
415
416			The class has a list of Result objects and a timestamp. The timestamp
417			is set to the current UTC time when the object is created. This is done
418			since each poll() call should normally complete within ms or us and the
419			target audience for the recorded metrics are eg. grafana graphs and
420			system alerts with resolutions down to 1 second typically. The difference
421			in time between the first recorded metric and last is generally going to
422			be much less than 1 second.
423
424			:Example:
425
426			>>>import plumd
427			>>>metrics = [ plumd.Float("idle", 0.0), plumd.Float("user", 100.0) ]
428			>>>res = Result("cpu", metrics)
429			>>>rset = ResultSet(results=[res])
430
431			>>>metrics = [ plumd.Float("free", 80.0), plumd.Float("used", 20.0) ]
432			>>>res = Result("disk", metrics)
433			>>>res.meta.add(plumd.String("dev", "sda1"))
434			>>>rset = ResultSet(results=[res])
435			"""
436			__slots__ = [ '_results', '_time' ] # do this to save memory
437
438			def __init__(self, results=None):
439			"""Class to encapsulate a collection of metrics eg. from poll()."""
440			self._results = list() if results is None else list(results)
441			self._time = arrow.utcnow()
442
443
444			@property
445			def time(self):
446			"""Return our timestamp.
447			:rtype: float
448			"""
449			return self._time
450
451
452			@property
453			def nresults(self):
454			"""Return the number of results recorded so far.
455			:rtype: int
456			"""
457			return len(self._results)
458
459
460			@property
461			def results(self):
462			"""Yields a tuple for each Result recorded in the format:
463
464			( time, result_name, result_meta, [metric, metric, metric] )
465
466			:rtype: generator
467			"""
468			for robj in self._results:
469			yield (self._time, robj.name, robj.meta, robj.metrics)
470
471
472			def __str__(self):
473			"""Return a human readable str.
474			:rtype: str"""
475			s = "ResultSet(results={0}, time={1})"
476			return s.format(len(self._results), self._time.timestamp)
477
478
479			def __repr__(self):
480			"""Return a human readable str.
481			:rtype: str"""
482			s = "ResultSet(results={0}, time={1})"
483			return s.format(len(self._results), self._time.timestamp)
484
485
486			def add(self, result):
487			"""Record a result in this ResultSet.
488
489			:param result: a Result object
490			:type results: Result
491			"""
492			self._results.append(result)
493
494
495			def add_list(self, results):
496			"""Record a list of results in this ResultSet.
497
498			:param results: a list of Result objects
499			:type results: list
500			"""
501			for result in results:
502			self._results.append(result)
503
504
505			class Render(object):
506			"""A helper class to render metrics, buffer them and return chunks of
507			metrics. Used as a superclass for the various metric writers (eg.
508			influxdb, graphite, etc).
509
510			The idea is to feed this object a ResultSet on each call to a writer
511			plugins push() and then consume chunks of metrics from it until a full
512			chunk cannot be returned.
513
514			To use this, a writer plugin subclasses Render and defines the process
515			method.The plugins implementation of process() should format the metrics it
516			gets passed in a format suitable for sending to its backend and store them
517			in the metrics dequeue.
518
519			The plugin can then create an instance of it's render object and in push()
520			simply call instance.process(results) and instance.get_chunk() to get the
521			next chunk of metrics.
522
523			On shutdown it can also call instance.get_chunk(partial=True) to get the
524			remaining metrics. Normally it would want to fetch full batches of metrics.
525
526			:param rconfig: config.Conf object passed from a writer plugin instance
527			:type rconfig: :class:`config.Conf`
528			"""
529			__metaclass__ = abc.ABCMeta
530
531			def __init__(self, rconfig):
532			# Use a double ended queue with a maximum size
533			maxlen = rconfig.get('maxqueue', default=8192)
534			self.metrics = collections.deque(maxlen=maxlen)
535			# define a metadata object and copy in any configured host metadata
536			self.meta = Meta()
537			clookup = {
538			int: Int,
539			float: Float,
540			str: String,
541			bool: Boolean
542			}
543			for key, val in rconfig.get('meta', default={}).items():
544			if val.__class__ not in clookup:
545			args = ( key, val.__class__.__name__ )
546			err = "host metadata: unsupported type: {0}={1}".format(*args)
547			raise ValueError(err)
548			self.meta.add(clookup[val.__class__](key, val))
549
550
551			@abc.abstractmethod
552			def process(self, rset):
553			"""Record a result set in our backends format. Subclasses for specific
554			backends must define this method.
555
556			:param rset: A :class:`ResultSet` object
557			:type rset: :class:`ResultSet`
558			"""
559			raise NotImplementedError("process must be defined by a subclass")
560
561
562			def next_batch(self, bsize, partial=None):
563			"""Return the next full batch of metrics - if partial is True and
564			self.metrics contains less than nmetrics metrics then return
565			the remaining metrics.
566
567			Returns a tuple of (queue size, the number of metrics returned,
568			list of metrics).
569
570			todo: also support number of bytes to return.
571
572			:param bsize: the batch size of metrics to return
573			:type bsize: int
574			:param partial: return less than chunk_size metrics if not enough
575			:type partial: bool
576			:rtype: tuple(int, int, list)
577			"""
578			qlen = len(self.metrics)
579			mlist = []
580			if qlen >= bsize:
581			mlist = [ self.metrics.popleft() for i in range(0, bsize) ]
582			qlen = len(self.metrics)
583			elif partial:
584			mlist = [ self.metrics.popleft() for i in range(0, qlen) ]
585			qlen = len(self.metrics)
586			return ( qlen, len(mlist), mlist )
587

s4z / plumd

Push — master ( 8d751b...7e5eed )

plumd.ResultSet.add_list() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like