`Arc` wrap `watcher` and/or `Controller` stream interfaces

[clux]

Would you like to work on this feature?

Maybe

What problem are you trying to solve?

As a final step to allow stream sharing #1080 we need to allow StreamExt::stream_subscribe output to be able to be passed into the controller streams interfaces:

• Controller::for_stream
• Controller::watches_stream
• Controller::owns_stream

Which currently is not possible because StreamExt::stream_subscribe produces Stream<Item = Result<Arc<K>, Error>>, but the input streams are normal watche streams of the form Stream<Item = Result<K, Error>>.

Arc wrapping can help reduce memory consumption + reduce awkward arc wrap points:

• reflector stores arc wrap already by cloning objects passed through from streams
• we already have to arc-wrap in Controller before calling out to the reconcile and error_policy fns anyway

Describe the solution you'd like

We need the streams that watcher produces to ultimately end up with equal types for both the subscribepath and the non-subscribe path. Options are considered below, but personally, I think that if it's viable, we should experiment with arc-wrapping earlier.

Describe alternatives you've considered

1. Always Arc wrap output from EventFlatten

Arc wrap once it's intended to be consumed; when it's flattened (::applied_objects() or ::touched_objects()). It's late enough that it won't really affect any store implementations, and it will work with ::stream_subscribe.

While this is less breaking than the suggestion below, it also creates another copy of the object; as each Store is cloning: https://github.com/kube-rs/kube/blob/c30b376a3a79dbfd04d5e4354d4e5a5324bf2596/kube-runtime/src/reflector/store.rs#L54-L69

2. Always Arc wrap output from watcher

A much more fundamental change to our api, but it will provide a much more consistent api, and possibly clone less huge k8s objects while passing through streams.

If we do it this deep, then we basically add it to step_trampolined where the objects are handled first: https://github.com/kube-rs/kube/blob/c30b376a3a79dbfd04d5e4354d4e5a5324bf2596/kube-runtime/src/watcher.rs#L371-L472

This has the potential to help reduce cloning between streams and reflectors. Given reflector stores and Kubernetes objects account for most of the memory usage of typical rust controllers, this could be a worthwhile avenue.

However, not sure if this is a practical solution.

3. Internally Arc with Arc wrapper helper for WatchStreamExt

1. Create a new WatchStreamExt::arc (or something like that) to map the Ok K elements to an Arc<K>.
2. Internally in Controller, call ::arc on all managed streams
3. Tell users using the unstable streams interface that they need an extra ::arc() call to be compatible

Fairly simple. Just propagate the Arc inside Controller only, and do it at an earlier point.

This does not solve memory issues with Store, and it puts the onus on the user to figure out when to arc-wrap or not.

Documentation, Adoption, Migration Strategy

Examples, doc tests, kube.rs controller guide updates where relevant. I suspect this can be done in a mostly non-disruptive way. We have arc-wrapped a bunch of things in the past, and it's generally been a good idea.

Target crate for feature

kube-runtime