Document Controller::reconcile_on and remove Err input requirement

[clux]

The unstable method currently suggests that this method can be used to help share a store with the reconciler. This is actually nothing specific to reconcile_on, and you can do the same with the streams interface with watches_stream.

(We made the reconcile_on right before watches_stream became a thing so this makes sense.)

Secondly, we pretended we would handle errors being passed in from a third-party resource which is not true. See #1304 (comment). We now no longer accept a stream of Result<ObjectRef, Error> but a stream of ObjectRef. Users creating reconcilers around third party apis / external resources need to do error handling when dealing with these separately.

The example now highlights that this has a better use-case with actually getting arbitrary third-party info, and then mapping that to kubernetes objects.

Planning on putting whatever we settle on here as the de-facto example for "external relations" on https://kube.rs/controllers/relations/#external-relations as it would be good to have something that compiled that i can refer it to.

(CI required a fixup in #1305)

[codecov]

Codecov Report

Merging #1304 (1a0912b) into main (68ae55c) will decrease coverage by 0.0%.
The diff coverage is 0.0%.

Additional details and impacted files

@@           Coverage Diff           @@
##            main   #1304     +/-   ##
=======================================
- Coverage   72.4%   72.4%   -0.0%     
=======================================
  Files         75      75             
  Lines       6341    6343      +2     
=======================================
  Hits        4586    4586             
- Misses      1755    1757      +2     

Files	Coverage Δ
kube-runtime/src/controller/mod.rs	34.1% <0.0%> (-0.2%)	⬇️

[Dav1dde]

For me the example is hard to understand and doesn't look like a real usecase. ~5 lines of setup which seems to be very constructed and a magic constant managed-cm1 (Question when reading the code that come to mind: Where does that come from? What is it used for? Won't this trigger always the same object?)

[Dav1dde]

Maybe it'd be better to hide the boilerplate from the example but instead just have an external_stream and show a mapping to an ObjectRef.

Pseudo:

struct ExternalObject {
    name: String,
}

let external_stream = watch_external_objects().map_ok(|ext| ObjectRef::new(&format!("{}-cm", ext.name)).within(&ns));

[clux]

That sounds a lot better to me as well. Ended up making a doc hidden function and refer to it using your name to help indicate that you are meant to write this yourself. Updated the post and image above.

Have used a very dumb signature for this hidden fn;

fn watch_external_objects() -> impl Stream<Item = ExternalObject> { stream::iter(vec![]) }

as a way to help the people who go digging, but can also show this line by default if we think it is helpful.

[clux]

Note there is also a bit of an interesting quirk about this fn as it stands; it needs a kube::runtime::watcher::Error as the error type for the input stream.. This is typically not what users would have available if they are reconciling from an external resource...

Not sure what to do about that at the moment. I can change the example to highlight it with:

diff --git a/kube-runtime/src/controller/mod.rs b/kube-runtime/src/controller/mod.rs
index 69399d89..0c46fc58 100644
--- a/kube-runtime/src/controller/mod.rs
+++ b/kube-runtime/src/controller/mod.rs
@@ -1107,13 +1107,13 @@ where
     /// # let client: Client = todo!();
     /// # async fn reconcile(_: Arc<ConfigMap>, _: Arc<()>) -> Result<Action, Error> { Ok(Action::await_change()) }
     /// # fn error_policy(_: Arc<ConfigMap>, _: &kube::Error, _: Arc<()>) -> Action { Action::await_change() }
-    /// # fn watch_external_objects() -> impl Stream<Item = ExternalObject> { stream::iter(vec![]) }
+    /// # fn watch_external_objects() -> impl Stream<Item = Result<ExternalObject, kube::runtime::watcher::Error>> { stream::iter(vec![]) }
     /// # let ns = "controller-ns".to_string();
     /// struct ExternalObject {
     ///     name: String,
     /// }
-    /// let external_stream = watch_external_objects().map(|ext| {
-    ///     Ok(ObjectRef::new(&format!("{}-cm", ext.name)).within(&ns))
+    /// let external_stream = watch_external_objects().map_ok(|ext| {
+    ///     ObjectRef::new(&format!("{}-cm", ext.name)).within(&ns)
     /// });
     ///
     /// Controller::new(Api::<ConfigMap>::namespaced(client, &ns), Config::default())

but it's probably better to shift the signature of this function to not get an user error type passed in?

[clux]

ok, if we are to omit the Error type, we would still have to pack it into a Result<X, watcher::Error> internally in that case because it all gets packed into the trigger_selector of type stream::SelectAll<BoxStream<'static, Result<ReconcileRequest<K>, watcher::Error>>> on Controller

We could do that in the signature by doing:

diff --git a/kube-runtime/src/controller/mod.rs b/kube-runtime/src/controller/mod.rs
index 69399d89..cd85f241 100644
--- a/kube-runtime/src/controller/mod.rs
+++ b/kube-runtime/src/controller/mod.rs
@@ -1113,7 +1113,7 @@ where
     ///     name: String,
     /// }
     /// let external_stream = watch_external_objects().map(|ext| {
-    ///     Ok(ObjectRef::new(&format!("{}-cm", ext.name)).within(&ns))
+    ///     ObjectRef::new(&format!("{}-cm", ext.name)).within(&ns)
     /// });
     ///
     /// Controller::new(Api::<ConfigMap>::namespaced(client, &ns), Config::default())
@@ -1125,15 +1125,14 @@ where
     /// ```
     #[cfg(feature = "unstable-runtime-reconcile-on")]
     #[must_use]
-    pub fn reconcile_on(
-        mut self,
-        trigger: impl Stream<Item = Result<ObjectRef<K>, watcher::Error>> + Send + 'static,
-    ) -> Self {
+    pub fn reconcile_on(mut self, trigger: impl Stream<Item = ObjectRef<K>> + Send + 'static) -> Self {
         self.trigger_selector.push(
             trigger
-                .map_ok(move |obj| ReconcileRequest {
-                    obj_ref: obj,
-                    reason: ReconcileReason::Unknown,
+                .map(move |obj| {
+                    Ok(ReconcileRequest {
+                        obj_ref: obj,
+                        reason: ReconcileReason::Unknown,
+                    })
                 })
                 .boxed(),
         );

which I think is better?

[Dav1dde]

Seems definitely simpler, also the watcher::Error doesn't really make sense for anything external, but is there any benefit to having the controller handle the errors?

[clux]

We don't really handle errors to a large extent here. Inside the applier we kind of just pass them through:

Input errors are turned into a QueueError:

kube/kube-runtime/src/controller/mod.rs

Lines 277 to 278 in 68ae55c

	queue
	.map_err(Error::QueueError)

which is a variant of the error type returned by the Controller stream meaning you can pick up on this as in the example:

kube/examples/configmapgen_controller.rs

Lines 114 to 118 in 68ae55c

	.for_each(|res| async move {
	match res {
	Ok(o) => info!("reconciled {:?}", o),
	Err(e) => warn!("reconcile failed: {}", e),
	}

so you can get back a more nested error than the one literally saw earlier 🙃

---

before it reaches the applier you do get one dubious benefit as we apply a global StreamBackoff on the queuestream:

kube/kube-runtime/src/controller/mod.rs

Lines 1271 to 1273 in 68ae55c

	StreamBackoff::new(self.trigger_selector, self.trigger_backoff)
	.take_until(future::select_all(self.graceful_shutdown_selector)),
	self.config,

so if users posted Err objects into reconcile_on they would get the backoff policy attached to the combined stream which i guess would help them "not retry errors too frequently" if they didn't do any error handling on the input stream and kept spamming us with errors.

We probably don't want to encourage this though. If they are spamming us with errors they can take down the controller (because the backoff is global and continued triggering of it could starve the input stream). Plus it sends the wrong message that they don't need to handle the errors from the 3rd party / external api because they can pass it to us - and that only hurts them long term.

[nightkr]

Users creating reconcilers around third party apis / external resources need to do error handling when dealing with these separately.

It's less about kube-rs handling those errors, and more about ensuring that applications can have the choice of propagating them through to wherever it makes sense to handle them.

At this point, whoever is calling reconcile_on with a fallible stream has to either: ignore any errors outright, promote them into panics, or split-and-rejoin the stream (á la trystream_split_result). TSSR is the only one of those options that works reasonably well with how you would expect a stream to work, and it is a massive footgun.

Going with your own example of watch_external_objects, virtually any non-trivial implementation is going to have error conditions. I agree that packing it into watcher::Error is super ugly, but I don't think removing the ability to handle errors entirely is an improvement.

[Dav1dde]

It's less about kube-rs handling those errors, and more about ensuring that applications can have the choice of propagating them through to wherever it makes sense to handle them.

But what can you reasonably do with these errors in the Controller? Yes you could have a common/joined error handler, but what option does that actually give you other than logging and ignoring?

I feel like you end up with a better design if you handle these errors where they occur and then you can make the decision whether to trigger the controller anyways or not.

[clux]

Yeah. The example is meant to be a type-only stand-in for a typically much larger external interface module in your application that deserves to have its own error handling not interwoven with the controller side (latter of which often is not so big).

All the use cases I have for this does not even have a stream as input, but will rather produce a synthetic stream via periodic fetching on external APIs / or listening to POST /reconcile events, and then chose to emit accepted items through some other way (put some ideas for this in a callout in this section).