more robust coll-index operations in operator and redis/kvrocks pod: (#3175)

- kvrocks: use main container instead of sidecar container to save index, to ensure it doesn't have timeout
- operator: add 'saved' state to determine when saving is actually complete
- operator: send custom signal instead of relying on sentinal file to indicate redis/kvrocks is done
- operator: track state of redis pod more accurately, by checking if containers completed successfully in operator

- load: don't attempt to overwrite kvrocks db if already exists
- load: download into temp dir first, then rename dir only if successful
- load: skip loading if remote file doesn't exist, nothing to do but reset index
- save: first upload to .tmp file, then rename to file destination (overwriting previous) only if successful
- save: bump max memory for save pod
- save: add upload concurrency to reduce memory consumption
- index import: set high backoff to ensure index jobs don't fail
- kvrocks: remove --appendonly flag, unsupported in kvrocks

Author: ikreymer

backend/btrixcloud/k8sapi.py

@@ -408,9 +408,9 @@ async def has_custom_jobs_with_label(self, plural, label) -> bool:
         except Exception:
             return False
 
-    async def send_signal_to_pod(self, pod_name, signame) -> bool:
+    async def send_signal_to_pod(self, pod_name, signame, container=None) -> bool:
         """send signal to all pods"""
-        command = ["bash", "-c", f"kill -s {signame} 1"]
+        command = ["sh", "-c", f"kill -s {signame} 1"]
         signaled = False
 
         try:
@@ -420,6 +420,7 @@ async def send_signal_to_pod(self, pod_name, signame) -> bool:
                 name=pod_name,
                 namespace=self.namespace,
                 command=command,
+                container=container,
                 stdout=True,
             )
             if res:

backend/btrixcloud/models.py

@@ -1593,7 +1593,7 @@ class CrawlOutWithResources(CrawlOut):
 ### COLLECTIONS ###
 
 TYPE_DEDUPE_INDEX_STATES = Literal[
-    "initing", "importing", "ready", "purging", "idle", "saving", "crawling"
+    "initing", "importing", "ready", "purging", "idle", "saving", "saved", "crawling"
 ]
 DEDUPE_INDEX_STATES = get_args(TYPE_DEDUPE_INDEX_STATES)
 

backend/btrixcloud/operator/collindexes.py

@@ -23,6 +23,17 @@
 from .baseoperator import BaseOperator
 
 
+# ============================================================================
+class IndexPodState(BaseModel):
+    """redis pod status"""
+
+    notFound: bool = False
+    running: bool = False
+    finished: bool = False
+    loaded: bool = False
+    savedAt: str = ""
+
+
 # ============================================================================
 class CollIndexStatus(BaseModel):
     """CollIndex Status"""
@@ -37,6 +48,9 @@ class CollIndexStatus(BaseModel):
 
     finishedAt: str = ""
 
+    # redis pod states
+    index: IndexPodState = IndexPodState()
+
 
 # ============================================================================
 class CollIndexSpec(BaseModel):
@@ -107,25 +121,25 @@ async def sync_index(self, data: MCSyncData):
         spec = CollIndexSpec(**data.parent.get("spec", {}))
         status = CollIndexStatus(**data.parent.get("status", {}))
 
-        coll_id = str(spec.id)
-        redis_name = "redis-coll-" + coll_id
+        coll_id = spec.id
+        redis_name = f"redis-coll-{coll_id}"
         new_children = []
 
         redis_pod = data.children[POD].get(redis_name)
 
         # check if redis should be skipped, eg. no pod active or complete
-        skip_redis = self.skip_redis(redis_pod)
+        self.sync_redis_pod_status(redis_pod, status)
 
         # allow deletion only if idle
         if data.finalizing:
             is_done = False
-            if status.state in ("idle", "saving") and not redis_pod:
-                # likely reentrant call still set to saving, just switch to idle
-                if status.state == "saving":
-                    status.state = "idle"
+            if status.state == "saved" and status.index.savedAt:
+                await self.set_state("idle", status, coll_id)
+                is_done = True
+            elif status.state == "idle" and status.index.notFound:
                 is_done = True
             # never inited, just remove
-            elif status.state == "initing" and skip_redis:
+            elif status.state == "initing" and status.index.notFound:
                 is_done = True
             else:
                 try:
@@ -154,15 +168,18 @@ async def sync_index(self, data: MCSyncData):
 
                 # Saving process
                 # 1. run bgsave while redis is active
-                if not skip_redis:
-                    await self.do_redis_save(spec.id, status)
+                if status.index.running:
+                    await self.do_save_redis(spec.id, status)
+
+                elif status.index.finished and not status.index.savedAt:
+                    await self.k8s.send_signal_to_pod(redis_name, "SIGUSR1", "save")
 
                 # 2. once redis has shutdown, check if fully finished
-                else:
-                    await self.check_redis_saved(redis_name, redis_pod, spec, status)
+                elif status.index.savedAt and status.index.savedAt != status.finishedAt:
+                    await self.mark_index_saved(redis_name, spec, status)
 
             else:
-                await self.update_state(skip_redis, data, spec.id, status)
+                await self.update_state(data, spec.id, status)
 
         # pylint: disable=broad-exception-caught
         except Exception as e:
@@ -187,32 +204,57 @@ async def sync_index(self, data: MCSyncData):
             "resyncAfterSeconds": resync_after,
         }
 
-    def skip_redis(self, pod):
+    def sync_redis_pod_status(self, pod, status: CollIndexStatus):
         """skip redis if no pod or redis container exited"""
         if not pod:
-            return True
+            status.index = IndexPodState(notFound=True)
+            return
 
-        if pod["status"].get("phase") != "Running":
-            return True
+        index = status.index
+        index.running = pod["status"].get("phase") == "Running"
 
+        terminated = None
         try:
-            if (
-                pod["status"]["containerStatuses"][0]["state"]["terminated"]["reason"]
-                == "Completed"
-            ):
-                return True
+            terminated = pod["status"]["containerStatuses"][0]["state"].get(
+                "terminated"
+            )
+            if terminated:
+                index.running = False
+                if terminated.get("reason") == "Completed":
+                    index.finished = True
         # pylint: disable=bare-except
         except:
             pass
 
-        return False
+        # redis pod likely running
+        if "initContainerStatuses" not in pod["status"]:
+            index.loaded = True
 
-    async def update_state(
-        self, skip_redis: bool, data, coll_id: UUID, status: CollIndexStatus
-    ):
+        else:
+            try:
+                index.loaded = (
+                    pod["status"]["initContainerStatuses"][0]["state"]["terminated"][
+                        "reason"
+                    ]
+                    == "Completed"
+                )
+            # pylint: disable=bare-except
+            except:
+                pass
+
+        if pod["status"].get("phase") == "Succeeded":
+            try:
+                index.savedAt = pod["status"]["containerStatuses"][1]["state"][
+                    "terminated"
+                ]["finishedAt"]
+            # pylint: disable=bare-except
+            except:
+                pass
+
+    async def update_state(self, data, coll_id: UUID, status: CollIndexStatus):
         """update state"""
         desired_state = status.state
-        if skip_redis:
+        if not status.index.loaded:
             desired_state = "initing"
 
         # first, handle any import or purge jobs
@@ -231,7 +273,7 @@ async def update_state(
             desired_state = "ready"
 
         # update stats if redis is available
-        if not skip_redis:
+        if status.index.running:
             await self.update_stats_from_redis(status, coll_id)
 
         if desired_state != status.state:
@@ -245,7 +287,7 @@ def is_expired(self, status: CollIndexStatus):
         if self.is_last_active_exceeds(status, self.idle_expire_time):
             return True
 
-        if status.state == "saving":
+        if status.state in ("saving", "saved"):
             return True
 
         return False
@@ -271,19 +313,19 @@ async def set_state(
         # self.run_task(self.coll_ops.update_dedupe_index_info(coll_id, state))
         await self.coll_ops.update_dedupe_index_info(coll_id, state)
 
-    async def do_delete(self, index_id: UUID):
+    async def do_delete(self, coll_id: UUID):
         """delete the CollIndex object"""
-        print(f"Deleting collindex {index_id}")
-        await self.k8s.delete_custom_object(f"collindex-{index_id}", "collindexes")
+        print(f"Deleting collindex {coll_id}")
+        await self.k8s.delete_custom_object(f"collindex-{coll_id}", "collindexes")
 
-    async def do_redis_save(self, coll_id: UUID, status: CollIndexStatus):
+    async def do_save_redis(self, coll_id: UUID, status: CollIndexStatus):
         """shutdown save redis"""
         try:
-            redis = await self.k8s.get_redis_connected("coll-" + str(coll_id))
+            redis = await self.k8s.get_redis_connected(f"coll-{coll_id}")
             if not redis:
                 return
 
-            if status.state != "saving":
+            if status.state not in ("saving", "saved"):
                 await redis.bgsave(False)
 
                 await self.set_state("saving", status, coll_id)
@@ -360,7 +402,7 @@ def get_related(self, data: MCBaseRequest):
 
     async def load_redis(
         self,
-        index_id: str,
+        coll_id: UUID,
         name: str,
         spec: CollIndexSpec,
         status: CollIndexStatus,
@@ -369,7 +411,7 @@ async def load_redis(
         params = {}
         params.update(self.shared_params)
         params["name"] = name
-        params["id"] = index_id
+        params["id"] = str(coll_id)
         params["init_redis"] = True
 
         params["load_dump"] = bool(status.indexLastSavedAt)
@@ -399,40 +441,25 @@ def get_index_storage_filename(self, coll_id: UUID, org: Organization):
         storage_path = org.storage.get_storage_extra_path(str(org.id))
         return storage_path + f"dedupe-index/{coll_id}"
 
-    async def check_redis_saved(
+    async def mark_index_saved(
         self,
         redis_name: str,
-        redis_pod,
         spec: CollIndexSpec,
         status: CollIndexStatus,
     ):
         """create sync job to save redis index data to s3 storage"""
 
-        if redis_pod and redis_pod["status"].get("phase") == "Succeeded":
-            finished_at = None
-            finished_at_str = ""
-            try:
-                finished_at_str = redis_pod["status"]["initContainerStatuses"][0][
-                    "state"
-                ]["terminated"]["finishedAt"]
-            # pylint: disable=bare-except
-            except:
-                pass
+        # update state immediately to speed up cleanup
+        print(f"Setting coll index state {status.state} -> saved {spec.id}")
+        status.state = "saved"
 
-            # update state immediately to speed up cleanup
-            print(f"Setting coll index state {status.state} -> idle {spec.id}")
-            status.state = "idle"
+        finished_at = str_to_date(status.index.savedAt)
 
-            if finished_at_str:
-                if status.finishedAt == finished_at_str:
-                    return
-                finished_at = str_to_date(finished_at_str)
-
-            await self.update_saved_dedupe_index_state_in_db(
-                spec.id, spec.oid, redis_name, finished_at or dt_now()
-            )
+        await self.update_saved_dedupe_index_state_in_db(
+            spec.id, spec.oid, redis_name, finished_at or dt_now()
+        )
 
-            status.finishedAt = finished_at_str
+        status.finishedAt = status.index.savedAt
 
     async def update_saved_dedupe_index_state_in_db(
         self, coll_id: UUID, oid: UUID, pod_name: str, finished_at: datetime.datetime

chart/app-templates/index-import-job.yaml

@@ -13,7 +13,9 @@ metadata:
 
 spec:
   ttlSecondsAfterFinished: 300
-  backoffLimit: 6
+
+  # keep retrying indefinitely
+  backoffLimit: 1000000
 
 # if purging, set completions to disallow changing parallelism
 # for import, omit to allow increasing later