Merge lp:~ev/oops-repository/bucket-versions into lp:~daisy-pluckers/oops-repository/trunk

[Evan]

This branch changes the updating of the package versions for each bucket. It now keeps track of which release as well as which version the bucket was updated for (ordering the latter using the dpkg binary version number comparator).

It uses three new column families:

BucketVersionsFull - A wide row of OOPS IDs for each combination of (bucket, DistroRelease, binary package version).

BucketVersionsCount - A counter column family that keeps track of the (DistroRelease, binary package version) count for each bucket. The binary package version is ordered using the dpkg comparator, which produces the following:
{('Ubuntu 12.04', '1.0~ppa1') : 12, ('Ubuntu 12.04', '1.0') : 9, ('Ubuntu 12.10', '1.0') : 36, ...}

BucketVersionsDay - A column family that's just used to keep track of which buckets we've updated the versions for today (today as when they were bucketed, not when the crash occurred - see the comment in the code). We'll use this to repair counters that were over/under-counted.

I'm tempted to only add to this when an exception is encountered when modifying the counter, rather than it grow with the BucketVersionsFull CF. We could also use a TTL to expire the data after a week or two (since it's not a counter CF). Can you think of any reason we might not want to do this? Is there any reason you can think of to keep around the knowledge of when we incremented the counters for BucketVersionsCount?

I haven't been able to think up one myself yet, but I'm not ruling it out :)

There's a script in lp:daisy that will be used to back-populate this data (which is updated to match this merge as https://code.launchpad.net/~ev/daisy/bucket-versions/+merge/158948).

[Brian Murray (brian-murray)]

In the Description you indicate that BucketVersionsCount is "(DistroRelease, binary package version)", however the code is written so that it is "bv_count.add(bucketid, (version, release))". I'm not sure why but I'd expect the ordering to have release before version instead of the other way around. Is there a specifc reason it is done with version before release?

[Brian Murray (brian-murray)]

Additionally, why is pycassa.util.uuid.UUID used? It seems to be just the same as uuid.UUID.

[Evan (ev)]

I've made the requested fixes as r68 (uuid) and r69 (release and version swap).

[Evan (ev)]

I made the supporting change to lp:daisy (as the lp:~ev/daisy/bucket-versions is now merged) as r316.

=== modified file 'oopsrepository/oopses.py'
--- oopsrepository/oopses.py	2013-04-12 14:10:15 +0000
+++ oopsrepository/oopses.py	2013-04-15 14:57:05 +0000
@@ -186,11 +186,37 @@
             daybucketcount_cf.add(resolution, bucketid)
     return day_key
 
-def update_bucket_versions(config, bucketid, version):
+def update_bucket_versions(config, bucketid, version,
+                           release=None, oopsid=None):
+
     pool = connection_pool(config)
-    bucketversions_cf = pycassa.ColumnFamily(pool, 'BucketVersions',
-                                             retry_counter_mutations=True)
-    bucketversions_cf.add(bucketid, version)
+    if release:
+        bv_full = pycassa.ColumnFamily(pool, 'BucketVersionsFull')
+        bv_count = pycassa.ColumnFamily(pool, 'BucketVersionsCount')
+        bv_day = pycassa.ColumnFamily(pool, 'BucketVersionsDay')
+
+        # Use the current day, rather than the day of the OOPS because this is
+        # specifically used for cleaning up counters nightly. If a very old
+        # OOPS gets processed here, we should still clean it up when we're
+        # handling the data for today.
+        day_key = time.strftime('%Y%m%d', time.gmtime())
+
+        o = pycassa.util.uuid.UUID(oopsid)
+        # When correcting the counts in bv_count, we'll iterate
+        # BucketVersionsDay for day_key. For each of these columns, we'll look
+        # up the correct value by calling bv_full.get_count(...).
+        bv_full.insert((bucketid, release, version), {o: ''})
+        bv_day.insert(day_key, {(bucketid, release, version): ''})
+
+        try:
+            bv_count.add(bucketid, (version, release))
+        except pycassa.MaximumRetryException:
+            # We'll clean these up nightly, so ignore timeouts.
+            pass
+    else:
+        bucketversions_cf = pycassa.ColumnFamily(pool, 'BucketVersions',
+                                                 retry_counter_mutations=True)
+        bucketversions_cf.add(bucketid, version)
 
 def query_bucket_versions(config, bucketid):
     pool = connection_pool(config)
 
=== modified file 'oopsrepository/schema.py'
--- oopsrepository/schema.py	2013-04-03 11:56:23 +0000
+++ oopsrepository/schema.py	2013-04-15 14:57:05 +0000
@@ -70,6 +70,28 @@
             workaround_1779(mgr.create_column_family, keyspace, 'BucketVersions',
                 comparator_type=UTF8_TYPE,
                 default_validation_class=CounterColumnType())
+        if 'BucketVersionsCount' not in cfs:
+            dpkg_comparator = 'com.canonical.dpkgversiontype.DpkgVersionType'
+            composite = CompositeType(dpkg_comparator, AsciiType())
+            workaround_1779(mgr.create_column_family,
+                            keyspace,
+                            'BucketVersionsCount',
+                            key_validation_class=UTF8_TYPE,
+                            comparator_type=composite,
+                            default_validation_class=CounterColumnType())
+        if 'BucketVersionsFull' not in cfs:
+            composite = CompositeType(UTF8Type(), AsciiType(), AsciiType())
+            workaround_1779(mgr.create_column_family,
+                            keyspace,
+                            'BucketVersionsFull',
+                            key_validation_class=composite,
+                            comparator_type=TIME_UUID_TYPE)
+        if 'BucketVersionsDay' not in cfs:
+            composite = CompositeType(UTF8Type(), AsciiType(), AsciiType())
+            workaround_1779(mgr.create_column_family,
+                            keyspace,
+                            'BucketVersionsDay',
+                            comparator_type=composite)
         if 'BucketSystems' not in cfs:
             workaround_1779(mgr.create_column_family, keyspace, 'BucketSystems',
                 key_validation_class=UTF8_TYPE,
 
=== modified file 'oopsrepository/tests/test_oopses.py'
--- oopsrepository/tests/test_oopses.py	2013-04-03 11:56:23 +0000
+++ oopsrepository/tests/test_oopses.py	2013-04-15 14:57:05 +0000
@@ -212,6 +212,32 @@
         oopses.update_bucket_versions(self.config, 'bucket-id', '1.2.3')
         self.assertEqual(bucketversions_cf.get('bucket-id')['1.2.3'], 1)
 
+        bv_full = pycassa.ColumnFamily(self.pool, 'BucketVersionsFull')
+        bv_count = pycassa.ColumnFamily(self.pool, 'BucketVersionsCount')
+        u = uuid.uuid1()
+        args = (self.config, 'bucket-id', '1.2.3', 'Ubuntu 12.04', str(u))
+        oopses.update_bucket_versions(*args)
+        d = bv_full.get(('bucket-id', 'Ubuntu 12.04', '1.2.3')).items()[0]
+        self.assertEqual((u, ''), d)
+        c = bv_count.get('bucket-id', columns=[('1.2.3', 'Ubuntu 12.04')])
+        c = c.values()[0]
+        self.assertEqual(1, c)
+    
+    def test_dpkg_comparator(self):
+        keyspace = self.useFixture(TemporaryOOPSDB()).keyspace
+        bv_count = pycassa.ColumnFamily(self.pool, 'BucketVersionsCount')
+        bv_count.add('bucket-id', ('1.0', 'release'))
+        bv_count.get('bucket-id', columns=[('1.0', 'release')])
+        bv_count.get('bucket-id', column_start=[('1.0', 'release')])
+        bv_count.get('bucket-id', column_finish=[('1.0', 'release')])
+        bv_count.add('bucket-id', ('1.0~ev1', 'release'))
+        self.assertEqual(1,
+            bv_count.get('bucket-id', column_count=1)[('1.0~ev1', 'release')])
+        bv_count.add('bucket-id', ('1.0+ev1', 'release'))
+        c = [('1.0+', 'release')]
+        self.assertEqual(1,
+            bv_count.get('bucket-id', column_start=c)[('1.0+ev1', 'release')])
+
     def test_update_errors_by_release(self):
         keyspace = self.useFixture(TemporaryOOPSDB()).keyspace
         firsterror = pycassa.ColumnFamily(self.pool, 'FirstError')