Reader.poll() - Code Metrics - Inspection of "fixup syntax" - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Push — master ( c60d6f...54bb83 )

by Kenny
created 2017-03-16 13:40 UTC
Reader.poll() A

↳ Parent: Project
Complexity

Conditions
Size

Total Lines
Duplication

Lines	0
Ratio	0 %
Importance

Changes	1
Bugs	0	Features	0
Metric	Value
c	1
b	0
f	0
dl	0
loc	12
rs	9.4285
cc	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 <[email protected]>

"""

import os
import time
import random
import logging
import threading
import collections
from collections import deque
from abc import ABCMeta, abstractmethod

from plumd.util import Interval, load_all_plugins, config_plugin_writers

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


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__(cls, name, value):
        """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
        """
        value = int(value)
        cls = super(Int, cls).__new__(cls, name, value)
        return cls


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__(cls, name, value):
        """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
        """
        value = float(value)
        cls = super(Float, cls).__new__(cls, name, value)
        return cls


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__(cls, name, value):
        """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
        """
        value = str(value)
        cls = super(String, cls).__new__(cls, name, value)
        return cls


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__(cls, name, value):
        """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
        """
        value = bool(value)
        cls = super(Boolean, cls).__new__(cls, name, value)
        return cls


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):
    """Encapsulate generic key=value metadata however ensure 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.
    """

    types = [Int, Float, String, Boolean]

    def __init__(self):
        """Encapsulate generic key=value metadata however ensure 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.
        """
        self._meta = collections.OrderedDict()

    def __str__(self):
        """Return a human readable str.

        :rtype: str
        """
        sval = "Meta(items={0})"
        vals = [ "{0}={1}".format(val[0], val[1]) for val in self._meta.items() ]
        return sval.format(",".join(vals))

    def __repr__(self):
        """Return a human readable str.

        :rtype: str
        """
        sval = "Meta(items={0})"
        vals = [ "{0}={1}".format(val[0], val[1]) for val in self._meta.items() ]
        return sval.format(",".join(vals))

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

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

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

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

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

    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):
    """Encapsulate a set of metrics and an optional Meta metadata object.

    :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
    """

    types = [Int, Float, String, Boolean, Gauge, Counter, Rate, Timer]

    def __init__(self, name, metrics=None):
        """Encapsulate a set of metrics and an optional Meta metadata object.

        :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
        """
        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
        """
        sval = "Result(name={0}, metrics={1})"
        return sval.format(self._name, self._metrics)

    def __repr__(self):
        """Return a human readable str.

        :rtype: str
        """
        sval = "Result(name={0}, metrics={1})"
        return sval.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])
    """

    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):
        """Yield 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
        """
        sval = "ResultSet(results={0}, time={1})"
        return sval.format(len(self._results), self._time)

    def __repr__(self):
        """Return a human readable str.

        :rtype: str
        """
        sval = "ResultSet(results={0}, time={1})"
        return sval.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__ = ABCMeta

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

    def __init__(self, rconfig):
        """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`
        """
        # 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 entry in rconfig.get('meta'):
            for key, val in entry.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()

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

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


class PluginLoader(object):
    """Load and control the reader and writer objects defined by
    the provided config.

    raises:
        ConfigError if a plugin is configured with an invalid path
        ConfigError if a plugin configuration is missing 'name'
        DuplicatePlugin if a plugin configuration has a duplicate 'name'
        PluginLoadError if there was an error loading the plugin

    :param config: a simple configuration helper object.
    :type config: plumd.config.conf
    :param log: a logger object
    :type log: logger
    :raises: ConfigError, DuplicatePlugin, PluginLoadError
    """

    def __init__(self, log, config):
        """Load and control the reader and writer objects defined by
        the provided config.

        raises:
            ConfigError if a plugin is configured with an invalid path
            ConfigError if a plugin configuration is missing 'name'
            DuplicatePlugin if a plugin configuration has a duplicate 'name'
            PluginLoadError if there was an error loading the plugin

        :param config: a simple configuration helper object.
        :type config: plumd.config.conf
        :param log: a logger object from structlog.get_logger()
        :type log: logger
        :raises: ConfigError, DuplicatePlugin, PluginLoadError
        """
        self.log = log
        msg = "initializing from: {0} conf: \n{1}"
        self.log.debug(msg.format(config.path, config))
        self.config = config
        self.readers = {}  # reader plugin io objects
        self.writers = {}  # writer plugin io objects
        self.load()

    def __str__(self):
        """Return a human readable str.

        :rtype: str
        """
        sval = "readers={0}, writers={1}".format(list(self.readers.keys()),
                                                 list(self.writers.keys()))
        return sval

    def __repr__(self):
        """Return a human readable str.

        :rtype: str
        """
        sval = "readers={0}, writers={1}".format(list(self.readers.keys()),
                                                 list(self.writers.keys()))
        return sval

    @property
    def nplugins(self):
        """Return the number of loaded plugins.

        :rtype: int
        """
        return len(list(self.readers.keys())) + len(list(self.writers.keys()))

    @property
    def nreaders(self):
        """Return the number of loaded reader plugins.

        :rtype: int
        """
        return len(list(self.readers.keys()))

    @property
    def nwriters(self):
        """Return the number of loaded writer plugins.

        :rtype: int
        """
        return len(list(self.writers.keys()))

    def load(self):
        """Load the configured plugin objects into self.readers and
        self.writers. Set the writers for each reader plugin.

        raises:
            ConfigError if ptype has an incorrect configuration path set
            ConfigError if a plugin configuration is missing 'name'
            DuplicatePlugin if a plugin configuration has a duplicate 'name'
            PluginLoadError if there was an error loading a plugin

        :raises: ConfigError, DuplicatePlugin
        """
        # get plugin configuration directory
        pdir = self.config.get('config.plugins')
        if not os.path.isdir(pdir):
            msg = "invalid plugin directory configured: {0}"
            raise ConfigError(msg.format(pdir))

        # load all plugins described by the plugin configurations
        msg = "loading plugins from: {0}"
        self.log.info(msg.format(self.config.get('config.plugins')))
        pobjs = load_all_plugins(self.log, self.config)

        # load_all_plugins returns a tuple of reader, writer objects
        self.readers, self.writers = pobjs

        # now the readers need to know what writers to write to
        config_plugin_writers(self)

    def start(self):
        """Start each reader/writer."""
        self.log.debug("starting")

        # start the writers then the readers
        for pdict in [self.writers, self.readers]:
            for pobj in pdict.values():
                self.log.debug("starting: {0}".format(pobj.name))
                pobj.start()

        # all done
        self.log.debug("started")

    def stop(self):
        """Stop each reader/writer."""
        self.log.debug("stopping plugins")

        # stop the readers first - they call each plugins onstop()
        self.log.debug("stopping readers")
        for pobj in list(self.readers.values()):
            self.log.debug("stopping reader: {0}".format(pobj.name))
            pobj.stop()

        # next stop the writers - they call each plugins onstop()
        self.log.debug("stopping writers")
        for pobj in list(self.writers.values()):
            self.log.debug("stopping writer: {0}".format(pobj.name))
            pobj.stop()

        # join threads now
        for pdict in [self.readers, self.writers]:
            for pname, pobj in list(pdict.items()):
                self.log.debug("waiting for plugin to stop: {0}".format(pname))
                pobj.join(timeout=self.config.get('shutdown_timeout'))

        # all done
        self.log.debug("stopped")

    def __del__(self):
        """Log __del__ calls."""
        self.log.debug("PluginLoader: __del__")


class Plugin(object):
    """The base class for all plugins - each plugin must sub class.

    :param config: an instance of plumd.conf configuration helper
    :type config: conf
    """

    __metaclass__ = ABCMeta
    defaults = {}

    def __init__(self, log, config):
        """The base class for all plugins - each plugin must sub class.

        :param config: an instance of plumd.conf configuration helper
        :type config: conf
        """
        self.config = config
        self.log = log
        # plugins can set their own defaults byt defining self.defaults
        self.config.defaults(self.defaults)
        # all plugins must be given a unique name
        self.name = config.get('name', 'unknown')
        # thread for reading or writing
        self.thread = None
        self.stop_evt = threading.Event()

    def __str__(self):
        """Return a human readable str.

        :rtype: str
        """
        sval = "Plugin(class={0})".format(self.__class__.__name__)
        return sval

    def __repr__(self):
        """Return a human readable str.

        :rtype: str
        """
        sval = "Plugin(class={0})".format(self.__class__.__name__)
        return sval

    def start(self):
        """Start run() in a new thread if not already running."""
        if self.thread is None or not self.thread.is_alive():
            self.thread = threading.Thread(target=self.run, group=None)
            self.thread.start()
            self.log.debug("Plugin {0} started".format(self.name))
        else:
            self.log.debug("Plugin {0} already running".format(self.name))

    def run(self):
        """Plugin sub classes must overide this."""
        raise NotImplementedError("Invalid Plugin")

    def stop(self):
        """Set our stop_evt Event."""
        # set the stop event and the self.read thread falls through.
        self.stop_evt.set()

    def join(self, timeout):
        """Join our thread."""
        if self.thread:
            self.thread.join(timeout)
            if self.thread.is_alive():
                msg = "Plugin {0} shutdown timeout - thread blocking exit"
                self.log.error(msg.format(self.name))

    def onstart(self):
        """Call before the first call to push()."""
        self.log.debug("Plugin: onstart: {0}".format(self.config.get('name')))

    def onstop(self):
        """Call just before the main process exits."""
        self.log.debug("Plugin: onstop: {0}".format(self.config.get('name')))

    def __del__(self):
        """Log __del__ calls."""
        self.log.debug("Plugin: {0} __del__".format(self.name))


class Reader(Plugin):
    """An abstract base class that all writer pluguns subclass."""

    __metaclass__ = ABCMeta

    defaults = {
        'poll.interval': 30,    # seconds between polls
        'delay.poll': 5         # random delay before first poll
    }

    def __init__(self, log, config):
        """Reader plugin abstract class.

        :param log: A logger
        :type log: logging.RootLogger
        :param config: a plumd.config.Conf configuration helper instance.
        :type config: plumd.config.Conf
        """
        super(Reader, self).__init__(log, config)
        self.config.defaults(Reader.defaults)
        # list of queues to write to
        self.queues = []

    def run(self):
        """Call the plugins onstart() function and then iterate over the
        plugins poll() function, pushing the result to each writer queue.

        When complete it calls the plugins onstop() function.
        """
        # starting up, call onstart
        self.onstart()

        # distribute metrics - don't everyone flood all at once
        time.sleep(random.randrange(int(self.config.get('delay.poll'))))

        # save a few cycles
        tloop = self.config.get('poll.interval')
        evt = self.stop_evt
        queues = self.queues

        # loop on calling the plugin objects poll and sending to writers
        while not evt.is_set():

            # loop every self.interval seconds
            with Interval(tloop, evt):

                # poll for metric result set
                rset = self.poll()

                # readers can return None values eg. a reader that polls
                # very often and periodically returns results
                if rset is None:
                    continue

                # write the value to each of our writer queues directly
                for queue_evt, queue in queues:
                    queue.append(rset)
                    queue_evt.set()

        # all done, call onstop
        self.onstop()

    @abstractmethod
    def poll(self):
        """Reader plugins must define this method, it measures and returns a
        metrics string.

        format <name:str> <value:float|int> <timestamp:time.time()>\n[...]
        set to change to a python object.

        :rtype: str
        """
        # this will never get called
        raise NotImplementedError("poll() not defined")


class Writer(Plugin):
    """An abstract base class that all writer pluguns subclass."""

    __metaclass__ = ABCMeta

    defaults = {
        'maxqueue': 8192
    }

    def __init__(self, log, config):
        """Reader plugin abstract class.

        :param log: A logger
        :type log: logging.RootLogger
        :param config: a plumd.config.Conf configuration helper instance.
        :type config: plumd.config.Conf
        """
        super(Writer, self).__init__(log, config)
        self.config.defaults(Writer.defaults)
        # list of queues to write to
        self.queue = deque()
        # Event signaling items on queue
        self.queue_evt = threading.Event()

    def run(self):
        """Call the plugins onstart() function and then consume
        ResultSets from self.queue, calling self.push() for each.

        When complete it calls the plugins onstop() function.
        """
        # starting up, call onstart
        self.onstart()

        # save a few cycles
        queue_evt = self.queue_evt
        queue = self.queue

        # loop on popping from queue and calling self.push()
        while not self.stop_evt.is_set():
            # wait for a ResultSet
            queue_evt.wait()
            # clear the event - race condition however
            # next reader poll() will set the queue_evt again
            queue_evt.clear()
            try:
                # write it
                self.push(queue.popleft())
            except IndexError:
                self.log.debug("Writer: {0}: queue empty".format(self.name))

        # all done, call onstop
        self.onstop()

    def stop(self):
        """Set our stop_evt and queue_evt Events."""
        # set the stop event and the self.read thread falls through.
        self.stop_evt.set()
        self.queue_evt.set()

    @abstractmethod
    def push(self, rset):
        """Writer plugins must define this method, it accepts a ResultSet.

        :param rset: ResultSet from a readers poll() call.
        :type rset: plumd.ResultSet
        :type metrics: str
        """
        # this will never get called
        raise NotImplementedError("push() not defined")

s4z / plumd

Push — master ( c60d6f...54bb83 )

Reader.poll() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
