WriterThread.run() - Code Metrics - Inspection of "more prometheus related changes - refactoring for..." - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Branch — prom (d3fc9e)

by Kenny
created 2017-03-01 08:01 UTC
WriterThread.run() B

↳ Parent: WriterThread
Complexity

Conditions
Size

Total Lines
Duplication

Lines	33
Ratio	100 %
Importance

Changes	4
Bugs	0	Features	0
Metric	Value
c	4
b	0
f	0
dl	33
loc	33
rs	7.5384
cc	6
# -*- 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.


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

"""

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

from plumd.util import Interval

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


DEFAULT_SERVER = 'null'

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
}

# possible values: threaded, basic
#   - basic for prometheus/server mode
#   - threaded for standard mode (readers run in threads, loop on sleep())
ENV_READER_CLASS = "PLUMD_READER_CLASS"
ENV_WRITER_CLASS = "PLUMD_WRITER_CLASS"


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):
        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):
        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):
        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):
        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})"
        return sval.format(sorted(self._meta.keys()))

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

        :rtype: str
        """
        sval = "Meta(items={0})"
        return sval.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 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.

    This is the return value from each Reader plugin poll() call.

    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.

        The tuple format is:

            ( 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 Plugin(object):
    """The base class for all Plugins."""

    __metaclass__ = ABCMeta
    defaults = {}

    def __init__(self, log, config):
        """Create a plugin.

        :param config: an instance of plumd.conf configuration helper
        :type config: conf
        """
        self.log = log
        self.config = config
        # 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', exception=True)

    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 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 reader pluguns must 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 base 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)
        self.interval = config.get("interval", exception=True)

    @abstractmethod
    def poll(self):
        """All Reader plugins must define this method, it returns a ResultSet.

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


class Render(Plugin):
    """Placeholder class to identify the various Render plugins."""

    pass


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

    __metaclass__ = ABCMeta

    defaults = {
        'maxqueue': 32
    }

    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)

    @abstractmethod
    def write(self, metrics):
        """Write the rendered metrics to our backend.

        What metrics is depends on the Render implementation.

        :param metrics: A rendered metric from our Render instance
        :type metrics: object
        """
        raise NotImplementedError("write must be defined by a subclass")

    @abstractmethod
    def flush(self, metrics):
        """Flush the rendered metrics to our backend.

        What metrics is depends on the Render implementation.

        This is used only during shutdown to send any remaining metrics
        from a Render plugin. Normally metrics are sent in chunks (depending
        on the Writer plugin) however during shutdown any partially chunked
        metrics need to be sent to the backend.

        Also, this function should return within a reasonable time (ie. it
        needs to have a timeout) as it is called during shutdown.

        :param metrics: A rendered metric from our Render instance
        :type metrics: object
        """
        raise NotImplementedError("write must be defined by a subclass")


class PluginThread(object):
    """Base class for Reader/Writer Thread objects."""

    __metaclass__ = ABCMeta
    defaults = {
        'thread.daemon': True
    }

    def __init__(self, log, config, plugin):
        """Base class for Reader/Writer Thread objects.

        :param log: A logger
        :type log: logging.RootLogger
        :param config: Thread configuration
        :type config: plumd.config.Conf
        :param plugin: A Reader/Writer plugin instance
        :type plugin: plumd.Plugin
        """
        self.log = log
        # plugins can set their own defaults byt defining self.defaults
        config.defaults(self.defaults)
        # thread for reading or writing
        self.thread = None
        self.stop_evt = threading.Event()
        self.plugin = plugin
        self.daemon = config.get("thread.daemon", exception=True)

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

    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)
            if self.daemon:
                self.thread.daemon = True
            self.thread.start()
            self.log.info("Plugin {0} started".format(self.plugin.name))
        else:
            self.log.warn("Plugin {0} already running".format(self.plugin.name))

    def stop(self):
        """Set our stop_evt Event to force our thread to (eventually) return."""
        self.stop_evt.set()

    def join(self, timeout):
        """Stop and wait up to timeout seconds for our thread to return.

        :param timeout: Number of seconds to wait for our thread to return.
        :type timeout: int
        """
        self.stop()
        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.plugin.name))


class ReaderThread(PluginThread):
    """Threading required to push metrics from Reader=>Writer."""

    __metaclass__ = ABCMeta

    defaults = {
        # random delay between 0 and delay.poll s for first poll() call
        # this is to prevent stampeeding herd after restarting many instances
        "delay.poll": 5
    }

    def __init__(self, log, config, plugin):
        """Create a ReaderThread, do not start it.

        :param log: A logger
        :type log: logging.RootLogger
        :param config: configuration for this reader thread
        :type config: plumd.config.Conf
        :param plugin: a plumd.Reader plugin instance.
        :type plugin: plumd.Reader
        """
        super(ReaderThread, self).__init__(log, config, plugin)
        config.defaults(ReaderThread.defaults)
        self.start_delay = config.get("delay.poll", exception=True)
        self.renders = deque()

    def register(self, render):
        """Add the Render object.

        :param render: A Render to send metrics to.
        :type render: plumd.Render
        """
        self.renders.append(render)

    def deregister(self, render):
        """Remove the Render object.

        :param render: A Render to send metrics to.
        :type render: plumd.Render
        """
        try:
            self.renders.remove(render)
        except ValueError as exc:
            err = "attempt to remove non-existant render: {0}"
            self.log.error(err.format(exc))

    def run(self):
        """Start reading from the reader Plugin after calling its onstart.

        Call stop() to get this thread to exit - before returning it will
        call the readers onstop().
        """
        if not self.queue or not self.render:
            err = "output queue/render not set for reader: {0}"
            raise PluginRuntimeError(err.format(self.plugin.name))

        # starting up, call onstart
        self.plugin.onstart()

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

        # save a few cycles
        tloop = self.plugin.interval
        evt = self.stop_evt
        poll = self.plugin.poll

        # 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 = 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 output queues directly
                for render in self.renders:
                    render.process(rset.results)

        # all done, call onstop
        self.plugin.onstop()


class WriterThread(PluginThread):
    """Threading required to isolate Readers from Writer backpressure."""

    __metaclass__ = ABCMeta

    defaults = {
        'maxqueue': 2048
    }

    def __init__(self, log, config, plugin):
        """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
        :param plugin: a plumd.Writer plugin instance.
        :type plugin: plumd.Writer
        """
        super(WriterThread, self).__init__(log, config, plugin)
        config.defaults(WriterThread.defaults)
        self.stop_evt = threading.Event()
        self.render = None
        self.queue = None

    def register(self, render):
        """Set the Render object.

        :param render: A Render to send metrics to.
        :type render: plumd.Render
        """
        if self.render:
            self.render.deregister(self.plugin.name)
        self.render = render
        self.queue = self.render.register(self.plugin.name)

    def deregister(self, render):
        """Un-Set the Render object.

        :param render: A Render instance.
        :type render: plumd.Render
        """
        if self.render:
            self.render.deregister(self.plugin.name)
            self.render = None
            self.queue = None

    def stop(self):
        """Set this threads stop event to force run to return."""
        self.stop_evt.set()
        # wakeup writer
        for i in range(0, 10):
            self.queue.put_nowait(None)

    def run(self):
        """Start writing metrics to self.writer from self.queue.

        First calls self.writer.onstart().

        Call stop() to get this thread to exit - before returning it will
        call the writers onstop().
        """

        if not self.queue or not self.render:
            err = "input queue/render not set for writer: {0}"
            raise PluginRuntimeError(err.format(self.plugin.name))

        # starting up, call onstart
        self.plugin.onstart()

        # save a few cycles
        stop_evt = self.stop_evt
        write = self.plugin.write

        # loop on popping from queue and calling self.push()
        while not stop_evt.is_set():
            entry = self.queue.get()
            if entry:
                write(entry)

        # flush any partially rendered metrics on shutdown
        partial = self.render.flush()
        if partial:
            self.plugin.flush(partial)

        # all done, call onstop
        self.plugin.onstop()
s4z / plumd

Branch — prom (d3fc9e)

WriterThread.run() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
