fix: purging sessions from the cache too early (#885)

Author: olevski

components/renku_data_services/k8s/clients.py

@@ -351,8 +351,12 @@ async def patch(self, meta: K8sObjectMeta, patch: dict[str, Any] | list[dict[str
     async def delete(self, meta: K8sObjectMeta) -> None:
         """Delete a k8s object."""
         await super().delete(meta)
-        if meta.gvk in self.__kinds_to_cache:
-            await self.__cache.delete(meta)
+        # NOTE: We use foreground deletion in the k8s client.
+        # This means that the parent resource is usually not deleted immediately and will
+        # wait for its children to be deleted before it is deleted.
+        # To avoid premature purging of resources from the cache we do not delete the resource here
+        # from the cache, rather we expect that the cache will sync itself properly and quickly purge
+        # stale resources.
 
     async def get(self, meta: K8sObjectMeta) -> K8sObject | None:
         """Get a specific k8s object, None is returned if the object does not exist."""

components/renku_data_services/k8s_watcher/core.py

@@ -41,40 +41,60 @@ def __init__(
         db_cache: K8sDbCache,
     ) -> None:
         self.__handler = handler
-        self.__tasks: dict[ClusterId, list[Task]] | None = None
+        self.__watch_tasks: dict[ClusterId, list[Task]] = {}
+        self.__full_sync_tasks: dict[ClusterId, Task] = {}
+        self.__full_sync_times: dict[ClusterId, datetime] = {}
+        self.__full_sync_running: set[ClusterId] = set()
         self.__kinds = kinds
         self.__clusters = clusters
-        self.__sync_period_seconds = 1800
+        self.__sync_period_seconds = 600
         self.__cache = db_cache
 
-    async def __sync(self) -> None:
+    async def __sync(self, cluster: Cluster, kind: GVK) -> None:
         """Upsert K8s objects in the cache and remove deleted objects from the cache."""
+        clnt = K8sClusterClient(cluster)
+        fltr = K8sObjectFilter(gvk=kind, namespace=cluster.namespace)
+        # Upsert new / updated objects
+        objects_in_k8s: dict[str, K8sObject] = {}
+        async for obj in clnt.list(fltr):
+            objects_in_k8s[obj.name] = obj
+            await self.__cache.upsert(obj)
+        # Remove objects that have been deleted from k8s but are still in cache
+        async for cache_obj in self.__cache.list(fltr):
+            cache_obj_is_in_k8s = objects_in_k8s.get(cache_obj.name) is not None
+            if cache_obj_is_in_k8s:
+                continue
+            await self.__cache.delete(cache_obj.meta)
+
+    async def __full_sync(self, cluster: Cluster) -> None:
+        """Run the full sync if it has never run or at the required interval."""
+        last_sync = self.__full_sync_times.get(cluster.id)
+        since_last_sync = datetime.now() - last_sync if last_sync is not None else None
+        if since_last_sync is not None and since_last_sync.total_seconds() < self.__sync_period_seconds:
+            return
+        self.__full_sync_running.add(cluster.id)
         for kind in self.__kinds:
-            for cluster in self.__clusters.values():
-                clnt = K8sClusterClient(cluster)
-                fltr = K8sObjectFilter(gvk=kind, namespace=cluster.namespace)
-                # Upsert new / updated objects
-                objects_in_k8s: dict[str, K8sObject] = {}
-                async for obj in clnt.list(fltr):
-                    objects_in_k8s[obj.name] = obj
-                    await self.__cache.upsert(obj)
-                # Remove objects that have been deleted from k8s but are still in cache
-                async for cache_obj in self.__cache.list(fltr):
-                    cache_obj_is_in_k8s = objects_in_k8s.get(cache_obj.name) is not None
-                    if cache_obj_is_in_k8s:
-                        continue
-                    await self.__cache.delete(cache_obj.meta)
+            logger.info(f"Starting full k8s cache sync for cluster {cluster} and kind {kind}")
+            await self.__sync(cluster, kind)
+        self.__full_sync_times[cluster.id] = datetime.now()
+        self.__full_sync_running.remove(cluster.id)
+
+    async def __periodic_full_sync(self, cluster: Cluster) -> None:
+        """Keeps trying to run the full sync."""
+        while True:
+            await self.__full_sync(cluster)
+            await asyncio.sleep(self.__sync_period_seconds / 10)
 
     async def __watch_kind(self, kind: GVK, cluster: Cluster) -> None:
-        last_sync: datetime | None = None
         while True:
             try:
-                if last_sync is None or (datetime.now() - last_sync).total_seconds() >= self.__sync_period_seconds:
-                    logger.info("Starting full k8s cache sync")
-                    await self.__sync()
-                    last_sync = datetime.now()
                 watch = cluster.api.async_watch(kind=kind.kr8s_kind, namespace=cluster.namespace)
                 async for event_type, obj in watch:
+                    while cluster.id in self.__full_sync_running:
+                        logger.info(
+                            f"Pausing k8s watch event processing for cluster {cluster} until full sync completes"
+                        )
+                        await asyncio.sleep(5)
                     await self.__handler(cluster.with_api_object(obj), event_type)
                     # in some cases, the kr8s loop above just never yields, especially if there's exceptions which
                     # can bypass async scheduling. This sleep here is as a last line of defence so this code does not
@@ -97,36 +117,41 @@ def __run_single(self, cluster: Cluster) -> list[Task]:
 
     async def start(self) -> None:
         """Start the watcher."""
-        if self.__tasks is None:
-            self.__tasks = {}
-        for cluster in self.__clusters.values():
-            self.__tasks[cluster.id] = self.__run_single(cluster)
+        for cluster in sorted(self.__clusters.values(), key=lambda x: x.id):
+            await self.__full_sync(cluster)
+            self.__full_sync_tasks[cluster.id] = asyncio.create_task(self.__periodic_full_sync(cluster))
+            self.__watch_tasks[cluster.id] = self.__run_single(cluster)
 
     async def wait(self) -> None:
         """Wait for all tasks.
 
         This is mainly used to block the main function.
         """
-        if self.__tasks is None:
-            return
-        await asyncio.gather(*[t for tl in self.__tasks.values() for t in tl])
+        all_tasks = list(self.__full_sync_tasks.values())
+        for tasks in self.__watch_tasks.values():
+            all_tasks.extend(tasks)
+        await asyncio.gather(*all_tasks)
 
     async def stop(self, timeout: timedelta = timedelta(seconds=10)) -> None:
         """Stop the watcher or timeout."""
-        if self.__tasks is None:
-            return
-        for task_list in self.__tasks.values():
+
+        async def stop_task(task: Task, timeout: timedelta) -> None:
+            if task.done():
+                return
+            task.cancel()
+            try:
+                async with asyncio.timeout(timeout.total_seconds()):
+                    with contextlib.suppress(CancelledError):
+                        await task
+            except TimeoutError:
+                logger.error("timeout trying to cancel k8s watcher task")
+                return
+
+        for task_list in self.__watch_tasks.values():
             for task in task_list:
-                if task.done():
-                    continue
-                task.cancel()
-                try:
-                    async with asyncio.timeout(timeout.total_seconds()):
-                        with contextlib.suppress(CancelledError):
-                            await task
-                except TimeoutError:
-                    logger.error("timeout trying to cancel k8s watcher task")
-                    continue
+                await stop_task(task, timeout)
+        for task in self.__full_sync_tasks.values():
+            await stop_task(task, timeout)
 
 
 async def collect_metrics(