plumd.plugins.PluginWriter

Complexity

Total Complexity

17

Size/Duplication

Total Lines	141
Duplicated Lines	0 %

9 Methods

Rating	Name	Duplication	Size	Complexity
A	PluginWriter.flush()	0	12	3
B	PluginWriter.dequeue()	0	29	5
A	PluginWriter.stop()	0	3	1
A	PluginWriter.__del__()	0	3	1
A	PluginWriter.name()	0	6	1
A	PluginWriter.onstop()	0	10	2
A	PluginWriter.start()	0	4	1
A	PluginWriter.__init__()	0	22	1
A	PluginWriter.join()	0	9	2

# -*- 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 Queue
import threading
import traceback


class PluginWriter(object):
    """Writer plugin helper that provides a queue for reader plugins to queue
    metrics onto. It runs a thread that writes to a plugin objects push().

    :param log: A logger
    :type log: logging.RootLogger
    :param pobj: An instance of a writer plugin.
    :type pobj: writer
    :param pname: The configured name of the plugin.
    :type pname: str
    :param qsize: The maximum queue size to set.
    :type qsize: int
    """


    def __init__(self, log, pobj, pname, qsize):
        """
        Writer plugin helper that provides a queue for reader plugins to queue
        metrics onto. It runs a thread that writes to a plugin objects push().

        :param log: A logger
        :type log: logging.RootLogger
        :param pobj: An instance of a writer plugin.
        :type pobj: writer
        :param pname: The configured name of the plugin.
        :type pname: str
        :param qsize: The maximum queue size to set.
        :type qsize: int
        """
        self.log = log
        self._name = pname
        # the queue readers enqueue metrics onto
        self.queue = Queue.Queue(maxsize=qsize)
        self.pobj = pobj
        # the thread that dequeues metrics to pboj.push
        self.thread = None
        self.stop_evt = threading.Event()


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


    def flush(self):
        """Flushes self.queue to self.pobj.push()."""
        while not self.queue.empty():
            rset = self.queue.get()
            # pretty much any exception can get raised in a plugins push call
            try:
                self.pobj.push(rset)
            except Exception as e:
                tb = traceback.format_exc()
                msg = "write: disabling writer {0}: exception in flush: {1}: {2}"
                self.log.debug(msg.format(self._name, e, tb))
                return


    def dequeue(self):
        """Dequeues from self.queue to self.pobj.push()."""
        # onstart needs to be called first
        try:
            self.pobj.onstart()
        except Exception as e:
            tb = traceback.format_exc()
            msg = "write: disabling writer {0}: exception in dequeue: {1}: {2}"
            self.log.error(msg.format(self._name, e, tb))
            self.queue = Queue.Queue(maxsize=1)
            return

        # can be stopped in between polls by setting self.run=>False
        while not self.stop_evt.is_set():
            rset = self.queue.get(block=True)
            # we push None values onto the queue when exiting
            if rset is None:
                msg = "write: dequeue for writer {0}: ignoring null value"
                self.log.debug(msg.format(self._name))
                continue
            try:
                self.pobj.push(rset)
            except Exception as e:
                tb = traceback.format_exc()
                msg = "write: disabling writer {0}: exception in push: {1}: {2}"
                self.log.debug(msg.format(self._name, e, tb))
                self.stop_evt.set()
                return
        self.onstop()


    def start(self):
        """Starts a writer thread that dequeues into pobj.push()."""
        self.thread = threading.Thread(target=self.dequeue, group=None)
        self.thread.start()


    def stop(self):
        """Stops our thread, flush()s and calls pobj.stop()."""
        self.stop_evt.set()


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


    def join(self):
        """Joins our thread."""
        # this function assumes self.stop() has been called
        # first we need to put a null entry onto the queue to wake the thread
        # if it is waiting on a blocking call to self.queue.get()
        if self.thread:
            self.queue.put(None)
            self.queue.put(None)
            self.thread.join()


    def __del__(self):
        """Mark __del__ in logs."""
        self.log.debug("write: del: {0}".format(self._name))