Track and reuse resourceVersion across watch reconnects; re-list only on 410 Gone

Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com>

Author: Copilot

kubernetes/informer/informer.py

@@ -90,6 +90,7 @@ def __init__(
         self._watch = None
         self._thread = None
         self._stop_event = threading.Event()
+        self._resource_version = None  # most recent RV seen; None forces a full re-list
 
     # ---------------------------------------------------------------- #
     # Public API                                                        #
@@ -185,34 +186,40 @@ def _fire(self, event_type, obj):
     def _initial_list(self):
         """Do the initial list and populate the cache."""
         kw = self._build_kwargs()
-        resource_version = "0"
         resp = self._list_func(**kw)
         items = getattr(resp, "items", []) or []
         self._cache._replace_all(items)
         rv = None
         meta = getattr(resp, "metadata", None)
         if meta is not None:
             rv = getattr(meta, "resource_version", None)
-        if rv:
-            resource_version = rv
-        return resource_version
+        self._resource_version = rv or "0"
 
     def _run_loop(self):
-        """Background loop: list then watch, reconnect on errors."""
+        """Background loop: list then watch, reconnect on errors.
+
+        A full re-list is only performed when ``self._resource_version`` is
+        ``None`` (first start or after a 410 Gone response).  On all other
+        reconnects the most recent ``resourceVersion`` is reused so that no
+        events are missed and the API server does not need to send a full
+        object snapshot.
+        """
         while not self._stop_event.is_set():
-            try:
-                resource_version = self._initial_list()
-            except Exception as exc:
-                logger.exception("Error during initial list; retrying")
-                self._fire(ERROR, exc)
-                self._stop_event.wait(timeout=5)
-                continue
+            # Full re-list only when we have no resource version to resume from.
+            if self._resource_version is None:
+                try:
+                    self._initial_list()
+                except Exception as exc:
+                    logger.exception("Error during initial list; retrying")
+                    self._fire(ERROR, exc)
+                    self._stop_event.wait(timeout=5)
+                    continue
 
             # Watch loop
             last_resync = time.monotonic()
             self._watch = Watch()
             kw = self._build_kwargs()
-            kw["resource_version"] = resource_version
+            kw["resource_version"] = self._resource_version
             try:
                 for event in self._watch.stream(self._list_func, **kw):
                     if self._stop_event.is_set():
@@ -245,13 +252,30 @@ def _run_loop(self):
                             self._fire(MODIFIED, cached_obj)
                         last_resync = time.monotonic()
             except ApiException as exc:
-                logger.warning(
-                    "Watch stream ended with ApiException (status=%s); reconnecting",
-                    exc.status,
-                )
+                if exc.status == 410:
+                    # The stored resource version is too old; force a full re-list.
+                    logger.warning(
+                        "Watch expired (410 Gone); will re-list from scratch"
+                    )
+                    self._resource_version = None
+                else:
+                    logger.warning(
+                        "Watch stream ended with ApiException (status=%s); reconnecting",
+                        exc.status,
+                    )
                 self._fire(ERROR, exc)
             except Exception as exc:
                 logger.exception("Unexpected error in watch loop; reconnecting")
                 self._fire(ERROR, exc)
             finally:
+                # Capture the most recent resource version seen by the Watch
+                # (updated on every ADDED/MODIFIED/DELETED/BOOKMARK event) so
+                # that the next watch connection can resume without re-listing.
+                # Do not overwrite a None that was set by a 410 handler above.
+                if (
+                    self._resource_version is not None
+                    and self._watch is not None
+                    and self._watch.resource_version
+                ):
+                    self._resource_version = self._watch.resource_version
                 self._watch = None

kubernetes/test/test_informer.py

@@ -359,6 +359,113 @@ def fake_stream(func, **kw):
         self.assertEqual(len(cached), 1)
         self.assertIs(cached[0], pod)
 
+    def test_resource_version_stored_from_watch(self):
+        """After the watch stream ends the latest RV is preserved for reconnect."""
+        pod = _make_pod("default", "rv-pod")
+        events = [{"type": "ADDED", "object": pod}]
+
+        list_func = MagicMock()
+        list_resp = MagicMock()
+        list_resp.items = []
+        list_resp.metadata = MagicMock(resource_version="10")
+        list_func.return_value = list_resp
+
+        informer = SharedInformer(list_func=list_func)
+
+        call_count = {"n": 0}
+
+        with patch("kubernetes.informer.informer.Watch") as MockWatch:
+            mock_w = MagicMock()
+            mock_w.resource_version = "99"
+
+            def fake_stream(func, **kw):
+                call_count["n"] += 1
+                yield from events
+                informer._stop_event.set()
+
+            mock_w.stream.side_effect = fake_stream
+            MockWatch.return_value = mock_w
+
+            informer.start()
+            informer._thread.join(timeout=3)
+
+        # The Watch reported RV "99"; the informer should have stored it.
+        self.assertEqual(informer._resource_version, "99")
+        # list_func should have been called once for the initial list only.
+        self.assertEqual(list_func.call_count, 1)
+
+    def test_reconnect_skips_relist_when_rv_known(self):
+        """On reconnect without 410 the informer must NOT call the list function again."""
+        pod = _make_pod("default", "reconnect-pod")
+
+        list_func = MagicMock()
+        list_resp = MagicMock()
+        list_resp.items = [pod]
+        list_resp.metadata = MagicMock(resource_version="5")
+        list_func.return_value = list_resp
+
+        informer = SharedInformer(list_func=list_func)
+
+        stream_calls = {"n": 0}
+
+        with patch("kubernetes.informer.informer.Watch") as MockWatch:
+            mock_w = MagicMock()
+            mock_w.resource_version = "7"
+
+            def fake_stream(func, **kw):
+                stream_calls["n"] += 1
+                if stream_calls["n"] == 1:
+                    # First stream: yield nothing then let it reconnect
+                    return iter([])
+                # Second stream: stop the informer
+                informer._stop_event.set()
+                return iter([])
+
+            mock_w.stream.side_effect = fake_stream
+            MockWatch.return_value = mock_w
+
+            informer.start()
+            informer._thread.join(timeout=3)
+
+        # list_func is called only once (initial list); reconnect reuses the RV.
+        self.assertEqual(list_func.call_count, 1)
+        self.assertEqual(stream_calls["n"], 2)
+
+    def test_410_gone_triggers_relist(self):
+        """A 410 Gone ApiException must reset resource_version and trigger re-list."""
+        from kubernetes.client.exceptions import ApiException
+
+        list_func = MagicMock()
+        list_resp = MagicMock()
+        list_resp.items = []
+        list_resp.metadata = MagicMock(resource_version="3")
+        list_func.return_value = list_resp
+
+        informer = SharedInformer(list_func=list_func)
+
+        stream_calls = {"n": 0}
+
+        with patch("kubernetes.informer.informer.Watch") as MockWatch:
+            mock_w = MagicMock()
+            mock_w.resource_version = "3"
+
+            def fake_stream(func, **kw):
+                stream_calls["n"] += 1
+                if stream_calls["n"] == 1:
+                    raise ApiException(status=410, reason="Gone")
+                # Second stream (after re-list): stop cleanly
+                informer._stop_event.set()
+                return iter([])
+
+            mock_w.stream.side_effect = fake_stream
+            MockWatch.return_value = mock_w
+
+            informer.start()
+            informer._thread.join(timeout=3)
+
+        # list_func called twice: initial list + re-list after 410.
+        self.assertEqual(list_func.call_count, 2)
+
 
 if __name__ == "__main__":
     unittest.main()