Render

Complexity

Total Complexity

7

Size/Duplication

Total Lines	86
Duplicated Lines	0 %

Importance

Changes	3
Bugs	2	Features	2

4 Methods

Rating	Name	Duplication	Size	Complexity
A	process()	0	9	1
A	__next__()	0	9	2
A	__init__()	0	22	3
A	__iter__()	0	7	1

# -*- 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 <kenny.freeman@gmail.com>

"""

__author__ = 'Kenny Freeman'
__email__ = 'kenny.freeman@gmail.com'
__version__ = '0.3.3'
__license__     = "ISCL"
__docformat__   = 'reStructuredText'

import abc
import time
import logging
import collections


DEFAULT_CONFIG_FILE = "/etc/plumd.yaml",    # path to main configuration file

DEFAULT_CONFIG = {
    'log.level':        "warn",               # crit, error, warn, info, debug
    'log.format':       "[%(asctime)s] %(levelname)s %(process)d %(message)s",
    '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
    'shutdown_timeout': 10                    # thread shutdown timeout
}


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 PluginRuntimeError(Exception):
    """Exception(s) were raised while running 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(list(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(self._name, self._metrics)


    def __repr__(self):
        """Return a human readable str.
        :rtype: str"""
        s = "Result(name={0}, metrics={1})"
        return s.format(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 = time.time()


    @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)


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


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

        :param result: a Result object
        :type results: Result
        """
        if not isinstance(result, Result):
            raise ValueError("Invalid result: {0}".format(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:
            if not isinstance(result, Result):
                raise ValueError("Invalid result: {0}".format(result))
            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

    defaults = {
        'max_queue': 8192,      # maxlen of deque used for metrics
        'max_items': 64,        # maximum number of items to send at once
        'max_bytes': 1024,      # maximum number of bytes to send at once
        'meta': {}              # metadata to add to each metric
    }

    def __init__(self, rconfig):
        # ensure default configuration is set
        rconfig.defaults(Render.defaults)
        self.max_bytes = rconfig.get('max_bytes')
        self.max_items = rconfig.get('max_items')
        # Use a double ended queue with a maximum size
        maxlen = rconfig.get('max_queue')
        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').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))


    def __iter__(self):
        """A Render iterator returns rendered metrics where the metrics are
        each as close to self.max_bytes in size as possible.

        :rtype: iterator
        """
        return self


    def __next__(self):
        """Return the next metric from the deque, or raise StopIteration() if
        there are no more metrics.

        :rtype: object
        """
        if len(self.metrics) < 1:
            raise StopIteration()
        return self.metrics.popleft()


    @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")