Merge pull request #80 from cheribral/master

Add the ability to use InfluxDB 0.9

graphios.cfg

@@ -111,11 +111,26 @@ librato_whitelist = [".*"]
 #nerf_librato = False
 
 #------------------------------------------------------------------------------
-# InfluxDB Details (comment in if you are using InfluxDB)
+# InfluxDB Details (if you are using InfluxDB 0.8)
 #------------------------------------------------------------------------------
 
 enable_influxdb = False
 
+#------------------------------------------------------------------------------
+# InfluxDB Details (if you are using InfluxDB 0.9)
+# This will work a bit differently because of the addition of tags in 
+# InfluxDB 0.9.  Now the metric will be named after the service description,
+# the perfdata field will be a tag, and the host name will be a tag.  The value
+# of the metric is stored in the 'value' column. 
+# Requires use_service_desc = True.
+#------------------------------------------------------------------------------
+
+enable_influxdb09 = False
+
+# Extra tags to add to metrics, like data center location etc.
+# Only valid for 0.9
+#influxdb_extra_tags = {"location": "la"}
+
 # Comma separated list of server:ports
 # defaults to 127.0.0.1:8086 (:8087 if using SSL).
 #influxdb_servers = 127.0.0.1:8087
@@ -136,6 +151,7 @@ enable_influxdb = False
 # Flag the InfluxDB backend as 'non essential' for the purposes of error checking
 #nerf_influxdb = False
 
+
 #------------------------------------------------------------------------------
 # STDOUT Details (comment in if you are using STDOUT)
 #------------------------------------------------------------------------------

graphios.py

@@ -487,6 +487,7 @@ def init_backends():
                       "statsd",
                       "librato",
                       "influxdb",
+                      "influxdb09",
                       "stdout",
                       )
     # populate the controller dict from avail + config. this assumes you named

graphios_backends.py

@@ -10,6 +10,7 @@
 import urllib2
 import json
 import os
+import ast
 # ###########################################################
 # #### Librato Backend
 
@@ -541,6 +542,30 @@ def chunks(self, l, n):
         for i in xrange(0, len(l), n):
             yield l[i:i+n]
 
+    def _send(self, server, chunk):
+        self.log.debug("Connecting to InfluxDB at %s" % server)
+        json_body = json.dumps(chunk)
+        req = urllib2.Request(self.build_url(server), json_body)
+        req.add_header('Content-Type', 'application/json')
+
+        try:
+            r = urllib2.urlopen(req, timeout=self.timeout)
+            r.close()
+            return True
+        except urllib2.HTTPError as e:
+            body = e.read()
+            self.log.warning('Failed to send metrics to InfluxDB. \
+                                Status code: %d: %s' % (e.code, body))
+            return False
+        except IOError as e:
+            fail_string = "Failed to send metrics to InfluxDB. "
+            if hasattr(e, 'code'):
+                fail_string = fail_string + "Status code: %s" % e.code
+            if hasattr(e, 'reason'):
+                fail_string = fail_string + str(e.reason)
+            self.log.warning(fail_string)
+            return False
+
     def send(self, metrics):
         """ Connect to influxdb and send metrics """
         ret = 0
@@ -575,27 +600,71 @@ def send(self, metrics):
         series_chunks = self.chunks(series, self.influxdb_max_metrics)
         for chunk in series_chunks:
             for s in self.influxdb_servers:
-                self.log.debug("Connecting to InfluxDB at %s" % s)
-                json_body = json.dumps(chunk)
-                req = urllib2.Request(self.build_url(s), json_body)
-                req.add_header('Content-Type', 'application/json')
+                if not self._send(s, chunk):
+                    ret = 0
+
+        return ret
+
+
+# ###########################################################
+# #### influxdb-0.9 backend  ####################################
+
+class influxdb09(influxdb):
+    def __init__(self, cfg):
+        influxdb.__init__(self, cfg)
+        if 'influxdb_extra_tags' in cfg:
+            self.influxdb_extra_tags = ast.literal_eval(
+                cfg['influxdb_extra_tags'])
+            print self.influxdb_extra_tags
+        else:
+            self.influxdb_extra_tags = {}
+
+    def build_url(self, server):
+        """ Returns a url to specified InfluxDB-server """
+        test_port = server.split(':')
+        if len(test_port) < 2:
+            server = "%s:%i" % (server, self.default_ports[self.scheme])
+
+        return "%s://%s/write?u=%s&p=%s" % (self.scheme, server,
+                                            self.influxdb_user,
+                                            self.influxdb_password)
+
+    def send(self, metrics):
+        """ Connect to influxdb and send metrics """
+        ret = 0
+        perfdata = []
+        for m in metrics:
+            ret += 1
 
+            if (m.SERVICEDESC == ''):
+                path = m.HOSTCHECKCOMMAND
+            else:
+                path = m.SERVICEDESC
+
+            # Ensure a int/float gets passed
+            try:
+                value = int(m.VALUE)
+            except ValueError:
                 try:
-                    r = urllib2.urlopen(req, timeout=self.timeout)
-                    r.close()
-                except urllib2.HTTPError as e:
-                    ret = 0
-                    body = e.read()
-                    self.log.warning('Failed to send metrics to InfluxDB. \
-                                        Status code: %d: %s' % (e.code, body))
-                except IOError as e:
+                    value = float(m.VALUE)
+                except ValueError:
+                    value = 0
+
+            tags = {"check": m.LABEL, "host": m.HOSTNAME}
+            tags.update(self.influxdb_extra_tags)
+
+            perfdata.append({
+                            "timestamp": int(m.TIMET),
+                            "name": path,
+                            "tags": tags,
+                            "fields": {"value": value}})
+
+        series_chunks = self.chunks(perfdata, self.influxdb_max_metrics)
+        for chunk in series_chunks:
+            series = {"database": self.influxdb_db, "points": chunk}
+            for s in self.influxdb_servers:
+                if not self._send(s, series):
                     ret = 0
-                    fail_string = "Failed to send metrics to InfluxDB. "
-                    if hasattr(e, 'code'):
-                        fail_string = fail_string + "Status code: %s" % e.code
-                    if hasattr(e, 'reason'):
-                        fail_string = fail_string + str(e.reason)
-                    self.log.warning(fail_string)
 
         return ret