Merge lp:~sidnei/txstatsd/platform-and-process-stats into lp:txstatsd

[Sidnei da Silva]

Add process and platform stats reading from U1. Allow for easily reporting a set of stats from the statsd server and from each statsd client.

[lp:~sidnei/txstatsd/platform-and-process-stats]

15. By Sidnei da Silva on 2011-07-07

- Get memory/cpu percent through psutil.

16. By Sidnei da Silva on 2011-07-07

- Report io counters if psutil implements them.

17. By Sidnei da Silva on 2011-07-07

- Wrong name

18. By Sidnei da Silva on 2011-07-07

- Add network reporting

19. By Sidnei da Silva on 2011-07-07

- Filter out non-tcp

20. By Sidnei da Silva on 2011-07-07

- Rename some keys

21. By Sidnei da Silva on 2011-07-07

- Add simple reporter classes for twisted threadpool and reactor.

22. By Sidnei da Silva on 2011-07-07

- Report reactor by default

23. By Sidnei da Silva on 2011-07-07

- Change meminfo and loadvg prefixes.

[Lucio Torre (lucio.torre)]

1. The metric prefixes have to end with ".", which seems error prone. The code should add that.
2. To report the metrics you are using counters, i know we need gauges and we dont have them, but incrementing because someone else divides later seems like we are leaking details.
3. Changing the interface of StatsDClientProtocol to take a function that is executed as an argument to report_stats that reads a file and then calls the target function seems like too much coupling for this specific use case.
4. socket.gethostname() will not work as metrics prefix on EC2, where the hostname is the internal ip and might change after rollouts.

I dont have solutions to some of those issues. Its your call.

[lp:~sidnei/txstatsd/platform-and-process-stats]

24. By Sidnei da Silva on 2011-07-12

- Do not require a trailing dot for prefix.
- Separate process stats reporting into a slightly more generic ReportingService.
- Make statsd plugin prefix configurable.

[Sidnei da Silva (sidnei)]

On Mon, Jul 11, 2011 at 9:48 PM, Lucio Torre <email address hidden> wrote:
> 1. The metric prefixes have to end with ".", which seems error prone. The code should add that.

The idea was that by not adding the '.' in the code, someone could use
an empty prefix, but that use-case is much more unlikely I guess.
Changed it to forcibly add '.'.

> 2. To report the metrics you are using counters, i know we need gauges and we dont have them, but incrementing because someone else divides later seems like we are leaking details.

Refactored a bit to make it easy to change from counters to gauges
when they are added (by passing a report_function around).

> 3. Changing the interface of StatsDClientProtocol to take a function that is executed as an argument to report_stats that reads a file and then calls the target function seems like too much coupling for this specific use case.

Removed that and created a ReportingService that keeps track of a list
of LoopingCall objects instead, with a schedule(function, interval,
report_function) method.

> 4. socket.gethostname() will not work as metrics prefix on EC2, where the hostname is the internal ip and might change after rollouts.

Indeed. I guess it'd be good practice to set a proper hostname once
you get the instance up. Anyway, made the prefix configurable through
the plugin by adding a --prefix argument.

[lp:~sidnei/txstatsd/platform-and-process-stats]

25. By Sidnei da Silva on 2011-07-12

- Don't fail if get_num_threads is not available.

26. By Sidnei da Silva on 2011-07-12

- Forgot to add report module

[Lucio Torre (lucio.torre)]

review: Approve

=== modified file '.bzrignore'
--- .bzrignore	2011-06-20 19:11:22 +0000
+++ .bzrignore	2011-07-12 13:56:27 +0000
@@ -1,1 +1,2 @@
 _trial_temp*
+twisted/plugins/dropin.cache

=== modified file 'example-stats-client.tac'
--- example-stats-client.tac	2011-06-23 14:19:21 +0000
+++ example-stats-client.tac	2011-07-12 13:56:27 +0000
@@ -8,11 +8,18 @@
 
 from txstatsd.protocol import StatsDClientProtocol
 from txstatsd.metrics import TransportMeter
+from txstatsd.process import PROCESS_STATS
+from txstatsd.report import ReportingService
 
 
 application = Application("example-stats-client")
-meter = TransportMeter(prefix=socket.gethostname())
-
+meter = TransportMeter(prefix=socket.gethostname() + ".example-client")
+
+reporting = ReportingService()
+reporting.setServiceParent(application)
+
+for report in PROCESS_STATS:
+    reporting.schedule(report, 10, meter.increment)
 
 def random_walker(name):
     """Meters a random walk."""
@@ -35,5 +42,5 @@
     t.start(0.5, now=False)
 
 
-protocol = StatsDClientProtocol("127.0.0.1", 8125, meter)
+protocol = StatsDClientProtocol("127.0.0.1", 8125, meter, 6000)
 reactor.listenUDP(0, protocol)

=== modified file 'statsd.tac'
--- statsd.tac	2011-07-04 21:43:01 +0000
+++ statsd.tac	2011-07-12 13:56:27 +0000
@@ -4,6 +4,5 @@
 
 application = Application("statsd")
 
-statsd_service = service.createService(service.StatsdOptions())
+statsd_service = service.createService(service.StatsDOptions())
 statsd_service.setServiceParent(application)
-

=== modified file 'txstatsd/metrics.py'
--- txstatsd/metrics.py	2011-07-05 05:27:22 +0000
+++ txstatsd/metrics.py	2011-07-12 13:56:27 +0000
@@ -55,6 +55,18 @@
         raise NotImplementedError()
 
 
+class InProcessMeter(BaseMeter):
+    """A meter that can be used inside the C{StatsD} daemon itself."""
+
+    def __init__(self, processor, prefix="", sample_rate=1):
+        self.processor = processor
+        BaseMeter.__init__(self, prefix=prefix, sample_rate=sample_rate)
+
+    def write(self, data):
+        """Pass the data along directly to the C{Processor}."""
+        self.processor.process(data)
+
+
 class Meter(BaseMeter):
     """A trivial, non-Twisted-dependent meter."""
 
@@ -118,12 +130,22 @@
     host = None
     port = None
 
-    def connected(self, transport, host, port):
+    def __init__(self, prefix="", sample_rate=1,
+                 connect_callback=None, disconnect_callback=None):
+        self.connect_callback = connect_callback
+        self.disconnect_callback = disconnect_callback
+        BaseMeter.__init__(self, prefix=prefix, sample_rate=sample_rate)
+
+    def connect(self, transport, host, port):
         self.transport = transport
         self.host = host
         self.port = port
+        if self.connect_callback is not None:
+            self.connect_callback()
 
-    def disconnected(self):
+    def disconnect(self):
+        if self.disconnect_callback is not None:
+            self.disconnect_callback()
         self.transport = self.host = self.port = None
 
     def write(self, data):

=== added file 'txstatsd/process.py'
--- txstatsd/process.py	1970-01-01 00:00:00 +0000
+++ txstatsd/process.py	2011-07-12 13:56:27 +0000
@@ -0,0 +1,171 @@
+import os
+import socket
+import psutil
+
+from functools import update_wrapper
+
+from twisted.internet import defer, fdesc, error
+
+
+MEMINFO_KEYS = ("MemTotal:", "MemFree:", "Buffers:",
+                "Cached:", "SwapCached:", "SwapTotal:",
+                "SwapFree:")
+
+MULTIPLIERS = {"kB": 1024, "mB": 1024 * 1024}
+
+
+def load_file(filename):
+    """Load a file into memory with non blocking reads."""
+
+    fd = os.open(filename, os.O_RDONLY)
+    fdesc.setNonBlocking(fd)
+
+    chunks = []
+    d = defer.Deferred()
+
+    def read_loop(data=None):
+        """Inner loop."""
+        if data is not None:
+            chunks.append(data)
+        r = fdesc.readFromFD(fd, read_loop)
+        if isinstance(r, error.ConnectionDone):
+            os.close(fd)
+            d.callback("".join(chunks))
+        elif r is not None:
+            os.close(fd)
+            d.errback(r)
+
+    read_loop("")
+    return d
+
+
+def parse_meminfo(data, prefix="sys.mem"):
+    """Parse data from /proc/meminfo."""
+    result = {}
+
+    for line in data.split("\n"):
+        if not line:
+            continue
+        parts = [x for x in line.split(" ") if x]
+        if not parts[0] in MEMINFO_KEYS:
+            continue
+
+        multiple = 1
+
+        if len(parts) == 3:
+            multiple = MULTIPLIERS[parts[2]]
+
+        # remove ':'
+        label = parts[0][:-1]
+        amount = int(parts[1]) * multiple
+        result[prefix + "." + label] = amount
+
+    return result
+
+
+def parse_loadavg(data, prefix="sys.loadavg"):
+    """Parse data from /proc/loadavg."""
+    return dict(zip(
+        (prefix + ".oneminute",
+         prefix + ".fiveminutes",
+         prefix + ".fifthteenminutes"),
+        [float(x) for x in data.split()[:3]]))
+
+
+def report_process_memory_and_cpu(process=psutil.Process(os.getpid()),
+                                  prefix="proc"):
+    """Report memory and CPU stats for C{process}."""
+    vsize, rss = process.get_memory_info()
+    utime, stime = process.get_cpu_times()
+    result = {prefix + ".cpu.percent": process.get_cpu_percent(),
+              prefix + ".cpu.user": utime,
+              prefix + ".cpu.system": stime,
+              prefix + ".memory.percent": process.get_memory_percent(),
+              prefix + ".memory.vsize": vsize,
+              prefix + ".memory.rss": rss}
+    if getattr(process, "get_num_threads", None) is not None:
+        result[prefix + ".threads"] = process.get_num_threads()
+    return result
+
+
+def report_process_io_counters(process=psutil.Process(os.getpid()),
+                               prefix="proc.io"):
+    """Report IO statistics for C{process}."""
+    result = {}
+    if getattr(process, "get_io_counters", None) is not None:
+        (read_count, write_count,
+         read_bytes, write_bytes) = process.get_io_counters()
+        result.update({
+            prefix + ".read.count": read_count,
+            prefix + ".write.count": write_count,
+            prefix + ".read.bytes": read_bytes,
+            prefix + ".write.bytes": write_bytes})
+    return result
+
+
+def report_process_net_stats(process=psutil.Process(os.getpid()),
+                             prefix="proc.net"):
+    """Report active connection statistics for C{process}."""
+    result = {}
+    if getattr(process, "get_connections", None) is not None:
+        for connection in process.get_connections():
+            fd, family, _type, laddr, raddr, status = connection
+            if _type == socket.SOCK_STREAM:
+                key = prefix + ".status.%s" % status.lower()
+                if not key in result:
+                    result[key] = 1
+                else:
+                    result[key] += 1
+    return result
+
+
+def report_system_stats(prefix="sys"):
+    cpu_times = psutil.cpu_times()
+    return {prefix + ".cpu.idle": cpu_times.idle,
+            prefix + ".cpu.iowait": cpu_times.iowait,
+            prefix + ".cpu.irq": cpu_times.irq,
+            prefix + ".cpu.nice": cpu_times.nice,
+            prefix + ".cpu.system": cpu_times.system,
+            prefix + ".cpu.user": cpu_times.user}
+
+
+def report_threadpool_stats(threadpool, prefix="threadpool"):
+    """Report stats about a given threadpool."""
+    def report():
+        return {prefix + ".working": len(threadpool.working),
+                prefix + ".queue": threadpool.q.qsize(),
+                prefix + ".waiters": len(threadpool.waiters),
+                prefix + ".threads": len(threadpool.threads)}
+    update_wrapper(report, report_threadpool_stats)
+    return report
+
+
+def report_reactor_stats(reactor, prefix="reactor"):
+    """Report statistics about a twisted reactor."""
+    def report():
+        return {prefix + ".readers": len(reactor.getReaders()),
+                prefix + ".writers": len(reactor.getWriters())}
+
+    update_wrapper(report, report_reactor_stats)
+    return report
+
+
+def report_file_stats(filename, parser):
+    """Read statistics from a file and report them."""
+    def report():
+        deferred = load_file(filename)
+        deferred.addCallback(parser)
+        return deferred
+    update_wrapper(report, report_file_stats)
+    return report
+
+
+PROCESS_STATS = (report_process_memory_and_cpu,)
+
+IO_STATS = (report_process_io_counters,)
+
+NET_STATS = (report_process_net_stats,)
+
+SYSTEM_STATS = (report_file_stats("/proc/meminfo", parse_meminfo),
+                report_file_stats("/proc/loadavg", parse_loadavg),
+                report_system_stats)

=== modified file 'txstatsd/protocol.py'
--- txstatsd/protocol.py	2011-06-22 00:18:58 +0000
+++ txstatsd/protocol.py	2011-07-12 13:56:27 +0000
@@ -1,7 +1,7 @@
 import logging
 
 from twisted.python import log
-from twisted.internet import task
+from twisted.internet import task, defer
 from twisted.protocols.basic import LineOnlyReceiver
 from twisted.internet.protocol import (
     DatagramProtocol, ReconnectingClientFactory)
@@ -27,21 +27,22 @@
 class StatsDClientProtocol(DatagramProtocol):
     """A Twisted-based implementation of the StatsD client protocol.
 
-    Data is sent via ConnectedUDP to a StatsD server for aggregation.
+    Data is sent via UDP to a StatsD server for aggregation.
     """
 
-    def __init__(self, host, port, meter):
+    def __init__(self, host, port, meter, interval=None):
         self.host = host
         self.port = port
         self.meter = meter
+        self.interval = interval
 
     def startProtocol(self):
         """Connect to destination host."""
-        self.meter.connected(self.transport, self.host, self.port)
+        self.meter.connect(self.transport, self.host, self.port)
 
     def stopProtocol(self):
         """Connection was lost."""
-        self.meter.disconnected()
+        self.meter.disconnect()
 
 
 class GraphiteProtocol(LineOnlyReceiver):
@@ -65,7 +66,7 @@
         log.msg("Connected. Scheduling flush to now + %ds." %
                 (self.interval / 1000), logLevel=logging.DEBUG)
         self.flush_task = task.LoopingCall(self.flushProcessor)
-        self.flush_task.start(self.interval / 1000)
+        self.flush_task.start(self.interval / 1000, False)
 
     def connectionLost(self, reason):
         """
@@ -77,12 +78,12 @@
             log.msg("Canceling scheduled flush.", logLevel=logging.DEBUG)
             self.flush_task.stop()
 
+    @defer.inlineCallbacks
     def flushProcessor(self):
         """Flush messages queued in the processor to Graphite."""
-        log.msg("Flushing messages.", logLevel=logging.DEBUG)
         for message in self.processor.flush(interval=self.interval):
             for line in message.splitlines():
-                self.sendLine(line)
+                yield self.sendLine(line)
 
 
 class GraphiteClientFactory(ReconnectingClientFactory):

=== added file 'txstatsd/report.py'
--- txstatsd/report.py	1970-01-01 00:00:00 +0000
+++ txstatsd/report.py	2011-07-12 13:56:27 +0000
@@ -0,0 +1,46 @@
+from twisted.internet.defer import maybeDeferred
+from twisted.internet.task import LoopingCall
+from twisted.python import log
+
+from functools import wraps
+
+from twisted.application.service import Service
+
+
+class ReportingService(Service):
+
+    def __init__(self):
+        self.tasks = []
+
+    def schedule(self, function, interval, report_function):
+        """
+        Schedule C{function} to be called every C{interval} seconds and then
+        report gathered metrics to C{Graphite} using C{report_function}.
+        """
+        task = LoopingCall(self.wrapped(function, report_function))
+        self.tasks.append((task, interval))
+
+    def wrapped(self, function, report_function):
+        def report_metrics(metrics):
+            """For each metric returned, call C{report_function} with it."""
+            for name, value in metrics.items():
+                report_function(name, value)
+            return metrics
+
+        @wraps(function)
+        def wrapper():
+            """Wrap C{function} to report metrics or log a failure."""
+            deferred = maybeDeferred(function)
+            deferred.addCallback(report_metrics)
+            deferred.addErrback(lambda failure: log.err(
+                failure, "Error while processing %s" % function.func_name))
+            return deferred
+        return wrapper
+
+    def startService(self):
+        for task, interval in self.tasks:
+            task.start(interval, now=False)
+
+    def stopService(self):
+        for task, interval in self.tasks:
+            task.stop()

=== modified file 'txstatsd/service.py'
--- txstatsd/service.py	2011-07-05 05:27:22 +0000
+++ txstatsd/service.py	2011-07-12 13:56:27 +0000
@@ -1,11 +1,16 @@
+import socket
 import ConfigParser
 
 from twisted.application.internet import TCPClient, UDPServer
 from twisted.application.service import MultiService
+from twisted.internet import reactor
 from twisted.python import usage, util
 
+from txstatsd import process
+from txstatsd.metrics import InProcessMeter
 from txstatsd.processor import MessageProcessor
 from txstatsd.protocol import GraphiteClientFactory, StatsDServerProtocol
+from txstatsd.report import ReportingService
 
 _unset = object()
 
@@ -73,13 +78,17 @@
     """
     glue_parameters = [
         ["carbon-cache-host", "h", "localhost",
-            "The host where carbon cache is listening."],
+         "The host where carbon cache is listening."],
         ["carbon-cache-port", "p", 2003,
-            "The port where carbon cache is listening.", int],
+         "The port where carbon cache is listening.", int],
         ["listen-port", "l", 8125,
-            "The UDP port where we will listen.", int],
+         "The UDP port where we will listen.", int],
         ["flush-interval", "i", 10000,
-            "The number of milliseconds between each flush.", int],
+         "The number of milliseconds between each flush.", int],
+        ["prefix", "p", None,
+         "Prefix to use when reporting stats.", str],
+        ["report", "r", None,
+         "Which additional stats to report {process|net|io|system}.", str],
     ]
 
 
@@ -89,11 +98,27 @@
     service = MultiService()
     service.setName("statsd")
     processor = MessageProcessor()
+    prefix = options["prefix"]
+    if prefix is None:
+        prefix = socket.gethostname() + ".statsd"
+
+    meter = InProcessMeter(processor, prefix=prefix)
+
+    if options["report"] is not None:
+        reporting = ReportingService()
+        reporting.setServiceParent(service)
+        reporting.schedule(
+            process.report_reactor_stats(reactor), 10, meter.increment)
+        reports = [name.strip() for name in options["report"].split(",")]
+        for report_name in reports:
+            for reporter in getattr(process, "%s_STATS" %
+                                    report_name.upper(), ()):
+                reporting.schedule(reporter, 10, meter.increment)
 
     factory = GraphiteClientFactory(processor, options["flush-interval"])
-    client = TCPClient(
-        options["carbon-cache-host"], options["carbon-cache-port"],
-        factory)
+    client = TCPClient(options["carbon-cache-host"],
+                       options["carbon-cache-port"],
+                       factory)
     client.setServiceParent(service)
 
     listener = UDPServer(options["listen-port"],

=== added file 'txstatsd/tests/test_process.py'
--- txstatsd/tests/test_process.py	1970-01-01 00:00:00 +0000
+++ txstatsd/tests/test_process.py	2011-07-12 13:56:27 +0000
@@ -0,0 +1,266 @@
+import os
+import psutil
+
+from mocker import MockerTestCase
+from twisted.internet import defer
+from twisted.trial.unittest import TestCase
+
+from txstatsd.process import (
+    load_file, parse_meminfo, parse_loadavg,
+    report_process_memory_and_cpu, report_process_io_counters,
+    report_process_net_stats, report_system_stats,
+    report_reactor_stats, report_threadpool_stats)
+
+
+meminfo = """\
+MemTotal:        8190436 kB
+MemFree:          995724 kB
+Buffers:            8052 kB
+Cached:           344824 kB
+SwapCached:       170828 kB
+Active:          4342436 kB
+Inactive:         907076 kB
+Active(anon):    4210168 kB
+Inactive(anon):   756096 kB
+Active(file):     132268 kB
+Inactive(file):   150980 kB
+Unevictable:      641692 kB
+Mlocked:          641676 kB
+SwapTotal:      23993336 kB
+SwapFree:       22750588 kB
+Dirty:               740 kB
+Writeback:             0 kB
+AnonPages:       5453396 kB
+Mapped:           259524 kB
+Shmem:             69952 kB
+Slab:             142444 kB
+SReclaimable:      83188 kB
+SUnreclaim:        59256 kB
+KernelStack:       16144 kB
+PageTables:        88384 kB
+NFS_Unstable:          0 kB
+Bounce:                0 kB
+WritebackTmp:          0 kB
+CommitLimit:    28088552 kB
+Committed_AS:   11178564 kB
+VmallocTotal:   34359738367 kB
+VmallocUsed:      156208 kB
+VmallocChunk:   34359579400 kB
+HardwareCorrupted:     0 kB
+HugePages_Total:       0
+HugePages_Free:        0
+HugePages_Rsvd:        0
+HugePages_Surp:        0
+Hugepagesize:       2048 kB
+DirectMap4k:     7968640 kB
+DirectMap2M:      415744 kB"""
+
+
+class TestSystemPerformance(TestCase, MockerTestCase):
+    """Test system performance monitoring."""
+
+    def assertSuccess(self, deferred, result=None):
+        """
+        Assert that the given C{deferred} results in the given C{result}.
+        """
+        self.assertTrue(isinstance(deferred, defer.Deferred))
+        return deferred.addCallback(self.assertEqual, result)
+
+    def test_read(self):
+        """We can read files non blocking."""
+        d = load_file(__file__)
+        return self.assertSuccess(d, open(__file__).read())
+
+    def test_loadinfo(self):
+        """We understand loadinfo."""
+        loadinfo = "1.02 1.08 1.14 2/2015 19420"
+        self.assertEqual(parse_loadavg(loadinfo), {
+            "sys.loadavg.oneminute": 1.02,
+            "sys.loadavg.fiveminutes": 1.08,
+            "sys.loadavg.fifthteenminutes": 1.14})
+
+    def test_meminfo(self):
+        """We understand meminfo."""
+        r = parse_meminfo(meminfo)
+        self.assertEqual(r['sys.mem.Buffers'], 8052 * 1024)
+        self.assert_('sys.mem.HugePages_Rsvd' not in r)
+
+    def test_statinfo(self):
+        """System stat info is collected through psutil."""
+        cpu_times = psutil.cpu_times()
+        mock = self.mocker.replace("psutil.cpu_times")
+        self.expect(mock()).result(cpu_times)
+        self.mocker.replay()
+
+        result = report_system_stats()
+        self.assertEqual(cpu_times.idle, result["sys.cpu.idle"])
+        self.assertEqual(cpu_times.iowait, result["sys.cpu.iowait"])
+        self.assertEqual(cpu_times.irq, result["sys.cpu.irq"])
+        self.assertEqual(cpu_times.nice, result["sys.cpu.nice"])
+        self.assertEqual(cpu_times.system, result["sys.cpu.system"])
+        self.assertEqual(cpu_times.user, result["sys.cpu.user"])
+
+    def test_self_statinfo(self):
+        """
+        Process stat info is collected through psutil.
+
+        If the L{Process} implementation does not have C{get_num_threads} then
+        the number of threads will not be included in the output.
+        """
+        process = psutil.Process(os.getpid())
+        vsize, rss = process.get_memory_info()
+        utime, stime = process.get_cpu_times()
+        cpu_percent = process.get_cpu_percent()
+        memory_percent = process.get_memory_percent()
+
+        mock = self.mocker.mock()
+        self.expect(mock.get_memory_info()).result((vsize, rss))
+        self.expect(mock.get_cpu_times()).result((utime, stime))
+        self.expect(mock.get_cpu_percent()).result(cpu_percent)
+        self.expect(mock.get_memory_percent()).result(memory_percent)
+        self.expect(mock.get_num_threads).result(None)
+        self.mocker.replay()
+
+        result = report_process_memory_and_cpu(process=mock)
+        self.assertEqual(utime, result["proc.cpu.user"])
+        self.assertEqual(stime, result["proc.cpu.system"])
+        self.assertEqual(cpu_percent, result["proc.cpu.percent"])
+        self.assertEqual(vsize, result["proc.memory.vsize"])
+        self.assertEqual(rss, result["proc.memory.rss"])
+        self.assertEqual(memory_percent, result["proc.memory.percent"])
+        self.failIf("proc.threads" in result)
+
+    def test_self_statinfo_with_num_threads(self):
+        """
+        Process stat info is collected through psutil.
+
+        If the L{Process} implementation contains C{get_num_threads} then the
+        number of threads will be included in the output.
+
+        """
+        process = psutil.Process(os.getpid())
+        vsize, rss = process.get_memory_info()
+        utime, stime = process.get_cpu_times()
+        cpu_percent = process.get_cpu_percent()
+        memory_percent = process.get_memory_percent()
+
+        mock = self.mocker.mock()
+        self.expect(mock.get_memory_info()).result((vsize, rss))
+        self.expect(mock.get_cpu_times()).result((utime, stime))
+        self.expect(mock.get_cpu_percent()).result(cpu_percent)
+        self.expect(mock.get_memory_percent()).result(memory_percent)
+        self.expect(mock.get_num_threads()).result(1)
+        self.mocker.replay()
+
+        result = report_process_memory_and_cpu(process=mock)
+        self.assertEqual(utime, result["proc.cpu.user"])
+        self.assertEqual(stime, result["proc.cpu.system"])
+        self.assertEqual(cpu_percent, result["proc.cpu.percent"])
+        self.assertEqual(vsize, result["proc.memory.vsize"])
+        self.assertEqual(rss, result["proc.memory.rss"])
+        self.assertEqual(memory_percent, result["proc.memory.percent"])
+        self.assertEqual(1, result["proc.threads"])
+
+    def test_ioinfo(self):
+        """Process IO info is collected through psutil."""
+        mock = self.mocker.mock()
+        self.expect(mock.get_io_counters).result(None)
+        self.mocker.replay()
+
+        # If the version of psutil doesn't have the C{get_io_counters},
+        # then io stats are not included in the output.
+        result = report_process_io_counters(process=mock)
+        self.failIf("proc.io.read.count" in result)
+        self.failIf("proc.io.write.count" in result)
+        self.failIf("proc.io.read.bytes" in result)
+        self.failIf("proc.io.write.bytes" in result)
+
+    def test_ioinfo_with_get_io_counters(self):
+        """
+        Process IO info is collected through psutil.
+
+        If C{get_io_counters} is implemented by the L{Process} object,
+        then io information will be returned with the process information.
+        """
+        io_counters = (10, 42, 125, 16)
+
+        mock = self.mocker.mock()
+        self.expect(mock.get_io_counters).result(mock)
+        self.expect(mock.get_io_counters()).result(io_counters)
+        self.mocker.replay()
+
+        result = report_process_io_counters(process=mock)
+        self.assertEqual(10, result["proc.io.read.count"])
+        self.assertEqual(42, result["proc.io.write.count"])
+        self.assertEqual(125, result["proc.io.read.bytes"])
+        self.assertEqual(16, result["proc.io.write.bytes"])
+
+    def test_netinfo_no_get_connections(self):
+        """
+        Process connection info is collected through psutil.
+
+        If the version of psutil doesn't implement C{get_connections} for
+        L{Process}, then no information is returned.
+        """
+        mock = self.mocker.mock()
+        self.expect(mock.get_connections).result(None)
+        self.mocker.replay()
+
+        # If the version of psutil doesn't have the C{get_io_counters},
+        # then io stats are not included in the output.
+        result = report_process_net_stats(process=mock)
+        self.failIf("proc.net.status.established" in result)
+
+    def test_netinfo_with_get_connections(self):
+        """
+        Process connection info is collected through psutil.
+
+        If the version of psutil implements C{get_connections} for L{Process},
+        then a count of connections in each state is returned.
+        """
+        connections = [
+            (115, 2, 1, ("10.0.0.1", 48776),
+             ("93.186.135.91", 80), "ESTABLISHED"),
+            (117, 2, 1, ("10.0.0.1", 43761),
+             ("72.14.234.100", 80), "CLOSING"),
+            (119, 2, 1, ("10.0.0.1", 60759),
+             ("72.14.234.104", 80), "ESTABLISHED"),
+            (123, 2, 1, ("10.0.0.1", 51314),
+             ("72.14.234.83", 443), "SYN_SENT")
+            ]
+
+        mock = self.mocker.mock()
+        self.expect(mock.get_connections).result(mock)
+        self.expect(mock.get_connections()).result(connections)
+        self.mocker.replay()
+
+        result = report_process_net_stats(process=mock)
+        self.assertEqual(2, result["proc.net.status.established"])
+        self.assertEqual(1, result["proc.net.status.closing"])
+        self.assertEqual(1, result["proc.net.status.syn_sent"])
+
+    def test_reactor_stats(self):
+        """Given a twisted reactor, pull out some stats from it."""
+        mock = self.mocker.mock()
+        self.expect(mock.getReaders()).result([None, None, None])
+        self.expect(mock.getWriters()).result([None, None])
+        self.mocker.replay()
+
+        result = report_reactor_stats(mock)()
+        self.assertEqual(3, result["reactor.readers"])
+        self.assertEqual(2, result["reactor.writers"])
+
+    def test_threadpool_stats(self):
+        """Given a twisted threadpool, pull out some stats from it."""
+        mock = self.mocker.mock()
+        self.expect(mock.q.qsize()).result(42)
+        self.expect(mock.threads).result(6 * [None])
+        self.expect(mock.waiters).result(2 * [None])
+        self.expect(mock.working).result(4 * [None])
+        self.mocker.replay()
+
+        result = report_threadpool_stats(mock)()
+        self.assertEqual(42, result["threadpool.queue"])
+        self.assertEqual(6, result["threadpool.threads"])
+        self.assertEqual(2, result["threadpool.waiters"])
+        self.assertEqual(4, result["threadpool.working"])

=== modified file 'txstatsd/tests/test_service.py'
--- txstatsd/tests/test_service.py	2011-07-05 05:27:22 +0000
+++ txstatsd/tests/test_service.py	2011-07-12 13:56:27 +0000
@@ -6,6 +6,7 @@
 
 
 class GlueOptionsTestCase(TestCase):
+
     def test_defaults(self):
         """
         Defaults get passed over to the instance.
@@ -94,6 +95,7 @@
 
 
 class ServiceTests(TestCase):
+
     def test_service(self):
         """
         The StatsD service can be created.