Add a command to search for add-ons inconsistencies between ES and MySQL

src/olympia/addons/management/commands/process_addons.py

@@ -5,6 +5,7 @@ from django.core.management.base import BaseCommand, CommandError
 from celery import chord, group
 
 from olympia.addons.models import Addon
+from olympia.addons.tasks import find_inconsistencies_between_es_and_db
 from olympia.amo.utils import chunked
 from olympia.devhub.tasks import convert_purified, get_preview_sizes
 from olympia.lib.crypto.tasks import sign_addons
@@ -12,6 +13,8 @@ from olympia.reviews.tasks import addon_review_aggregates
 
 
 tasks = {
+    'find_inconsistencies_between_es_and_db': {
+        'method': find_inconsistencies_between_es_and_db, 'qs': []},
     'get_preview_sizes': {'method': get_preview_sizes, 'qs': []},
     'convert_purified': {'method': convert_purified, 'qs': []},
     'addon_review_aggregates': {'method': addon_review_aggregates, 'qs': []},
@@ -33,6 +36,10 @@ class Command(BaseCommand):
     option_list = BaseCommand.option_list + (
         make_option('--task', action='store', type='string',
                     dest='task', help='Run task on the addons.'),
+        make_option('--with-deleted', action='store_true',
+                    dest='with_deleted',
+                    help='Include deleted add-ons when determining which '
+                         'add-ons to process.'),
     )
 
     def handle(self, *args, **options):
@@ -40,9 +47,13 @@ class Command(BaseCommand):
         if not task:
             raise CommandError('Unknown task provided. Options are: %s'
                                % ', '.join(tasks.keys()))
-        pks = (Addon.objects.filter(*task['qs'])
+        if options.get('with_deleted'):
+            addon_manager = Addon.unfiltered
+        else:
+            addon_manager = Addon.objects
+        pks = (addon_manager.filter(*task['qs'])
                             .values_list('pk', flat=True)
-                            .order_by('-last_updated'))
+                            .order_by('id'))
         if 'pre' in task:
             # This is run in process to ensure its run before the tasks.
             pks = task['pre'](pks)

src/olympia/addons/tasks.py

@@ -5,6 +5,7 @@ from django.conf import settings
 from django.core.files.storage import default_storage as storage
 from django.db import transaction
 
+from elasticsearch_dsl import Search
 from PIL import Image
 
 import olympia.core.logger
@@ -374,3 +375,37 @@ def calc_checksum(theme_id, **kw):
         theme.save()
     except IOError as e:
         log.error(str(e))
+
+
+@task
+@write  # To bypass cache and use the primary replica.
+def find_inconsistencies_between_es_and_db(ids, **kw):
+    length = len(ids)
+    log.info(
+        'Searching for inconsistencies between db and es %d-%d [%d].',
+        ids[0], ids[-1], length)
+    db_addons = Addon.unfiltered.in_bulk(ids)
+    es_addons = Search(
+        doc_type=AddonIndexer.get_doctype_name(),
+        index=AddonIndexer.get_index_alias(),
+        using=amo.search.get_es()).filter('ids', values=ids)[:length].execute()
+    es_addons = es_addons
+    db_len = len(db_addons)
+    es_len = len(es_addons)
+    if db_len != es_len:
+        log.info('Inconsistency found: %d in db vs %d in es.',
+                 db_len, es_len)
+    for result in es_addons.hits.hits:
+        pk = result['_source']['id']
+        db_modified = db_addons[pk].modified.isoformat()
+        es_modified = result['_source']['modified']
+        if db_modified != es_modified:
+            log.info('Inconsistency found for addon %d: '
+                     'modified is %s in db vs %s in es.',
+                     pk, db_modified, es_modified)
+        db_status = db_addons[pk].status
+        es_status = result['_source']['status']
+        if db_status != es_status:
+            log.info('Inconsistency found for addon %d: '
+                     'status is %s in db vs %s in es.',
+                     pk, db_status, es_status)