Method _run_loop

kubernetes/informer/informer.py:238–342

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 ar

(self)

        self._resource_version = rv or "0"

    def _run_loop(self):
        """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():
            # 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"] = self._resource_version
            # When a resync period is configured, set a matching server-side
            # watch timeout so that the stream exits after resync_period seconds
            # even if no events arrive.  Without this, a quiet period longer
            # than resync_period would never trigger a resync because the check
            # below only runs when the generator yields an event.
            if self._resync_period > 0:
                kw["timeout_seconds"] = max(1, int(self._resync_period))
            try:
                for event in self._watch.stream(self._list_func, **kw):
                    if self._stop_event.is_set():
                        break
                    evt_type = event.get("type")
                    obj = event.get("object")
                    # Sync the most recent resource version from the Watch
                    # instance (updated by unmarshal_event before yielding).
                    # Do this before firing handlers so consumers that wake on
                    # an event immediately see the advanced resource version.
                    if self._watch is not None and self._watch.resource_version:
                        self._resource_version = self._watch.resource_version
                    if evt_type == ADDED:
                        self._cache._put(obj)
                        self._fire(ADDED, obj)
                    elif evt_type == MODIFIED:
                        self._cache._put(obj)
                        self._fire(MODIFIED, obj)
                    elif evt_type == DELETED:
                        self._cache._remove(obj)
                        self._fire(DELETED, obj)
                    elif evt_type == BOOKMARK:
                        # BOOKMARK events carry an updated resource version but
                        # no object state change; the Watch instance already
                        # records the new resource_version internally.
                        self._fire(BOOKMARK, event.get("raw_object", obj))