Merge lp:~theiw/txstatsd/txstatsd-metermetric into lp:txstatsd

[Ian Wilkinson]

Introduce support for Coda Hale's meter metric.

[Lucio Torre (lucio.torre)]

review: Approve

[Sidnei da Silva (sidnei)]

im under the impression that those would be better implemented as functions on the graphite side, but other than that looks good to me. +1

review: Approve

=== modified file 'README'
--- README	2011-06-20 19:11:22 +0000
+++ README	2011-08-26 12:28:49 +0000
@@ -11,6 +11,9 @@
 * 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
 -------
 
 
=== added file 'txstatsd/metrics/metermetric.py'
--- txstatsd/metrics/metermetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/metermetric.py	2011-08-26 12:28:49 +0000
@@ -0,0 +1,104 @@
+
+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>}
+    """
+
+    MESSAGE = (
+        "stats.meter.%(key)s.count %(count)s %(timestamp)s\n"
+        "stats.meter.%(key)s.mean_rate %(mean_rate)s %(timestamp)s\n"
+        "stats.meter.%(key)s.1min_rate %(rate_1min)s %(timestamp)s\n"
+        "stats.meter.%(key)s.5min_rate %(rate_5min)s %(timestamp)s\n"
+        "stats.meter.%(key)s.15min_rate %(rate_15min)s %(timestamp)s\n")
+
+    def __init__(self, name, wall_time_func=time.time):
+        """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
+
+        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):
+        return MeterMetricReporter.MESSAGE % {
+            "key": self.name,
+            "count": self.count,
+            "mean_rate": self.mean_rate(),
+            "rate_1min": self.one_minute_rate(),
+            "rate_5min": self.five_minute_rate(),
+            "rate_15min": self.fifteen_minute_rate(),
+            "timestamp": timestamp}
+
+    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
 
=== modified file 'txstatsd/metrics/metrics.py'
--- txstatsd/metrics/metrics.py	2011-07-27 15:02:32 +0000
+++ txstatsd/metrics/metrics.py	2011-08-26 12:28:49 +0000
@@ -1,5 +1,6 @@
 
 from txstatsd.metrics.gaugemetric import GaugeMetric
+from txstatsd.metrics.metermetric import MeterMetric
 from txstatsd.metrics.metric import Metric
 
 
@@ -28,6 +29,16 @@
             self._metrics[name] = gauge_metric
         self._metrics[name].mark(value)
 
+    def meter(self, name, value, 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)
 
=== modified file 'txstatsd/server/processor.py'
--- txstatsd/server/processor.py	2011-07-25 16:09:37 +0000
+++ txstatsd/server/processor.py	2011-08-26 12:28:49 +0000
@@ -5,6 +5,8 @@
 
 from twisted.python import log
 
+from txstatsd.metrics.metermetric import MeterMetricReporter
+
 
 SPACES = re.compile("\s+")
 SLASHES = re.compile("\/+")
@@ -43,6 +45,7 @@
         self.timer_metrics = {}
         self.counter_metrics = {}
         self.gauge_metrics = deque()
+        self.meter_metrics = {}
 
     def fail(self, message):
         """Log and discard malformed message."""
@@ -72,6 +75,8 @@
             self.process_timer_metric(key, fields[0], message)
         elif fields[1] == "g":
             self.process_gauge_metric(key, fields[0], message)
+        elif fields[1] == "m":
+            self.process_meter_metric(key, fields[0], message)
         else:
             return self.fail(message)
 
@@ -110,6 +115,21 @@
         except (TypeError, ValueError):
             self.fail(message)
 
+    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)
+
+        if not key in self.meter_metrics:
+            metric = MeterMetricReporter(key, self.time_function)
+            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
@@ -136,6 +156,11 @@
             messages.extend(gauge_metrics)
             num_stats += events
 
+        meter_metrics, events = self.flush_meter_metrics(timestamp)
+        if events > 0:
+            messages.extend(meter_metrics)
+            num_stats += events
+
         messages.append("statsd.numStats %s %s" % (num_stats, timestamp))
         return messages
 
@@ -211,3 +236,17 @@
         self.gauge_metrics.clear()
 
         return (metrics, events)
+
+    def flush_meter_metrics(self, timestamp):
+        metrics = []
+        events = 0
+        for metric in self.meter_metrics.itervalues():
+            message = metric.report(timestamp)
+            metrics.append(message)
+            events += 1
+
+        return (metrics, events)
+
+    def update_metrics(self):
+        for metric in self.meter_metrics.itervalues():
+            metric.tick()
 
=== modified file 'txstatsd/service.py'
--- txstatsd/service.py	2011-07-27 20:31:33 +0000
+++ txstatsd/service.py	2011-08-26 12:28:49 +0000
@@ -117,6 +117,12 @@
                                     report_name.upper(), ()):
                 reporting.schedule(reporter, 10, metrics.gauge)
 
+    # Schedule updates for those metrics expecting to be
+    # periodically updated, for example the meter metric.
+    metrics_updater = ReportingService()
+    metrics_updater.setServiceParent(service)
+    metrics_updater.schedule(processor.update_metrics, 5, None)
+
     factory = GraphiteClientFactory(processor, options["flush-interval"])
     client = TCPClient(options["carbon-cache-host"],
                        options["carbon-cache-port"],
 
=== 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	2011-08-26 12:28:49 +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 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	2011-08-26 12:28:49 +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')
 
=== modified file 'txstatsd/tests/test_metrics.py'
--- txstatsd/tests/test_metrics.py	2011-07-03 11:49:05 +0000
+++ txstatsd/tests/test_metrics.py	2011-08-26 12:28:49 +0000
@@ -32,6 +32,12 @@
         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)
 
=== modified file 'txstatsd/tests/test_processor.py'
--- txstatsd/tests/test_processor.py	2011-07-04 03:00:04 +0000
+++ txstatsd/tests/test_processor.py	2011-08-26 12:28:49 +0000
@@ -1,3 +1,5 @@
+import time
+
 from unittest import TestCase
 
 from txstatsd.server.processor import MessageProcessor
@@ -224,3 +226,73 @@
         self.assertEqual(
             "statsd.numStats 1 42", messages[1])
         self.assertEqual(0, len(self.processor.gauge_metrics))
+
+
+class FlushMeterMetricMessagesTest(TestCase):
+
+    def setUp(self):
+        self.processor = MessageProcessor(time_function=self.wall_clock_time)
+        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(self):
+        """
+        Test the correct rendering of the Graphite report for
+        a meter metric.
+        """
+        self.processor.process("gorets:3.0|m")
+
+        self.time_now += 1
+        messages = self.processor.flush()
+        self.assertEqual(2, len(messages))
+        meter_metric = messages[0].splitlines()
+        self.assertEqual(
+            "stats.meter.gorets.count 3.0 %s" % self.time_now,
+            meter_metric[0])
+        self.assertEqual(
+            "stats.meter.gorets.mean_rate 3.0 %s" % self.time_now,
+            meter_metric[1])
+        self.assertEqual(
+            "stats.meter.gorets.1min_rate 0.0 %s" % self.time_now,
+            meter_metric[2])
+        self.assertEqual(
+            "stats.meter.gorets.5min_rate 0.0 %s" % self.time_now,
+            meter_metric[3])
+        self.assertEqual(
+            "stats.meter.gorets.15min_rate 0.0 %s" % self.time_now,
+            meter_metric[4])
+        self.assertEqual(
+            "statsd.numStats 1 %s" % self.time_now, messages[1])
+
+        # As we are employing the expected results from test_ewma.py
+        # we perform the initial tick(), before advancing the clock 60sec.
+        self.processor.update_metrics()
+
+        self.mark_minutes(1)
+        self.time_now += 60
+        messages = self.processor.flush()
+        self.assertEqual(2, len(messages))
+        meter_metric = messages[0].splitlines()
+        self.assertEqual(
+            "stats.meter.gorets.count 3.0 %s" % self.time_now,
+            meter_metric[0])
+        self.assertTrue(
+            meter_metric[1].startswith(
+                "stats.meter.gorets.mean_rate 0.04918032"))
+        self.assertTrue(
+            meter_metric[2].startswith(
+                "stats.meter.gorets.1min_rate 0.22072766"))
+        self.assertTrue(
+            meter_metric[3].startswith(
+                "stats.meter.gorets.5min_rate 0.49123845"))
+        self.assertTrue(
+            meter_metric[4].startswith(
+                "stats.meter.gorets.15min_rate 0.5613041"))
+        self.assertEqual(
+            "statsd.numStats 1 %s" % self.time_now, messages[1])
 
=== modified file 'txstatsd/version.py'
--- txstatsd/version.py	2011-08-23 02:05:21 +0000
+++ txstatsd/version.py	2011-08-26 12:28:49 +0000
@@ -1,1 +1,1 @@
-txstatsd = "0.1.3"
+txstatsd = "0.2.0"