Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -321,6 +321,9 @@ private <P extends HasMetadata> ResolvedControllerConfiguration<P> controllerCon
var triggerReconcilerOnAllEvents =
annotation != null && annotation.triggerReconcilerOnAllEvents();

var onUpdateFilterCombinedWithOr =
annotation != null && annotation.onUpdateFilterCombinedWithOr();

InformerConfiguration<P> informerConfig =
InformerConfiguration.builder(resourceClass)
.initFromAnnotation(annotation != null ? annotation.informer() : null, context)
Expand All @@ -341,7 +344,8 @@ private <P extends HasMetadata> ResolvedControllerConfiguration<P> controllerCon
dependentFieldManager,
this,
informerConfig,
triggerReconcilerOnAllEvents);
triggerReconcilerOnAllEvents,
onUpdateFilterCombinedWithOr);
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -121,4 +121,8 @@ default boolean triggerReconcilerOnAllEvent() {
default boolean triggerReconcilerOnAllEvents() {
return false;
}

default boolean isOnUpdateFilterCombinedWithOr() {
return false;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ public class ControllerConfigurationOverrider<R extends HasMetadata> {
private Map<DependentResourceSpec, Object> configurations;
private final InformerConfiguration<R>.Builder config;
private boolean triggerReconcilerOnAllEvents;
private boolean onUpdateFilterCombinedWithOr;

private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
this.finalizer = original.getFinalizerName();
Expand All @@ -59,6 +60,7 @@ private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
this.name = original.getName();
this.fieldManager = original.fieldManager();
this.triggerReconcilerOnAllEvents = original.triggerReconcilerOnAllEvents();
this.onUpdateFilterCombinedWithOr = original.isOnUpdateFilterCombinedWithOr();
}

public ControllerConfigurationOverrider<R> withFinalizer(String finalizer) {
Expand Down Expand Up @@ -186,6 +188,12 @@ public ControllerConfigurationOverrider<R> withTriggerReconcilerOnAllEvents(
return this;
}

public ControllerConfigurationOverrider<R> withOnUpdateFilterCombinedWithOr(
boolean onUpdateFilterCombinedWithOr) {
this.onUpdateFilterCombinedWithOr = onUpdateFilterCombinedWithOr;
return this;
}

/**
* Sets a max page size limit when starting the informer. This will result in pagination while
* populating the cache. This means that longer lists will take multiple requests to fetch. See
Expand Down Expand Up @@ -231,6 +239,7 @@ public ControllerConfiguration<R> build() {
original.getConfigurationService(),
config.buildForController(),
triggerReconcilerOnAllEvents,
onUpdateFilterCombinedWithOr,
original.getWorkflowSpec().orElse(null));
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ public class ResolvedControllerConfiguration<P extends HasMetadata>
private final ConfigurationService configurationService;
private final String fieldManager;
private final boolean triggerReconcilerOnAllEvents;
private final boolean onUpdateFilterCombinedWithOr;
private WorkflowSpec workflowSpec;

public ResolvedControllerConfiguration(ControllerConfiguration<P> other) {
Expand All @@ -61,6 +62,7 @@ public ResolvedControllerConfiguration(ControllerConfiguration<P> other) {
other.getConfigurationService(),
other.getInformerConfig(),
other.triggerReconcilerOnAllEvents(),
other.isOnUpdateFilterCombinedWithOr(),
other.getWorkflowSpec().orElse(null));
}

Expand All @@ -77,6 +79,7 @@ public ResolvedControllerConfiguration(
ConfigurationService configurationService,
InformerConfiguration<P> informerConfig,
boolean triggerReconcilerOnAllEvents,
boolean onUpdateFilterCombinedWithOr,
WorkflowSpec workflowSpec) {
this(
name,
Expand All @@ -90,7 +93,8 @@ public ResolvedControllerConfiguration(
fieldManager,
configurationService,
informerConfig,
triggerReconcilerOnAllEvents);
triggerReconcilerOnAllEvents,
onUpdateFilterCombinedWithOr);
setWorkflowSpec(workflowSpec);
}

Expand All @@ -106,7 +110,8 @@ protected ResolvedControllerConfiguration(
String fieldManager,
ConfigurationService configurationService,
InformerConfiguration<P> informerConfig,
boolean triggerReconcilerOnAllEvents) {
boolean triggerReconcilerOnAllEvents,
boolean onUpdateFilterCombinedWithOr) {
this.informerConfig = informerConfig;
this.configurationService = configurationService;
this.name = ControllerConfiguration.ensureValidName(name, associatedReconcilerClassName);
Expand All @@ -120,6 +125,7 @@ protected ResolvedControllerConfiguration(
ControllerConfiguration.ensureValidFinalizerName(finalizer, getResourceTypeName());
this.fieldManager = fieldManager;
this.triggerReconcilerOnAllEvents = triggerReconcilerOnAllEvents;
this.onUpdateFilterCombinedWithOr = onUpdateFilterCombinedWithOr;
}

protected ResolvedControllerConfiguration(
Expand All @@ -139,6 +145,7 @@ protected ResolvedControllerConfiguration(
null,
configurationService,
InformerConfiguration.builder(resourceClass).buildForController(),
false,
false);
}

Expand Down Expand Up @@ -234,4 +241,9 @@ public String fieldManager() {
public boolean triggerReconcilerOnAllEvents() {
return triggerReconcilerOnAllEvents;
}

@Override
public boolean isOnUpdateFilterCombinedWithOr() {
return onUpdateFilterCombinedWithOr;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -105,4 +105,16 @@ MaxReconciliationInterval maxReconciliationInterval() default
* documentation for further details.
*/
boolean triggerReconcilerOnAllEvents() default false;

/**
* When set to {@code true}, the {@link
* io.javaoperatorsdk.operator.processing.event.source.filter.OnUpdateFilter} configured via
* {@link Informer#onUpdateFilter()} is combined with JOSDK's internal update filters using OR
* instead of the default AND logic. This allows the user filter to expand the set of events that
* trigger reconciliation — for example, to also reconcile on specific status field updates even
* when the resource generation has not changed.
*
* @return whether the user-provided update filter is combined with internal filters using OR
*/
boolean onUpdateFilterCombinedWithOr() default false;
}
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,13 @@ public ControllerEventSource(Controller<T> controller) {
Optional.ofNullable(informerConfig.getOnAddFilter()).ifPresent(this::setOnAddFilter);
Optional.ofNullable(informerConfig.getOnUpdateFilter())
.ifPresentOrElse(
filter -> setOnUpdateFilter(filter.and(internalOnUpdateFilter)),
filter -> {
if (config.isOnUpdateFilterCombinedWithOr()) {
setOnUpdateFilter(filter.or(internalOnUpdateFilter));
} else {
setOnUpdateFilter(filter.and(internalOnUpdateFilter));
}
},
Comment on lines +65 to +71
() -> setOnUpdateFilter(internalOnUpdateFilter));
Optional.ofNullable(informerConfig.getGenericFilter()).ifPresent(this::setGenericFilter);
setControllerConfiguration(config);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -140,13 +140,45 @@ void callsBroadcastsOnResourceEvents() {
eq(ResourceAction.UPDATED), eq(customResource1), eq(customResource1));
}

@Test
void orCombinedFilterTriggersEventWhenInternalFilterWouldReject() {
TestCustomResource cr = TestUtils.testCustomResource();
cr.getMetadata().setFinalizers(List.of(FINALIZER));
cr.getMetadata().setGeneration(1L);

OnUpdateFilter<TestCustomResource> userFilter = (newRes, oldRes) -> true;
source = new ControllerEventSource<>(new TestController(null, userFilter, null, true));
setUpSource(source, true, controllerConfig);

source.handleEvent(ResourceAction.UPDATED, cr, cr, null);

verify(eventHandler, times(1)).handleEvent(any());
}

@Test
void orCombinedFilterDoesNotTriggerWhenUserFilterAlsoRejects() {
TestCustomResource cr = TestUtils.testCustomResource();
cr.getMetadata().setFinalizers(List.of(FINALIZER));
cr.getMetadata().setGeneration(1L);

OnUpdateFilter<TestCustomResource> userFilter = (newRes, oldRes) -> false;
source = new ControllerEventSource<>(new TestController(null, userFilter, null, true));
setUpSource(source, true, controllerConfig);

source.handleEvent(ResourceAction.UPDATED, cr, cr, null);

verify(eventHandler, never()).handleEvent(any());
}

@Test
void filtersOutEventsOnAddAndUpdate() {
TestCustomResource cr = TestUtils.testCustomResource();

OnAddFilter<TestCustomResource> onAddFilter = (res) -> false;
OnUpdateFilter<TestCustomResource> onUpdatePredicate = (res, res2) -> false;
source = new ControllerEventSource<>(new TestController(onAddFilter, onUpdatePredicate, null));
source =
new ControllerEventSource<>(
new TestController(onAddFilter, onUpdatePredicate, null, false));
setUpSource(source, true, controllerConfig);

source.handleEvent(ResourceAction.ADDED, cr, null, null);
Expand All @@ -159,7 +191,7 @@ void filtersOutEventsOnAddAndUpdate() {
void genericFilterFiltersOutAddUpdateAndDeleteEvents() {
TestCustomResource cr = TestUtils.testCustomResource();

source = new ControllerEventSource<>(new TestController(null, null, res -> false));
source = new ControllerEventSource<>(new TestController(null, null, res -> false, false));
setUpSource(source, true, controllerConfig);

source.handleEvent(ResourceAction.ADDED, cr, null, null);
Expand All @@ -174,7 +206,7 @@ void ownUpdateEchoIsFilteredOutByEventFilter() throws InterruptedException {
// End-to-end smoke for the event-filter wiring on the controller path: an event for our
// own write must not propagate. Detail-level filter scenarios are covered in
// EventingDetailTest / EventFilterSupportTest.
source = spy(new ControllerEventSource<>(new TestController(null, null, null)));
source = spy(new ControllerEventSource<>(new TestController(null, null, null, false)));
setUpSource(source, true, controllerConfig);
doReturn(Optional.empty()).when(source).get(any());

Expand All @@ -189,7 +221,7 @@ void ownUpdateEchoIsFilteredOutByEventFilter() throws InterruptedException {
@Test
void foreignUpdateDuringFilteringPropagatesAsUpdate() {
// An external event during the filter window must surface (not be filtered as own).
source = spy(new ControllerEventSource<>(new TestController(null, null, null)));
source = spy(new ControllerEventSource<>(new TestController(null, null, null, false)));
setUpSource(source, true, controllerConfig);

var latch = sendForEventFilteringUpdate(2);
Expand All @@ -203,7 +235,7 @@ void foreignUpdateDuringFilteringPropagatesAsUpdate() {
void deleteEventDuringFilteringPropagatesAsDelete() {
// A DELETE arriving during the filter window must surface — the resource has gone,
// so the filter must not silence it just because our own write is still tracking RVs.
source = spy(new ControllerEventSource<>(new TestController(null, null, null)));
source = spy(new ControllerEventSource<>(new TestController(null, null, null, false)));
setUpSource(source, true, controllerConfig);

var latch = sendForEventFilteringUpdate(2);
Expand All @@ -223,7 +255,7 @@ void deleteEventDuringFilteringPropagatesAsDelete() {
void multipleForeignEventsDuringFilteringMergeIntoSingleEvent() {
// Several external events during one filter window collapse into a single
// synthesized event spanning prev → latest seen.
source = spy(new ControllerEventSource<>(new TestController(null, null, null)));
source = spy(new ControllerEventSource<>(new TestController(null, null, null, false)));
setUpSource(source, true, controllerConfig);

var latch = sendForEventFilteringUpdate(2);
Expand Down Expand Up @@ -266,17 +298,19 @@ private static class TestController extends Controller<TestCustomResource> {
public TestController(
OnAddFilter<TestCustomResource> onAddFilter,
OnUpdateFilter<TestCustomResource> onUpdateFilter,
GenericFilter<TestCustomResource> genericFilter) {
GenericFilter<TestCustomResource> genericFilter,
boolean onUpdateFilterCombinedWithOr) {
super(
reconciler,
new TestConfiguration(true, onAddFilter, onUpdateFilter, genericFilter),
new TestConfiguration(
true, onAddFilter, onUpdateFilter, genericFilter, onUpdateFilterCombinedWithOr),
MockKubernetesClient.client(TestCustomResource.class));
}

public TestController(boolean generationAware) {
super(
reconciler,
new TestConfiguration(generationAware, null, null, null),
new TestConfiguration(generationAware, null, null, null, false),
MockKubernetesClient.client(TestCustomResource.class));
}

Expand All @@ -298,7 +332,8 @@ public TestConfiguration(
boolean generationAware,
OnAddFilter<TestCustomResource> onAddFilter,
OnUpdateFilter<TestCustomResource> onUpdateFilter,
GenericFilter<TestCustomResource> genericFilter) {
GenericFilter<TestCustomResource> genericFilter,
boolean onUpdateFilterCombinedWithOr) {
super(
"test",
generationAware,
Expand All @@ -316,7 +351,8 @@ public TestConfiguration(
.withGenericFilter(genericFilter)
.withComparableResourceVersions(true)
.buildForController(),
false);
false,
onUpdateFilterCombinedWithOr);
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
/*
* Copyright Java Operator SDK Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.javaoperatorsdk.operator.baseapi.filter;

import java.time.Duration;
import java.util.Map;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;

import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

class OrCombinedFilterIT {

public static final String RESOURCE_NAME = "or-filter-test";
public static final int POLL_DELAY = 150;

@RegisterExtension
LocallyRunOperatorExtension operator =
LocallyRunOperatorExtension.builder()
.withReconciler(new OrCombinedFilterTestReconciler())
.build();

@Test
void orCombinedFilterTriggersReconciliationEvenWhenInternalFilterWouldReject() {
operator.create(createResource());

await()
.pollDelay(Duration.ofMillis(POLL_DELAY))
.untilAsserted(() -> assertThat(reconciler().getNumberOfExecutions()).isEqualTo(1));

// Spec update bumps generation — internal generation-aware filter accepts → reconcile
var res = operator.get(FilterTestCustomResource.class, RESOURCE_NAME);
res.getSpec().setValue("updated");
operator.replace(res);

await()
.pollDelay(Duration.ofMillis(POLL_DELAY))
.untilAsserted(() -> assertThat(reconciler().getNumberOfExecutions()).isEqualTo(2));

// Annotation-only update does not bump generation — internal filter would reject,
// but with OR combination the user filter accepts
res = operator.get(FilterTestCustomResource.class, RESOURCE_NAME);
res.getMetadata()
.setAnnotations(Map.of(OrCombinedFilterTestReconciler.TRIGGER_ANNOTATION, "true"));
operator.replace(res);

await()
.pollDelay(Duration.ofMillis(POLL_DELAY))
.untilAsserted(() -> assertThat(reconciler().getNumberOfExecutions()).isEqualTo(3));

// Removing the annotation: user filter rejects, no generation change, should not reconcile
res = operator.get(FilterTestCustomResource.class, RESOURCE_NAME);
res.getMetadata().getAnnotations().remove(OrCombinedFilterTestReconciler.TRIGGER_ANNOTATION);
operator.replace(res);

await()
.pollDelay(Duration.ofMillis(POLL_DELAY))
.untilAsserted(() -> assertThat(reconciler().getNumberOfExecutions()).isEqualTo(3));
}

private OrCombinedFilterTestReconciler reconciler() {
return operator.getReconcilerOfType(OrCombinedFilterTestReconciler.class);
}

FilterTestCustomResource createResource() {
var resource = new FilterTestCustomResource();
resource.setMetadata(new ObjectMetaBuilder().withName(RESOURCE_NAME).build());
resource.setSpec(new FilterTestResourceSpec());
resource.getSpec().setValue("initial");
return resource;
}
}
Loading
Loading