Merge pull request #100 from directionless/seph/limit

Limit metric length going into ganglia

.gitignore

@@ -3,3 +3,4 @@ build
 dist
 *.egg-info
 *.log
+venv

debian/changelog

@@ -1,3 +1,9 @@
+python-statsd (1.1) testing; urgency=low
+
+  *  Fix to truncate long metrics names before sending to ganglia
+
+ -- seph <seph@directionless.org>  Fri, 15 Sep 2017 20:18:57 +0000
+
 python-statsd (1.0-4) testing; urgency=low
 
   * update to 36a59d3b126ded4658aff25bce94e844a1c6413e
@@ -9,7 +15,7 @@ python-statsd (1.0-2) UNRELEASED; urgency=low
 
   [ Rob Terhaar ]
   * add requirement module for python-argparse
-  * added upstart script to /usr/share/docs/python-statsd/ 
+  * added upstart script to /usr/share/docs/python-statsd/
 
   [ Gábor Farkas ]
   * do not override the threshold

debian/control

@@ -5,9 +5,9 @@ Priority: optional
 Standards-Version: 3.8.3
 Build-Depends: debhelper (>= 7.3), python-support (>= 1.0.3), python, python-setuptools
 XS-Python-Version: >= 2.6
-Vcs-Git: https://github.com/robbyt/py-statsd
-Vcs-Browser: https://github.com/robbyt/py-statsd
-Homepage: https://github.com/robbyt/py-statsd
+Vcs-Git: https://github.com/sivy/py-statsd
+Vcs-Browser: https://github.com/sivy/py-statsd
+Homepage: https://github.com/sivy/py-statsd
 
 Package: python-statsd
 Architecture: all

pystatsd/server.py

@@ -4,19 +4,24 @@
 import time
 import types
 import logging
-from . import gmetric
 from subprocess import call
 from warnings import warn
 # from xdrlib import Packer, Unpacker
 
-log = logging.getLogger(__name__)
-
 try:
     from setproctitle import setproctitle
 except ImportError:
     setproctitle = None
 
-from .daemon import Daemon
+# Messily get the import for things we're distributing. This is in a
+# try block, since we seem to need different syntax based on some set
+# of python version and whetehr or not we're in a library.
+try:
+    from . import gmetric
+    from .daemon import Daemon
+except ValueError:
+    import gmetric
+    from daemon import Daemon
 
 
 __all__ = ['Server']
@@ -33,8 +38,6 @@ def _clean_key(k):
         )
     )
 
-
-
 TIMER_MSG = '''%(prefix)s.%(key)s.lower %(min)s %(ts)s
 %(prefix)s.%(key)s.count %(count)s %(ts)s
 %(prefix)s.%(key)s.mean %(mean)s %(ts)s
@@ -47,7 +50,7 @@ class Server(object):
 
     def __init__(self, pct_threshold=90, debug=False, transport='graphite',
                  ganglia_host='localhost', ganglia_port=8649,
-                 ganglia_spoof_host='statsd:statsd',
+                 ganglia_spoof_host='statsd:statsd', ganglia_max_length=100,
                  gmetric_exec='/usr/bin/gmetric', gmetric_options = '-d',
                  graphite_host='localhost', graphite_port=2003, global_prefix=None, 
                  flush_interval=10000,
@@ -64,6 +67,8 @@ def __init__(self, pct_threshold=90, debug=False, transport='graphite',
         # Use gmetric
         self.gmetric_exec = gmetric_exec
         self.gmetric_options = gmetric_options
+        # Common Ganglia
+        self.ganglia_max_length = ganglia_max_length
         # Set DMAX to flush interval plus 20%. That should avoid metrics to prematurely expire if there is
         # some type of a delay when flushing
         self.dmax = int(self.flush_interval * 1.2)
@@ -88,7 +93,10 @@ def __init__(self, pct_threshold=90, debug=False, transport='graphite',
         self.flusher = 0
 
     def send_to_ganglia_using_gmetric(self,k,v,group, units):
-        call([self.gmetric_exec, self.gmetric_options, "-u", units, "-g", group, "-t", "double", "-n",  k, "-v", str(v) ])
+        if len(k) >= self.ganglia_max_length:
+            log.debug("Ganglia metric too long. Ignoring: %s" % k)
+        else:
+            call([self.gmetric_exec, self.gmetric_options, "-u", units, "-g", group, "-t", "double", "-n",  k, "-v", str(v) ])
 
 
     def process(self, data):
@@ -174,7 +182,10 @@ def flush(self):
             elif self.transport == 'ganglia':
                 # We put counters in _counters group. Underscore is to make sure counters show up
                 # first in the GUI. Change below if you disagree
-                g.send(k, v, "double", "count", "both", 60, self.dmax, "_counters", self.ganglia_spoof_host)
+                if len(k) >= self.ganglia_max_length:
+                    log.debug("Ganglia metric too long. Ignoring: %s" % k)
+                else:
+                    g.send(k, v, "double", "count", "both", 60, self.dmax, "_counters", self.ganglia_spoof_host)
             elif self.transport == 'ganglia-gmetric':
                 self.send_to_ganglia_using_gmetric(k,v, "_counters", "count")
 
@@ -198,7 +209,10 @@ def flush(self):
                 msg = '%s.%s %s %s\n' % (self.counters_prefix, k, v, ts)
                 stat_string += msg
             elif self.transport == 'ganglia':
-                g.send(k, v, "double", "count", "both", 60, self.dmax, "_gauges", self.ganglia_spoof_host)
+                if len(k) >= self.ganglia_max_length:
+                    log.debug("Ganglia metric too long. Ignoring: %s" % k)
+                else:
+                    g.send(k, v, "double", "count", "both", 60, self.dmax, "_gauges", self.ganglia_spoof_host)
             elif self.transport == 'ganglia-gmetric':
                 self.send_to_ganglia_using_gmetric(k,v, "_gauges", "gauge")
 
@@ -250,12 +264,15 @@ def flush(self):
                     # We are gonna convert all times into seconds, then let rrdtool add proper SI unit. This avoids things like
                     # 3521 k ms which is 3.521 seconds
                     # What group should these metrics be in. For the time being we'll set it to the name of the key
-                    group = k
-                    g.send(k + "_min", min / 1000, "double", "seconds", "both", 60, self.dmax, group, self.ganglia_spoof_host)
-                    g.send(k + "_mean", mean / 1000, "double", "seconds", "both", 60, self.dmax, group, self.ganglia_spoof_host)
-                    g.send(k + "_max", max / 1000, "double", "seconds", "both", 60, self.dmax, group, self.ganglia_spoof_host)
-                    g.send(k + "_count", count, "double", "count", "both", 60, self.dmax, group, self.ganglia_spoof_host)
-                    g.send(k + "_" + str(self.pct_threshold) + "pct", max_threshold / 1000, "double", "seconds", "both", 60, self.dmax, group, self.ganglia_spoof_host)
+                    if len(k) >= self.ganglia_max_length:
+                        log.debug("Ganglia metric too long. Ignoring: %s" % k)
+                    else:
+                        group = k
+                        g.send(k + "_min", min / 1000, "double", "seconds", "both", 60, self.dmax, group, self.ganglia_spoof_host)
+                        g.send(k + "_mean", mean / 1000, "double", "seconds", "both", 60, self.dmax, group, self.ganglia_spoof_host)
+                        g.send(k + "_max", max / 1000, "double", "seconds", "both", 60, self.dmax, group, self.ganglia_spoof_host)
+                        g.send(k + "_count", count, "double", "count", "both", 60, self.dmax, group, self.ganglia_spoof_host)
+                        g.send(k + "_" + str(self.pct_threshold) + "pct", max_threshold / 1000, "double", "seconds", "both", 60, self.dmax, group, self.ganglia_spoof_host)
                 elif self.transport == 'ganglia-gmetric':
                     # We are gonna convert all times into seconds, then let rrdtool add proper SI unit. This avoids things like
                     # 3521 k ms which is 3.521 seconds
@@ -364,6 +381,8 @@ def run_server():
     # Use gmetric
     parser.add_argument('--ganglia-gmetric-exec', dest='gmetric_exec', help='Use gmetric executable. Defaults to /usr/bin/gmetric', type=str, default="/usr/bin/gmetric")
     parser.add_argument('--ganglia-gmetric-options', dest='gmetric_options', help='Options to pass to gmetric. Defaults to -d 60', type=str, default="-d 60")
+    # Common for ganglia
+    parser.add_argument('--ganglia-max-length', dest='ganglia_max_length', help='Maximum length of metric names for ganglia. Defaults to 100 characters', type=str, default=100)
     # 
     parser.add_argument('--flush-interval', dest='flush_interval', help='how often to send data to graphite in millis (default: 10000)', type=int, default=10000)
     parser.add_argument('--no-aggregate-counters', dest='no_aggregate_counters', help='should statsd report counters as absolute instead of count/sec', action='store_true')
@@ -381,6 +400,7 @@ def run_server():
     log_level = logging.DEBUG if options.debug else logging.INFO
     logging.basicConfig(level=log_level,format='%(asctime)s [%(levelname)s] %(message)s')
 
+    log.info("Starting up on %s" % options.port)
     daemon = ServerDaemon(options.pidfile)
     if options.daemonize:
         daemon.start(options)
@@ -391,5 +411,8 @@ def run_server():
     else:
         daemon.run(options)
 
+
+log = logging.getLogger(__name__)
+
 if __name__ == '__main__':
     run_server()