PluginWriter

Complexity

Total Complexity

24

Size/Duplication

Total Lines	160
Duplicated Lines	18.13 %

Importance

Changes	1
Bugs	1	Features	1

9 Methods

Rating	Name	Duplication	Size	Complexity
A	flush()	14	17	4
B	dequeue()	15	28	5
A	stop()	0	3	1
A	__del__()	0	3	1
A	name()	0	6	1
A	onstop()	0	10	2
A	start()	0	8	2
A	__init__()	0	22	1
C	join()	0	32	7

# -*- 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,)

if PY3:
    import queue as Queue
else:
    import Queue

import time
import threading
import traceback

import plumd


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()."""
        start = time.time()
        while not self.queue.empty():

This code seems to be duplicated in your project.

            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: writer {0}: exception in flush: {1}: {2}"
                self.log.debug(msg.format(self._name, e, tb))
                return
            if time.time() - start > 1:
                msg = "write: flush: flush taking too long, dropping metrics"
                self.log.error(msg)
                break


    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 onstart: {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():

This code seems to be duplicated in your project.

            rset = self.queue.get(block=True)
            # we push None values onto the queue when exiting
            if rset is None:
                msg = "write: dequeued None value for writer {0}"
                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


    def start(self):
        """Starts a writer thread that dequeues into pobj.push()."""
        self.thread = threading.Thread(target=self.dequeue, group=None)
        self.thread.start()
        self.log.debug("Writer {0} thread started".format(self._name))
        if not self.thread.is_alive():
            err = "Writer {0} thread failed to start".format(self._name)
            raise plumd.PluginRuntimeError(err)


    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, timeout):
        """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:
            # give the writer a short time to drain the queue
            # then drop all metrics and force an exit
            try:
                self.queue.put(None, block=False)
                self.queue.put(None, block=False)
            except Queue.Full:
                msg = "write: writer {0} queue full on stop"
                self.log.warn(msg.format(self._name))
            # give the writer some time to flush it's queue
            self.thread.join(timeout)
            if not self.thread.isAlive():
                return
            # if it's still running then forcibly empty its queue
            if not self.queue.empty():
                msg = "write: timeout exceeded: dropping metrics: writer {0}"
                self.log.error(msg.format(self._name))
                while not self.queue.empty():
                    self.queue.get(block=False)
            # now put None values onto the queue to force a loop and exit
            try:
                self.queue.put(None, block=False)
                self.queue.put(None, block=False)
            except Queue.Full:
                pass
            # the thread should exit now
            self.thread.join()


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