Schedule OpenSearch index cleanup

tests/unit/cli/test_search.py

@@ -76,9 +76,15 @@ def test_delete_older_indices_success(self, cli):
             ),
             get=pretend.call_recorder(
                 lambda **kw: {
-                    "production-2024-03-01": {},
-                    "production-2024-02-01": {},
-                    "production-2024-01-01": {},
+                    "production-2024-03-01": {
+                        "settings": {"index": {"creation_date": "3"}}
+                    },
+                    "production-2024-02-01": {
+                        "settings": {"index": {"creation_date": "2"}}
+                    },
+                    "production-2024-01-01": {
+                        "settings": {"index": {"creation_date": "1"}}
+                    },
                 }
             ),
             delete=pretend.call_recorder(lambda **kw: {"acknowledged": True}),
@@ -113,11 +119,21 @@ def test_delete_older_indices_multiple_deletions(self, cli):
             ),
             get=pretend.call_recorder(
                 lambda **kw: {
-                    "production-2024-05-01": {},
-                    "production-2024-04-01": {},
-                    "production-2024-03-01": {},
-                    "production-2024-02-01": {},
-                    "production-2024-01-01": {},
+                    "production-2024-05-01": {
+                        "settings": {"index": {"creation_date": "5"}}
+                    },
+                    "production-2024-04-01": {
+                        "settings": {"index": {"creation_date": "4"}}
+                    },
+                    "production-2024-03-01": {
+                        "settings": {"index": {"creation_date": "3"}}
+                    },
+                    "production-2024-02-01": {
+                        "settings": {"index": {"creation_date": "2"}}
+                    },
+                    "production-2024-01-01": {
+                        "settings": {"index": {"creation_date": "1"}}
+                    },
                 }
             ),
             delete=pretend.call_recorder(lambda **kw: {"acknowledged": True}),
@@ -162,7 +178,13 @@ def test_delete_older_indices_only_current(self, cli):
             get_alias=pretend.call_recorder(
                 lambda **kw: {"production-2024-03-01": {"aliases": {"production": {}}}}
             ),
-            get=pretend.call_recorder(lambda **kw: {"production-2024-03-01": {}}),
+            get=pretend.call_recorder(
+                lambda **kw: {
+                    "production-2024-03-01": {
+                        "settings": {"index": {"creation_date": "3"}}
+                    }
+                }
+            ),
             delete=pretend.call_recorder(lambda **kw: {"acknowledged": True}),
         )
         client = pretend.stub(indices=indices_client)
@@ -187,8 +209,12 @@ def test_delete_older_indices_keeps_two(self, cli):
             ),
             get=pretend.call_recorder(
                 lambda **kw: {
-                    "production-2024-02-01": {},
-                    "production-2024-01-01": {},
+                    "production-2024-02-01": {
+                        "settings": {"index": {"creation_date": "2"}}
+                    },
+                    "production-2024-01-01": {
+                        "settings": {"index": {"creation_date": "1"}}
+                    },
                 }
             ),
             delete=pretend.call_recorder(lambda **kw: {"acknowledged": True}),

tests/unit/search/test_init.py

@@ -3,6 +3,8 @@
 import opensearchpy
 import pretend
 
+from celery.schedules import crontab
+
 from warehouse import search
 
 from ...common.db.packaging import ProjectFactory, ReleaseFactory
@@ -160,6 +162,10 @@ def test_includeme(monkeypatch):
             iface=search.interfaces.ISearchService,
         ),
     ]
+    assert config.add_periodic_task.calls == [
+        pretend.call(crontab(minute=0, hour=6), search.reindex),
+        pretend.call(crontab(minute=0, hour=8), search.delete_older_indices),
+    ]
 
 
 def test_execute_reindex_no_service():

tests/unit/search/test_tasks.py

@@ -17,7 +17,9 @@
 from warehouse.packaging.models import LifecycleStatus
 from warehouse.search.tasks import (
     SearchLock,
+    _delete_index,
     _project_docs,
+    delete_older_indices,
     reindex,
     reindex_project,
     unindex_project,
@@ -165,7 +167,14 @@ def __init__(self):
     def exists_alias(self, name):
         return name in self.aliases
 
+    def get(self, index):
+        return self.indices
+
     def get_alias(self, name):
+        if name not in self.aliases:
+            raise opensearchpy.exceptions.NotFoundError(
+                404, "alias_not_found", "no alias"
+            )
         return self.aliases[name]
 
     def put_alias(self, name, index):
@@ -503,6 +512,121 @@ def project_docs(db):
         ]
 
 
+class TestDeleteIndex:
+    def test_deletes(self):
+        es_client = FakeESClient()
+
+        _delete_index(es_client, "warehouse-aaaaaaaaaa")
+
+        assert es_client.indices.delete.calls == [
+            pretend.call(index="warehouse-aaaaaaaaaa")
+        ]
+
+    def test_swallows_snapshot_in_progress(self, monkeypatch):
+        es_client = FakeESClient()
+        err = opensearchpy.exceptions.RequestError(
+            400, "snapshot_in_progress_exception", {}
+        )
+
+        def delete(index):
+            raise err
+
+        es_client.indices.delete = delete
+
+        captured = []
+        monkeypatch.setattr(
+            warehouse.search.tasks.sentry_sdk,
+            "capture_exception",
+            captured.append,
+        )
+
+        _delete_index(es_client, "warehouse-aaaaaaaaaa")
+
+        assert captured == [err]
+
+
+class TestDeleteOlderIndices:
+    def _setup(self, db_request, monkeypatch, es_client):
+        db_request.registry.update({"opensearch.index": "production"})
+        db_request.registry.settings = {
+            "opensearch.url": "http://some.url",
+            "celery.scheduler_url": "redis://redis:6379/0",
+        }
+        monkeypatch.setattr(
+            warehouse.search.tasks.opensearchpy,
+            "OpenSearch",
+            lambda *a, **kw: es_client,
+        )
+        monkeypatch.setattr(warehouse.search.tasks, "SearchLock", NotLock)
+
+    def test_deletes_older_than_two(self, db_request, monkeypatch):
+        es_client = FakeESClient()
+        es_client.indices.indices = {
+            "production-0001": {"settings": {"index": {"creation_date": "1"}}},
+            "production-0002": {"settings": {"index": {"creation_date": "2"}}},
+            "production-0003": {"settings": {"index": {"creation_date": "3"}}},
+            "production-0004": {"settings": {"index": {"creation_date": "4"}}},
+        }
+        es_client.indices.aliases = {"production": ["production-0004"]}
+        self._setup(db_request, monkeypatch, es_client)
+
+        delete_older_indices(pretend.stub(), db_request)
+
+        assert es_client.indices.delete.calls == [
+            pretend.call(index="production-0002"),
+            pretend.call(index="production-0001"),
+        ]
+
+    def test_keeps_two_most_recent(self, db_request, monkeypatch):
+        es_client = FakeESClient()
+        es_client.indices.indices = {
+            "production-0001": {"settings": {"index": {"creation_date": "1"}}},
+            "production-0002": {"settings": {"index": {"creation_date": "2"}}},
+        }
+        es_client.indices.aliases = {"production": ["production-0002"]}
+        self._setup(db_request, monkeypatch, es_client)
+
+        delete_older_indices(pretend.stub(), db_request)
+
+        assert es_client.indices.delete.calls == []
+
+    def test_only_live_index(self, db_request, monkeypatch):
+        es_client = FakeESClient()
+        es_client.indices.indices = {
+            "production-0001": {"settings": {"index": {"creation_date": "1"}}}
+        }
+        es_client.indices.aliases = {"production": ["production-0001"]}
+        self._setup(db_request, monkeypatch, es_client)
+
+        delete_older_indices(pretend.stub(), db_request)
+
+        assert es_client.indices.delete.calls == []
+
+    def test_no_alias_is_noop(self, db_request, monkeypatch):
+        es_client = FakeESClient()
+        es_client.indices.indices = {"production-0001": None}
+        self._setup(db_request, monkeypatch, es_client)
+
+        delete_older_indices(pretend.stub(), db_request)
+
+        assert es_client.indices.delete.calls == []
+
+    def test_retry_on_lock(self, db_request, monkeypatch):
+        task = pretend.stub(
+            retry=pretend.call_recorder(pretend.raiser(celery.exceptions.Retry))
+        )
+
+        db_request.registry.settings = {"celery.scheduler_url": "redis://redis:6379/0"}
+
+        le = redis.exceptions.LockError("Failed to acquire lock")
+        monkeypatch.setattr(SearchLock, "acquire", pretend.raiser(le))
+
+        with pytest.raises(celery.exceptions.Retry):
+            delete_older_indices(task, db_request)
+
+        assert task.retry.calls == [pretend.call(countdown=60, exc=le)]
+
+
 class TestPartialReindex:
     def test_reindex_fails_when_raising(self, db_request, monkeypatch):
         docs = pretend.stub()

warehouse/cli/search.py

@@ -2,10 +2,8 @@
 
 import click
 
-from opensearchpy.exceptions import NotFoundError
-
 from warehouse.cli import warehouse
-from warehouse.search.tasks import reindex as _reindex
+from warehouse.search.tasks import prune_older_indices, reindex as _reindex
 
 
 @warehouse.group()
@@ -63,29 +61,13 @@ def delete_older_indices(config, env_name):
     """
     client = config.registry["opensearch.client"]
 
-    # Gets alias of current "live" index, don't remove that one
-    try:
-        alias = client.indices.get_alias(name=env_name)
-    except NotFoundError:
+    result = prune_older_indices(client, env_name)
+    if result is None:
         click.echo(f"No alias found for {env_name}, aborting.", err=True)
         raise click.Abort
 
-    current_index = next(iter(alias.keys()))
-    click.echo(f"Current index: {current_index}")
-
-    indices = client.indices.get(index=f"{env_name}-*")
-    # sort the response by date, keep most recent 2
-    indices = sorted(indices.keys(), reverse=True)
-    # remove current index from the list
-    indices.remove(current_index)
-    # Remove the most recent, non-alias one from the list
-    if indices:
-        indices.pop(0)
-    # Remaining indices are older than the most recent two, delete them
-    click.echo(f"Found {len(indices)} older indices to delete.")
-
-    for index in indices:
+    click.echo(f"Current index: {result.current_index}")
+    click.echo(f"Found {len(result.deleted)} older indices to delete.")
+    for index in result.deleted:
         click.echo(f"Deleting index: {index}")
-        client.indices.delete(index=index)
-
     click.echo("Done.")

warehouse/search/__init__.py

@@ -14,7 +14,7 @@
 from warehouse.packaging.models import LifecycleStatus, Project, Release
 from warehouse.search.interfaces import ISearchService
 from warehouse.search.services import SearchService
-from warehouse.search.tasks import reindex
+from warehouse.search.tasks import delete_older_indices, reindex
 from warehouse.search.utils import get_index
 
 
@@ -112,5 +112,6 @@ def includeme(config):
     config.add_request_method(opensearch, name="opensearch", reify=True)
 
     config.add_periodic_task(crontab(minute=0, hour=6), reindex)
+    config.add_periodic_task(crontab(minute=0, hour=8), delete_older_indices)
 
     config.register_service_factory(SearchService.create_service, iface=ISearchService)