Merge ~cjwatson/launchpad:number-cruncher-librarian into launchpad:master

[Colin Watson]

number-cruncher: Add librarian metrics

From lp:tuolumne-lp-configs. This query is fairly heavyweight (it takes a couple of minutes to run on staging), so only run it once an hour.

[Tom Wardill (twom)]

review: Approve

diff --git a/lib/lp/services/statsd/numbercruncher.py b/lib/lp/services/statsd/numbercruncher.py
index 6a43c36..27c3a30 100644
--- a/lib/lp/services/statsd/numbercruncher.py
+++ b/lib/lp/services/statsd/numbercruncher.py
@@ -10,6 +10,10 @@ __all__ = ['NumberCruncher']
 
 import logging
 
+from storm.expr import (
+    Count,
+    Sum,
+    )
 import transaction
 from twisted.application import service
 from twisted.internet import (
@@ -26,6 +30,8 @@ from lp.buildmaster.enums import (
     )
 from lp.buildmaster.interfaces.builder import IBuilderSet
 from lp.buildmaster.manager import PrefetchedBuilderFactory
+from lp.services.database.interfaces import IStore
+from lp.services.librarian.model import LibraryFileContent
 from lp.services.statsd.interfaces.statsd_client import IStatsdClient
 
 NUMBER_CRUNCHER_LOG_NAME = "number-cruncher"
@@ -36,6 +42,7 @@ class NumberCruncher(service.Service):
 
     QUEUE_INTERVAL = 60
     BUILDER_INTERVAL = 60
+    LIBRARIAN_INTERVAL = 3600
 
     def __init__(self, clock=None, builder_factory=None):
         if clock is None:
@@ -68,6 +75,10 @@ class NumberCruncher(service.Service):
         stopping_deferred = loop.start(interval)
         return loop, stopping_deferred
 
+    def _sendGauge(self, gauge_name, value):
+        self.logger.debug("{}: {}".format(gauge_name, value))
+        self.statsd_client.gauge(gauge_name, value)
+
     def updateBuilderQueues(self):
         """Update statsd with the build queue lengths.
 
@@ -82,8 +93,7 @@ class NumberCruncher(service.Service):
                     gauge_name = (
                         "buildqueue,virtualized={},arch={},env={}".format(
                             virt, arch, self.statsd_client.lp_environment))
-                    self.logger.debug("{}: {}".format(gauge_name, value[0]))
-                    self.statsd_client.gauge(gauge_name, value[0])
+                    self._sendGauge(gauge_name, value[0])
             self.logger.debug("Build queue stats update complete.")
         except Exception:
             self.logger.exception("Failure while updating build queue stats:")
@@ -118,8 +128,7 @@ class NumberCruncher(service.Service):
             for count_name, count_value in counts.items():
                 gauge_name = "builders,status={},arch={},env={}".format(
                     count_name, processor, self.statsd_client.lp_environment)
-                self.logger.debug("{}: {}".format(gauge_name, count_value))
-                self.statsd_client.gauge(gauge_name, count_value)
+                self._sendGauge(gauge_name, count_value)
         self.logger.debug("Builder stats update complete.")
 
     def updateBuilderStats(self):
@@ -134,20 +143,43 @@ class NumberCruncher(service.Service):
             self.logger.exception("Failure while updating builder stats:")
         transaction.abort()
 
+    def updateLibrarianStats(self):
+        """Update librarian statistics.
+
+        This aborts the current transaction before returning.
+        """
+        try:
+            self.logger.debug("Updating librarian stats.")
+            store = IStore(LibraryFileContent)
+            total_files, total_filesize = store.find(
+                (Count(), Sum(LibraryFileContent.filesize))).one()
+            self._sendGauge(
+                "librarian.total_files,env={}".format(
+                    self.statsd_client.lp_environment),
+                total_files)
+            self._sendGauge(
+                "librarian.total_filesize,env={}".format(
+                    self.statsd_client.lp_environment),
+                total_filesize)
+            self.logger.debug("Librarian stats update complete.")
+        except Exception:
+            self.logger.exception("Failure while updating librarian stats:")
+        transaction.abort()
+
     def startService(self):
         self.logger.info("Starting number-cruncher service.")
-        self.update_queues_loop, self.update_queues_deferred = (
-            self._startLoop(self.QUEUE_INTERVAL, self.updateBuilderQueues))
-        self.update_builder_loop, self.update_builder_deferred = (
-            self._startLoop(self.BUILDER_INTERVAL, self.updateBuilderStats))
+        self.loops = []
+        self.stopping_deferreds = []
+        for interval, callback in (
+                (self.QUEUE_INTERVAL, self.updateBuilderQueues),
+                (self.BUILDER_INTERVAL, self.updateBuilderStats),
+                (self.LIBRARIAN_INTERVAL, self.updateLibrarianStats),
+                ):
+            loop, stopping_deferred = self._startLoop(interval, callback)
+            self.loops.append(loop)
+            self.stopping_deferreds.append(stopping_deferred)
 
     def stopService(self):
-        deferreds = []
-        deferreds.append(self.update_queues_deferred)
-        deferreds.append(self.update_builder_deferred)
-
-        self.update_queues_loop.stop()
-        self.update_builder_loop.stop()
-
-        d = defer.DeferredList(deferreds, consumeErrors=True)
-        return d
+        for loop in self.loops:
+            loop.stop()
+        return defer.DeferredList(self.stopping_deferreds, consumeErrors=True)
diff --git a/lib/lp/services/statsd/tests/test_numbercruncher.py b/lib/lp/services/statsd/tests/test_numbercruncher.py
index 020ef76..0dff3a0 100644
--- a/lib/lp/services/statsd/tests/test_numbercruncher.py
+++ b/lib/lp/services/statsd/tests/test_numbercruncher.py
@@ -10,6 +10,7 @@ __metaclass__ = type
 from testtools.matchers import (
     Equals,
     MatchesListwise,
+    Not,
     )
 from testtools.twistedsupport import AsynchronousDeferredRunTest
 import transaction
@@ -153,6 +154,57 @@ class TestNumberCruncher(StatsMixin, TestCaseWithFactory):
             "Failure while updating build queue stats:",
             cruncher.logger.getLogBuffer())
 
+    def test_updateLibrarianStats(self):
+        clock = task.Clock()
+        cruncher = NumberCruncher(clock=clock)
+        cruncher.updateLibrarianStats()
+
+        self.assertFalse(is_transaction_in_progress())
+        self.assertEqual(2, self.stats_client.gauge.call_count)
+        self.assertThat(
+            [x[0] for x in self.stats_client.gauge.call_args_list],
+            MatchesListwise([
+                MatchesListwise([
+                    Equals('librarian.total_files,env=test'),
+                    Not(Equals(0)),
+                    ]),
+                MatchesListwise([
+                    Equals('librarian.total_filesize,env=test'),
+                    Not(Equals(0)),
+                    ]),
+                ]))
+        total_files = self.stats_client.gauge.call_args_list[0][0][1]
+        total_filesize = self.stats_client.gauge.call_args_list[1][0][1]
+
+        self.factory.makeLibraryFileAlias(content=b'x' * 1000, db_only=True)
+        self.factory.makeLibraryFileAlias(content=b'x' * 2000, db_only=True)
+        self.stats_client.gauge.reset_mock()
+        cruncher.updateLibrarianStats()
+
+        self.assertFalse(is_transaction_in_progress())
+        self.assertEqual(2, self.stats_client.gauge.call_count)
+        self.assertThat(
+            [x[0] for x in self.stats_client.gauge.call_args_list],
+            MatchesListwise([
+                Equals(('librarian.total_files,env=test', total_files + 2)),
+                Equals((
+                    'librarian.total_filesize,env=test',
+                    total_filesize + 3000,
+                    )),
+                ]))
+
+    def test_updateLibrarianStats_error(self):
+        clock = task.Clock()
+        cruncher = NumberCruncher(clock=clock)
+        cruncher.logger = BufferLogger()
+        with DatabaseBlockedPolicy():
+            cruncher.updateLibrarianStats()
+
+        self.assertFalse(is_transaction_in_progress())
+        self.assertIn(
+            "Failure while updating librarian stats:",
+            cruncher.logger.getLogBuffer())
+
     def test_startService_starts_update_queues_loop(self):
         clock = task.Clock()
         cruncher = NumberCruncher(clock=clock)
@@ -174,3 +226,14 @@ class TestNumberCruncher(StatsMixin, TestCaseWithFactory):
         advance = NumberCruncher.BUILDER_INTERVAL + 1
         clock.advance(advance)
         self.assertNotEqual(0, cruncher.updateBuilderStats.call_count)
+
+    def test_startService_starts_update_librarian_loop(self):
+        clock = task.Clock()
+        cruncher = NumberCruncher(clock=clock)
+
+        cruncher.updateLibrarianStats = FakeMethod()
+
+        cruncher.startService()
+        advance = NumberCruncher.LIBRARIAN_INTERVAL + 1
+        clock.advance(advance)
+        self.assertNotEqual(0, cruncher.updateLibrarianStats.call_count)