GraphiteWriteTCP.write() - 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

GraphiteWriteTCP.write() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	5
Bugs	1	Features	3

Metric	Value
c	5
b	1
f	3
dl	0
loc	23
rs	8.2508
cc	5

# -*- coding: utf-8 -*-
"""Built-in Plumd Graphite Writer supporting plain text tcp/udp."""
import sys
import time

import socket
import threading

from plumd import Writer
from plumd.util import Interval

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


PY3 = sys.version_info > (3,)


class Graphite(Writer):
    """Graphite TCP writer plugin."""

    # todo: move things to render, writers, remove from here
    # also todo: print warning for queue size in render not in Graphite
    defaults = {
        'prefix':           "servers",
        'host':             '127.0.0.1',
        'port':             2003,
        'seperator':        "=",     # metadata seperator
        # any characters not in this value are removed from metrics/meta data
        'filter_chars':     'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-',
        'meta_key':         False,   # include metadata keys in metrics
        'max_bytes':        1400,    # send metrics in chunks of this many bytes
        'retry_time':       30,      # delay between connection attempts
        'timeout':          5,       # timeout for socket operations
        'maxqueue':         64,      # maximum number of metrics to queue
        'warnqueue':        32,      # print warning if queue size > this
        'tcp_ka_idle':      60,      # send keepalives if idle (seconds)
        'tcp_ka_intvl':     5,       # send unacked keepalives (seconds)
        'tcp_ka_count':     5        # up to this many unacked keepalives
    }

    def __init__(self, log, config):
        """Graphite sender.

        :param log: A logger
        :type log: logging.RootLogger
        :param config: a plumd.config.Conf configuration helper instance.
        :type config: plumd.config.Conf
        """
        super(Graphite, self).__init__(log, config)
        config.defaults(Graphite.defaults)
        self.addr = (config.get('host'), config.get('port'))
        self.socket = None
        self.evt_stop = threading.Event()

    def onstop(self):
        """Set stop event to force any writes to fall through and exit."""
        self.evt_stop.set()
        msg = "Graphite: onstop: {0}: setting stop event"
        self.log.info(msg.format(self.config.get('name')))

    def connect(self):
        """Connect to graphite and maintain a connection, return success of operation.

        :rtype: bool
        """
        timeout = self.config.get('timeout')
        tcp_ka_idle = self.config.get('tcp_ka_idle')
        tcp_ka_intvl = self.config.get('tcp_ka_intvl')
        tcp_ka_count = self.config.get('tcp_ka_count')
        # disconnect if we're already connected
        if self.socket:
            self.disconnect()
        try:
            # create the socket + connection w/ timeout
            self.socket = socket.create_connection(self.addr, timeout=timeout)
            # enable tcp keepalive with more aggressive values than default
            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
            self.socket.setsockopt(socket.SOL_TCP, socket.TCP_KEEPIDLE, tcp_ka_idle)
            self.socket.setsockopt(socket.SOL_TCP, socket.TCP_KEEPINTVL, tcp_ka_intvl)
            self.socket.setsockopt(socket.SOL_TCP, socket.TCP_KEEPCNT, tcp_ka_count)
            # set timeout for socket operations
            self.socket.settimeout(timeout)
            # log and return
            msg = "Graphite: connected: {0}"
            self.log.info(msg.format(self.addr))
            return True
        except Exception as e:

            # log exception, ensure cleanup is done (disconnect)
            msg = "Graphite: {0}: connect: exception: {1}"
            self.log.error(msg.format(self.addr, e))
            if self.socket:
                try:
                    self.socket.close()
                except Exception as exc:

                    self.socket = None
                self.socket = None
        return False

    def disconnect(self):
        """Close the tcp connection to the graphite server."""
        if self.socket:
            try:
                self.socket.close()
                self.socket = None
            except Exception as e:

                msg = "Graphite: disconnect: exception {0}".format(e)
                self.log.error(msg)
                self.socket = None

    def write(self, metrics):
        """Send the already rendered metrics string to graphite with retry.

        :param metrics: the metrics string to send.
        :type metrics: str
        :rtype: bool
        """
        # connect to graphite
        while (not self.evt_stop.is_set()) and (not self.socket) and (not self.connect()):
            # reconnect on an interval
            with Interval(self.config.get('retry_time'), self.evt_stop) as wait:
                msg = "re-connecting to graphite host {0}:{1} in {2:.1f} s"
                self.log.info(msg.format(self.addr[0], self.addr[1], wait.remaining))
        # "this should never happen"
        if (not self.evt_stop.is_set()) and (not self.socket):
            msg = "graphite: socket not connected after connect, dropping metrics"
            self.log.error(msg)
            return
        # send metrics with retry
        while not self.evt_stop.is_set():
            try:
                if PY3:
                    self.socket.sendall(bytes(metrics, 'utf8'))
                else:
                    self.socket.sendall(metrics)
                break
            except Exception as e:

                msg = "Graphite: {0}:{1}: write: exception: {2}"
                self.log.error(msg.format(self.addr[0], self.addr[1], e))
                self.connect()

    def flush(self, metrics):
        """Flush the metrics string to graphite, return immediately if not connected.

        This is used on shutdown.

        :param metrics: the metrics string to send.
        :type metrics: str
        :rtype: bool
        """
        nbytes = len(metrics)
        msg = "Graphite: {0}:{1} flushing {2} bytes"
        self.log.debug(msg.format(self.addr[0], self.addr[1], nbytes))
        if not self.socket:
            msg = "Graphite: {0}:{1} not connected, dropping metrics"
            self.log.debug(msg.format(self.addr[0], self.addr[1]))
            return
        try:
            if PY3:
                self.socket.sendall(bytes(metrics, 'utf8'))
            else:
                self.socket.sendall(metrics)
            msg = "Graphite: {0}:{1} {2} bytes flushed"
            self.log.debug(msg.format(self.addr[0], self.addr[1], nbytes))
        except Exception as e:

            msg = "Graphite: flush: {0}:{1}: write: exception: dropping metrics: {2}"
            self.log.error(msg.format(self.addr[0], self.addr[1], e))



1			# -- coding: utf-8 --
2			"""Built-in Plumd Graphite Writer supporting plain text tcp/udp."""
3			import sys
4			import time
			0 ignored issues – show Unused Code introduced 2016-09-10 09:25 UTC by Report Bug Copy Issue Report The import `time` seems to be unused. Loading history...
5			import socket
6			import threading
7
8			from plumd import Writer
9			from plumd.util import Interval
10
11			__author__ = 'Kenny Freeman'
12			__email__ = '[email protected]'
13			__license__ = "ISCL"
14			__docformat__ = 'reStructuredText'
15
16
17			PY3 = sys.version_info > (3,)
18
19
20			class Graphite(Writer):
21			"""Graphite TCP writer plugin."""
22
23			# todo: move things to render, writers, remove from here
24			# also todo: print warning for queue size in render not in Graphite
25			defaults = {
26			'prefix': "servers",
27			'host': '127.0.0.1',
28			'port': 2003,
29			'seperator': "=", # metadata seperator
30			# any characters not in this value are removed from metrics/meta data
31			'filter_chars': 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-',
32			'meta_key': False, # include metadata keys in metrics
33			'max_bytes': 1400, # send metrics in chunks of this many bytes
34			'retry_time': 30, # delay between connection attempts
35			'timeout': 5, # timeout for socket operations
36			'maxqueue': 64, # maximum number of metrics to queue
37			'warnqueue': 32, # print warning if queue size > this
38			'tcp_ka_idle': 60, # send keepalives if idle (seconds)
39			'tcp_ka_intvl': 5, # send unacked keepalives (seconds)
40			'tcp_ka_count': 5 # up to this many unacked keepalives
41			}
42
43			def __init__(self, log, config):
44			"""Graphite sender.
45
46			:param log: A logger
47			:type log: logging.RootLogger
48			:param config: a plumd.config.Conf configuration helper instance.
49			:type config: plumd.config.Conf
50			"""
51			super(Graphite, self).__init__(log, config)
52			config.defaults(Graphite.defaults)
53			self.addr = (config.get('host'), config.get('port'))
54			self.socket = None
55			self.evt_stop = threading.Event()
56
57			def onstop(self):
58			"""Set stop event to force any writes to fall through and exit."""
59			self.evt_stop.set()
60			msg = "Graphite: onstop: {0}: setting stop event"
61			self.log.info(msg.format(self.config.get('name')))
62
63			def connect(self):
64			"""Connect to graphite and maintain a connection, return success of operation.
65
66			:rtype: bool
67			"""
68			timeout = self.config.get('timeout')
69			tcp_ka_idle = self.config.get('tcp_ka_idle')
70			tcp_ka_intvl = self.config.get('tcp_ka_intvl')
71			tcp_ka_count = self.config.get('tcp_ka_count')
72			# disconnect if we're already connected
73			if self.socket:
74			self.disconnect()
75			try:
76			# create the socket + connection w/ timeout
77			self.socket = socket.create_connection(self.addr, timeout=timeout)
78			# enable tcp keepalive with more aggressive values than default
79			self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
80			self.socket.setsockopt(socket.SOL_TCP, socket.TCP_KEEPIDLE, tcp_ka_idle)
81			self.socket.setsockopt(socket.SOL_TCP, socket.TCP_KEEPINTVL, tcp_ka_intvl)
82			self.socket.setsockopt(socket.SOL_TCP, socket.TCP_KEEPCNT, tcp_ka_count)
83			# set timeout for socket operations
84			self.socket.settimeout(timeout)
85			# log and return
86			msg = "Graphite: connected: {0}"
87			self.log.info(msg.format(self.addr))
88			return True
89			except Exception as e:
			0 ignored issues – show Best Practice introduced 2017-03-01 08:02 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
90			# log exception, ensure cleanup is done (disconnect)
91			msg = "Graphite: {0}: connect: exception: {1}"
92			self.log.error(msg.format(self.addr, e))
93			if self.socket:
94			try:
95			self.socket.close()
96			except Exception as exc:
			0 ignored issues – show Best Practice introduced 2017-03-01 08:02 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history... Unused Code introduced 2017-03-01 08:02 UTC by Report Bug Copy Issue Report The variable `exc` seems to be unused. Loading history...
97			self.socket = None
98			self.socket = None
99			return False
100
101			def disconnect(self):
102			"""Close the tcp connection to the graphite server."""
103			if self.socket:
104			try:
105			self.socket.close()
106			self.socket = None
107			except Exception as e:
			0 ignored issues – show Best Practice introduced 2017-03-01 08:02 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
108			msg = "Graphite: disconnect: exception {0}".format(e)
109			self.log.error(msg)
110			self.socket = None
111
112			def write(self, metrics):
113			"""Send the already rendered metrics string to graphite with retry.
114
115			:param metrics: the metrics string to send.
116			:type metrics: str
117			:rtype: bool
118			"""
119			# connect to graphite
120			while (not self.evt_stop.is_set()) and (not self.socket) and (not self.connect()):
121			# reconnect on an interval
122			with Interval(self.config.get('retry_time'), self.evt_stop) as wait:
123			msg = "re-connecting to graphite host {0}:{1} in {2:.1f} s"
124			self.log.info(msg.format(self.addr[0], self.addr[1], wait.remaining))
125			# "this should never happen"
126			if (not self.evt_stop.is_set()) and (not self.socket):
127			msg = "graphite: socket not connected after connect, dropping metrics"
128			self.log.error(msg)
129			return
130			# send metrics with retry
131			while not self.evt_stop.is_set():
132			try:
133			if PY3:
134			self.socket.sendall(bytes(metrics, 'utf8'))
135			else:
136			self.socket.sendall(metrics)
137			break
138			except Exception as e:
			0 ignored issues – show Best Practice introduced 2017-03-01 08:02 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
139			msg = "Graphite: {0}:{1}: write: exception: {2}"
140			self.log.error(msg.format(self.addr[0], self.addr[1], e))
141			self.connect()
142
143			def flush(self, metrics):
144			"""Flush the metrics string to graphite, return immediately if not connected.
145
146			This is used on shutdown.
147
148			:param metrics: the metrics string to send.
149			:type metrics: str
150			:rtype: bool
151			"""
152			nbytes = len(metrics)
153			msg = "Graphite: {0}:{1} flushing {2} bytes"
154			self.log.debug(msg.format(self.addr[0], self.addr[1], nbytes))
155			if not self.socket:
156			msg = "Graphite: {0}:{1} not connected, dropping metrics"
157			self.log.debug(msg.format(self.addr[0], self.addr[1]))
158			return
159			try:
160			if PY3:
161			self.socket.sendall(bytes(metrics, 'utf8'))
162			else:
163			self.socket.sendall(metrics)
164			msg = "Graphite: {0}:{1} {2} bytes flushed"
165			self.log.debug(msg.format(self.addr[0], self.addr[1], nbytes))
166			except Exception as e:
			0 ignored issues – show Best Practice introduced 2017-03-01 08:02 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
167			msg = "Graphite: flush: {0}:{1}: write: exception: dropping metrics: {2}"
168			self.log.error(msg.format(self.addr[0], self.addr[1], e))
169
170

s4z / plumd

Branch — prom (d3fc9e)

GraphiteWriteTCP.write() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like