First cut of logger-based statsd

gunicorn/arbiter.py

@@ -480,6 +480,11 @@ class Arbiter(object):
             (pid, _) = workers.pop(0)
             self.kill_worker(pid, signal.SIGQUIT)
 
+        self.log.info("%d workers",
+                      metric="gunicorn.workers",
+                      value=len(self.WORKERS),
+                      mtype="gauge")
+
     def spawn_worker(self):
         self.worker_age += 1
         worker = self.worker_class(self.worker_age, self.pid, self.LISTENERS,

gunicorn/config.py

@@ -421,6 +421,15 @@ def validate_file(val):
 
     return path
 
+def validate_hostport(val):
+    val = validate_string(val)
+    if val is None:
+        return None
+    elements = val.split(":")
+    if len(elements) == 2:
+        return (elements[0], int(elements[1]))
+    else:
+        raise TypeError("Value must consist of: hostname:port")
 
 def get_default_config_file():
     config_path = os.path.join(os.path.abspath(os.getcwd()),
@@ -1554,3 +1563,15 @@ if sys.version_info >= (2, 7):
         desc = """\
         Ciphers to use (see stdlib ssl module's)
         """
+
+# statsD monitoring
+class StatsdTo(Setting):
+    name = "statsd_to"
+    section = "Logging"
+    cli = ["--log-statsd-to"]
+    meta = "STATSD_ADDR"
+    default = "localhost:8125"
+    validator = validate_hostport
+    desc ="""\
+    host:port of the statsd server to log to
+    """

gunicorn/instrument/statsd.py

@@ -0,0 +1,107 @@
+# -*- coding: utf-8 -
+#
+# This file is part of gunicorn released under the MIT license.
+# See the NOTICE for more information.
+
+"""Bare-bones implementation of statsD's protocol, client-side
+"""
+import socket
+from gunicorn.glogging import Logger
+
+# Instrumentation constants
+STATSD_DEFAULT_PORT = 8125
+METRIC_VAR = "metric"
+VALUE_VAR = "value"
+MTYPE_VAR = "mtype"
+SAMPLING_VAR = "sampling"
+
+class Statsd(Logger):
+    """statsD-based instrumentation, that passes as a logger
+    """
+    def __init__(self, cfg):
+        """host, port: statsD server
+        """
+        Logger.__init__(self, cfg)
+        try:
+            host, port = cfg.statsd_to
+            self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            self.sock.connect((host, int(port)))
+        except Exception:
+            self.sock = None
+
+    # Log errors and warnings
+    def critical(self, msg, *args, **kwargs):
+        Logger.critical(self, msg, *args, **kwargs)
+        self.increment("gunicorn.log.critical", 1)
+
+    def error(self, msg, *args, **kwargs):
+        Logger.error(self, msg, *args, **kwargs)
+        self.increment("gunicorn.log.error", 1)
+
+    def warning(self, msg, *args, **kwargs):
+        Logger.warning(self, msg, *args, **kwargs)
+        self.increment("gunicorn.log.warning", 1)
+
+    def exception(self, msg, *args, **kwargs):
+        Logger.exception(self, msg, *args, **kwargs)
+        self.increment("gunicorn.log.exception", 1)
+
+    def info(self, msg, *args, **kwargs):
+        """Log a given statistic if metric, value, sampling are present
+        """
+        Logger.info(self, msg, *args, **kwargs)
+        metric = kwargs.get(METRIC_VAR, None)
+        value = kwargs.get(VALUE_VAR, None)
+        typ = kwargs.get(MTYPE_VAR, None)
+        if metric and value and typ:
+            if typ == "gauge":
+                self.gauge(metric, value)
+            elif typ == "counter":
+                sampling = kwargs.get(SAMPLING_VAR, 1.0)
+                self.increment(metric, value, sampling)
+            else:
+                pass
+
+    # skip the run-of-the-mill logs
+    def debug(self, msg, *args, **kwargs):
+        Logger.debug(self, msg, *args, **kwargs)
+
+    def log(self, lvl, msg, *args, **kwargs):
+        Logger.log(self, lvl, msg, *args, **kwargs)
+
+    # access logging
+    def access(self, resp, req, environ, request_time):
+        """Measure request duration
+        """
+        Logger.access(self, resp, req, environ, request_time)
+        self.histogram("gunicorn.request.duration", request_time)
+        self.increment("gunicorn.requests", 1)
+
+    # statsD methods
+    def gauge(self, name, value):
+        try:
+            if self.sock:
+                self.sock.send("%s:%s|g\n" % (name, value))
+        except Exception:
+            pass
+
+    def increment(self, name, value, sampling_rate=1.0):
+        try:
+            if self.sock:
+                self.sock.send("%s:%s|c@%s\n" % (name, value, sampling_rate))
+        except Exception:
+            pass
+
+    def decrement(self, name, value, sampling_rate=1.0):
+        try:
+            if self.sock:
+                self.sock.send("%s:-%s|c@%s\n" % (name, value, sampling_rate))
+        except Exception:
+            pass
+
+    def histogram(self, name, value, sampling_rate=1.0):
+        try:
+            if self.sock:
+                self.sock.send("%s:%s|h@%s\n" % (name, value, sampling_rate))
+        except Exception:
+            pass