PluginReader

Complexity

Total Complexity

26

Size/Duplication

Total Lines	178
Duplicated Lines	0 %

Importance

Changes	1
Bugs	1	Features	1

11 Methods

Rating	Name	Duplication	Size	Complexity
A	onstart()	0	9	2
A	__del__()	0	3	1
B	read()	0	23	6
A	start()	0	8	2
A	join()	0	7	3
A	onstop()	0	9	2
A	stop()	0	4	1
A	__init__()	0	23	1
A	name()	0	6	1
A	poll()	0	22	3
B	write()	0	27	4

# -*- coding: utf-8 -*-
"""Defines a class to read from a queue and write to a writer plugin.

.. moduleauthor:: Kenny Freeman <kenny.freeman@gmail.com>

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

import sys
PY3 = sys.version_info > (3,)

import time
if PY3:
    import queue as Queue
else:
    import Queue
import random
import threading
import traceback

import plumd
import plumd.util


class PluginReader(object):
    """Reads from a plugin instance by calling pobj.poll() and writes
    to a list of PluginWriters by calling <writerobj>.queue.put() - loops
    on a configured interval.

    :param log: A logger
    :type log: logging.get_logger()
    :param pobj: An instance of a Reader plugin.
    :type pobj: Reader
    :param pname: The configured name of the plugin.
    :type pname: str
    :param interval: The conigured polling interval for thep plugin.
    :type interval: int
    :raises: ConfigError, PluginLoadError
    """


    def __init__(self, log, pobj, pname):
        """Reads from a plugin instance by calling pobj.poll() and writes
        to a list of PluginWriters by calling <writerobj>.queue.put() - loops
        on a configured interval.

        :param log: A logger
        :type log: structlog.get_logger()
        :param pobj: An instance of a Reader plugin.
        :type pobj: Reader
        :param pname: The configured name of the plugin.
        :type pname: str
        :raises: ConfigError, PluginLoadError
        """
        self.log = log
        # poll() interval - get from plugins configuration
        self.interval = pobj.config.get('poll.interval')
        self._name = pname
        self.pobj = pobj
        # what writers should we write metrics to?
        self.writers = []
        # thread that polls for metrics and writes to writers
        self.thread = None
        self.stop_evt = threading.Event()


    @property
    def name(self):
        """Simply return our configured name.
        :rtype: str
        """
        return self._name


    def onstart(self):
        """Call the plugins onstart function."""
        try:
            self.pobj.onstart()
        except Exception as e:
            tb = traceback.format_exc()
            msg = "read: disabling reader {0}: exception in onstart: {1}: {2}"
            self.log.error(msg.format(self._name, e, tb))
            return


    def poll(self):
        """Calls the plugins poll() function and either returns a valid
        :class:`plumd.ResultSet` or None.

        :rtype: plumd.ResultSet
        """
        rset = None
        # plugins can raise pretty much any exception, wrap in catch all try
        try:
            rset = self.pobj.poll()
            if not isinstance(rset, plumd.ResultSet):
                msg = "read: invalid result set from reader {0}: {1}"
                self.log.error(msg.format(self._name, rset))
                rset = plumd.ResultSet()
        except Exception as e:
            # print a traceback and ensure the reader thread exits
            tb = traceback.format_exc()
            msg="read: disabling reader {0} due to exception in poll: {1} : {2}"
            self.log.error(msg.format(self._name, e, tb))
            self.stop_evt.set()
            rset = plumd.ResultSet()
        return rset


    def write(self, rset):
        """Iterates through each writer plugin and writes the
        :class:`plumd.ResultSet` to the writers input queue. Uses a
        :class:`plumd.utils.Interval` to determine if it will block when
        pushing to a writers queue. If there is time remaining in the current
        poll interval it will block. Also, if a writers queue is full it will
        drop the metric for that writer. Future versions may use persistance
        to queue result sets to disk if the backend metrics services are down.

        :param rset: a result set from a plugin poll() call.
        :type rset: plumd.ResultSet
        :param lint: an interval helper.
        :type lint: plumd.util.Interval
        """
        # only send result sets that have metrics
        if rset.nresults < 1:
            return

        # enqueue onto each writers queue, wait for any remaining time
        for wobj in self.writers:
            try:
                # allow queue puts to block for a short time
                # this cannot block for long as it will also block shutdown
                wobj.queue.put(rset, block=True, timeout=1)
            except Queue.Full:
                msg = "read: {0}: dropping metrics: writer plugin queue full: {1}"
                self.log.warn(msg.format(self._name, wobj.name))


    def read(self):
        """Calls the plugins onstart() function and then iterates over the
        plugins poll() function, pushing the result to each writer queue."""
        self.onstart()

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

        # loop on calling the plugin objects poll and sending to writers
        while not self.stop_evt.is_set():
            # loop every self.interval seconds
            with plumd.util.Interval(self.interval, self.stop_evt) as lint:
                if self.stop_evt.is_set():
                    break
                rset = self.poll()
                # readers can return None values eg. a reader that polls
                # very often and periodically returns results
                if rset:
                    self.write(rset)
                # warn if loop time exceeded
                if lint.remaining < 0:
                    msg = "read: {0} reader loop time exceeded"
                    self.log.warn(msg.format(self._name))


    def start(self):
        """Calls pobj.onstart() and starts a thread to poll() metrics."""
        self.thread = threading.Thread(target=self.read, group=None)
        self.thread.start()
        self.log.debug("Reader {0} thread started".format(self._name))
        if not self.thread.is_alive():
            err = "Reader {0} thread failed to start".format(self._name)
            raise plumd.PluginRuntimeError(err)


    def stop(self):
        """Stops the poll() thread and calls pobj.onstop()."""
        # set the stop event and the self.read thread falls through.
        self.stop_evt.set()


    def onstop(self):
        """Calls the plugins onstop callable."""
        # this can raise pretty much any exception
        try:
            self.pobj.onstop()
        except Exception as e:
            tb = traceback.format_exc()
            msg = "read: exception during onstop for reader {0}: {1}: {2}"
            self.log.error(msg.format(self._name, e, tb))


    def join(self, timeout):
        """Joins our thread."""
        if self.thread:
            self.thread.join(timeout)
            if self.thread.isAlive():
                msg = "read: plugin {0} shutdown timeout - blocking exit"
                self.log.error(msg.format(self._name))


    def __del__(self):
        """Logs __del__ calls."""
        self.log.debug("read: del: {0}".format(self._name))