Merge lp:~sidnei/txstatsd/cooperate into lp:txstatsd

[Sidnei da Silva]

Use cooperator to yield more often and avoid blocking the reactor for too long.

Use cooperator to yield more often and avoid blocking the reactor for too long.

[lp:~sidnei/txstatsd/cooperate]

113. By Sidnei da Silva on 2013-06-10

- Add reactor inspector tests

[Guillermo Gonzalez (verterok)]

nice!

review: Approve

=== modified file 'txstatsd/metrics/distinctmetric.py'
--- txstatsd/metrics/distinctmetric.py	2013-03-08 13:27:16 +0000
+++ txstatsd/metrics/distinctmetric.py	2013-06-10 14:01:47 +0000
@@ -165,7 +165,7 @@
                  ".count_1min": self.count_1min(now),
                  ".count_1hour": self.count_1hour(now),
                  ".count_1day": self.count_1day(now)}
-        for item, value in items.iteritems():
+        for item, value in sorted(items.iteritems()):
             metrics.append((self.prefix + self.name + item, value, timestamp))
         return metrics
 

=== modified file 'txstatsd/metrics/timermetric.py'
--- txstatsd/metrics/timermetric.py	2012-11-20 11:56:07 +0000
+++ txstatsd/metrics/timermetric.py	2013-06-10 14:01:47 +0000
@@ -118,7 +118,7 @@
         @param percentiles: One or more percentiles.
         """
         return [percentile for percentile in
-            self.histogram.percentiles(*percentiles)]
+                self.histogram.percentiles(*percentiles)]
 
     def get_values(self):
         """Returns a list of all recorded durations in the timer's sample."""
@@ -145,8 +145,8 @@
                  ".999percentile": percentiles[5],
                  ".count": self.count,
                  ".rate": self.rate(timestamp),
-            }
-        for item, value in items.iteritems():
+                 }
+        for item, value in sorted(items.iteritems()):
             metrics.append((self.prefix + self.name + item,
                             round(value, 6), timestamp))
         self.clear(timestamp)

=== modified file 'txstatsd/protocol.py'
--- txstatsd/protocol.py	2012-12-27 20:49:06 +0000
+++ txstatsd/protocol.py	2013-06-10 14:01:47 +0000
@@ -21,7 +21,7 @@
 
 import socket
 
-from twisted.internet.defer import inlineCallbacks, returnValue, Deferred
+from twisted.internet import abstract
 from twisted.internet.protocol import DatagramProtocol
 from twisted.python import log
 
@@ -139,6 +139,9 @@
         self.transport = None
         self.transport_gateway = None
 
+        if abstract.isIPAddress(host):
+            self.host_resolved(host)
+
     def __str__(self):
         return "%s:%d" % (self.host, self.port)
 

=== modified file 'txstatsd/report.py'
--- txstatsd/report.py	2012-06-28 17:29:26 +0000
+++ txstatsd/report.py	2013-06-10 14:01:47 +0000
@@ -19,6 +19,14 @@
 # TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
 # SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
+import os
+import sys
+import time
+import logging
+import threading
+import traceback
+import Queue
+
 from twisted.internet.defer import maybeDeferred
 from twisted.internet.task import LoopingCall
 from twisted.python import log
@@ -82,3 +90,113 @@
         for task, interval in self.tasks:
             task.stop()
         Service.stopService(self)
+
+
+class ReactorInspector(threading.Thread):
+    """Log message with a time delta from the last call."""
+
+    def __init__(self, reactor_call, metrics, loop_time=3, log=log.msg):
+        self.running = False
+        self.stopped = False
+        self.queue = Queue.Queue()
+        self.reactor_call = reactor_call
+        self.loop_time = loop_time
+        self.last_responsive_ts = 0
+        self.reactor_thread = None
+        self.metrics = metrics
+        super(ReactorInspector, self).__init__()
+        self.daemon = True
+        self.log = log
+
+    def start(self):
+        """Start the thread. Should be called from the reactor main thread."""
+        self.reactor_thread = threading.currentThread().ident
+        if not self.running:
+            self.running = True
+            super(ReactorInspector, self).start()
+
+    def stop(self):
+        """Stop the thread."""
+        self.stopped = True
+        self.log("ReactorInspector: stopped")
+
+    def dump_frames(self):
+        """Dump frames info to log file."""
+        current = threading.currentThread().ident
+        frames = sys._current_frames()
+        for frame_id, frame in frames.iteritems():
+            if frame_id == current:
+                continue
+
+            stack = ''.join(traceback.format_stack(frame))
+
+            if frame_id == self.reactor_thread:
+                title = "Dumping Python frame for reactor main thread"
+            else:
+                title = "Dumping Python frame"
+            self.log("%s %s (pid: %d):\n%s" %
+                     (title, frame_id, os.getpid(), stack),
+                     logLevel=logging.DEBUG)
+
+    def run(self):
+        """Start running the thread."""
+        self.log("ReactorInspector: started")
+        msg_id = 0
+        oldest_pending_request_ts = time.time()
+        while not self.stopped:
+            def task(msg_id=msg_id, tini=time.time()):
+                """Put result in queue with initial and completed times."""
+                self.queue.put((msg_id, tini, time.time()))
+            self.reactor_call(task)
+            time.sleep(self.loop_time)
+            try:
+                id_sent, tini, tsent = self.queue.get_nowait()
+            except Queue.Empty:
+                # Oldest pending request is still out there
+                delay = time.time() - oldest_pending_request_ts
+                self.metrics.gauge("delay", delay)
+                self.log("ReactorInspector: detected unresponsive!"
+                         " (current: %d, pid: %d) delay: %.3f" % (
+                             msg_id, os.getpid(), delay),
+                         logLevel=logging.CRITICAL)
+                self.dump_frames()
+            else:
+                delay = tsent - tini
+                self.metrics.gauge("delay", delay)
+                if msg_id > id_sent:
+                    self.log("ReactorInspector: late (current: %d, "
+                             "got: %d, pid: %d, cleaning queue) "
+                             "delay: %.3f" % (msg_id, id_sent,
+                                              os.getpid(), delay),
+                             logLevel=logging.WARNING)
+                    while not self.queue.empty():
+                        self.queue.get_nowait()
+                    # About to start a new request with nothing pending
+                    oldest_pending_request_ts = time.time()
+                else:
+                    assert msg_id == id_sent
+                    # About to start a new request with nothing pending
+                    self.last_responsive_ts = time.time()
+                    oldest_pending_request_ts = self.last_responsive_ts
+                    self.log("ReactorInspector: ok (msg: %d, "
+                             "pid: %d) delay: %.3f" % (
+                                 msg_id, os.getpid(), delay),
+                             logLevel=logging.DEBUG)
+            finally:
+                msg_id += 1
+
+
+class ReactorInspectorService(Service):
+    """Start/stop the reactor inspector service."""
+
+    def __init__(self, reactor, metrics, loop_time=3):
+        self.inspector = ReactorInspector(
+            reactor.callFromThread, metrics, loop_time)
+
+    def startService(self):
+        Service.startService(self)
+        self.inspector.start()
+
+    def stopService(self):
+        self.inspector.stop()
+        Service.stopService(self)

=== modified file 'txstatsd/server/configurableprocessor.py'
--- txstatsd/server/configurableprocessor.py	2012-06-28 17:29:26 +0000
+++ txstatsd/server/configurableprocessor.py	2013-06-10 14:01:47 +0000
@@ -60,8 +60,9 @@
 
     def compose_timer_metric(self, key, duration):
         if not key in self.timer_metrics:
-            metric = TimerMetricReporter(key,
-                wall_time_func=self.time_function, prefix=self.message_prefix)
+            metric = TimerMetricReporter(
+                key, wall_time_func=self.time_function,
+                prefix=self.message_prefix)
             self.timer_metrics[key] = metric
         self.timer_metrics[key].update(duration)
 
@@ -93,31 +94,16 @@
         self.meter_metrics[key].mark(value)
 
     def flush_counter_metrics(self, interval, timestamp):
-        metrics = []
-        events = 0
         for metric in self.counter_metrics.itervalues():
             messages = metric.report(timestamp)
-            metrics.extend(messages)
-            events += 1
-
-        return (metrics, events)
+            yield messages
 
     def flush_gauge_metrics(self, timestamp):
-        metrics = []
-        events = 0
         for metric in self.gauge_metrics.itervalues():
             messages = metric.report(timestamp)
-            metrics.extend(messages)
-            events += 1
-
-        return (metrics, events)
+            yield messages
 
     def flush_timer_metrics(self, percent, timestamp):
-        metrics = []
-        events = 0
         for metric in self.timer_metrics.itervalues():
             messages = metric.report(timestamp)
-            metrics.extend(messages)
-            events += 1
-
-        return (metrics, events)
+            yield messages

=== modified file 'txstatsd/server/loggingprocessor.py'
--- txstatsd/server/loggingprocessor.py	2012-06-28 17:29:26 +0000
+++ txstatsd/server/loggingprocessor.py	2013-06-10 14:01:47 +0000
@@ -48,8 +48,7 @@
 
     def flush(self, interval=10000, percent=90):
         """Log all received metric samples to the supplied logger."""
-        messages = list(super(LoggingMessageProcessor, self).flush(
-            interval=interval, percent=percent))
-        for msg in messages:
+        parent = super(LoggingMessageProcessor, self)
+        for msg in parent.flush(interval=interval, percent=percent):
             self.logger.info("Out: %s %s %s" % msg)
-        return messages
+            yield msg

=== modified file 'txstatsd/server/processor.py'
--- txstatsd/server/processor.py	2012-06-28 17:29:26 +0000
+++ txstatsd/server/processor.py	2013-06-10 14:01:47 +0000
@@ -25,6 +25,7 @@
 import logging
 
 from twisted.python import log
+from twisted.internet.task import Cooperator
 
 from txstatsd.metrics.metermetric import MeterMetricReporter
 
@@ -226,73 +227,75 @@
         Flush all queued stats, computing a normalized count based on
         C{interval} and mean timings based on C{threshold}.
         """
-        messages = []
         per_metric = {}
         num_stats = 0
         interval = interval / 1000
         timestamp = int(self.time_function())
 
         start = self.time_function()
-        counter_metrics, events = self.flush_counter_metrics(interval,
-                                                             timestamp)
+        events = 0
+        for metrics in self.flush_counter_metrics(interval, timestamp):
+            for metric in metrics:
+                yield metric
+            events += 1
         duration = self.time_function() - start
-        if events > 0:
-            messages.extend(sorted(counter_metrics))
-            num_stats += events
+        num_stats += events
         per_metric["counter"] = (events, duration)
 
         start = self.time_function()
-        timer_metrics, events = self.flush_timer_metrics(percent, timestamp)
+        events = 0
+        for metrics in self.flush_timer_metrics(percent, timestamp):
+            for metric in metrics:
+                yield metric
+            events += 1
         duration = self.time_function() - start
-        if events > 0:
-            messages.extend(sorted(timer_metrics))
-            num_stats += events
+        num_stats += events
         per_metric["timer"] = (events, duration)
 
         start = self.time_function()
-        gauge_metrics, events = self.flush_gauge_metrics(timestamp)
+        events = 0
+        for metrics in self.flush_gauge_metrics(timestamp):
+            for metric in metrics:
+                yield metric
+            events += 1
         duration = self.time_function() - start
-        if events > 0:
-            messages.extend(sorted(gauge_metrics))
-            num_stats += events
+        num_stats += events
         per_metric["gauge"] = (events, duration)
 
         start = self.time_function()
-        meter_metrics, events = self.flush_meter_metrics(timestamp)
+        events = 0
+        for metrics in self.flush_meter_metrics(timestamp):
+            for metric in metrics:
+                yield metric
+            events += 1
         duration = self.time_function() - start
-        if events > 0:
-            messages.extend(sorted(meter_metrics))
-            num_stats += events
+        num_stats += events
         per_metric["meter"] = (events, duration)
 
         start = self.time_function()
-        plugin_metrics, events = self.flush_plugin_metrics(interval, timestamp)
+        events = 0
+        for metrics in self.flush_plugin_metrics(interval, timestamp):
+            for metric in metrics:
+                yield metric
+            events += 1
         duration = self.time_function() - start
-        if events > 0:
-            messages.extend(sorted(plugin_metrics))
-            num_stats += events
+        num_stats += events
         per_metric["plugin"] = (events, duration)
 
-        self.flush_metrics_summary(messages, num_stats, per_metric, timestamp)
-        return messages
+        for metrics in self.flush_metrics_summary(num_stats, per_metric,
+                                                  timestamp):
+            for metric in metrics:
+                yield metric
 
     def flush_counter_metrics(self, interval, timestamp):
-        metrics = []
-        events = 0
         for key, count in self.counter_metrics.iteritems():
             self.counter_metrics[key] = 0
 
             value = count / interval
-            metrics.append((self.stats_prefix + key, value, timestamp))
-            metrics.append((self.count_prefix + key, count, timestamp))
-            events += 1
-
-        return (metrics, events)
+            yield ((self.stats_prefix + key, value, timestamp),
+                   (self.count_prefix + key, count, timestamp))
 
     def flush_timer_metrics(self, percent, timestamp):
-        metrics = []
-        events = 0
-
         threshold_value = ((100 - percent) / 100.0)
         for key, timers in self.timer_metrics.iteritems():
             count = len(timers)
@@ -318,78 +321,52 @@
                          ".upper_%s" % percent: threshold_upper,
                          ".lower": lower,
                          ".count": count}
-                for item, value in items.iteritems():
-                    metrics.append((self.timer_prefix + key + item,
-                                    value, timestamp))
-                events += 1
-
-        return (metrics, events)
+                yield sorted((self.timer_prefix + key + item, value, timestamp)
+                             for item, value in items.iteritems())
 
     def flush_gauge_metrics(self, timestamp):
-        metrics = []
-        events = 0
         for metric in self.gauge_metrics:
             value = metric[0]
             key = metric[1]
 
-            metrics.append((self.gauge_prefix + key + ".value",
-                            value, timestamp))
-            events += 1
+            yield ((self.gauge_prefix + key + ".value", value, timestamp),)
 
         self.gauge_metrics.clear()
 
-        return (metrics, events)
-
     def flush_meter_metrics(self, timestamp):
-        metrics = []
-        events = 0
         for metric in self.meter_metrics.itervalues():
             messages = metric.report(timestamp)
-            metrics.extend(messages)
-            events += 1
-
-        return (metrics, events)
+            yield messages
 
     def flush_plugin_metrics(self, interval, timestamp):
-        metrics = []
-        events = 0
-
         for metric in self.plugin_metrics.itervalues():
             messages = metric.flush(interval, timestamp)
-            metrics.extend(messages)
-            events += 1
-
-        return (metrics, events)
-
-    def flush_metrics_summary(self, messages, num_stats,
-                              per_metric, timestamp):
-
-        messages.append((self.internal_metrics_prefix + "numStats",
-                         num_stats, timestamp))
+            yield messages
+
+    def flush_metrics_summary(self, num_stats, per_metric, timestamp):
+        yield ((self.internal_metrics_prefix + "numStats",
+                num_stats, timestamp),)
 
         self.last_flush_duration = 0
         for name, (value, duration) in per_metric.iteritems():
-            messages.extend([
-                (self.internal_metrics_prefix +
-                 "flush.%s.count" % name,
-                 value, timestamp),
-                (self.internal_metrics_prefix +
-                 "flush.%s.duration" % name,
-                 duration * 1000, timestamp)])
+            yield ((self.internal_metrics_prefix +
+                    "flush.%s.count" % name,
+                    value, timestamp),
+                   (self.internal_metrics_prefix +
+                    "flush.%s.duration" % name,
+                    duration * 1000, timestamp))
             log.msg("Flushed %d %s metrics in %.6f" %
                     (value, name, duration))
             self.last_flush_duration += duration
 
         self.last_process_duration = 0
         for metric_type, duration in self.process_timings.iteritems():
-            messages.extend([
-                (self.internal_metrics_prefix +
-                 "receive.%s.count" %
-                 metric_type, self.by_type[metric_type], timestamp),
-                (self.internal_metrics_prefix +
-                 "receive.%s.duration" %
-                 metric_type, duration * 1000, timestamp)
-                ])
+            yield ((self.internal_metrics_prefix +
+                    "receive.%s.count" %
+                    metric_type, self.by_type[metric_type], timestamp),
+                   (self.internal_metrics_prefix +
+                    "receive.%s.duration" %
+                    metric_type, duration * 1000, timestamp))
             log.msg("Processing %d %s metrics took %.6f" %
                     (self.by_type[metric_type], metric_type, duration))
             self.last_process_duration += duration

=== modified file 'txstatsd/server/protocol.py'
--- txstatsd/server/protocol.py	2012-06-28 17:29:26 +0000
+++ txstatsd/server/protocol.py	2013-06-10 14:01:47 +0000
@@ -31,8 +31,8 @@
     server via TCP.
     """
 
-    def __init__(self, processor,
-                 monitor_message=None, monitor_response=None):
+    def __init__(self, processor, monitor_message=None,
+                 monitor_response=None):
         self.processor = processor
         self.monitor_message = monitor_message
         self.monitor_response = monitor_response
@@ -42,9 +42,10 @@
         if data == self.monitor_message:
             # Send the expected response to the
             # monitoring agent.
-            self.transport.write(self.monitor_response, (host, port))
-        else:
-            self.processor.process(data)
+            return self.transport.write(
+                self.monitor_response, (host, port))
+        return self.transport.reactor.callLater(
+            0, self.processor.process, data)
 
 
 class StatsDTCPServerProtocol(LineReceiver):
@@ -54,8 +55,8 @@
     server via TCP.
     """
 
-    def __init__(self, processor,
-                 monitor_message=None, monitor_response=None):
+    def __init__(self, processor, monitor_message=None,
+                 monitor_response=None):
         self.processor = processor
         self.monitor_message = monitor_message
         self.monitor_response = monitor_response
@@ -65,20 +66,20 @@
         if data == self.monitor_message:
             # Send the expected response to the
             # monitoring agent.
-            self.transport.write(self.monitor_response)
-        else:
-            self.processor.process(data)
+            return self.transport.write(self.monitor_response)
+        return self.transport.reactor.callLater(
+            0, self.processor.process, data)
 
 
 class StatsDTCPServerFactory(Factory):
 
-    def __init__(self, processor,
-                 monitor_message=None, monitor_response=None):
+    def __init__(self, processor, monitor_message=None,
+                 monitor_response=None):
         self.processor = processor
         self.monitor_message = monitor_message
         self.monitor_response = monitor_response
 
     def buildProtocol(self, addr):
-        return StatsDTCPServerProtocol(self.processor,
-            self.monitor_message, self.monitor_response)
-
+        return StatsDTCPServerProtocol(
+            self.processor, self.monitor_message,
+            self.monitor_response)

=== modified file 'txstatsd/service.py'
--- txstatsd/service.py	2012-06-28 17:29:26 +0000
+++ txstatsd/service.py	2013-06-10 14:01:47 +0000
@@ -41,7 +41,7 @@
     StatsDServerProtocol, StatsDTCPServerFactory)
 from txstatsd.server.router import Router
 from txstatsd.server import httpinfo
-from txstatsd.report import ReportingService
+from txstatsd.report import ReportingService, ReactorInspectorService
 from txstatsd.itxstatsd import IMetricFactory
 from twisted.application.service import Service
 from twisted.internet import task
@@ -211,19 +211,26 @@
         self.processor = processor
         self.flush_interval = flush_interval
         self.flush_task = task.LoopingCall(self.flushProcessor)
+        self.coop = task.Cooperator()
         if clock is not None:
             self.flush_task.clock = clock
 
     def flushProcessor(self):
         """Flush messages queued in the processor to Graphite."""
-        flushed = 0
         start = time.time()
-        for metric, value, timestamp in self.processor.flush(
-                interval=self.flush_interval):
-            self.carbon_client.sendDatapoint(metric, (timestamp, value))
-            flushed += 1
-        log.msg("Flushed total %d metrics in %.6f" %
-                (flushed, time.time() - start))
+        interval = self.flush_interval
+        flush = self.processor.flush
+
+        def doWork():
+            flushed = 0
+            for metric, value, timestamp in flush(interval=interval):
+                yield self.carbon_client.sendDatapoint(
+                    metric, (timestamp, value))
+                flushed += 1
+            log.msg("Flushed total %d metrics in %.6f" %
+                    (flushed, time.time() - start))
+
+        self.coop.coiterate(doWork())
 
     def startService(self):
         self.flush_task.start(self.flush_interval / 1000, False)
@@ -276,7 +283,7 @@
         # LoggingMessageProcessor supersedes
         #  any other processor class in "dump-mode"
         assert not hasattr(log, 'info')
-        log.info = log.msg # for compatibility with LMP logger interface
+        log.info = log.msg  # for compatibility with LMP logger interface
         processor = functools.partial(LoggingMessageProcessor, logger=log)
 
     if options["statsd-compliance"]:
@@ -315,6 +322,11 @@
             process.report_reactor_stats(reactor), 60, metrics.gauge)
         reports = [name.strip() for name in options["report"].split(",")]
         for report_name in reports:
+            if report_name == "reactor":
+                inspector = ReactorInspectorService(reactor, metrics,
+                                                    loop_time=0.05)
+                inspector.setServiceParent(root_service)
+
             for reporter in getattr(process, "%s_STATS" %
                                     report_name.upper(), ()):
                 reporting.schedule(reporter, 60, metrics.gauge)
@@ -343,12 +355,12 @@
 
     if options["listen-tcp-port"] is not None:
         statsd_tcp_server_factory = StatsDTCPServerFactory(
-                input_router,
-                monitor_message=options["monitor-message"],
-                monitor_response=options["monitor-response"])
+            input_router,
+            monitor_message=options["monitor-message"],
+            monitor_response=options["monitor-response"])
 
         listener = TCPServer(options["listen-tcp-port"],
-            statsd_tcp_server_factory)
+                             statsd_tcp_server_factory)
         listener.setServiceParent(root_service)
 
     httpinfo_service = httpinfo.makeService(options, processor, statsd_service)

=== modified file 'txstatsd/tests/test_client.py'
--- txstatsd/tests/test_client.py	2013-03-07 16:33:45 +0000
+++ txstatsd/tests/test_client.py	2013-06-10 14:01:47 +0000
@@ -198,11 +198,18 @@
 
         self.assertIsInstance(self.client.data_queue, DataQueue)
 
-    def test_starts_without_transport_gateway(self):
+    def test_starts_with_transport_gateway_if_ip(self):
         """The client starts without a TransportGateway."""
         self.client = TwistedStatsDClient('127.0.0.1', 8000)
         self.build_protocol()
 
+        self.assertTrue(self.client.transport_gateway is not None)
+
+    def test_starts_without_transport_gateway_if_not_ip(self):
+        """The client starts without a TransportGateway."""
+        self.client = TwistedStatsDClient('localhost', 8000)
+        self.build_protocol()
+
         self.assertTrue(self.client.transport_gateway is None)
 
     def test_passes_transport_to_gateway(self):
@@ -269,7 +276,8 @@
         callback = Mock()
         self.client.transport_gateway.write.return_value = bytes_sent
         self.assertEqual(self.client.write(message, callback), bytes_sent)
-        self.client.transport_gateway.write.assert_called_once_with(message, callback)
+        self.client.transport_gateway.write.assert_called_once_with(
+            message, callback)
 
     def test_sends_messages_to_queue_before_host_resolves(self):
         """Before the host is resolved, send messages to the DataQueue."""
@@ -277,7 +285,7 @@
         self.build_protocol()
 
         message = 'some data'
-        self.client.data_queue =  Mock(spec=DataQueue)
+        self.client.data_queue = Mock(spec=DataQueue)
         callback = Mock()
         self.client.data_queue.write.return_value = None
         result = self.client.write(message, callback)

=== modified file 'txstatsd/tests/test_configurableprocessor.py'
--- txstatsd/tests/test_configurableprocessor.py	2012-06-28 17:29:26 +0000
+++ txstatsd/tests/test_configurableprocessor.py	2013-06-10 14:01:47 +0000
@@ -40,7 +40,7 @@
         configurable_processor = ConfigurableMessageProcessor(
             time_function=lambda: 42)
         configurable_processor.process("gorets:17|c")
-        messages = configurable_processor.flush()
+        messages = list(configurable_processor.flush())
         self.assertEqual(("gorets.count", 17, 42), messages[0])
         self.assertEqual(("statsd.numStats", 1, 42), messages[1])
 
@@ -51,7 +51,7 @@
         configurable_processor = ConfigurableMessageProcessor(
             time_function=lambda: 42, message_prefix="test.metric")
         configurable_processor.process("gorets:17|c")
-        messages = configurable_processor.flush()
+        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])
@@ -64,7 +64,7 @@
             time_function=lambda: 42, message_prefix="test.metric",
             internal_metrics_prefix="statsd.foo.")
         configurable_processor.process("gorets:17|c")
-        messages = configurable_processor.flush()
+        messages = list(configurable_processor.flush())
         self.assertEqual(("test.metric.gorets.count", 17, 42), messages[0])
         self.assertEqual(("statsd.foo.numStats", 1, 42),
                          messages[1])
@@ -77,7 +77,7 @@
             time_function=lambda: 42, message_prefix="test.metric",
             plugins=[distinct_metric_factory])
         configurable_processor.process("gorets:17|pd")
-        messages = configurable_processor.flush()
+        messages = list(configurable_processor.flush())
         self.assertEquals(("test.metric.gorets.count", 1, 42), messages[0])
 
     def test_flush_single_timer_single_time(self):
@@ -94,7 +94,7 @@
         configurable_processor.process("glork:24|ms")
         _now = 42
 
-        messages = configurable_processor.flush()
+        messages = list(configurable_processor.flush())
         messages.sort()
 
         expected = [
@@ -128,7 +128,7 @@
         configurable_processor.process("glork:42|ms")
 
         _now = 42
-        messages = configurable_processor.flush()
+        messages = list(configurable_processor.flush())
         messages.sort()
 
         expected = [
@@ -169,7 +169,7 @@
         self.configurable_processor.process("gorets:3.0|m")
 
         self.time_now += 1
-        messages = self.configurable_processor.flush()
+        messages = list(self.configurable_processor.flush())
         self.assertEqual(("test.metric.gorets.count", 3.0, self.time_now),
                          messages[0])
         self.assertEqual(("test.metric.gorets.rate", 3.0, self.time_now),

=== added file 'txstatsd/tests/test_inspector.py'
--- txstatsd/tests/test_inspector.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/test_inspector.py	2013-06-10 14:01:47 +0000
@@ -0,0 +1,204 @@
+# Copyright (C) 2011-2012 Canonical Services Ltd
+#
+# Permission is hereby granted, free of charge, to any person obtaining
+# a copy of this software and associated documentation files (the
+# "Software"), to deal in the Software without restriction, including
+# without limitation the rights to use, copy, modify, merge, publish,
+# distribute, sublicense, and/or sell copies of the Software, and to
+# permit persons to whom the Software is furnished to do so, subject to
+# the following conditions:
+#
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+# CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+"""Tests for the ReactorInspector."""
+
+import re
+import time
+import logging
+import threading
+
+from twisted.trial.unittest import TestCase as TwistedTestCase
+from twisted.internet import reactor, defer
+
+from txstatsd.report import ReactorInspector
+
+
+def parse_delay(msg):
+    return float(re.search("delay: (\d+.\d{1,3})", msg).group(1))
+
+
+class ReactorInspectorTestCase(TwistedTestCase):
+    """Test the ReactorInspector class."""
+
+    def setUp(self):
+        """Set up."""
+
+        class Helper(object):
+            """Fake object with a controllable call."""
+            def __init__(self):
+                self.call_count = 1
+                self.calls = []
+                self.ri = None
+
+            def call(self, func):
+                """Call function when counter is 0, then stop running."""
+                self.call_count -= 1
+                self.calls.append(func)
+                if self.call_count == 0:
+                    for f in self.calls:
+                        f()
+                if self.call_count <= 0:
+                    self.ri.stop()
+
+        class FakeMetrics(object):
+            """Fake Metrics object that records calls."""
+            def __init__(self):
+                """Initialize calls."""
+                self.calls = []
+
+            def meter(self, name, count):
+                """Record call to meter()."""
+                self.calls.append(("meter", name, count))
+
+            def gauge(self, name, val):
+                """Record call to gauge()."""
+                self.calls.append(("gauge", name, round(val, 3)))
+
+        def log(msg, logLevel=None):
+            self.logged.append((msg, logLevel))
+
+        self.helper = Helper()
+        self.fake_metrics = FakeMetrics()
+        self.ri = ReactorInspector(self.helper.call, self.fake_metrics,
+                                   loop_time=.1, log=log)
+        self.helper.ri = self.ri
+        self.logged = []
+
+    def check_log(self, *expected, **kw):
+        logLevel = kw.get("logLevel", None)
+        for (msg, level) in self.logged:
+            if level == logLevel and all(m in msg for m in expected):
+                return msg
+        return False
+
+    def run_ri(self, call_count=None, join=True):
+        """Set the call count and then run the ReactorInspector."""
+        if call_count is not None:
+            self.helper.call_count = call_count
+        # pylint: disable=W0201
+        self.start_ts = time.time()
+        self.ri.start()
+        # Reactor will stop after call_count calls, thanks to helper
+        if join:
+            self.ri.join()
+
+    def test_stop(self):
+        """It stops."""
+        self.run_ri(1000, join=False)
+        assert self.ri.is_alive()
+        self.ri.stop()
+        self.ri.join()
+        self.assertFalse(self.ri.is_alive())
+
+    @defer.inlineCallbacks
+    def test_dump_frames(self):
+        """Test how frames are dumped.
+
+        Rules:
+        - own frame must not be logged
+        - must log all other threads
+        - main reactor thread must have special title
+        """
+        # other thread, whose frame must be logged
+        waitingd = defer.Deferred()
+
+        def waiting_function():
+            """Function with funny name to be checked later."""
+            reactor.callFromThread(waitingd.callback, True)
+            # wait have a default value; pylint: disable=E1120
+            event.wait()
+
+        event = threading.Event()
+        threading.Thread(target=waiting_function).start()
+        # Make sure the thread has entered the waiting_function
+        yield waitingd
+
+        # Set reactor_thread since we're not starting the ReactorInspector
+        # thread here.
+        self.ri.reactor_thread = threading.currentThread().ident
+
+        # dump frames in other thread, also
+        def dumping_function():
+            """Function with funny name to be checked later."""
+            time.sleep(.1)
+            self.ri.dump_frames()
+            reactor.callFromThread(d.callback, True)
+
+        d = defer.Deferred()
+        threading.Thread(target=dumping_function).start()
+        yield d
+        event.set()
+
+        # check
+        self.assertFalse(self.check_log("dumping_function",
+                                        logLevel=logging.DEBUG))
+        self.assertTrue(self.check_log("Dumping Python frame",
+                                       "waiting_function",
+                                       logLevel=logging.DEBUG))
+        self.assertTrue(self.check_log("Dumping Python frame",
+                                       "reactor main thread",
+                                       logLevel=logging.DEBUG))
+
+    def test_reactor_ok(self):
+        """Reactor working fast."""
+        self.run_ri()
+        ok_line = self.assertTrue(self.check_log("ReactorInspector: ok",
+                                                 logLevel=logging.DEBUG))
+        # Check the metrics
+        delay = parse_delay(ok_line)
+        expected_metric = ("gauge", "delay", delay)
+        self.assertEqual([expected_metric], self.fake_metrics.calls)
+        self.assertTrue(self.ri.last_responsive_ts >= self.start_ts)
+
+    @defer.inlineCallbacks
+    def test_reactor_blocked(self):
+        """Reactor not working fast."""
+        dump_frames_called = defer.Deferred()
+        self.ri.dump_frames = lambda: dump_frames_called.callback(True)
+        self.run_ri(0)
+        yield dump_frames_called
+        log_line = self.check_log("ReactorInspector",
+                                  "detected unresponsive",
+                                  logLevel=logging.CRITICAL)
+        self.assertTrue(log_line)
+        delay = parse_delay(log_line)
+        self.assertTrue(delay >= .1)  # waited for entire loop time
+        # Check the metrics
+        expected_metric = ("gauge", "delay", delay)
+        self.assertEqual([expected_metric], self.fake_metrics.calls)
+
+        self.assertTrue(self.ri.last_responsive_ts < self.start_ts)
+
+    def test_reactor_back_alive(self):
+        """Reactor resurrects after some loops."""
+        self.run_ri(3)
+        late_line = self.check_log("ReactorInspector: late",
+                                   "got: 0", logLevel=logging.WARNING)
+        self.assertTrue(late_line)
+        delay = parse_delay(late_line)
+        self.assertTrue(delay >= .2)  # At least 2 cycles of delay
+        # Check the metrics
+        expected_metric = ("gauge", "delay", delay)
+        self.assertEqual(expected_metric, self.fake_metrics.calls[-1])
+
+        self.assertTrue(self.ri.queue.empty())
+        # A late reactor is not considered responsive (until a successful loop)
+        self.assertTrue(self.ri.last_responsive_ts < self.start_ts)

=== modified file 'txstatsd/tests/test_loggingprocessor.py'
--- txstatsd/tests/test_loggingprocessor.py	2012-06-28 17:29:26 +0000
+++ txstatsd/tests/test_loggingprocessor.py	2013-06-10 14:01:47 +0000
@@ -30,6 +30,7 @@
     def report(self, *args):
         return [('Sample report', 1, 2)]
 
+
 class TestLogger(object):
     def __init__(self):
         self.log = ''
@@ -63,7 +64,7 @@
         processor = LoggingMessageProcessor(logger)
         metric = FakeMeterMetric()
         processor.meter_metrics['test'] = metric
-        processor.flush()
+        list(processor.flush())
         expected = ["Out: %s %s %s" % message
                     for message in metric.report()]
         self.assertFalse(set(expected).difference(logger.log.splitlines()))
@@ -75,7 +76,7 @@
             time_function=lambda: 42)
         msg_in = "gorets:17|pd"
         processor.process(msg_in)
-        processor.flush()
+        list(processor.flush())
         messages = processor.plugin_metrics['gorets'].flush(
             10, processor.time_function())
         expected = ["In: %s" % msg_in] + ["Out: %s %s %s" % message

=== modified file 'txstatsd/tests/test_processor.py'
--- txstatsd/tests/test_processor.py	2012-06-28 17:29:26 +0000
+++ txstatsd/tests/test_processor.py	2013-06-10 14:01:47 +0000
@@ -189,22 +189,25 @@
         flushed.
         """
         self.timer.set([0,
-                        0, 1, # counter
-                        1, 3, # timer
-                        3, 6, # gauge
-                        6, 10, # meter
-                        10, 15, # plugin
+                        0, 1,  # counter
+                        1, 3,  # timer
+                        3, 6,  # gauge
+                        6, 10,  # meter
+                        10, 15,  # plugin
                         ])
-        def flush_metrics_summary(messages, num_stats, per_metric, timestamp):
+
+        def flush_metrics_summary(num_stats, per_metric, timestamp):
             self.assertEqual((0, 1), per_metric["counter"])
             self.assertEqual((0, 2), per_metric["timer"])
             self.assertEqual((0, 3), per_metric["gauge"])
             self.assertEqual((0, 4), per_metric["meter"])
             self.assertEqual((0, 5), per_metric["plugin"])
+            yield ()
+
         self.addCleanup(setattr, self.processor, "flush_metrics_summary",
                         self.processor.flush_metrics_summary)
         self.processor.flush_metrics_summary = flush_metrics_summary
-        self.processor.flush()
+        list(self.processor.flush())
 
     def test_flush_metrics_summary(self):
         """
@@ -215,14 +218,15 @@
         self.processor.process_timings = {"c": 1}
         self.processor.by_type = {"c": 42}
         messages = []
-        self.processor.flush_metrics_summary(messages, 1, per_metric, 42)
+        map(messages.extend, self.processor.flush_metrics_summary(
+            1, per_metric, 42))
         self.assertEqual(5, len(messages))
         self.assertEqual([('statsd.numStats', 1, 42),
                           ('statsd.flush.counter.count', 10, 42),
                           ('statsd.flush.counter.duration', 1000, 42),
                           ('statsd.receive.c.count', 42, 42),
                           ('statsd.receive.c.duration', 1000, 42)],
-                          messages)
+                         messages)
         self.assertEquals({}, self.processor.process_timings)
         self.assertEquals({}, self.processor.by_type)
 
@@ -238,7 +242,8 @@
         Flushing the message processor when there are no stats available should
         still produce one message where C{statsd.numStats} is set to zero.
         """
-        self.assertEqual(("statsd.numStats", 0, 42), self.processor.flush()[0])
+        self.assertEqual(("statsd.numStats", 0, 42),
+                         list(self.processor.flush())[0])
 
     def test_flush_counter(self):
         """
@@ -246,7 +251,7 @@
         normalized to the default interval.
         """
         self.processor.counter_metrics["gorets"] = 42
-        messages = self.processor.flush()
+        messages = list(self.processor.flush())
         self.assertEqual(("stats.gorets", 4, 42), messages[0])
         self.assertEqual(("stats_counts.gorets", 42, 42), messages[1])
         self.assertEqual(("statsd.numStats", 1, 42), messages[2])
@@ -258,7 +263,7 @@
         case the counter value will be unchanged.
         """
         self.processor.counter_metrics["gorets"] = 42
-        messages = self.processor.flush(interval=1000)
+        messages = list(self.processor.flush(interval=1000))
         self.assertEqual(("stats.gorets", 42, 42), messages[0])
         self.assertEqual(("stats_counts.gorets", 42, 42), messages[1])
         self.assertEqual(("statsd.numStats", 1, 42), messages[2])
@@ -271,7 +276,7 @@
         reset after flush is called.
         """
         self.processor.timer_metrics["glork"] = [24]
-        messages = self.processor.flush()
+        messages = list(self.processor.flush())
         self.assertEqual(("stats.timers.glork.count", 1, 42), messages[0])
         self.assertEqual(("stats.timers.glork.lower", 24, 42), messages[1])
         self.assertEqual(("stats.timers.glork.mean", 24, 42), messages[2])
@@ -290,7 +295,7 @@
         - mean will be the mean value within the 90th percentile
         """
         self.processor.timer_metrics["glork"] = [4, 8, 15, 16, 23, 42]
-        messages = self.processor.flush()
+        messages = list(self.processor.flush())
         self.assertEqual(("stats.timers.glork.count", 6, 42), messages[0])
         self.assertEqual(("stats.timers.glork.lower", 4, 42), messages[1])
         self.assertEqual(("stats.timers.glork.mean", 13, 42), messages[2])
@@ -312,7 +317,7 @@
         - mean will be the mean value within the 50th percentile
         """
         self.processor.timer_metrics["glork"] = [4, 8, 15, 16, 23, 42]
-        messages = self.processor.flush(percent=50)
+        messages = list(self.processor.flush(percent=50))
         self.assertEqual(("stats.timers.glork.count", 6, 42), messages[0])
         self.assertEqual(("stats.timers.glork.lower", 4, 42), messages[1])
         self.assertEqual(("stats.timers.glork.mean", 9, 42), messages[2])
@@ -329,7 +334,7 @@
 
         self.processor.process("gorets:9.6|g")
 
-        messages = self.processor.flush()
+        messages = list(self.processor.flush())
         self.assertEqual(
             ("stats.gauge.gorets.value", 9.6, 42), messages[0])
         self.assertEqual(
@@ -344,14 +349,14 @@
 
         self.processor.process("gorets:item|pd")
 
-        messages = self.processor.flush()
+        messages = list(self.processor.flush())
         self.assertEqual(("stats.pdistinct.gorets.count", 1, 42), messages[0])
         self.assertEqual(("stats.pdistinct.gorets.count_1day",
-                        5552568545, 42), messages[1])
+                          5552568545, 42), messages[1])
         self.assertEqual(("stats.pdistinct.gorets.count_1hour",
-                        5552568545, 42), messages[2])
+                          5552568545, 42), messages[2])
         self.assertEqual(("stats.pdistinct.gorets.count_1min",
-                        5552568545, 42), messages[3])
+                          5552568545, 42), messages[3])
 
     def test_flush_plugin_arguments(self):
         """Test the passing of arguments for flush."""
@@ -362,9 +367,9 @@
                 return []
 
         self.processor.plugin_metrics["somemetric"] = FakeMetric()
-        self.processor.flush(41000)
-        self.assertEquals((41, 42),
-            self.processor.plugin_metrics["somemetric"].data)
+        list(self.processor.flush(41000))
+        self.assertEquals(
+            (41, 42), self.processor.plugin_metrics["somemetric"].data)
 
 
 class FlushMeterMetricMessagesTest(TestCase):
@@ -384,7 +389,7 @@
         self.processor.process("gorets:3.0|m")
 
         self.time_now += 1
-        messages = self.processor.flush()
+        messages = list(self.processor.flush())
         self.assertEqual(
             ("stats.meter.gorets.count", 3.0, self.time_now),
             messages[0])
@@ -396,7 +401,7 @@
             messages[2])
 
         self.time_now += 60
-        messages = self.processor.flush()
+        messages = list(self.processor.flush())
         self.assertEqual(
             ("stats.meter.gorets.count", 3.0, self.time_now),
             messages[0])

=== modified file 'txstatsd/tests/test_service.py'
--- txstatsd/tests/test_service.py	2013-03-06 21:50:26 +0000
+++ txstatsd/tests/test_service.py	2013-06-10 14:01:47 +0000
@@ -307,4 +307,3 @@
 
         reactor.callWhenRunning(exercise)
         reactor.run()
-