Merge lp:~theiw/txstatsd/txstatsd-coda-hale-counters into lp:txstatsd

[Ian Wilkinson]

Include support for reporting counter metrics, fixes an issue with the prefix for meter metrics.

Include support for reporting counter metrics.

The specialised ExtendedMetrics was introducing imagining an interface away from that of StatsD.

(This also fixes an issue with the prefix for meter metrics.)

[lp:~theiw/txstatsd/txstatsd-coda-hale-counters]

34. By Ian Wilkinson on 2011-09-09

Correctly update the counter reporter.

[Lucio Torre (lucio.torre)]

review: Approve

[Sidnei da Silva (sidnei)]

review: Approve

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

=== added file 'txstatsd/metrics/extendedmetrics.py'
--- txstatsd/metrics/extendedmetrics.py	1970-01-01 00:00:00 +0000
+++ txstatsd/metrics/extendedmetrics.py	2011-09-09 15:51:23 +0000
@@ -0,0 +1,39 @@
+
+from txstatsd.metrics.countermetric import CounterMetric
+from txstatsd.metrics.metrics import Metrics
+
+
+class ExtendedMetrics(Metrics):
+    def __init__(self, connection=None, namespace=""):
+        """A convenience class for reporting metric samples
+        to a C{txstatsd} server configured with the
+        L{ConfigurableProcessor<txstatsd.server.configurableprocessor>}
+        processor.
+
+        @param connection: The connection endpoint representing
+            the C{txstatsd} server.
+        @param namespace: The top-level namespace identifying the
+            origin of the samples.
+        """
+
+        super(ExtendedMetrics, self).__init__(connection, namespace)
+
+    def increment(self, name, value=1, sample_rate=1):
+        """Report and increase in name by count."""
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = CounterMetric(self.connection,
+                                   name,
+                                   sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].increment(value)
+
+    def decrement(self, name, value=1, sample_rate=1):
+        """Report and decrease in name by count."""
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = CounterMetric(self.connection,
+                                   name,
+                                   sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].decrement(value)

=== modified file 'txstatsd/metrics/metermetric.py'
--- txstatsd/metrics/metermetric.py	2011-09-02 10:11:11 +0000
+++ txstatsd/metrics/metermetric.py	2011-09-09 15:51:23 +0000
@@ -60,8 +60,10 @@
         self.name = name
         self.wall_time_func = wall_time_func
 
+        if prefix:
+            prefix += '.'
         self.message = Template(MeterMetricReporter.MESSAGE).substitute(
-            prefix=prefix + '.')
+            prefix=prefix)
 
         self.m1_rate = Ewma.one_minute_ewma()
         self.m5_rate = Ewma.five_minute_ewma()

=== modified file 'txstatsd/server/configurableprocessor.py'
--- txstatsd/server/configurableprocessor.py	2011-09-02 10:11:11 +0000
+++ txstatsd/server/configurableprocessor.py	2011-09-09 15:51:23 +0000
@@ -3,6 +3,7 @@
 
 import time
 
+from txstatsd.metrics.countermetric import CounterMetricReporter
 from txstatsd.metrics.metermetric import MeterMetricReporter
 from txstatsd.server.processor import MessageProcessor
 
@@ -14,6 +15,7 @@
     Currently, this extends to:
     - Allow a prefix to be added to the composed messages sent
       to the Graphite server.
+    - Report an incrementing and decrementing counter metric.
     """
 
     # Notice: These messages replicate those seen in the
@@ -21,10 +23,6 @@
     # In a future release they will be placed in their
     # respective metric reporter class.
     # See MeterMetricReporter.
-    COUNTERS_MESSAGE = (
-        "$prefix%(key)s %(value)s %(timestamp)s\n"
-        "$prefix%(key)s %(count)s %(timestamp)s\n")
-
     TIMERS_MESSAGE = (
         "$prefix%(key)s.mean %(mean)s %(timestamp)s\n"
         "$prefix%(key)s.upper %(upper)s %(timestamp)s\n"
@@ -43,10 +41,6 @@
         if message_prefix:
             message_prefix += '.'
 
-        message = Template(ConfigurableMessageProcessor.COUNTERS_MESSAGE)
-        self.counters_message = message.substitute(
-            prefix=message_prefix)
-
         message = Template(ConfigurableMessageProcessor.TIMERS_MESSAGE)
         self.timers_message = message.substitute(
             prefix=message_prefix)
@@ -55,9 +49,33 @@
         self.gauge_metric_message = message.substitute(
             prefix=message_prefix)
 
+    def process_counter_metric(self, key, composite, message):
+        try:
+            value = float(composite[0])
+        except (TypeError, ValueError):
+            return self.fail(message)
+
+        self.compose_counter_metric(key, value)
+
+    def compose_counter_metric(self, key, value):
+        if not key in self.counter_metrics:
+            metric = CounterMetricReporter(key, prefix=self.message_prefix)
+            self.counter_metrics[key] = metric
+        self.counter_metrics[key].mark(value)
+
     def compose_meter_metric(self, key, value):
         if not key in self.meter_metrics:
             metric = MeterMetricReporter(key, self.time_function,
                                          prefix=self.message_prefix)
             self.meter_metrics[key] = metric
         self.meter_metrics[key].mark(value)
+
+    def flush_counter_metrics(self, interval, timestamp):
+        metrics = []
+        events = 0
+        for metric in self.counter_metrics.itervalues():
+            message = metric.report(timestamp)
+            metrics.append(message)
+            events += 1
+
+        return (metrics, events)

=== modified file 'txstatsd/server/processor.py'
--- txstatsd/server/processor.py	2011-09-02 10:11:11 +0000
+++ txstatsd/server/processor.py	2011-09-09 15:51:23 +0000
@@ -114,6 +114,10 @@
             if match is None:
                 return self.fail(message)
             rate = match.group(1)
+
+        self.compose_counter_metric(key, value, rate)
+
+    def compose_counter_metric(self, key, value, rate):
         if key not in self.counter_metrics:
             self.counter_metrics[key] = 0
         self.counter_metrics[key] += value * (1 / float(rate))

=== modified file 'txstatsd/tests/test_processor.py'
--- txstatsd/tests/test_processor.py	2011-09-02 10:11:11 +0000
+++ txstatsd/tests/test_processor.py	2011-09-09 15:51:23 +0000
@@ -148,18 +148,18 @@
     def test_flush_counter_with_empty_prefix(self):
         """
         Ensure no prefix features if none is supplied.
+        B{Note}: The C{ConfigurableMessageProcessor} reports
+        the counter value, and not the normalized version as
+        seen in the StatsD-compliant C{Processor}.
         """
         configurable_processor = ConfigurableMessageProcessor(
             time_function=lambda: 42)
-        configurable_processor.counter_metrics["gorets"] = 42
+        configurable_processor.process("gorets:17|c")
         messages = configurable_processor.flush()
         self.assertEqual(2, len(messages))
         counters = messages[0].splitlines()
-        self.assertEqual("gorets 4 42", counters[0])
-        self.assertEqual("gorets 42 42", counters[1])
+        self.assertEqual("gorets.count 17 42", counters[0])
         self.assertEqual("statsd.numStats 1 42", messages[1])
-        self.assertEqual(
-            0, configurable_processor.counter_metrics["gorets"])
 
     def test_flush_counter_with_prefix(self):
         """
@@ -167,15 +167,12 @@
         """
         configurable_processor = ConfigurableMessageProcessor(
             time_function=lambda: 42, message_prefix="test.metric")
-        configurable_processor.counter_metrics["gorets"] = 42
+        configurable_processor.process("gorets:17|c")
         messages = configurable_processor.flush()
         self.assertEqual(2, len(messages))
         counters = messages[0].splitlines()
-        self.assertEqual("test.metric.gorets 4 42", counters[0])
-        self.assertEqual("test.metric.gorets 42 42", counters[1])
+        self.assertEqual("test.metric.gorets.count 17 42", counters[0])
         self.assertEqual("statsd.numStats 1 42", messages[1])
-        self.assertEqual(
-            0, configurable_processor.counter_metrics["gorets"])
 
     def test_flush_counter_one_second_interval(self):
         """

=== modified file 'txstatsd/version.py'
--- txstatsd/version.py	2011-09-07 10:18:11 +0000
+++ txstatsd/version.py	2011-09-09 15:51:23 +0000
@@ -1,1 +1,1 @@
-txstatsd = "0.3.0"
+txstatsd = "0.4.0"