Merge lp:~bac/charmworld/make-queue-a-job into lp:charmworld

[Brad Crittenden]

Run charm/bundle queue and ingest from one script

Originally, queueing was run from a cronjob every fifteen minutes. Stuff was
added without regard to it being queued up already.

Ingest was run continuously but slept fifteen minutes when it completed before
looking at the queue again.

This change makes the supervisord-run worker do the queue job, then ingest
charms and bundles. At the end, it sleeps for fifteen minutes minus the
amount spent doing the work.

The charmworld charm needs to be fixed to not create the cronjob for
queueing. It hasn't been done yet but the bin/queued script that is called by
the cronjob needs to be disabled before this change lands or the queueing will
be run by two different approaches.

https://codereview.appspot.com/45440043/

R=benji

Run charm/bundle queue and ingest from one script

Originally, queueing was run from a cronjob every fifteen minutes. Stuff was
added without regard to it being queued up already.

Ingest was run continuously but slept fifteen minutes when it completed before
looking at the queue again.

This change makes the supervisord-run worker do the queue job, then ingest
charms and bundles. At the end, it sleeps for fifteen minutes minus the
amount spent doing the work.

The charmworld charm needs to be fixed to not create the cronjob for
queueing. It hasn't been done yet but the bin/queued script that is called by
the cronjob needs to be disabled before this change lands or the queueing will
be run by two different approaches.

https://codereview.appspot.com/45440043/

[Brad Crittenden (bac)]

Reviewers: mp+200027_code.launchpad.net,

Message:
Please take a look.

Description:
Run charm/bundle queue and ingest from one script

Originally, queueing was run from a cronjob every fifteen minutes.
Stuff was
added without regard to it being queued up already.

Ingest was run continuously but slept fifteen minutes when it completed
before
looking at the queue again.

This change makes the supervisord-run worker do the queue job, then
ingest
charms and bundles. At the end, it sleeps for fifteen minutes minus the
amount spent doing the work.

The charmworld charm needs to be fixed to not create the cronjob for
queueing. It hasn't been done yet but the bin/queued script that is
called by
the cronjob needs to be disabled before this change lands or the
queueing will
be run by two different approaches.

https://code.launchpad.net/~bac/charmworld/make-queue-a-job/+merge/200027

(do not edit description out of merge proposal)

Please review this at https://codereview.appspot.com/45440043/

Affected files (+147, -43 lines):
   A [revision details]
   M charmworld/jobs/lp.py
   M charmworld/jobs/tests/test_lp.py
   M charmworld/jobs/tests/test_worker.py
   M charmworld/jobs/worker.py
   M charmworld/testing/__init__.py

[Brad Crittenden (bac)]

Ignore my comment about the cronjob. It seems bin/enqueue is handy to
keep around so I'll just manually disable the cronjob on staging before
this branch lands and then follow quickly with an updated charm.

https://codereview.appspot.com/45440043/

[lp:~bac/charmworld/make-queue-a-job]

469. By Brad Crittenden on 2013-12-24

Updated hacking.rst to remove bin/enqueue.

[Benji York (benji)]

LGTM -- there is one critical bug that should be easy to fix (and
probably deserves a test so something like that doesn't happen again).

https://codereview.appspot.com/45440043/diff/1/charmworld/jobs/lp.py
File charmworld/jobs/lp.py (right):

https://codereview.appspot.com/45440043/diff/1/charmworld/jobs/lp.py#newcode206
charmworld/jobs/lp.py:206: def main(arglist):
The bin/enqueue script is an entry point wrapper around this function
and doesn't pass any arguments, so if you run bin/enqueue you'll get

     TypeError: main() takes exactly 1 argument (0 given)

I would use a default argument of None and add this to the body of main:

if arglist is None:
     arglist = sys.argv[1:]

https://codereview.appspot.com/45440043/diff/1/charmworld/jobs/tests/test_lp.py
File charmworld/jobs/tests/test_lp.py (right):

https://codereview.appspot.com/45440043/diff/1/charmworld/jobs/tests/test_lp.py#newcode374
charmworld/jobs/tests/test_lp.py:374: namespace = Namespace()
It took a couple of readings for me to understand this test. Perhaps
naming "namespace" "expected" instead would have helped.

https://codereview.appspot.com/45440043/diff/1/charmworld/jobs/tests/test_lp.py#newcode375
charmworld/jobs/tests/test_lp.py:375: setattr(namespace, 'prefix',
'~sinzui')
I'm curious as to why you used setattr instead of "namespace.prefix =
'~sinzui'".

https://codereview.appspot.com/45440043/diff/1/charmworld/jobs/tests/test_lp.py#newcode380
charmworld/jobs/tests/test_lp.py:380:
This looks like an errant newline. Or maybe a space reserved for a
forgotten comment.

https://codereview.appspot.com/45440043/diff/1/charmworld/jobs/tests/test_worker.py
File charmworld/jobs/tests/test_worker.py (right):

https://codereview.appspot.com/45440043/diff/1/charmworld/jobs/tests/test_worker.py#newcode231
charmworld/jobs/tests/test_worker.py:231: def test_lp_run_limit(self):
I'm a big fan of the
first-line-of-a-test-is-a-comment-describing-what-the-test-is-about
pattern. Hint, hint. ;)

https://codereview.appspot.com/45440043/

[lp:~bac/charmworld/make-queue-a-job]

470. By Brad Crittenden on 2014-01-02

Fixes from review. Allow lp-main to be called with no args since that's what bin/enqueue does. Add tests. Make existing tests more readable and comment intent.

[Juju Gui Bot (juju-gui-bot)]

review: Approve (continuous-integration)

=== modified file 'charmworld/jobs/lp.py'
--- charmworld/jobs/lp.py	2013-10-23 21:53:17 +0000
+++ charmworld/jobs/lp.py	2014-01-02 19:14:28 +0000
@@ -8,6 +8,7 @@
 )
 import logging
 import pymongo
+import sys
 
 from charmworld.charmstore import (
     CharmStore,
@@ -169,10 +170,7 @@
         log.warn(str(error))
 
 
-# XXX bug=1130732: This function does not appear to be tested.
-def main():
-    configure_logging()
-    parser = argparse.ArgumentParser()
+def configure_parser(parser):
     parser.add_argument(
         "--limit",
         help="Maximum number of charms to queue.",
@@ -181,7 +179,9 @@
         "--prefix",
         help="Only branches matching this prefix are queued.",
         type=str, default=unspecified)
-    args = parser.parse_args()
+
+
+def run(args):
     log = logging.getLogger("charm.launchpad")
     settings = get_ini()
     if args.limit is None:
@@ -203,5 +203,15 @@
         log.warn(str(error))
 
 
+def main(arglist=None):
+    if arglist is None:
+        arglist = sys.argv[1:]
+    configure_logging()
+    parser = argparse.ArgumentParser()
+    configure_parser(parser)
+    args = parser.parse_args(arglist)
+    run(args)
+
+
 if __name__ == '__main__':
-    main()
+    main(sys.argv[1:])

=== modified file 'charmworld/jobs/tests/test_lp.py'
--- charmworld/jobs/tests/test_lp.py	2013-10-24 17:24:43 +0000
+++ charmworld/jobs/tests/test_lp.py	2014-01-02 19:14:28 +0000
@@ -1,6 +1,7 @@
 # Copyright 2012, 2013 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+from argparse import Namespace
 import json
 from requests.exceptions import ConnectionError
 
@@ -16,6 +17,8 @@
     JobTestBase,
     MONGO_DATABASE,
     MongoTestBase,
+    redirect_stderr,
+    substitute_argv,
 )
 
 from charmworld.jobs.lp import (
@@ -24,6 +27,7 @@
     CHARM_QUEUE,
     db_charms,
     dequeue,
+    main,
     queue_from_branches,
     unspecified,
 )
@@ -354,3 +358,46 @@
                    '~charmers/charms/precise'):
             self.queue_from_branches()
         self.assertIs(None, self.charm_queue.next())
+
+    def run_main(self, extra_args=None):
+        with patch('charmworld.jobs.lp.configure_logging') as mock_cl:
+            with patch('charmworld.jobs.lp.run') as mock_run:
+                fake_argv = None
+                if extra_args is not None:
+                   # Listify and add.
+                    fake_argv = extra_args.split()
+                if fake_argv is not None:
+                    main(fake_argv)
+                else:
+                    # We must be able to call main with no arguments.
+                    main()
+        self.assertTrue(mock_cl.called)
+        return mock_run
+
+    def test_main_args(self):
+        # Show arguments are passed and used by main.
+        mock = self.run_main('--limit 10 --prefix ~sinzui')
+        expected = Namespace()
+        expected.prefix = '~sinzui'
+        expected.limit = 10
+        mock.assert_called_with(expected)
+
+    def test_main_bad_args(self):
+        # Show bad arguments passed to main are caught.
+        with redirect_stderr() as stderr:
+            with self.assertRaises(SystemExit) as e:
+                self.run_main('--nolimit 10 --prefix ~sinzui')
+                self.assertEqual(2, e.exception.code)
+        self.assertIn(
+            'error: unrecognized arguments: --nolimit 10', stderr.read())
+
+    def test_main_no_args(self):
+        # Show main can be called with no arguments and in that case uses
+        # sys.argv.
+        test_args = ['ignore', '--prefix', '~benji', '--limit', '101']
+        with substitute_argv(test_args):
+            mock = self.run_main()
+        expected = Namespace()
+        expected.prefix = '~benji'
+        expected.limit = 101
+        mock.assert_called_with(expected)

=== modified file 'charmworld/jobs/tests/test_worker.py'
--- charmworld/jobs/tests/test_worker.py	2013-07-30 19:14:07 +0000
+++ charmworld/jobs/tests/test_worker.py	2014-01-02 19:14:28 +0000
@@ -59,6 +59,10 @@
         self.charms.append(charm_data)
 
 
+class FakeArgs(object):
+    run_forever = False
+
+
 @contextmanager
 def main_environment(charm, index_client, queue):
     """An environment in which worker.main() can be run."""
@@ -67,16 +71,18 @@
             yield
 
 
-def FauxWorkerFactory(test_case):
+def FauxWorkerFactory(test_case, expected_arg_hash):
     class FauxWorker:
         run_called = False
 
-        def __init__(self, *args):
+        def __init__(self, *ignored):
             pass
 
         @classmethod
-        def run(cls, run_forever):
-            test_case.assertFalse(run_forever)
+        def run(cls, args_):
+            for argname, expected in expected_arg_hash.items():
+                actual = args_.__getattribute__(argname)
+                test_case.assertEqual(expected, actual)
             cls.run_called = True
 
     return FauxWorker
@@ -90,6 +96,7 @@
         self.basket_queue = get_queue(self.db, BASKET_QUEUE)
         self.charm_queue.put({'charm': 'foo'})
         self.charm_queue.put({'charm': 'bar'})
+        self.args = FakeArgs()
 
     def tearDown(self):
         self.charm_queue.clear()
@@ -103,10 +110,11 @@
             GatheringRun(),
         )
         queue_jobs = {
-            self.charm_queue: ingest_jobs
+            self.charm_queue: ingest_jobs,
         }
-        worker = QueueWorker(self.db, queue_jobs)
-        worker.run()
+        worker = QueueWorker(self.db, None, queue_jobs)
+
+        worker.run(self.args)
 
         self.assertEqual(0, self.charm_queue.size())
         expected = [
@@ -119,10 +127,10 @@
     def test_failure_while_running(self):
         ingest_jobs = (FailureRun(), GatheringRun())
         queue_jobs = {
-            self.charm_queue: ingest_jobs
+            self.charm_queue: ingest_jobs,
         }
-        worker = QueueWorker(self.db, queue_jobs)
-        worker.run()
+        worker = QueueWorker(self.db, None, queue_jobs)
+        worker.run(self.args)
 
         self.assertEqual(0, self.charm_queue.size())
         expected = [
@@ -150,26 +158,30 @@
             SuccessfulRun(),
             GatheringRun(),
         )
-        worker = QueueWorker(self.db, {LockedJobQueue: ingest_jobs})
-        worker.run()
+        worker = QueueWorker(self.db, None, {LockedJobQueue: ingest_jobs})
+        worker.run(self.args)
 
     def test_ingest_job_contract(self):
         # All ingest jobs accept the right inputs.
         ingest_jobs = [create_autospec(UpdateCharmJob)()]
-        worker = QueueWorker(self.db, {self.charm_queue: ingest_jobs})
-        worker.run()
+        worker = QueueWorker(self.db, None, {self.charm_queue: ingest_jobs})
+        worker.run(self.args)
         self.assertEqual(0, self.charm_queue.size())
 
     def test_interval_is_int(self):
-        worker = QueueWorker(self.db, {self.charm_queue: []})
+        worker = QueueWorker(self.db, None, {self.charm_queue: []})
         self.assertIs(int, type(worker.interval),
                       'QueueWorker.interval is a %s, not an int.' %
                       type(worker.interval))
 
-    def run_main_with_exit(self):
+    def run_main_with_exit(self, extra_args=None):
         with patch('charmworld.jobs.worker.configure_logging'):
             with self.assertRaises(SystemExit) as e:
-                main(['ingest-queued'], db=self.db)
+                fake_argv = ['ingest-queued']
+                if extra_args is not None:
+                    # Listify and add.
+                    fake_argv.extend(extra_args.split())
+                main(fake_argv, db=self.db)
         return e.exception.code
 
     def test_elasticsearch_failure(self):
@@ -184,7 +196,8 @@
                         values={'es_urls': 'http://localhost:70'}):
             with main_environment(charm, index_client, self.charm_queue):
                 with logger_buffer('charm.worker') as handler:
-                    exit_code = self.run_main_with_exit()
+                    with patch('charmworld.jobs.worker.lp.run'):
+                        exit_code = self.run_main_with_exit()
         self.assertEqual(40, handler.buffer[1].levelno)
         self.assertIn('Could not connect', handler.buffer[1].msg)
         self.assertEqual(1, exit_code)
@@ -195,19 +208,38 @@
         queue = get_queue(self.db, BASKET_QUEUE)
         queue_entry = {'bundle': 'foo'}
         queue.put(queue_entry)
-        with patch('charmworld.jobs.worker.run_job') as run_job:
-            self.run_main_with_exit()
-            queue_entries_run = [x[0][1] for x in run_job.call_args_list]
-            self.assertIn(queue_entry, queue_entries_run)
+        with patch('charmworld.jobs.worker.lp.run'):
+            with patch('charmworld.jobs.worker.run_job') as run_job:
+                self.run_main_with_exit()
+                queue_entries_run = [x[0][1] for x in run_job.call_args_list]
+                self.assertIn(queue_entry, queue_entries_run)
 
     def test_worker_does_not_run_forever_by_default(self):
-        FauxWorker = FauxWorkerFactory(self)
+        expected = {'run_forever': False}
+        FauxWorker = FauxWorkerFactory(self, expected)
         with patch('charmworld.jobs.worker.QueueWorker', new=FauxWorker):
             self.run_main_with_exit()
         self.assertTrue(FauxWorker.run_called)
 
     def test_worker_runs_forever_if_command_line_says_to(self):
-        FauxWorker = FauxWorkerFactory(self)
-        with patch('charmworld.jobs.worker.QueueWorker', new=FauxWorker):
-            self.run_main_with_exit()
+        expected = {'run_forever': True}
+        FauxWorker = FauxWorkerFactory(self, expected)
+        with patch('charmworld.jobs.worker.QueueWorker', new=FauxWorker):
+            self.run_main_with_exit('--run-forever')
+        self.assertTrue(FauxWorker.run_called)
+
+    def test_lp_run_limit(self):
+        # Show that the '--limit' argument is passed and used.
+        expected = {'limit': 5}
+        FauxWorker = FauxWorkerFactory(self, expected)
+        with patch('charmworld.jobs.worker.QueueWorker', new=FauxWorker):
+            self.run_main_with_exit('--limit 5')
+        self.assertTrue(FauxWorker.run_called)
+
+    def test_lp_run_prefix(self):
+        # Show that the '--prefix' argument is passed and used.
+        expected = {'prefix': 'abentley'}
+        FauxWorker = FauxWorkerFactory(self, expected)
+        with patch('charmworld.jobs.worker.QueueWorker', new=FauxWorker):
+            self.run_main_with_exit('--prefix abentley')
         self.assertTrue(FauxWorker.run_called)

=== modified file 'charmworld/jobs/worker.py'
--- charmworld/jobs/worker.py	2013-07-22 15:49:43 +0000
+++ charmworld/jobs/worker.py	2014-01-02 19:14:28 +0000
@@ -3,6 +3,7 @@
 
 import argparse
 import logging
+import math
 import sys
 import time
 
@@ -15,6 +16,7 @@
     UpdateBundleJob,
     UpdateCharmJob,
 )
+from charmworld.jobs import lp
 from charmworld.jobs.utils import get_queue
 from charmworld.models import getconnection
 from charmworld.models import getdb
@@ -32,20 +34,29 @@
 class QueueWorker(object):
     """Processes raw charm data"""
 
-    def __init__(self, db, queue_jobs, interval=INTERVAL):
+    def __init__(self, db, queue_jobs, ingest_jobs, interval=INTERVAL):
         self.db = db
-        self.queue_jobs = queue_jobs
+        self.queue_jobs = queue_jobs or []
+        self.ingest_jobs = ingest_jobs
         self.log = logging.getLogger('charm.worker')
         self.interval = interval
 
-    def run(self, run_forever=False):
+    @staticmethod
+    def wait_seconds(interval, start, now):
+        remaining = interval - (now - start)
+        return math.max(0, remaining)
+
+    def run(self, args, now=None):
         self.log.info('Starting processing')
+        self.start_time = now if now is not None else time.time()
         return_code = 0
-        for jobs in self.queue_jobs.values():
+        for jobs in self.ingest_jobs.values():
             for job in jobs:
                 job.setup(db=self.db)
         while True:
-            for queue, jobs in self.queue_jobs.items():
+            for job in self.queue_jobs:
+                job(args)
+            for queue, jobs in self.ingest_jobs.items():
                 while True:
                     item = queue.next()
                     if item is None:
@@ -64,15 +75,17 @@
                     finally:
                         item.complete()
 
-            if not run_forever:
+            if not args.run_forever:
                 self.log.info('Empty queue and not running forever; quitting.')
                 break
 
-            wait_time_in_minutes = self.interval / 60
+            wait_seconds = self.wait_seconds(
+                self.interval, self.start_time, time.time())
+            wait_time_in_minutes = wait_seconds / 60
             self.log.info(
                 'Empty queue.  Waiting %s minutes to recheck.' %
                 wait_time_in_minutes)
-            time.sleep(self.interval)
+            time.sleep(wait_seconds)
         return return_code
 
 
@@ -88,11 +101,13 @@
         help='Keep running after the queues are processed and process more '
         'later.',
         action='store_true', dest='run_forever')
+    lp.configure_parser(parser)
     # Explicitly pass sys.argv to make monkeypatching more reliable.
     args = parser.parse_args(argv[1:])
-    queue_jobs = {
+    queue_jobs = (lp.run, )
+    ingest_jobs = {
         get_queue(db, CHARM_QUEUE): [UpdateCharmJob()],
         get_queue(db, BASKET_QUEUE): [UpdateBundleJob()],
     }
-    worker = QueueWorker(db, queue_jobs)
-    sys.exit(worker.run(run_forever=args.run_forever))
+    worker = QueueWorker(db, queue_jobs, ingest_jobs)
+    sys.exit(worker.run(args))

=== modified file 'charmworld/testing/__init__.py'
--- charmworld/testing/__init__.py	2013-10-14 19:36:57 +0000
+++ charmworld/testing/__init__.py	2014-01-02 19:14:28 +0000
@@ -6,6 +6,7 @@
 import os
 import shutil
 from StringIO import StringIO
+import sys
 from tempfile import mkdtemp
 import unittest
 from urllib2 import URLError
@@ -281,3 +282,31 @@
         filename,
     )
     return open(data_path).read()
+
+
+@contextmanager
+def redirect_stderr():
+    """Context manager for redirecting sys.stderr.
+
+    The new stream where sys.stderr is redirected is return.
+    Upon exiting the context manager the stream is set to the first byte
+    written so it can easily be read.
+    """
+    orig_stderr = sys.stderr
+    stream = StringIO()
+    sys.stderr = stream
+    try:
+        yield stream
+    finally:
+        stream.seek(0)
+        sys.stderr = orig_stderr
+
+
+@contextmanager
+def substitute_argv(args):
+    orig_argv = sys.argv
+    sys.argv = args
+    try:
+        yield
+    finally:
+        sys.argv = orig_argv

=== modified file 'docs/hacking.rst'
--- docs/hacking.rst	2013-10-30 18:35:32 +0000
+++ docs/hacking.rst	2014-01-02 19:14:28 +0000
@@ -95,8 +95,7 @@
     $ make sysdeps
     $ make install
     $ make run
-    $ bin/enqueue --prefix=~charmers/charms/precise --limit=100
-    $ bin/ingest-queued
+    $ bin/ingest-queued --prefix=~charmers/charms/precise --limit=100
 
 
 Contributing work