async def _watch_and_update(self):
    "\n        Keeps the current list of resources up-to-date\n\n        We first fetch the list of current resources, and store that. Then we\n        register to be notified of changes to those resources, and keep our\n        local store up-to-date based on these notifications.\n\n        We also perform exponential backoff, giving up after we hit 32s\n        wait time. This should protect against network connections dropping\n        and intermittent unavailability of the api-server. Every time we\n        recover from an exception we also do a full fetch, to pick up\n        changes that might've been missed in the time we were not doing\n        a watch.\n\n        Since the resources are read-only in the Spawner (where they are\n        used), then this is safe.  The Spawner's view of the world might be\n        out-of-date, but it's not going to corrupt any data.\n        "
    selectors = []
    if self.label_selector:
        selectors.append(('label selector=%r' % self.label_selector))
    if self.field_selector:
        selectors.append(('field selector=%r' % self.field_selector))
    log_selector = ', '.join(selectors)
    cur_delay = 0.1
    if self.omit_namespace:
        ns_str = 'all namespaces'
    else:
        ns_str = 'namespace {}'.format(self.namespace)
    self.log.info('watching for %s with %s in %s', self.kind, log_selector, ns_str)
    while True:
        self.log.debug('Connecting %s watcher', self.kind)
        start = time.monotonic()
        w = watch.Watch()
        try:
            resource_version = (await self._list_and_update())
            watch_args = {'label_selector': self.label_selector, 'field_selector': self.field_selector, 'resource_version': resource_version}
            if (not self.omit_namespace):
                watch_args['namespace'] = self.namespace
            if self.request_timeout:
                watch_args['_request_timeout'] = self.request_timeout
            if self.timeout_seconds:
                watch_args['timeout_seconds'] = self.timeout_seconds
            method = partial(getattr(self.api, self.list_method_name), _preload_content=False)
            async with w.stream(method, **watch_args) as stream:
                async for watch_event in stream:
                    cur_delay = 0.1
                    resource = watch_event['raw_object']
                    ref_key = '{}/{}'.format(resource['metadata']['namespace'], resource['metadata']['name'])
                    if (watch_event['type'] == 'DELETED'):
                        self.resources.pop(ref_key, None)
                    else:
                        self.resources[ref_key] = resource
                    if self._stopping:
                        self.log.info('%s watcher stopped: inner', self.kind)
                        break
                    watch_duration = (time.monotonic() - start)
                    if (watch_duration >= self.restart_seconds):
                        self.log.debug('Restarting %s watcher after %i seconds', self.kind, watch_duration)
                        break
        except ReadTimeoutError:
            self.log.warning('Read timeout watching %s, reconnecting', self.kind)
            continue
        except asyncio.CancelledError:
            self.log.debug('Cancelled watching %s', self.kind)
            raise
        except Exception:
            cur_delay = (cur_delay * 2)
            if (cur_delay > 30):
                self.log.exception('Watching resources never recovered, giving up')
                if self.on_failure:
                    self.on_failure()
                return
            self.log.exception('Error when watching resources, retrying in %ss', cur_delay)
            (await asyncio.sleep(cur_delay))
            continue
        else:
            self.log.debug('%s watcher timeout', self.kind)
        finally:
            w.stop()
            if self._stopping:
                self.log.info('%s watcher stopped: outer', self.kind)
                break
    self.log.warning('%s watcher finished', self.kind)
Keeps the current list of resources up-to-date

We first fetch the list of current resources, and store that. Then we
register to be notified of changes to those resources, and keep our
local store up-to-date based on these notifications.

We also perform exponential backoff, giving up after we hit 32s
wait time. This should protect against network connections dropping
and intermittent unavailability of the api-server. Every time we
recover from an exception we also do a full fetch, to pick up
changes that might've been missed in the time we were not doing
a watch.

Since the resources are read-only in the Spawner (where they are
used), then this is safe.  The Spawner's view of the world might be
out-of-date, but it's not going to corrupt any data.
async def _watch_and_update(self):
    "\n        Keeps the current list of resources up-to-date\n\n        We first fetch the list of current resources, and store that. Then we\n        register to be notified of changes to those resources, and keep our\n        local store up-to-date based on these notifications.\n\n        We also perform exponential backoff, giving up after we hit 32s\n        wait time. This should protect against network connections dropping\n        and intermittent unavailability of the api-server. Every time we\n        recover from an exception we also do a full fetch, to pick up\n        changes that might've been missed in the time we were not doing\n        a watch.\n\n        Since the resources are read-only in the Spawner (where they are\n        used), then this is safe.  The Spawner's view of the world might be\n        out-of-date, but it's not going to corrupt any data.\n        "
    selectors = []
    if self.label_selector:
        selectors.append(('label selector=%r' % self.label_selector))
    if self.field_selector:
        selectors.append(('field selector=%r' % self.field_selector))
    log_selector = ', '.join(selectors)
    cur_delay = 0.1
    if self.omit_namespace:
        ns_str = 'all namespaces'
    else:
        ns_str = 'namespace {}'.format(self.namespace)
    self.log.info('watching for %s with %s in %s', self.kind, log_selector, ns_str)
    while True:
        self.log.debug('Connecting %s watcher', self.kind)
        start = time.monotonic()
        w = watch.Watch()
        try:
            resource_version = (await self._list_and_update())
            watch_args = {'label_selector': self.label_selector, 'field_selector': self.field_selector, 'resource_version': resource_version}
            if (not self.omit_namespace):
                watch_args['namespace'] = self.namespace
            if self.request_timeout:
                watch_args['_request_timeout'] = self.request_timeout
            if self.timeout_seconds:
                watch_args['timeout_seconds'] = self.timeout_seconds
            method = partial(getattr(self.api, self.list_method_name), _preload_content=False)
            async with w.stream(method, **watch_args) as stream:
                async for watch_event in stream:
                    cur_delay = 0.1
                    resource = watch_event['raw_object']
                    ref_key = '{}/{}'.format(resource['metadata']['namespace'], resource['metadata']['name'])
                    if (watch_event['type'] == 'DELETED'):
                        self.resources.pop(ref_key, None)
                    else:
                        self.resources[ref_key] = resource
                    if self._stopping:
                        self.log.info('%s watcher stopped: inner', self.kind)
                        break
                    watch_duration = (time.monotonic() - start)
                    if (watch_duration >= self.restart_seconds):
                        self.log.debug('Restarting %s watcher after %i seconds', self.kind, watch_duration)
                        break
        except ReadTimeoutError:
            self.log.warning('Read timeout watching %s, reconnecting', self.kind)
            continue
        except asyncio.CancelledError:
            self.log.debug('Cancelled watching %s', self.kind)
            raise
        except Exception:
            cur_delay = (cur_delay * 2)
            if (cur_delay > 30):
                self.log.exception('Watching resources never recovered, giving up')
                if self.on_failure:
                    self.on_failure()
                return
            self.log.exception('Error when watching resources, retrying in %ss', cur_delay)
            (await asyncio.sleep(cur_delay))
            continue
        else:
            self.log.debug('%s watcher timeout', self.kind)
        finally:
            w.stop()
            if self._stopping:
                self.log.info('%s watcher stopped: outer', self.kind)
                break
    self.log.warning('%s watcher finished', self.kind)