GraphiteWriteUDP

Complexity

Total Complexity

5

Size/Duplication

Total Lines	32
Duplicated Lines	0 %

Importance

Changes	3
Bugs	0	Features	1

3 Methods

Rating	Name	Duplication	Size	Complexity
A	__init__()	0	5	1
A	write_blocking()	0	11	2
A	write()	0	11	2

# -*- coding: utf-8 -*-

There seems to be a cyclic import (plumd -> plumd.util).

"""Built-in Plumd Graphite Writer supporting plain text tcp/udp."""
import sys
import time
import socket

import plumd

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


PY3 = sys.version_info > (3,)


def format_meta(meta, sep, meta_key, filter_chars):
    """Return a formatted metadata string from the Meta object.

    :param meta: :class:`plumd.Meta` object that describes a metrics metadata.
    :type meta: plumd.Meta
    :param sep: seperator character for metadata values (eg. =).
    :type sep: str
    :param meta_key: flag to include meta data key values or not
    :type meta_key: bool
    :param filter_chars: string value consisting of the characters to filter
    :type filter_chars` : str
    :rtype: str
    """
    # generate metadata string
    mmeta_dat = []
    for mkey, mval in meta.items:
        if PY3:
            mkey = mkey.translate(dict.fromkeys(map(ord, filter_chars), None))
        else:
            mkey = mkey.translate(None, filter_chars)
        # mval is one of Int, Float, String, etc metric types
        # and so mval[1] is the actual value where mval[0] is the name of it
        mval = mval[1]
        if isinstance(mval, str):
            if PY3:
                mval = mval.translate(dict.fromkeys(
                    map(ord, filter_chars), None))
            else:
                mval = mval.translate(None, filter_chars)
        elif isinstance(mval, bool):
            mval = "true" if mval else "false"
        # include the metadata key in the metric? eg. .region_uswest.
        if meta_key:
            mmeta_dat.append("{0}{1}{2}".format(mkey, sep, mval))
        # put the metadata value in the metric only eg. .uswest.
        else:
            mmeta_dat.append(mval)
    return ".".join(mmeta_dat)


class GraphiteRenderPlain(plumd.Render):
    """Renders metrics in graphite plain text format:

    <prefix>.<hmeta=hval>.<measurement>.<metric>.<mmeta=mval>.value <value> <ts>

    where host and metric metadata keys are sorted before being added to the
    metric name and the seperator character is configurable. eg.

    servers.env=test.region=test.cpu.idle.value 0.0 1452240380
    servers.env=test.region=test.disk.free.dev=sda1.value 1213234232 1452240380

    or with a . seperator character:

    servers.env.test.region.test.cpu.idle.value 0.0 1452240380
    servers.env.test.region.test.disk.free.dev.sda1.value 1213234232 1452240380

    :param rconfig: A Conf object initialized with the plugin configuration.
    :type rconfig: plumd.conf.Conf
    :raises: plumd.ConfigError
    """

    def __init__(self, rconfig):
        super(GraphiteRenderPlain, self).__init__(rconfig)
        # get the configured values from the plugin configuration
        self.prefix = rconfig.get("prefix")
        # record rendered metrics here, when bytes > max_bytes put into deque
        self.buff = list()
        self.buff_bytes = 0
        self.seperator = rconfig.get("seperator")
        self.meta_key = rconfig.get("meta_key")
        # need to build a list of characters to remove from metric/meta strings
        # this method is significantly faster/easier on the cpu than re.sub
        # build filter_chars with eg. chars = string.letters + string.digits +
        # "_-"
        keep_chars = rconfig.get("filter_chars")
        self.filter_chars = "".join(char for char in map(chr, range(256))
                                    if char not in keep_chars)
        if self.seperator and isinstance(self.seperator, str):
            self.seperator = self.seperator[0]

        # generate the host meta data string - assumes static host metadata
        self.hmeta_str = format_meta(self.meta, self.seperator, self.meta_key,
                                     self.filter_chars)

    def flush(self):
        """Return and clear the partial metric buffer, also clear queue."""
        metrics = self.buff
        self.buff = []
        self.buff_bytes = 0
        self.metrics.clear()
        return "".join(metrics)

    def process(self, results):
        """Process the metrics in the result set into the graphite plain text
        format.

        :param results: a generator that returns metrics in a standard format
        :type results: generator
        """
        hmeta_str = self.hmeta_str          # host metadata string, if any
        prefix = self.prefix                # metric prefix string
        meta_key = self.meta_key            # bool: include metadata in metric?
        sep = self.seperator                # meta data seperator character
        filter_chars = self.filter_chars    # remove these chars from strings

        # render each metric in the graphite plaintext format
        for (ts, rname, rmeta, metrics) in results:
            # get the metadata for this metric, if any
            mmeta_dat = []
            for mkey, mval in rmeta.items:
                mkey = mkey.translate(None, filter_chars)
                mval = mval[1]  # mval[0] is the name, mval[1] is the value
                if isinstance(mval, str):
                    mval = mval.translate(None, filter_chars)
                elif isinstance(mval, bool):
                    mval = "true" if mval else "false"
                # include the metadata key in the metric? eg. .region_uswest.
                if meta_key:
                    mmeta_dat.append("{0}{1}{2}".format(mkey, sep, mval))
                # put the metadata value in the metric only eg. .uswest.
                else:
                    mmeta_dat.append(mval)

            # generate a metric string for each metric
            for metric in metrics:
                mclass = metric.__class__
                # silently ignore strings
                if mclass == plumd.String:
                    continue
                # booleans are either really big or 0
                mval = metric[1]
                if mclass == plumd.Boolean:
                    mval = sys.maxint if mval else 0

The Module sys does not seem to have a member named maxint.

                # get the full metric string
                if hmeta_str:
                    args = [prefix, hmeta_str, rname, metric[0]]
                else:
                    args = [prefix, rname, metric[0]]
                # now combine the values
                if mmeta_dat:
                    fmstr = "{0}.{1} {2:.2f} {3}\n".format(".".join(args),
                                                           ".".join(mmeta_dat),
                                                           int(mval), int(ts))
                else:
                    fmstr = "{0} {1:.2f} {2}\n".format(".".join(args), mval,
                                                       int(ts))
                fmstr_len = len(fmstr)

                # can we append to the existing buffer?
                if len(self.buff) + fmstr_len < self.max_bytes:
                    self.buff.append(fmstr)
                    self.buff_bytes += fmstr_len
                    continue

                # or do we need to queue and start a new buffer?
                self.metrics.append(self.buff)
                self.buff = list(fmstr)
                self.buff_bytes = fmstr_len


class Graphite(plumd.Writer):
    """Graphite sender."""
    # 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,
        'protocol':         "tcp",
        'format':           "plain",  # plain or pickle
        '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':         8192,    # maximum number of metrics to queue
        'warnqueue':        1024,    # 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.render = None
        fmt = config.get('format')
        fmts = {
            'plain': GraphiteRenderPlain
        }
        if fmt in fmts:
            self.render = fmts[fmt](self.config)
        else:
            raise plumd.ConfigError("unkown format: {0}".format(fmt))
        proto = config.get('protocol')
        self.writer = None
        writers = {
            'tcp': GraphiteWriteTCP,
            'udp': GraphiteWriteUDP
        }
        if proto in writers:
            self.writer = writers[proto](self.log, config)
        else:
            raise plumd.ConfigError("unkown protocol: {0}".format(proto))

    def onstop(self):
        """Flush remaining metrics and disconnect socket on shutdown."""
        msg = "Graphite: onstop: {0}: flushing metrics"
        self.log.info(msg.format(self.config.get('name')))
        # if we're not connected then drop all metrics and return
        if not self.writer.socket:
            msg = "Graphite: onstop: not connected, dropping all queued metrics"
            self.log.error(msg)
            self.render.flush()
            return

        # ensure we don't take too long trying to flush metrics on shutdown
        start = time.time()
        for mbuff in self.render:
            self.writer.write("".join(mbuff))
            if time.time() - start > 1.0:
                self.render.flush()
                msg = "Graphite: onstop: {0}: unable to send metrics, dropping"
                self.log.error(msg.format(self.config.get('name')))
                self.writer.disconnect()
                return
        mbuff = self.render.flush()
        if mbuff:
            self.writer.write("".join(mbuff))
        self.writer.disconnect()
        msg = "Graphite: onstop: {0}: done"
        self.log.info(msg.format(self.config.get('name')))

    def push(self, rset):
        """Render metrics onto our deque and send all full batches.

        :param metrics: The list of pre-formated metrics from our renderer.
        :type metrics: list
        """
        # record this batch of metrics
        if rset:
            self.render.process(rset.results)

        # get the next full batch of metrics to send - next_batch pop_left's
        resend = None
        for mbuff in self.render:
            metrics = self.writer.write("".join(mbuff))
            if metrics:
                resend = metrics
                # do not continue if we're unable to send
                # this can cause significant delays on shutdown
                break

        # add metrics back onto queue if any
        if resend:
            self.log.debug("Graphite: write: requeuing metrics")
            self.render.metrics.append(resend)


class GraphiteWriteTCP(object):
    """Graphite TCP writer."""

    def __init__(self, log, config):
        """Graphite TCP writer."""
        self.log = log
        self.addr = (config.get('host'), config.get('port'))
        self.socket = None
        self.timeout = config.get('timeout')
        self.retry_time = config.get('retry_time')
        self.tcp_ka_idle = config.get('tcp_ka_idle')
        self.tcp_ka_intvl = config.get('tcp_ka_intvl')
        self.tcp_ka_count = config.get('tcp_ka_count')

    def connect(self):
        """Connect to graphite and maintain a connection. Returns True if
        the connection was made or False if it failed.

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

Catching very general exceptions such as Exception is usually not recommended.

            # log exception, ensure cleanup is done (disconnect)
            msg = "Graphite: {0}: connect: excception: {1}"
            self.log.error(msg.format(self.addr, e))
            # pause before reconnecting
            time.sleep(self.retry_time)
            self.disconnect()
        return False

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

Catching very general exceptions such as Exception is usually not recommended.

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

    def write(self, metrics):
        """Send the metrics string to graphite.

        :param metrics: the metrics string to send.
        :type metrics: str
        :rtype: bool
        """
        if not self.socket:
            self.connect()
        if not self.socket:
            msg = "Graphite: write: connection failed"
            self.log.error(msg)
            return metrics  # resend them
        try:
            if PY3:
                self.socket.sendall(bytes(metrics, 'utf8'))
            else:
                self.socket.sendall(metrics)
        except Exception as e:

Catching very general exceptions such as Exception is usually not recommended.

            msg = "Graphite: {0}: write: exception: {1}"
            self.log.error(msg.format(self.addr, e))
            self.connect()
            return metrics  # resend them


class GraphiteWriteUDP(object):
    """Graphite UDP writer."""

    def __init__(self, log, config):
        """Graphite UDP writer."""
        self.log = log
        self.addr = (config.get('host'), config.get('port'))
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    def write_blocking(self, metrics):
        """Send the metrics to graphite and return number of bytes sent.

        :param metrics: the metrics string to send.
        :type metrics: str
        :rtype: int
        """
        if PY3:
            return self.socket.sendto(bytes(metrics, 'utf8'), self.addr)
        else:
            return self.socket.sendto(metrics, self.addr)

    def write(self, metrics):
        """Send the metrics to graphite and return number of bytes sent.

        :param metrics: the metrics string to send.
        :type metrics: str
        :rtype: int
        """
        if PY3:
            return self.socket.sendto(bytes(metrics, 'utf8'), self.addr)
        else:
            return self.socket.sendto(metrics, self.addr)