Merge lp:~jderose/dmedia/downgrade-unknown into lp:dmedia

[Jason Gerard DeRose]

For more info, see this bug:
https://bugs.launchpad.net/dmedia/+bug/1131742

This reworks MetaStore.downgrade_by_store_atime() to be driven by the "file/stored" view rather than the "store/atime" view.

Importantly, this change means copies in a store will always be downgraded when the store lacks a corresponding doc in CouchDB. Previously, a missing dmedia/store doc meant these copies never got downgraded by the store atime (although they would eventually get downgraded when the DOWNGRADE_BY_LAST_VERIFIED threshold was reached).

To be a bit more conservative as we test and tune the remaining automation behaviours, I also lowered DOWNGRADE_BY_LAST_VERIFIED from 4 weeks to 2 weeks.

Lastly, the threshold for when files would start to be re-verified was previous hard-coded using the ONE_WEEK constant. Instead, it's now using new VERIFY_THRESHOLD constant (also in the tests), making this parameter easy to tune.

[xzcvczx (xzcvczx)]

Nice

review: Approve

=== modified file 'dmedia/metastore.py'
--- dmedia/metastore.py	2013-02-21 13:38:52 +0000
+++ dmedia/metastore.py	2013-02-22 14:17:19 +0000
@@ -48,12 +48,13 @@
 
 
 log = logging.getLogger()
+
 DAY = 24 * 60 * 60
-ONE_WEEK = 7 * DAY
-
-DOWNGRADE_BY_STORE_ATIME = 7 * DAY  # 1 week
-DOWNGRADE_BY_NEVER_VERIFIED = 2 * DAY  # 48 hours
-DOWNGRADE_BY_LAST_VERIFIED = 28 * DAY  # 4 weeks
+WEEK = 7 * DAY
+DOWNGRADE_BY_NEVER_VERIFIED = 2 * DAY
+VERIFY_THRESHOLD = WEEK
+DOWNGRADE_BY_STORE_ATIME = WEEK
+DOWNGRADE_BY_LAST_VERIFIED = 2 * WEEK
 
 
 class MTimeMismatch(Exception):
@@ -71,7 +72,7 @@
         return time.perf_counter() - self.start
 
     def log(self, msg, *args):
-        log.info('[%.3fs] ' + msg, self.delta, *args)
+        log.info('[%.3f] ' + msg, self.delta, *args)
 
 
 def get_dict(d, key):
@@ -361,13 +362,24 @@
         if curtime is None:
             curtime = int(time.time())
         assert isinstance(curtime, int) and curtime >= 0
-        rows = self.db.view('store', 'atime',
-            endkey=(curtime - DOWNGRADE_BY_STORE_ATIME)
-        )['rows']
-        ids = [row['id'] for row in rows]
-        for store_id in ids:
-            self.downgrade_store(store_id)
-        return ids
+        threshold = curtime - DOWNGRADE_BY_STORE_ATIME
+        t = TimeDelta()
+        result = {}
+        rows = self.db.view('file', 'stored', reduce=True, group=True)['rows']
+        for row in rows:
+            store_id = row['key']
+            try:
+                doc = self.db.get(store_id)
+                atime = doc.get('atime')
+                if isinstance(atime, int) and atime > threshold:
+                    log.info('Store %s okay at atime %s', store_id, atime)
+                    continue
+            except NotFound:
+                log.warning('No doc found for store %s, forcing downgrade')
+            result[store_id] = self.downgrade_store(store_id)
+        total = sum(result.values())
+        t.log('downgraded %d total copies in %d stores', total, len(result))
+        return result
 
     def downgrade_store(self, store_id):
         t = TimeDelta()
@@ -545,7 +557,7 @@
 
     def verify_all(self, fs):
         start = [fs.id, None]
-        end = [fs.id, int(time.time()) - ONE_WEEK]
+        end = [fs.id, int(time.time()) - VERIFY_THRESHOLD]
         count = 0
         t = TimeDelta()
         log.info('verifying %r', fs)

=== modified file 'dmedia/tests/test_metastore.py'
--- dmedia/tests/test_metastore.py	2013-02-21 09:31:46 +0000
+++ dmedia/tests/test_metastore.py	2013-02-22 14:17:19 +0000
@@ -1019,54 +1019,128 @@
         for (old, new) in zip(docs, db.get_many(ids)):
             self.assertEqual(old, new)
 
-    def test_downgrade_by_store_atime(self):   
-        db = util.get_db(self.env, True)
-
-        class Dummy(metastore.MetaStore):
-            def __init__(self, db):
-                super().__init__(db)
-                self._calls = []
-
+    def test_downgrade_by_store_atime(self):
+        class PassThrough(metastore.MetaStore):
             def downgrade_store(self, store_id):
                 self._calls.append(store_id)
+                return super().downgrade_store(store_id)
 
-        # Test when empty
-        ms = Dummy(db)
-        self.assertEqual(ms.downgrade_by_store_atime(), [])
-        self.assertEqual(ms._calls, [])
+        db = util.get_db(self.env, True)
+        ms = PassThrough(db)
         curtime = int(time.time())
-        self.assertEqual(ms.downgrade_by_store_atime(curtime), [])
-        self.assertEqual(ms._calls, [])
-
-        # Test when some need to be downgraded
         base = curtime - metastore.DOWNGRADE_BY_STORE_ATIME
-        docs = []
-        for i in range(8):
-            doc = {
-                '_id': random_id(),
-                'type': 'dmedia/store',
-                'atime': base + i,
-            }
-            docs.append(doc)
-        db.save_many(docs)
-        ids = [doc['_id'] for doc in docs]
-        self.assertEqual(ms.downgrade_by_store_atime(curtime - 1), [])
+
+        # Test when empty:
+        ms._calls = []
+        self.assertEqual(ms.downgrade_by_store_atime(), {})
+        self.assertEqual(ms.downgrade_by_store_atime(curtime), {})
         self.assertEqual(ms._calls, [])
-        for i in range(8):
-            expected = ids[:i+1]
-            self.assertEqual(
-                ms.downgrade_by_store_atime(curtime + i),
-                expected
-            )
-            self.assertEqual(ms._calls, expected)
-            ms._calls = []
-
-        # Once more with feeling
-        self.assertEqual(
-            ms.downgrade_by_store_atime(curtime),
-            [ids[0]]
-        )
-        self.assertEqual(ms._calls, [ids[0]])
+
+        # One store that's missing its doc:
+        store_id1 = random_id()
+        ids1 = tuple(random_id() for i in range(17))
+        docs = [
+            {
+                '_id': _id,
+                'type': 'dmedia/file',
+                'stored': {store_id1: {'copies': 1}},
+            }
+            for _id in ids1
+        ]
+        db.save_many(docs)
+
+        # Another store with an atime old enough to trigger a downgrade:
+        store_id2 = random_id()
+        doc2 = {'_id': store_id2, 'atime': base}
+        db.save(doc2)
+        ids2 = tuple(random_id() for i in range(18))
+        docs = [
+            {
+                '_id': _id,
+                'type': 'dmedia/file',
+                'stored': {store_id2: {'copies': 1}},
+            }
+            for _id in ids2
+        ]
+        db.save_many(docs)
+
+        # A store with an atime new enough to be okay:
+        store_id3 = random_id()
+        doc3 = {'_id': store_id3, 'atime': base + 1}
+        db.save(doc3)
+        ids3 = tuple(random_id() for i in range(19))
+        docs = [
+            {
+                '_id': _id,
+                'type': 'dmedia/file',
+                'stored': {store_id3: {'copies': 1}},
+            }
+            for _id in ids3
+        ]
+        db.save_many(docs)
+
+        # And finally a store missing its doc['atime']:
+        store_id4 = random_id()
+        doc4 = {'_id': store_id4}
+        db.save(doc4)
+        ids4 = tuple(random_id() for i in range(20))
+        docs = [
+            {
+                '_id': _id,
+                'type': 'dmedia/file',
+                'stored': {store_id4: {'copies': 1}},
+            }
+            for _id in ids4
+        ]
+        db.save_many(docs)
+
+        # Test at curtime:
+        self.assertEqual(ms.downgrade_by_store_atime(curtime),
+            {store_id1: 17, store_id2: 18, store_id4: 20}
+        )
+        self.assertEqual(ms._calls,
+            sorted([store_id1, store_id2, store_id4])
+        )
+        for doc in db.get_many(ids1):
+            self.assertTrue(doc['_rev'].startswith('2-'))
+            self.assertEqual(doc['stored'], {store_id1: {'copies': 0}})
+        for doc in db.get_many(ids2):
+            self.assertTrue(doc['_rev'].startswith('2-'))
+            self.assertEqual(doc['stored'], {store_id2: {'copies': 0}})
+        for doc in db.get_many(ids3):
+            self.assertTrue(doc['_rev'].startswith('1-'))
+            self.assertEqual(doc['stored'], {store_id3: {'copies': 1}})
+        for doc in db.get_many(ids4):
+            self.assertTrue(doc['_rev'].startswith('2-'))
+            self.assertEqual(doc['stored'], {store_id4: {'copies': 0}})
+
+        # Test at curtime + 1:
+        ms._calls = []
+        self.assertEqual(ms.downgrade_by_store_atime(curtime + 1),
+            {store_id1: 0, store_id2: 0, store_id3: 19, store_id4: 0}
+        )
+        self.assertEqual(ms._calls,
+            sorted([store_id1, store_id2, store_id3, store_id4])
+        )
+        for doc in db.get_many(ids1):
+            self.assertTrue(doc['_rev'].startswith('2-'))
+            self.assertEqual(doc['stored'], {store_id1: {'copies': 0}})
+        for doc in db.get_many(ids2):
+            self.assertTrue(doc['_rev'].startswith('2-'))
+            self.assertEqual(doc['stored'], {store_id2: {'copies': 0}})
+        for doc in db.get_many(ids3):
+            self.assertTrue(doc['_rev'].startswith('2-'))
+            self.assertEqual(doc['stored'], {store_id3: {'copies': 0}})
+        for doc in db.get_many(ids4):
+            self.assertTrue(doc['_rev'].startswith('2-'))
+            self.assertEqual(doc['stored'], {store_id4: {'copies': 0}})
+
+        # Make sure the dmedia/store docs aren't modified:
+        with self.assertRaises(microfiber.NotFound) as cm:
+            db.get(store_id1)
+        self.assertEqual(db.get(store_id2), doc2)
+        self.assertEqual(db.get(store_id3), doc3)
+        self.assertEqual(db.get(store_id4), doc4)
 
     def test_downgrade_store(self):    
         db = util.get_db(self.env, True)
@@ -1554,7 +1628,7 @@
         again = docs2[:4]
         assert len(again) == 4
         for doc in again:
-            doc['stored'][fs.id]['verified'] -= (metastore.ONE_WEEK + 1)
+            doc['stored'][fs.id]['verified'] -= (metastore.VERIFY_THRESHOLD + 1)
             db.save(doc)
         self.assertEqual(ms.verify_all(fs), 4)
         self.assertEqual(ms.verify_all(fs), 0)