Merge pull request #748 from alq666/statsd-logger

gunicorn instrumentation via statsD and a new logger class

README.rst

@@ -143,6 +143,35 @@ For example:
 It is all here. No configuration files nor additional python modules to
 write !!
 
+Instrumentation
+---------------
+
+Gunicorn provides an optional instrumentation of the arbiter and
+workers using the statsD_ protocol over UDP. Thanks to the 
+`gunicorn.instrument.statsd` module, Gunicorn becomes a statsD client
+The use of UDP cleanly isolates Gunicorn from the receiving end of the statsD
+metrics so that instrumentation does not cause Gunicorn to be heldeup by a slow
+statsD consumer.
+
+To use statsD, just tell gunicorn where the statsD server is:
+
+    $ gunicorn --statsd-host=localhost:8125 ...
+
+The `Statsd` logger overrides `gunicorn.glogging.Logger` to track
+all requests. The following metrics are generated:
+
+  * ``gunicorn.requests``: request rate per second
+  * ``gunicorn.request.duration``: histogram of request duration
+  * ``gunicorn.workers``: number of workers managed by the arbiter (gauge)
+  * ``gunicorn.log.critical``: rate of critical log messages
+  * ``gunicorn.log.error``: rate of error log messages
+  * ``gunicorn.log.warning``: rate of warning log messages
+  * ``gunicorn.log.exception``: rate of exceptional log messages
+
+To generate new metrics you can `log.info` with a few additional keywords::
+
+    log.info("...", extra={"metric": "my.metric", "value": "1.2", "mtype": "gauge"})
+
 LICENSE
 -------
 
@@ -161,4 +190,5 @@ details.
 .. _`production page`: http://docs.gunicorn.org/en/latest/deploy.html
 .. _`config file`: http://docs.gunicorn.org/en/latest/configure.html
 .. _setproctitle: http://pypi.python.org/pypi/setproctitle/
+.. _statsD: http://github.com/etsy/statsd
 .. _LICENSE: http://github.com/benoitc/gunicorn/blob/master/LICENSE

gunicorn/arbiter.py

@@ -482,6 +482,11 @@ class Arbiter(object):
             (pid, _) = workers.pop(0)
             self.kill_worker(pid, signal.SIGTERM)
 
+        self.log.info("{0} workers".format(len(workers)),
+                      extra={"metric": "gunicorn.workers",
+                          "value": len(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

@@ -143,7 +143,12 @@ class Config(object):
             # support the default
             uri = "gunicorn.glogging.Logger"
         
-        logger_class = util.load_class(uri,
+        # if statsd is on, automagically switch to the statsd logger
+        if 'statsd_host' in self.settings:
+            logger_class = util.load_class("gunicorn.instrument.statsd.Statsd",
+                section="gunicorn.loggers")
+        else:
+            logger_class = util.load_class(uri,
                 default="gunicorn.glogging.Logger",
                 section="gunicorn.loggers")
 
@@ -436,6 +441,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()),
@@ -1625,3 +1639,15 @@ if sys.version_info >= (2, 7):
         desc = """\
         Ciphers to use (see stdlib ssl module's)
         """
+
+# statsD monitoring
+class StatsdHost(Setting):
+    name = "statsd_host"
+    section = "Logging"
+    cli = ["--statsd-host"]
+    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,117 @@
+# -*- 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"
+GAUGE_TYPE = "gauge"
+COUNTER_TYPE = "counter"
+HISTOGRAM_TYPE = "histogram"
+
+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_host
+            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)
+
+    # Special treatement for info, the most common log level
+    def info(self, msg, *args, **kwargs):
+        """Log a given statistic if metric, value and type are present
+        """
+        Logger.info(self, msg, *args, **kwargs)
+        extra = kwargs.get("extra", None)
+        if extra is not None:
+            metric = extra.get(METRIC_VAR, None)
+            value = extra.get(VALUE_VAR, None)
+            typ = extra.get(MTYPE_VAR, None)
+            if metric and value and typ:
+                if typ == GAUGE_TYPE:
+                    self.gauge(metric, value)
+                elif typ == COUNTER_TYPE:
+                    self.increment(metric, value)
+                elif typ == HISTOGRAM_TYPE:
+                    self.histogram(metric, value)
+                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
+        request_time is a datetime.timedelta
+        """
+        Logger.access(self, resp, req, environ, request_time)
+        duration_in_s = request_time.seconds + float(request_time.microseconds)/10**6
+        self.histogram("gunicorn.request.duration", duration_in_s)
+        self.increment("gunicorn.requests", 1)
+        self.increment("gunicorn.request.status.%d" % int(resp.status.split()[0]), 1)
+
+    # statsD methods
+    # you can use those directly if you want
+    def gauge(self, name, value):
+        try:
+            if self.sock:
+                self.sock.send("{0}:{1}|g".format(name, value))
+        except Exception:
+            pass
+
+    def increment(self, name, value, sampling_rate=1.0):
+        try:
+            if self.sock:
+                self.sock.send("{0}:{1}|c|@{2}".format(name, value, sampling_rate))
+        except Exception:
+            pass
+
+    def decrement(self, name, value, sampling_rate=1.0):
+        try:
+            if self.sock:
+                self.sock.send("{0}:-{1}|c|@{2}".format(name, value, sampling_rate))
+        except Exception:
+            pass
+
+    def histogram(self, name, value):
+        try:
+            if self.sock:
+                self.sock.send("{0}:{1}|ms".format(name, value))
+        except Exception:
+            pass

tests/test_010-statsd.py

@@ -0,0 +1,54 @@
+from datetime import timedelta
+
+import t
+
+from gunicorn.config import Config
+from gunicorn.instrument.statsd import Statsd
+
+class TestException(Exception): pass
+
+class MockSocket(object):
+    "Pretend to be a UDP socket"
+    def __init__(self, failp):
+        self.failp = failp
+        self.msgs = []  # accumulate messages for later inspection
+
+    def send(self, msg):
+        if self.failp:
+            raise TestException("Should not interrupt the logger")
+        self.msgs.append(msg)
+
+    def reset(self):
+        self.msgs = []
+
+class MockResponse(object):
+    def __init__(self, status):
+        self.status = status
+
+def test_statsd_fail():
+    "UDP socket fails"
+    logger = Statsd(Config())
+    logger.sock = MockSocket(True)
+    logger.info("No impact on logging")
+    logger.debug("No impact on logging")
+    logger.critical("No impact on logging")
+    logger.error("No impact on logging")
+    logger.warning("No impact on logging")
+    logger.exception("No impact on logging")
+
+def test_instrument():
+    logger = Statsd(Config())
+    logger.sock = MockSocket(False)
+
+    logger.info("Blah", extra={"mtype": "gauge", "metric": "gunicorn.test", "value": 666})
+    t.eq(logger.sock.msgs[0], "gunicorn.test:666|g")
+    logger.sock.reset()
+
+    logger.critical("Boom")
+    t.eq(logger.sock.msgs[0], "gunicorn.log.critical:1|c|@1.0")
+    logger.sock.reset()
+
+    logger.access(MockResponse("200 OK"), None, {}, timedelta(seconds=7))
+    t.eq(logger.sock.msgs[0], "gunicorn.request.duration:7.0|ms")
+    t.eq(logger.sock.msgs[1], "gunicorn.requests:1|c|@1.0")
+    t.eq(logger.sock.msgs[2], "gunicorn.request.status.200:1|c|@1.0")