launchpad-reviewers team mailing list archive

[Tom Wardill <tom.wardill@xxxxxxxxxxxxx>]

Tom Wardill has proposed merging ~twom/launchpad:stats-code-import-stats into launchpad:master.

Commit message:
Add CodeImportJob stats to numbercruncher

Requested reviews:
  Launchpad code reviewers (launchpad-reviewers)

For more details, see:
https://code.launchpad.net/~twom/launchpad/+git/launchpad/+merge/397427
-- 
Your team Launchpad code reviewers is requested to review the proposed merge of ~twom/launchpad:stats-code-import-stats into launchpad:master.

diff --git a/lib/lp/services/statsd/numbercruncher.py b/lib/lp/services/statsd/numbercruncher.py
index 3f1a6e1..199dab0 100644
--- a/lib/lp/services/statsd/numbercruncher.py
+++ b/lib/lp/services/statsd/numbercruncher.py
@@ -4,12 +4,15 @@
 """Out of process statsd reporting."""
 
 from __future__ import absolute_import, print_function, unicode_literals
+from datetime import datetime
+
 
 __metaclass__ = type
 __all__ = ['NumberCruncher']
 
 import logging
 
+import pytz
 from storm.expr import (
     Count,
     Sum,
@@ -27,6 +30,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 +45,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 +161,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 +190,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..8b45548 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,58 @@ 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'),
+                    Not(Equals(0)),
+                    ]),
+                MatchesListwise([
+                    Equals('codeimport.overdue,env=test'),
+                    Not(Equals(0)),
+                    ]),
+                ]))
+
+        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'),
+                    Not(Equals(0)),
+                    ]),
+                ]))
+
+
+    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 +318,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.LIBRARIAN_INTERVAL + 1
+        clock.advance(advance)
+        self.assertNotEqual(0, cruncher.updateCodeImportStats.call_count)