Streaming paginated items when using list under watch api

[echelonh]

I'm using kube::runtime::watcher to watch certain resources in my cluster.

It works great, but I suspect that the initial list stage causes a rather big memory spike when dealing with a large amount of objects (e.g. 500+). I think that what happens is that the list action tries to list+deserialize all the objects first, and then provide them to the iterator after the fact. The data itself, plus the deserialization effort take their toll on the process memory and we see a spike.

I tried to set page_size to relatively low numbers (10?), but it didn't seem to affect the memory usage pattern. It's possible that I don't consume fast enough or that I'm doing something wrong though.

This is only my assesment, so I would love to confirm this suspicion with someone here who's more familiar with the code.
In addition - I wanted to know if ask if there's a possible workaround, or ask if list could be changed to stream the objects one-by-one instead of getting all of them first.

Thanks a lot!

[echelonh]

So, I confirmed the issue is still happening, this time armed with a bit more insights -
https://github.com/kube-rs/kube/blob/main/kube-runtime/src/watcher.rs#L474-L494
We use an object vec and then objects.extend per each page iterattion and yield all the objects once the list stage is complete.

While it is technically correct, iiuc one of the leading reasons for the k8s pagination mechanism is to be able to list objects while keeping a low memory footprint. This is exactly what I'm trying to achieve 🙃 . Unfortunately this is bad news when dealing with a big amount of objects (I crossed 1gb barrier, release build and all.. 😅 )

I'm going to try and open a PR, but I'm not too fluent with the inners of kube-rs. If anyone wants to take the lead, code review me or collaborate - I'll be more than happy.

Thanks!

[clux]

Hi there. Yeah, you are correct here that there are work that could be done here to improve pagination and the way it we allocate and hand out chunks (currently we stack up pages internally until we can send out a complete watcher::Event. This is done so that reflector's stores can be swapped out atomically (i.e. there's never any time the store is half-full), and handing out pages directly can undermine that property.

We know this is not an optimal solution, but it is a local maximum in the sense that it maintains atomicity while not demanding the same huge memory requirements from the apiserver (who would also be batching up the list without paging).

The best solution imo is moving to the already supported 1.27 streaming lists feature that drops list chunking altogether and can be turned on via the watcher config's parameter InitialListStrategy if your cluster is new enough.

You can try it out with the pod_watcher example.

[echelonh]

Hi, thanks for the quick reply!

I didn't account for the reflector, but now I understand why that particular piece of code was written the way it is.

Unfortunately, using streaming lists is currently not an option for me. I actually looked into this particular feature, and naturally preferred to use it, but I realized only too late that this feature is still off by default for must k8s deployments (link).

Going forward, I can solve the problem locally for my needs but I'd love to try and contribute still -

I'm guessing one approach would be some WatchConfig value stating whether to stream the initial list or aggregate it. This way the reflector uses the aggregation option and the user can decide on his own. Another alternative is to let the user the option to watch without any initial listing and take care of the listing stage on its own. I think the first option is better.

Any other ideas or thoughts?

[clux]

Hm, I understand.

My original plan here was to move entirely to streaming lists and remove paging entirely once streaming lists is supported everywhere; i.e. when it's enabled by default in our MK8SV. But given that this would not be before 1.35 is out if it is enabled by default in 1.30, it would be nice to have an improvement for people that can't use it yet (2 years is a long time).

You could possibly (along the lines of your first suggestion) try to extend watcher::Event with a Page(Vec<...>) and PageLast(Vec<...>) event, expose those events from watcher and then intercept them in the Store instead in:

kube/kube-runtime/src/reflector/store.rs

Lines 60 to 84 in c7054b5

	/// Applies a single watcher event to the store
	pub fn apply_watcher_event(&mut self, event: &watcher::Event<K>) {
	match event {
	watcher::Event::Applied(obj) => {
	let key = ObjectRef::from_obj_with(obj, self.dyntype.clone());
	let obj = Arc::new(obj.clone());
	self.store.write().insert(key, obj);
	}
	watcher::Event::Deleted(obj) => {
	let key = ObjectRef::from_obj_with(obj, self.dyntype.clone());
	self.store.write().remove(&key);
	}
	watcher::Event::Restarted(new_objs) => {
	let new_objs = new_objs
	.iter()
	.map(|obj| {
	(
	ObjectRef::from_obj_with(obj, self.dyntype.clone()),
	Arc::new(obj.clone()),
	)
	})
	.collect::<AHashMap<_, _>>();
	*self.store.write() = new_objs;
	}
	}

and move the caching there (where it is actually needed - then later we could maybe make atomicity optional). It might require modifying some WatchStreamExt methods that relies on the watcher::Event and more, but it's also possible that is not going to be too bad. Feel free to have a try!