Merge lp:~mk-fraggod/txstatsd/txstatsd into lp:txstatsd

[Mike Kazantsev]

Adds --dump-mode option with passing data further to graphite (as per suggestion on the IRC) and derive_plugin for instant-rate meters, tests included.

I've squashed commits per-feature, original history can be found in the github fork (github.com/mk-fg/txstatsd, and upstream_merge branch there).

[Lucio Torre (lucio.torre)]

Theres already a plugin using the "d" kay, that counts distinct values exaclty using redis for real time reports and postgres for long term storage. maybe we could change this metric to "dx" or something like that?

see https://code.launchpad.net/~txstatsd-dev/txstatsd/distinct-plugin

Is there any reason to do this and not use graphite to calculate the derivative?

[Mike Kazantsev (mk-fraggod)]

Will update to use "dx", sure, just failed to notice the conflict.

About graphite-side derivative calculation - I assume you question the need for "rate" value with "count" already present.

1. I find it harder (from usability standpoint) to make graphite calculate derivative (than just pre-calculate one) every time the value is needed. Especially true for wildcard graphs, dashboards and any custom data interface.

2. "derivative()" in graphite is incapable to handle overflow/restart correctly, producing huge spike whenever that happens, totally screwing up any graph, while "rate" sent from statsd does not suffer from that issue at all.

3. "count" offset (say, when the absolute level was known-baseline) can be calculated from rate by summing all the rate values since then, and the damage from overflow/restart with this approach is minimal, compared to "reset to a totally arbitrary value".

4. "count" value does not make any sense in nearly all cases - it's a completely arbitrary value offset from some point in time. All the real-world use-cases I have for such metric type are for "rate" only.

Initially I've left "count" there just for consistency with "meter" metric, which can be used if you need just "count", I guess, but at this point I think it was a mistake.

So what do you think about leaving only the "rate" value there?

It would definitely be better for my purposes - minus one nesting level, lesser load and disk usage.

If you believe it'd make sense to have only ".count" in mainline, existing "meter" metric should probably be used instead, and the plugin isn't needed there at all, I guess.

=== added file '.bzrignore'
--- .bzrignore	1970-01-01 00:00:00 +0000
+++ .bzrignore	2012-05-14 16:57:19 +0000
@@ -0,0 +1,2 @@
+_trial_temp*
+twisted/plugins/dropin.cache

=== renamed file '.bzrignore' => '.bzrignore.moved'
=== added file 'LICENSE'
--- LICENSE	1970-01-01 00:00:00 +0000
+++ LICENSE	2012-05-14 16:57:19 +0000
@@ -0,0 +1,20 @@
+Copyright (C) 2011 Canonical Services Ltd
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

=== renamed file 'LICENSE' => 'LICENSE.moved'
=== added file 'MANIFEST.in'
--- MANIFEST.in	1970-01-01 00:00:00 +0000
+++ MANIFEST.in	2012-05-14 16:57:19 +0000
@@ -0,0 +1,5 @@
+include README
+include LICENSE
+include setup.py
+recursive-include txstatsd *.py
+graft twisted

=== renamed file 'MANIFEST.in' => 'MANIFEST.in.moved'
=== added file 'README'
--- README	1970-01-01 00:00:00 +0000
+++ README	2012-05-14 16:57:19 +0000
@@ -0,0 +1,21 @@
+Dependencies
+------------
+
+* Python
+
+* The twisted python package (python-twisted on Debian or similar systems)
+
+Things present here
+-------------------
+
+* The txstatsd python package, containing a server and a client implementation
+  for the statsd protocol.
+
+* The metrics support borrows from Coda Hale's Metrics project
+  https://github.com/codahale/metrics.
+
+License
+-------
+
+txStatsD is open source software, MIT License. See the LICENSE file for more
+details.

=== renamed file 'README' => 'README.moved'
=== added file 'example-stats-client.tac'
--- example-stats-client.tac	1970-01-01 00:00:00 +0000
+++ example-stats-client.tac	2012-05-14 16:57:19 +0000
@@ -0,0 +1,52 @@
+import socket
+import random
+
+from twisted.internet import reactor
+from twisted.internet import task
+
+from twisted.application.service import Application
+
+from txstatsd.client import (
+    TwistedStatsDClient, StatsDClientProtocol)
+from txstatsd.metrics.metrics import Metrics
+from txstatsd.process import PROCESS_STATS
+from txstatsd.report import ReportingService
+
+
+STATSD_HOST = "127.0.0.1"
+STATSD_PORT = 8125
+
+application = Application("example-stats-client")
+statsd_client = TwistedStatsDClient(STATSD_HOST, STATSD_PORT)
+metrics = Metrics(connection=statsd_client,
+                  namespace=socket.gethostname() + ".example-client")
+
+reporting = ReportingService()
+reporting.setServiceParent(application)
+
+for report in PROCESS_STATS:
+    reporting.schedule(report, 10, metrics.increment)
+
+def random_walker(name):
+    """Meters a random walk."""
+    if random.random() > 0.5:
+        metrics.increment(name)
+    else:
+        metrics.decrement(name)
+
+def random_normal(name):
+    """Meters samples from a normal distribution."""
+    metrics.timing(name, random.normalvariate(10, 3))
+
+
+for n in range(5):
+    t = task.LoopingCall(random_walker, name="walker%i" % n)
+    t.start(0.5, now=False)
+
+for n in range(5):
+    t = task.LoopingCall(random_normal, name="normal%i" % n)
+    t.start(0.5, now=False)
+
+
+protocol = StatsDClientProtocol(statsd_client)
+reactor.listenUDP(0, protocol)

=== renamed file 'example-stats-client.tac' => 'example-stats-client.tac.moved'
=== added file 'requirements.txt'
--- requirements.txt	1970-01-01 00:00:00 +0000
+++ requirements.txt	2012-05-14 16:57:19 +0000
@@ -0,0 +1,7 @@
+Twisted==11.0.0
+mocker==1.1
+numpy==1.6.1
+psutil==0.4.0
+scipy==0.9.0
+wsgiref==0.1.2
+zope.interface==3.8.0

=== renamed file 'requirements.txt' => 'requirements.txt.moved'
=== added file 'setup.py'
--- setup.py	1970-01-01 00:00:00 +0000
+++ setup.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,67 @@
+from distutils.core import setup
+from distutils.command.install import install
+from glob import glob
+import os
+
+from twisted.plugin import IPlugin, getPlugins
+
+from txstatsd import version
+
+# If setuptools is present, use it to find_packages(), and also
+# declare our dependency on epsilon.
+extra_setup_args = {}
+try:
+    import setuptools
+    from setuptools import find_packages
+except ImportError:
+    def find_packages():
+        """
+        Compatibility wrapper.
+
+        Taken from storm setup.py.
+        """
+        packages = []
+        for directory, subdirectories, files in os.walk("txstatsd"):
+            if '__init__.py' in files:
+                packages.append(directory.replace(os.sep, '.'))
+        return packages
+
+long_description = """
+Twisted-based implementation of a statsd-compatible server and client.
+"""
+
+
+class TxPluginInstaller(install):
+    def run(self):
+        install.run(self)
+        # Make sure we refresh the plugin list when installing, so we know
+        # we have enough write permissions.
+        # see http://twistedmatrix.com/documents/current/core/howto/plugin.html
+        # "when installing or removing software which provides Twisted plugins,
+        # the site administrator should be sure the cache is regenerated"
+        list(getPlugins(IPlugin))
+
+setup(
+    cmdclass = {'install': TxPluginInstaller},
+    name="txStatsD",
+    version=version.txstatsd,
+    description="A network daemon for aggregating statistics",
+    author="txStatsD Developers",
+    url="https://launchpad.net/txstatsd",
+    license="MIT",
+    packages=find_packages() + ["twisted.plugins"],
+    scripts=glob("./bin/*"),
+    long_description=long_description,
+    classifiers=[
+        "Development Status :: 4 - Beta",
+        "Intended Audience :: Developers",
+        "Intended Audience :: System Administrators",
+        "Intended Audience :: Information Technology",
+        "Programming Language :: Python",
+        "Topic :: Database",
+        "Topic :: Internet :: WWW/HTTP",
+        "License :: OSI Approved :: MIT License",
+       ],
+    **extra_setup_args
+    )
+

=== renamed file 'setup.py' => 'setup.py.moved'
=== added file 'statsd.tac'
--- statsd.tac	1970-01-01 00:00:00 +0000
+++ statsd.tac	2012-05-14 16:57:19 +0000
@@ -0,0 +1,8 @@
+from twisted.application.service import Application
+
+from txstatsd import service
+
+application = Application("statsd")
+
+statsd_service = service.createService(service.StatsDOptions())
+statsd_service.setServiceParent(application)

=== renamed file 'statsd.tac' => 'statsd.tac.moved'
=== added directory 'twisted'
=== renamed directory 'twisted' => 'twisted.moved'
=== added directory 'twisted/plugins'
=== added file 'twisted/plugins/derive_plugin.py'
--- twisted/plugins/derive_plugin.py	1970-01-01 00:00:00 +0000
+++ twisted/plugins/derive_plugin.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,69 @@
+from zope.interface import implements
+
+from twisted.plugin import IPlugin
+from txstatsd.itxstatsd import IMetric, IMetricFactory
+from txstatsd.metrics.metric import Metric
+
+import logging, time
+
+
+class DeriveMetricReporter(object):
+    """
+    A simplier meter metric which measures instant throughput rate for each interval.
+    """
+    implements(IMetric)
+
+    def __init__(self, name, wall_time_func=time.time, prefix=""):
+        """Construct a metric we expect to be periodically updated.
+
+        @param name: Indicates what is being instrumented.
+        @param wall_time_func: Function for obtaining wall time.
+        """
+        self.name = name
+        self.wall_time_func = wall_time_func
+        if prefix: prefix += "."
+        self.prefix = prefix
+        self.update = self.count = 0
+        self.poll_time = self.wall_time_func()
+
+    def process(self, fields):
+        """
+        Process new data for this metric.
+
+        @type fields: C{list}
+        @param fields: The list of message parts. Usually in the form of
+        (value, metric_type, [sample_rate])
+        """
+        val, k = fields
+        self.update += float(val)
+
+    def flush(self, interval, timestamp):
+        """
+        Returns a string with new line separated list of metrics to report.
+
+        @type interval: C{float}
+        @param interval: The time since last flush.
+        @type timestamp: C{float}
+        @param timestamp: The timestamp for now.
+        """
+        poll_prev, self.poll_time = self.poll_time, self.wall_time_func()
+        if self.poll_time == poll_prev: return list()
+        rate = float(self.update) / (self.poll_time - poll_prev)
+        self.count, self.update = self.count + self.update, 0
+        return list(
+            (self.prefix + self.name + "." + item, round(value, 6), timestamp)
+            for item, value in [("count", self.count), ("rate", rate)] )
+
+
+class DeriveMetricFactory(object):
+    implements(IMetricFactory, IPlugin)
+
+    name = "derive"
+    metric_type = "dx"
+
+    def build_metric(self, prefix, name, wall_time_func=None):
+        return DeriveMetricReporter(name, prefix=prefix, wall_time_func=wall_time_func)
+
+    def configure(self, options): pass
+
+derive_metric_factory = DeriveMetricFactory()

=== added file 'twisted/plugins/distinct_plugin.py'
--- twisted/plugins/distinct_plugin.py	1970-01-01 00:00:00 +0000
+++ twisted/plugins/distinct_plugin.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,20 @@
+from zope.interface import implements
+
+from twisted.plugin import IPlugin
+from txstatsd.itxstatsd import IMetricFactory
+from txstatsd.metrics.distinctmetric import DistinctMetricReporter
+
+class DistinctMetricFactory(object):
+    implements(IMetricFactory, IPlugin)
+
+    name = "pdistinct"
+    metric_type = "pd"
+
+    def build_metric(self, prefix, name, wall_time_func=None):
+        return DistinctMetricReporter(name, prefix=prefix,
+                                      wall_time_func=wall_time_func)
+
+    def configure(self, options):
+        pass
+
+distinct_metric_factory = DistinctMetricFactory()

=== added file 'twisted/plugins/txstatsd_plugin.py'
--- twisted/plugins/txstatsd_plugin.py	1970-01-01 00:00:00 +0000
+++ twisted/plugins/txstatsd_plugin.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,22 @@
+from zope.interface import implements
+
+from twisted.plugin import IPlugin
+from twisted.application.service import IServiceMaker
+
+from txstatsd import service
+
+
+class StatsDServiceMaker(object):
+    implements(IServiceMaker, IPlugin)
+    tapname = "statsd"
+    description = "Collect and aggregate stats for graphite."
+    options = service.StatsDOptions
+
+    def makeService(self, options):
+        """
+        Construct a StatsD service.
+        """
+        return service.createService(options)
+
+# Now construct an object which *provides* the relevant interfaces
+serviceMaker = StatsDServiceMaker()

=== added directory 'txstatsd'
=== added file 'txstatsd.conf-example'
--- txstatsd.conf-example	1970-01-01 00:00:00 +0000
+++ txstatsd.conf-example	2012-05-14 16:57:19 +0000
@@ -0,0 +1,26 @@
+[statsd]
+# The host where carbon cache is listening.
+carbon-cache-host: localhost
+# The port where carbon cache is listening.
+carbon-cache-port: 2003
+# The UDP port where we will listen.
+listen-port: 8125
+
+# The number of milliseconds between each flush.
+flush-interval: 60000
+
+# Which additional stats to report {process|net|io|system}.
+report:
+# Prefix to use when reporting stats.
+prefix:
+# Produce StatsD-compliant messages.
+statsd-compliance: 1
+
+# Support application monitoring. UDP echo is initially supported.
+# Should we receive the monitor-message, we respond with the
+# configured monitor-response.
+monitor-message: txstatsd ping
+monitor-response: txstatsd pong
+
+[plugin_sample]
+sample-key: sample-value
\ No newline at end of file

=== renamed file 'txstatsd.conf-example' => 'txstatsd.conf-example.moved'
=== renamed directory 'txstatsd' => 'txstatsd.moved'
=== added file 'txstatsd/__init__.py'
=== added file 'txstatsd/client.py'
--- txstatsd/client.py	1970-01-01 00:00:00 +0000
+++ txstatsd/client.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,165 @@
+import socket
+
+from twisted.internet.defer import inlineCallbacks, returnValue
+from twisted.internet.protocol import DatagramProtocol
+from twisted.python import log
+
+
+class StatsDClientProtocol(DatagramProtocol):
+    """A Twisted-based implementation of the StatsD client protocol.
+
+    Data is sent via UDP to a StatsD server for aggregation.
+    """
+
+    def __init__(self, client):
+        self.client = client
+
+    def startProtocol(self):
+        """Connect to destination host."""
+        self.client.connect(self.transport)
+
+    def stopProtocol(self):
+        """Connection was lost."""
+        self.client.disconnect()
+
+
+class TwistedStatsDClient(object):
+
+    def __init__(self, host, port,
+                 connect_callback=None,
+                 disconnect_callback=None,
+                 resolver_errback=None):
+        """
+        Build a connection that reports to the endpoint (on C{host} and
+        C{port}) using UDP.
+
+        @param host: The StatsD server host.
+        @param port: The StatsD server port.
+        @param resolver_errback: The errback to invoke should
+            issues occur resolving the supplied C{host}.
+        @param connect_callback: The callback to invoke on connection.
+        @param disconnect_callback: The callback to invoke on disconnection.
+        """
+        from twisted.internet import reactor
+
+        self.reactor = reactor
+
+        @inlineCallbacks
+        def resolve(host):
+            self.host = yield reactor.resolve(host)
+            returnValue(self.host)
+
+        self.host = None
+        self.resolver = resolve(host)
+        if resolver_errback is None:
+            self.resolver.addErrback(log.err)
+        else:
+            self.resolver.addErrback(resolver_errback)
+
+        self.port = port
+        self.connect_callback = connect_callback
+        self.disconnect_callback = disconnect_callback
+
+        self.transport = None
+
+    @inlineCallbacks
+    def connect(self, transport=None):
+        """Connect to the StatsD server."""
+        host = yield self.resolver
+        if host is not None:
+            self.transport = transport
+            if self.transport is not None:
+                if self.connect_callback is not None:
+                    self.connect_callback()
+
+    def disconnect(self):
+        """Disconnect from the StatsD server."""
+        if self.disconnect_callback is not None:
+            self.disconnect_callback()
+        self.transport = None
+
+    def write(self, data, callback=None):
+        """Send the metric to the StatsD server.
+
+        @param data: The data to be sent.
+        @param callback: The callback to which the result should be sent.
+            B{Note}: The C{callback} will be called in the C{reactor}
+            thread, and not in the thread of the original caller.
+        """
+        self.reactor.callFromThread(self._write, data, callback)
+
+    def _write(self, data, callback):
+        """Send the metric to the StatsD server.
+
+        @param data: The data to be sent.
+        @param callback: The callback to which the result should be sent.
+        @raise twisted.internet.error.MessageLengthError: If the size of data
+            is too large.
+        """
+        if self.host is not None and self.transport is not None:
+            try:
+                bytes_sent = self.transport.write(data, (self.host, self.port))
+                if callback is not None:
+                    callback(bytes_sent)
+            except (OverflowError, TypeError, socket.error, socket.gaierror):
+                if callback is not None:
+                    callback(None)
+
+
+class UdpStatsDClient(object):
+
+    def __init__(self, host=None, port=None):
+        """Build a connection that reports to C{host} and C{port})
+        using UDP.
+
+        @param host: The StatsD host.
+        @param port: The StatsD port.
+        @raise ValueError: If the C{host} and C{port} cannot be
+            resolved (for the case where they are not C{None}).
+        """
+        self.host = host
+        self.port = port
+
+        if host is not None and port is not None:
+            try:
+                self.host, self.port = socket.getaddrinfo(
+                    host, port, socket.AF_INET,
+                    socket.SOCK_DGRAM, socket.SOL_UDP)[0][4]
+            except (TypeError, IndexError, socket.error, socket.gaierror):
+                raise ValueError("The address cannot be resolved.")
+
+        self.socket = None
+
+    def connect(self):
+        """Connect to the StatsD server."""
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.socket.setblocking(0)
+
+    def disconnect(self):
+        """Disconnect from the StatsD server."""
+        if self.socket is not None:
+            self.socket.close()
+        self.socket = None
+
+    def write(self, data):
+        """Send the metric to the StatsD server."""
+        if self.host is None or self.port is None or self.socket is None:
+            return
+        try:
+            return self.socket.sendto(data, (self.host, self.port))
+        except (socket.error, socket.gaierror):
+            return None
+
+
+class InternalClient(object):
+    """A connection that can be used inside the C{StatsD} daemon itself."""
+
+    def __init__(self, processor):
+        """
+        A connection that writes directly to the C{MessageProcessor}.
+        """
+        self._processor = processor
+
+    def write(self, data):
+        """Write directly to the C{MessageProcessor}."""
+        self._processor.process(data)

=== added file 'txstatsd/itxstatsd.py'
--- txstatsd/itxstatsd.py	1970-01-01 00:00:00 +0000
+++ txstatsd/itxstatsd.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,54 @@
+from zope.interface import Interface, Attribute
+
+
+class IMetricFactory(Interface):
+    name = Attribute("""
+        @type name C{str}
+        @ivar name: The name of this kind of metric
+        """)
+
+    metric_type = Attribute("""
+        @type metric_type: C{str}
+        @ivar metric_type: The string that will be used by clients to send
+        metrics of this kind to the server.
+        """)
+
+    def build_metric(prefix, name, wall_time_func=None):
+        """
+        Returns an object that implements the C{IMetric} interface for name.
+
+        @type prefix: C{str}
+        @param prefix: The prefix used for reporting this metric.
+        @type name: C{str}
+        @param name: The name used for reporting this metric.
+        """
+
+    def configure(options):
+        """
+        Configures the factory. Will be called at startup by the service
+        factory.
+
+        @type options: C{twisted.python.usage.Options}
+        @param options: The configuration options.
+        """
+
+
+class IMetric(Interface):
+    def process(fields):
+        """
+        Process new data for this metric.
+
+        @type fields: C{list}
+        @param fields: The list of message parts. Usually in the form of
+        (value, metric_type, [sample_rate])
+        """
+
+    def flush(interval, timestamp):
+        """
+        Returns a string with new line separated list of metrics to report.
+
+        @type interval: C{float}
+        @param interval: The time since last flush.
+        @type timestamp: C{float}
+        @param timestamp: The timestamp for now.
+        """

=== added directory 'txstatsd/metrics'
=== added file 'txstatsd/metrics/__init__.py'
=== added file 'txstatsd/metrics/countermetric.py'
--- txstatsd/metrics/countermetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/countermetric.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,67 @@
+import math
+
+from txstatsd.metrics.metric import Metric
+
+
+class CounterMetric(Metric):
+    """An incrementing and decrementing counter metric."""
+
+    def __init__(self, connection, name, sample_rate=1):
+        """Construct a metric that reports samples to the supplied
+        C{connection}.
+
+        @param connection: The connection endpoint representing
+            the StatsD server.
+        @param name: Indicates what is being instrumented.
+        @param sample_rate: Restrict the number of samples sent
+            to the StatsD server based on the supplied C{sample_rate}.
+        """
+        Metric.__init__(self, connection, name, sample_rate=sample_rate)
+
+        self._count = 0
+
+    def increment(self, value):
+        """Increment the counter by C{value}"""
+        self._count += value
+        self._update(self._count)
+
+    def decrement(self, value):
+        """Decrement the counter by C{value}"""
+        self._count -= value
+        self._update(self._count)
+
+    def count(self):
+        """Returns the counter's current value."""
+        return self._count
+
+    def clear(self):
+        """Resets the counter to 0."""
+        self._count = 0
+        self._update(self._count)
+
+    def _update(self, value):
+        """Report the counter."""
+        self.send("%s|c" % value)
+
+
+class CounterMetricReporter(object):
+    """An incrementing and decrementing counter metric."""
+
+    def __init__(self, name, prefix=""):
+        """Construct a metric we expect to be periodically updated.
+
+        @param name: Indicates what is being instrumented.
+        """
+        self.name = name
+
+        if prefix:
+            prefix += "."
+        self.prefix = prefix
+        self.count = 0
+
+    def mark(self, value):
+        self.count = value
+
+    def report(self, timestamp):
+        return [(self.prefix + self.name + ".count",
+                 math.trunc(self.count), timestamp)]

=== added file 'txstatsd/metrics/distinctmetric.py'
--- txstatsd/metrics/distinctmetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/distinctmetric.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,150 @@
+# Copyright (C) 2011 Canonical
+# All Rights Reserved
+"""
+Implements a probabilistic distinct counter with sliding windows.
+
+Based on:
+http://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.12.7100
+
+And extended for sliding windows.
+"""
+import random
+import time
+
+from zope.interface import implements
+
+from txstatsd.metrics.metric import Metric
+from txstatsd.itxstatsd import IMetric
+
+
+class SBoxHash(object):
+    """A very fast hash.
+
+    This class create a random hash function that is very fast.
+    Based on SBOXes. Not Crypto Strong.
+
+    Two instances of this class will hash differently.
+    """
+
+    def __init__(self):
+        self.table = [random.randint(0, 0xFFFFFFFF - 1) for i in range(256)]
+
+    def hash(self, data):
+        value = 0
+        for c in data:
+            value = value ^ self.table[ord(c)]
+            value = value * 3
+            value = value & 0xFFFFFFFF
+        return value
+
+
+def hash(data):
+    """Hash data using a random hasher."""
+    p = SBoxHash()
+    return p.hash(data)
+
+
+def zeros(n):
+    """Count the zeros to the right of the binary representation of n."""
+    count = 0
+    i = 0
+    while True:
+        v = (n >> i)
+        if v <= 0:
+            return count
+        if v & 1:
+            return count
+        count += 1
+        i += 1
+    return count
+
+
+class SlidingDistinctCounter(object):
+    """A probabilistic distinct counter with sliding windows."""
+
+    def __init__(self, n_hashes, n_buckets):
+        self.n_hashes = n_hashes
+        self.n_buckets = n_buckets
+
+        self.hashes = [SBoxHash() for i in range(n_hashes)]
+        self.buckets = [[0] * n_buckets for i in range(n_hashes)]
+
+    def add(self, when, item):
+        hashes = (h.hash(item) for h in self.hashes)
+        for i, value in enumerate(hashes):
+            self.buckets[i][min(self.n_buckets - 1, zeros(value))] = when
+
+    def distinct(self, since=0):
+        total = 0.0
+        for i in range(self.n_hashes):
+            least0 = 0
+            for b in range(self.n_buckets):
+                if self.buckets[i][b] <= since:
+                    break
+                least0 += 1
+            total += least0
+        v = total / self.n_hashes
+        return int((2 ** v) / 0.77351)
+
+
+class DistinctMetric(Metric):
+    """
+    Keeps an estimate of the distinct numbers of items seen on various
+    sliding windows of time.
+    """
+
+    def mark(self, item):
+        """Report this item was seen."""
+        self.send("%s|d" % item)
+
+
+class DistinctMetricReporter(object):
+    """
+    Keeps an estimate of the distinct numbers of items seen on various
+    sliding windows of time.
+    """
+    implements(IMetric)
+
+    def __init__(self, name, wall_time_func=time.time, prefix=""):
+        """Construct a metric we expect to be periodically updated.
+
+        @param name: Indicates what is being instrumented.
+        @param wall_time_func: Function for obtaining wall time.
+        @param prefix: If present, a string to prepend to the message
+            composed when C{report} is called.
+        """
+        self.name = name
+        self.wall_time_func = wall_time_func
+        self.counter = SlidingDistinctCounter(32, 32)
+        if prefix:
+            prefix += "."
+        self.prefix = prefix
+
+    def count(self):
+        return self.counter.distinct()
+
+    def count_1min(self, now):
+        return self.counter.distinct(now - 60)
+
+    def count_1hour(self, now):
+        return self.counter.distinct(now - 60 * 60)
+
+    def count_1day(self, now):
+        return self.counter.distinct(now - 60 * 60 * 24)
+
+    def process(self, fields):
+        self.update(fields[0])
+
+    def update(self, item):
+        self.counter.add(self.wall_time_func(), item)
+
+    def flush(self, interval, timestamp):
+        now = self.wall_time_func()
+        metrics = []
+        items = {".count": self.count(),
+                 ".count_1min": self.count_1min(now),
+                 ".count_1hour": self.count_1hour(now),
+                 ".count_1day": self.count_1day(now)}
+        for item, value in items.iteritems():
+            metrics.append((self.prefix + self.name + item, value, timestamp))
+        return metrics

=== added file 'txstatsd/metrics/extendedmetrics.py'
--- txstatsd/metrics/extendedmetrics.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/extendedmetrics.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,53 @@
+
+from txstatsd.metrics.countermetric import CounterMetric
+from txstatsd.metrics.timermetric import TimerMetric
+from txstatsd.metrics.metrics import Metrics
+
+
+class ExtendedMetrics(Metrics):
+
+    def __init__(self, connection=None, namespace=""):
+        """A convenience class for reporting metric samples
+        to a C{txstatsd} server configured with the
+        L{ConfigurableProcessor<txstatsd.server.configurableprocessor>}
+        processor.
+
+        @param connection: The connection endpoint representing
+            the C{txstatsd} server.
+        @param namespace: The top-level namespace identifying the
+            origin of the samples.
+        """
+
+        super(ExtendedMetrics, self).__init__(connection, namespace)
+
+    def increment(self, name, value=1, sample_rate=1):
+        """Report and increase in name by count."""
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = CounterMetric(self.connection,
+                                   name,
+                                   sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].increment(value)
+
+    def decrement(self, name, value=1, sample_rate=1):
+        """Report and decrease in name by count."""
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = CounterMetric(self.connection,
+                                   name,
+                                   sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].decrement(value)
+
+    def timing(self, name, duration=None, sample_rate=1):
+        """Report this sample performed in duration seconds."""
+        if duration is None:
+            duration = self.calculate_duration()
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = TimerMetric(self.connection,
+                                 name,
+                                 sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].mark(duration)

=== added file 'txstatsd/metrics/gaugemetric.py'
--- txstatsd/metrics/gaugemetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/gaugemetric.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,43 @@
+from txstatsd.metrics.metric import Metric
+
+
+class GaugeMetric(Metric):
+    """A gauge metric is an instantaneous reading of a particular value."""
+
+    def __init__(self, connection, name, sample_rate=1):
+        """Construct a metric that reports samples to the supplied
+        C{connection}.
+
+        @param connection: The connection endpoint representing
+            the StatsD server.
+        @param name: Indicates what is being instrumented.
+        @param sample_rate: Restrict the number of samples sent
+            to the StatsD server based on the supplied C{sample_rate}.
+        """
+        Metric.__init__(self, connection, name, sample_rate=sample_rate)
+
+    def mark(self, value):
+        """Report the C{value} for this gauge."""
+        self.send("%s|g" % value)
+
+
+class GaugeMetricReporter(object):
+    """A gauge metric is an instantaneous reading of a particular value."""
+
+    def __init__(self, name, prefix=""):
+        """Construct a metric we expect to be periodically updated.
+
+        @param name: Indicates what is being instrumented.
+        """
+        self.name = name
+
+        if prefix:
+            prefix += "."
+        self.prefix = prefix
+        self.value = 0
+
+    def mark(self, value):
+        self.value = value
+
+    def report(self, timestamp):
+        return [(self.prefix + self.name + ".value", self.value, timestamp)]

=== added file 'txstatsd/metrics/histogrammetric.py'
--- txstatsd/metrics/histogrammetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/histogrammetric.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,209 @@
+# -*- coding: utf-8 *-*
+import math
+
+from txstatsd.stats.exponentiallydecayingsample \
+    import ExponentiallyDecayingSample
+from txstatsd.stats.uniformsample import UniformSample
+
+
+class HistogramMetricReporter(object):
+    """
+    A metric which calculates the distribution of a value.
+
+    See:
+    - U{Accurately computing running variance
+          <http://www.johndcook.com/standard_deviation.html>}
+    """
+
+    @classmethod
+    def using_uniform_sample(cls, prefix=""):
+        """
+        Uses a uniform sample of 1028 elements, which offers a 99.9%
+        confidence level with a 5% margin of error assuming a normal
+        distribution.
+         """
+        sample = UniformSample(1028)
+        return HistogramMetricReporter(sample, prefix=prefix)
+
+    @classmethod
+    def using_exponentially_decaying_sample(cls, prefix=""):
+        """
+        Uses an exponentially decaying sample of 1028 elements, which offers
+        a 99.9% confidence level with a 5% margin of error assuming a normal
+        distribution, and an alpha factor of 0.015, which heavily biases
+        the sample to the past 5 minutes of measurements.
+        """
+        sample = ExponentiallyDecayingSample(1028, 0.015)
+        return HistogramMetricReporter(sample, prefix=prefix)
+
+    def __init__(self, sample, prefix=""):
+        """Creates a new HistogramMetric with the given sample.
+
+        @param sample: The sample to create a histogram from.
+        """
+        self.sample = sample
+
+        if prefix:
+            prefix += "."
+        self.prefix = prefix
+
+        self._min = 0
+        self._max = 0
+        self._sum = 0
+
+        # These are for the Welford algorithm for calculating running
+        # variance without floating-point doom.
+        self.variance = [-1.0, 0.0]    # M, S
+        self.count = 0
+
+        self.clear()
+
+    def clear(self):
+        """Clears all recorded values."""
+        self.sample.clear()
+        self.count = 0
+        self._max = None
+        self._min = None
+        self._sum = 0
+        self.variance = [-1.0, 0.0]
+
+    def update(self, value, name=""):
+        """Adds a recorded value.
+
+        @param value: The length of the value.
+        """
+        self.count += 1
+        self.sample.update(value)
+        self.set_max(value)
+        self.set_min(value)
+        self._sum += value
+        self.update_variance(value)
+
+    def report(self, timestamp):
+        # median, 75, 95, 98, 99, 99.9 percentile
+        percentiles = self.percentiles(0.5, 0.75, 0.95, 0.98, 0.99, 0.999)
+        metrics = []
+        items = {
+            ".min": self.min(),
+            ".max": self.max(),
+            ".mean": self.mean(),
+            ".stddev": self.std_dev(),
+            ".median": percentiles[0],
+            ".75percentile": percentiles[1],
+            ".95percentile": percentiles[2],
+            ".98percentile": percentiles[3],
+            ".99percentile": percentiles[4],
+            ".999percentile": percentiles[5]}
+
+        for item, value in items.itervalues():
+            metrics.append((self.prefix + self.name + item, value, timestamp))
+        return metrics
+
+    def min(self):
+        """Returns the smallest recorded value."""
+        return self._min if self.count > 0 else 0.0
+
+    def max(self):
+        """Returns the largest recorded value."""
+        return self._max if self.count > 0 else 0.0
+
+    def mean(self):
+        """Returns the arithmetic mean of all recorded values."""
+        return float(self._sum) / self.count if self.count > 0 else 0.0
+
+    def std_dev(self):
+        """Returns the standard deviation of all recorded values."""
+        return math.sqrt(self.get_variance()) if self.count > 0 else 0.0
+
+    def percentiles(self, *percentiles):
+        """Returns a list of values at the given percentiles.
+
+        @param percentiles one or more percentiles
+        """
+
+        scores = [0.0] * len(percentiles)
+        if self.count > 0:
+            values = self.sample.get_values()
+            values.sort()
+
+            for i in range(len(percentiles)):
+                p = percentiles[i]
+                pos = p * (len(values) + 1)
+                if pos < 1:
+                    scores[i] = values[0]
+                elif pos >= len(values):
+                    scores[i] = values[-1]
+                else:
+                    lower = values[int(pos) - 1]
+                    upper = values[int(pos)]
+                    scores[i] = lower + (pos - math.floor(pos)) * (
+                        upper - lower)
+
+        return scores
+
+    def histogram(self):
+        """Returns an histogram of the sample.
+        """
+
+        # If we dont have data, build an empty histogram.
+        if not self.count > 0:
+            return [0.0] * 10
+
+        # Sturges Rule for selecting the number of bins
+        # Sturges, H. A. (1926) The choice of a class interval.
+        # Journal of the American Statistical Association 21, 65–66.
+        n_bins = int(math.ceil(1 + math.log(self.count, 2)))
+
+        scores = [0.0] * n_bins
+
+        values = self.sample.get_values()
+        max_value = float(max(values))
+        min_value = float(min(values))
+        value_range = max_value - min_value
+
+        for value in values:
+            pos = int(((value - min_value) / value_range) * n_bins)
+            if pos == n_bins:
+                pos -= 1
+
+            scores[pos] += 1
+        return scores
+
+    def get_values(self):
+        """Returns a list of all values in the histogram's sample."""
+        return self.sample.get_values()
+
+    def get_variance(self):
+        if self.count <= 1:
+            return 0.0
+        return self.variance[1] / (self.count - 1)
+
+    def set_max(self, potential_max):
+        if self._max is None:
+            self._max = potential_max
+        else:
+            self._max = max(self.max(), potential_max)
+
+    def set_min(self, potential_min):
+        if self._min is None:
+            self._min = potential_min
+        else:
+            self._min = min(self.min(), potential_min)
+
+    def update_variance(self, value):
+        old_values = self.variance
+        new_values = [0.0, 0.0]
+        if old_values[0] == -1:
+            new_values[0] = value
+            new_values[1] = 0.0
+        else:
+            old_m = old_values[0]
+            old_s = old_values[1]
+
+            new_m = old_m + (float(value - old_m) / self.count)
+            new_s = old_s + (float(value - old_m) * (value - new_m))
+
+            new_values[0] = new_m
+            new_values[1] = new_s
+
+        self.variance = new_values

=== added file 'txstatsd/metrics/imetrics.py'
--- txstatsd/metrics/imetrics.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/imetrics.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,18 @@
+from zope.interface import Interface
+
+class IMetrics(Interface):
+    """
+    Provides a global utility for sending out metric information.
+    """
+
+    def gauge(name, value, sample_rate=1):
+        """Record an absolute reading for C{name} with C{value}."""
+
+    def increment(name, value=1, sample_rate=1):
+        """Increment counter C{name} by C{count}."""
+
+    def decrement(name, value=1, sample_rate=1):
+        """Decrement counter C{name} by C{count}."""
+
+    def timing(name, duration=None, sample_rate=1):
+        """Report that C{name} took C{duration} seconds."""

=== added file 'txstatsd/metrics/metermetric.py'
--- txstatsd/metrics/metermetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/metermetric.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,103 @@
+import time
+
+from txstatsd.metrics.metric import Metric
+from txstatsd.stats.ewma import Ewma
+
+
+class MeterMetric(Metric):
+    """
+    A meter metric which measures mean throughput and one-, five-, and
+    fifteen-minute exponentially-weighted moving average throughputs.
+
+    See:
+    - U{EMA
+    <http://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average>}
+    """
+
+    def __init__(self, connection, name, sample_rate=1):
+        """Construct a metric that reports samples to the supplied
+        C{connection}.
+
+        @param connection: The connection endpoint representing
+            the StatsD server.
+        @param name: Indicates what is being instrumented.
+        @param sample_rate: Restrict the number of samples sent
+            to the StatsD server based on the supplied C{sample_rate}.
+        """
+        Metric.__init__(self, connection, name, sample_rate=sample_rate)
+
+    def mark(self, value=1):
+        """Mark the occurrence of a given number (C{value}) of events."""
+        self.send("%s|m" % value)
+
+
+class MeterMetricReporter(object):
+    """
+    A meter metric which measures mean throughput and one-, five-, and
+    fifteen-minute exponentially-weighted moving average throughputs.
+
+    See:
+    - U{EMA
+    <http://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average>}
+    """
+
+    def __init__(self, name, wall_time_func=time.time, prefix=""):
+        """Construct a metric we expect to be periodically updated.
+
+        @param name: Indicates what is being instrumented.
+        @param wall_time_func: Function for obtaining wall time.
+        """
+        self.name = name
+        self.wall_time_func = wall_time_func
+
+        if prefix:
+            prefix += "."
+        self.prefix = prefix
+
+        self.m1_rate = Ewma.one_minute_ewma()
+        self.m5_rate = Ewma.five_minute_ewma()
+        self.m15_rate = Ewma.fifteen_minute_ewma()
+        self.count = 0
+        self.start_time = self.wall_time_func()
+
+    def mark(self, value=1):
+        """Mark the occurrence of a given number of events."""
+        self.count += value
+        self.m1_rate.update(value)
+        self.m5_rate.update(value)
+        self.m15_rate.update(value)
+
+    def tick(self):
+        """Updates the moving averages."""
+        self.m1_rate.tick()
+        self.m5_rate.tick()
+        self.m15_rate.tick()
+
+    def report(self, timestamp):
+        metrics = []
+        items = {".count": self.count,
+                 ".mean_rate": self.mean_rate(),
+                 ".1min_rate": self.one_minute_rate(),
+                 ".5min_rate": self.five_minute_rate(),
+                 ".15min_rate": self.fifteen_minute_rate()}
+
+        for item, value in items.iteritems():
+            metrics.append((self.prefix + self.name + item,
+                            round(value, 6), timestamp))
+        return metrics
+
+    def fifteen_minute_rate(self):
+        return self.m15_rate.rate
+
+    def five_minute_rate(self):
+        return self.m5_rate.rate
+
+    def one_minute_rate(self):
+        return self.m1_rate.rate
+
+    def mean_rate(self):
+        if self.count == 0:
+            return 0.0
+        else:
+            elapsed = self.wall_time_func() - self.start_time
+            return float(self.count) / elapsed

=== added file 'txstatsd/metrics/metric.py'
--- txstatsd/metrics/metric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/metric.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,47 @@
+
+import random
+
+
+class Metric(object):
+    """
+    The foundation metric from which the specialized
+    metrics are derived.
+    """
+
+    def __init__(self, connection, name, sample_rate=1):
+        """Construct a metric that reports samples to the supplied
+        C{connection}.
+
+        @param connection: The connection endpoint representing
+            the StatsD server.
+        @param name: Specific description for this metric.
+        @param sample_rate: Restrict the number of samples sent
+            to the StatsD server based on the supplied C{sample_rate}.
+        """
+        self.connection = connection
+        self.name = name
+        self.sample_rate = sample_rate
+
+    def clear(self):
+        """Responsibility of the specialized metrics."""
+        pass
+
+    def send(self, data):
+        """
+        Message the C{data} to the C{StatsD} server according to the
+        C{sample_rate}.
+        """
+
+        if self.sample_rate < 1:
+            if random.random() > self.sample_rate:
+                return
+            data += "|@%s" % (self.sample_rate,)
+
+        data = self.name + ":" + data
+
+        self.write(data)
+
+    def write(self, data):
+        """Message the C{data} to the C{StatsD} server."""
+        if self.connection is not None:
+            self.connection.write(data)

=== added file 'txstatsd/metrics/metrics.py'
--- txstatsd/metrics/metrics.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/metrics.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,140 @@
+
+import time
+from txstatsd.metrics.gaugemetric import GaugeMetric
+from txstatsd.metrics.metermetric import MeterMetric
+from txstatsd.metrics.distinctmetric import DistinctMetric
+from txstatsd.metrics.metric import Metric
+
+
+class GenericMetric(Metric):
+    def __init__(self, connection, key, name, sample_rate=1):
+        super(GenericMetric, self).__init__(connection, name,
+                                            sample_rate=sample_rate)
+        self.key = key
+
+    def mark(self, value):
+        self.send("%s|%s" % (value, self.key))
+
+
+class Metrics(object):
+    def __init__(self, connection=None, namespace=""):
+        """A convenience class for reporting metric samples
+        to a StatsD server (C{connection}).
+
+        @param connection: The connection endpoint representing
+            the StatsD server.
+        @param namespace: The top-level namespace identifying the
+            origin of the samples.
+        """
+
+        self.connection = connection
+        self.namespace = namespace
+        self._metrics = {}
+        self.last_time = 0
+
+    def report(self, name, value, metric_type, sample_rate=1):
+        """Report a generic metric.
+
+        Used for server side plugins without client support.
+        """
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = GenericMetric(self.connection,
+                                        metric_type,
+                                        name,
+                                        sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].mark(value)
+
+    def gauge(self, name, value, sample_rate=1):
+        """Report an instantaneous reading of a particular value."""
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            gauge_metric = GaugeMetric(self.connection,
+                                       name,
+                                       sample_rate)
+            self._metrics[name] = gauge_metric
+        self._metrics[name].mark(value)
+
+    def meter(self, name, value=1, sample_rate=1):
+        """Mark the occurrence of a given number of events."""
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            meter_metric = MeterMetric(self.connection,
+                                       name,
+                                       sample_rate)
+            self._metrics[name] = meter_metric
+        self._metrics[name].mark(value)
+
+    def increment(self, name, value=1, sample_rate=1):
+        """Report and increase in name by count."""
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = Metric(self.connection,
+                            name,
+                            sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].send("%s|c" % value)
+
+    def decrement(self, name, value=1, sample_rate=1):
+        """Report and decrease in name by count."""
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = Metric(self.connection,
+                            name,
+                            sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].send("%s|c" % -value)
+
+    def reset_timing(self):
+        """Resets the duration timer for the next call to timing()"""
+        self.last_time = time.time()
+
+    def calculate_duration(self):
+        """Resets the duration timer and returns the elapsed duration"""
+        current_time = time.time()
+        duration = current_time - self.last_time
+        self.last_time = current_time
+        return duration
+
+    def timing(self, name, duration=None, sample_rate=1):
+        """Report that this sample performed in duration seconds.
+           Default duration is the actual elapsed time since
+           the last call to this method or reset_timing()"""
+        if duration is None:
+            duration = self.calculate_duration()
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = Metric(self.connection,
+                            name,
+                            sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].send("%s|ms" % (duration * 1000))
+
+    def distinct(self, name, item):
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = DistinctMetric(self.connection, name)
+            self._metrics[name] = metric
+        self._metrics[name].mark(item)
+
+    def clear(self, name):
+        """Allow the metric to re-initialize its internal state."""
+        name = self.fully_qualify_name(name)
+        if name in self._metrics:
+            metric = self._metrics[name]
+            if getattr(metric, 'clear', None) is not None:
+                metric.clear()
+
+    def fully_qualify_name(self, name):
+        """Compose the fully-qualified name: namespace and name."""
+        fully_qualified_name = ""
+        if self.namespace is not None:
+            fully_qualified_name = self.namespace
+        if name is not None:
+            # prepend the separator should we have a namespace
+            if self.namespace is not None and len(self.namespace) > 0:
+                fully_qualified_name += "." + name
+            else:
+                fully_qualified_name = name
+        return fully_qualified_name

=== added file 'txstatsd/metrics/timermetric.py'
--- txstatsd/metrics/timermetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/timermetric.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,161 @@
+import json
+import time
+
+from twisted.web import resource
+
+from txstatsd.metrics.histogrammetric import HistogramMetricReporter
+from txstatsd.metrics.metermetric import MeterMetricReporter
+from txstatsd.metrics.metric import Metric
+from txstatsd.stats.exponentiallydecayingsample \
+    import ExponentiallyDecayingSample
+
+
+class TimerMetric(Metric):
+    """
+    A timer metric which aggregates timing durations and provides duration
+    statistics, plus throughput statistics via L{MeterMetric}.
+    """
+
+    def __init__(self, connection, name, sample_rate=1):
+        """Construct a metric that reports samples to the supplied
+        C{connection}.
+
+        @param connection: The connection endpoint representing
+            the StatsD server.
+        @param name: Indicates what is being instrumented.
+        @param sample_rate: Restrict the number of samples sent
+            to the StatsD server based on the supplied C{sample_rate}.
+        """
+        Metric.__init__(self, connection, name, sample_rate=sample_rate)
+
+    def mark(self, duration):
+        """Report this sample performed in duration (measured in seconds)."""
+        self.send("%s|ms" % (duration * 1000))
+
+
+class TimerResource(resource.Resource):
+    isLeaf = True
+
+    def __init__(self, reporter):
+        resource.Resource.__init__(self)
+        self.reporter = reporter
+
+    def render_GET(self, request):
+        result = dict(
+            histogram=self.reporter.histogram.histogram(),
+            max_value=self.reporter.max(),
+            min_value=self.reporter.min())
+        return json.dumps(result)
+
+
+class TimerMetricReporter(object):
+    """
+    A timer metric which aggregates timing durations and provides duration
+    statistics, plus throughput statistics via L{MeterMetricReporter}.
+    """
+
+    def __init__(self, name, wall_time_func=time.time, prefix=""):
+        """Construct a metric we expect to be periodically updated.
+
+        @param name: Indicates what is being instrumented.
+        @param wall_time_func: Function for obtaining wall time.
+        @param prefix: If present, a string to prepend to the message
+            composed when C{report} is called.
+        """
+        self.name = name
+        self.wall_time_func = wall_time_func
+
+        if prefix:
+            prefix += "."
+        self.prefix = prefix
+
+        sample = ExponentiallyDecayingSample(1028, 0.015)
+        self.histogram = HistogramMetricReporter(sample)
+        self.meter = MeterMetricReporter(
+            "calls", wall_time_func=self.wall_time_func)
+        self.clear()
+
+    def clear(self):
+        """Clears all recorded durations."""
+        self.histogram.clear()
+
+    def count(self):
+        return self.histogram.count
+
+    def fifteen_minute_rate(self):
+        return self.meter.fifteen_minute_rate()
+
+    def five_minute_rate(self):
+        return self.meter.five_minute_rate()
+
+    def mean_rate(self):
+        return self.meter.mean_rate()
+
+    def one_minute_rate(self):
+        return self.meter.one_minute_rate()
+
+    def max(self):
+        """Returns the longest recorded duration."""
+        return self.histogram.max()
+
+    def min(self):
+        """Returns the shortest recorded duration."""
+        return self.histogram.min()
+
+    def mean(self):
+        """Returns the arithmetic mean of all recorded durations."""
+        return self.histogram.mean()
+
+    def std_dev(self):
+        """Returns the standard deviation of all recorded durations."""
+        return self.histogram.std_dev()
+
+    def getResource(self):
+        """Return an http resource to represent this."""
+        return TimerResource(self)
+
+    def percentiles(self, *percentiles):
+        """
+        Returns an array of durations at the given percentiles.
+
+        @param percentiles: One or more percentiles.
+        """
+        return [percentile for percentile in
+            self.histogram.percentiles(*percentiles)]
+
+    def get_values(self):
+        """Returns a list of all recorded durations in the timer's sample."""
+        return [value for value in self.histogram.get_values()]
+
+    def update(self, duration):
+        """Adds a recorded duration.
+
+        @param duration: The length of the duration in seconds.
+        """
+        if duration >= 0:
+            self.histogram.update(duration)
+            self.meter.mark()
+
+    def tick(self):
+        """Updates the moving averages."""
+        self.meter.tick()
+
+    def report(self, timestamp):
+        # median, 75, 95, 98, 99, 99.9 percentile
+        percentiles = self.percentiles(0.5, 0.75, 0.95, 0.98, 0.99, 0.999)
+        metrics = []
+        items = {".min": self.min(),
+                 ".max": self.max(),
+                 ".mean": self.mean(),
+                 ".stddev": self.std_dev(),
+                 ".99percentile": percentiles[4],
+                 ".999percentile": percentiles[5],
+                 ".count": self.meter.count,
+                 ".1min_rate": self.meter.one_minute_rate(),
+                 ".5min_rate": self.meter.five_minute_rate(),
+                 ".15min_rate": self.meter.fifteen_minute_rate()}
+
+        for item, value in items.iteritems():
+            metrics.append((self.prefix + self.name + item,
+                            round(value, 6), timestamp))
+        return metrics

=== added file 'txstatsd/process.py'
--- txstatsd/process.py	1970-01-01 00:00:00 +0000
+++ txstatsd/process.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,217 @@
+import os
+import socket
+import psutil
+
+from functools import update_wrapper
+
+
+MEMINFO_KEYS = ("MemTotal:", "MemFree:", "Buffers:",
+                "Cached:", "SwapCached:", "SwapTotal:",
+                "SwapFree:")
+
+MULTIPLIERS = {"kB": 1024, "mB": 1024 * 1024}
+
+
+def load_file(filename):
+    """Load a file into memory."""
+    with open(filename, "r") as f:
+        return f.read()
+
+
+def parse_meminfo(data, prefix="sys.mem"):
+    """Parse data from /proc/meminfo."""
+    result = {}
+
+    for line in data.split("\n"):
+        if not line:
+            continue
+        parts = [x for x in line.split(" ") if x]
+        if not parts[0] in MEMINFO_KEYS:
+            continue
+
+        multiple = 1
+
+        if len(parts) == 3:
+            multiple = MULTIPLIERS[parts[2]]
+
+        # remove ':'
+        label = parts[0][:-1]
+        amount = int(parts[1]) * multiple
+        result[prefix + "." + label] = amount
+
+    return result
+
+
+def parse_loadavg(data, prefix="sys.loadavg"):
+    """Parse data from /proc/loadavg."""
+    return dict(zip(
+        (prefix + ".oneminute",
+         prefix + ".fiveminutes",
+         prefix + ".fifthteenminutes"),
+        [float(x) for x in data.split()[:3]]))
+
+
+def parse_netdev(data, prefix="sys.net"):
+    """Parse data from /proc/net/dev."""
+    lines = data.splitlines()
+    # Parse out the column headers as keys.
+    _, receive_columns, transmit_columns = lines[1].split("|")
+    columns = ["recv_%s" % column for column in receive_columns.split()]
+    columns.extend(["send_%s" % column for column in transmit_columns.split()])
+
+    # Parse out the network devices.
+    result = {}
+    for line in lines[2:]:
+        if not ":" in line:
+            continue
+        device, data = line.split(":")
+        device = device.strip()
+        data = dict(zip(columns, map(int, data.split())))
+        result.update({
+            "%s.%s.bytes.received" % (prefix, device): data["recv_bytes"],
+            "%s.%s.bytes.sent" % (prefix, device): data["send_bytes"],
+            "%s.%s.packets.received" % (prefix, device): data["recv_packets"],
+            "%s.%s.packets.sent" % (prefix, device): data["send_packets"]})
+    return result
+
+
+class ProcessReport(object):
+
+    def __init__(self, process=None):
+        self._process = process
+
+    @property
+    def process(self):
+        """Override property with current process on first access."""
+        if self._process is None:
+            self._process = psutil.Process(os.getpid())
+        return self._process
+
+    def get_memory_and_cpu(self, prefix="proc"):
+        """Report memory and CPU stats for C{process}."""
+        vsize, rss = self.process.get_memory_info()
+        result = {prefix + ".cpu.percent": self.process.get_cpu_percent(),
+                  prefix + ".memory.percent":
+                    self.process.get_memory_percent(),
+                  prefix + ".memory.vsize": vsize,
+                  prefix + ".memory.rss": rss}
+        if getattr(self.process, "get_num_threads", None) is not None:
+            result[prefix + ".threads"] = self.process.get_num_threads()
+        return result
+
+    def get_cpu_counters(self, prefix="proc"):
+        """Report memory and CPU counters for C{process}."""
+        utime, stime = self.process.get_cpu_times()
+        result = {prefix + ".cpu.user": utime,
+                  prefix + ".cpu.system": stime}
+        return result
+
+    def get_io_counters(self, prefix="proc.io"):
+        """Report IO statistics for C{process}."""
+        result = {}
+        if getattr(self.process, "get_io_counters", None) is not None:
+            (read_count, write_count,
+             read_bytes, write_bytes) = self.process.get_io_counters()
+            result.update({
+                prefix + ".read.count": read_count,
+                prefix + ".write.count": write_count,
+                prefix + ".read.bytes": read_bytes,
+                prefix + ".write.bytes": write_bytes})
+        return result
+
+    def get_net_stats(self, prefix="proc.net"):
+        """Report active connection statistics for C{process}."""
+        result = {}
+        if getattr(self.process, "get_connections", None) is not None:
+            for connection in self.process.get_connections():
+                fd, family, _type, laddr, raddr, status = connection
+                if _type == socket.SOCK_STREAM:
+                    key = prefix + ".status.%s" % status.lower()
+                    if not key in result:
+                        result[key] = 1
+                    else:
+                        result[key] += 1
+        return result
+
+
+def report_counters(report_function):
+    """
+    Report difference between last value and current value for wrapped
+    function.
+    """
+    def generate():
+        last_value = None
+        while True:
+            result = {}
+            new_value = report_function()
+            if last_value is None:
+                last_value = new_value
+            else:
+                for key, value in new_value.iteritems():
+                    result[key] = value - last_value[key]
+                last_value = new_value
+            yield result
+    generator = generate()
+    def report():
+        return generator.next()
+    update_wrapper(report, report_function)
+    return report
+
+
+process_report = ProcessReport()
+report_process_memory_and_cpu = process_report.get_memory_and_cpu
+report_process_cpu_counters = report_counters(process_report.get_cpu_counters)
+report_process_io_counters = report_counters(process_report.get_io_counters)
+report_process_net_stats = process_report.get_net_stats
+
+
+def report_system_stats(prefix="sys"):
+    cpu_times = psutil.cpu_times()._asdict()
+    system_stats = {}
+    for mode, time in cpu_times.iteritems():
+        system_stats[prefix + ".cpu." + mode] = time
+    return system_stats
+
+
+def report_threadpool_stats(threadpool, prefix="threadpool"):
+    """Report stats about a given threadpool."""
+    def report():
+        return {prefix + ".working": len(threadpool.working),
+                prefix + ".queue": threadpool.q.qsize(),
+                prefix + ".waiters": len(threadpool.waiters),
+                prefix + ".threads": len(threadpool.threads)}
+    update_wrapper(report, report_threadpool_stats)
+    return report
+
+
+def report_reactor_stats(reactor, prefix="reactor"):
+    """Report statistics about a twisted reactor."""
+    def report():
+        return {prefix + ".readers": len(reactor.getReaders()),
+                prefix + ".writers": len(reactor.getWriters())}
+
+    update_wrapper(report, report_reactor_stats)
+    return report
+
+
+def report_file_stats(filename, parser):
+    """Read statistics from a file and report them."""
+    def report():
+        return parser(load_file(filename))
+    update_wrapper(report, report_file_stats)
+    return report
+
+
+PROCESS_STATS = (report_process_memory_and_cpu,)
+
+COUNTER_STATS = (report_process_cpu_counters,)
+
+IO_STATS = (report_process_io_counters,)
+
+NET_STATS = (report_process_net_stats,)
+
+SYSTEM_STATS = (report_file_stats("/proc/meminfo", parse_meminfo),
+                report_file_stats("/proc/loadavg", parse_loadavg),
+                report_counters(report_file_stats("/proc/net/dev", parse_netdev)),
+                report_counters(report_system_stats),
+                )

=== added file 'txstatsd/report.py'
--- txstatsd/report.py	1970-01-01 00:00:00 +0000
+++ txstatsd/report.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,64 @@
+from twisted.internet.defer import maybeDeferred
+from twisted.internet.task import LoopingCall
+from twisted.python import log
+
+from functools import wraps
+
+from twisted.application.service import Service
+from twisted.internet import threads
+
+
+class ReportingService(Service):
+
+    def __init__(self, instance_name="", clock=None):
+        self.tasks = []
+        self.clock = clock
+        self.instance_name = instance_name
+
+    def schedule(self, function, interval, report_function):
+        """
+        Schedule C{function} to be called every C{interval} seconds and then
+        report gathered metrics to C{Graphite} using C{report_function}.
+
+        If C{report_function} is C{None}, it just calls the function without
+        reporting the metrics.
+        """
+        if report_function is not None:
+            call = self.wrapped(function, report_function)
+        else:
+            call = function
+        task = LoopingCall(call)
+        if self.clock is not None:
+            task.clock = self.clock
+        self.tasks.append((task, interval))
+        if self.running:
+            task.start(interval, now=True)
+
+    def wrapped(self, function, report_function):
+        def report_metrics(metrics):
+            """For each metric returned, call C{report_function} with it."""
+            for name, value in metrics.items():
+                if self.instance_name:
+                    name = self.instance_name + "." + name
+                report_function(name, value)
+            return metrics
+
+        @wraps(function)
+        def wrapper():
+            """Wrap C{function} to report metrics or log a failure."""
+            deferred = maybeDeferred(function)
+            deferred.addCallback(report_metrics)
+            deferred.addErrback(lambda failure: log.err(
+                failure, "Error while processing %s" % function.func_name))
+            return deferred
+        return wrapper
+
+    def startService(self):
+        Service.startService(self)
+        for task, interval in self.tasks:
+            task.start(interval, now=False)
+
+    def stopService(self):
+        for task, interval in self.tasks:
+            task.stop()
+        Service.stopService(self)

=== added directory 'txstatsd/server'
=== added file 'txstatsd/server/__init__.py'
=== added file 'txstatsd/server/configurableprocessor.py'
--- txstatsd/server/configurableprocessor.py	1970-01-01 00:00:00 +0000
+++ txstatsd/server/configurableprocessor.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,108 @@
+
+import time
+
+from txstatsd.metrics.countermetric import CounterMetricReporter
+from txstatsd.metrics.gaugemetric import GaugeMetricReporter
+from txstatsd.metrics.metermetric import MeterMetricReporter
+from txstatsd.metrics.timermetric import TimerMetricReporter
+from txstatsd.server.processor import MessageProcessor
+
+
+class ConfigurableMessageProcessor(MessageProcessor):
+    """
+    This specialised C{MessageProcessor} supports behaviour
+    that is not StatsD-compliant.
+
+    Currently, this extends to:
+    - Allow a prefix to be added to the composed messages sent
+      to the Graphite server.
+    - Report an instantaneous reading of a particular value.
+    - Report an incrementing and decrementing counter metric.
+    - Report a timer metric which aggregates timing durations and provides
+      duration statistics, plus throughput statistics.
+    """
+
+    def __init__(self, time_function=time.time, message_prefix="",
+                 internal_metrics_prefix="", plugins=None):
+        super(ConfigurableMessageProcessor, self).__init__(
+            time_function=time_function, plugins=plugins)
+
+        if not internal_metrics_prefix and not message_prefix:
+            internal_metrics_prefix = "statsd."
+        elif message_prefix and not internal_metrics_prefix:
+            internal_metrics_prefix = message_prefix + "." + "statsd."
+        self.internal_metrics_prefix = internal_metrics_prefix
+        self.message_prefix = message_prefix
+        self.gauge_metrics = {}
+
+    def get_message_prefix(self, kind):
+        return self.message_prefix
+
+    def compose_timer_metric(self, key, duration):
+        if not key in self.timer_metrics:
+            metric = TimerMetricReporter(key, prefix=self.message_prefix)
+            self.timer_metrics[key] = metric
+        self.timer_metrics[key].update(duration)
+
+    def process_counter_metric(self, key, composite, message):
+        try:
+            value = float(composite[0])
+        except (TypeError, ValueError):
+            return self.fail(message)
+
+        self.compose_counter_metric(key, value)
+
+    def compose_counter_metric(self, key, value):
+        if not key in self.counter_metrics:
+            metric = CounterMetricReporter(key, prefix=self.message_prefix)
+            self.counter_metrics[key] = metric
+        self.counter_metrics[key].mark(value)
+
+    def compose_gauge_metric(self, key, value):
+        if not key in self.gauge_metrics:
+            metric = GaugeMetricReporter(key, prefix=self.message_prefix)
+            self.gauge_metrics[key] = metric
+        self.gauge_metrics[key].mark(value)
+
+    def compose_meter_metric(self, key, value):
+        if not key in self.meter_metrics:
+            metric = MeterMetricReporter(key, self.time_function,
+                                         prefix=self.message_prefix)
+            self.meter_metrics[key] = metric
+        self.meter_metrics[key].mark(value)
+
+    def flush_counter_metrics(self, interval, timestamp):
+        metrics = []
+        events = 0
+        for metric in self.counter_metrics.itervalues():
+            messages = metric.report(timestamp)
+            metrics.extend(messages)
+            events += 1
+
+        return (metrics, events)
+
+    def flush_gauge_metrics(self, timestamp):
+        metrics = []
+        events = 0
+        for metric in self.gauge_metrics.itervalues():
+            messages = metric.report(timestamp)
+            metrics.extend(messages)
+            events += 1
+
+        return (metrics, events)
+
+    def flush_timer_metrics(self, percent, timestamp):
+        metrics = []
+        events = 0
+        for metric in self.timer_metrics.itervalues():
+            messages = metric.report(timestamp)
+            metrics.extend(messages)
+            events += 1
+
+        return (metrics, events)
+
+    def update_metrics(self):
+        super(ConfigurableMessageProcessor, self).update_metrics()
+
+        for metric in self.timer_metrics.itervalues():
+            metric.tick()

=== added file 'txstatsd/server/httpinfo.py'
--- txstatsd/server/httpinfo.py	1970-01-01 00:00:00 +0000
+++ txstatsd/server/httpinfo.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,60 @@
+# -*- coding: utf-8 *-*
+import json
+
+from twisted.application import service, internet
+from twisted.web import server, resource, http
+
+
+class Status(resource.Resource):
+    isLeaf = True
+    time_high_water = 0.7
+
+    def __init__(self, processor, statsd_service):
+        resource.Resource.__init__(self)
+        self.processor = processor
+        self.statsd_service = statsd_service
+
+    def render_GET(self, request):
+        data = dict(flush_time=self.processor.last_flush_duration,
+                    process_time=self.processor.last_process_duration,
+                    flush_interval=self.statsd_service.flush_interval)
+        if data["flush_interval"] * self.time_high_water < (
+                data["process_time"] + data["flush_time"]):
+            data["status"] = "ERROR"
+            request.setResponseCode(http.INTERNAL_SERVER_ERROR)
+            return json.dumps(data)
+        data["status"] = "OK"
+        return json.dumps(data)
+
+
+class Metrics(resource.Resource):
+
+    def __init__(self, processor):
+        resource.Resource.__init__(self)
+        self.processor = processor
+
+    def getChild(self, name, request):
+        metric = self.processor.timer_metrics.get(name, None) or \
+                self.processor.plugin_metrics.get(name, None)
+        if metric is None:
+            return resource.NoResource()
+
+        meth = getattr(metric, "getResource", None)
+
+        if meth is None:
+            return resource.NoResource()
+
+        return meth()
+
+
+def makeService(options, processor, statsd_service):
+
+    if options["http-port"] is None:
+        return service.MultiService()
+
+    root = resource.Resource()
+    root.putChild("status", Status(processor, statsd_service))
+    root.putChild("metrics", Metrics(processor))
+    site = server.Site(root)
+    s = internet.TCPServer(int(options["http-port"]), site)
+    return s

=== added file 'txstatsd/server/loggingprocessor.py'
--- txstatsd/server/loggingprocessor.py	1970-01-01 00:00:00 +0000
+++ txstatsd/server/loggingprocessor.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,36 @@
+
+import time
+
+from txstatsd.server.configurableprocessor import ConfigurableMessageProcessor
+
+
+class LoggingMessageProcessor(ConfigurableMessageProcessor):
+    """
+    This specialised C{MessageProcessor} logs the received metrics
+    using the supplied logger (which should have a callable C{info}
+    attribute.)
+    """
+
+    def __init__( self, logger, time_function=time.time,
+            message_prefix="", plugins=None, **kwz ):
+        super(LoggingMessageProcessor, self).__init__(
+            time_function=time_function, message_prefix=message_prefix,
+            plugins=plugins, **kwz )
+
+        logger_info = getattr(logger, "info", None)
+        if logger_info is None or not callable(logger_info):
+            raise TypeError()
+        self.logger = logger
+
+    def process_message(self, message, metric_type, key, fields):
+        self.logger.info("In: %s" % message)
+        return super(LoggingMessageProcessor, self)\
+            .process_message(message, metric_type, key, fields)
+
+    def flush(self, interval=10000, percent=90):
+        """Log all received metric samples to the supplied logger."""
+        messages = list( super(LoggingMessageProcessor, self)\
+            .flush(interval=interval, percent=percent) )
+        for msg in messages:
+            self.logger.info("Out: %s %s %s" % msg)
+        return messages

=== added file 'txstatsd/server/processor.py'
--- txstatsd/server/processor.py	1970-01-01 00:00:00 +0000
+++ txstatsd/server/processor.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,371 @@
+from collections import deque
+import re
+import time
+import logging
+
+from twisted.python import log
+
+from txstatsd.metrics.metermetric import MeterMetricReporter
+
+
+SPACES = re.compile("\s+")
+SLASHES = re.compile("\/+")
+NON_ALNUM = re.compile("[^a-zA-Z_\-0-9\.]")
+RATE = re.compile("^@([\d\.]+)")
+
+
+def normalize_key(key):
+    """
+    Normalize a key that might contain spaces, forward-slashes and other
+    special characters into something that is acceptable by graphite.
+    """
+    key = SPACES.sub("_", key)
+    key = SLASHES.sub("-", key)
+    key = NON_ALNUM.sub("", key)
+    return key
+
+
+class BaseMessageProcessor(object):
+
+    def process(self, message):
+        """
+        """
+        if not ":" in message:
+            return self.fail(message)
+
+        key, data = message.strip().split(":", 1)
+        if not "|" in data:
+            return self.fail(message)
+
+        fields = data.split("|")
+        if len(fields) < 2 or len(fields) > 3:
+            return self.fail(message)
+
+        key = normalize_key(key)
+        metric_type = fields[1]
+        return self.process_message(message, metric_type, key, fields)
+
+    def rebuild_message(self, metric_type, key, fields):
+        return key + ":" + "|".join(fields)
+
+    def fail(self, message):
+        """Log and discard malformed message."""
+        log.msg("Bad line: %r" % message, logLevel=logging.DEBUG)
+
+
+class MessageProcessor(BaseMessageProcessor):
+    """
+    This C{MessageProcessor} produces StatsD-compliant messages
+    for publishing to a Graphite server.
+    Metrics behaviour that varies from StatsD should be placed in
+    some specialised C{MessageProcessor} (see L{ConfigurableMessageProcessor
+    <txstatsd.server.configurableprocessor.ConfigurableMessageProcessor>}).
+    """
+
+    def __init__(self, time_function=time.time, plugins=None):
+        self.time_function = time_function
+
+        self.stats_prefix = "stats."
+        self.internal_metrics_prefix = "statsd."
+        self.count_prefix = "stats_counts."
+        self.timer_prefix = self.stats_prefix + "timers."
+        self.gauge_prefix = self.stats_prefix + "gauge."
+
+        self.process_timings = {}
+        self.by_type = {}
+        self.last_flush_duration = 0
+        self.last_process_duration = 0
+
+        self.timer_metrics = {}
+        self.counter_metrics = {}
+        self.gauge_metrics = deque()
+        self.meter_metrics = {}
+
+        self.plugins = {}
+        self.plugin_metrics = {}
+
+        if plugins is not None:
+            for plugin in plugins:
+                self.plugins[plugin.metric_type] = plugin
+
+    def process_message(self, message, metric_type, key, fields):
+        """
+        Process a single entry, adding it to either C{counters}, C{timers},
+        or C{gauge_metrics} depending on which kind of message it is.
+        """
+        start = self.time_function()
+        if metric_type == "c":
+            self.process_counter_metric(key, fields, message)
+        elif metric_type == "ms":
+            self.process_timer_metric(key, fields[0], message)
+        elif metric_type == "g":
+            self.process_gauge_metric(key, fields[0], message)
+        elif metric_type == "m":
+            self.process_meter_metric(key, fields[0], message)
+        elif metric_type in self.plugins:
+            self.process_plugin_metric(metric_type, key, fields, message)
+        else:
+            return self.fail(message)
+        self.process_timings.setdefault(metric_type, 0)
+        self.process_timings[metric_type] += self.time_function() - start
+        self.by_type.setdefault(metric_type, 0)
+        self.by_type[metric_type] += 1
+
+    def get_message_prefix(self, kind):
+        return "stats." + kind
+
+    def process_plugin_metric(self, metric_type, key, items, message):
+        if not key in self.plugin_metrics:
+            factory = self.plugins[metric_type]
+            metric = factory.build_metric(
+                self.get_message_prefix(factory.name),
+                name=key, wall_time_func=self.time_function)
+            self.plugin_metrics[key] = metric
+        self.plugin_metrics[key].process(items)
+
+    def process_timer_metric(self, key, duration, message):
+        try:
+            duration = float(duration)
+        except (TypeError, ValueError):
+            return self.fail(message)
+
+        self.compose_timer_metric(key, duration)
+
+    def compose_timer_metric(self, key, duration):
+        if key not in self.timer_metrics:
+            self.timer_metrics[key] = []
+        self.timer_metrics[key].append(duration)
+
+    def process_counter_metric(self, key, composite, message):
+        try:
+            value = float(composite[0])
+        except (TypeError, ValueError):
+            return self.fail(message)
+        rate = 1
+        if len(composite) == 3:
+            match = RATE.match(composite[2])
+            if match is None:
+                return self.fail(message)
+            rate = match.group(1)
+
+        self.compose_counter_metric(key, value, rate)
+
+    def compose_counter_metric(self, key, value, rate):
+        if key not in self.counter_metrics:
+            self.counter_metrics[key] = 0
+        self.counter_metrics[key] += value * (1 / float(rate))
+
+    def process_gauge_metric(self, key, composite, message):
+        values = composite.split(":")
+        if not len(values) == 1:
+            return self.fail(message)
+
+        try:
+            value = float(values[0])
+        except (TypeError, ValueError):
+            self.fail(message)
+
+        self.compose_gauge_metric(key, value)
+
+    def compose_gauge_metric(self, key, value):
+        metric = [value, key]
+        self.gauge_metrics.append(metric)
+
+    def process_meter_metric(self, key, composite, message):
+        values = composite.split(":")
+        if not len(values) == 1:
+            return self.fail(message)
+
+        try:
+            value = float(values[0])
+        except (TypeError, ValueError):
+            self.fail(message)
+
+        self.compose_meter_metric(key, value)
+
+    def compose_meter_metric(self, key, value):
+        if not key in self.meter_metrics:
+            metric = MeterMetricReporter(key, self.time_function,
+                                         prefix="stats.meter")
+            self.meter_metrics[key] = metric
+        self.meter_metrics[key].mark(value)
+
+    def flush(self, interval=10000, percent=90):
+        """
+        Flush all queued stats, computing a normalized count based on
+        C{interval} and mean timings based on C{threshold}.
+        """
+        messages = []
+        per_metric = {}
+        num_stats = 0
+        interval = interval / 1000
+        timestamp = int(self.time_function())
+
+        start = self.time_function()
+        counter_metrics, events = self.flush_counter_metrics(interval,
+                                                             timestamp)
+        duration = self.time_function() - start
+        if events > 0:
+            messages.extend(sorted(counter_metrics))
+            num_stats += events
+        per_metric["counter"] = (events, duration)
+
+        start = self.time_function()
+        timer_metrics, events = self.flush_timer_metrics(percent, timestamp)
+        duration = self.time_function() - start
+        if events > 0:
+            messages.extend(sorted(timer_metrics))
+            num_stats += events
+        per_metric["timer"] = (events, duration)
+
+        start = self.time_function()
+        gauge_metrics, events = self.flush_gauge_metrics(timestamp)
+        duration = self.time_function() - start
+        if events > 0:
+            messages.extend(sorted(gauge_metrics))
+            num_stats += events
+        per_metric["gauge"] = (events, duration)
+
+        start = self.time_function()
+        meter_metrics, events = self.flush_meter_metrics(timestamp)
+        duration = self.time_function() - start
+        if events > 0:
+            messages.extend(sorted(meter_metrics))
+            num_stats += events
+        per_metric["meter"] = (events, duration)
+
+        start = self.time_function()
+        plugin_metrics, events = self.flush_plugin_metrics(interval, timestamp)
+        duration = self.time_function() - start
+        if events > 0:
+            messages.extend(sorted(plugin_metrics))
+            num_stats += events
+        per_metric["plugin"] = (events, duration)
+
+        self.flush_metrics_summary(messages, num_stats, per_metric, timestamp)
+        return messages
+
+    def flush_counter_metrics(self, interval, timestamp):
+        metrics = []
+        events = 0
+        for key, count in self.counter_metrics.iteritems():
+            self.counter_metrics[key] = 0
+
+            value = count / interval
+            metrics.append((self.stats_prefix + key, value, timestamp))
+            metrics.append((self.count_prefix + key, count, timestamp))
+            events += 1
+
+        return (metrics, events)
+
+    def flush_timer_metrics(self, percent, timestamp):
+        metrics = []
+        events = 0
+
+        threshold_value = ((100 - percent) / 100.0)
+        for key, timers in self.timer_metrics.iteritems():
+            count = len(timers)
+            if count > 0:
+                self.timer_metrics[key] = []
+
+                timers.sort()
+                lower = timers[0]
+                upper = timers[-1]
+                count = len(timers)
+
+                mean = lower
+                threshold_upper = upper
+
+                if count > 1:
+                    index = count - int(round(threshold_value * count))
+                    timers = timers[:index]
+                    threshold_upper = timers[-1]
+                    mean = sum(timers) / index
+
+                items = {".mean": mean,
+                         ".upper": upper,
+                         ".upper_%s" % percent: threshold_upper,
+                         ".lower": lower,
+                         ".count": count}
+                for item, value in items.iteritems():
+                    metrics.append((self.timer_prefix + key + item,
+                                    value, timestamp))
+                events += 1
+
+        return (metrics, events)
+
+    def flush_gauge_metrics(self, timestamp):
+        metrics = []
+        events = 0
+        for metric in self.gauge_metrics:
+            value = metric[0]
+            key = metric[1]
+
+            metrics.append((self.gauge_prefix + key + ".value",
+                            value, timestamp))
+            events += 1
+
+        self.gauge_metrics.clear()
+
+        return (metrics, events)
+
+    def flush_meter_metrics(self, timestamp):
+        metrics = []
+        events = 0
+        for metric in self.meter_metrics.itervalues():
+            messages = metric.report(timestamp)
+            metrics.extend(messages)
+            events += 1
+
+        return (metrics, events)
+
+    def flush_plugin_metrics(self, interval, timestamp):
+        metrics = []
+        events = 0
+
+        for metric in self.plugin_metrics.itervalues():
+            messages = metric.flush(interval, timestamp)
+            metrics.extend(messages)
+            events += 1
+
+        return (metrics, events)
+
+    def flush_metrics_summary(self, messages, num_stats,
+                              per_metric, timestamp):
+
+        messages.append((self.internal_metrics_prefix + "numStats",
+                         num_stats, timestamp))
+
+        self.last_flush_duration = 0
+        for name, (value, duration) in per_metric.iteritems():
+            messages.extend([
+                (self.internal_metrics_prefix +
+                 "flush.%s.count" % name,
+                 value, timestamp),
+                (self.internal_metrics_prefix +
+                 "flush.%s.duration" % name,
+                 duration * 1000, timestamp)])
+            log.msg("Flushed %d %s metrics in %.6f" %
+                    (value, name, duration))
+            self.last_flush_duration += duration
+
+        self.last_process_duration = 0
+        for metric_type, duration in self.process_timings.iteritems():
+            messages.extend([
+                (self.internal_metrics_prefix +
+                 "receive.%s.count" %
+                 metric_type, self.by_type[metric_type], timestamp),
+                (self.internal_metrics_prefix +
+                 "receive.%s.duration" %
+                 metric_type, duration * 1000, timestamp)
+                ])
+            log.msg("Processing %d %s metrics took %.6f" %
+                    (self.by_type[metric_type], metric_type, duration))
+            self.last_process_duration += duration
+
+        self.process_timings.clear()
+        self.by_type.clear()
+
+    def update_metrics(self):
+        for metric in self.meter_metrics.itervalues():
+            metric.tick()

=== added file 'txstatsd/server/protocol.py'
--- txstatsd/server/protocol.py	1970-01-01 00:00:00 +0000
+++ txstatsd/server/protocol.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,63 @@
+from twisted.internet.protocol import (
+    DatagramProtocol, Protocol, Factory)
+from twisted.protocols.basic import LineReceiver
+
+
+class StatsDServerProtocol(DatagramProtocol):
+    """A Twisted-based implementation of the StatsD server.
+
+    Data is received via UDP for local aggregation and then sent to a Graphite
+    server via TCP.
+    """
+
+    def __init__(self, processor,
+                 monitor_message=None, monitor_response=None):
+        self.processor = processor
+        self.monitor_message = monitor_message
+        self.monitor_response = monitor_response
+
+    def datagramReceived(self, data, (host, port)):
+        """Process received data and store it locally."""
+        if data == self.monitor_message:
+            # Send the expected response to the
+            # monitoring agent.
+            self.transport.write(self.monitor_response, (host, port))
+        else:
+            self.processor.process(data)
+
+
+class StatsDTCPServerProtocol(LineReceiver):
+    """A Twisted-based implementation of the StatsD server over TCP.
+
+    Data is received via TCP for local aggregation and then sent to a Graphite
+    server via TCP.
+    """
+
+    def __init__(self, processor,
+                 monitor_message=None, monitor_response=None):
+        self.processor = processor
+        self.monitor_message = monitor_message
+        self.monitor_response = monitor_response
+
+    def lineReceived(self, data):
+        """Process received data and store it locally."""
+        if data == self.monitor_message:
+            # Send the expected response to the
+            # monitoring agent.
+            self.transport.write(self.monitor_response)
+        else:
+            self.processor.process(data)
+
+
+class StatsDTCPServerFactory(Factory):
+
+    def __init__(self, processor,
+                 monitor_message=None, monitor_response=None):
+        self.processor = processor
+        self.monitor_message = monitor_message
+        self.monitor_response = monitor_response
+
+    def buildProtocol(self, addr):
+        return StatsDTCPServerProtocol(self.processor,
+            self.monitor_message, self.monitor_response)
+

=== added file 'txstatsd/server/router.py'
--- txstatsd/server/router.py	1970-01-01 00:00:00 +0000
+++ txstatsd/server/router.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,308 @@
+"""
+Routes messages to different processors.
+
+Rules are of the form:
+    condition => target
+Where each on of condition and target are of the form:
+    name [arguments]*
+And the arguments are dependant on the name.
+
+Each rule is applied to the message on the order they are specified.
+
+Conditions supported:
+    any: will match all messages
+    metric_type [type]+: will match a metric of any of the types specified
+    path_like fnmatch_exp: will match the path against the expression with
+        fnmatch.
+    not [rule..]: will return the negation of the result of rule
+
+
+Targets supported:
+    drop: will drop the message, stopping any further processing.
+    redirect_udp host port: will send to (host, port) by udp
+    redirect_tcp host port: will send to (host, port) by tcp
+    rewrite pattern repl: will rewrite the path like re.sub
+    set_metric_type metric_type: will make the metric of type metric_type
+
+"""
+import fnmatch
+import re
+import time
+
+from zope.interface import implements
+
+from twisted.application.internet import UDPServer
+from twisted.application.service import Service
+from twisted.internet import interfaces
+from twisted.internet.protocol import (
+    DatagramProtocol, ReconnectingClientFactory, Protocol)
+from twisted.internet import defer
+from twisted.python import log
+
+from txstatsd.server.processor import BaseMessageProcessor
+
+
+class StopProcessingException(Exception):
+
+    pass
+
+
+class TCPRedirectService(Service):
+
+    def __init__(self, host, port, factory):
+        self.host = host
+        self.port = port
+        self.factory = factory
+
+    def startService(self):
+        from twisted.internet import reactor
+
+        reactor.connectTCP(self.host, self.port, self.factory)
+        return Service.startService(self)
+
+    def stopService(self):
+        self.factory.stopTrying()
+        if self.factory.protocol:
+            self.factory.protocol.transport.loseConnection()
+        return Service.stopService(self)
+
+
+class TCPRedirectClientFactory(ReconnectingClientFactory):
+
+    def __init__(self, callback=None):
+        self.callback = callback
+        self.protocol = None
+
+    def buildProtocol(self, addr):
+        from twisted.internet import reactor
+
+        self.resetDelay()
+        self.protocol = TCPRedirectProtocol()
+        if self.callback:
+            reactor.callLater(0, self.callback)
+            self.callback = None
+
+        return self.protocol
+
+    def write(self, data):
+        if self.protocol:
+            self.protocol.write(data)
+
+
+class TCPRedirectProtocol(Protocol):
+    """A client protocol for redicting messages over tcp.
+    """
+
+    implements(interfaces.IPushProducer)
+
+    def __init__(self):
+        self.paused = False
+        self.last_paused = None
+        self.dropped = 0
+
+    def connectionMade(self):
+        """
+        A connection has been made, register ourselves as a producer for the
+        bound transport.
+        """
+        self.transport.registerProducer(self, True)
+
+    def pauseProducing(self):
+        """Pause producing messages, since the buffer is full."""
+        self.last_paused = int(time.time())
+        self.paused = True
+
+    stopProducing = pauseProducing
+
+    def resumeProducing(self):
+        """We can write to the transport again. Yay!."""
+        time_now = int(time.time())
+        log("Resumed TCP redirect. "
+            "Dropped %s messages during %s seconds ",
+                self.dropped, time_now - self.last_paused)
+        self.paused = False
+        self.dropped = 0
+        self.last_paused = None
+
+    def write(self, line):
+        if self.paused:
+            self.dropped += 1
+            return
+
+        if line[-2:] != "\r\n":
+            if line[-1] == "\r":
+                line += "\n"
+            else:
+                line += "\r\n"
+        self.transport.write(line)
+
+
+class UDPRedirectProtocol(DatagramProtocol):
+
+    def __init__(self, host, port, callback):
+        self.host = host
+        self.port = port
+        self.callback = callback
+
+    def startProtocol(self):
+        self.transport.connect(self.host, self.port)
+        self.callback()
+
+    def write(self, data):
+        if self.transport is not None:
+            self.transport.write(data)
+
+
+class Router(BaseMessageProcessor):
+
+    def __init__(self, message_processor, rules_config, service=None):
+        """Configure a router with rules_config.
+
+        rules_config is a new_line separeted list of rules.
+        """
+        self.rules_config = rules_config
+        self.message_processor = message_processor
+        self.flush = message_processor.flush
+        self.ready = defer.succeed(None)
+        self.service = service
+        self.rules = self.build_rules(rules_config)
+
+    def build_condition(self, condition):
+        condition_parts = [
+            p.strip() for p in condition.split(" ") if p]
+        condition_factory = getattr(self,
+            "build_condition_" + condition_parts[0], None)
+        if condition_factory is None:
+            raise ValueError("unknown condition %s" %
+                            (condition_parts[0],))
+        condition_function = condition_factory(*condition_parts[1:])
+        return condition_function
+
+    def build_rules(self, rules_config):
+        rules = []
+        for line in rules_config.split("\n"):
+            if not line:
+                continue
+
+            condition, target = line.split("=>")
+            condition_function = self.build_condition(condition)
+
+            target_parts = [
+                p.strip() for p in target.split(" ") if p]
+
+            target_factory = getattr(self,
+                "build_target_" + target_parts[0], None)
+
+            if target_factory is None:
+                raise ValueError("unknown target %s" %
+                                (target_parts[0],))
+
+            rules.append((
+                condition_function,
+                target_factory(*target_parts[1:])))
+        return rules
+
+    def build_condition_any(self):
+        """Returns a condition that always matches."""
+        return lambda *args: True
+
+    def build_condition_not(self, *args):
+        """
+        Returns a condition that negates the condition from its arguments.
+        """
+        other_condition = self.build_condition(" ".join(args))
+
+        def not_condition(metric_type, key, fields):
+            return not other_condition(metric_type, key, fields)
+        return not_condition
+
+    def build_condition_metric_type(self, *metric_types):
+        """Returns a condition that matched on metric kind."""
+        def metric_type_condition(metric_type, key, fields):
+            return (metric_type in metric_types)
+        return metric_type_condition
+
+    def build_condition_path_like(self, pattern):
+        def path_like_condition(metric_type, key, fields):
+            return fnmatch.fnmatch(key, pattern)
+        return path_like_condition
+
+    def build_target_drop(self):
+        """Returns a target that stops the processing of a message."""
+        def drop(*args):
+            return
+        return drop
+
+    def build_target_rewrite(self, pattern, repl, dup="no-dup"):
+        rexp = re.compile(pattern)
+
+        def rewrite_target(metric_type, key, fields):
+            if dup == "dup" and rexp.match(key) is not None:
+                yield metric_type, key, fields
+            key = rexp.sub(repl, key)
+            yield metric_type, key, fields
+
+        return rewrite_target
+
+    def build_target_set_metric_type(self, metric_type, dup="no-dup"):
+        def set_metric_type(_, key, fields):
+            if dup == "dup":
+                yield _, key, fields
+            yield metric_type, key, fields
+        return set_metric_type
+
+    def build_target_redirect_udp(self, host, port):
+        if self.service is None:
+            return lambda *args: True
+
+        port = int(port)
+        d = defer.Deferred()
+        self.ready.addCallback(lambda _: d)
+        protocol = UDPRedirectProtocol(host, port, lambda: d.callback(None))
+
+        client = UDPServer(0, protocol)
+        client.setServiceParent(self.service)
+
+        def redirect_udp_target(metric_type, key, fields):
+            message = self.rebuild_message(metric_type, key, fields)
+            protocol.write(message)
+            yield metric_type, key, fields
+        return redirect_udp_target
+
+    def build_target_redirect_tcp(self, host, port):
+        if self.service is None:
+            return lambda *args: True
+
+        port = int(port)
+        d = defer.Deferred()
+        self.ready.addCallback(lambda _: d)
+        factory = TCPRedirectClientFactory(lambda: d.callback(None))
+
+        redirect_service = TCPRedirectService(host, port, factory)
+        redirect_service.setServiceParent(self.service)
+
+        def redirect_tcp_target(metric_type, key, fields):
+            message = self.rebuild_message(metric_type, key, fields)
+            factory.write(message)
+            yield metric_type, key, fields
+        return redirect_tcp_target
+
+    def process_message(self, message, metric_type, key, fields):
+        metrics = [(metric_type, key, fields)]
+        if self.rules:
+            for condition, target in self.rules:
+                pending, metrics = metrics, []
+                if not pending:
+                    return
+                for metric_type, key, fields in pending:
+                    if not condition(metric_type, key, fields):
+                        metrics.append((metric_type, key, fields))
+                        continue
+                    result = target(metric_type, key, fields)
+                    if result is not None:
+                        metrics.extend(result)
+
+        for (metric_type, key, fields) in metrics:
+            message = self.rebuild_message(metric_type, key, fields)
+            self.message_processor.process_message(message, metric_type,
+                                                   key, fields)

=== added file 'txstatsd/service.py'
--- txstatsd/service.py	1970-01-01 00:00:00 +0000
+++ txstatsd/service.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,340 @@
+
+import getopt
+import sys
+import time
+import ConfigParser
+import platform
+import functools
+
+from twisted.application.internet import UDPServer, TCPServer
+from twisted.application.service import MultiService
+from twisted.python import usage, log
+from twisted.plugin import getPlugins
+
+from txstatsd.client import InternalClient
+from txstatsd.metrics.metrics import Metrics
+from txstatsd.metrics.extendedmetrics import ExtendedMetrics
+from txstatsd.server.processor import MessageProcessor
+from txstatsd.server.configurableprocessor import ConfigurableMessageProcessor
+from txstatsd.server.loggingprocessor import LoggingMessageProcessor
+from txstatsd.server.protocol import (
+    StatsDServerProtocol, StatsDTCPServerFactory)
+from txstatsd.server.router import Router
+from txstatsd.server import httpinfo
+from txstatsd.report import ReportingService
+from txstatsd.itxstatsd import IMetricFactory
+from twisted.application.service import Service
+from twisted.internet import task
+
+
+def accumulateClassList(classObj, attr, listObj,
+                        baseClass=None, excludeClass=None):
+    """Accumulate all attributes of a given name in a class hierarchy
+    into a single list.
+
+    Assuming all class attributes of this name are lists.
+    """
+    for base in classObj.__bases__:
+        accumulateClassList(base, attr, listObj, excludeClass=excludeClass)
+    if excludeClass != classObj:
+        if baseClass is None or baseClass in classObj.__bases__:
+            listObj.extend(classObj.__dict__.get(attr, []))
+
+
+class OptionsGlue(usage.Options):
+    """Extends usage.Options to also read parameters from a config file."""
+
+    optParameters = [
+        ["config", "c", None, "Config file to use."]]
+
+    def __init__(self):
+        parameters = []
+        accumulateClassList(self.__class__, 'optParameters',
+                            parameters, excludeClass=OptionsGlue)
+        for parameter in parameters:
+            if parameter[0] == "config" or parameter[1] == "c":
+                raise ValueError("the --config/-c parameter is reserved.")
+
+        self.overridden_options = []
+
+        super(OptionsGlue, self).__init__()
+
+    def opt_config(self, config_path):
+        self['config'] = config_path
+
+    opt_c = opt_config
+
+    def parseOptions(self, options=None):
+        """Obtain overridden options."""
+
+        if options is None:
+            options = sys.argv[1:]
+        try:
+            opts, args = getopt.getopt(options,
+                                       self.shortOpt, self.longOpt)
+        except getopt.error, e:
+            raise usage.UsageError(str(e))
+
+        for opt, arg in opts:
+            if opt[1] == '-':
+                opt = opt[2:]
+            else:
+                opt = opt[1:]
+            self.overridden_options.append(opt)
+
+        super(OptionsGlue, self).parseOptions(options=options)
+
+    def postOptions(self):
+        """Read the configuration file if one is provided."""
+        if self['config'] is not None:
+            config_file = ConfigParser.RawConfigParser()
+            config_file.read(self['config'])
+
+            self.configure(config_file)
+
+    def overridden_option(self, opt):
+        """Return whether this option was overridden."""
+        return opt in self.overridden_options
+
+    def configure(self, config_file):
+        """Read the configuration items, coercing types as required."""
+        for name, value in config_file.items(self.config_section):
+            self._coerce_option(name, value)
+
+        for section in sorted(config_file.sections()):
+            if section.startswith("plugin_"):
+                self[section] = config_file.items(section)
+            if section.startswith("carbon-cache"):
+                for name, value in config_file.items(section):
+                    self._coerce_option(name, value)
+
+    def _coerce_option(self, name, value):
+        """Coerce a single option, checking for overriden options."""
+        # Overridden options have precedence
+        if not self.overridden_option(name):
+            # Options appends '=' when gathering the parameters
+            if (name + '=') in self.longOpt:
+                # Coerce the type if required
+                if name in self._dispatch:
+                    if isinstance(self._dispatch[name], usage.CoerceParameter):
+                        value = self._dispatch[name].coerce(value)
+                    else:
+                        self._dispatch[name](name, value)
+                        return
+                self[name] = value
+
+
+class StatsDOptions(OptionsGlue):
+    """
+    The set of configuration settings for txStatsD.
+    """
+
+    optParameters = [
+        ["carbon-cache-host", "h", None,
+         "The host where carbon cache is listening.", str],
+        ["carbon-cache-port", "p", None,
+         "The port where carbon cache is listening.", int],
+        ["carbon-cache-name", "n", None,
+         "An identifier for the carbon-cache instance."],
+        ["listen-port", "l", 8125,
+         "The UDP port where we will listen.", int],
+        ["flush-interval", "i", 60000,
+         "The number of milliseconds between each flush.", int],
+        ["prefix", "x", None,
+         "Prefix to use when reporting stats.", str],
+        ["instance-name", "N", None,
+         "Instance name for our own stats reporting.", str],
+        ["report", "r", None,
+         "Which additional stats to report {process|net|io|system}.", str],
+        ["monitor-message", "m", "txstatsd ping",
+         "Message we expect from monitoring agent.", str],
+        ["monitor-response", "o", "txstatsd pong",
+         "Response we should send monitoring agent.", str],
+        ["statsd-compliance", "s", 1,
+         "Produce StatsD-compliant messages.", int],
+        ["dump-mode", "d", 0,
+         "Dump received and aggregated metrics"
+             " before passing them to carbon.", int],
+        ["routing", "g", "",
+         "Routing rules", str],
+        ["listen-tcp-port", "t", None,
+         "The TCP port where we will listen.", int],
+        ["max-queue-size", "Q", 20000,
+         "Maximum send queue size per destination.", int],
+        ["max-datapoints-per-message", "M", 1000,
+         "Maximum datapoints per message to carbon-cache.", int],
+        ["http-port", "P", None,
+         "The httpinfo port.", int],
+        ]
+
+    def __init__(self):
+        self.config_section = 'statsd'
+        super(StatsDOptions, self).__init__()
+        self["carbon-cache-host"] = []
+        self["carbon-cache-port"] = []
+        self["carbon-cache-name"] = []
+
+    def opt_carbon_cache_host(self, host):
+        self["carbon-cache-host"].append(host)
+
+    def opt_carbon_cache_port(self, port):
+        self["carbon-cache-port"].append(usage.portCoerce(port))
+
+    def opt_carbon_cache_name(self, name):
+        self["carbon-cache-name"].append(name)
+
+
+class StatsDService(Service):
+
+    def __init__(self, carbon_client, processor, flush_interval, clock=None):
+        self.carbon_client = carbon_client
+        self.processor = processor
+        self.flush_interval = flush_interval
+        self.flush_task = task.LoopingCall(self.flushProcessor)
+        if clock is not None:
+            self.flush_task.clock = clock
+
+    def flushProcessor(self):
+        """Flush messages queued in the processor to Graphite."""
+        flushed = 0
+        start = time.time()
+        for metric, value, timestamp in self.processor.flush(
+                interval=self.flush_interval):
+            self.carbon_client.sendDatapoint(metric, (timestamp, value))
+            flushed += 1
+        log.msg("Flushed total %d metrics in %.6f" %
+                (flushed, time.time() - start))
+
+    def startService(self):
+        self.flush_task.start(self.flush_interval / 1000, False)
+
+    def stopService(self):
+        if self.flush_task.running:
+            self.flush_task.stop()
+
+
+def report_client_manager_stats():
+    from carbon.instrumentation import stats
+
+    current_stats = stats.copy()
+    for name, value in list(current_stats.items()):
+        del current_stats[name]
+        if name.startswith("destinations"):
+            current_stats[name.replace(":", "_")] = value
+        stats[name] = 0
+    return current_stats
+
+
+def createService(options):
+    """Create a txStatsD service."""
+    from carbon.routers import ConsistentHashingRouter
+    from carbon.client import CarbonClientManager
+    from carbon.conf import settings
+
+    settings.MAX_QUEUE_SIZE = options["max-queue-size"]
+    settings.MAX_DATAPOINTS_PER_MESSAGE = options["max-datapoints-per-message"]
+
+    root_service = MultiService()
+    root_service.setName("statsd")
+
+    prefix = options["prefix"]
+    if prefix is None:
+        prefix = "statsd"
+
+    instance_name = options["instance-name"]
+    if not instance_name:
+        instance_name = platform.node()
+
+    # initialize plugins
+    plugin_metrics = []
+    for plugin in getPlugins(IMetricFactory):
+        plugin.configure(options)
+        plugin_metrics.append(plugin)
+
+    processor = None
+    if options["dump-mode"]:
+        # LoggingMessageProcessor supersedes
+        #  any other processor class in "dump-mode"
+        assert not hasattr(log, 'info')
+        log.info = log.msg # for compatibility with LMP logger interface
+        processor = functools.partial(LoggingMessageProcessor, logger=log)
+
+    if options["statsd-compliance"]:
+        processor = (processor or MessageProcessor)(plugins=plugin_metrics)
+        input_router = Router(processor, options['routing'], root_service)
+        connection = InternalClient(input_router)
+        metrics = Metrics(connection)
+    else:
+        processor = (processor or ConfigurableMessageProcessor)(
+            message_prefix=prefix,
+            internal_metrics_prefix=prefix + "." + instance_name + ".",
+            plugins=plugin_metrics)
+        input_router = Router(processor, options['routing'], root_service)
+        connection = InternalClient(input_router)
+        metrics = ExtendedMetrics(connection)
+
+    if not options["carbon-cache-host"]:
+        options["carbon-cache-host"].append("127.0.0.1")
+    if not options["carbon-cache-port"]:
+        options["carbon-cache-port"].append(2004)
+    if not options["carbon-cache-name"]:
+        options["carbon-cache-name"].append(None)
+
+    reporting = ReportingService(instance_name)
+    reporting.setServiceParent(root_service)
+
+    # Schedule updates for those metrics expecting to be
+    # periodically updated, for example the meter metric.
+    reporting.schedule(processor.update_metrics, 10, None)
+    reporting.schedule(report_client_manager_stats,
+                       options["flush-interval"] / 1000,
+                       metrics.gauge)
+
+    if options["report"] is not None:
+        from txstatsd import process
+        from twisted.internet import reactor
+
+        reporting.schedule(
+            process.report_reactor_stats(reactor), 60, metrics.gauge)
+        reports = [name.strip() for name in options["report"].split(",")]
+        for report_name in reports:
+            for reporter in getattr(process, "%s_STATS" %
+                                    report_name.upper(), ()):
+                reporting.schedule(reporter, 60, metrics.gauge)
+
+    # XXX Make this configurable.
+    router = ConsistentHashingRouter()
+    carbon_client = CarbonClientManager(router)
+    carbon_client.setServiceParent(root_service)
+
+    for host, port, name in zip(options["carbon-cache-host"],
+                                options["carbon-cache-port"],
+                                options["carbon-cache-name"]):
+        carbon_client.startClient((host, port, name))
+
+    statsd_service = StatsDService(carbon_client, input_router,
+                                   options["flush-interval"])
+    statsd_service.setServiceParent(root_service)
+
+    statsd_server_protocol = StatsDServerProtocol(
+        input_router,
+        monitor_message=options["monitor-message"],
+        monitor_response=options["monitor-response"])
+
+    listener = UDPServer(options["listen-port"], statsd_server_protocol)
+    listener.setServiceParent(root_service)
+
+    if options["listen-tcp-port"] is not None:
+        statsd_tcp_server_factory = StatsDTCPServerFactory(
+                input_router,
+                monitor_message=options["monitor-message"],
+                monitor_response=options["monitor-response"])
+
+        listener = TCPServer(options["listen-tcp-port"],
+            statsd_tcp_server_factory)
+        listener.setServiceParent(root_service)
+
+    httpinfo_service = httpinfo.makeService(options, processor, statsd_service)
+    httpinfo_service.setServiceParent(root_service)
+
+    return root_service

=== added directory 'txstatsd/stats'
=== added file 'txstatsd/stats/__init__.py'
=== added file 'txstatsd/stats/ewma.py'
--- txstatsd/stats/ewma.py	1970-01-01 00:00:00 +0000
+++ txstatsd/stats/ewma.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,70 @@
+
+"""
+An exponentially-weighted moving average.
+
+See:
+- U{UNIX Load Average Part 1: How It Works
+    <http://www.teamquest.com/pdfs/whitepaper/ldavg1.pdf>}
+- U{UNIX Load Average Part 2: Not Your Average Average
+    <http://www.teamquest.com/pdfs/whitepaper/ldavg2.pdf>}
+"""
+
+import math
+
+
+class Ewma(object):
+    M1_ALPHA  = 1 - math.exp(-5 / 60.0)
+    M5_ALPHA  = 1 - math.exp(-5 / 60.0 / 5)
+    M15_ALPHA = 1 - math.exp(-5 / 60.0 / 15)
+
+    @classmethod
+    def one_minute_ewma(cls):
+        """
+        Creates a new C{Ewma} which is equivalent to the UNIX one minute
+        load average and which expects to be ticked every 5 seconds.
+        """
+        return Ewma(Ewma.M1_ALPHA, 5)
+
+    @classmethod
+    def five_minute_ewma(cls):
+        """
+        Creates a new C{Ewma} which is equivalent to the UNIX five minute
+        load average and which expects to be ticked every 5 seconds.
+        """
+        return Ewma(Ewma.M5_ALPHA, 5)
+
+    @classmethod
+    def fifteen_minute_ewma(cls):
+        """
+        Creates a new C{Ewma} which is equivalent to the UNIX fifteen
+        minute load average and which expects to be ticked every 5 seconds.
+        """
+        return Ewma(Ewma.M15_ALPHA, 5)
+
+    def __init__(self, alpha, interval):
+        """Create a new C{Ewma} with a specific smoothing constant.
+        
+        @param alpha: The smoothing constant.
+        @param interval: The expected tick interval in seconds.
+        """
+        self.interval = interval
+        self.alpha = float(alpha)
+
+        self.initialized = False
+        self.rate = 0.0
+        self.uncounted = 0
+
+    def update(self, n):
+        """Update the moving average with a new value."""
+        self.uncounted += n
+
+    def tick(self):
+        """Mark the passage of time and decay the current rate accordingly."""
+        count = self.uncounted
+        self.uncounted = 0
+        instant_rate = float(count) / self.interval
+        if self.initialized:
+            self.rate += (self.alpha * (instant_rate  - self.rate))
+        else:
+            self.rate = instant_rate
+            self.initialized = True

=== added file 'txstatsd/stats/exponentiallydecayingsample.py'
--- txstatsd/stats/exponentiallydecayingsample.py	1970-01-01 00:00:00 +0000
+++ txstatsd/stats/exponentiallydecayingsample.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,124 @@
+import bisect
+import math
+import random
+import time
+
+
+class ExponentiallyDecayingSample(object):
+    """
+    An exponentially-decaying random sample of values. Uses Cormode et
+    al's forward-decaying priority reservoir sampling method to produce a
+    statistically representative sample, exponentially biased towards newer
+    entries.
+
+    See:
+    - U{Cormode et al. Forward Decay: A Practical Time Decay Model for
+      Streaming Systems. ICDE '09: Proceedings of the 2009 IEEE International
+      Conference on Data Engineering (2009)
+      <http://www.research.att.com/people/Cormode_Graham/
+      library/publications/CormodeShkapenyukSrivastavaXu09.pdf>}
+    """
+
+    # 10 minutes (in seconds)
+    RESCALE_THRESHOLD = 60 * 10
+
+    def __init__(self, reservoir_size, alpha, wall_time=None):
+        """Creates a new C{ExponentiallyDecayingSample}.
+
+        @param reservoir_size: The number of samples to keep in the sampling
+            reservoir.
+        @parama alpha: The exponential decay factor; the higher this is,
+            the more biased the sample will be towards newer values.
+        """
+        self._values = []
+        self.alpha = alpha
+        self.reservoir_size = reservoir_size
+
+        self.count = 0
+        self.start_time = 0
+        self.next_scale_time = 0
+
+        if wall_time is None:
+            wall_time = time.time
+        self._wall_time = wall_time
+        self.clear()
+
+    def clear(self):
+        self._values = []
+        self.count = 0
+        self.start_time = self.tick()
+        self.next_scale_time = (
+            self._wall_time() + self.RESCALE_THRESHOLD)
+
+    def size(self):
+        return min(self.reservoir_size, self.count)
+
+    def update(self, value, timestamp=None):
+        """Adds an old value with a fixed timestamp to the sample.
+
+        @param value: The value to be added.
+        @param timestamp: The epoch timestamp of *value* in seconds.
+        """
+
+        now = self._wall_time()
+        next = self.next_scale_time
+        if now >= next:
+            self.rescale(now, next)
+
+        if timestamp is None:
+            timestamp = self.tick()
+
+        priority = self.weight(timestamp - self.start_time) / random.random()
+        self.count += 1
+        new_count = self.count
+        if new_count <= self.reservoir_size:
+            bisect.insort(self._values, (priority, value))
+        else:
+            first = self._values[0][0]
+
+            if first < priority:
+                bisect.insort(self._values, (priority, value))
+                self._values = self._values[1:]
+
+    def get_values(self):
+        return [v for (k, v) in self._values]
+
+    def tick(self):
+        return self._wall_time()
+
+    def weight(self, t):
+        return math.exp(self.alpha * t)
+
+    def rescale(self, now, next):
+        """
+        A common feature of the above techniques - indeed, the key technique
+        that allows us to track the decayed weights efficiently - is that they
+        maintain counts and other quantities based on g(ti - L), and only
+        scale by g(t - L) at query time. But while g(ti - L)/g(t-L) is
+        guaranteed to lie between zero and one, the intermediate values of
+        g(ti - L) could become very large. For polynomial functions, these
+        values should not grow too large, and should be effectively
+        represented in practice by floating point values without loss of
+        precision. For exponential functions, these values could grow quite
+        large as new values of (ti - L) become large, and potentially exceed
+        the capacity of common floating point types. However, since the values
+        stored by the algorithms are linear combinations of g values (scaled
+        sums), they can be rescaled relative to a new landmark. That is, by
+        the analysis of exponential decay in Section III-A, the choice of L
+        does not affect the final result. We can therefore multiply each value
+        based on L by a factor of exp(-alpha(L' - L)), and obtain the correct
+        value as if we had instead computed relative to a new landmark L' (and
+        then use this new L' at query time). This can be done with a linear
+        pass over whatever data structure is being used.
+        """
+
+        self.next_scale_time = (
+            now + self.RESCALE_THRESHOLD)
+        old_start_time = self.start_time
+        self.start_time = self.tick()
+
+        new_values = []
+        for k, v in self._values:
+            nk = k * math.exp(-self.alpha * (self.start_time - old_start_time))
+            new_values.append((nk, v))
+        self._values = new_values

=== added file 'txstatsd/stats/uniformsample.py'
--- txstatsd/stats/uniformsample.py	1970-01-01 00:00:00 +0000
+++ txstatsd/stats/uniformsample.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,45 @@
+
+import random
+import sys
+
+
+class UniformSample(object):
+    """
+    A random sample of a stream of values. Uses Vitter's Algorithm R to
+    produce a statistically representative sample.
+    
+    See:
+    - U{Random Sampling with a Reservoir
+        <http://www.cs.umd.edu/~samir/498/vitter.pdf>}
+    """
+
+    def __init__(self, reservoir_size):
+        """Creates a new C{UniformSample}.
+
+        @param reservoir_size: The number of samples to keep in the sampling
+            reservoir.
+        """
+        self._values = [0 for i in range(reservoir_size)]
+        self._count = 0
+        self.clear()
+    
+    def clear(self):
+        self._values = [0 for i in range(len(self._values))]
+        self._count = 0
+
+    def size(self):
+        c = self._count
+        return len(self._values) if c > len(self._values) else c
+
+    def update(self, value):
+        self._count += 1
+        if self._count <= len(self._values):
+            self._values[self._count - 1] = value
+        else:
+            r = random.randint(1, sys.maxint) % self._count
+            if r < len(self._values):
+                self._values[r] = value
+
+    def get_values(self):
+        s = self.size()
+        return [self._values[i] for i in range(0, s)]

=== added directory 'txstatsd/tests'
=== added file 'txstatsd/tests/__init__.py'
=== added file 'txstatsd/tests/helper.py'
--- txstatsd/tests/helper.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/helper.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,18 @@
+from txstatsd.metrics.metrics import Metrics
+
+class FakeStatsDClient(object):
+    """A fake C{StatsDClient} that simply appends to metrics.data on write."""
+
+    def __init__(self, metrics):
+        self.metrics = metrics
+
+    def write(self, data):
+        self.metrics.data.append(data)
+
+
+class FakeMetrics(Metrics):
+    """A fake C{IMeter} that simply stores metrics locally."""
+
+    def __init__(self, namespace=""):
+        Metrics.__init__(self, FakeStatsDClient(self), namespace=namespace)
+        self.data = []

=== added directory 'txstatsd/tests/metrics'
=== added file 'txstatsd/tests/metrics/__init__.py'
--- txstatsd/tests/metrics/__init__.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/metrics/__init__.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,2 @@
+
+

=== added file 'txstatsd/tests/metrics/test_derive.py'
--- txstatsd/tests/metrics/test_derive.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/metrics/test_derive.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,77 @@
+import itertools as it, operator as op, functools as ft
+
+import random
+
+from twisted.trial.unittest import TestCase
+from twisted.plugin import getPlugins
+from twisted.plugins import derive_plugin
+from txstatsd.itxstatsd import IMetricFactory
+
+
+class TestDeriveMetricReporter(TestCase):
+
+    mt = derive_plugin.DeriveMetricFactory.metric_type
+
+
+    def test_fastpoll(self):
+        random.seed(1)
+
+        wall_time = random.randint(1, 1000)
+        reporter = derive_plugin.DeriveMetricReporter(
+            'test', wall_time_func=lambda: wall_time )
+        self.assertEquals(reporter.flush(random.randint(1, 1000), wall_time), list())
+
+        for j in xrange(random.randint(1, 100)):
+            reporter.process([random.random() * 1e3, self.mt])
+        self.assertEquals(reporter.flush(random.randint(1, 1000), wall_time), list())
+
+
+    def test_interface(self):
+        random.seed(1)
+
+        wall_time = random.random() * 1e6
+        reporter = derive_plugin.DeriveMetricReporter(
+            'test', prefix='some.prefix', wall_time_func=lambda: wall_time )
+
+        count = random.randint(1, 100)
+        for j in xrange(count):
+            wall_time += 1
+            reporter.process([random.random() * 1e3, self.mt])
+
+        ts = random.random() * 1e6
+        values = reporter.flush(count, ts)
+        self.assertEquals(len(values), 2)
+        self.assertEquals(set(it.imap(len, values)), {3})
+        self.assertEquals(set(it.imap(op.itemgetter(2), values)), {ts})
+        self.assertEquals(
+            set(it.imap(op.itemgetter(0), values)),
+            {'some.prefix.test.count', 'some.prefix.test.rate'} )
+
+
+    def test_values(self):
+        random.seed(1)
+
+        wall_time = 0
+        reporter = derive_plugin.DeriveMetricReporter(
+            'test', wall_time_func=lambda: wall_time )
+
+        val_count = 0
+        for i in xrange(100):
+            rate = (random.random() - 0.5) * 1e3
+            count = random.randint(2, 100)
+
+            for j in xrange(1, count):
+                wall_time += 1
+                reporter.process([rate, self.mt])
+                val_count += rate
+
+            values = dict(it.imap(op.itemgetter(0, 1), reporter.flush(count, wall_time)))
+            self.assertTrue(abs(values['test.count'] - val_count) < count**-1)
+            self.assertTrue(abs(values['test.rate'] - rate) < count**-1)
+
+
+class TestPlugin(TestCase):
+
+    def test_factory(self):
+        self.assertTrue(derive_plugin.derive_metric_factory in \
+                        list(getPlugins(IMetricFactory)))

=== added file 'txstatsd/tests/metrics/test_distinct.py'
--- txstatsd/tests/metrics/test_distinct.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/metrics/test_distinct.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,94 @@
+# Copyright (C) 2011 Canonical
+# All Rights Reserved
+import random
+
+from scipy.stats import chi2
+
+from twisted.trial.unittest import TestCase
+from twisted.plugin import getPlugins
+from twisted.plugins import distinct_plugin
+import txstatsd.metrics.distinctmetric as distinct
+from txstatsd.itxstatsd import IMetricFactory
+
+
+class TestHash(TestCase):
+
+    def test_hash_chars(self):
+        "For one table, all chars map to different chars"
+        results = set()
+        for c in range(256):
+            random.seed(1)
+            h = distinct.hash(chr(c))
+            results.add(h)
+        self.assertEquals(len(results), 256)
+
+    def test_chi_square(self):
+        N = 10000
+
+        for (bits, buckets) in [(-1, 1024), (24, 256),
+                                (16, 256), (8, 256), (0, 256)]:
+            bins = [0] * buckets
+            for i in range(N):
+                v = distinct.hash(str(i))
+                if bits < 0:
+                    bin = v / (0xFFFFFFFF / buckets)
+                else:
+                    bin = (v >> bits) & 0xFF
+                bins[bin] += 1
+            value = sum(((x - N / buckets) ** 2) / (N / buckets) for x in bins)
+            pval = chi2.cdf(value, N)
+            if pval > 0.5:
+                print bins, pval
+            self.assertTrue(pval < 0.5, "bits %s, pval == %s" % (bits, pval))
+    test_chi_square.skip = "Takes too long to run every time."
+
+
+class TestZeros(TestCase):
+
+    def test_zeros(self):
+        self.assertEquals(distinct.zeros(1), 0)
+        self.assertEquals(distinct.zeros(2), 1)
+        self.assertEquals(distinct.zeros(4), 2)
+        self.assertEquals(distinct.zeros(5), 0)
+        self.assertEquals(distinct.zeros(8), 3)
+        self.assertEquals(distinct.zeros(9), 0)
+
+
+class TestDistinct(TestCase):
+
+    def test_all(self):
+        random.seed(1)
+
+        for r in [1000, 10000]:
+            cd = distinct.SlidingDistinctCounter(32, 32)
+            for i in range(r):
+                cd.add(1, str(i))
+            error = abs(cd.distinct() - r)
+            self.assertTrue(error < 0.15 * r)
+
+
+class TestDistinctMetricReporter(TestCase):
+
+    def test_reports(self):
+        random.seed(1)
+        _wall_time = [0]
+        def _time():
+            return _wall_time[0]
+
+        dmr = distinct.DistinctMetricReporter("test", wall_time_func=_time)
+        for i in range(3000):
+            _wall_time[0] = i * 50
+            dmr.update(str(i))
+        now = _time()
+        self.assertTrue(abs(dmr.count() - 3000) < 600)
+        self.assertTrue(abs(dmr.count_1min(now) - 1) < 2)
+        self.assertTrue(abs(dmr.count_1hour(now) - 72) < 15)
+        self.assertTrue(abs(dmr.count_1day(now) - 1728) < 500)
+
+
+class TestPlugin(TestCase):
+    
+    def test_factory(self):
+        self.assertTrue(distinct_plugin.distinct_metric_factory in \
+                        list(getPlugins(IMetricFactory)))
+

=== added file 'txstatsd/tests/metrics/test_histogrammetric.py'
--- txstatsd/tests/metrics/test_histogrammetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/metrics/test_histogrammetric.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,91 @@
+
+import math
+from unittest import TestCase
+
+from txstatsd.metrics.histogrammetric import HistogramMetricReporter
+from txstatsd.stats.uniformsample import UniformSample
+
+
+class TestHistogramReporterMetric(TestCase):
+
+    def test_histogram_with_zero_recorded_values(self):
+        sample = UniformSample(100)
+        histogram = HistogramMetricReporter(sample)
+
+        self.assertEqual(histogram.count, 0, 'Should have a count of 0')
+        self.assertEqual(histogram.max(), 0,
+                         'Should have a max of 0')
+        self.assertEqual(histogram.min(), 0,
+                         'Should have a min of 0')
+
+        self.assertEqual(histogram.mean(), 0,
+                         'Should have a mean of 0')
+
+        self.assertEqual(histogram.std_dev(), 0,
+                         'Should have a standard deviation of 0')
+
+        percentiles = histogram.percentiles(0.5, 0.75, 0.99)
+        self.assertTrue(
+            (math.fabs(percentiles[0] - 0) < 0.01),
+            'Should calculate percentiles')
+        self.assertTrue(
+            (math.fabs(percentiles[1] - 0) < 0.01),
+            'Should calculate percentiles')
+        self.assertTrue(
+            (math.fabs(percentiles[2] - 0) < 0.01),
+            'Should calculate percentiles')
+
+        self.assertEqual(len(histogram.get_values()), 0,
+                         'Should have no values')
+
+    def test_histogram_of_numbers_1_through_10000(self):
+        sample = UniformSample(100000)
+        histogram = HistogramMetricReporter(sample)
+        for i in range(1, 10001):
+            histogram.update(i)
+
+        self.assertEqual(histogram.count, 10000,
+                         'Should have a count of 10000')
+
+        self.assertEqual(histogram.max(), 10000,
+                         'Should have a max of 10000')
+        self.assertEqual(histogram.min(), 1,
+                         'Should have a min of 1')
+
+        self.assertTrue(
+            (math.fabs(histogram.mean() - 5000.5) < 0.01),
+            'Should have a mean value of 5000.5')
+
+        self.assertTrue(
+            (math.fabs(histogram.std_dev() - 2886.89) < 0.01),
+            'Should have a standard deviation of X')
+
+        percentiles = histogram.percentiles(0.5, 0.75, 0.99)
+        self.assertTrue(
+            (math.fabs(percentiles[0] - 5000.5) < 0.01),
+            'Should calculate percentiles')
+        self.assertTrue(
+            (math.fabs(percentiles[1] - 7500.75) < 0.01),
+            'Should calculate percentiles')
+        self.assertTrue(
+            (math.fabs(percentiles[2] - 9900.99) < 0.01),
+            'Should calculate percentiles')
+
+        values = [i for i in range(1, 10001)]
+        self.assertEqual(histogram.get_values(), values,
+                         'Should have 10000 values')
+
+    def test_histogram_histogram(self):
+        sample = UniformSample(100000)
+        histogram = HistogramMetricReporter(sample)
+        for i in range(1001, 11001):
+            histogram.update(i)
+
+        hist = histogram.histogram()
+        self.assertEquals(sum(hist), 10000)
+
+        total = sum(hist)
+        binsize = int(total / len(hist))
+        for i in hist:
+            self.assertTrue(abs(i - binsize) <= 1)
+

=== added file 'txstatsd/tests/metrics/test_timermetric.py'
--- txstatsd/tests/metrics/test_timermetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/metrics/test_timermetric.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,139 @@
+
+import math
+
+from twisted.trial.unittest import TestCase
+
+from txstatsd.metrics.timermetric import TimerMetricReporter
+
+
+class TestBlankTimerMetric(TestCase):
+    def setUp(self):
+        self.timer = TimerMetricReporter('test')
+        self.timer.tick()
+
+    def test_max(self):
+        self.assertEqual(
+            self.timer.max(), 0,
+            'Should have a max of zero')
+
+    def test_min(self):
+        self.assertEqual(
+            self.timer.min(), 0,
+            'Should have a min of zero')
+
+    def test_mean(self):
+        self.assertEqual(
+            self.timer.max(), 0,
+            'Should have a mean of zero')
+
+    def test_count(self):
+        self.assertEqual(
+            self.timer.count(), 0,
+            'Should have a count of zero')
+
+    def test_std_dev(self):
+        self.assertEqual(
+            self.timer.std_dev(), 0,
+            'Should have a standard deviation of zero')
+
+    def test_percentiles(self):
+        percentiles = self.timer.percentiles(0.5, 0.95, 0.98, 0.99, 0.999)
+        self.assertEqual(
+            percentiles[0], 0,
+            'Should have median of zero')
+        self.assertEqual(
+            percentiles[1], 0,
+            'Should have p95 of zero')
+        self.assertEqual(
+            percentiles[2], 0,
+            'Should have p98 of zero')
+        self.assertEqual(
+            percentiles[3], 0,
+            'Should have p99 of zero')
+        self.assertEqual(
+            percentiles[4], 0,
+            'Should have p99.9 of zero')
+
+    def test_mean_rate(self):
+        self.assertEqual(
+            self.timer.mean_rate(), 0,
+            'Should have a mean rate of zero')
+
+    def test_one_minute_rate(self):
+        self.assertEqual(
+            self.timer.one_minute_rate(), 0,
+            'Should have a one-minute rate of zero`')
+
+    def test_five_minute_rate(self):
+        self.assertEqual(
+            self.timer.five_minute_rate(), 0,
+            'Should have a five-minute rate of zero')
+
+    def test_fifteen_minute_rate(self):
+        self.assertEqual(
+            self.timer.fifteen_minute_rate(), 0,
+            'Should have a fifteen-minute rate of zero')
+
+    def test_no_values(self):
+        self.assertEqual(
+            len(self.timer.get_values()), 0,
+            'Should have no values')
+
+
+class TestTimingSeriesEvents(TestCase):
+    def setUp(self):
+        self.timer = TimerMetricReporter('test')
+        self.timer.tick()
+        self.timer.update(10)
+        self.timer.update(20)
+        self.timer.update(20)
+        self.timer.update(30)
+        self.timer.update(40)
+
+    def test_count(self):
+        self.assertEqual(
+            self.timer.count(), 5,
+            'Should record the count')
+
+    def test_min(self):
+        self.assertTrue(
+            (math.fabs(self.timer.min() - 10.0) < 0.001),
+            'Should calculate the minimum duration')
+
+    def test_max(self):
+        self.assertTrue(
+            (math.fabs(self.timer.max() - 40.0) < 0.001),
+            'Should calculate the maximum duration')
+
+    def test_mean(self):
+        self.assertTrue(
+            (math.fabs(self.timer.mean() - 24.0) < 0.001),
+            'Should calculate the mean duration')
+
+    def test_std_dev(self):
+        self.assertTrue(
+            (math.fabs(self.timer.std_dev() - 11.401) < 0.001),
+            'Should calculate the standard deviation')
+
+    def test_percentiles(self):
+        percentiles = self.timer.percentiles(0.5, 0.95, 0.98, 0.99, 0.999)
+        self.assertTrue(
+            (math.fabs(percentiles[0] - 20.0) < 0.001),
+            'Should calculate the median')
+        self.assertTrue(
+            (math.fabs(percentiles[1] - 40.0) < 0.001),
+            'Should calculate the p95')
+        self.assertTrue(
+            (math.fabs(percentiles[2] - 40.0) < 0.001),
+            'Should calculate the p98')
+        self.assertTrue(
+            (math.fabs(percentiles[3] - 40.0) < 0.001),
+            'Should calculate the p99')
+        self.assertTrue(
+            (math.fabs(percentiles[4] - 40.0) < 0.001),
+            'Should calculate the p999')
+
+    def test_values(self):
+        self.assertEqual(
+            set(self.timer.get_values()), set([10, 20, 20, 30, 40]),
+            'Should have a series of values')

=== added directory 'txstatsd/tests/stats'
=== added file 'txstatsd/tests/stats/__init__.py'
=== added file 'txstatsd/tests/stats/test_ewma.py'
--- txstatsd/tests/stats/test_ewma.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/stats/test_ewma.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,315 @@
+
+import math
+from unittest import TestCase
+
+from txstatsd.stats.ewma import Ewma
+
+
+def mark_minutes(minutes, ewma):
+    for i in range(1, minutes * 60, 5):
+        ewma.tick()
+
+class TestEwmaOneMinute(TestCase):
+    def setUp(self):
+        self.ewma = Ewma.one_minute_ewma()
+        self.ewma.update(3)
+        self.ewma.tick()
+
+    def test_first_tick(self):
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.6) < 0.000001),
+            'Should have a rate of 0.6 events/sec after the first tick')
+
+    def test_one_minute(self):
+        mark_minutes(1, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.22072766) < 0.00000001),
+            'Should have a rate of 0.22072766 events/sec after 1 minute')
+
+    def test_two_minutes(self):
+        mark_minutes(2, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.08120117) < 0.00000001),
+            'Should have a rate of 0.08120117 events/sec after 2 minutes')
+
+    def test_three_minutes(self):
+        mark_minutes(3, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.02987224) < 0.00000001),
+            'Should have a rate of 0.02987224 events/sec after 3 minutes')
+
+    def test_four_minutes(self):
+        mark_minutes(4, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.01098938) < 0.00000001),
+            'Should have a rate of 0.01098938 events/sec after 4 minutes')
+
+    def test_five_minutes(self):
+        mark_minutes(5, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00404277) < 0.00000001),
+            'Should have a rate of 0.00404277 events/sec after 5 minutes')
+
+    def test_six_minutes(self):
+        mark_minutes(6, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00148725) < 0.00000001),
+            'Should have a rate of 0.00148725 events/sec after 6 minutes')
+
+    def test_seven_minutes(self):
+        mark_minutes(7, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00054713) < 0.00000001),
+            'Should have a rate of 0.00054713 events/sec after 7 minutes')
+
+    def test_eight_minutes(self):
+        mark_minutes(8, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00020128) < 0.00000001),
+            'Should have a rate of 0.00020128 events/sec after 8 minutes')
+
+    def test_nine_minutes(self):
+        mark_minutes(9, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00007405) < 0.00000001),
+            'Should have a rate of 0.00007405 events/sec after 9 minutes')
+
+    def test_ten_minutes(self):
+        mark_minutes(10, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00002724) < 0.00000001),
+            'Should have a rate of 0.00002724 events/sec after 10 minutes')
+
+    def test_eleven_minutes(self):
+        mark_minutes(11, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00001002) < 0.00000001),
+            'Should have a rate of 0.00001002 events/sec after 11 minutes')
+
+    def test_twelve_minutes(self):
+        mark_minutes(12, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00000369) < 0.00000001),
+            'Should have a rate of 0.00000369 events/sec after 12 minutes')
+
+    def test_thirteen_minutes(self):
+        mark_minutes(13, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00000136) < 0.00000001),
+            'Should have a rate of 0.00000136 events/sec after 13 minutes')
+
+    def test_fourteen_minutes(self):
+        mark_minutes(14, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00000050) < 0.00000001),
+            'Should have a rate of 0.00000050 events/sec after 14 minutes')
+
+    def test_fifteen_minutes(self):
+        mark_minutes(15, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.00000018) < 0.00000001),
+            'Should have a rate of 0.00000018 events/sec after 15 minutes')
+
+
+class TestEwmaFiveMinute(TestCase):
+    def setUp(self):
+        self.ewma = Ewma.five_minute_ewma()
+        self.ewma.update(3)
+        self.ewma.tick()
+
+    def test_first_tick(self):
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.6) < 0.000001),
+            'Should have a rate of 0.6 events/sec after the first tick')
+
+    def test_one_minute(self):
+        mark_minutes(1, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.49123845) < 0.00000001),
+            'Should have a rate of 0.49123845 events/sec after 1 minute')
+
+    def test_two_minutes(self):
+        mark_minutes(2, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.40219203) < 0.00000001),
+            'Should have a rate of 0.40219203 events/sec after 2 minutes')
+
+    def test_three_minutes(self):
+        mark_minutes(3, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.32928698) < 0.00000001),
+            'Should have a rate of 0.32928698 events/sec after 3 minutes')
+
+    def test_four_minutes(self):
+        mark_minutes(4, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.26959738) < 0.00000001),
+            'Should have a rate of 0.26959738 events/sec after 4 minutes')
+
+    def test_five_minutes(self):
+        mark_minutes(5, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.22072766) < 0.00000001),
+            'Should have a rate of 0.22072766 events/sec after 5 minutes')
+
+    def test_six_minutes(self):
+        mark_minutes(6, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.18071653) < 0.00000001),
+            'Should have a rate of 0.18071653 events/sec after 6 minutes')
+
+    def test_seven_minutes(self):
+        mark_minutes(7, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.14795818) < 0.00000001),
+            'Should have a rate of 0.14795818 events/sec after 7 minutes')
+
+    def test_eight_minutes(self):
+        mark_minutes(8, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.12113791) < 0.00000001),
+            'Should have a rate of 0.12113791 events/sec after 8 minutes')
+
+    def test_nine_minutes(self):
+        mark_minutes(9, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.09917933) < 0.00000001),
+            'Should have a rate of 0.09917933 events/sec after 9 minutes')
+
+    def test_ten_minutes(self):
+        mark_minutes(10, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.08120117) < 0.00000001),
+            'Should have a rate of 0.08120117 events/sec after 10 minutes')
+
+    def test_eleven_minutes(self):
+        mark_minutes(11, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.06648190) < 0.00000001),
+            'Should have a rate of 0.06648190 events/sec after 11 minutes')
+
+    def test_twelve_minutes(self):
+        mark_minutes(12, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.05443077) < 0.00000001),
+            'Should have a rate of 0.05443077 events/sec after 12 minutes')
+
+    def test_thirteen_minutes(self):
+        mark_minutes(13, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.04456415) < 0.00000001),
+            'Should have a rate of 0.04456415 events/sec after 13 minutes')
+
+    def test_fourteen_minutes(self):
+        mark_minutes(14, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.03648604) < 0.00000001),
+            'Should have a rate of 0.03648604 events/sec after 14 minutes')
+
+    def test_fifteen_minutes(self):
+        mark_minutes(15, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.02987224) < 0.00000001),
+            'Should have a rate of 0.02987224 events/sec after 15 minutes')
+
+
+class TestEwmaFifteenMinute(TestCase):
+    def setUp(self):
+        self.ewma = Ewma.fifteen_minute_ewma()
+        self.ewma.update(3)
+        self.ewma.tick()
+
+    def test_first_tick(self):
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.6) < 0.000001),
+            'Should have a rate of 0.6 events/sec after the first tick')
+
+    def test_one_minute(self):
+        mark_minutes(1, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.56130419) < 0.00000001),
+            'Should have a rate of 0.56130419 events/sec after 1 minute')
+
+    def test_two_minutes(self):
+        mark_minutes(2, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.52510399) < 0.00000001),
+            'Should have a rate of 0.52510399 events/sec after 2 minutes')
+
+    def test_three_minutes(self):
+        mark_minutes(3, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.49123845) < 0.00000001),
+            'Should have a rate of 0.49123845 events/sec after 3 minutes')
+
+    def test_four_minutes(self):
+        mark_minutes(4, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.45955700) < 0.00000001),
+            'Should have a rate of 0.45955700 events/sec after 4 minutes')
+
+    def test_five_minutes(self):
+        mark_minutes(5, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.42991879) < 0.00000001),
+            'Should have a rate of 0.42991879 events/sec after 5 minutes')
+
+    def test_six_minutes(self):
+        mark_minutes(6, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.40219203) < 0.00000001),
+            'Should have a rate of 0.40219203 events/sec after 6 minutes')
+
+    def test_seven_minutes(self):
+        mark_minutes(7, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.37625345) < 0.00000001),
+            'Should have a rate of 0.37625345 events/sec after 7 minutes')
+
+    def test_eight_minutes(self):
+        mark_minutes(8, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.35198773) < 0.00000001),
+            'Should have a rate of 0.35198773 events/sec after 8 minutes')
+
+    def test_nine_minutes(self):
+        mark_minutes(9, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.32928698) < 0.00000001),
+            'Should have a rate of 0.32928698 events/sec after 9 minutes')
+
+    def test_ten_minutes(self):
+        mark_minutes(10, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.30805027) < 0.00000001),
+            'Should have a rate of 0.30805027 events/sec after 10 minutes')
+
+    def test_eleven_minutes(self):
+        mark_minutes(11, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.28818318) < 0.00000001),
+            'Should have a rate of 0.28818318 events/sec after 11 minutes')
+
+    def test_twelve_minutes(self):
+        mark_minutes(12, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.26959738) < 0.00000001),
+            'Should have a rate of 0.26959738 events/sec after 12 minutes')
+
+    def test_thirteen_minutes(self):
+        mark_minutes(13, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.25221023) < 0.00000001),
+            'Should have a rate of 0.25221023 events/sec after 13 minutes')
+
+    def test_fourteen_minutes(self):
+        mark_minutes(14, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.23594443) < 0.00000001),
+            'Should have a rate of 0.23594443 events/sec after 14 minutes')
+
+    def test_fifteen_minutes(self):
+        mark_minutes(15, self.ewma)
+        self.assertTrue(
+            (math.fabs(self.ewma.rate - 0.22072766) < 0.00000001),
+            'Should have a rate of 0.22072766 events/sec after 15 minutes')

=== added file 'txstatsd/tests/stats/test_exponentiallydecayingsample.py'
--- txstatsd/tests/stats/test_exponentiallydecayingsample.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/stats/test_exponentiallydecayingsample.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,84 @@
+import random
+
+from twisted.trial.unittest import TestCase
+
+from txstatsd.stats.exponentiallydecayingsample import (
+    ExponentiallyDecayingSample)
+
+
+class TestExponentiallyDecayingSample(TestCase):
+
+    def test_100_out_of_1000_elements(self):
+        population = [i for i in range(0, 100)]
+        sample = ExponentiallyDecayingSample(1000, 0.99)
+        for i in population:
+            sample.update(i)
+
+        self.assertEqual(sample.size(), 100, 'Should have 100 elements')
+        self.assertEqual(len(sample.get_values()), 100,
+                         'Should have 100 elements')
+        self.assertEqual(
+            len(set(sample.get_values()).difference(set(population))), 0,
+            'Should only have elements from the population')
+
+    def test_100_out_of_10_elements(self):
+        population = [i for i in range(0, 10)]
+        sample = ExponentiallyDecayingSample(100, 0.99)
+        for i in population:
+            sample.update(i)
+
+        self.assertEqual(sample.size(), 10)
+        self.assertEqual(len(sample.get_values()), 10,
+                         'Should have 10 elements')
+        self.assertEqual(
+            len(set(sample.get_values()).difference(set(population))), 0,
+            'Should only have elements from the population')
+
+    def test_heavily_biased_100_out_of_1000_elements(self):
+        population = [i for i in range(0, 100)]
+        sample = ExponentiallyDecayingSample(1000, 0.01)
+        for i in population:
+            sample.update(i)
+
+        self.assertEqual(sample.size(), 100, 'Should have 100 elements')
+        self.assertEqual(len(sample.get_values()), 100,
+                         'Should have 100 elements')
+
+        self.assertEqual(
+            len(set(sample.get_values()).difference(set(population))), 0,
+            'Should only have elements from the population')
+
+    def test_ewma_sample_load(self):
+
+        _time = [10000]
+
+        def wtime():
+            return _time[0]
+
+        sample = ExponentiallyDecayingSample(100, 0.99, wall_time=wtime)
+        sample.RESCALE_THRESHOLD = 100
+        sample.clear()
+        for i in xrange(10000000):
+            sample.update(random.normalvariate(0, 10))
+            _time[0] += 1
+
+        self.assertEqual(sample.size(), 100)
+        self.assertEqual(len(sample.get_values()), 100,
+                         'Should have 100 elements')
+    test_ewma_sample_load.skip = "takes too long to run"
+
+    def test_ewma_overflow(self):
+        """Long pauses on metric input should not overflow weight."""
+        _time = [10000]
+
+        def wtime():
+            return _time[0]
+
+        sample = ExponentiallyDecayingSample(100, 0.99, wall_time=wtime)
+        for i in xrange(100):
+            sample.update(random.normalvariate(0, 10))
+            _time[0] += 10000
+
+        self.assertEqual(sample.size(), 100)
+        self.assertEqual(len(sample.get_values()), 100,
+                         'Should have 100 elements')

=== added file 'txstatsd/tests/stats/test_uniformsample.py'
--- txstatsd/tests/stats/test_uniformsample.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/stats/test_uniformsample.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,32 @@
+
+from unittest import TestCase
+
+from txstatsd.stats.uniformsample import UniformSample
+
+
+class TestUniformSample(TestCase):
+    def test_100_out_of_1000_elements(self):
+        population = [i for i in range(0, 1000)]
+        sample = UniformSample(100)
+        for i in population:
+            sample.update(i)
+
+        self.assertEqual(sample.size(), 100, 'Should have 100 elements')
+        self.assertEqual(len(sample.get_values()), 100,
+                         'Should have 100 elements')
+        self.assertEqual(
+            len(set(sample.get_values()).difference(set(population))), 0,
+            'Should only have elements from the population')
+
+    def test_100_out_of_10_elements(self):
+        population = [i for i in range(0, 10)]
+        sample = UniformSample(100)
+        for i in population:
+            sample.update(i)
+
+        self.assertEqual(sample.size(), 10, 'Should have 10 elements')
+        self.assertEqual(len(sample.get_values()), 10,
+                         'Should have 10 elements')
+        self.assertEqual(
+            len(set(sample.get_values()).difference(set(population))), 0,
+            'Should only have elements from the population')

=== added file 'txstatsd/tests/test_client.py'
--- txstatsd/tests/test_client.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/test_client.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,93 @@
+"""Tests for the various client classes."""
+
+from twisted.internet import reactor
+from twisted.internet.defer import inlineCallbacks, Deferred
+from twisted.python import log
+from twisted.trial.unittest import TestCase
+
+from txstatsd.client import (
+    StatsDClientProtocol, TwistedStatsDClient, UdpStatsDClient)
+
+
+class TestClient(TestCase):
+
+    def setUp(self):
+        super(TestClient, self).setUp()
+        self.client = None
+        self.exception = None
+
+    def tearDown(self):
+        if self.client:
+            self.client.transport.stopListening()
+        super(TestClient, self).tearDown()
+
+    def test_twistedstatsd_write_with_wellformed_address(self):
+        self.client = TwistedStatsDClient('127.0.0.1', 8000)
+        protocol = StatsDClientProtocol(self.client)
+        reactor.listenUDP(0, protocol)
+
+        def ensure_bytes_sent(bytes_sent):
+            self.assertEqual(bytes_sent, len('message'))
+
+        def exercise(callback):
+            self.client.write('message', callback=callback)
+
+        d = Deferred()
+        d.addCallback(ensure_bytes_sent)
+        reactor.callWhenRunning(exercise, d.callback)
+        return d
+
+    @inlineCallbacks
+    def test_twistedstatsd_with_malformed_address_and_errback(self):
+        def ensure_exception_raised(ignore):
+            self.assertTrue(self.exception.startswith("DNS lookup failed"))
+
+        def capture_exception_raised(failure):
+            self.exception = failure.getErrorMessage()
+
+        yield TwistedStatsDClient(
+            '256.0.0.0', 1,
+            resolver_errback=capture_exception_raised)
+
+        d = Deferred()
+        d.addCallback(ensure_exception_raised)
+        reactor.callLater(.5, d.callback, None)
+        yield d
+
+    @inlineCallbacks
+    def test_twistedstatsd_with_malformed_address_and_no_errback(self):
+        def ensure_exception_raised(ignore):
+            self.assertTrue(self.exception.startswith("DNS lookup failed"))
+
+        def capture_exception_raised(failure):
+            self.exception = failure.getErrorMessage()
+
+        self.patch(log, "err", capture_exception_raised)
+
+        yield TwistedStatsDClient('256.0.0.0', 1)
+
+        d = Deferred()
+        d.addCallback(ensure_exception_raised)
+        reactor.callLater(.5, d.callback, None)
+        yield d
+
+    def test_udpstatsd_wellformed_address(self):
+        client = UdpStatsDClient('localhost', 8000)
+        self.assertEqual(client.host, '127.0.0.1')
+        client = UdpStatsDClient(None, None)
+        self.assertEqual(client.host, None)
+
+    def test_udpstatsd_malformed_address(self):
+        self.assertRaises(ValueError,
+                          UdpStatsDClient, 'localhost', -1)
+        self.assertRaises(ValueError,
+                          UdpStatsDClient, 'localhost', 'malformed')
+        self.assertRaises(ValueError,
+                          UdpStatsDClient, 0, 8000)
+
+    def test_udpstatsd_socket_nonblocking(self):
+        client = UdpStatsDClient('localhost', 8000)
+        client.connect()
+        # According to the python docs (and the source, I've checked)
+        # setblocking(0) is the same as settimeout(0.0).
+        self.assertEqual(client.socket.gettimeout(), 0.0)

=== added file 'txstatsd/tests/test_configurableprocessor.py'
--- txstatsd/tests/test_configurableprocessor.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/test_configurableprocessor.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,152 @@
+import time
+
+from unittest import TestCase
+
+from twisted.plugins.distinct_plugin import distinct_metric_factory
+
+from txstatsd.server.configurableprocessor import ConfigurableMessageProcessor
+
+class FlushMessagesTest(TestCase):
+
+    def test_flush_counter_with_empty_prefix(self):
+        """
+        Ensure no prefix features if none is supplied.
+        B{Note}: The C{ConfigurableMessageProcessor} reports
+        the counter value, and not the normalized version as
+        seen in the StatsD-compliant C{Processor}.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42)
+        configurable_processor.process("gorets:17|c")
+        messages = configurable_processor.flush()
+        self.assertEqual(("gorets.count", 17, 42), messages[0])
+        self.assertEqual(("statsd.numStats", 1, 42), messages[1])
+
+    def test_flush_counter_with_prefix(self):
+        """
+        Ensure the prefix features if one is supplied.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42, message_prefix="test.metric")
+        configurable_processor.process("gorets:17|c")
+        messages = configurable_processor.flush()
+        self.assertEqual(("test.metric.gorets.count", 17, 42), messages[0])
+        self.assertEqual(("test.metric.statsd.numStats", 1, 42),
+                         messages[1])
+
+    def test_flush_counter_with_internal_prefix(self):
+        """
+        Ensure the prefix features if one is supplied.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42, message_prefix="test.metric",
+            internal_metrics_prefix="statsd.foo.")
+        configurable_processor.process("gorets:17|c")
+        messages = configurable_processor.flush()
+        self.assertEqual(("test.metric.gorets.count", 17, 42), messages[0])
+        self.assertEqual(("statsd.foo.numStats", 1, 42),
+                         messages[1])
+
+    def test_flush_plugin(self):
+        """
+        Ensure the prefix features if one is supplied.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42, message_prefix="test.metric",
+            plugins=[distinct_metric_factory])
+        configurable_processor.process("gorets:17|pd")
+        messages = configurable_processor.flush()
+        self.assertEquals(("test.metric.gorets.count", 1, 42), messages[0])
+
+    def test_flush_single_timer_single_time(self):
+        """
+        If a single timer with a single data point is present, all
+        percentiles will be set to the same value.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42)
+
+        configurable_processor.process("glork:24|ms")
+        messages = configurable_processor.flush()
+
+        self.assertEqual(('glork.15min_rate', 0.0, 42), messages[0])
+        self.assertEqual(('glork.1min_rate', 0.0, 42), messages[1])
+        self.assertEqual(('glork.5min_rate', 0.0, 42), messages[2])
+        self.assertEqual(("glork.999percentile", 24.0, 42), messages[3])
+        self.assertEqual(("glork.99percentile", 24.0, 42), messages[4])
+        self.assertEqual(("glork.count", 1., 42), messages[5])
+        self.assertEqual(("glork.max", 24.0, 42), messages[6])
+        self.assertEqual(("glork.mean", 24.0, 42), messages[7])
+        self.assertEqual(("glork.min", 24.0, 42), messages[8])
+        self.assertEqual(("glork.stddev", 0.0, 42), messages[9])
+
+    def test_flush_single_timer_multiple_times(self):
+        """
+        Test reporting of multiple timer metric samples.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42)
+
+        configurable_processor.process("glork:4|ms")
+        configurable_processor.update_metrics()
+        configurable_processor.process("glork:8|ms")
+        configurable_processor.update_metrics()
+        configurable_processor.process("glork:15|ms")
+        configurable_processor.update_metrics()
+        configurable_processor.process("glork:16|ms")
+        configurable_processor.update_metrics()
+        configurable_processor.process("glork:23|ms")
+        configurable_processor.update_metrics()
+        configurable_processor.process("glork:42|ms")
+        configurable_processor.update_metrics()
+
+        messages = configurable_processor.flush()
+
+        self.assertEqual(('glork.15min_rate', 0.20000000000000001, 42),
+            messages[0])
+        self.assertEqual(('glork.1min_rate', 0.20000000000000001, 42),
+            messages[1])
+        self.assertEqual(('glork.5min_rate', 0.20000000000000001, 42),
+            messages[2])
+        self.assertEqual(("glork.999percentile", 42.0, 42), messages[3])
+        self.assertEqual(("glork.99percentile", 42.0, 42), messages[4])
+        self.assertEqual(('glork.count', 6.0, 42), messages[5])
+        self.assertEqual(("glork.max", 42.0, 42), messages[6])
+        self.assertEqual(("glork.mean", 18.0, 42), messages[7])
+        self.assertEqual(("glork.min", 4.0, 42), messages[8])
+        self.assertEqual(("glork.stddev", 13.490738, 42), messages[9])
+
+
+class FlushMeterMetricMessagesTest(TestCase):
+
+    def setUp(self):
+        self.configurable_processor = ConfigurableMessageProcessor(
+            time_function=self.wall_clock_time, message_prefix="test.metric")
+        self.time_now = int(time.time())
+
+    def wall_clock_time(self):
+        return self.time_now
+
+    def mark_minutes(self, minutes):
+        for i in range(1, minutes * 60, 5):
+            self.processor.update_metrics()
+
+    def test_flush_meter_metric_with_prefix(self):
+        """
+        Test the correct rendering of the Graphite report for
+        a meter metric when a prefix is supplied.
+        """
+        self.configurable_processor.process("gorets:3.0|m")
+
+        self.time_now += 1
+        messages = self.configurable_processor.flush()
+        self.assertEqual(("test.metric.gorets.15min_rate", 0.0, self.time_now),
+                         messages[0])
+        self.assertEqual(("test.metric.gorets.1min_rate", 0.0, self.time_now),
+                         messages[1])
+        self.assertEqual(("test.metric.gorets.5min_rate", 0.0, self.time_now),
+                         messages[2])
+        self.assertEqual(("test.metric.gorets.count", 3.0, self.time_now),
+                         messages[3])
+        self.assertEqual(("test.metric.gorets.mean_rate", 3.0, self.time_now),
+                         messages[4])

=== added file 'txstatsd/tests/test_httpinfo.py'
--- txstatsd/tests/test_httpinfo.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/test_httpinfo.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,128 @@
+# -*- coding: utf-8 *-*
+import json
+
+from twisted.trial.unittest import TestCase
+
+from twisted.internet import reactor, defer, protocol
+from twisted.application.service import IService
+from twisted.web.client import Agent
+
+from txstatsd.metrics.timermetric import TimerMetricReporter
+from txstatsd.server import httpinfo
+from txstatsd import service
+
+
+class Dummy:
+    flush_interval = 10
+    last_flush_duration = 3
+    last_process_duration = 2
+
+
+class ResponseCollector(protocol.Protocol):
+
+    def __init__(self, finished):
+        self.finished = finished
+        self.data = []
+
+    def dataReceived(self, bytes):
+        self.data.append(bytes)
+
+    def connectionLost(self, reason):
+        self.finished.callback("".join(self.data))
+
+
+def collect_response(response):
+    d = defer.Deferred()
+    c = ResponseCollector(d)
+    response.deliverBody(c)
+    return d
+
+
+class HttpException(Exception):
+
+    def __init__(self, response):
+        super(HttpException, self).__init__(response.phrase)
+        self.response = response
+
+
+class ServiceTestsBuilder(TestCase):
+
+    def setUp(self):
+        self.service = None
+
+    @defer.inlineCallbacks
+    def get_results(self, path, **kwargs):
+        webport = 12323
+        o = service.StatsDOptions()
+        o["http-port"] = webport
+        d = Dummy()
+        d.__dict__.update(kwargs)
+        self.service = s = httpinfo.makeService(o, d, d)
+        s.startService()
+        agent = Agent(reactor)
+
+        result = yield agent.request('GET',
+            'http://localhost:%s/%s' % (webport, path))
+        if result.code != 200:
+            raise HttpException(result)
+        data = yield collect_response(result)
+        defer.returnValue(data)
+
+    def tearDown(self):
+        if self.service is not None:
+            self.service.stopService()
+
+    @defer.inlineCallbacks
+    def test_httpinfo_ok(self):
+        data = yield self.get_results("status")
+        self.assertEquals(json.loads(data)["status"], "OK")
+
+    @defer.inlineCallbacks
+    def test_httpinfo_error(self):
+        try:
+            data = yield self.get_results("status", last_flush_duration=30)
+        except HttpException, e:
+            self.assertEquals(e.response.code, 500)
+        else:
+            self.fail("Not 500")
+
+    @defer.inlineCallbacks
+    def test_httpinfo_timer(self):
+        try:
+            data = yield self.get_results("metrics/gorets",
+                timer_metrics={'gorets': 100})
+        except HttpException, e:
+            self.assertEquals(e.response.code, 404)
+        else:
+            self.fail("Not 404")
+
+    @defer.inlineCallbacks
+    def test_httpinfo_timer2(self):
+        """Returns the canonical empty histogram without data."""
+        tmr = TimerMetricReporter('gorets')
+        data = yield self.get_results("metrics/gorets",
+            timer_metrics={'gorets': tmr})
+        self.assertEquals(json.loads(data)["histogram"], [0.] * 10)
+
+    @defer.inlineCallbacks
+    def test_httpinfo_timer3(self):
+        """Returns a valid histogram with data."""
+
+        tmr = TimerMetricReporter('gorets')
+        for i in range(1, 1001):
+            tmr.histogram.update(i)
+        data = yield self.get_results("metrics/gorets",
+            timer_metrics={'gorets': tmr})
+        hist = json.loads(data)
+        self.assertTrue(isinstance(hist, dict))
+        self.assertEquals(sum(hist["histogram"]), 1000)
+
+    @defer.inlineCallbacks
+    def test_httpinfo_fake_plugin(self):
+        """Also works for plugins."""
+
+        tmr = TimerMetricReporter('gorets')
+        data = yield self.get_results("metrics/gorets",
+            timer_metrics={}, plugin_metrics={'gorets': tmr})
+        hist = json.loads(data)
+        self.assertTrue(isinstance(hist, dict))

=== added file 'txstatsd/tests/test_loggingprocessor.py'
--- txstatsd/tests/test_loggingprocessor.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/test_loggingprocessor.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,64 @@
+
+from unittest import TestCase
+
+from twisted.plugins.distinct_plugin import distinct_metric_factory
+
+from txstatsd.server.loggingprocessor import LoggingMessageProcessor
+
+
+class FakeMeterMetric(object):
+    def report(self, *args):
+        return [('Sample report', 1, 2)]
+
+class TestLogger(object):
+    def __init__(self):
+        self.log = ''
+
+    def info(self, measurement):
+        self.log += measurement + "\n"
+
+
+class TestLoggingMessageProcessor(TestCase):
+
+    def test_logger_with_no_info(self):
+        def invoker():
+            logger = 'logger'
+            LoggingMessageProcessor(logger)
+
+        self.assertRaises(TypeError, invoker)
+
+    def test_logger_with_non_callable_info(self):
+        def invoker():
+            class Logger(object):
+                def __init__(self):
+                    self.info = 'logger'
+
+            logger = Logger()
+            LoggingMessageProcessor(logger)
+
+        self.assertRaises(TypeError, invoker)
+
+    def test_logger(self):
+        logger = TestLogger()
+        processor = LoggingMessageProcessor(logger)
+        metric = FakeMeterMetric()
+        processor.meter_metrics['test'] = metric
+        processor.flush()
+        expected = ["Out: %s %s %s" % message
+                for message in metric.report()]
+        self.assertFalse(set(expected).difference(logger.log.splitlines()))
+
+    def test_logger_plugin(self):
+        logger = TestLogger()
+        processor = LoggingMessageProcessor(
+            logger, plugins=[distinct_metric_factory],
+            time_function=lambda: 42)
+        msg_in = "gorets:17|pd"
+        processor.process(msg_in)
+        processor.flush()
+        messages = processor.plugin_metrics['gorets'].flush(
+            10, processor.time_function())
+        expected = ["In: %s" % msg_in]\
+            + ["Out: %s %s %s" % message
+                for message in messages]
+        self.assertFalse(set(expected).difference(logger.log.splitlines()))

=== added file 'txstatsd/tests/test_metrics.py'
--- txstatsd/tests/test_metrics.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/test_metrics.py	2012-05-14 16:57:19 +0000
@@ -0,0 +1,120 @@
+"""Tests for the Metrics convenience class."""
+
+import re
+import time
+from unittest import TestCase
+from txstatsd.metrics.extendedmetrics import ExtendedMetrics
+from txstatsd.metrics.metrics import Metrics
+
+
+class FakeStatsDClient(object):
+
+    def connect(self):
+        """Connect to the StatsD server."""
+        pass
+
+    def disconnect(self):
+        """Disconnect from the StatsD server."""
+        pass
+
+    def write(self, data):
+        """Send the metric to the StatsD server."""
+        self.data = data
+
+
+class TestMetrics(TestCase):
+
+    def setUp(self):
+        self.connection = FakeStatsDClient()
+        self.metrics = Metrics(self.connection, 'txstatsd.tests')
+
+    def test_gauge(self):
+        """Test reporting of a gauge metric sample."""
+        self.metrics.gauge('gauge', 102)
+        self.assertEqual(self.connection.data,
+                         'txstatsd.tests.gauge:102|g')
+
+    def test_meter(self):
+        """Test reporting of a meter metric sample."""
+        self.metrics.meter('meter', 3)
+        self.assertEqual(self.connection.data,
+                         'txstatsd.tests.meter:3|m')
+
+    def test_counter(self):
+        """Test the increment and decrement operations."""
+        self.metrics.increment('counter', 18)
+        self.assertEqual(self.connection.data,
+                         'txstatsd.tests.counter:18|c')
+        self.metrics.decrement('counter', 9)
+        self.assertEqual(self.connection.data,
+                         'txstatsd.tests.counter:-9|c')
+
+    def test_timing(self):
+        """Test the timing operation."""
+        self.metrics.timing('timing', 101.1234)
+        self.assertEqual(self.connection.data,
+                         'txstatsd.tests.timing:101123.4|ms')
+
+    def test_timing_automatic(self):
+        """Test the automatic timing operation with explicit reset"""
+        start_time = time.time()
+
+        self.metrics.reset_timing()
+        time.sleep(.1)
+        self.metrics.timing('timing')
+
+        elapsed = time.time() - start_time
+
+        label, val, units = re.split(":|\|", self.connection.data)
+        self.assertEqual(label, 'txstatsd.tests.timing')
+        self.assertEqual(units, 'ms')
+        self.assertTrue(100 <= float(val) <= elapsed * 1000)
+
+    def test_timing_automatic_implicit_reset(self):
+        """Test the automatic timing operation with implicit reset"""
+        start_time = time.time()
+
+        self.metrics.timing('something_else')
+        time.sleep(.1)
+        self.metrics.timing('timing')
+
+        elapsed = time.time() - start_time
+
+        label, val, units = re.split(":|\|", self.connection.data)
+        self.assertEqual(label, 'txstatsd.tests.timing')
+        self.assertEqual(units, 'ms')
+        self.assertTrue(100 <= float(val) <= elapsed * 1000)
+
+    def test_generic(self):
+        """Test the GenericMetric class."""
+        self.metrics.report('users', "pepe", "pd")
+        self.assertEqual(self.connection.data,
+                         'txstatsd.tests.users:pepe|pd')
+
+    def test_empty_namespace(self):
+        """Test reporting of an empty namespace."""
+        self.metrics.namespace = None
+        self.metrics.gauge('gauge', 213)
+        self.assertEqual(self.connection.data,
+                         'gauge:213|g')
+