plumd.plugins.writers.InfluxdbRender

Complexity

Total Complexity

25

Size/Duplication

Total Lines	102
Duplicated Lines	0 %

2 Methods

Rating	Name	Duplication	Size	Complexity
F	process()	0	62	18
B	__init__()	0	18	7

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

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

import copy
import socket
import logging
import httplib

import requests # pip install requests

import plumd
import plumd.plugins
import plumd

# from: http://stackoverflow.com/questions/11029717/how-do-i-disable-log-messages-from-the-requests-library
# http requests were generating log entries.. were.
httplib.HTTPConnection.debuglevel = 0
rlog = logging.getLogger("requests.packages.urllib3")
rlog.addHandler(logging.NullHandler())
rlog.setLevel(logging.ERROR)
rlog.propagate = True


class InfluxdbRender(plumd.Render):
    """Renders metrics in an influxdb writable format.

    As of this writing the latest influxdb line protocol documentation is at:
    https://docs.influxdata.com/influxdb/v0.10/write_protocols/line/

    Plumd renders metrics like:

    cpu,host=some,meta=values idle=90.0,user=5.0,system=5.0 <ts>\n
    disk,host=some,meta=values,dev=/dev/sda free=95.0,used=5.0 <ts>\n

    where host and metric metadata map to measurement tags and the metric
    values for a single result get comma seperated as field values.

    Each result in a result set generates a new metric line and each metric
    line is queued in a deque for sending.

    :param rconfig: A Conf object initialized with the plugin configuration.
    :type rconfig: plumd.conf.Conf
    """
    def __init__(self, rconfig):
        super(InfluxdbRender, self).__init__(rconfig)

        # use a lambda for escaping
        escmd = lambda t: t.replace(" ", "\ ").replace(",", "\,").replace("\n", "\\n")

        # generate host metadata string once on init
        meta_dat = []
        for hmkey, hmval in self.meta.items:
            hmkey = escmd(hmkey)
            if isinstance(hmval, str):
                hmval = escmd(hmval)
            elif isinstance(hmval, int):
                hmval = "{0}i".format(hmval)
            elif isinstance(hmval, bool):
                hmval = "t" if hmval else "f"
            meta_dat.append("{0}={1}".format(hmkey, hmval))
        self.meta_str = ",".join(meta_dat)


    def process(self, results):
        """Process the metrics into the influxdb line protocol (as of v0.9).

        :param metrics: a generator that returns metrics in a standard format
        :type metrics: generator
        """
        # use lambdas for string fixups in metadata/fields
        escmd = lambda t: t.replace(" ", "\ ").replace(",", "\,").replace("\n", "\\n")
        escmf = lambda f: '""{0}""'.format(f).replace(" ", "\ ").replace(",", "\,").replace("\n", "\\n")

        # render each metric in the influxdb line format, append to self.metrics
        # this is a long function as I don't want to call outside the loop
        # ( time, result_name, result_meta, [metric, metric, metric] )
        for (ts, rname, rmeta, metrics) in results:
            # generate a metrics string eg. idle=0.0,user=90.0,sys=10.0
            met_str = ""
            for metric in metrics:
                mclass = metric.__class__
                key_str = escmd(metric[0])
                val_str = None
                # value => int:append i, bool: t or f, string dbl qut
                if mclass == plumd.Int or mclass == plumd.Counter:
                    val_str = "{0}i".format(metric[1])
                elif mclass == plumd.Boolean:
                    val_str = "t" if metric[1] else "f"
                elif mclass == plumd.String:
                    val_str = escmf(metric[1])
                else:
                    val_str = metric[1]
                if met_str:
                    met_str += ",{0}={1}".format(key_str, val_str)
                else:
                    met_str = "{0}={1}".format(key_str, val_str)

            # next add any metadata from the result
            rmdat = []
            # combine host and metric metadata
            for r_mkey, r_mval in rmeta.items:
                r_mkey = escmd(r_mkey)
                if isinstance(r_mval, str):
                    r_mval = escmd(r_mval)
                elif isinstance(r_mval, int):
                    r_mval = "{0}i".format(r_mval)
                elif isinstance(r_mval, bool):
                    r_mval = "t" if r_mval else "f"
                rmdat.append("{0}={1}".format(r_mkey, r_mval))
            rmstr = ",".join(rmdat)

            # finally create the combined metadata string
            meta_str = rname
            if self.meta_str:
                meta_str += "," + self.meta_str
            if rmstr:
                meta_str += "," + rmstr

            # now combine it all - note influx uses nanosecond timestamps
            # see http://crsmithdev.com/arrow/, X is timestamp and SSS is ms
            ts_fmt = "XSSS000000"
            mstr = "{0} {1} {2}".format(meta_str, met_str, ts.format(ts_fmt))
            # add the rendered metric string to our deque of metrics
            # this allows for "\n".join() when sending
            self.metrics.append(mstr)


class Influxdb(plumd.plugins.Writer):
    """Influxdb sender."""
    defaults = {
        'username': 'admin',
        'password': 'admin',
        'database': 'metrics',
        'host':     '127.0.0.1',
        'port':     8086,
        'protocol': "http", # http or udp : todo: udp
        'batch':    64,     # send batches of this size
        'retries':  3,      # retry failed requests
        'timeout':  5,      # timeout http posts in 5 seconds
        'maxqueue': 8192,   # maximum number of metrics to queue
        'warnqueue': 1024   # print warning if queue size > this
    }

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

        :param log: a structlog logger instance.
        :type log: structlog logger
        :param config: a plumd.config.Conf configuration helper instance.
        :type config: plumd.config.Conf
        """
        super(Influxdb, self).__init__(log, config)
        self.config = config
        proto = config.get('protocol')
        self.batch = config.get('batch')
        self.warn = config.get('warnqueue')
        self.socket = None
        if proto.lower() == "http":
            self.session = requests.Session()
            self.session.auth = (config.get('username'), config.get('password'))
            content_type = {'Content-type': 'application/octet-stream'}
            self.session.headers.update(content_type)
            self.send = self.send_http
            args = [ config.get('host'), config.get('port'),
                     config.get('database') ]
            self.url = "http://{0}:{1}/write?db={2}".format(*args)
            self.retries = config.get('retries')
            self.timeout = config.get('timeout')
            self.log.debug("__init__", timeout=self.timeout, url=self.url)
        elif proto.lower() == "udp":
            # udp has no auth or db (configured on receiver side)
            self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self.addr = (config.get('host'), config.get('port'))
            self.send = self.send_udp
            self.log.debug("__init__", timeout=self.timeout, address=self.addr)
        else:
            raise plumd.ConfigError("unknown protocol: {0}".format(self.proto))
        self.render = InfluxdbRender(self.config)


    def send_udp(self, metrics):
        """Send the batch of metrics to influxdb via UDP protocol.

        :param metrics: The list of pre-formated metrics from our renderer.
        :type metrics: list
        """
        mstr = "\n".join(metrics) + "\n"
        str_len = len(mstr)
        try:
            self.socket.sendto(mstr, self.addr)
        except Exception as e:
            self.log.debug("send_udp", exception=e)
            return


    def send_http(self, metrics):
        """Send the batch of metrics to influxb.

        :param metrics: The list of pre-formated metrics from our renderer.
        :type metrics: list
        """
        rdat = "\n".join(metrics) + "\n"
        url = self.url
        nretries = self.retries
        timeout = self.timeout
        while nretries > 0:
            nretries -= 1
            resp = None
            try:
                resp = self.session.post(url, data=rdat, timeout=timeout)
            except Exception as e:
                msg = "exception during post"
                rtxt = "None"
                if resp is not None and hasattr(resp, "text"):
                    rtxt = resp.text
                self.log.error("send_http", exception=e, msg=msg, response=rtxt)
                continue
            # requests module ok codes doesn't include 204...
            if resp.status_code  >= 200 and resp.status_code < 300:
                return
            else:
                self.log.debug("send_http", action="retrying",
                               response=resp.status_code, text=resp.text)
        self.log.debug("send_http", action="failed", reason="too many retries")


    def onstop(self):

This code seems to be duplicated in your project.

        """Flush any remaining metrics."""
        # get the next full batch of metrics to send
        self.log.debug("onstop", action="flushing")
        qsize, nmetrics, bmetrics = self.render.next_batch(self.batch, True)
        if qsize > self.warn:
            self.log.warn("onstop", qsize=qsize, warn=self.warn)
        while nmetrics > 0:
            self.log.debug("onstop", action="sending", metrics=bmetrics)
            self.send(bmetrics)
            # and the next batch until there are no more full batches
            qsize, nmetrics, bmetrics = self.render.next_batch(self.batch, True)
        self.log.debug("onstop", action="flushed")


    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
        self.render.process(rset.results)

        # get the next full batch of metrics to send
        qsize, nmetrics, bmetrics = self.render.next_batch(self.batch, False)
        if qsize > self.warn:
            self.log.warn("push", qsize=qsize)
        while nmetrics > 0:
            self.send(bmetrics)
            # and the next batch until there are no more full batches
            qsize, nmetrics, bmetrics = self.render.next_batch(self.batch,
                                                               False)