plumd.plugins.writers.Elastic

Complexity

Total Complexity

23

Size/Duplication

Total Lines	127
Duplicated Lines	0 %

4 Methods

Rating	Name	Duplication	Size	Complexity
F	send()	0	50	14
A	onstop()	0	12	3
B	__init__()	0	27	3
A	push()	0	17	3

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

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

import itertools
import logging
import httplib
import json

import requests # pip install requests

import plumd
import plumd.plugins

# 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 ElasticRender(plumd.Render):
    """Renders metrics in an elasticsearch writable format which is simply
    a dictionary with the required bulk api parameters and metric details.

    NOTE: this class is considered experimental and is not fully functional.

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

    def __init__(self, rconfig):
        super(ElasticRender, self).__init__(rconfig)
        # get the configured values from the plugin configuration
        self.index = rconfig.get("index")
        self.type = rconfig.get("type")


    def process(self, results):
        """Process the results into a format suitable for bulk api calls.

        Results are in the format:

        ( time, result_name, result_meta, [metric, metric, metric] )

        :param metrics: a generator that returns metrics in a standard format
        :type metrics: generator
        """
        field_map = { # this is here to save a lookup in the loop below
            plumd.Int:        ('int', 'int_val'),
            plumd.Counter:    ('counter', 'int_val'),
            plumd.Float:      ('float', 'float_val'),
            plumd.Gauge:      ('gauge', 'float_val'),
            plumd.Rate:       ('rate', 'float_val'),
            plumd.Timer:      ('timer', 'float_val'),
            plumd.String:     ('string', 'str_val'),
            plumd.Boolean:    ('bool', 'bool_val')
        }
        for (ts, rname, rmeta, metrics) in results:
            rmetric = {
                'measurement': mname,
                'metric': name,
                'time': ts.isoformat()
            }
            if mclass not in field_map:
                rmetric['unk_val'] = val
                rmetric['metric_type'] = "unknown"
            else:
                metric_type, metric_key = field_map[mclass]
                rmetric[metric_key] = val
                rmetric["metric_type"] = metric_type
            # todo: how to handle this better? (key collisions this way)
            # fold host meta data into metric document
            for key, val in self.host_meta.items():
                if key not in rmetric:
                    rmetric[key] = val
            # fold metric meta data into metric document
            for key, val in meta.items():
                if key not in rmetric:
                    rmetric[key] = val
            # es bulk api wants newline separated json strings
            index = { "index": {"_index" : self.index } }
            action_str = "{0}\n{1}\n".format(json.dumps(index),
                                             json.dumps(rmetric))
            self.metrics.append(action_str)


class Elastic(plumd.plugins.Writer):
    """Elasticsearch sender."""
    defaults = {
        'proto':    'http',
        'hosts':    ['127.0.0.1:9200'],     # default list of hosts
        'timeout':  30,                     # todo: timeouts in es calls
        'index':    'metrics',              # name of es index for metrics
        'type':     'metric',               # type of document to create
        'batch':    64,                     # number of metrics per bulk req
        'retries':  3,                      # retry failed requests
        'timeout':  5,                      # timeout http post requests
        'maxqueue': 8192,                   # maximum number of metrics to queue
        'warnqueue': 256                    # print a warning if the queue > this
    }

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

        NOTE: this class is considered experimental and is not fully functional.

        :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(Elastic, self).__init__(log, config)
        self.host_meta = config.get('meta')    # host level metadata
        self.hosts = config.get('hosts')
        self.index = config.get('index')
        self.dtype = self.config.get('type')
        self.batch = self.config.get('batch')
        self.warn = self.config.get('warnqueue')
        url = "{0}://{1}/{2}/{3}/_bulk/" # the documents have the index in them
        base = self.config.get('proto')
        hosts = self.config.get('hosts')
        urls = [ url.format(base, h, self.index, self.dtype) for h in hosts ]
        sessions = [ requests.Session() for host in hosts ]
        self.sessions = itertools.cycle(zip(urls, sessions))
        self.retries = config.get('retries')
        self.timeout = config.get('timeout')
        self.render = ElasticRender(self.config)
        self.log.debug("__init__", timeout=self.timeout)


    def send(self, metrics):
        """Render metrics onto our deque and send all full batches using
        the bulk api.

        :param metrics: The list of pre-formated metrics from our renderer.
        :type metrics: list
        """
        rdat = "\n".join(metrics) # es bulk api wants \n separated json strings
        self.log.debug("send", index=self.index, action="posting")
        # cycle through each url, retry up to self.retries count
        nretries = self.retries
        timeout = self.timeout
        for url, sess in self.sessions:
            nretries -= 1
            if nretries < 0:
                break
            (resp, took, errs, items, failed) = (None, "unknown", True, [], [])
            try:
                self.log.debug("send", request=url, data=rdat, timeout=timeout)
                resp = sess.post(url, data=rdat)
            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", exception=e, msg=msg, response=rtxt)
                continue
            if resp.status_code  == requests.codes.ok:
                try:
                    robj = resp.json()
                    if 'took' in robj:
                        took = robj['took']
                    if 'errors' in robj:
                        errs = robj['errors']
                    if errs and 'items' in robj:
                        for item in robj['items']:
                            failed.append(item)
                except Exception as e:
                    msg = "error parsing response"
                    self.log.debug("send", exception=e, msg=msg)
                self.log.debug("send", action="SUCCESS", took=took,
                               errors=errs, failed=failed)
                break
            else:
                self.log.debug("send", action="retrying",
                               response=resp.status_code)
        if nretries > 0:
            self.log.debug("send", action="complete")
        else:
            self.log.debug("send", action="failed", reason="too many retries")


    def onstop(self):

This code seems to be duplicated in your project.

        """Flush any remaining metrics."""
        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("push", 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):
        """Send metrics to elasticsearch."""
        self.log.debug("push", action="processing", index=self.index,
                       dtype=self.dtype)
        # 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.log.debug("push", 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, False)
        self.log.debug("push", action="processed")