Add Controller::reconcile_on

[co42]

Allow to feed the controller with a stream of ObjectRef<K> to reconcile.

Motivation

When creating a controller which watches some pods I wanted to be able to cache them with a reflector, and use the cache later in the reconcile function.
I does nos seems to be possible right now. Some PRs and issues related : #824 #1080.
This solution allow to do it with only one watch, which ensure that resources in the store are up to date when queried within the reconcile function.

Solution

Add a function Controller::reconcile_on which take a stream of ObjectRef<K> as parameter.

        let (pod_store, writer) = reflector::store();
        // Store can be used in the reconcile loop instead of querying Kube
        ctx.pod_store = pod_store;
        let stream = reflector(writer, watcher(pod_api.clone(), ListParams::default()))
            .applied_objects()
            .try_filter_map(|pod| future::ready(Ok(pod_mapper(pod))))
            .boxed();

        info!("Operator ready");
        let drainer = Controller::new(crd_api, ListParams::default())
            .shutdown_on_signal()
            // NEW FUNCTION
            .reconcile_on(stream)
            .run(reconcile, error_policy, Arc::new(ctx))
            .for_each(|_result| async move {})
            .boxed();

If the design is accepted I'll add documentation and tests

[codecov]

Codecov Report

Merging #1163 (511b78b) into main (e01187e) will decrease coverage by 0.97%.
The diff coverage is 0.00%.

❗ Current head 511b78b differs from pull request most recent head e2f86d5. Consider uploading reports for the commit e2f86d5 to get more accurate results

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1163      +/-   ##
==========================================
- Coverage   73.49%   72.52%   -0.97%     
==========================================
  Files          68       67       -1     
  Lines        5349     5216     -133     
==========================================
- Hits         3931     3783     -148     
- Misses       1418     1433      +15     

Impacted Files	Coverage Δ
kube-runtime/src/controller/mod.rs	35.86% <0.00%> (+0.44%)	⬆️

... and 12 files with indirect coverage changes

[clux]

Hey there. Thanks a lot for this. This looks like a high potential building block going forward!

There are actually two potential use cases for the signature you've proposed:

• stream sharing Sharing watch streams and caches between Controllers #1080
• being able to trigger individual reconciles from something like an http route

it might be more natural to sell this in the docs as a trigger helper initially, because i feel that maybe this is too low-level to recommend as the main interface for re-using streams. e.g. maybe we should have something like Controller::watches_stream that does the try_filter_map internally. Still, both use-cases can probably be experimented with under an unstable feature flag initially.

Anyway, I am at least kind of keen to try this out to verify. It looks like it can at the very least bypass some of the stream recreation problems. Probably would reference this PR in a controller-rs PR to try it out a bit first.

[clux]

quick comment, haven't tested out anything, and currently this isn't tested

[co42]

I added a no_run doc test example.

Should I add some unit tests also ?

it might be more natural to sell this in the docs as a trigger helper initially, because i feel that maybe this is too low-level to recommend as the main interface for re-using streams. e.g. maybe we should have something like Controller::watches_stream that does the try_filter_map internally. Still, both use-cases can probably be experimented with under an unstable feature flag initially.

I agree, the good thing about it being low level, and not related to watches, is that it can used for others use cases. For example you can watch an external database and trigger the reconcile process on arbitrary resources.

[clux]

looks sensible to me, although doc tests are failing. left minor notes on it.

Should I add some unit tests also ?

it's probably hard to find a good setup for this in Controller. The doc test is probably ok. Although i am open to suggestions.

[clux]

minor doc nits plus help for making tests pass.

you can probably run cargo test -p kube-runtime --all-features --doc locally also

[co42]

minor doc nits plus help for making tests pass.

you can probably run cargo test -p kube-runtime --all-features --doc locally also

Sorry about that. Tests are working now and I applied your comments

[clux]

Ah, thanks for this. This looks better. Although you got screwed by being behind the PR #1162. Think this will pass after swapping ListParams to watcher::Config after updating.

[clux]

Also, do you mind adding an unstable flag for this initially?

• #[cfg(feature = "unstable-runtime-reconcile-on")] in front of the method
• list the feature in kube-runtime/Cargo.toml and add it to the unstable-runtime feature list

If it's good after a few versions we'll remove this.

[Dav1dde]

Maybe reconcile_on should not take ObjectRef<K> but a ReconcileRequest so it works with the existing trigger_owners, trigger_with and trigger_self functions?

[clux]

Maybe reconcile_on should not take ObjectRef<K> but a ReconcileRequest so it works with the existing trigger_owners, trigger_with and trigger_self functions?

I think for a main stream sharing interface, yes, we would rather want an input of raw K events. To me, reconcile_on fits more the external-trigger use case (e.g. via a http route) even though currently the example is geared towards stream sharing. We can change that later as we get more precise helpers.

I think going forward we should explore (in other PRs) functionality like:

• Controller::watches_stream
• Controller::owns_stream

That can take a stream of K as inputs created from user watchers.

[co42]

I rebased and added a feature flag.

Do I need to change something else ?

[clux]

Looks good. Thank you very much!