prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Tune Elasticsearch client for reindex separately from main search client
  #3892 was necessary to alleviate the hard outage experienced due to ES cluster being down.
  
  We were waiting 30s per request for Elasticsearch then retrying. This was added in #1471 to handle reindex issues.
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `warehouse/search/tasks.py`
  Content:
  ```
  1 # Licensed under the Apache License, Version 2.0 (the "License");
  2 # you may not use this file except in compliance with the License.
  3 # You may obtain a copy of the License at
  4 #
  5 # http://www.apache.org/licenses/LICENSE-2.0
  6 #
  7 # Unless required by applicable law or agreed to in writing, software
  8 # distributed under the License is distributed on an "AS IS" BASIS,
  9 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10 # See the License for the specific language governing permissions and
  11 # limitations under the License.
  12 
  13 import binascii
  14 import os
  15 
  16 from elasticsearch.helpers import parallel_bulk
  17 from sqlalchemy import and_, func
  18 from sqlalchemy.orm import aliased
  19 
  20 from warehouse.packaging.models import (
  21     Classifier, Project, Release, release_classifiers)
  22 from warehouse.packaging.search import Project as ProjectDocType
  23 from warehouse.search.utils import get_index
  24 from warehouse import tasks
  25 from warehouse.utils.db import windowed_query
  26 
  27 
  28 def _project_docs(db):
  29 
  30     releases_list = (
  31         db.query(Release.name, Release.version)
  32         .order_by(
  33             Release.name,
  34             Release.is_prerelease.nullslast(),
  35             Release._pypi_ordering.desc(),
  36         )
  37         .distinct(Release.name)
  38         .subquery("release_list")
  39     )
  40 
  41     r = aliased(Release, name="r")
  42 
  43     all_versions = (
  44         db.query(func.array_agg(r.version))
  45         .filter(r.name == Release.name)
  46         .correlate(Release)
  47         .as_scalar()
  48         .label("all_versions")
  49     )
  50 
  51     classifiers = (
  52         db.query(func.array_agg(Classifier.classifier))
  53         .select_from(release_classifiers)
  54         .join(Classifier, Classifier.id == release_classifiers.c.trove_id)
  55         .filter(Release.name == release_classifiers.c.name)
  56         .filter(Release.version == release_classifiers.c.version)
  57         .correlate(Release)
  58         .as_scalar()
  59         .label("classifiers")
  60     )
  61 
  62     release_data = (
  63         db.query(
  64             Release.description,
  65             Release.name,
  66             Release.version.label("latest_version"),
  67             all_versions,
  68             Release.author,
  69             Release.author_email,
  70             Release.maintainer,
  71             Release.maintainer_email,
  72             Release.home_page,
  73             Release.summary,
  74             Release.keywords,
  75             Release.platform,
  76             Release.download_url,
  77             Release.created,
  78             classifiers,
  79             Project.normalized_name,
  80             Project.name,
  81         )
  82         .select_from(releases_list)
  83         .join(Release, and_(
  84             Release.name == releases_list.c.name,
  85             Release.version == releases_list.c.version))
  86         .outerjoin(Release.project)
  87         .order_by(Release.name)
  88     )
  89 
  90     for release in windowed_query(release_data, Release.name, 50000):
  91         p = ProjectDocType.from_db(release)
  92         p.full_clean()
  93         yield p.to_dict(include_meta=True)
  94 
  95 
  96 @tasks.task(ignore_result=True, acks_late=True)
  97 def reindex(request):
  98     """
  99     Recreate the Search Index.
  100     """
  101     client = request.registry["elasticsearch.client"]
  102     number_of_replicas = request.registry.get("elasticsearch.replicas", 0)
  103     refresh_interval = request.registry.get("elasticsearch.interval", "1s")
  104 
  105     # We use a randomly named index so that we can do a zero downtime reindex.
  106     # Essentially we'll use a randomly named index which we will use until all
  107     # of the data has been reindexed, at which point we'll point an alias at
  108     # our randomly named index, and then delete the old randomly named index.
  109 
  110     # Create the new index and associate all of our doc types with it.
  111     index_base = request.registry["elasticsearch.index"]
  112     random_token = binascii.hexlify(os.urandom(5)).decode("ascii")
  113     new_index_name = "{}-{}".format(index_base, random_token)
  114     doc_types = request.registry.get("search.doc_types", set())
  115     shards = request.registry.get("elasticsearch.shards", 1)
  116 
  117     # Create the new index with zero replicas and index refreshes disabled
  118     # while we are bulk indexing.
  119     new_index = get_index(
  120         new_index_name,
  121         doc_types,
  122         using=client,
  123         shards=shards,
  124         replicas=0,
  125         interval="-1",
  126     )
  127     new_index.create(wait_for_active_shards=shards)
  128 
  129     # From this point on, if any error occurs, we want to be able to delete our
  130     # in progress index.
  131     try:
  132         request.db.execute("SET statement_timeout = '600s'")
  133 
  134         for _ in parallel_bulk(client, _project_docs(request.db)):
  135             pass
  136     except:  # noqa
  137         new_index.delete()
  138         raise
  139     finally:
  140         request.db.rollback()
  141         request.db.close()
  142 
  143     # Now that we've finished indexing all of our data we can optimize it and
  144     # update the replicas and refresh intervals.
  145     client.indices.forcemerge(index=new_index_name)
  146     client.indices.put_settings(
  147         index=new_index_name,
  148         body={
  149             "index": {
  150                 "number_of_replicas": number_of_replicas,
  151                 "refresh_interval": refresh_interval,
  152             }
  153         }
  154     )
  155 
  156     # Point the alias at our new randomly named index and delete the old index.
  157     if client.indices.exists_alias(name=index_base):
  158         to_delete = set()
  159         actions = []
  160         for name in client.indices.get_alias(name=index_base):
  161             to_delete.add(name)
  162             actions.append({"remove": {"index": name, "alias": index_base}})
  163         actions.append({"add": {"index": new_index_name, "alias": index_base}})
  164         client.indices.update_aliases({"actions": actions})
  165         client.indices.delete(",".join(to_delete))
  166     else:
  167         client.indices.put_alias(name=index_base, index=new_index_name)
  168
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/warehouse/search/tasks.py b/warehouse/search/tasks.py
  --- a/warehouse/search/tasks.py
  +++ b/warehouse/search/tasks.py
  @@ -11,11 +11,15 @@
   # limitations under the License.
   
   import binascii
  +import urllib
   import os
   
   from elasticsearch.helpers import parallel_bulk
  +from elasticsearch_dsl import serializer
   from sqlalchemy import and_, func
   from sqlalchemy.orm import aliased
  +import certifi
  +import elasticsearch
   
   from warehouse.packaging.models import (
       Classifier, Project, Release, release_classifiers)
  @@ -98,7 +102,15 @@
       """
       Recreate the Search Index.
       """
  -    client = request.registry["elasticsearch.client"]
  +    p = urllib.parse.urlparse(request.registry.settings["elasticsearch.url"])
  +    client = elasticsearch.Elasticsearch(
  +        [urllib.parse.urlunparse(p[:2] + ("",) * 4)],
  +        verify_certs=True,
  +        ca_certs=certifi.where(),
  +        timeout=30,
  +        retry_on_timeout=True,
  +        serializer=serializer.serializer,
  +    )
       number_of_replicas = request.registry.get("elasticsearch.replicas", 0)
       refresh_interval = request.registry.get("elasticsearch.interval", "1s")
  
verification_info: {"golden_diff": "diff --git a/warehouse/search/tasks.py b/warehouse/search/tasks.py\n--- a/warehouse/search/tasks.py\n+++ b/warehouse/search/tasks.py\n@@ -11,11 +11,15 @@\n # limitations under the License.\n \n import binascii\n+import urllib\n import os\n \n from elasticsearch.helpers import parallel_bulk\n+from elasticsearch_dsl import serializer\n from sqlalchemy import and_, func\n from sqlalchemy.orm import aliased\n+import certifi\n+import elasticsearch\n \n from warehouse.packaging.models import (\n     Classifier, Project, Release, release_classifiers)\n@@ -98,7 +102,15 @@\n     \"\"\"\n     Recreate the Search Index.\n     \"\"\"\n-    client = request.registry[\"elasticsearch.client\"]\n+    p = urllib.parse.urlparse(request.registry.settings[\"elasticsearch.url\"])\n+    client = elasticsearch.Elasticsearch(\n+        [urllib.parse.urlunparse(p[:2] + (\"\",) * 4)],\n+        verify_certs=True,\n+        ca_certs=certifi.where(),\n+        timeout=30,\n+        retry_on_timeout=True,\n+        serializer=serializer.serializer,\n+    )\n     number_of_replicas = request.registry.get(\"elasticsearch.replicas\", 0)\n     refresh_interval = request.registry.get(\"elasticsearch.interval\", \"1s\")\n", "issue": "Tune Elasticsearch client for reindex separately from main search client\n#3892 was necessary to alleviate the hard outage experienced due to ES cluster being down.\r\n\r\nWe were waiting 30s per request for Elasticsearch then retrying. This was added in #1471 to handle reindex issues.\n", "before_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport binascii\nimport os\n\nfrom elasticsearch.helpers import parallel_bulk\nfrom sqlalchemy import and_, func\nfrom sqlalchemy.orm import aliased\n\nfrom warehouse.packaging.models import (\n    Classifier, Project, Release, release_classifiers)\nfrom warehouse.packaging.search import Project as ProjectDocType\nfrom warehouse.search.utils import get_index\nfrom warehouse import tasks\nfrom warehouse.utils.db import windowed_query\n\n\ndef _project_docs(db):\n\n    releases_list = (\n        db.query(Release.name, Release.version)\n        .order_by(\n            Release.name,\n            Release.is_prerelease.nullslast(),\n            Release._pypi_ordering.desc(),\n        )\n        .distinct(Release.name)\n        .subquery(\"release_list\")\n    )\n\n    r = aliased(Release, name=\"r\")\n\n    all_versions = (\n        db.query(func.array_agg(r.version))\n        .filter(r.name == Release.name)\n        .correlate(Release)\n        .as_scalar()\n        .label(\"all_versions\")\n    )\n\n    classifiers = (\n        db.query(func.array_agg(Classifier.classifier))\n        .select_from(release_classifiers)\n        .join(Classifier, Classifier.id == release_classifiers.c.trove_id)\n        .filter(Release.name == release_classifiers.c.name)\n        .filter(Release.version == release_classifiers.c.version)\n        .correlate(Release)\n        .as_scalar()\n        .label(\"classifiers\")\n    )\n\n    release_data = (\n        db.query(\n            Release.description,\n            Release.name,\n            Release.version.label(\"latest_version\"),\n            all_versions,\n            Release.author,\n            Release.author_email,\n            Release.maintainer,\n            Release.maintainer_email,\n            Release.home_page,\n            Release.summary,\n            Release.keywords,\n            Release.platform,\n            Release.download_url,\n            Release.created,\n            classifiers,\n            Project.normalized_name,\n            Project.name,\n        )\n        .select_from(releases_list)\n        .join(Release, and_(\n            Release.name == releases_list.c.name,\n            Release.version == releases_list.c.version))\n        .outerjoin(Release.project)\n        .order_by(Release.name)\n    )\n\n    for release in windowed_query(release_data, Release.name, 50000):\n        p = ProjectDocType.from_db(release)\n        p.full_clean()\n        yield p.to_dict(include_meta=True)\n\n\n@tasks.task(ignore_result=True, acks_late=True)\ndef reindex(request):\n    \"\"\"\n    Recreate the Search Index.\n    \"\"\"\n    client = request.registry[\"elasticsearch.client\"]\n    number_of_replicas = request.registry.get(\"elasticsearch.replicas\", 0)\n    refresh_interval = request.registry.get(\"elasticsearch.interval\", \"1s\")\n\n    # We use a randomly named index so that we can do a zero downtime reindex.\n    # Essentially we'll use a randomly named index which we will use until all\n    # of the data has been reindexed, at which point we'll point an alias at\n    # our randomly named index, and then delete the old randomly named index.\n\n    # Create the new index and associate all of our doc types with it.\n    index_base = request.registry[\"elasticsearch.index\"]\n    random_token = binascii.hexlify(os.urandom(5)).decode(\"ascii\")\n    new_index_name = \"{}-{}\".format(index_base, random_token)\n    doc_types = request.registry.get(\"search.doc_types\", set())\n    shards = request.registry.get(\"elasticsearch.shards\", 1)\n\n    # Create the new index with zero replicas and index refreshes disabled\n    # while we are bulk indexing.\n    new_index = get_index(\n        new_index_name,\n        doc_types,\n        using=client,\n        shards=shards,\n        replicas=0,\n        interval=\"-1\",\n    )\n    new_index.create(wait_for_active_shards=shards)\n\n    # From this point on, if any error occurs, we want to be able to delete our\n    # in progress index.\n    try:\n        request.db.execute(\"SET statement_timeout = '600s'\")\n\n        for _ in parallel_bulk(client, _project_docs(request.db)):\n            pass\n    except:  # noqa\n        new_index.delete()\n        raise\n    finally:\n        request.db.rollback()\n        request.db.close()\n\n    # Now that we've finished indexing all of our data we can optimize it and\n    # update the replicas and refresh intervals.\n    client.indices.forcemerge(index=new_index_name)\n    client.indices.put_settings(\n        index=new_index_name,\n        body={\n            \"index\": {\n                \"number_of_replicas\": number_of_replicas,\n                \"refresh_interval\": refresh_interval,\n            }\n        }\n    )\n\n    # Point the alias at our new randomly named index and delete the old index.\n    if client.indices.exists_alias(name=index_base):\n        to_delete = set()\n        actions = []\n        for name in client.indices.get_alias(name=index_base):\n            to_delete.add(name)\n            actions.append({\"remove\": {\"index\": name, \"alias\": index_base}})\n        actions.append({\"add\": {\"index\": new_index_name, \"alias\": index_base}})\n        client.indices.update_aliases({\"actions\": actions})\n        client.indices.delete(\",\".join(to_delete))\n    else:\n        client.indices.put_alias(name=index_base, index=new_index_name)\n", "path": "warehouse/search/tasks.py"}], "after_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport binascii\nimport urllib\nimport os\n\nfrom elasticsearch.helpers import parallel_bulk\nfrom elasticsearch_dsl import serializer\nfrom sqlalchemy import and_, func\nfrom sqlalchemy.orm import aliased\nimport certifi\nimport elasticsearch\n\nfrom warehouse.packaging.models import (\n    Classifier, Project, Release, release_classifiers)\nfrom warehouse.packaging.search import Project as ProjectDocType\nfrom warehouse.search.utils import get_index\nfrom warehouse import tasks\nfrom warehouse.utils.db import windowed_query\n\n\ndef _project_docs(db):\n\n    releases_list = (\n        db.query(Release.name, Release.version)\n        .order_by(\n            Release.name,\n            Release.is_prerelease.nullslast(),\n            Release._pypi_ordering.desc(),\n        )\n        .distinct(Release.name)\n        .subquery(\"release_list\")\n    )\n\n    r = aliased(Release, name=\"r\")\n\n    all_versions = (\n        db.query(func.array_agg(r.version))\n        .filter(r.name == Release.name)\n        .correlate(Release)\n        .as_scalar()\n        .label(\"all_versions\")\n    )\n\n    classifiers = (\n        db.query(func.array_agg(Classifier.classifier))\n        .select_from(release_classifiers)\n        .join(Classifier, Classifier.id == release_classifiers.c.trove_id)\n        .filter(Release.name == release_classifiers.c.name)\n        .filter(Release.version == release_classifiers.c.version)\n        .correlate(Release)\n        .as_scalar()\n        .label(\"classifiers\")\n    )\n\n    release_data = (\n        db.query(\n            Release.description,\n            Release.name,\n            Release.version.label(\"latest_version\"),\n            all_versions,\n            Release.author,\n            Release.author_email,\n            Release.maintainer,\n            Release.maintainer_email,\n            Release.home_page,\n            Release.summary,\n            Release.keywords,\n            Release.platform,\n            Release.download_url,\n            Release.created,\n            classifiers,\n            Project.normalized_name,\n            Project.name,\n        )\n        .select_from(releases_list)\n        .join(Release, and_(\n            Release.name == releases_list.c.name,\n            Release.version == releases_list.c.version))\n        .outerjoin(Release.project)\n        .order_by(Release.name)\n    )\n\n    for release in windowed_query(release_data, Release.name, 50000):\n        p = ProjectDocType.from_db(release)\n        p.full_clean()\n        yield p.to_dict(include_meta=True)\n\n\n@tasks.task(ignore_result=True, acks_late=True)\ndef reindex(request):\n    \"\"\"\n    Recreate the Search Index.\n    \"\"\"\n    p = urllib.parse.urlparse(request.registry.settings[\"elasticsearch.url\"])\n    client = elasticsearch.Elasticsearch(\n        [urllib.parse.urlunparse(p[:2] + (\"\",) * 4)],\n        verify_certs=True,\n        ca_certs=certifi.where(),\n        timeout=30,\n        retry_on_timeout=True,\n        serializer=serializer.serializer,\n    )\n    number_of_replicas = request.registry.get(\"elasticsearch.replicas\", 0)\n    refresh_interval = request.registry.get(\"elasticsearch.interval\", \"1s\")\n\n    # We use a randomly named index so that we can do a zero downtime reindex.\n    # Essentially we'll use a randomly named index which we will use until all\n    # of the data has been reindexed, at which point we'll point an alias at\n    # our randomly named index, and then delete the old randomly named index.\n\n    # Create the new index and associate all of our doc types with it.\n    index_base = request.registry[\"elasticsearch.index\"]\n    random_token = binascii.hexlify(os.urandom(5)).decode(\"ascii\")\n    new_index_name = \"{}-{}\".format(index_base, random_token)\n    doc_types = request.registry.get(\"search.doc_types\", set())\n    shards = request.registry.get(\"elasticsearch.shards\", 1)\n\n    # Create the new index with zero replicas and index refreshes disabled\n    # while we are bulk indexing.\n    new_index = get_index(\n        new_index_name,\n        doc_types,\n        using=client,\n        shards=shards,\n        replicas=0,\n        interval=\"-1\",\n    )\n    new_index.create(wait_for_active_shards=shards)\n\n    # From this point on, if any error occurs, we want to be able to delete our\n    # in progress index.\n    try:\n        request.db.execute(\"SET statement_timeout = '600s'\")\n\n        for _ in parallel_bulk(client, _project_docs(request.db)):\n            pass\n    except:  # noqa\n        new_index.delete()\n        raise\n    finally:\n        request.db.rollback()\n        request.db.close()\n\n    # Now that we've finished indexing all of our data we can optimize it and\n    # update the replicas and refresh intervals.\n    client.indices.forcemerge(index=new_index_name)\n    client.indices.put_settings(\n        index=new_index_name,\n        body={\n            \"index\": {\n                \"number_of_replicas\": number_of_replicas,\n                \"refresh_interval\": refresh_interval,\n            }\n        }\n    )\n\n    # Point the alias at our new randomly named index and delete the old index.\n    if client.indices.exists_alias(name=index_base):\n        to_delete = set()\n        actions = []\n        for name in client.indices.get_alias(name=index_base):\n            to_delete.add(name)\n            actions.append({\"remove\": {\"index\": name, \"alias\": index_base}})\n        actions.append({\"add\": {\"index\": new_index_name, \"alias\": index_base}})\n        client.indices.update_aliases({\"actions\": actions})\n        client.indices.delete(\",\".join(to_delete))\n    else:\n        client.indices.put_alias(name=index_base, index=new_index_name)\n", "path": "warehouse/search/tasks.py"}]}