Merge lp:~wesmason/txstatsd/fix-counters into lp:txstatsd

[Wes Mason]

Fix ConfigurableMessageProcessor not processing counter stats correctly, and add rate tests.

This a fix for ConfigurableMessageProcessor to actually update the value for counters rather than overwriting, and flushes correctly too.

Patch by ~james-w, and I've tested it with values from canonistack deployed SSO and and manually with `nc`.

[James Westby (james-w)]

Hi,

Is there a test in the other processor for the RATE regex matching that we could
co-opt here? Or maybe there's even a way to re-use that code directly.

Thanks,

James

[Wes Mason (wesmason)]

Will investigate, I think we discussed that before the Christmas break but I'd forgotten.

Thanks!

[Wes Mason (wesmason)]

There's no ready-to-reuse RATE test so working on one.

[Wes Mason (wesmason)]

Added tests to ensure rate is picked from stat datagram and that both whole and fractional decimals are parsed (and applied to the rate correctly).

[James Westby (james-w)]

review: Approve

=== modified file 'txstatsd/metrics/countermetric.py'
--- txstatsd/metrics/countermetric.py	2012-06-28 17:29:26 +0000
+++ txstatsd/metrics/countermetric.py	2015-01-22 14:26:14 +0000
@@ -80,9 +80,12 @@
         self.prefix = prefix
         self.count = 0
 
-    def mark(self, value):
-        self.count = value
+    def update(self, value, rate):
+        self.count += value * (1 / float(rate))
 
     def report(self, timestamp):
         return [(self.prefix + self.name + ".count",
                  math.trunc(self.count), timestamp)]
+
+    def reset(self):
+        self.count = 0

=== modified file 'txstatsd/server/configurableprocessor.py'
--- txstatsd/server/configurableprocessor.py	2014-10-16 11:13:07 +0000
+++ txstatsd/server/configurableprocessor.py	2015-01-22 14:26:14 +0000
@@ -25,7 +25,10 @@
 from txstatsd.metrics.gaugemetric import GaugeMetricReporter
 from txstatsd.metrics.metermetric import MeterMetricReporter
 from txstatsd.metrics.timermetric import TimerMetricReporter
-from txstatsd.server.processor import MessageProcessor
+from txstatsd.server.processor import (
+    MessageProcessor,
+    RATE,
+)
 
 
 class ConfigurableMessageProcessor(MessageProcessor):
@@ -71,14 +74,20 @@
             value = float(composite[0])
         except (TypeError, ValueError):
             return self.fail(message)
-
-        self.compose_counter_metric(key, value)
-
-    def compose_counter_metric(self, key, value):
+        rate = 1
+        if len(composite) == 3:
+            match = RATE.match(composite[2])
+            if match is None:
+                return self.fail(message)
+            rate = match.group(1)
+
+        self.compose_counter_metric(key, value, rate)
+
+    def compose_counter_metric(self, key, value, rate):
         if not key in self.counter_metrics:
             metric = CounterMetricReporter(key, prefix=self.message_prefix)
             self.counter_metrics[key] = metric
-        self.counter_metrics[key].mark(value)
+        self.counter_metrics[key].update(value, rate)
 
     def compose_gauge_metric(self, key, value, delta=False):
         if not key in self.gauge_metrics:
@@ -96,6 +105,7 @@
     def flush_counter_metrics(self, interval, timestamp):
         for metric in self.counter_metrics.itervalues():
             messages = metric.report(timestamp)
+            metric.reset()
             yield messages
 
     def flush_gauge_metrics(self, timestamp):

=== modified file 'txstatsd/tests/test_configurableprocessor.py'
--- txstatsd/tests/test_configurableprocessor.py	2014-10-16 11:29:24 +0000
+++ txstatsd/tests/test_configurableprocessor.py	2015-01-22 14:26:14 +0000
@@ -20,7 +20,6 @@
 # SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 import time
-
 from unittest import TestCase
 
 from twisted.plugins.distinct_plugin import distinct_metric_factory
@@ -53,8 +52,7 @@
         configurable_processor.process("gorets:17|c")
         messages = list(configurable_processor.flush())
         self.assertEqual(("test.metric.gorets.count", 17, 42), messages[0])
-        self.assertEqual(("test.metric.statsd.numStats", 1, 42),
-                         messages[1])
+        self.assertEqual(("test.metric.statsd.numStats", 1, 42), messages[1])
 
     def test_flush_counter_with_internal_prefix(self):
         """
@@ -66,8 +64,62 @@
         configurable_processor.process("gorets:17|c")
         messages = list(configurable_processor.flush())
         self.assertEqual(("test.metric.gorets.count", 17, 42), messages[0])
-        self.assertEqual(("statsd.foo.numStats", 1, 42),
-                         messages[1])
+        self.assertEqual(("statsd.foo.numStats", 1, 42), messages[1])
+
+    def test_flush_counter_with_empty_prefix_with_rate(self):
+        """
+        Ensure no prefix features if none is supplied.
+        B{Note}: The C{ConfigurableMessageProcessor} reports
+        the counter value, and not the normalized version as
+        seen in the StatsD-compliant C{Processor}.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42)
+
+        configurable_processor.process("gorets:17|c|@2")
+        messages = list(configurable_processor.flush())
+        self.assertEqual(("gorets.count", 8, 42), messages[0])
+        self.assertEqual(("statsd.numStats", 1, 42), messages[1])
+
+        configurable_processor.process("gorets:17|c|@1.5")
+        messages = list(configurable_processor.flush())
+        self.assertEqual(("gorets.count", 11, 42), messages[0])
+        self.assertEqual(("statsd.numStats", 1, 42), messages[1])
+
+    def test_flush_counter_with_prefix_with_rate(self):
+        """
+        Ensure the prefix features if one is supplied.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42, message_prefix="test.metric")
+
+        configurable_processor.process("gorets:17|c|@2")
+        messages = list(configurable_processor.flush())
+        self.assertEqual(("test.metric.gorets.count", 8, 42), messages[0])
+        self.assertEqual(("test.metric.statsd.numStats", 1, 42), messages[1])
+
+        configurable_processor.process("gorets:17|c|@1.5")
+        messages = list(configurable_processor.flush())
+        self.assertEqual(("test.metric.gorets.count", 11, 42), messages[0])
+        self.assertEqual(("test.metric.statsd.numStats", 1, 42), messages[1])
+
+    def test_flush_counter_with_internal_prefix_with_rate(self):
+        """
+        Ensure the prefix features if one is supplied.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42, message_prefix="test.metric",
+            internal_metrics_prefix="statsd.foo.")
+
+        configurable_processor.process("gorets:17|c|@2")
+        messages = list(configurable_processor.flush())
+        self.assertEqual(("test.metric.gorets.count", 8, 42), messages[0])
+        self.assertEqual(("statsd.foo.numStats", 1, 42), messages[1])
+
+        configurable_processor.process("gorets:17|c|@1.5")
+        messages = list(configurable_processor.flush())
+        self.assertEqual(("test.metric.gorets.count", 11, 42), messages[0])
+        self.assertEqual(("statsd.foo.numStats", 1, 42), messages[1])
 
     def test_flush_plugin(self):
         """
@@ -178,6 +230,14 @@
 
 class ComposeTests(TestCase):
 
+    def test_compose_counter_metric(self):
+        configurable_processor = ConfigurableMessageProcessor()
+        configurable_processor.compose_counter_metric('a', 1, 1.0)
+        configurable_processor.compose_counter_metric('a', 1, 1.0)
+        self.assertEqual(
+            configurable_processor.counter_metrics['a'].count,
+            2)
+
     def test_compose_gauge_metric(self):
         configurable_processor = ConfigurableMessageProcessor()
         configurable_processor.compose_gauge_metric('a', 1)