Merge lp:~adeuring/charmworld/queue-review-mutex into lp:~juju-jitsu/charmworld/trunk

[Abel Deuring]

Use locks to prevent the jobs lp, core_review and review from running concurrently.

Locking for jobs

Charmworld runs a few cron jobs to keep the list of avaliable charms
up to date. If charmworld is installed on more than one machines, running
these jobs on all machines would cause unnecessary duplicate work,
like queueing the charms for further processing.

This branch adds a simple locking mechansim for these jobs. The core idea:
A lock is a record in the collection "locks" of the MongoDB with a given
ID. MongoDB can raise a DuplicateKey exception when a record with the ID
'some_name' already exists during a call of
collection.insert({'_id': 'some_name', ...})

("can" means: When I first attempted to use the lock for the jobs review and
core_review, locking simply did not work, while it worked fine in tests and
for the queueing job. It turned out that the parameter "fsync" msut be
specified when pymongo.Connection is instantiated. As the test
test_bad_connection_setup shows, the actual value of this parameter is
not important -- but it must be present...)

Hence lock() ensures that fsync was specified for the given DB connection.

Locks have a lease time; if the lease expires, a now stale lock is deleted
when another process wants to acquire the lock. The related call of
locks.remove(existing_lock) is a possible race condition: Two processes
might try concurrently to remove the stale lock. A simple tests shows that
duplicate remove(existing_lock) calls succeed, even if the second call
just does nothing.

https://codereview.appspot.com/7379053/

[Abel Deuring (adeuring)]

Reviewers: mp+150853_code.launchpad.net,

Message:
Please take a look.

Description:
Locking for jobs

Charmworld runs a few cron jobs to keep the list of avaliable charms
up to date. If charmworld is installed on more than one machines,
running
these jobs on all machines would cause unnecessary duplicate work,
like queueing the charms for further processing.

This branch adds a simple locking mechansim for these jobs. The core
idea:
A lock is a record in the collection "locks" of the MongoDB with a given
ID. MongoDB can raise a DuplicateKey exception when a record with the ID
'some_name' already exists during a call of
collection.insert({'_id': 'some_name', ...})

("can" means: When I first attempted to use the lock for the jobs review
and
core_review, locking simply did not work, while it worked fine in tests
and
for the queueing job. It turned out that the parameter "fsync" msut be
specified when pymongo.Connection is instantiated. As the test
test_bad_connection_setup shows, the actual value of this parameter is
not important -- but it must be present...)

Hence lock() ensures that fsync was specified for the given DB
connection.

Locks have a lease time; if the lease expires, a now stale lock is
deleted
when another process wants to acquire the lock. The related call of
locks.remove(existing_lock) is a possible race condition: Two processes
might try concurrently to remove the stale lock. A simple tests shows
that
duplicate remove(existing_lock) calls succeed, even if the second call
just does nothing.

https://code.launchpad.net/~adeuring/charmworld/queue-review-mutex/+merge/150853

(do not edit description out of merge proposal)

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

Affected files:
   A [revision details]
   M charmworld/jobs/core_review.py
   M charmworld/jobs/lp.py
   M charmworld/jobs/review.py
   A charmworld/jobs/tests/test_utils.py
   M charmworld/jobs/utils.py
   M charmworld/testing/__init__.py
   M default.ini

[Richard Harding (rharding)]

lgtm with a check on using safe vs fsync. If safe works then it should
be a bit more performance than an fsync.

I also would prefer to see something that catches locks that were not
released. I know they timeout, but we should have some notice that they
were hit so we can look into the issue. I'm worried about a charm
instance dying without us realizing there's an issue. It might take some
coordination with the charm though to get an error/notice out and up to
nagios or email or something.

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

https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/core_review.py#newcode107
charmworld/jobs/core_review.py:107: c = pymongo.Connection(MONGO_URL,
fsync=True)
can we get away with just safe=True vs fsync? It's a bit lighter than
fsync and we changed to this in test runs to help fix issues from mongo
syncing.

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

https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/lp.py#newcode73
charmworld/jobs/lp.py:73: connection = getconnection(settings)
does this need to have the safe= flag set to get a safe connection?

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

https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/review.py#newcode117
charmworld/jobs/review.py:117: c = pymongo.Connection(MONGO_URL,
fsync=True)
again, can safe= work?

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

https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/tests/test_utils.py#newcode38
charmworld/jobs/tests/test_utils.py:38: self.assertIs(None,
self.db['locks'].find_one({'_id': LOCK_NAME}))
should this be wrapped into a helper? utils.check_lock(db, LOCK_NAME) or
something?

https://codereview.appspot.com/7379053/diff/1/default.ini
File default.ini (right):

https://codereview.appspot.com/7379053/diff/1/default.ini#newcode42
default.ini:42: script_lease_time = 30
we need the bug to note that the charm should be updated to create this
time based on the config time used to cron the importer.

https://codereview.appspot.com/7379053/

[Abel Deuring (adeuring)]

On 27.02.2013 18:30, <email address hidden> wrote:
> lgtm with a check on using safe vs fsync. If safe works then it should
> be a bit more performance than an fsync.

Right, using safe=True works also. Checking if this parameter has been
specififed, is a bit more tricky, so I changed the function lock() to
just try to insert a test record twice.

>
> I also would prefer to see something that catches locks that were not
> released. I know they timeout, but we should have some notice that they
> were hit so we can look into the issue. I'm worried about a charm
> instance dying without us realizing there's an issue. It might take some
> coordination with the charm though to get an error/notice out and up to
> nagios or email or something.

If a job "just dies" without releasing the job, we have a problem that
is likely much more serious that just a stale lock:

    try:
        yield
    finally:
        locks.remove(my_lock)

ensures that the lock is released for "regular Python errors". If the
job dies due to a segfault, this segfault would be my main concern, not
the "orphaned" lock data.

>
>
> https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/core_review.py
>
> File charmworld/jobs/core_review.py (right):
>
> https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/core_review.py#newcode107
>
> charmworld/jobs/core_review.py:107: c = pymongo.Connection(MONGO_URL,
> fsync=True)
> can we get away with just safe=True vs fsync? It's a bit lighter than
> fsync and we changed to this in test runs to help fix issues from mongo
> syncing.

Done. Remember though that we can use fsync=False -- the odd thing is
that specifying fsync=True as well as fsync=False ensure that inserting
a doc with an existing key raises a DuplicateKeyError, while _not_
specifying it suppresses the DuplicateKeyError.

>
> https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/lp.py
> File charmworld/jobs/lp.py (right):
>
> https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/lp.py#newcode73
>
> charmworld/jobs/lp.py:73: connection = getconnection(settings)
> does this need to have the safe= flag set to get a safe connection?

No, the trick here are the default parameters of getconnection():

def getconnection(settings, fsync=False, safe=False):
    [...]
    connection = pymongo.Connection(
        url_or_host, port,
        fsync=fsync,
        safe=safe,
    )

So, fsync _is_ specified and everyting is fine.

>
> https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/review.py
> File charmworld/jobs/review.py (right):
>
> https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/review.py#newcode117
>
> charmworld/jobs/review.py:117: c = pymongo.Connection(MONGO_URL,
> fsync=True)
> again, can safe= work?

changed.

>
> https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/tests/test_utils.py
>
> File charmworld/jobs/tests/test_utils.py (right):
>
> https://codereview.appspot.com/7379053/diff/1/charmworld/jobs/tests/test_utils.py#newcode38
>
> charmworld/jobs/tests/test_utils.py:38: self.assertIs(None,
> self.db['locks'].find_one({'_id': LOCK_NAME}))
> should this be wrapped into a helper? utils.check_l...

[lp:~adeuring/charmworld/queue-review-mutex]

159. By Abel Deuring on 2013-02-28

implemented reviewr comments

[Abel Deuring (adeuring)]

review: Approve

=== modified file 'charmworld/jobs/core_review.py'
--- charmworld/jobs/core_review.py	2013-02-12 16:15:07 +0000
+++ charmworld/jobs/core_review.py	2013-02-28 13:04:21 +0000
@@ -21,7 +21,10 @@
 import logging
 import time
 
+from charmworld.utils import get_ini
 from config import MONGO_URL
+from utils import lock
+from utils import LockHeld
 
 log = logging.getLogger("core.review")
 
@@ -101,9 +104,15 @@
     logging.basicConfig(
         level=logging.DEBUG,
         format="%(asctime)s: %(name)s@%(levelname)s: %(message)s")
-    c = pymongo.Connection(MONGO_URL)
+    c = pymongo.Connection(MONGO_URL, safe=True)
     db = c['juju']
-    update_review_queue(db)
+    settings = get_ini()
+    try:
+        with lock('ingest-core-review',
+                  int(settings['script_lease_time']) * 60, db, log):
+            update_review_queue(db)
+    except LockHeld, error:
+        log.warn(str(error))
 
 if __name__ == '__main__':
     main()

=== modified file 'charmworld/jobs/lp.py'
--- charmworld/jobs/lp.py	2013-02-19 12:40:36 +0000
+++ charmworld/jobs/lp.py	2013-02-28 13:04:21 +0000
@@ -6,9 +6,14 @@
 import urllib
 import json
 
+from charmworld.models import getconnection
+from charmworld.models import getdb
+from charmworld.utils import get_ini
 from config import CHARM_IMPORT_FILTER
 from config import CHARM_IMPORT_LIMIT
 from config import CHARM_QUEUE
+from utils import lock
+from utils import LockHeld
 from utils import get_queue
 from utils import parse_branch
 
@@ -63,5 +68,14 @@
     logging.basicConfig(
         level=logging.INFO,
         format="%(asctime)s: %(name)s@%(levelname)s: %(message)s")
-    queue = get_queue(CHARM_QUEUE)
-    queue_charms(queue)
+    log = logging.getLogger("charm.launchpad")
+    settings = get_ini()
+    connection = getconnection(settings)
+    db = getdb(connection, settings.get('mongo.database'))
+    try:
+        with lock('ingest-queue', int(settings['script_lease_time']) * 60,
+                  db, log):
+            queue = get_queue(CHARM_QUEUE)
+            queue_charms(queue)
+    except LockHeld, error:
+        log.warn(str(error))

=== modified file 'charmworld/jobs/review.py'
--- charmworld/jobs/review.py	2013-02-12 16:15:07 +0000
+++ charmworld/jobs/review.py	2013-02-28 13:04:21 +0000
@@ -22,7 +22,10 @@
 import logging
 import time
 
+from charmworld.utils import get_ini
 from config import MONGO_URL
+from utils import lock
+from utils import LockHeld
 
 log = logging.getLogger("charm.review")
 
@@ -111,9 +114,15 @@
     logging.basicConfig(
         level=logging.DEBUG,
         format="%(asctime)s: %(name)s@%(levelname)s: %(message)s")
-    c = pymongo.Connection(MONGO_URL)
+    c = pymongo.Connection(MONGO_URL, safe=True)
     db = c['juju']
-    update_review_queue(db)
+    settings = get_ini()
+    try:
+        with lock('ingest-review',
+                  int(settings['script_lease_time']) * 60, db, log):
+            update_review_queue(db)
+    except LockHeld, error:
+        log.warn(str(error))
 
 if __name__ == '__main__':
     main()

=== added file 'charmworld/jobs/tests/test_utils.py'
--- charmworld/jobs/tests/test_utils.py	1970-01-01 00:00:00 +0000
+++ charmworld/jobs/tests/test_utils.py	2013-02-28 13:04:21 +0000
@@ -0,0 +1,152 @@
+# Copyright 2013 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+import logging
+from pymongo import Connection
+from pymongo.errors import DuplicateKeyError
+from time import time
+
+from charmworld.jobs.utils import BadDBConnection
+from charmworld.jobs.utils import lock
+from charmworld.jobs.utils import LockHeld
+from charmworld.testing import MONGO_DATABASE
+from charmworld.testing import MONGO_URL
+from charmworld.testing import MongoTestBase
+
+
+LOCK_NAME = u'test_lock'
+
+
+class LockTest(MongoTestBase):
+
+    def setUp(self):
+        super(LockTest, self).setUp()
+        self.locked_method_called = False
+        self.log = logging.getLogger('locktest')
+        self.log.setLevel(logging.WARN)
+        self.handler = self.get_handler('locktest')
+
+    def test_regular_locking(self):
+        # If no lock with the given name is held, a lock() call
+        # is successful.
+        self.assertIs(None, self.db['locks'].find_one({'_id': LOCK_NAME}))
+        with lock(LOCK_NAME, 1, self.db, self.log):
+            # Now a lock exists.
+            self.assertIsNot(
+                None, self.db['locks'].find_one({'_id': LOCK_NAME}))
+        # The lock has been removed.
+        self.assertIs(None, self.db['locks'].find_one({'_id': LOCK_NAME}))
+
+    def use_lock(self, exc=None, db=None):
+        # Acquire a lock; optionally raise an exception.
+        if db is None:
+            db = self.db
+        with lock(LOCK_NAME, 1, db, self.log):
+            self.locked_method_called = True
+            if exc is not None:
+                raise exc
+
+    def make_lock(self, lease_time, pid=1):
+        return {
+            '_id': LOCK_NAME,
+            'lease_until': int(time() + lease_time),
+            'host': 'another-host',
+            'pid': pid,
+        }
+
+    def test_lock_already_held(self):
+        # If another process has already created a lock, another call
+        # to lock() raises a LockHeld exception.
+        existing_lock = self.make_lock(5)
+        self.db['locks'].insert(existing_lock)
+        self.assertRaises(LockHeld, self.use_lock)
+
+    def test_stale_lock(self):
+        # Stale locks are removed.
+        existing_lock = self.make_lock(-1)
+        self.db['locks'].insert(existing_lock)
+        self.use_lock()
+        warning = self.handler.buffer[0]
+        self.assertEqual(
+            'Stale lock removed. Host: %(host)s, PID: %(pid)s, '
+            'lease_until: %(lease_until)s' % existing_lock, warning.msg)
+
+    def test_lock_released_when_locked_function_fails(self):
+        # The lock is also released when the locked function raises an
+        # exception.
+        error_occurred = False
+        try:
+            self.use_lock(exc=ValueError)
+        except ValueError:
+            error_occurred = True
+        self.assertTrue(error_occurred)
+        self.assertIs(None, self.db['locks'].find_one({'_id': LOCK_NAME}))
+
+    def test_duplicate_deletion_of_stale_lock(self):
+        # MongoDB's Collection.remove() call does not fail when called
+        # for a non-existing document. This means that two processes
+        # can attempt to delete a stale lock without failing.
+        locks = self.db['locks']
+        lock = self.make_lock(-1)
+        locks.insert(lock)
+        locks.remove(lock)
+        self.assertIs(None, locks.find_one(lock))
+        # The second call to remove the lock does not raise an exception.
+        locks.remove(lock)
+
+    def test_lock_removal_succeeds_only_for_known_lock(self):
+        # A possible race condition: Two processes detect a stale lock.
+        # Process A removes the lock and creates a new lock...
+        stale_lock = self.make_lock(-2, pid=1)
+        lock_process_a = self.make_lock(2, pid=2)
+        locks = self.db.locks
+        locks.insert(lock_process_a)
+        # ...before process B tries to remove the stale lock again.
+        locks.remove(stale_lock)
+        # The new lock still exists.
+        self.assertEqual(lock_process_a, locks.find_one({'_id': LOCK_NAME}))
+
+    def duplicate_insert(self, conn, expect_duplicate_key_error):
+        # Try to insert the same document twice into the Mongo DB.
+        db = conn[MONGO_DATABASE]
+        locks = db['locks']
+        lock_1 = self.make_lock(1, pid=42)
+        locks.insert(lock_1)
+        stored_lock = locks.find_one({'_id': LOCK_NAME})
+        # A second insert call may succeed..
+        lock_2 = self.make_lock(1, pid=43)
+        if expect_duplicate_key_error:
+            self.assertRaises(DuplicateKeyError, locks.insert, lock_2)
+        else:
+            locks.insert(lock_2)
+            # ...but has no effect.
+            stored_lock = locks.find_one({'_id': LOCK_NAME})
+            self.assertEqual(lock_1, stored_lock)
+        locks.remove({'_id': LOCK_NAME})
+
+    def test_bad_connection_setup(self):
+        # If the DB connection is created without the parameter fsync,
+        # no DuplicateKeyError is raised.
+        conn = Connection(MONGO_URL)
+        self.duplicate_insert(conn, expect_duplicate_key_error=False)
+
+        # A DuplicateKeyError is raised if the parameter fsync is specified.
+        conn = Connection(MONGO_URL, fsync=True)
+        self.duplicate_insert(conn, expect_duplicate_key_error=True)
+        # Note that the value of fsync does not matter...
+        conn = Connection(MONGO_URL, fsync=False)
+        self.duplicate_insert(conn, expect_duplicate_key_error=True)
+
+        # Alternatively, using safe=True ensures also that a
+        # DuplicateKeyError is raised.
+        conn = Connection(MONGO_URL, safe=False)
+        self.duplicate_insert(conn, expect_duplicate_key_error=False)
+        conn = Connection(MONGO_URL, safe=True)
+        self.duplicate_insert(conn, expect_duplicate_key_error=True)
+
+    def test_lock_creation_without_fsync_or_safe_param(self):
+        # A DB connection that was created without the fsync parameter
+        # or safe=True cannot be used to create a lock.
+        conn = Connection(MONGO_URL)
+        db = conn['locks']
+        self.assertRaises(BadDBConnection, self.use_lock, db=db)

=== modified file 'charmworld/jobs/utils.py'
--- charmworld/jobs/utils.py	2013-02-19 12:40:36 +0000
+++ charmworld/jobs/utils.py	2013-02-28 13:04:21 +0000
@@ -2,8 +2,13 @@
 # licensed under the GNU Affero General Public License version 3 (see
 # the file LICENSE).
 
+from contextlib import contextmanager
 from mongoqueue import MongoQueue
 from nose.tools import nottest
+from pymongo.errors import DuplicateKeyError
+from os import getpid
+from socket import gethostname
+from time import time
 
 from charmworld.models import getconnection
 from charmworld.models import getdb
@@ -64,6 +69,67 @@
 BRANCH_TIPS = "https://api.launchpad.net/devel/charm?ws.op=getBranchTips"
 
 
+class LockHeld(Exception):
+    pass
+
+
+class BadDBConnection(Exception):
+    pass
+
+
+@contextmanager
+def lock(name, lease_time, db, log):
+    """A simple MongoDB based mutex lock.
+
+    :param name: The name of the mutex.
+    :param lease_until: The time until which the lease is held.
+    """
+    locks = db['locks']
+    # check that a DuplicateKeyError is raised when a doc with an existing
+    # key is inserted.
+    raises_key_error = False
+    test_doc = {'_id': '_test'}
+    try:
+        locks.insert(test_doc)
+        locks.insert(test_doc)
+    except DuplicateKeyError:
+        raises_key_error = True
+    if not raises_key_error:
+        raise BadDBConnection(
+            'The MongoDB connection is not configured to raise '
+            'DuplicateKeyErrors. Use the parameter safe=True or '
+            'fsync (with any value) when creating the conenction.'
+            'See test_bad_connection_setup() in '
+            'charmworld/jobs/tests/test_utils.py for more details.')
+
+    existing_lock = locks.find_one({'_id': name})
+    now = time()
+    if existing_lock is not None and existing_lock['lease_until'] < now:
+        # Note that two calls that remove the same record do not fail, hence
+        # there is no need to check the result call.
+        locks.remove(existing_lock)
+        log.warn(
+            'Stale lock removed. Host: %(host)s, PID: %(pid)s, '
+            'lease_until: %(lease_until)s' % existing_lock)
+    my_lock = {
+        '_id': name,
+        'lease_until': now + lease_time,
+        'host': gethostname(),
+        'pid': getpid(),
+    }
+    try:
+        locks.insert(my_lock)
+    except DuplicateKeyError:
+        existing_lock = locks.find_one({'_id': name})
+        raise LockHeld(
+            'Lock %s already held until %s. Current time: %s' % (
+                name, existing_lock['lease_until'], now))
+    try:
+        yield
+    finally:
+        locks.remove(my_lock)
+
+
 @nottest
 def test():
     import urllib

=== modified file 'charmworld/testing/__init__.py'
--- charmworld/testing/__init__.py	2013-02-21 16:30:35 +0000
+++ charmworld/testing/__init__.py	2013-02-28 13:04:21 +0000
@@ -46,6 +46,16 @@
         # Drop the Mongo database.
         self.connection.drop_database(MONGO_DATABASE)
 
+    def get_handler(self, log_name):
+        logger = logging.getLogger(log_name)
+        handler = logging.handlers.BufferingHandler(20)
+        logger.addHandler(handler)
+
+        def cleanup():
+            logger.removeHandler(handler)
+        self.addCleanup(cleanup)
+        return handler
+
 
 class WebTestBase(MongoTestBase):
 
@@ -125,16 +135,6 @@
         self.addCleanup(cleanup)
         return (in_queue, out_queue)
 
-    def get_handler(self, log_name):
-        logger = logging.getLogger(log_name)
-        handler = logging.handlers.BufferingHandler(20)
-        logger.addHandler(handler)
-
-        def cleanup():
-            logger.removeHandler(handler)
-        self.addCleanup(cleanup)
-        return handler
-
 
 @contextmanager
 def login_request(app, groups=None):

=== modified file 'default.ini'
--- default.ini	2013-02-21 13:06:17 +0000
+++ default.ini	2013-02-28 13:04:21 +0000
@@ -36,6 +36,11 @@
 mongo.url = mongodb://localhost:27017
 mongo.database = juju
 
+# The time in minutes a lock is held by the scripts "queue" and "review".
+# This value should be greater than the time interval between two runs of
+# this script.
+script_lease_time = 30
+
 [server:main]
 use = egg:Paste#http
 host = 0.0.0.0