PluginWriter.join() - Code Metrics - Inspection of "added gauges, rates configuration values to proc r..." - s4z/plumd - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — master (7e8cc2)

by Kenny

created 2016-08-18 13:55 UTC

PluginWriter.join() C

↳ Parent: PluginWriter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
dl	0
loc	32
rs	5.5
c	0
b	0
f	0

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

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

"""
__author__ = 'Kenny Freeman'
__email__ = '[email protected]'
__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():

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

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


1		# -- coding: utf-8 --
2		"""Defines a class to read from a queue and write to a writer plugin.
3
4		.. moduleauthor:: Kenny Freeman <[email protected]>
5
6		"""
7		__author__ = 'Kenny Freeman'
8		__email__ = '[email protected]'
9		__license__ = "ISCL"
10		__docformat__ = 'reStructuredText'
11
12		import sys
13		PY3 = sys.version_info > (3,)
14
15		if PY3:
16		import queue as Queue
17		else:
18		import Queue
19
20		import time
21		import threading
22		import traceback
23
24		import plumd
25
26
27		class PluginWriter(object):
28		"""Writer plugin helper that provides a queue for reader plugins to queue
29		metrics onto. It runs a thread that writes to a plugin objects push().
30
31		:param log: A logger
32		:type log: logging.RootLogger
33		:param pobj: An instance of a writer plugin.
34		:type pobj: writer
35		:param pname: The configured name of the plugin.
36		:type pname: str
37		:param qsize: The maximum queue size to set.
38		:type qsize: int
39		"""
40
41
42		def __init__(self, log, pobj, pname, qsize):
43		"""
44		Writer plugin helper that provides a queue for reader plugins to queue
45		metrics onto. It runs a thread that writes to a plugin objects push().
46
47		:param log: A logger
48		:type log: logging.RootLogger
49		:param pobj: An instance of a writer plugin.
50		:type pobj: writer
51		:param pname: The configured name of the plugin.
52		:type pname: str
53		:param qsize: The maximum queue size to set.
54		:type qsize: int
55		"""
56		self.log = log
57		self._name = pname
58		# the queue readers enqueue metrics onto
59		self.queue = Queue.Queue(maxsize=qsize)
60		self.pobj = pobj
61		# the thread that dequeues metrics to pboj.push
62		self.thread = None
63		self.stop_evt = threading.Event()
64
65
66		@property
67		def name(self):
68		"""Simply return our configured name.
69		:rtype: str
70		"""
71		return self._name
72
73
74		def flush(self):
75		"""Flushes self.queue to self.pobj.push()."""
76		start = time.time()
77	View Code Duplication	while not self.queue.empty():
		1 ignored issue – show Duplication introduced 2016-08-13 11:41 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
78		rset = self.queue.get()
79		# pretty much any exception can get raised in a plugins push call
80		try:
81		self.pobj.push(rset)
82		except Exception as e:
83		tb = traceback.format_exc()
84		msg = "write: writer {0}: exception in flush: {1}: {2}"
85		self.log.debug(msg.format(self._name, e, tb))
86		return
87		if time.time() - start > 1:
88		msg = "write: flush: flush taking too long, dropping metrics"
89		self.log.error(msg)
90		break
91
92
93		def dequeue(self):
94		"""Dequeues from self.queue to self.pobj.push()."""
95		# onstart needs to be called first
96		try:
97		self.pobj.onstart()
98		except Exception as e:
99		tb = traceback.format_exc()
100		msg = "write: disabling writer {0}: exception in onstart: {1}: {2}"
101		self.log.error(msg.format(self._name, e, tb))
102		self.queue = Queue.Queue(maxsize=1)
103		return
104
105		# can be stopped in between polls by setting self.run=>False
106	View Code Duplication	while not self.stop_evt.is_set():
		1 ignored issue – show Duplication introduced 2016-08-13 11:41 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
107		rset = self.queue.get(block=True)
108		# we push None values onto the queue when exiting
109		if rset is None:
110		msg = "write: dequeued None value for writer {0}"
111		self.log.debug(msg.format(self._name))
112		continue
113		try:
114		self.pobj.push(rset)
115		except Exception as e:
116		tb = traceback.format_exc()
117		msg = "write: disabling writer {0}: exception in push: {1}: {2}"
118		self.log.debug(msg.format(self._name, e, tb))
119		self.stop_evt.set()
120		return
121
122
123		def start(self):
124		"""Starts a writer thread that dequeues into pobj.push()."""
125		self.thread = threading.Thread(target=self.dequeue, group=None)
126		self.thread.start()
127		self.log.debug("Writer {0} thread started".format(self._name))
128		if not self.thread.is_alive():
129		err = "Writer {0} thread failed to start".format(self._name)
130		raise plumd.PluginRuntimeError(err)
131
132
133		def stop(self):
134		"""Stops our thread, flush()s and calls pobj.stop()."""
135		self.stop_evt.set()
136
137
138		def onstop(self):
139		"""Calls the plugins onstop callable."""
140		self.flush()
141		# this can raise pretty much any exception
142		try:
143		self.pobj.onstop()
144		except Exception as e:
145		tb = traceback.format_exc()
146		msg = "write: writer {0}: exception in onstop: {1}: {2}"
147		self.log.error(msg.format(self._name, e, tb))
148
149
150		def join(self, timeout):
151		"""Joins our thread."""
152		# this function assumes self.stop() has been called
153		# first we need to put a null entry onto the queue to wake the thread
154		# if it is waiting on a blocking call to self.queue.get()
155		if self.thread:
156		# give the writer a short time to drain the queue
157		# then drop all metrics and force an exit
158		try:
159		self.queue.put(None, block=False)
160		self.queue.put(None, block=False)
161		except Queue.Full:
162		msg = "write: writer {0} queue full on stop"
163		self.log.warn(msg.format(self._name))
164		# give the writer some time to flush it's queue
165		self.thread.join(timeout)
166		if not self.thread.isAlive():
167		return
168		# if it's still running then forcibly empty its queue
169		if not self.queue.empty():
170		msg = "write: timeout exceeded: dropping metrics: writer {0}"
171		self.log.error(msg.format(self._name))
172		while not self.queue.empty():
173		self.queue.get(block=False)
174		# now put None values onto the queue to force a loop and exit
175		try:
176		self.queue.put(None, block=False)
177		self.queue.put(None, block=False)
178		except Queue.Full:
179		pass
180		# the thread should exit now
181		self.thread.join()
182
183
184		def __del__(self):
185		"""Mark __del__ in logs."""
186		self.log.debug("write: del: {0}".format(self._name))
187

s4z / plumd

Branch — master (7e8cc2)

PluginWriter.join() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like