Fluentd check for Datadog (https://app.datadoghq.com)

fluentd.py

# stdlib
import re
import urllib2
import urlparse

# project
from util import headers
from checks import AgentCheck
from checks.utils import add_basic_auth

# 3rd party
import simplejson as json

class Fluentd(AgentCheck):
    """Tracks basic fluentd metrics via the monitor_agent plugin
    * number of retry_count
    * number of buffer_queue_length
    * number of buffer_total_queued_size

    $ curl http://localhost:24220/api/plugins.json
    {"plugins":[{"type": "monitor_agent", ...}, {"type": "forward", ...}]}

    """
    def check(self, instance):
        if 'monitor_agent_url' not in instance:
            raise Exception('Fluentd instance missing "monitor_agent_url" value.')
        tags = instance.get('tags', [])

        response = self._get_data(instance)
        metrics = self._parse_json(response, tags)

        funcs = {
            'gauge': self.gauge,
            'histogram': self.histogram,
        }
        for row in metrics:
            try:
                name, value, tags, metric_type = row
                func = funcs[metric_type]
                func(name, value, tags)
            except Exception:
                self.log.error(u'Could not submit metric: %s' % repr(row))

    def _get_data(self, instance):
        url = instance.get('monitor_agent_url')
        req = urllib2.Request(url, None, headers(self.agentConfig))
        parsed_url = urlparse.urlparse(url)
        try:
            response = urllib2.urlopen(req)
        except Exception:
            raise
        return response.read()

    def _parse_json(self, raw, tags=None):
        if tags is None:
            tags = []
        output = []

        try:
            parsed = json.loads(raw)
        except Exception:
            self.log.error(u'Could not parse json: %s' % repr(row))
            output.append(['fluentd.alive', 0, tags, 'gauge'])
        else:
            output.append(['fluentd.alive', 1, tags, 'gauge'])

        for plg in parsed['plugins']:
            for metric in ('retry_count', 'buffer_total_queued_size', 'buffer_queue_length'):
                if metric in plg:
                    output.append(['fluentd.monitor_agent.%s' % (metric), plg[metric], tags, 'histogram'])

        return output

fluentd.yaml

init_config:

instances:
    - monitor_agent_url: 'http://127.0.0.1:24220/api/plugins.json'