Merge lp:~ahasenack/txstatsd/txstatsd-packaging into lp:txstatsd

[Andreas Hasenack]

This branch adds a debian/ structure to the txstatsd source tree. It builds and installs, and python sees the module after installation, as does twistd.

The {post,pre}inst and {post,pre}rm scripts are straight from dh_make, I didn't change those, and seem to do the right thing. After installation, I checked and postinst had a section added by dh_pysupport.

These are the lint warnings I get:
W: python-txstatsd source: out-of-date-standards-version 3.8.3 (current is 3.9.1)
W: python-txstatsd: maintainer-script-empty postrm
W: python-txstatsd: maintainer-script-empty preinst

[Sidnei da Silva (sidnei)]

python-psutil needs to be added as a dependency.

[lp:~ahasenack/txstatsd/txstatsd-packaging]

15. By Andreas Hasenack on 2011-07-12

Merged with trunk (the ~landscape one, not upstream).

16. By Andreas Hasenack on 2011-07-12

Added dependency for python-psutil.

17. By Andreas Hasenack on 2011-07-12

Include example-stats-client.tac in the documentation.

18. By Andreas Hasenack on 2011-07-12

One more example file for the docs.

19. By Andreas Hasenack on 2011-07-12

- Merged with trunk again.
- Added statsd.tac to the documentation directory.

20. By Andreas Hasenack on 2011-07-12

Change source package name to just txstatsd.

21. By Andreas Hasenack on 2011-07-12

- Changed source package name to just txstatsd
- Changed build dependency from python-twisted to python-twisted-core

22. By Andreas Hasenack on 2011-07-12

Merged with trunk to grab missing report.py.

23. By Andreas Hasenack on 2011-07-12

Adjusted dependencies.

24. By Andreas Hasenack on 2011-07-12

Actually, no need for python-dev as this is noarch.

25. By Andreas Hasenack on 2011-07-12

Fixed version/release numbers.

Unmerged revisions

25. By Andreas Hasenack on 2011-07-12

Fixed version/release numbers.

24. By Andreas Hasenack on 2011-07-12

Actually, no need for python-dev as this is noarch.

23. By Andreas Hasenack on 2011-07-12

Adjusted dependencies.

22. By Andreas Hasenack on 2011-07-12

Merged with trunk to grab missing report.py.

21. By Andreas Hasenack on 2011-07-12

- Changed source package name to just txstatsd
- Changed build dependency from python-twisted to python-twisted-core

20. By Andreas Hasenack on 2011-07-12

Change source package name to just txstatsd.

19. By Andreas Hasenack on 2011-07-12

- Merged with trunk again.
- Added statsd.tac to the documentation directory.

18. By Andreas Hasenack on 2011-07-12

One more example file for the docs.

17. By Andreas Hasenack on 2011-07-12

Include example-stats-client.tac in the documentation.

16. By Andreas Hasenack on 2011-07-12

Added dependency for python-psutil.

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

=== added directory 'debian'
=== added file 'debian/changelog'
--- debian/changelog	1970-01-01 00:00:00 +0000
+++ debian/changelog	2011-07-12 14:22:56 +0000
@@ -0,0 +1,5 @@
+txstatsd (0.1.0-0ubuntu1) unstable; urgency=low
+
+  * Initial Release.
+
+ -- Andreas Hasenack <[e-mail]>  Mon, 11 Jul 2011 19:29:34 -0300

=== added file 'debian/compat'
--- debian/compat	1970-01-01 00:00:00 +0000
+++ debian/compat	2011-07-12 14:22:56 +0000
@@ -0,0 +1,1 @@
+7

=== added file 'debian/control'
--- debian/control	1970-01-01 00:00:00 +0000
+++ debian/control	2011-07-12 14:22:56 +0000
@@ -0,0 +1,15 @@
+Source: txstatsd
+Section: python
+Priority: extra
+Maintainer: Andreas Hasenack <[e-mail]>
+Build-Depends: debhelper (>= 7), python-twisted-core, python-support
+Standards-Version: 3.8.3
+Homepage: https://launchpad.net/txstatsd
+
+Package: python-txstatsd
+Architecture: all
+Depends: ${python:Depends}, ${misc:Depends}, python-psutil, python-twisted-core
+Description: A network daemon for aggregating statistics
+ A network daemon for aggregating statistics (counters and timers), rolling
+ them up, then sending them to graphite. Really a port of
+ https://github.com/etsy/statsd to twisted.

=== added file 'debian/copyright'
--- debian/copyright	1970-01-01 00:00:00 +0000
+++ debian/copyright	2011-07-12 14:22:56 +0000
@@ -0,0 +1,43 @@
+This work was packaged for Debian by:
+
+    Andreas Hasenack <[e-mail]> on Mon, 11 Jul 2011 19:29:34 -0300
+
+It was downloaded from:
+
+    https://launchpad.net/txstatsd
+
+Upstream Author(s):
+
+    Sidnei da Silva <[e-mail]>
+
+Copyright:
+
+    <Copyright (C) 2011 Canonical Services Ltd>
+
+License:
+
+    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.
+
+The Debian packaging is:
+
+    Copyright (C) 2011 Andreas Hasenack <[e-mail]>
+
+and is licensed under MIT.
+

=== added file 'debian/docs'
--- debian/docs	1970-01-01 00:00:00 +0000
+++ debian/docs	2011-07-12 14:22:56 +0000
@@ -0,0 +1,4 @@
+README
+example-stats-client.tac
+txstatsd.conf-example
+statsd.tac

=== added file 'debian/postinst'
--- debian/postinst	1970-01-01 00:00:00 +0000
+++ debian/postinst	2011-07-12 14:22:56 +0000
@@ -0,0 +1,39 @@
+#!/bin/sh
+# postinst script for python-txstatsd
+#
+# see: dh_installdeb(1)
+
+set -e
+
+# summary of how this script can be called:
+#        * <postinst> `configure' <most-recently-configured-version>
+#        * <old-postinst> `abort-upgrade' <new version>
+#        * <conflictor's-postinst> `abort-remove' `in-favour' <package>
+#          <new-version>
+#        * <postinst> `abort-remove'
+#        * <deconfigured's-postinst> `abort-deconfigure' `in-favour'
+#          <failed-install-package> <version> `removing'
+#          <conflicting-package> <version>
+# for details, see http://www.debian.org/doc/debian-policy/ or
+# the debian-policy package
+
+
+case "$1" in
+    configure)
+    ;;
+
+    abort-upgrade|abort-remove|abort-deconfigure)
+    ;;
+
+    *)
+        echo "postinst called with unknown argument \`$1'" >&2
+        exit 1
+    ;;
+esac
+
+# dh_installdeb will replace this with shell code automatically
+# generated by other debhelper scripts.
+
+#DEBHELPER#
+
+exit 0

=== added file 'debian/postrm'
--- debian/postrm	1970-01-01 00:00:00 +0000
+++ debian/postrm	2011-07-12 14:22:56 +0000
@@ -0,0 +1,37 @@
+#!/bin/sh
+# postrm script for python-txstatsd
+#
+# see: dh_installdeb(1)
+
+set -e
+
+# summary of how this script can be called:
+#        * <postrm> `remove'
+#        * <postrm> `purge'
+#        * <old-postrm> `upgrade' <new-version>
+#        * <new-postrm> `failed-upgrade' <old-version>
+#        * <new-postrm> `abort-install'
+#        * <new-postrm> `abort-install' <old-version>
+#        * <new-postrm> `abort-upgrade' <old-version>
+#        * <disappearer's-postrm> `disappear' <overwriter>
+#          <overwriter-version>
+# for details, see http://www.debian.org/doc/debian-policy/ or
+# the debian-policy package
+
+
+case "$1" in
+    purge|remove|upgrade|failed-upgrade|abort-install|abort-upgrade|disappear)
+    ;;
+
+    *)
+        echo "postrm called with unknown argument \`$1'" >&2
+        exit 1
+    ;;
+esac
+
+# dh_installdeb will replace this with shell code automatically
+# generated by other debhelper scripts.
+
+#DEBHELPER#
+
+exit 0

=== added file 'debian/preinst'
--- debian/preinst	1970-01-01 00:00:00 +0000
+++ debian/preinst	2011-07-12 14:22:56 +0000
@@ -0,0 +1,35 @@
+#!/bin/sh
+# preinst script for python-txstatsd
+#
+# see: dh_installdeb(1)
+
+set -e
+
+# summary of how this script can be called:
+#        * <new-preinst> `install'
+#        * <new-preinst> `install' <old-version>
+#        * <new-preinst> `upgrade' <old-version>
+#        * <old-preinst> `abort-upgrade' <new-version>
+# for details, see http://www.debian.org/doc/debian-policy/ or
+# the debian-policy package
+
+
+case "$1" in
+    install|upgrade)
+    ;;
+
+    abort-upgrade)
+    ;;
+
+    *)
+        echo "preinst called with unknown argument \`$1'" >&2
+        exit 1
+    ;;
+esac
+
+# dh_installdeb will replace this with shell code automatically
+# generated by other debhelper scripts.
+
+#DEBHELPER#
+
+exit 0

=== added file 'debian/prerm'
--- debian/prerm	1970-01-01 00:00:00 +0000
+++ debian/prerm	2011-07-12 14:22:56 +0000
@@ -0,0 +1,38 @@
+#!/bin/sh
+# prerm script for python-txstatsd
+#
+# see: dh_installdeb(1)
+
+set -e
+
+# summary of how this script can be called:
+#        * <prerm> `remove'
+#        * <old-prerm> `upgrade' <new-version>
+#        * <new-prerm> `failed-upgrade' <old-version>
+#        * <conflictor's-prerm> `remove' `in-favour' <package> <new-version>
+#        * <deconfigured's-prerm> `deconfigure' `in-favour'
+#          <package-being-installed> <version> `removing'
+#          <conflicting-package> <version>
+# for details, see http://www.debian.org/doc/debian-policy/ or
+# the debian-policy package
+
+
+case "$1" in
+    remove|upgrade|deconfigure)
+    ;;
+
+    failed-upgrade)
+    ;;
+
+    *)
+        echo "prerm called with unknown argument \`$1'" >&2
+        exit 1
+    ;;
+esac
+
+# dh_installdeb will replace this with shell code automatically
+# generated by other debhelper scripts.
+
+#DEBHELPER#
+
+exit 0

=== added file 'debian/rules'
--- debian/rules	1970-01-01 00:00:00 +0000
+++ debian/rules	2011-07-12 14:22:56 +0000
@@ -0,0 +1,13 @@
+#!/usr/bin/make -f
+# -*- makefile -*-
+# Sample debian/rules that uses debhelper.
+# This file was originally written by Joey Hess and Craig Small.
+# As a special exception, when this file is copied by dh-make into a
+# dh-make output file, you may use that output file without restriction.
+# This special exception was added by Craig Small in version 0.37 of dh-make.
+
+# Uncomment this to turn on verbose mode.
+#export DH_VERBOSE=1
+
+%:
+	dh  $@

=== modified file 'example-stats-client.tac'
--- example-stats-client.tac	2011-06-23 14:19:21 +0000
+++ example-stats-client.tac	2011-07-12 14:22:56 +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 14:22:56 +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 14:22:56 +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 14:22:56 +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 14:22:56 +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 14:22:56 +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 14:22:56 +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 14:22:56 +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 14:22:56 +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.