Merge lp:~sidnei/txstatsd/split-counters into lp:txstatsd

[Sidnei da Silva]

Fix counter-like reports to report delta since last call

Fix counter-like reporters to report delta since last call.

[Lucio Torre (lucio.torre)]

review: Approve

=== modified file 'txstatsd/process.py'
--- txstatsd/process.py	2011-11-18 17:13:52 +0000
+++ txstatsd/process.py	2012-01-28 00:28:23 +0000
@@ -4,8 +4,6 @@
 
 from functools import update_wrapper
 
-from twisted.internet import defer, fdesc, error
-
 
 MEMINFO_KEYS = ("MemTotal:", "MemFree:", "Buffers:",
                 "Cached:", "SwapCached:", "SwapTotal:",
@@ -15,27 +13,9 @@
 
 
 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
+    """Load a file into memory."""
+    with open(filename, "r") as f:
+        return f.read()
 
 
 def parse_meminfo(data, prefix="sys.mem"):
@@ -110,10 +90,7 @@
     def get_memory_and_cpu(self, prefix="proc"):
         """Report memory and CPU stats for C{process}."""
         vsize, rss = self.process.get_memory_info()
-        utime, stime = self.process.get_cpu_times()
         result = {prefix + ".cpu.percent": self.process.get_cpu_percent(),
-                  prefix + ".cpu.user": utime,
-                  prefix + ".cpu.system": stime,
                   prefix + ".memory.percent":
                     self.process.get_memory_percent(),
                   prefix + ".memory.vsize": vsize,
@@ -122,6 +99,13 @@
             result[prefix + ".threads"] = self.process.get_num_threads()
         return result
 
+    def get_cpu_counters(self, prefix="proc"):
+        """Report memory and CPU counters for C{process}."""
+        utime, stime = self.process.get_cpu_times()
+        result = {prefix + ".cpu.user": utime,
+                  prefix + ".cpu.system": stime}
+        return result
+
     def get_io_counters(self, prefix="proc.io"):
         """Report IO statistics for C{process}."""
         result = {}
@@ -150,9 +134,34 @@
         return result
 
 
+def report_counters(report_function):
+    """
+    Report difference between last value and current value for wrapped
+    function.
+    """
+    def generate():
+        last_value = None
+        while True:
+            result = {}
+            new_value = report_function()
+            if last_value is None:
+                last_value = new_value
+            else:
+                for key, value in new_value.iteritems():
+                    result[key] = value - last_value[key]
+                last_value = new_value
+            yield result
+    generator = generate()
+    def report():
+        return generator.next()
+    update_wrapper(report, report_function)
+    return report
+
+
 process_report = ProcessReport()
 report_process_memory_and_cpu = process_report.get_memory_and_cpu
-report_process_io_counters = process_report.get_io_counters
+report_process_cpu_counters = report_counters(process_report.get_cpu_counters)
+report_process_io_counters = report_counters(process_report.get_io_counters)
 report_process_net_stats = process_report.get_net_stats
 
 
@@ -188,20 +197,21 @@
 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
+        return parser(load_file(filename))
     update_wrapper(report, report_file_stats)
     return report
 
 
 PROCESS_STATS = (report_process_memory_and_cpu,)
 
+COUNTER_STATS = (report_process_cpu_counters,)
+
 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_file_stats("/proc/net/dev", parse_netdev),
-                report_system_stats)
+                report_counters(report_file_stats("/proc/net/dev", parse_netdev)),
+                report_counters(report_system_stats),
+                )

=== modified file 'txstatsd/service.py'
--- txstatsd/service.py	2012-01-27 16:11:50 +0000
+++ txstatsd/service.py	2012-01-28 00:28:23 +0000
@@ -11,6 +11,7 @@
 
 from txstatsd.client import InternalClient
 from txstatsd.metrics.metrics import Metrics
+from txstatsd.metrics.extendedmetrics import ExtendedMetrics
 from txstatsd.server.processor import MessageProcessor
 from txstatsd.server.configurableprocessor import ConfigurableMessageProcessor
 from txstatsd.server.protocol import (
@@ -235,13 +236,13 @@
         processor = MessageProcessor(plugins=plugin_metrics)
         input_router = Router(processor, options['routing'], root_service)
         connection = InternalClient(input_router)
-        metrics = Metrics(connection, namespace=prefix)
+        metrics = Metrics(connection)
     else:
         processor = ConfigurableMessageProcessor(message_prefix=prefix,
                                                  plugins=plugin_metrics)
         input_router = Router(processor, options['routing'], root_service)
         connection = InternalClient(input_router)
-        metrics = Metrics(connection)
+        metrics = ExtendedMetrics(connection)
 
     if not options["carbon-cache-host"]:
         options["carbon-cache-host"].append("127.0.0.1")
@@ -259,20 +260,23 @@
 
     # Schedule updates for those metrics expecting to be
     # periodically updated, for example the meter metric.
-    reporting.schedule(processor.update_metrics, 5, None)
-    reporting.schedule(report_client_manager_stats, 10, metrics.gauge)
+    reporting.schedule(processor.update_metrics, 10, None)
+    reporting.schedule(report_client_manager_stats,
+                       options["flush-interval"] / 1000,
+                       metrics.gauge)
 
     if options["report"] is not None:
         from txstatsd import process
         from twisted.internet import reactor
 
         reporting.schedule(
-            process.report_reactor_stats(reactor), 10, metrics.gauge)
+            process.report_reactor_stats(reactor), 60, metrics.gauge)
         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, metrics.gauge)
+                reporting.schedule(reporter, 60, metrics.gauge)
+
 
     # XXX Make this configurable.
     router = ConsistentHashingRouter()

=== modified file 'txstatsd/tests/test_process.py'
--- txstatsd/tests/test_process.py	2011-08-08 09:28:45 +0000
+++ txstatsd/tests/test_process.py	2012-01-28 00:28:23 +0000
@@ -3,12 +3,11 @@
 import sys
 
 from mocker import MockerTestCase
-from twisted.internet import defer
 from twisted.trial.unittest import TestCase
 
 from txstatsd.process import (
     ProcessReport, load_file, parse_meminfo, parse_loadavg, parse_netdev,
-    report_system_stats, report_reactor_stats, report_threadpool_stats)
+    report_system_stats, report_reactor_stats, report_threadpool_stats, report_counters)
 
 
 meminfo = """\
@@ -65,18 +64,6 @@
 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"
@@ -91,8 +78,8 @@
         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."""
+    def test_cpu_counters(self):
+        """System cpu counters are collected through psutil."""
         cpu_times = psutil.cpu_times()
         mock = self.mocker.replace("psutil.cpu_times")
         self.expect(mock()).result(cpu_times)
@@ -121,39 +108,50 @@
             self.assertEqual(cpu_times.idle, result["sys.cpu.idle"])
             self.assertEqual(cpu_times.irq, result["sys.cpu.irq"])
 
-    def test_self_statinfo(self):
+    def test_self_cpu_and_memory_stats(self):
         """
-        Process stat info is collected through psutil.
+        Process cpu and memory stats are 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 = ProcessReport(process=mock).get_memory_and_cpu()
-        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.
+    def test_self_cpu_counters(self):
+        """
+        Process cpu counters are collected through psutil.
+        """
+        process = psutil.Process(os.getpid())
+        utime, stime = process.get_cpu_times()
+
+        mock = self.mocker.mock()
+        self.expect(mock.get_cpu_times()).result((utime, stime))
+        self.mocker.replay()
+
+        result = ProcessReport(process=mock).get_cpu_counters()
+        self.assertEqual(utime, result["proc.cpu.user"])
+        self.assertEqual(stime, result["proc.cpu.system"])
+
+    def test_self_cpu_and_memory_stats_with_num_threads(self):
+        """
+        Process cpu and memory stats are collected through psutil.
 
         If the L{Process} implementation contains C{get_num_threads} then the
         number of threads will be included in the output.
@@ -161,21 +159,17 @@
         """
         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 = ProcessReport(process=mock).get_memory_and_cpu()
-        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"])
@@ -304,3 +298,25 @@
             "sys.net.tun0.bytes.sent": 5226635,
             "sys.net.tun0.packets.received": 24837,
             "sys.net.tun0.packets.sent": 26986})
+
+    def test_report_counters(self):
+        """
+        C{report_counters} keeps the last value of a called function and on the
+        next call returns the difference between current return value and
+        previous return value.
+        """
+        def generate():
+            yield {"foo": 1}
+            yield {"foo": 5}
+            yield {"foo": 10}
+            yield {"foo": 17}
+        generate = generate()
+        def reporter():
+            return generate.next()
+        wrapped = report_counters(reporter)
+        self.assertEqual({}, wrapped())
+        self.assertEqual({"foo": 4}, wrapped())
+        self.assertEqual({"foo": 5}, wrapped())
+        self.assertEqual({"foo": 7}, wrapped())
+        
+