Merge lp:~theiw/txstatsd/deprecate-meter into lp:txstatsd

[Ian Wilkinson]

[1] service.py placed at top-level.

[2] client.py placed at top-level

[3] Renamed StatsDClient to TwistedStatsDClient.

[4] report.py to top-level.

[5] processor.py refactored following the style introduced with gauge metric handling.

[6] meter.py to be removed.

[Lucio Torre (lucio.torre)]

 946 + metric = [float(v) for v in values]
 947 + metric.append(key)

 that code seems to imply that the contents of metric could be *values, key
 when in reality it has to be [value, key]
 so maybe it can be replaced with a one liner: metric = [ float(value[0]), key]

review: Approve

[Sidnei da Silva (sidnei)]

Looks good, although there's a few problems that need fixing before merge. I'll approve it and trust that you'll fix them. :)

[1] Some pep8 warnings:

txstatsd/server/processor.py:124:80: E501 line too long (81 characters)
txstatsd/client.py:51:39: W291 trailing whitespace
txstatsd/client.py:115:1: W391 blank line at end of file

[2] Some pyflakes warnings:

txstatsd/service.py:113: undefined name 'meter'

[3] A test failure:

===============================================================================
[ERROR]
Traceback (most recent call last):
  File "/usr/lib/python2.7/dist-packages/twisted/trial/runner.py", line 575, in loadPackage
    module = modinfo.load()
  File "/usr/lib/python2.7/dist-packages/twisted/python/modules.py", line 383, in load
    return self.pathEntry.pythonPath.moduleLoader(self.name)
  File "/usr/lib/python2.7/dist-packages/twisted/python/reflect.py", line 464, in namedAny
    topLevelPackage = _importAndCheckStack(trialname)
  File "/src/txstatsd/deprecate-meter/txstatsd/tests/test_service.py", line 5, in <module>
    from txstatsd import service
  File "/src/txstatsd/deprecate-meter/txstatsd/service.py", line 12, in <module>
    from txstatsd.processor import MessageProcessor
exceptions.ImportError: No module named processor

txstatsd.tests.test_service

[4] Seems like some files where removed and added again. This not only breaks history but it can also cause conflicts for anyone that has pending branches which get merged after yours. I suggest starting fresh from trunk and reapplying your changes. You could even just take a patch from this branch and apply it on top of a clean branch.

review: Approve

[lp:~theiw/txstatsd/deprecate-meter]

19. By Ian Wilkinson on 2011-07-25

pep8 and correct failing test.

=== modified file 'example-stats-client.tac'
--- example-stats-client.tac	2011-07-12 04:54:49 +0000
+++ example-stats-client.tac	2011-07-25 16:11:30 +0000
@@ -6,14 +6,20 @@
 
 from twisted.application.service import Application
 
-from txstatsd.protocol import StatsDClientProtocol
-from txstatsd.metrics import TransportMeter
+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")
-meter = TransportMeter(prefix=socket.gethostname() + ".example-client")
+statsd_client = TwistedStatsDClient(STATSD_HOST, STATSD_PORT)
+metrics = Metrics(connection=statsd_client,
+                  namespace=socket.gethostname() + ".example-client")
 
 reporting = ReportingService()
 reporting.setServiceParent(application)
@@ -24,13 +30,13 @@
 def random_walker(name):
     """Meters a random walk."""
     if random.random() > 0.5:
-        meter.increment(name)
+        metrics.increment(name)
     else:
-        meter.decrement(name)
+        metrics.decrement(name)
 
 def random_normal(name):
     """Meters samples from a normal distribution."""
-    meter.timing(name, random.normalvariate(10, 3))
+    metrics.timing(name, random.normalvariate(10, 3))
 
 
 for n in range(5):
@@ -42,5 +48,6 @@
     t.start(0.5, now=False)
 
 
-protocol = StatsDClientProtocol("127.0.0.1", 8125, meter, 6000)
+protocol = StatsDClientProtocol(STATSD_HOST, STATSD_PORT,
+                                statsd_client, 6000)
 reactor.listenUDP(0, protocol)

=== added file 'txstatsd/client.py'
--- txstatsd/client.py	1970-01-01 00:00:00 +0000
+++ txstatsd/client.py	2011-07-25 16:11:30 +0000
@@ -0,0 +1,114 @@
+
+import socket
+
+from twisted.internet.protocol import DatagramProtocol
+
+
+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, host, port, client, interval=None):
+        self.host = host
+        self.port = port
+        self.client = client
+        self.interval = interval
+
+    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):
+        """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 connect_callback: The callback to invoke on connection.
+        @param disconnect_callback: The callback to invoke on disconnection.
+        """
+
+        self.host = host
+        self.port = port
+        self.connect_callback = connect_callback
+        self.disconnect_callback = disconnect_callback
+
+        self.transport = None
+
+    def connect(self, transport=None):
+        """Connect to the StatsD server."""
+        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 = self.host = self.port = None
+
+    def write(self, data):
+        """Send the metric to the StatsD server."""
+        if self.transport is not None:
+            self.transport.write(data, (self.host, self.port))
+
+
+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.
+        """
+        self.host = host
+        self.port = port
+
+    def connect(self):
+        """Connect to the StatsD server."""
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.addr = (self.host, self.port)
+
+    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.addr is None or self.socket is None:
+            return
+        self.socket.sendto(data, self.addr)
+
+
+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 connect(self):
+        pass
+
+    def disconnect(self):
+        pass
+
+    def write(self, data):
+        """Write directly to the C{MessageProcessor}."""
+        self._processor.process(data)

=== added directory 'txstatsd/metrics'
=== removed file 'txstatsd/metrics.py'
--- txstatsd/metrics.py	2011-07-12 04:54:49 +0000
+++ txstatsd/metrics.py	1970-01-01 00:00:00 +0000
@@ -1,154 +0,0 @@
-import time
-import socket
-import random
-
-
-class BaseMeter(object):
-
-    def __init__(self, prefix="", sample_rate=1):
-        """Base implementation for a C{StatsD} client utility.
-
-        All networking logic is left for the subclasses.
-
-        prefix: a string to prepend to all metrics. useful to insert stuff like
-            server name.
-        sample_rate: only write C{sample_rate} percent of messages to the wire.
-        """
-        self.prefix = prefix
-        self.sample_rate = sample_rate
-
-    def increment(self, name, count=1):
-        """Report and increase in name by count."""
-        self.send("%s:%s|c" % (name, count))
-
-    def decrement(self, name, count=1):
-        """Report and decrease in name by count."""
-        self.increment(name, count * -1)
-
-    def done(self, name):
-        """Report that name completed successfully."""
-        self.increment(name + ".done")
-
-    def error(self, name):
-        """Report that name failed."""
-        self.increment(name + ".error")
-
-    def timing(self, name, duration):
-        """Report that name took duration ms."""
-        self.send("%s:%s|ms" % (name, duration))
-
-    def send(self, data):
-        """Send out C{data} to C{StatsD} if over C{sample_rate}."""
-
-        if self.sample_rate < 1:
-            if random.random() > self.sample_rate:
-                return
-            data += "|@%s" % (self.sample_rate,)
-
-        if self.prefix:
-            data = self.prefix + "." + data
-
-        self.write(data)
-
-    def write(self, data):
-        """Write out C{data} to the network."""
-        raise NotImplementedError()
-
-
-class InProcessMeter(BaseMeter):
-    """A meter that can be used inside the C{StatsD} daemon itself."""
-
-    def __init__(self, processor, prefix="", sample_rate=1):
-        self.processor = processor
-        BaseMeter.__init__(self, prefix=prefix, sample_rate=sample_rate)
-
-    def write(self, data):
-        """Pass the data along directly to the C{Processor}."""
-        self.processor.process(data)
-
-
-class Meter(BaseMeter):
-    """A trivial, non-Twisted-dependent meter."""
-
-    def __init__(self, host=None, port=None, prefix="", sample_rate=1):
-        """Build a meter that reports to host:port over udp.
-
-        host: statsd host
-        port: statsd port
-        prefix: a string to prepend to all metrics. useful to insert stuff like
-            server name.
-        sample_rate: only write C{sample_rate} percent of messages to the wire.
-        """
-        BaseMeter.__init__(self, prefix=prefix, sample_rate=sample_rate)
-        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        self.addr = (host, port)
-
-    def write(self, data):
-        """Write out C{data} to the network."""
-        if self.addr is None or self.socket is None:
-            return
-        self.socket.sendto(data, self.addr)
-
-    def close(self):
-        """Close the socket."""
-        self.socket.close()
-        self.socket = None
-
-
-class Measure(object):
-    """Context Manager for generic measuring."""
-
-    def __init__(self, prefix, operation_name, meter=None):
-        if meter is None:
-            meter = Meter(prefix)
-        self.meter = meter
-        self.operation_name = operation_name
-        self.before = None
-
-    def __enter__(self):
-        """Increase operation meter."""
-        self.before = time.time()
-        self.meter.increment(self.operation_name)
-
-    def __exit__(self, exception_type, value, traceback):
-        """
-        Record time since __enter__, increase either the .done or the
-        .error meter, and decrease the operation meter.
-
-        """
-        if exception_type is None:
-            self.meter.timing(self.operation_name, time.time() - self.before)
-            self.meter.increment(self.operation_name + '.done')
-        else:
-            self.meter.increment(self.operation_name + '.error')
-        self.meter.decrement(self.operation_name)
-
-
-class TransportMeter(BaseMeter):
-
-    transport = None
-    host = None
-    port = None
-
-    def __init__(self, prefix="", sample_rate=1,
-                 connect_callback=None, disconnect_callback=None):
-        self.connect_callback = connect_callback
-        self.disconnect_callback = disconnect_callback
-        BaseMeter.__init__(self, prefix=prefix, sample_rate=sample_rate)
-
-    def connect(self, transport, host, port):
-        self.transport = transport
-        self.host = host
-        self.port = port
-        if self.connect_callback is not None:
-            self.connect_callback()
-
-    def disconnect(self):
-        if self.disconnect_callback is not None:
-            self.disconnect_callback()
-        self.transport = self.host = self.port = None
-
-    def write(self, data):
-        """Send metrics to the StatsD server using the transport."""
-        if self.transport is not None:
-            self.transport.write(data, (self.host, self.port))

=== added file 'txstatsd/metrics/__init__.py'
=== added file 'txstatsd/metrics/gaugemetric.py'
--- txstatsd/metrics/gaugemetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/gaugemetric.py	2011-07-25 16:11:30 +0000
@@ -0,0 +1,22 @@
+
+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)

=== added file 'txstatsd/metrics/metric.py'
--- txstatsd/metrics/metric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/metric.py	2011-07-25 16:11:30 +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	2011-07-25 16:11:30 +0000
@@ -0,0 +1,80 @@
+
+from txstatsd.metrics.gaugemetric import GaugeMetric
+from txstatsd.metrics.metric import Metric
+
+
+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 = {}
+
+    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 increment(self, name, value, 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, 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 timing(self, name, duration, sample_rate=1):
+        """Report this sample performed in duration ms."""
+        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)
+
+    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

=== removed file 'txstatsd/processor.py'
--- txstatsd/processor.py	2011-07-05 05:27:22 +0000
+++ txstatsd/processor.py	1970-01-01 00:00:00 +0000
@@ -1,140 +0,0 @@
-import re
-import time
-import logging
-
-from twisted.python import log
-
-
-SPACES = re.compile("\s+")
-SLASHES = re.compile("\/+")
-NON_ALNUM = re.compile("[^a-zA-Z_\-0-9\.]")
-RATE = re.compile("^@([\d\.]+)")
-COUNTERS_MESSAGE = (
-    "stats.%(key)s %(value)s %(timestamp)s\n"
-    "stats_counts.%(key)s %(count)s %(timestamp)s\n")
-TIMERS_MESSAGE = (
-    "stats.timers.%(key)s.mean %(mean)s %(timestamp)s\n"
-    "stats.timers.%(key)s.upper %(upper)s %(timestamp)s\n"
-    "stats.timers.%(key)s.upper_%(percent)s %(threshold_upper)s"
-        " %(timestamp)s\n"
-    "stats.timers.%(key)s.lower %(lower)s %(timestamp)s\n"
-    "stats.timers.%(key)s.count %(count)s %(timestamp)s\n")
-
-
-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 MessageProcessor(object):
-
-    def __init__(self, time_function=time.time):
-        self.time_function = time_function
-        self.timers = {}
-        self.counters = {}
-
-    def fail(self, message):
-        """Log and discard malformed message."""
-        log.msg("Bad line: %r" % message, logLevel=logging.DEBUG)
-
-    def process(self, message):
-        """
-        Process a single entry, adding it to either C{counters} or C{timers}
-        depending on which kind of message it is.
-        """
-        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)
-
-        try:
-            value = float(fields[0])
-        except (TypeError, ValueError):
-            return self.fail(message)
-
-        key = normalize_key(key)
-
-        if fields[1] == "c":
-            rate = 1
-            if len(fields) == 3:
-                match = RATE.match(fields[2])
-                if match is None:
-                    return self.fail(message)
-                rate = match.group(1)
-            if key not in self.counters:
-                self.counters[key] = 0
-            self.counters[key] += value * (1 / float(rate))
-        elif fields[1] == "ms":
-            if key not in self.timers:
-                self.timers[key] = []
-            self.timers[key].append(value)
-        else:
-            return self.fail(message)
-
-    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 = []
-        num_stats = 0
-        interval = interval / 1000
-        timestamp = int(self.time_function())
-
-        for key, count in self.counters.iteritems():
-            self.counters[key] = 0
-
-            value = count / interval
-            messages.append(COUNTERS_MESSAGE % {
-                "key": key,
-                "value": value,
-                "count": count,
-                "timestamp": timestamp})
-            num_stats += 1
-
-        threshold_value = ((100 - percent) / 100.0)
-        for key, timers in self.timers.iteritems():
-            count = len(timers)
-            if count > 0:
-                self.timers[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
-
-                num_stats += 1
-
-                messages.append(TIMERS_MESSAGE % {
-                    "key": key,
-                    "mean": mean,
-                    "upper": upper,
-                    "percent": percent,
-                    "threshold_upper": threshold_upper,
-                    "lower": lower,
-                    "count": count,
-                    "timestamp": timestamp})
-
-        messages.append("statsd.numStats %s %s" % (num_stats, timestamp))
-        return messages

=== removed file 'txstatsd/protocol.py'
--- txstatsd/protocol.py	2011-07-13 16:58:42 +0000
+++ txstatsd/protocol.py	1970-01-01 00:00:00 +0000
@@ -1,82 +0,0 @@
-from twisted.internet import task, defer
-from twisted.protocols.basic import LineOnlyReceiver
-from twisted.internet.protocol import (
-    DatagramProtocol, ReconnectingClientFactory)
-
-
-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):
-        self.processor = processor
-
-    def datagramReceived(self, data, (host, port)):
-        """Process received data and store it locally."""
-        self.processor.process(data)
-
-
-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, host, port, meter, interval=None):
-        self.host = host
-        self.port = port
-        self.meter = meter
-        self.interval = interval
-
-    def startProtocol(self):
-        """Connect to destination host."""
-        self.meter.connect(self.transport, self.host, self.port)
-
-    def stopProtocol(self):
-        """Connection was lost."""
-        self.meter.disconnect()
-
-
-class GraphiteProtocol(LineOnlyReceiver):
-    """A client protocol for talking to Graphite.
-
-    Messages to Graphite are line-based and C{\n}-separated.
-    """
-
-    delimiter = "\n"
-
-    def __init__(self, processor, interval):
-        self.processor = processor
-        self.interval = interval
-        self.flush_task = task.LoopingCall(self.flushProcessor)
-        self.flush_task.start(self.interval / 1000, False)
-
-    @defer.inlineCallbacks
-    def flushProcessor(self):
-        """Flush messages queued in the processor to Graphite."""
-        for message in self.processor.flush(interval=self.interval):
-            for line in message.splitlines():
-                if self.connected:
-                    self.sendLine(line)
-                yield
-
-
-class GraphiteClientFactory(ReconnectingClientFactory):
-    """A reconnecting Graphite client."""
-
-    def __init__(self, processor, interval):
-        self.processor = processor
-        self.interval = interval
-
-    def buildProtocol(self, addr):
-        """
-        Build a new instance of the L{Graphite} protocol, bound to the
-        L{MessageProcessor}.
-        """
-        self.resetDelay()
-        protocol = GraphiteProtocol(self.processor, self.interval)
-        protocol.factory = self
-        return protocol

=== added file 'txstatsd/report.py'
--- txstatsd/report.py	1970-01-01 00:00:00 +0000
+++ txstatsd/report.py	2011-07-25 16:11:30 +0000
@@ -0,0 +1,46 @@
+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
+
+
+class ReportingService(Service):
+
+    def __init__(self):
+        self.tasks = []
+
+    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}.
+        """
+        task = LoopingCall(self.wrapped(function, report_function))
+        self.tasks.append((task, interval))
+
+    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():
+                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):
+        for task, interval in self.tasks:
+            task.start(interval, now=False)
+
+    def stopService(self):
+        for task, interval in self.tasks:
+            task.stop()

=== removed file 'txstatsd/report.py'
--- txstatsd/report.py	2011-07-12 13:54:16 +0000
+++ txstatsd/report.py	1970-01-01 00:00:00 +0000
@@ -1,46 +0,0 @@
-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
-
-
-class ReportingService(Service):
-
-    def __init__(self):
-        self.tasks = []
-
-    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}.
-        """
-        task = LoopingCall(self.wrapped(function, report_function))
-        self.tasks.append((task, interval))
-
-    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():
-                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):
-        for task, interval in self.tasks:
-            task.start(interval, now=False)
-
-    def stopService(self):
-        for task, interval in self.tasks:
-            task.stop()

=== added directory 'txstatsd/server'
=== added file 'txstatsd/server/__init__.py'
=== added file 'txstatsd/server/processor.py'
--- txstatsd/server/processor.py	1970-01-01 00:00:00 +0000
+++ txstatsd/server/processor.py	2011-07-25 16:11:30 +0000
@@ -0,0 +1,213 @@
+from collections import deque
+import re
+import time
+import logging
+
+from twisted.python import log
+
+
+SPACES = re.compile("\s+")
+SLASHES = re.compile("\/+")
+NON_ALNUM = re.compile("[^a-zA-Z_\-0-9\.]")
+RATE = re.compile("^@([\d\.]+)")
+COUNTERS_MESSAGE = (
+    "stats.%(key)s %(value)s %(timestamp)s\n"
+    "stats_counts.%(key)s %(count)s %(timestamp)s\n")
+TIMERS_MESSAGE = (
+    "stats.timers.%(key)s.mean %(mean)s %(timestamp)s\n"
+    "stats.timers.%(key)s.upper %(upper)s %(timestamp)s\n"
+    "stats.timers.%(key)s.upper_%(percent)s %(threshold_upper)s"
+        " %(timestamp)s\n"
+    "stats.timers.%(key)s.lower %(lower)s %(timestamp)s\n"
+    "stats.timers.%(key)s.count %(count)s %(timestamp)s\n")
+
+GAUGE_METRIC_MESSAGE = (
+    "stats.gauge.%(key)s.value %(value)s %(timestamp)s\n")
+
+
+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 MessageProcessor(object):
+
+    def __init__(self, time_function=time.time):
+        self.time_function = time_function
+        self.timer_metrics = {}
+        self.counter_metrics = {}
+        self.gauge_metrics = deque()
+
+    def fail(self, message):
+        """Log and discard malformed message."""
+        log.msg("Bad line: %r" % message, logLevel=logging.DEBUG)
+
+    def process(self, message):
+        """
+        Process a single entry, adding it to either C{counters}, C{timers},
+        or C{gauge_metrics} depending on which kind of message it is.
+        """
+        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)
+
+        if fields[1] == "c":
+            self.process_counter_metric(key, fields, message)
+        elif fields[1] == "ms":
+            self.process_timer_metric(key, fields[0], message)
+        elif fields[1] == "g":
+            self.process_gauge_metric(key, fields[0], message)
+        else:
+            return self.fail(message)
+
+    def process_timer_metric(self, key, value, message):
+        try:
+            value = float(value)
+        except (TypeError, ValueError):
+            return self.fail(message)
+        if key not in self.timer_metrics:
+            self.timer_metrics[key] = []
+        self.timer_metrics[key].append(value)
+
+    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)
+        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:
+            metric = [float(values[0]), key]
+            self.gauge_metrics.append(metric)
+        except (TypeError, ValueError):
+            self.fail(message)
+
+    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 = []
+        num_stats = 0
+        interval = interval / 1000
+        timestamp = int(self.time_function())
+
+        counter_metrics, events = self.flush_counter_metrics(interval,
+                                                             timestamp)
+        if events > 0:
+            messages.extend(counter_metrics)
+            num_stats += events
+
+        timer_metrics, events = self.flush_timer_metrics(percent, timestamp)
+        if events > 0:
+            messages.extend(timer_metrics)
+            num_stats += events
+
+        gauge_metrics, events = self.flush_gauge_metrics(timestamp)
+        if events > 0:
+            messages.extend(gauge_metrics)
+            num_stats += events
+
+        messages.append("statsd.numStats %s %s" % (num_stats, 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
+            message = COUNTERS_MESSAGE % {
+                "key": key,
+                "value": value,
+                "count": count,
+                "timestamp": timestamp}
+            metrics.append(message)
+            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
+
+                message = TIMERS_MESSAGE % {
+                    "key": key,
+                    "mean": mean,
+                    "upper": upper,
+                    "percent": percent,
+                    "threshold_upper": threshold_upper,
+                    "lower": lower,
+                    "count": count,
+                    "timestamp": timestamp}
+                metrics.append(message)
+                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]
+
+            message = GAUGE_METRIC_MESSAGE % {
+                "key": key,
+                "value": value,
+                "timestamp": timestamp}
+            metrics.append(message)
+            events += 1
+
+        self.gauge_metrics.clear()
+
+        return (metrics, events)

=== added file 'txstatsd/server/protocol.py'
--- txstatsd/server/protocol.py	1970-01-01 00:00:00 +0000
+++ txstatsd/server/protocol.py	2011-07-25 16:11:30 +0000
@@ -0,0 +1,61 @@
+from twisted.internet import task, defer
+from twisted.protocols.basic import LineOnlyReceiver
+from twisted.internet.protocol import (
+    DatagramProtocol, ReconnectingClientFactory)
+
+
+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):
+        self.processor = processor
+
+    def datagramReceived(self, data, (host, port)):
+        """Process received data and store it locally."""
+        self.processor.process(data)
+
+
+class GraphiteProtocol(LineOnlyReceiver):
+    """A client protocol for talking to Graphite.
+
+    Messages to Graphite are line-based and C{\n}-separated.
+    """
+
+    delimiter = "\n"
+
+    def __init__(self, processor, interval):
+        self.processor = processor
+        self.interval = interval
+        self.flush_task = task.LoopingCall(self.flushProcessor)
+        self.flush_task.start(self.interval / 1000, False)
+
+    @defer.inlineCallbacks
+    def flushProcessor(self):
+        """Flush messages queued in the processor to Graphite."""
+        for message in self.processor.flush(interval=self.interval):
+            for line in message.splitlines():
+                if self.connected:
+                    self.sendLine(line)
+                yield
+
+
+class GraphiteClientFactory(ReconnectingClientFactory):
+    """A reconnecting Graphite client."""
+
+    def __init__(self, processor, interval):
+        self.processor = processor
+        self.interval = interval
+
+    def buildProtocol(self, addr):
+        """
+        Build a new instance of the L{Graphite} protocol, bound to the
+        L{MessageProcessor}.
+        """
+        self.resetDelay()
+        protocol = GraphiteProtocol(self.processor, self.interval)
+        protocol.factory = self
+        return protocol

=== added file 'txstatsd/service.py'
--- txstatsd/service.py	1970-01-01 00:00:00 +0000
+++ txstatsd/service.py	2011-07-25 16:11:30 +0000
@@ -0,0 +1,130 @@
+import socket
+import ConfigParser
+
+from twisted.application.internet import TCPClient, UDPServer
+from twisted.application.service import MultiService
+from twisted.internet import reactor
+from twisted.python import usage, util
+
+from txstatsd import process
+from txstatsd.client import InternalClient
+from txstatsd.metrics.metrics import Metrics
+from txstatsd.server.processor import MessageProcessor
+from txstatsd.server.protocol import GraphiteClientFactory, StatsDServerProtocol
+from txstatsd.report import ReportingService
+
+_unset = object()
+
+
+class OptionsGlue(usage.Options):
+    """
+    Extends usage.Options to also read parameters from a config file.
+    """
+
+    optParameters = []
+
+    def __init__(self):
+        self.glue_defaults = {}
+        self._config_file = None
+        config = ["config", "c", None, "Config file to use."]
+
+        new_params = []
+
+        def process_parameter(parameter):
+            long, short, default, doc, paramType = util.padTo(5, parameter)
+            self.glue_defaults[long] = default
+            new_params.append(
+                [long, short, _unset, doc, paramType])
+
+        for parameter in self.glue_parameters:
+            if parameter[0] == "config" or parameter[1] == "c":
+                raise ValueError("the --config/-c parameter is reserved.")
+            process_parameter(parameter)
+        process_parameter(config)
+
+        # we need to change self.__class__.optParameters as usage.Options
+        # will collect the config from there, not self.optParameters:
+        # reflect.accumulateClassList(self.__class__, 'optParameters',
+        #                            parameters)
+        self.__class__.optParameters = new_params
+
+        super(OptionsGlue, self).__init__()
+
+    def __getitem__(self, item):
+        result = super(OptionsGlue, self).__getitem__(item)
+        if result is not _unset:
+            return result
+
+        fname = super(OptionsGlue, self).__getitem__("config")
+        if fname is not _unset:
+            self._config_file = ConfigParser.RawConfigParser()
+            self._config_file.read(fname)
+
+        if self._config_file is not None:
+            try:
+                result = self._config_file.get("statsd", item)
+            except ConfigParser.NoOptionError:
+                pass
+            else:
+                if item in self._dispatch:
+                    result = self._dispatch[item].coerce(result)
+                return result
+
+        return self.glue_defaults[item]
+
+
+class StatsDOptions(OptionsGlue):
+    """
+    The set of configuration settings for txStatsD.
+    """
+    glue_parameters = [
+        ["carbon-cache-host", "h", "localhost",
+         "The host where carbon cache is listening."],
+        ["carbon-cache-port", "p", 2003,
+         "The port where carbon cache is listening.", int],
+        ["listen-port", "l", 8125,
+         "The UDP port where we will listen.", int],
+        ["flush-interval", "i", 10000,
+         "The number of milliseconds between each flush.", int],
+        ["prefix", "p", None,
+         "Prefix to use when reporting stats.", str],
+        ["report", "r", None,
+         "Which additional stats to report {process|net|io|system}.", str],
+    ]
+
+
+def createService(options):
+    """Create a txStatsD service."""
+
+    service = MultiService()
+    service.setName("statsd")
+    processor = MessageProcessor()
+    prefix = options["prefix"]
+    if prefix is None:
+        prefix = socket.gethostname() + ".statsd"
+
+    connection = InternalClient(processor)
+    metrics = Metrics(connection, namespace=prefix)
+
+    if options["report"] is not None:
+        reporting = ReportingService()
+        reporting.setServiceParent(service)
+        reporting.schedule(
+            process.report_reactor_stats(reactor), 10, metrics.increment)
+        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, 10, metrics.increment)
+
+    factory = GraphiteClientFactory(processor, options["flush-interval"])
+    client = TCPClient(options["carbon-cache-host"],
+                       options["carbon-cache-port"],
+                       factory)
+    client.setServiceParent(service)
+
+    listener = UDPServer(options["listen-port"],
+        StatsDServerProtocol(processor))
+    listener.setServiceParent(service)
+
+    return service

=== removed file 'txstatsd/service.py'
--- txstatsd/service.py	2011-07-12 04:54:49 +0000
+++ txstatsd/service.py	1970-01-01 00:00:00 +0000
@@ -1,128 +0,0 @@
-import socket
-import ConfigParser
-
-from twisted.application.internet import TCPClient, UDPServer
-from twisted.application.service import MultiService
-from twisted.internet import reactor
-from twisted.python import usage, util
-
-from txstatsd import process
-from txstatsd.metrics import InProcessMeter
-from txstatsd.processor import MessageProcessor
-from txstatsd.protocol import GraphiteClientFactory, StatsDServerProtocol
-from txstatsd.report import ReportingService
-
-_unset = object()
-
-
-class OptionsGlue(usage.Options):
-    """
-    Extends usage.Options to also read parameters from a config file.
-    """
-
-    optParameters = []
-
-    def __init__(self):
-        self.glue_defaults = {}
-        self._config_file = None
-        config = ["config", "c", None, "Config file to use."]
-
-        new_params = []
-
-        def process_parameter(parameter):
-            long, short, default, doc, paramType = util.padTo(5, parameter)
-            self.glue_defaults[long] = default
-            new_params.append(
-                [long, short, _unset, doc, paramType])
-
-        for parameter in self.glue_parameters:
-            if parameter[0] == "config" or parameter[1] == "c":
-                raise ValueError("the --config/-c parameter is reserved.")
-            process_parameter(parameter)
-        process_parameter(config)
-
-        # we need to change self.__class__.optParameters as usage.Options
-        # will collect the config from there, not self.optParameters:
-        # reflect.accumulateClassList(self.__class__, 'optParameters',
-        #                            parameters)
-        self.__class__.optParameters = new_params
-
-        super(OptionsGlue, self).__init__()
-
-    def __getitem__(self, item):
-        result = super(OptionsGlue, self).__getitem__(item)
-        if result is not _unset:
-            return result
-
-        fname = super(OptionsGlue, self).__getitem__("config")
-        if fname is not _unset:
-            self._config_file = ConfigParser.RawConfigParser()
-            self._config_file.read(fname)
-
-        if self._config_file is not None:
-            try:
-                result = self._config_file.get("statsd", item)
-            except ConfigParser.NoOptionError:
-                pass
-            else:
-                if item in self._dispatch:
-                    result = self._dispatch[item].coerce(result)
-                return result
-
-        return self.glue_defaults[item]
-
-
-class StatsDOptions(OptionsGlue):
-    """
-    The set of configuration settings for txStatsD.
-    """
-    glue_parameters = [
-        ["carbon-cache-host", "h", "localhost",
-         "The host where carbon cache is listening."],
-        ["carbon-cache-port", "p", 2003,
-         "The port where carbon cache is listening.", int],
-        ["listen-port", "l", 8125,
-         "The UDP port where we will listen.", int],
-        ["flush-interval", "i", 10000,
-         "The number of milliseconds between each flush.", int],
-        ["prefix", "p", None,
-         "Prefix to use when reporting stats.", str],
-        ["report", "r", None,
-         "Which additional stats to report {process|net|io|system}.", str],
-    ]
-
-
-def createService(options):
-    """Create a txStatsD service."""
-
-    service = MultiService()
-    service.setName("statsd")
-    processor = MessageProcessor()
-    prefix = options["prefix"]
-    if prefix is None:
-        prefix = socket.gethostname() + ".statsd"
-
-    meter = InProcessMeter(processor, prefix=prefix)
-
-    if options["report"] is not None:
-        reporting = ReportingService()
-        reporting.setServiceParent(service)
-        reporting.schedule(
-            process.report_reactor_stats(reactor), 10, meter.increment)
-        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, 10, meter.increment)
-
-    factory = GraphiteClientFactory(processor, options["flush-interval"])
-    client = TCPClient(options["carbon-cache-host"],
-                       options["carbon-cache-port"],
-                       factory)
-    client.setServiceParent(service)
-
-    listener = UDPServer(options["listen-port"],
-        StatsDServerProtocol(processor))
-    listener.setServiceParent(service)
-
-    return service

=== modified file 'txstatsd/tests/test_metrics.py'
--- txstatsd/tests/test_metrics.py	2011-06-21 17:14:01 +0000
+++ txstatsd/tests/test_metrics.py	2011-07-25 16:11:30 +0000
@@ -1,72 +1,60 @@
-"""Tests for metrics context manager."""
-
-from mocker import ANY, MockerTestCase
-from txstatsd.metrics import Measure
-
-
-class MeasureTest(MockerTestCase):
-    """Test case for the L{Measure} context manager."""
-
-    def test_measure(self):
-        """Basic test."""
-        operation_name = 'fake_function'
-        meter = self.mocker.mock()
-        meter.increment(operation_name)
-        meter.timing(operation_name, ANY)
-        meter.increment(operation_name + '.done')
-        meter.decrement(operation_name)
-        self.mocker.replay()
-        result = []
-
-        def fake_function():
-            """Useles method."""
-            result.append(0)
-
-        with Measure('test_prefix', 'fake_function', meter):
-            fake_function()
-        self.assertEqual([0], result)
-
-    def test_measure_timing(self):
-        """Test the timing works."""
-        operation_name = 'fake_function'
-        MockTime = self.mocker.replace('time.time')  # pylint: disable=C0103
-        self.expect(MockTime()).result(10)
-        self.expect(MockTime()).result(15)
-        meter = self.mocker.mock()
-        meter.increment(operation_name)
-        meter.timing(operation_name, 5)
-        meter.increment(operation_name + '.done')
-        meter.decrement(operation_name)
-        self.mocker.replay()
-
-        def fake_function():
-            """Useless method."""
-
-        with Measure('test_prefix', 'fake_function', meter):
-            fake_function()
-
-    def test_measure_handle_exceptions(self):
-        """Test exceptions."""
-
-        class TestException(Exception):
-            """Exception used to test the wrapper."""
-            pass
-
-        operation_name = 'fake_function'
-        meter = self.mocker.mock()
-        meter.increment(operation_name)
-        meter.increment(operation_name + '.error')
-        meter.decrement(operation_name)
-        self.mocker.replay()
-
-        def fake_function():
-            """Fake Method that raises an exception."""
-            raise TestException()
-
-        try:
-            with Measure('test_prefix', 'fake_function', meter):
-                fake_function()
-        except TestException:
-            self.assertTrue(True)
-        else:
-            self.fail('measure context manager did not reraise exception.')
+"""Tests for the Metrics convenience class."""
+
+from unittest import TestCase
+
+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_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', 101123)
+        self.assertEqual(self.connection.data,
+                         'txstatsd.tests.timing:101123|ms')
+
+    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')
+
+        self.metrics.namespace = ''
+        self.metrics.gauge('gauge', 413)
+        self.assertEqual(self.connection.data,
+                         'gauge:413|g')

=== modified file 'txstatsd/tests/test_processor.py'
--- txstatsd/tests/test_processor.py	2011-06-23 14:24:04 +0000
+++ txstatsd/tests/test_processor.py	2011-07-25 16:11:30 +0000
@@ -1,6 +1,6 @@
 from unittest import TestCase
 
-from txstatsd.processor import MessageProcessor
+from txstatsd.server.processor import MessageProcessor
 
 
 class TestMessageProcessor(MessageProcessor):
@@ -25,8 +25,8 @@
         unit. 'c' is simply used to signal that this is a counter message.
         """
         self.processor.process("gorets:1|c")
-        self.assertEqual(1, len(self.processor.counters))
-        self.assertEqual(1.0, self.processor.counters["gorets"])
+        self.assertEqual(1, len(self.processor.counter_metrics))
+        self.assertEqual(1.0, self.processor.counter_metrics["gorets"])
 
     def test_receive_counter_rate(self):
         """
@@ -37,8 +37,8 @@
         counter value.
         """
         self.processor.process("gorets:1|c|@0.1")
-        self.assertEqual(1, len(self.processor.counters))
-        self.assertEqual(10.0, self.processor.counters["gorets"])
+        self.assertEqual(1, len(self.processor.counter_metrics))
+        self.assertEqual(10.0, self.processor.counter_metrics["gorets"])
 
     def test_receive_timer(self):
         """
@@ -46,16 +46,28 @@
         identifier and '320' is the time in milliseconds.
         """
         self.processor.process("glork:320|ms")
-        self.assertEqual(1, len(self.processor.timers))
-        self.assertEqual([320], self.processor.timers["glork"])
+        self.assertEqual(1, len(self.processor.timer_metrics))
+        self.assertEqual([320], self.processor.timer_metrics["glork"])
+
+    def test_receive_gauge_metric(self):
+        """
+        A gauge metric message takes the form:
+        '<name>:<count>|g'.
+        'g' indicates this is a gauge metric message.
+        """
+        self.processor.process("gorets:9.6|g")
+        self.assertEqual(1, len(self.processor.gauge_metrics))
+        self.assertEqual(
+            [9.6, 'gorets'],
+            self.processor.gauge_metrics.pop())
 
     def test_receive_message_no_fields(self):
         """
         If a timer message has no fields, it is logged and discarded.
         """
         self.processor.process("glork")
-        self.assertEqual(0, len(self.processor.timers))
-        self.assertEqual(0, len(self.processor.counters))
+        self.assertEqual(0, len(self.processor.timer_metrics))
+        self.assertEqual(0, len(self.processor.counter_metrics))
         self.assertEqual(["glork"], self.processor.failures)
 
     def test_receive_counter_no_value(self):
@@ -63,7 +75,7 @@
         If a counter message has no value, it is logged and discarded.
         """
         self.processor.process("gorets:|c")
-        self.assertEqual(0, len(self.processor.counters))
+        self.assertEqual(0, len(self.processor.counter_metrics))
         self.assertEqual(["gorets:|c"], self.processor.failures)
 
     def test_receive_timer_no_value(self):
@@ -71,7 +83,7 @@
         If a timer message has no value, it is logged and discarded.
         """
         self.processor.process("glork:|ms")
-        self.assertEqual(0, len(self.processor.timers))
+        self.assertEqual(0, len(self.processor.timer_metrics))
         self.assertEqual(["glork:|ms"], self.processor.failures)
 
     def test_receive_not_enough_fields(self):
@@ -79,8 +91,8 @@
         If a timer message has not enough fields, it is logged and discarded.
         """
         self.processor.process("glork:1")
-        self.assertEqual(0, len(self.processor.timers))
-        self.assertEqual(0, len(self.processor.counters))
+        self.assertEqual(0, len(self.processor.timer_metrics))
+        self.assertEqual(0, len(self.processor.counter_metrics))
         self.assertEqual(["glork:1"], self.processor.failures)
 
     def test_receive_too_many_fields(self):
@@ -88,8 +100,8 @@
         If a timer message has too many fields, it is logged and discarded.
         """
         self.processor.process("gorets:1|c|@0.1|yay")
-        self.assertEqual(0, len(self.processor.timers))
-        self.assertEqual(0, len(self.processor.counters))
+        self.assertEqual(0, len(self.processor.timer_metrics))
+        self.assertEqual(0, len(self.processor.counter_metrics))
         self.assertEqual(["gorets:1|c|@0.1|yay"], self.processor.failures)
 
 
@@ -110,28 +122,28 @@
         If a counter is present, flushing it will generate a counter message
         normalized to the default interval.
         """
-        self.processor.counters["gorets"] = 42
+        self.processor.counter_metrics["gorets"] = 42
         messages = self.processor.flush()
         self.assertEqual(2, len(messages))
         counters = messages[0].splitlines()
         self.assertEqual("stats.gorets 4 42", counters[0])
         self.assertEqual("stats_counts.gorets 42 42", counters[1])
         self.assertEqual("statsd.numStats 1 42", messages[1])
-        self.assertEqual(0, self.processor.counters["gorets"])
+        self.assertEqual(0, self.processor.counter_metrics["gorets"])
 
     def test_flush_counter_one_second_interval(self):
         """
         It is possible to flush counters with a one-second interval, in which
         case the counter value will be unchanged.
         """
-        self.processor.counters["gorets"] = 42
+        self.processor.counter_metrics["gorets"] = 42
         messages = self.processor.flush(interval=1000)
         self.assertEqual(2, len(messages))
         counters = messages[0].splitlines()
         self.assertEqual("stats.gorets 42 42", counters[0])
         self.assertEqual("stats_counts.gorets 42 42", counters[1])
         self.assertEqual("statsd.numStats 1 42", messages[1])
-        self.assertEqual(0, self.processor.counters["gorets"])
+        self.assertEqual(0, self.processor.counter_metrics["gorets"])
 
     def test_flush_single_timer_single_time(self):
         """
@@ -139,7 +151,7 @@
         threshold_upper, lower, mean will be set to the same value. Timer is
         reset after flush is called.
         """
-        self.processor.timers["glork"] = [24]
+        self.processor.timer_metrics["glork"] = [24]
         messages = self.processor.flush()
         self.assertEqual(2, len(messages))
         timers = messages[0].splitlines()
@@ -149,7 +161,7 @@
         self.assertEqual("stats.timers.glork.lower 24 42", timers[3])
         self.assertEqual("stats.timers.glork.count 1 42", timers[4])
         self.assertEqual("statsd.numStats 1 42", messages[1])
-        self.assertEqual([], self.processor.timers["glork"])
+        self.assertEqual([], self.processor.timer_metrics["glork"])
 
     def test_flush_single_timer_multiple_times(self):
         """
@@ -160,7 +172,7 @@
         - count will be the count of data points
         - mean will be the mean value within the 90th percentile
         """
-        self.processor.timers["glork"] = [4, 8, 15, 16, 23, 42]
+        self.processor.timer_metrics["glork"] = [4, 8, 15, 16, 23, 42]
         messages = self.processor.flush()
         self.assertEqual(2, len(messages))
         timers = messages[0].splitlines()
@@ -170,7 +182,7 @@
         self.assertEqual("stats.timers.glork.lower 4 42", timers[3])
         self.assertEqual("stats.timers.glork.count 6 42", timers[4])
         self.assertEqual("statsd.numStats 1 42", messages[1])
-        self.assertEqual([], self.processor.timers["glork"])
+        self.assertEqual([], self.processor.timer_metrics["glork"])
 
     def test_flush_single_timer_50th_percentile(self):
         """
@@ -184,7 +196,7 @@
         - count will be the count of data points
         - mean will be the mean value within the 50th percentile
         """
-        self.processor.timers["glork"] = [4, 8, 15, 16, 23, 42]
+        self.processor.timer_metrics["glork"] = [4, 8, 15, 16, 23, 42]
         messages = self.processor.flush(percent=50)
         self.assertEqual(2, len(messages))
         timers = messages[0].splitlines()
@@ -194,4 +206,21 @@
         self.assertEqual("stats.timers.glork.lower 4 42", timers[3])
         self.assertEqual("stats.timers.glork.count 6 42", timers[4])
         self.assertEqual("statsd.numStats 1 42", messages[1])
-        self.assertEqual([], self.processor.timers["glork"])
+        self.assertEqual([], self.processor.timer_metrics["glork"])
+
+    def test_flush_gauge_metric(self):
+        """
+        Test the correct rendering of the Graphite report for
+        a gauge metric.
+        """
+
+        self.processor.process("gorets:9.6|g")
+
+        messages = self.processor.flush()
+        self.assertEqual(2, len(messages))
+        gauge_metric = messages[0].splitlines()
+        self.assertEqual(
+            "stats.gauge.gorets.value 9.6 42", gauge_metric[0])
+        self.assertEqual(
+            "statsd.numStats 1 42", messages[1])
+        self.assertEqual(0, len(self.processor.gauge_metrics))

=== modified file 'txstatsd/version.py'
--- txstatsd/version.py	2011-06-20 19:11:22 +0000
+++ txstatsd/version.py	2011-07-25 16:11:30 +0000
@@ -1,1 +1,1 @@
-txstatsd = "0.1.0"
+txstatsd = "0.1.1"