Add shared stream interfaces

[mateiidavid]

Motivation

Stream sharing has been missing from the library for some time now. We've had a few attempts at making it happen, e.g. new controller interfaces (such as Controller::for_stream) and a stream subscriber adapter. #1189 describes the original issue we sought to solve before shipping shared streams: getting all of the interfaces to play well together.

This change is a departure from the original issue described in #1189. After some exploration, it became a bit clearer that the stream subscriber interface may not be the best primitive to use:

• Using a broadcast channel that cannot do backpressure forces us to shed load (i.e. drop events when a subscriber isn't being driven)
• Sharing entire events is probably not ideal; it complicates the API without providing much added value.

This PR (and most of the exploration done as part of it) aims to address these two issues by changing and adding new interfaces.

Solution

Broadly speaking:

1. Instead of sharing events, we share objects. The way this has been envisioned to work is by having the runtime (async executor) own the "main" watch on a specific resource. For example, a task that polls and handles errors is submitted to the scheduler. The stream propagates inner objects to readers.
2. Readers (i.e. streams that yield shared objects) can be piped into controllers through a new interface to decouple the shared stream implementation from other interfaces that are used to thread events through a controller's applier loop.

As a general rule of thumb, the solution here should be backwards compatible and avoid breaking current functionality. For now, it made more sense to introduce more duplication to allow for more flexibility in the future to change the current implementation (or in other words this should be marked as experimental).

More focused:

1. Streams now "flow" from a store, since reflectors will always cache an object by putting it behind an arc. The writer side includes:

• A broadcast sender.
• An inactive receiver.
• A new method to create a store that can be used for shared streams (it takes in a buffering factor).
• A method to "subscribe" to changes.
• A stream impl. The stream impl is the same as before; we take ownership of a stream that yields a watch result and apply the object. Now, we also share a reference to the object (i.e. an ObjectRef) on the broadcast channel.

2. A ReflectHandle contains a store reader and a broadcast receiver. It implements Stream and yields K-typed objects. The stream implementation reads a reference from the broadcast channel and retrieves the object from the store.

Contract:

1. To create a shareable stream reflector, the store must be created with new_with_dispatch, otherwise the broadcast channel will never be used.
2. Backpressure is applied always when there is no capacity on the broadcast channel. Note: an improperly sized buffer can have consequences and implications to parallelism (since the main stream will not receive new events until there's space on the broadcast channel).
3. At least one subscriber must be created up-front, before the store is passed to reflect(). More subscribers can be created after by cloning the initial handle.
4. Subscribers must be polled to avoid deadlocking.
5. Controllers must use a concrete interface to use a shared stream (Controller::for_shared_stream).

Potential future work:

• Adapt example to include owned objects.
• Ensure an interface exists to use owned objects (Controller::for_owned_shared, naming TBD ofc).
• Potentially merge dispatch_event with apply_watcher_event

Feedback welcome! Especially if we can simplify the current implementation. Some things may be out of place for now. I think in the spirit of keeping this as flexible as possible, my recommendation would be to duplicate code until we can get this in the hands of people that have more niche use cases or that can test it out at scale.

There's a big commit history for people curious on how this evolved over time. Ultimately settled on some approaches suggested in #1426.

[codecov]

Codecov Report

Attention: Patch coverage is 90.00000% with 18 lines in your changes are missing coverage. Please review.

Project coverage is 74.7%. Comparing base (c1df726) to head (eca6be1).

Additional details and impacted files

@@           Coverage Diff           @@
##            main   #1449     +/-   ##
=======================================
+ Coverage   74.3%   74.7%   +0.5%     
=======================================
  Files         78      79      +1     
  Lines       6689    6868    +179     
=======================================
+ Hits        4964    5125    +161     
- Misses      1725    1743     +18     

Files	Coverage Δ
kube-runtime/src/reflector/mod.rs	100.0% <100.0%> (ø)
kube-runtime/src/reflector/store.rs	99.2% <100.0%> (+0.3%)	⬆️
kube-runtime/src/utils/watch_ext.rs	21.1% <100.0%> (+9.3%)	⬆️
kube-runtime/src/reflector/dispatcher.rs	95.5% <95.5%> (ø)
kube-runtime/src/controller/mod.rs	32.3% <0.0%> (-1.4%)	⬇️

... and 1 file with indirect coverage changes

[clux]

This is huge! Amazing work. I've left some initial thoughts on the public interface and some of the internals.

[clux]

a couple more comments

[clux]

looks great. nothing major stands out so 👍 from me

some minor nits, feel free to fix/ignore