Merge lp:~sidnei/txstatsd/derive-metric into lp:txstatsd

[Sidnei da Silva]

Replace meter metric by a simpler direct rate metric not using EWMA.

Replace meter metric by a simpler direct rate metric not using EWMA.

[Lucio Torre (lucio.torre)]

review: Approve

[lp:~sidnei/txstatsd/derive-metric]

95. By Sidnei da Silva on 2012-05-24

- Use provided timestamp, cleanup some formatting.

96. By Sidnei da Silva on 2012-05-24

- Restore accidentally deleted file

97. By Sidnei da Silva on 2012-05-24

- More formatting fixes

98. By Mike Kazantsev on 2012-05-24

Add Mike Kazantsev as author.

=== modified file 'txstatsd/metrics/metermetric.py'
--- txstatsd/metrics/metermetric.py	2012-02-08 23:30:35 +0000
+++ txstatsd/metrics/metermetric.py	2012-05-24 19:44:17 +0000
@@ -1,17 +1,12 @@
 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>}
+    A simplier meter metric which measures instant throughput rate for each
+    interval.
     """
 
     def __init__(self, connection, name, sample_rate=1):
@@ -33,12 +28,8 @@
 
 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>}
+    A simplier meter metric which measures instant throughput rate for each
+    interval.
     """
 
     def __init__(self, name, wall_time_func=time.time, prefix=""):
@@ -54,50 +45,40 @@
             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()
+        self.value = self.count = 0
+        self.poll_time = self.wall_time_func()
+
+    def mark(self, value):
+        """
+        Process new data for this metric.
+
+        @type value: C{float}
+        @param value: The reported value, to be aggregate into the meter.
+        """
+        self.value += value
 
     def report(self, timestamp):
+        """
+        Returns a list of metrics to report.
+
+        @type timestamp: C{float}
+        @param timestamp: The timestamp for now.
+        """
+        poll_prev, self.poll_time = self.poll_time, timestamp
+
+        if self.poll_time == poll_prev:
+            return list()
+
+        rate = float(self.value) / (self.poll_time - poll_prev)
+        self.count, self.value = self.count + self.value, 0
+
         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()}
+        items = {
+            ".count": self.count,
+            ".rate": rate
+            }
 
-        for item, value in items.iteritems():
+        for item, value in sorted(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

=== modified file 'txstatsd/metrics/timermetric.py'
--- txstatsd/metrics/timermetric.py	2012-05-22 19:12:00 +0000
+++ txstatsd/metrics/timermetric.py	2012-05-24 19:44:17 +0000
@@ -131,9 +131,6 @@
         if duration >= 0:
             self.histogram.update(duration)
 
-    def tick(self):
-        pass
-
     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)

=== modified file 'txstatsd/server/configurableprocessor.py'
--- txstatsd/server/configurableprocessor.py	2012-05-22 19:12:00 +0000
+++ txstatsd/server/configurableprocessor.py	2012-05-24 19:44:17 +0000
@@ -101,9 +101,3 @@
             events += 1
 
         return (metrics, events)
-
-    def update_metrics(self):
-        super(ConfigurableMessageProcessor, self).update_metrics()
-
-        for metric in self.timer_metrics.itervalues():
-            metric.tick()

=== modified file 'txstatsd/server/loggingprocessor.py'
--- txstatsd/server/loggingprocessor.py	2011-12-02 17:09:32 +0000
+++ txstatsd/server/loggingprocessor.py	2012-05-24 19:44:17 +0000
@@ -1,4 +1,3 @@
-
 import time
 
 from txstatsd.server.configurableprocessor import ConfigurableMessageProcessor
@@ -11,33 +10,26 @@
     attribute.)
     """
 
-    def __init__(self, logger, time_function=time.time, message_prefix="", plugins=None):
+    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)
+            plugins=plugins, **kwz)
 
-        logger_info = getattr(logger, 'info', None)
+        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 log file."""
-        timestamp = int(self.time_function())
-        interval = interval / 1000
-
-        def log_metrics(metrics):
-            for metric in metrics.itervalues():
-                messages = metric.report(timestamp)
-                for measurement in messages:
-                    self.logger.info("%s %s %s" % measurement)
-
-        log_metrics(self.counter_metrics)
-        log_metrics(self.gauge_metrics)
-        log_metrics(self.meter_metrics)
-        log_metrics(self.timer_metrics)
-        
-        for metric in self.plugin_metrics.itervalues():
-            for measurement in metric.flush(interval, timestamp):
-                self.logger.info("%s %s %s" % measurement)
-
+        """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

=== modified file 'txstatsd/server/processor.py'
--- txstatsd/server/processor.py	2012-04-26 19:21:24 +0000
+++ txstatsd/server/processor.py	2012-05-24 19:44:17 +0000
@@ -375,7 +375,3 @@
 
         self.process_timings.clear()
         self.by_type.clear()
-
-    def update_metrics(self):
-        for metric in self.meter_metrics.itervalues():
-            metric.tick()

=== modified file 'txstatsd/service.py'
--- txstatsd/service.py	2012-02-08 23:30:59 +0000
+++ txstatsd/service.py	2012-05-24 19:44:17 +0000
@@ -4,6 +4,7 @@
 import time
 import ConfigParser
 import platform
+import functools
 
 from twisted.application.internet import UDPServer, TCPServer
 from twisted.application.service import MultiService
@@ -15,6 +16,7 @@
 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
@@ -150,6 +152,9 @@
          "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,
@@ -246,13 +251,21 @@
         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 = MessageProcessor(plugins=plugin_metrics)
+        processor = (processor or MessageProcessor)(plugins=plugin_metrics)
         input_router = Router(processor, options['routing'], root_service)
         connection = InternalClient(input_router)
         metrics = Metrics(connection)
     else:
-        processor = ConfigurableMessageProcessor(
+        processor = (processor or ConfigurableMessageProcessor)(
             message_prefix=prefix,
             internal_metrics_prefix=prefix + "." + instance_name + ".",
             plugins=plugin_metrics)
@@ -270,9 +283,6 @@
     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)

=== added file 'txstatsd/tests/metrics/test_metermetric.py'
--- txstatsd/tests/metrics/test_metermetric.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/metrics/test_metermetric.py	2012-05-24 19:44:17 +0000
@@ -0,0 +1,33 @@
+import random
+from twisted.trial.unittest import TestCase
+
+from txstatsd.metrics.metermetric import MeterMetricReporter
+
+
+class TestDeriveMetricReporter(TestCase):
+
+    def test_fastpoll(self):
+        wall_time = 42
+        reporter = MeterMetricReporter(
+            "test", wall_time_func=lambda: wall_time)
+
+        self.assertEquals([], reporter.report(wall_time))
+
+    def test_interface(self):
+        random.seed(1)
+
+        wall_time = [0]
+        reporter = MeterMetricReporter("test", prefix="some.prefix",
+                                       wall_time_func=lambda: wall_time[0])
+        reporter.mark(42)
+        reporter.mark(60)
+        reporter.mark(38)
+        wall_time = [10]
+
+        reported = reporter.report(10)
+        self.assertEqual(2, len(reported))
+        self.assertEqual(140, reported[0][1])
+        self.assertEqual(14, reported[1][1])
+        self.assertEquals(
+            ['some.prefix.test.count', 'some.prefix.test.rate'],
+            [reported[0][0], reported[1][0]])

=== modified file 'txstatsd/tests/test_configurableprocessor.py'
--- txstatsd/tests/test_configurableprocessor.py	2012-05-22 19:12:00 +0000
+++ txstatsd/tests/test_configurableprocessor.py	2012-05-24 19:44:17 +0000
@@ -100,17 +100,11 @@
             time_function=lambda: _now)
 
         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()
 
         _now = 42
         messages = configurable_processor.flush()
@@ -155,13 +149,7 @@
 
         self.time_now += 1
         messages = self.configurable_processor.flush()
-        self.assertEqual(("test.metric.gorets.15min_rate", 0.0, self.time_now),
+        self.assertEqual(("test.metric.gorets.count", 3.0, self.time_now),
                          messages[0])
-        self.assertEqual(("test.metric.gorets.1min_rate", 0.0, self.time_now),
+        self.assertEqual(("test.metric.gorets.rate", 3.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])

=== modified file 'txstatsd/tests/test_loggingprocessor.py'
--- txstatsd/tests/test_loggingprocessor.py	2011-12-02 17:09:32 +0000
+++ txstatsd/tests/test_loggingprocessor.py	2012-05-24 19:44:17 +0000
@@ -1,4 +1,3 @@
-
 from unittest import TestCase
 
 from twisted.plugins.distinct_plugin import distinct_metric_factory
@@ -44,20 +43,20 @@
         metric = FakeMeterMetric()
         processor.meter_metrics['test'] = metric
         processor.flush()
-        expected = "\n".join(["%s %s %s" % message
-                              for message in metric.report()])
-        self.assertEqual(expected + "\n", logger.log)
+        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)
-        processor.process("gorets:17|pd")
+        msg_in = "gorets:17|pd"
+        processor.process(msg_in)
         processor.flush()
         messages = processor.plugin_metrics['gorets'].flush(
             10, processor.time_function())
-        expected = "\n".join(["%s %s %s" % message
-                              for message in messages])
-        self.assertEqual(expected + "\n", logger.log)
-
+        expected = ["In: %s" % msg_in] + ["Out: %s %s %s" % message
+                                          for message in messages]
+        self.assertFalse(set(expected).difference(logger.log.splitlines()))

=== modified file 'txstatsd/tests/test_processor.py'
--- txstatsd/tests/test_processor.py	2012-04-26 19:21:24 +0000
+++ txstatsd/tests/test_processor.py	2012-05-24 19:44:17 +0000
@@ -355,10 +355,6 @@
     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
@@ -369,45 +365,22 @@
         self.time_now += 1
         messages = self.processor.flush()
         self.assertEqual(
-            ("stats.meter.gorets.15min_rate", 0.0, self.time_now),
+            ("stats.meter.gorets.count", 3.0, self.time_now),
             messages[0])
         self.assertEqual(
-            ("stats.meter.gorets.1min_rate", 0.0, self.time_now),
+            ("stats.meter.gorets.rate", 3.0, self.time_now),
             messages[1])
         self.assertEqual(
-            ("stats.meter.gorets.5min_rate", 0.0, self.time_now),
-            messages[2])
-        self.assertEqual(
-            ("stats.meter.gorets.count", 3.0, self.time_now),
-            messages[3])
-        self.assertEqual(
-            ("stats.meter.gorets.mean_rate", 3.0, self.time_now),
-            messages[4])
-        self.assertEqual(
             ("statsd.numStats", 1, self.time_now),
-            messages[5])
-
-        # 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)
+            messages[2])
+
         self.time_now += 60
         messages = self.processor.flush()
         self.assertEqual(
-            ("stats.meter.gorets.15min_rate", 0.561304, self.time_now),
+            ("stats.meter.gorets.count", 3.0, self.time_now),
             messages[0])
         self.assertEqual(
-            ("stats.meter.gorets.1min_rate", 0.220728, self.time_now),
+            ("stats.meter.gorets.rate", 0.0, self.time_now),
             messages[1])
         self.assertEqual(
-            ("stats.meter.gorets.5min_rate", 0.491238, self.time_now),
-            messages[2])
-        self.assertEqual(
-            ("stats.meter.gorets.count", 3.0, self.time_now),
-            messages[3])
-        self.assertEqual(
-            ("stats.meter.gorets.mean_rate", 0.049180, self.time_now),
-            messages[4])
-        self.assertEqual(
-            ("statsd.numStats", 1, self.time_now), messages[5])
+            ("statsd.numStats", 1, self.time_now), messages[2])

=== modified file 'txstatsd/tests/test_service.py'
--- txstatsd/tests/test_service.py	2012-02-07 21:11:42 +0000
+++ txstatsd/tests/test_service.py	2012-05-24 19:44:17 +0000
@@ -1,13 +1,13 @@
+import tempfile
 import ConfigParser
-import tempfile
-from unittest import TestCase
 from StringIO import StringIO
 
+from twisted.trial.unittest import TestCase
+
 from carbon.client import CarbonClientManager
 
 from twisted.internet.defer import inlineCallbacks, Deferred
 from twisted.internet.protocol import DatagramProtocol
-from twisted.internet.test.reactormixins import ReactorBuilder
 from twisted.application.internet import UDPServer
 
 from txstatsd import service
@@ -186,7 +186,7 @@
         self.monitor_response = data
 
 
-class ServiceTestsBuilder(ReactorBuilder):
+class ServiceTestsBuilder(TestCase):
 
     def test_service(self):
         """
@@ -243,7 +243,7 @@
         The StatsD service messages the expected response to the
         monitoring agent.
         """
-        reactor = self.buildReactor()
+        from twisted.internet import reactor
 
         options = service.StatsDOptions()
         processor = MessageProcessor()
@@ -280,7 +280,5 @@
                 reactor.stop()
 
         reactor.callWhenRunning(exercise)
-        self.runReactor(reactor)
-
-
-globals().update(ServiceTestsBuilder.makeTestCaseClasses())
+        reactor.run()
+