Merge lp:~lucio.torre/txstatsd/add-plugins into lp:txstatsd

[Lucio Torre]

Add pluggable metric kind support to txstatsd.

Add pluggable metric kind support to txstatsd.

If you add an IMetricFactory provider in a twisted/plugins/ directory of your project and it is in the python path, it will be picked up and available for reporting. (see the probabilistic distinct count as an example of this).

[Ian Wilkinson (theiw)]

+1

Some quick comments:

* version.py needs a bump
* including a plugin_ section in txstatsd.conf-example might be useful
* rather than metric plugins, I would be tempted by processor plugins. This would involve:
- processor.py becoming an abstract processor, instantiated with a registry of metrics
- statsdprocessor.py specialising the abstract processor with its metrics registry populated by the supported StatsD metrics
- pluggableprocessor.py a specialised processor, with a metrics processor and functionality for handling processor plugins
- processorplugin.py replaces the metrics-oriented extensions introduced with the distinct_plugin.py
- twisted/plugins/codahale_plugin.py would then allow the inclusion of the Coda Hale metrics
- the pluggableprocessor.py could allow aggregation of processor plugins, for example, we would be able to support Coda Hale metrics, and logging capability
* itxstatsd.py could be txstatsd/metrics/imetric.py
* introduce txstatsd/config.py with the accumulateClassList and OptionsGlue from service.py. Have an extended OptionsGlue that includes the plugin support
* best stop there!

review: Approve

[Ian Wilkinson (theiw)]

In the earlier comment, the following:

"- pluggableprocessor.py a specialised processor, with a metrics processor and functionality for handling processor plugins"

should have read:

"- pluggableprocessor.py a specialised processor, with a metrics registry and functionality for handling processor plugins"

[Sidnei da Silva (sidnei)]

1. Please rename itxstatsd.py to interfaces.py

2. Please fix this docstring:

+ def test_generic(self):
+ """Test the timing operation."""

3. The IMetricFactory attribute is named 'metric_kind_key' but elsewhere in the code 'metric_type' is used. Maybe rename the attribute to 'metric_type'?

4. Should this variable be named 'generic_metric'?

   gauge_metric = GenericMetric

5. I couldn't figure out what GenericMetric is used for. Maybe a refactoring leftover?

6. Please keep at least a line between class and first function:

+class TestPlugin(TestCase):
+ def test_factory(self):

review: Needs Fixing

[lp:~lucio.torre/txstatsd/add-plugins]

52. By Lucio Torre on 2011-11-18

fixed per review comments

53. By Lucio Torre on 2011-11-18

unify naming

[Lucio Torre (lucio.torre)]

ian,
- done
- done
- too much refactoring there to add features and rewrite everything on one branch.
- "A convention for defining interfaces is do so in a file named like ProjectName/projectname/iprojectname.py" http://twistedmatrix.com/documents/current/core/howto/plugin.html
- left for the big refactoring.

sidnei,
1. "A convention for defining interfaces is do so in a file named like ProjectName/projectname/iprojectname.py" http://twistedmatrix.com/documents/current/core/howto/plugin.html

2. done

3. done (and changed other locations to use metric_type too as discussed on irc)

4. done

5. for sending metric that have no specific interface support in the client side and not adding dependencies on plugins there too.

6. fixed

[lp:~lucio.torre/txstatsd/add-plugins]

54. By Lucio Torre on 2011-11-18

adding tests for passing arguments on flush

[Sidnei da Silva (sidnei)]

Looks good. +1!

review: Approve

=== added file 'twisted/plugins/distinct_plugin.py'
--- twisted/plugins/distinct_plugin.py	1970-01-01 00:00:00 +0000
+++ twisted/plugins/distinct_plugin.py	2011-11-18 20:11:23 +0000
@@ -0,0 +1,20 @@
+from zope.interface import implements
+
+from twisted.plugin import IPlugin
+from txstatsd.itxstatsd import IMetricFactory
+from txstatsd.metrics.distinctmetric import DistinctMetricReporter
+
+class DistinctMetricFactory(object):
+    implements(IMetricFactory, IPlugin)
+
+    name = "pdistinct"
+    metric_type = "pd"
+
+    def build_metric(self, prefix, name, wall_time_func=None):
+        return DistinctMetricReporter(name, prefix=prefix,
+                                      wall_time_func=wall_time_func)
+
+    def configure(self, options):
+        pass
+
+distinct_metric_factory = DistinctMetricFactory()

=== modified file 'txstatsd.conf-example'
--- txstatsd.conf-example	2011-09-02 10:11:11 +0000
+++ txstatsd.conf-example	2011-11-18 20:11:23 +0000
@@ -21,3 +21,6 @@
 # configured monitor-response.
 monitor-message: txstatsd ping
 monitor-response: txstatsd pong
+
+[plugin_sample]
+sample-key: sample-value
\ No newline at end of file

=== added file 'txstatsd/itxstatsd.py'
--- txstatsd/itxstatsd.py	1970-01-01 00:00:00 +0000
+++ txstatsd/itxstatsd.py	2011-11-18 20:11:23 +0000
@@ -0,0 +1,54 @@
+from zope.interface import Interface, Attribute
+
+
+class IMetricFactory(Interface):
+    name = Attribute("""
+        @type name C{str}
+        @ivar name: The name of this kind of metric
+        """)
+
+    metric_type = Attribute("""
+        @type metric_type: C{str}
+        @ivar metric_type: The string that will be used by clients to send
+        metrics of this kind to the server.
+        """)
+
+    def build_metric(prefix, name, wall_time_func=None):
+        """
+        Returns an object that implements the C{IMetric} interface for name.
+
+        @type prefix: C{str}
+        @param prefix: The prefix used for reporting this metric.
+        @type name: C{str}
+        @param name: The name used for reporting this metric.
+        """
+
+    def configure(options):
+        """
+        Configures the factory. Will be called at startup by the service
+        factory.
+
+        @type options: C{twisted.python.usage.Options}
+        @param options: The configuration options.
+        """
+
+
+class IMetric(Interface):
+    def process(fields):
+        """
+        Process new data for this metric.
+
+        @type fields: C{list}
+        @param fields: The list of message parts. Usually in the form of
+        (value, metric_type, [sample_rate])
+        """
+
+    def flush(interval, timestamp):
+        """
+        Returns a string with new line separated list of metrics to report.
+
+        @type interval: C{float}
+        @param interval: The time since last flush.
+        @type timestamp: C{float}
+        @param timestamp: The timestamp for now.
+        """

=== modified file 'txstatsd/metrics/distinctmetric.py'
--- txstatsd/metrics/distinctmetric.py	2011-10-26 16:48:42 +0000
+++ txstatsd/metrics/distinctmetric.py	2011-11-18 20:11:23 +0000
@@ -13,21 +13,23 @@
 
 from string import Template
 
+from zope.interface import implements
 from txstatsd.metrics.metric import Metric
-
-   
+from txstatsd.itxstatsd import IMetric
+
+
 class SBoxHash(object):
     """A very fast hash.
-    
+
     This class create a random hash function that is very fast.
     Based on SBOXes. Not Crypto Strong.
-    
+
     Two instances of this class will hash differently.
     """
-    
+
     def __init__(self):
         self.table = [random.randint(0, 0xFFFFFFFF - 1) for i in range(256)]
-        
+
     def hash(self, data):
         value = 0
         for c in data:
@@ -35,14 +37,14 @@
             value = value * 3
             value = value & 0xFFFFFFFF
         return value
-    
-        
+
+
 def hash(data):
     """Hash data using a random hasher."""
     p = SBoxHash()
     return p.hash(data)
-    
-    
+
+
 def zeros(n):
     """Count the zeros to the right of the binary representation of n."""
     count = 0
@@ -60,19 +62,19 @@
 
 class SlidingDistinctCounter(object):
     """A probabilistic distinct counter with sliding windows."""
-    
+
     def __init__(self, n_hashes, n_buckets):
         self.n_hashes = n_hashes
         self.n_buckets = n_buckets
-        
+
         self.hashes = [SBoxHash() for i in range(n_hashes)]
         self.buckets = [[0] * n_buckets for i in range(n_hashes)]
-        
+
     def add(self, when, item):
         hashes = (h.hash(item) for h in self.hashes)
         for i, value in enumerate(hashes):
             self.buckets[i][min(self.n_buckets - 1, zeros(value))] = when
-    
+
     def distinct(self, since=0):
         total = 0.0
         for i in range(self.n_hashes):
@@ -84,7 +86,7 @@
             total += least0
         v = total / self.n_hashes
         return int((2 ** v) / 0.77351)
-        
+
 
 class DistinctMetric(Metric):
     """
@@ -102,13 +104,13 @@
     Keeps an estimate of the distinct numbers of items seen on various
     sliding windows of time.
     """
+    implements(IMetric)
 
     MESSAGE = (
         "$prefix%(key)s.count_1min %(count_1min)s %(timestamp)s\n"
         "$prefix%(key)s.count_1hour %(count_1hour)s %(timestamp)s\n"
         "$prefix%(key)s.count_1day %(count_1day)s %(timestamp)s\n"
-        "$prefix%(key)s.count %(count)s %(timestamp)s\n"
-    )
+        "$prefix%(key)s.count %(count)s %(timestamp)s\n")
 
     def __init__(self, name, wall_time_func=time.time, prefix=""):
         """Construct a metric we expect to be periodically updated.
@@ -131,18 +133,20 @@
 
     def count_1min(self, now):
         return self.counter.distinct(now - 60)
-    
+
     def count_1hour(self, now):
         return self.counter.distinct(now - 60 * 60)
-    
+
     def count_1day(self, now):
         return self.counter.distinct(now - 60 * 60 * 24)
 
+    def process(self, fields):
+        self.update(fields[0])
+
     def update(self, item):
-        """Adds a seen item."""
         self.counter.add(self.wall_time_func(), item)
-    
-    def report(self, timestamp):
+
+    def flush(self, interval, timestamp):
         now = self.wall_time_func()
         return self.message % {
             "key": self.name,

=== modified file 'txstatsd/metrics/metrics.py'
--- txstatsd/metrics/metrics.py	2011-10-25 20:22:10 +0000
+++ txstatsd/metrics/metrics.py	2011-11-18 20:11:23 +0000
@@ -5,6 +5,16 @@
 from txstatsd.metrics.metric import Metric
 
 
+class GenericMetric(Metric):
+    def __init__(self, connection, key, name, sample_rate=1):
+        super(GenericMetric, self).__init__(connection, name,
+                                            sample_rate=sample_rate)
+        self.key = key
+
+    def mark(self, value):
+        self.send("%s|%s" % (value, self.key))
+
+
 class Metrics(object):
     def __init__(self, connection=None, namespace=""):
         """A convenience class for reporting metric samples
@@ -20,6 +30,20 @@
         self.namespace = namespace
         self._metrics = {}
 
+    def report(self, name, value, metric_type, sample_rate=1):
+        """Report a generic metric.
+
+        Used for server side plugins without client support.
+        """
+        name = self.fully_qualify_name(name)
+        if not name in self._metrics:
+            metric = GenericMetric(self.connection,
+                                        metric_type,
+                                        name,
+                                        sample_rate)
+            self._metrics[name] = metric
+        self._metrics[name].mark(value)
+
     def gauge(self, name, value, sample_rate=1):
         """Report an instantaneous reading of a particular value."""
         name = self.fully_qualify_name(name)
@@ -76,7 +100,7 @@
             metric = DistinctMetric(self.connection, name)
             self._metrics[name] = metric
         self._metrics[name].mark(item)
-        
+
     def clear(self, name):
         """Allow the metric to re-initialize its internal state."""
         name = self.fully_qualify_name(name)

=== modified file 'txstatsd/process.py'
--- txstatsd/process.py	2011-08-08 09:28:45 +0000
+++ txstatsd/process.py	2011-11-18 20:11:23 +0000
@@ -114,7 +114,8 @@
         result = {prefix + ".cpu.percent": self.process.get_cpu_percent(),
                   prefix + ".cpu.user": utime,
                   prefix + ".cpu.system": stime,
-                  prefix + ".memory.percent": self.process.get_memory_percent(),
+                  prefix + ".memory.percent":
+                    self.process.get_memory_percent(),
                   prefix + ".memory.vsize": vsize,
                   prefix + ".memory.rss": rss}
         if getattr(self.process, "get_num_threads", None) is not None:

=== modified file 'txstatsd/server/configurableprocessor.py'
--- txstatsd/server/configurableprocessor.py	2011-10-25 20:22:10 +0000
+++ txstatsd/server/configurableprocessor.py	2011-11-18 20:11:23 +0000
@@ -24,9 +24,9 @@
 
     METRICS_SUMMARY = "statsd.numStats %s %s\n"
 
-    def __init__(self, time_function=time.time, message_prefix=""):
+    def __init__(self, time_function=time.time, message_prefix="", plugins=None):
         super(ConfigurableMessageProcessor, self).__init__(
-            time_function=time_function)
+            time_function=time_function, plugins=plugins)
 
         if message_prefix:
             self.metrics_summary = message_prefix + '.' + \
@@ -38,6 +38,9 @@
         self.message_prefix = message_prefix
         self.gauge_metrics = {}
 
+    def get_message_prefix(self, kind):
+        return self.message_prefix
+    
     def compose_timer_metric(self, key, duration):
         if not key in self.timer_metrics:
             metric = TimerMetricReporter(key, prefix=self.message_prefix)

=== modified file 'txstatsd/server/loggingprocessor.py'
--- txstatsd/server/loggingprocessor.py	2011-10-04 16:22:19 +0000
+++ txstatsd/server/loggingprocessor.py	2011-11-18 20:11:23 +0000
@@ -11,18 +11,20 @@
     attribute.)
     """
 
-    def __init__(self, logger, time_function=time.time, message_prefix=""):
+    def __init__(self, logger, time_function=time.time, message_prefix="", plugins=None):
         super(LoggingMessageProcessor, self).__init__(
-            time_function=time_function, message_prefix=message_prefix)
+            time_function=time_function, message_prefix=message_prefix,
+            plugins=plugins)
 
         logger_info = getattr(logger, 'info', None)
         if logger_info is None or not callable(logger_info):
             raise TypeError()
         self.logger = logger
 
-    def flush(self):
+    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():
@@ -34,4 +36,9 @@
         log_metrics(self.gauge_metrics)
         log_metrics(self.meter_metrics)
         log_metrics(self.timer_metrics)
+        
+        for metric in self.plugin_metrics.itervalues():
+            report = metric.flush(interval, timestamp)
+            for measurement in report.splitlines():
+                self.logger.info(measurement)
 

=== modified file 'txstatsd/server/processor.py'
--- txstatsd/server/processor.py	2011-10-26 16:48:42 +0000
+++ txstatsd/server/processor.py	2011-11-18 20:11:23 +0000
@@ -50,7 +50,7 @@
     GAUGE_METRIC_MESSAGE = (
         "stats.gauge.%(key)s.value %(value)s %(timestamp)s\n")
 
-    def __init__(self, time_function=time.time):
+    def __init__(self, time_function=time.time, plugins=None):
         self.time_function = time_function
 
         self.counters_message = MessageProcessor.COUNTERS_MESSAGE
@@ -63,6 +63,13 @@
         self.meter_metrics = {}
         self.distinct_metrics = {}
 
+        self.plugins = {}
+        self.plugin_metrics = {}
+
+        if plugins is not None:
+            for plugin in plugins:
+                self.plugins[plugin.metric_type] = plugin
+
     def fail(self, message):
         """Log and discard malformed message."""
         log.msg("Bad line: %r" % message, logLevel=logging.DEBUG)
@@ -84,30 +91,33 @@
             return self.fail(message)
 
         key = normalize_key(key)
+        metric_type = fields[1]
 
-        if fields[1] == "c":
+        if metric_type == "c":
             self.process_counter_metric(key, fields, message)
-        elif fields[1] == "ms":
+        elif metric_type == "ms":
             self.process_timer_metric(key, fields[0], message)
-        elif fields[1] == "g":
+        elif metric_type == "g":
             self.process_gauge_metric(key, fields[0], message)
-        elif fields[1] == "m":
+        elif metric_type == "m":
             self.process_meter_metric(key, fields[0], message)
-        elif fields[1] == "d":
-            self.process_distinct_metric(key, fields[0], message)            
+        elif metric_type in self.plugins:
+            self.process_plugin_metric(metric_type, key, fields, message)
         else:
             return self.fail(message)
 
-    def process_distinct_metric(self, key, item, message):
-        self.compose_distinct_metric(key, str(item))
-        
-    def compose_distinct_metric(self, key, item):
-        if not key in self.distinct_metrics:
-            metric = DistinctMetricReporter(key, self.time_function,
-                                         prefix="stats.distinct")
-            self.distinct_metrics[key] = metric
-        self.distinct_metrics[key].update(item)
-    
+    def get_message_prefix(self, kind):
+        return "stats." + kind
+
+    def process_plugin_metric(self, metric_type, key, items, message):
+        if not key in self.plugin_metrics:
+            factory = self.plugins[metric_type]
+            metric = factory.build_metric(
+                self.get_message_prefix(factory.name),
+                name=key, wall_time_func=self.time_function)
+            self.plugin_metrics[key] = metric
+        self.plugin_metrics[key].process(items)
+
     def process_timer_metric(self, key, duration, message):
         try:
             duration = float(duration)
@@ -206,11 +216,11 @@
             messages.extend(meter_metrics)
             num_stats += events
 
-        distinct_metrics, events = self.flush_distinct_metrics(timestamp)
+        plugin_metrics, events = self.flush_plugin_metrics(interval, timestamp)
         if events > 0:
-            messages.extend(distinct_metrics)
+            messages.extend(plugin_metrics)
             num_stats += events
-            
+
         self.flush_metrics_summary(messages, num_stats, timestamp)
         return messages
 
@@ -297,16 +307,17 @@
 
         return (metrics, events)
 
-    def flush_distinct_metrics(self, timestamp):
+    def flush_plugin_metrics(self, interval, timestamp):
         metrics = []
         events = 0
-        for metric in self.distinct_metrics.itervalues():
-            message = metric.report(timestamp)
+
+        for metric in self.plugin_metrics.itervalues():
+            message = metric.flush(interval, timestamp)
             metrics.append(message)
             events += 1
 
         return (metrics, events)
-        
+
     def flush_metrics_summary(self, messages, num_stats, timestamp):
         messages.append("statsd.numStats %s %s\n" % (num_stats, timestamp))
 

=== modified file 'txstatsd/service.py'
--- txstatsd/service.py	2011-10-25 14:33:14 +0000
+++ txstatsd/service.py	2011-11-18 20:11:23 +0000
@@ -7,6 +7,7 @@
 from twisted.application.internet import TCPClient, UDPServer
 from twisted.application.service import MultiService
 from twisted.python import usage
+from twisted.plugin import getPlugins
 
 from txstatsd.client import InternalClient
 from txstatsd.metrics.metrics import Metrics
@@ -15,11 +16,12 @@
 from txstatsd.server.protocol import (
     GraphiteClientFactory, StatsDServerProtocol)
 from txstatsd.report import ReportingService
+from txstatsd.itxstatsd import IMetricFactory
 
 
 def accumulateClassList(classObj, attr, listObj,
                         baseClass=None, excludeClass=None):
-    """Accumulate all attributes of a given name in a class hierarchy 
+    """Accumulate all attributes of a given name in a class hierarchy
     into a single list.
 
     Assuming all class attributes of this name are lists.
@@ -35,8 +37,7 @@
     """Extends usage.Options to also read parameters from a config file."""
 
     optParameters = [
-        ["config", "c", None, "Config file to use."]
-    ]
+        ["config", "c", None, "Config file to use."]]
 
     def __init__(self):
         parameters = []
@@ -99,6 +100,10 @@
                         value = self._dispatch[name].coerce(value)
                     self[name] = value
 
+        for section in config_file.sections():
+            if section.startswith("plugin_"):
+                self[section] = config_file.items(section)
+
 
 class StatsDOptions(OptionsGlue):
     """
@@ -122,8 +127,7 @@
         ["monitor-response", "o", "txstatsd pong",
          "Response we should send monitoring agent.", str],
         ["statsd-compliance", "s", 1,
-         "Produce StatsD-compliant messages.", int]
-    ]
+         "Produce StatsD-compliant messages.", int]]
 
     def __init__(self):
         self.config_section = 'statsd'
@@ -140,12 +144,19 @@
     if prefix is None:
         prefix = socket.gethostname() + ".statsd"
 
+    # initialize plugins
+    plugin_metrics = []
+    for plugin in getPlugins(IMetricFactory):
+        plugin.configure(options)
+        plugin_metrics.append(plugin)
+
     if options["statsd-compliance"]:
-        processor = MessageProcessor()
+        processor = MessageProcessor(plugins=plugin_metrics)
         connection = InternalClient(processor)
         metrics = Metrics(connection, namespace=prefix)
     else:
-        processor = ConfigurableMessageProcessor(message_prefix=prefix)
+        processor = ConfigurableMessageProcessor(message_prefix=prefix,
+                                                 plugins=plugin_metrics)
         connection = InternalClient(processor)
         metrics = Metrics(connection)
 

=== modified file 'txstatsd/tests/metrics/test_distinct.py'
--- txstatsd/tests/metrics/test_distinct.py	2011-10-26 03:26:21 +0000
+++ txstatsd/tests/metrics/test_distinct.py	2011-11-18 20:11:23 +0000
@@ -5,11 +5,13 @@
 from scipy.stats import chi2
 
 from twisted.trial.unittest import TestCase
-
+from twisted.plugin import getPlugins
+from twisted.plugins import distinct_plugin
 import txstatsd.metrics.distinctmetric as distinct
-
-
-class TestHash(TestCase):        
+from txstatsd.itxstatsd import IMetricFactory
+
+
+class TestHash(TestCase):
     def test_hash_chars(self):
         "For one table, all chars map to different chars"
         results = set()
@@ -18,10 +20,10 @@
             h = distinct.hash(chr(c))
             results.add(h)
         self.assertEquals(len(results), 256)
-        
+
     def test_chi_square(self):
         N = 10000
-        
+
         for (bits, buckets) in [(-1, 1024), (24, 256),
                                 (16, 256), (8, 256), (0, 256)]:
             bins = [0] * buckets
@@ -48,26 +50,27 @@
         self.assertEquals(distinct.zeros(5), 0)
         self.assertEquals(distinct.zeros(8), 3)
         self.assertEquals(distinct.zeros(9), 0)
-        
+
+
 class TestDistinct(TestCase):
     def test_all(self):
         random.seed(1)
-        
+
         for r in [1000, 10000]:
             cd = distinct.SlidingDistinctCounter(32, 32)
             for i in range(r):
                 cd.add(1, str(i))
             error = abs(cd.distinct() - r)
             self.assertTrue(error < 0.15 * r)
-        
-        
+
+
 class TestDistinctMetricReporter(TestCase):
     def test_reports(self):
         random.seed(1)
         _wall_time = [0]
         def _time():
             return _wall_time[0]
-            
+
         dmr = distinct.DistinctMetricReporter("test", wall_time_func=_time)
         for i in range(3000):
             _wall_time[0] = i * 50
@@ -77,4 +80,12 @@
         self.assertTrue(abs(dmr.count_1min(now) - 1) < 2)
         self.assertTrue(abs(dmr.count_1hour(now) - 72) < 15)
         self.assertTrue(abs(dmr.count_1day(now) - 1728) < 500)
-        self.assertTrue("count_1hour" in dmr.report(now))
+        self.assertTrue("count_1hour" in dmr.flush(1, now))
+
+
+class TestPlugin(TestCase):
+    
+    def test_factory(self):
+        self.assertTrue(distinct_plugin.distinct_metric_factory in \
+                        list(getPlugins(IMetricFactory)))
+

=== modified file 'txstatsd/tests/test_configurableprocessor.py'
--- txstatsd/tests/test_configurableprocessor.py	2011-10-10 16:08:12 +0000
+++ txstatsd/tests/test_configurableprocessor.py	2011-11-18 20:11:23 +0000
@@ -2,9 +2,10 @@
 
 from unittest import TestCase
 
+from twisted.plugins.distinct_plugin import distinct_metric_factory
+
 from txstatsd.server.configurableprocessor import ConfigurableMessageProcessor
 
-
 class FlushMessagesTest(TestCase):
 
     def test_flush_counter_with_empty_prefix(self):
@@ -36,7 +37,20 @@
         self.assertEqual("test.metric.gorets.count 17 42", counters[0])
         self.assertEqual("test.metric.statsd.numStats 1 42",
                          messages[1].splitlines()[0])
-
+        
+    def test_flush_plugin(self):
+        """
+        Ensure the prefix features if one is supplied.
+        """
+        configurable_processor = ConfigurableMessageProcessor(
+            time_function=lambda: 42, message_prefix="test.metric",
+            plugins=[distinct_metric_factory])
+        configurable_processor.process("gorets:17|pd")
+        messages = configurable_processor.flush()
+        self.assertEqual(2, len(messages))
+        self.assertTrue("test.metric.gorets" in messages[0])
+        
+        
     def test_flush_single_timer_single_time(self):
         """
         If a single timer with a single data point is present, all

=== modified file 'txstatsd/tests/test_loggingprocessor.py'
--- txstatsd/tests/test_loggingprocessor.py	2011-10-05 15:55:01 +0000
+++ txstatsd/tests/test_loggingprocessor.py	2011-11-18 20:11:23 +0000
@@ -1,6 +1,8 @@
 
 from unittest import TestCase
 
+from twisted.plugins.distinct_plugin import distinct_metric_factory
+
 from txstatsd.server.loggingprocessor import LoggingMessageProcessor
 
 
@@ -8,6 +10,13 @@
     def report(self, *args):
         return 'Sample report'
 
+class TestLogger(object):
+    def __init__(self):
+        self.log = ''
+
+    def info(self, measurement):
+        self.log += measurement + "\n"
+
 
 class TestLoggingMessageProcessor(TestCase):
 
@@ -30,16 +39,21 @@
         self.assertRaises(TypeError, invoker)
 
     def test_logger(self):
-        class Logger(object):
-            def __init__(self):
-                self.log = ''
-
-            def info(self, measurement):
-                self.log += measurement
-
-        logger = Logger()
+        logger = TestLogger()
         processor = LoggingMessageProcessor(logger)
         metric = FakeMeterMetric()
         processor.meter_metrics['test'] = metric
         processor.flush()
-        self.assertEqual(metric.report(), logger.log)
+        self.assertEqual(metric.report() + "\n", logger.log)
+
+    def test_logger_plugin(self):
+        logger = TestLogger()
+        processor = LoggingMessageProcessor(
+            logger, plugins=[distinct_metric_factory],
+            time_function=lambda: 42)
+        processor.process("gorets:17|pd")
+        processor.flush()
+        self.assertEqual(processor.plugin_metrics['gorets'].flush(
+                                10, processor.time_function()),
+                         logger.log)
+

=== modified file 'txstatsd/tests/test_metrics.py'
--- txstatsd/tests/test_metrics.py	2011-08-26 12:16:45 +0000
+++ txstatsd/tests/test_metrics.py	2011-11-18 20:11:23 +0000
@@ -53,6 +53,12 @@
         self.assertEqual(self.connection.data,
                          'txstatsd.tests.timing:101123|ms')
 
+    def test_generic(self):
+        """Test the GenericMetric class."""
+        self.metrics.report('users', "pepe", "pd")
+        self.assertEqual(self.connection.data,
+                         'txstatsd.tests.users:pepe|pd')
+
     def test_empty_namespace(self):
         """Test reporting of an empty namespace."""
         self.metrics.namespace = None

=== modified file 'txstatsd/tests/test_processor.py'
--- txstatsd/tests/test_processor.py	2011-10-26 16:48:42 +0000
+++ txstatsd/tests/test_processor.py	2011-11-18 20:11:23 +0000
@@ -1,14 +1,16 @@
 import time
 
 from unittest import TestCase
+from twisted.plugin import getPlugins
 
 from txstatsd.server.processor import MessageProcessor
-
+from txstatsd.itxstatsd import IMetricFactory
 
 class TestMessageProcessor(MessageProcessor):
 
     def __init__(self):
-        super(TestMessageProcessor, self).__init__()
+        super(TestMessageProcessor, self).__init__(
+            plugins=getPlugins(IMetricFactory))
         self.failures = []
 
     def fail(self, message):
@@ -66,12 +68,12 @@
     def test_receive_distinct_metric(self):
         """
         A distinct metric message takes the form:
-        '<name>:<item>|d'.
-        'distinct' indicates this is a distinct metric message.
+        '<name>:<item>|pd'.
+        'pd' indicates this is a probabilistic distinct metric message.
         """
-        self.processor.process("gorets:one|d")
-        self.assertEqual(1, len(self.processor.distinct_metrics))
-        self.assertTrue(self.processor.distinct_metrics["gorets"].count() > 0)
+        self.processor.process("gorets:one|pd")
+        self.assertEqual(1, len(self.processor.plugin_metrics))
+        self.assertTrue(self.processor.plugin_metrics["gorets"].count() > 0)
 
     def test_receive_message_no_fields(self):
         """
@@ -120,7 +122,8 @@
 class FlushMessagesTest(TestCase):
 
     def setUp(self):
-        self.processor = MessageProcessor(time_function=lambda: 42)
+        self.processor = MessageProcessor(time_function=lambda: 42,
+                                          plugins=getPlugins(IMetricFactory))
 
     def test_flush_no_stats(self):
         """
@@ -243,15 +246,27 @@
         a distinct metric.
         """
 
-        self.processor.process("gorets:item|d")
-        
+        self.processor.process("gorets:item|pd")
+
         messages = self.processor.flush()
         self.assertEqual(2, len(messages))
         metrics = messages[0]
-        self.assertTrue("stats.distinct.gorets.count " in metrics)
-        self.assertTrue("stats.distinct.gorets.count_1hour" in metrics)
-        self.assertTrue("stats.distinct.gorets.count_1min" in metrics)
-        self.assertTrue("stats.distinct.gorets.count_1day" in metrics)
+        self.assertTrue("stats.pdistinct.gorets.count " in metrics)
+        self.assertTrue("stats.pdistinct.gorets.count_1hour" in metrics)
+        self.assertTrue("stats.pdistinct.gorets.count_1min" in metrics)
+        self.assertTrue("stats.pdistinct.gorets.count_1day" in metrics)
+
+    def test_flush_plugin_arguments(self):
+        """Test the passing of arguments for flush."""
+
+        class FakeMetric(object):
+            def flush(self, interval, timestamp):
+                self.data = interval, timestamp
+
+        self.processor.plugin_metrics["somemetric"] = FakeMetric()
+        self.processor.flush(41000)
+        self.assertEquals((41,42),
+            self.processor.plugin_metrics["somemetric"].data)
 
 
 class FlushMeterMetricMessagesTest(TestCase):

=== modified file 'txstatsd/tests/test_service.py'
--- txstatsd/tests/test_service.py	2011-09-29 08:50:43 +0000
+++ txstatsd/tests/test_service.py	2011-11-18 20:11:23 +0000
@@ -2,6 +2,7 @@
 import ConfigParser
 import tempfile
 from unittest import TestCase
+from StringIO import StringIO
 
 from twisted.internet.defer import inlineCallbacks, Deferred
 from twisted.internet.protocol import DatagramProtocol
@@ -104,6 +105,17 @@
         f, o = self.get_file_parser([["number", "n", 5, "help", int]])
         o.parseOptions(["--config", f.name])
         self.assertEquals(5, o["number"])
+        
+    def test_support_plugin_sections(self):
+        class TestOptions(service.OptionsGlue):
+            optParameters = [["test", "t", "default", "help"]]
+            config_section = "statsd"
+            
+        o = TestOptions()
+        config_file = ConfigParser.RawConfigParser()
+        config_file.readfp(StringIO("[statsd]\n\n[plugin_test]\nfoo = bar\n"))
+        o.configure(config_file)
+        self.assertEquals(o["plugin_test"], config_file.items("plugin_test"))
 
 
 class Agent(DatagramProtocol):

=== modified file 'txstatsd/version.py'
--- txstatsd/version.py	2011-11-02 10:50:24 +0000
+++ txstatsd/version.py	2011-11-18 20:11:23 +0000
@@ -1,1 +1,1 @@
-txstatsd = "0.6.4"
+txstatsd = "0.7.0"