Understanding filter predicates and a possible race condition?

[sjbodzo]

When writing a fairly standard controller that watches a stream of K: CustomResource, I notice that the following filter does not do what I expect:

    let sink_stream = reflector(sink_writer, watcher(sinks, cfg.clone()))
        .default_backoff()
        .touched_objects()
        .predicate_filter((predicates::generation).fallback(predicates::resource_version));
    Controller::for_stream(sink_stream, sink_reader)
    // start the stream and visit each item, doing things

In this case my custom resource is type Sink, and the Api<Sink> details are bog-standard / default. I expect that creating a Sink in the cluster will trigger the stream and drive the reconciliation process on this controller.

What I observe is that the predicate_filter completely misses the new object creation. The same holds true if I remove the fallback from the predicate_filter as well. If I remove the predicate_filter, then the stream triggers on Sink creation as I would expect.

Some open questions:

• Could this be a race where the object triggers before the initial metadata is populated in the object? The Kubernetes SIG spec uses "should" as a weak constraint on the logical clock (e.g. generation, resourceVersion) fields, so perhaps they are not set before the object is filtered out of the queue?
• If I delete the pod running my controller, the new pod that comes up immediately triggers on the Sink creation event and runs the reconciliation loop. Does this give credence to the race condition theory?
• If I remove the predicate_filter I do not observe infinite re-triggering, which I would expect given that the controller watches Sink resources and updates the status subresource of the Sink in its reconciliation loop. I find this result confusing, given that nothing in touched_objects suggests this behavior. Why doesn't this cause re-triggering on the same stream? Perhaps Event::Applied<K> does not include updates to the subresource?

Thanks in advance for your time / help!

[clux]

some small answers, i don't know exactly what's going on here.

Could this be a race where the object triggers before the initial metadata is populated in the object?

if that's the case, then that would be a pretty big footgun in predicates. i've never seen objects missing data population like this, but I've not particularly looked for it either.

if you have an environment to test, a .map() with a println("saw unfiltered {obj:?}") after the predicate filter should help confirm / deny this.

If I delete the pod running my controller, the new pod that comes up immediately triggers on the Sink creation event and runs the reconciliation loop.

This part makes sense. The new controller will not have anything built up in its temporary memory predicate cache so every new event flows through unfiltered once.

If I remove the predicate_filter I do not observe infinite re-triggering, which I would expect given that the controller watches Sink resources and updates the status subresource of the Sink in its reconciliation loop. I find this result confusing, given that nothing in touched_objects suggests this behavior. Why doesn't this cause re-triggering on the same stream? Perhaps Event::Applied does not include updates to the subresource?

Without a filter, that would indeed be weird if your reconciler is writing non deterministic updates to the status object - as that should indeed re-trigger. It's a normal watcher so it should include status objects in the watch stream if you have included these in the crd schema for Sink. A print thing like the above should also help confirm / deny this.

[clux]

Actually, yeah. The stream sharing stuff has some unclear behaviour on how to broadcast events for cases like this. Basically, not everything is passed on atm as it was a bit unclear how broadcasts should interact with touched_objects() / applied_objects().

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

Lines 138 to 164 in f9902f1

	/// Broadcast an event to any downstream listeners subscribed on the store
	pub(crate) async fn dispatch_event(&mut self, event: &watcher::Event<K>) {
	if let Some(ref mut dispatcher) = self.dispatcher {
	match event {
	watcher::Event::Apply(obj) => {
	let obj_ref = obj.to_object_ref(self.dyntype.clone());
	// TODO (matei): should this take a timeout to log when backpressure has
	// been applied for too long, e.g. 10s
	dispatcher.broadcast(obj_ref).await;
	}
	watcher::Event::InitDone => {
	let obj_refs: Vec<_> = {
	let store = self.store.read();
	store.keys().cloned().collect()
	};
	for obj_ref in obj_refs {
	dispatcher.broadcast(obj_ref).await;
	}
	}
	_ => {}
	}
	}
	}
	}

is current, and #1472 is one follow-up ticket.

[sjbodzo]

Interesting. Thank you for all of the help and feedback.

Im going to mark the above as the answer to highlight the important caveat, though in my case the watches are not shared between the controllers. The for_stream is using a stream isolated to its controller, separate from the other controllers (which both use standard Controller::new style).

The codebase in question is running on 0.88.1, which predates the introduction of the stream sharing functionality. I will play with this some more, and if anything interesting comes to light will share it here as follow-up. Thanks again!

[clux]

Ah, apologies. If it's only using the regular streams interface (and not shared streams), then your case is a little more suspicious. The general streams interface should not have as many rough edges.

You should be able to test this in isolation with just a watcher stream and a println after each element going through the predicate side without going through a controller (that side should not matter for you if the streams are not shared between controllers).

    let stream = watcher(api, watcher::Config::default().any_semantic())
        .default_backoff()
        .modify(|x| println!("pre-filter {}", x.name_any()))
        .applied_objects()
        .predicate_filter(predicates::resource_version);
    let mut stream = pin!(stream);

    while let Some(pod) = stream.try_next().await? {
        println!("post-filter {}", pod.name_any());
    }

[sjbodzo]

@ivan-kiselev identified our root issue.

The first time a new object is seen, it does not have a finalizer set. Therefore the code will attempt to patch the object and await_change. Patching the object does not increment the object's generation so our predicate filter then prevents the object from triggering the controller. This is because the initial trigger caused generation = 1 to already be seen.

Restarting the controller instance fixed the problem by presenting a new store to cache.

The solution implemented was to write a custom predicate filter on K that inputs the hash of the generation (if present) and then adds in the hash of the finalizers:

pub fn generation_and_finalizers<K: Resource>(obj: &K) -> Option<u64> {
    let mut hasher = DefaultHasher::default();
    if let Some(gen) = obj.meta().generation {
        gen.hash(&mut hasher)
    };
    obj.finalizers().hash(&mut hasher);
    let hash = hasher.finish();
    Some(hash)
}

This prevents your stream-based controller from missing the initial object when using a generation predicate filter with a finalizer in your controller.

@clux marking this as the solution for posterity

[clux]

Thanks for this! If it is of interest, we could add predicates.finalizers as a standalone fn, then you could create a predicate like your above fn in the future with predicates::generation.fallback(predicates::finalizers).

[ivan-kiselev]

I will throw an item to my TODO list for this week to submit a PR, it doesn't seem to be much work :)

[ivan-kiselev]

here we go #1560