Fix reindex (#3989)

* Make sure data always goes to new index

Fixes #3746

* Make celery workers more short lived

* Fix tests

Author: honzakral

Procfile.cabotage

@@ -1,5 +1,5 @@
 release: bin/release-cabotage
 web: bin/start-web-cabotage python -m gunicorn.app.wsgiapp -c gunicorn-cabotage.conf warehouse.wsgi:application
 web-uploads: bin/start-web-cabotage python -m gunicorn.app.wsgiapp -c gunicorn-uploads-cabotage.conf warehouse.wsgi:application
-worker: bin/start-worker celery -A warehouse worker -l info --max-tasks-per-child 128
+worker: bin/start-worker celery -A warehouse worker -l info --max-tasks-per-child 32
 worker-beat: bin/start-worker celery -A warehouse beat -S redbeat.RedBeatScheduler -l info

tests/unit/search/test_tasks.py

@@ -128,9 +128,10 @@ def project_docs(db):
         class TestException(Exception):
             pass
 
-        def parallel_bulk(client, iterable):
+        def parallel_bulk(client, iterable, index=None):
             assert client is es_client
             assert iterable is docs
+            assert index == "warehouse-cbcbcbcbcb"
             raise TestException
 
         monkeypatch.setattr(
@@ -176,15 +177,19 @@ def project_docs(db):
             lambda *a, **kw: es_client
         )
 
-        parallel_bulk = pretend.call_recorder(lambda client, iterable: [None])
+        parallel_bulk = pretend.call_recorder(
+            lambda client, iterable, index: [None]
+        )
         monkeypatch.setattr(
             warehouse.search.tasks, "parallel_bulk", parallel_bulk)
 
         monkeypatch.setattr(os, "urandom", lambda n: b"\xcb" * n)
 
         reindex(db_request)
 
-        assert parallel_bulk.calls == [pretend.call(es_client, docs)]
+        assert parallel_bulk.calls == [
+            pretend.call(es_client, docs, index="warehouse-cbcbcbcbcb")
+        ]
         assert es_client.indices.create.calls == [
             pretend.call(
                 body={
@@ -247,15 +252,19 @@ def project_docs(db):
             lambda *a, **kw: es_client
         )
 
-        parallel_bulk = pretend.call_recorder(lambda client, iterable: [None])
+        parallel_bulk = pretend.call_recorder(
+            lambda client, iterable, index: [None]
+        )
         monkeypatch.setattr(
             warehouse.search.tasks, "parallel_bulk", parallel_bulk)
 
         monkeypatch.setattr(os, "urandom", lambda n: b"\xcb" * n)
 
         reindex(db_request)
 
-        assert parallel_bulk.calls == [pretend.call(es_client, docs)]
+        assert parallel_bulk.calls == [
+            pretend.call(es_client, docs, index="warehouse-cbcbcbcbcb")
+        ]
         assert es_client.indices.create.calls == [
             pretend.call(
                 body={

warehouse/search/tasks.py

@@ -94,7 +94,9 @@ def _project_docs(db):
     for release in windowed_query(release_data, Release.name, 50000):
         p = ProjectDocType.from_db(release)
         p.full_clean()
-        yield p.to_dict(include_meta=True)
+        doc = p.to_dict(include_meta=True)
+        doc.pop('_index', None)
+        yield doc
 
 
 @tasks.task(ignore_result=True, acks_late=True)
@@ -143,7 +145,8 @@ def reindex(request):
     try:
         request.db.execute("SET statement_timeout = '600s'")
 
-        for _ in parallel_bulk(client, _project_docs(request.db)):
+        for _ in parallel_bulk(client, _project_docs(request.db),
+                               index=new_index_name):
             pass
     except:  # noqa
         new_index.delete()