simplify deletion of dedupe index: (#3200)

- if index not used by a crawl, allow fast deletion
- delete any index jobs and the collindex object right away
- ensure index state not reset to 'idle' if empty/marked for deletion

---------

Co-authored-by: Tessa Walsh <tessa@bitarchivist.net>

Author: ikreymer

backend/btrixcloud/colls.py

@@ -803,9 +803,11 @@ async def delete_dedupe_index(
         if not coll.indexState:
             raise HTTPException(status_code=404, detail="no_dedupe_index")
 
-        # if index is not idle, can't delete it yet
-        if coll.indexState != "idle":
-            raise HTTPException(status_code=400, detail="dedupe_index_is_in_use")
+        # if index is not idle/ready, check if any crawls running
+        if coll.indexState not in ("idle", "ready"):
+            # if crawls running using index, can't delete
+            if await self.crawl_ops.has_active_crawls_with_dedupe_coll(org.id, coll.id):
+                raise HTTPException(status_code=400, detail="dedupe_index_is_in_use")
 
         if coll.indexFile:
             if not await self.storage_ops.delete_file_object(org, coll.indexFile):
@@ -816,17 +818,24 @@ async def delete_dedupe_index(
                 raise HTTPException(status_code=400, detail="file_deletion_error")
 
         await self.collections.find_one_and_update(
-            {"_id": coll.id, "indexState": "idle"},
+            {"_id": coll.id},
             {
                 "$set": {
                     "indexStats": None,
                     "indexState": None,
                     "indexFile": None,
                     "indexLastSavedAt": None,
+                    "indexDiskSpaceUsed": None,
                 }
             },
         )
 
+        # if not idle, delete k8s dedupe resources
+        if coll.indexState != "idle":
+            await self.crawl_manager.delete_dedupe_index_resources(
+                str(org.id), str(coll.id)
+            )
+
         if remove_from_workflows:
             await self.crawl_configs.update_many(
                 {"oid": org.id, "dedupeCollId": coll.id},
@@ -840,7 +849,7 @@ async def update_dedupe_index_stats(
     ):
         """update dedupe index stats for specified collection"""
         self.collections.find_one_and_update(
-            {"_id": coll_id},
+            {"_id": coll_id, "indexState": {"$ne": None}},
             {
                 "$set": {
                     "indexStats": stats.dict(),
@@ -855,14 +864,20 @@ async def update_dedupe_index_info(
         state: TYPE_DEDUPE_INDEX_STATES,
         index_file: Optional[DedupeIndexFile] = None,
         dt: Optional[datetime] = None,
+        if_exists=False,
     ):
         """update the state, and optionally, dedupe index file info"""
         query: dict[str, Any] = {"indexState": state}
         if index_file and dt:
             query["indexLastSavedAt"] = dt
             query["indexFile"] = index_file.model_dump()
 
-        res = self.collections.find_one_and_update({"_id": coll_id}, {"$set": query})
+        match: dict[str, Any] = {"_id": coll_id}
+        # only update if index already exists
+        if if_exists:
+            match["indexState"] = {"$ne": None}
+
+        res = self.collections.find_one_and_update(match, {"$set": query})
         return res is not None
 
     async def get_dedupe_index_saved(self, coll_id: UUID) -> Optional[datetime]:
@@ -886,6 +901,13 @@ async def get_dedupe_index_disk_size(self, coll_id: UUID) -> int:
             return coll.get("indexDiskSpaceUsed", 0)
         return 0
 
+    async def has_dedupe_index(self, coll_id: UUID, oid: UUID) -> bool:
+        """return true if collection exists and indexState is set on collection"""
+        coll = await self.collections.find_one(
+            {"_id": coll_id, "oid": oid}, projection={"indexState"}
+        )
+        return coll and coll.get("indexState") is not None
+
     # END DEDUPE OPS
 
     async def recalculate_org_collection_stats(self, org: Organization):

backend/btrixcloud/crawlmanager.py

@@ -269,6 +269,12 @@ async def run_index_import_job(
 
         return name
 
+    async def delete_dedupe_index_resources(self, oid: str, coll_id: str) -> None:
+        """Delete dedupe index-related jobs and index itself"""
+        await self._delete_jobs(f"role=index-import-job,oid={oid},coll={coll_id}")
+
+        await self.delete_custom_object(f"collindex-{coll_id}", "collindexes")
+
     async def ensure_cleanup_seed_file_cron_job_exists(self):
         """ensure cron background job to clean up unused seed files weekly exists"""
 

backend/btrixcloud/crawls.py

@@ -412,6 +412,19 @@ async def get_active_crawls_pending_size(self, oid: UUID) -> int:
 
         return results[0].get("totalSum") or 0
 
+    async def has_active_crawls_with_dedupe_coll(
+        self, oid: UUID, coll_id: UUID
+    ) -> bool:
+        """return true/false if any active crawls exist that use given coll_id for dedupe"""
+        res = await self.crawls.find_one(
+            {
+                "state": {"$in": RUNNING_AND_WAITING_STATES},
+                "oid": oid,
+                "dedupeCollId": coll_id,
+            }
+        )
+        return bool(res)
+
     async def delete_crawls(
         self,
         org: Organization,

backend/btrixcloud/k8sapi.py

@@ -461,6 +461,14 @@ async def list_crawl_jobs(self, label: str = "") -> List[dict[str, Any]]:
         )
         return resp.get("items", [])
 
+    async def _delete_jobs(self, label: str) -> None:
+        """Delete namespaced jobs"""
+        await self.batch_api.delete_collection_namespaced_job(
+            namespace=self.namespace,
+            label_selector=label,
+            propagation_policy="Foreground",
+        )
+
     async def _delete_cron_jobs(self, label: str) -> None:
         """Delete namespaced cron jobs (e.g. crawl configs, bg jobs)"""
         await self.batch_api.delete_collection_namespaced_cron_job(

backend/btrixcloud/models.py

@@ -1720,7 +1720,7 @@ class Collection(BaseMongoModel):
     indexStats: Optional[DedupeIndexStats] = None
 
     # size of db on disk when in use
-    indexDiskSpaceUsed: int = 0
+    indexDiskSpaceUsed: Optional[int] = None
 
 
 # ============================================================================

backend/btrixcloud/operator/baseoperator.py

@@ -4,6 +4,7 @@
 import os
 import json
 from typing import TYPE_CHECKING, Any
+from uuid import UUID
 from kubernetes.utils import parse_quantity
 
 import yaml
@@ -238,7 +239,9 @@ async def ensure_coll_index_ready(
 
         # if index not found, create it
         if not found:
-            await self.k8s.create_or_update_coll_index(coll_id, oid)
+            # ensure dedupe index exists
+            if await self.coll_ops.has_dedupe_index(UUID(coll_id), UUID(oid)):
+                await self.k8s.create_or_update_coll_index(coll_id, oid)
 
         return False
 

backend/btrixcloud/operator/collindexes.py

@@ -140,7 +140,7 @@ async def mc_finalize_index(data: MCSyncData):
         async def mc_related(data: MCBaseRequest):
             return self.get_related(data)
 
-    # pylint: disable=too-many-locals, too-many-branches
+    # pylint: disable=too-many-locals, too-many-branches, too-many-statements
     async def sync_index(self, data: MCSyncData):
         """sync CollIndex object with existing state"""
         spec = CollIndexSpec(**data.parent.get("spec", {}))
@@ -168,15 +168,28 @@ async def sync_index(self, data: MCSyncData):
                 is_done = True
             else:
                 try:
-                    await self.coll_ops.get_collection_raw(spec.id, spec.oid)
+                    coll = await self.coll_ops.get_collection_raw(spec.id, spec.oid)
+                    # if index state is not set, index has been deleted
+                    # also delete immediately
+                    if not coll.get("indexState"):
+                        # if pod still exists, send SIGUSR2 to exit immediately
+                        if redis_pod:
+                            await self.k8s.send_signal_to_pod(
+                                redis_name, "SIGUSR2", "save"
+                            )
+                        is_done = True
                 # pylint: disable=bare-except
                 except:
                     # collection not found, delete index
                     is_done = True
 
             if is_done:
                 print(f"CollIndex removed: {spec.id}")
-                return {"status": status.dict(), "children": [], "finalized": True}
+                return {
+                    "status": status.dict(),
+                    "children": [],
+                    "finalized": not redis_pod,
+                }
 
         try:
             # determine if index was previously saved before initing redis
@@ -337,8 +350,7 @@ async def set_state(
         status.state = state
         status.lastStateChangeAt = date_to_str(dt_now())
 
-        # self.run_task(self.coll_ops.update_dedupe_index_info(coll_id, state))
-        await self.coll_ops.update_dedupe_index_info(coll_id, state)
+        await self.coll_ops.update_dedupe_index_info(coll_id, state, if_exists=True)
 
     async def do_delete(self, coll_id: UUID):
         """delete the CollIndex object"""

chart/app-templates/redis.yaml

@@ -152,6 +152,9 @@ spec:
       command: ["sh", "-c"]
       args:
         - |
+          # exit on SIGUSR2
+          trap 'echo "SIGUSR2, aborting save" && exit 0' SIGUSR2
+
           # wait for SIGUSR1
           trap ':' SIGUSR1