Merge ~twom/launchpad:stats-code-import-stats into launchpad:master

[Tom Wardill]

Add CodeImportJob stats to numbercruncher

[Colin Watson (cjwatson)]

review: Approve

diff --git a/lib/lp/services/statsd/numbercruncher.py b/lib/lp/services/statsd/numbercruncher.py
index 3f1a6e1..b143927 100644
--- a/lib/lp/services/statsd/numbercruncher.py
+++ b/lib/lp/services/statsd/numbercruncher.py
@@ -8,8 +8,10 @@ from __future__ import absolute_import, print_function, unicode_literals
 __metaclass__ = type
 __all__ = ['NumberCruncher']
 
+from datetime import datetime
 import logging
 
+import pytz
 from storm.expr import (
     Count,
     Sum,
@@ -27,6 +29,8 @@ from zope.component import getUtility
 from lp.buildmaster.enums import BuilderCleanStatus
 from lp.buildmaster.interfaces.builder import IBuilderSet
 from lp.buildmaster.manager import PrefetchedBuilderFactory
+from lp.code.enums import CodeImportJobState
+from lp.code.model.codeimportjob import CodeImportJob
 from lp.services.database.interfaces import IStore
 from lp.services.librarian.model import LibraryFileContent
 from lp.services.statsd.interfaces.statsd_client import IStatsdClient
@@ -40,6 +44,7 @@ class NumberCruncher(service.Service):
     QUEUE_INTERVAL = 60
     BUILDER_INTERVAL = 60
     LIBRARIAN_INTERVAL = 3600
+    CODE_IMPORT_INTERVAL = 60
 
     def __init__(self, clock=None, builder_factory=None):
         if clock is None:
@@ -155,6 +160,27 @@ class NumberCruncher(service.Service):
             self.logger.exception("Failure while updating librarian stats:")
         transaction.abort()
 
+    def updateCodeImportStats(self):
+        """Update stats about code imports.
+
+        This aborts the current transaction before returning.
+        """
+        try:
+            self.logger.debug("Update code import stats.")
+            store = IStore(CodeImportJob)
+            pending = store.find(
+                CodeImportJob,
+                CodeImportJob.state == CodeImportJobState.PENDING).count()
+            overdue = store.find(
+                CodeImportJob,
+                CodeImportJob.date_due < datetime.now(pytz.UTC)).count()
+            self._sendGauge("codeimport.pending", pending)
+            self._sendGauge("codeimport.overdue", overdue)
+            self.logger.debug("Code import stats update complete")
+        except Exception:
+            self.logger.exception("Failure while updating code import stats.")
+        transaction.abort()
+
     def startService(self):
         self.logger.info("Starting number-cruncher service.")
         self.loops = []
@@ -163,6 +189,7 @@ class NumberCruncher(service.Service):
                 (self.QUEUE_INTERVAL, self.updateBuilderQueues),
                 (self.BUILDER_INTERVAL, self.updateBuilderStats),
                 (self.LIBRARIAN_INTERVAL, self.updateLibrarianStats),
+                (self.CODE_IMPORT_INTERVAL, self.updateCodeImportStats),
                 ):
             loop, stopping_deferred = self._startLoop(interval, callback)
             self.loops.append(loop)
diff --git a/lib/lp/services/statsd/tests/test_numbercruncher.py b/lib/lp/services/statsd/tests/test_numbercruncher.py
index 41b6bd4..f2ea44e 100644
--- a/lib/lp/services/statsd/tests/test_numbercruncher.py
+++ b/lib/lp/services/statsd/tests/test_numbercruncher.py
@@ -17,12 +17,14 @@ from testtools.twistedsupport import AsynchronousDeferredRunTest
 import transaction
 from twisted.internet import task
 from zope.component import getUtility
+from zope.security.proxy import removeSecurityProxy
 
 from lp.buildmaster.enums import BuilderCleanStatus
 from lp.buildmaster.interactor import BuilderSlave
 from lp.buildmaster.interfaces.processor import IProcessorSet
 from lp.buildmaster.model.buildqueue import BuildQueue
 from lp.buildmaster.tests.mock_slaves import OkSlave
+from lp.code.enums import CodeImportJobState
 from lp.services.database.isolation import is_transaction_in_progress
 from lp.services.database.policy import DatabaseBlockedPolicy
 from lp.services.log.logger import BufferLogger
@@ -232,6 +234,57 @@ class TestNumberCruncher(StatsMixin, TestCaseWithFactory):
             "Failure while updating librarian stats:",
             cruncher.logger.getLogBuffer())
 
+    def test_updateCodeImportStats(self):
+        clock = task.Clock()
+        cruncher = NumberCruncher(clock=clock)
+        cruncher.updateCodeImportStats()
+
+        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('codeimport.pending,env=test'),
+                    Equals(1),
+                    ]),
+                MatchesListwise([
+                    Equals('codeimport.overdue,env=test'),
+                    Equals(1),
+                    ]),
+                ]))
+
+        job = removeSecurityProxy(self.factory.makeCodeImportJob())
+        job.state = CodeImportJobState.PENDING
+        self.stats_client.gauge.reset_mock()
+        cruncher.updateCodeImportStats()
+
+        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('codeimport.pending,env=test'),
+                    Equals(2),
+                    ]),
+                MatchesListwise([
+                    Equals('codeimport.overdue,env=test'),
+                    Equals(2),
+                    ]),
+                ]))
+
+    def test_updateCodeImportStats_error(self):
+        clock = task.Clock()
+        cruncher = NumberCruncher(clock=clock)
+        cruncher.logger = BufferLogger()
+        with DatabaseBlockedPolicy():
+            cruncher.updateCodeImportStats()
+
+        self.assertFalse(is_transaction_in_progress())
+        self.assertIn(
+            "Failure while updating code import stats.",
+            cruncher.logger.getLogBuffer())
+
     def test_startService_starts_update_queues_loop(self):
         clock = task.Clock()
         cruncher = NumberCruncher(clock=clock)
@@ -264,3 +317,14 @@ class TestNumberCruncher(StatsMixin, TestCaseWithFactory):
         advance = NumberCruncher.LIBRARIAN_INTERVAL + 1
         clock.advance(advance)
         self.assertNotEqual(0, cruncher.updateLibrarianStats.call_count)
+
+    def test_startService_starts_update_code_import_loop(self):
+        clock = task.Clock()
+        cruncher = NumberCruncher(clock=clock)
+
+        cruncher.updateCodeImportStats = FakeMethod()
+
+        cruncher.startService()
+        advance = NumberCruncher.CODE_IMPORT_INTERVAL + 1
+        clock.advance(advance)
+        self.assertNotEqual(0, cruncher.updateCodeImportStats.call_count)