Skip to content

Commit a17da69

Browse files
csviricsviri
committed
add force filtering option
Signed-off-by: Attila Mészáros <a_meszaros@apple.com>
1 parent b813247 commit a17da69

5 files changed

Lines changed: 112 additions & 30 deletions

File tree

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperations.java‎

Lines changed: 57 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,10 @@ public ResourceOperations(Context<P> context) {
6868
* @param <R> resource type
6969
*/
7070
public <R extends HasMetadata> R serverSideApply(R resource) {
71+
return serverSideApply(resource, (Options) null);
72+
}
73+
74+
public <R extends HasMetadata> R serverSideApply(R resource, Options options) {
7175
return resourcePatch(
7276
resource,
7377
r ->
@@ -79,7 +83,13 @@ public <R extends HasMetadata> R serverSideApply(R resource) {
7983
.withForce(true)
8084
.withFieldManager(context.getControllerConfiguration().fieldManager())
8185
.withPatchType(PatchType.SERVER_SIDE_APPLY)
82-
.build()));
86+
.build()),
87+
options);
88+
}
89+
90+
public <R extends HasMetadata> R serverSideApply(
91+
R resource, InformerEventSource<R, P> informerEventSource) {
92+
return serverSideApply(resource, informerEventSource, null);
8393
}
8494

8595
/**
@@ -97,7 +107,7 @@ public <R extends HasMetadata> R serverSideApply(R resource) {
97107
* @param <R> resource type
98108
*/
99109
public <R extends HasMetadata> R serverSideApply(
100-
R resource, InformerEventSource<R, P> informerEventSource) {
110+
R resource, InformerEventSource<R, P> informerEventSource, Options options) {
101111
if (informerEventSource == null) {
102112
return serverSideApply(resource);
103113
}
@@ -208,6 +218,10 @@ public P serverSideApplyPrimaryStatus(P resource) {
208218
context.eventSourceRetriever().getControllerEventSource());
209219
}
210220

221+
public <R extends HasMetadata> R update(R resource) {
222+
return update(resource, (Options) null);
223+
}
224+
211225
/**
212226
* Updates the resource and caches the response if needed, thus making sure that next
213227
* reconciliation will see to updated resource - or more recent one if additional update happened
@@ -221,8 +235,13 @@ public P serverSideApplyPrimaryStatus(P resource) {
221235
* @return updated resource
222236
* @param <R> resource type
223237
*/
224-
public <R extends HasMetadata> R update(R resource) {
225-
return resourcePatch(resource, r -> context.getClient().resource(r).update());
238+
public <R extends HasMetadata> R update(R resource, Options options) {
239+
return resourcePatch(resource, r -> context.getClient().resource(r).update(), options);
240+
}
241+
242+
public <R extends HasMetadata> R update(
243+
R resource, InformerEventSource<R, P> informerEventSource) {
244+
return update(resource, informerEventSource, null);
226245
}
227246

228247
/**
@@ -240,12 +259,12 @@ public <R extends HasMetadata> R update(R resource) {
240259
* @param <R> resource type
241260
*/
242261
public <R extends HasMetadata> R update(
243-
R resource, InformerEventSource<R, P> informerEventSource) {
262+
R resource, InformerEventSource<R, P> informerEventSource, Options options) {
244263
if (informerEventSource == null) {
245264
return update(resource);
246265
}
247266
return resourcePatch(
248-
resource, r -> context.getClient().resource(r).update(), informerEventSource);
267+
resource, r -> context.getClient().resource(r).update(), informerEventSource, options);
249268
}
250269

251270
/**
@@ -262,7 +281,8 @@ public <R extends HasMetadata> R update(
262281
* @param <R> resource type
263282
*/
264283
public <R extends HasMetadata> R create(R resource) {
265-
return resourcePatch(resource, r -> context.getClient().resource(r).create());
284+
return resourcePatch(
285+
resource, r -> context.getClient().resource(r).create(), new Options(true));
266286
}
267287

268288
/**
@@ -284,8 +304,12 @@ public <R extends HasMetadata> R create(
284304
if (informerEventSource == null) {
285305
return create(resource);
286306
}
307+
// it is safe to do event filtering for create since check if the resource already exists.
287308
return resourcePatch(
288-
resource, r -> context.getClient().resource(r).create(), informerEventSource);
309+
resource,
310+
r -> context.getClient().resource(r).create(),
311+
informerEventSource,
312+
new Options(true));
289313
}
290314

291315
/**
@@ -518,6 +542,10 @@ public P jsonMergePatchPrimaryStatus(P resource) {
518542
context.eventSourceRetriever().getControllerEventSource());
519543
}
520544

545+
public <R extends HasMetadata> R resourcePatch(R resource, UnaryOperator<R> updateOperation) {
546+
return resourcePatch(resource, updateOperation, (Options) null);
547+
}
548+
521549
/**
522550
* Utility method to patch a resource and cache the result. Automatically discovers the event
523551
* source for the resource type and delegates to {@link #resourcePatch(HasMetadata, UnaryOperator,
@@ -530,7 +558,8 @@ public P jsonMergePatchPrimaryStatus(P resource) {
530558
* @throws IllegalStateException if no event source or multiple event sources are found
531559
*/
532560
@SuppressWarnings({"rawtypes", "unchecked"})
533-
public <R extends HasMetadata> R resourcePatch(R resource, UnaryOperator<R> updateOperation) {
561+
public <R extends HasMetadata> R resourcePatch(
562+
R resource, UnaryOperator<R> updateOperation, Options options) {
534563

535564
var esList = context.eventSourceRetriever().getEventSourcesFor(resource.getClass());
536565
if (esList.isEmpty()) {
@@ -544,7 +573,8 @@ public <R extends HasMetadata> R resourcePatch(R resource, UnaryOperator<R> upda
544573
es.name());
545574
}
546575
if (es instanceof ManagedInformerEventSource mes) {
547-
return resourcePatch(resource, updateOperation, (ManagedInformerEventSource<R, P, ?>) mes);
576+
return resourcePatch(
577+
resource, updateOperation, (ManagedInformerEventSource<R, P, ?>) mes, options);
548578
} else {
549579
throw new IllegalStateException(
550580
"Target event source must be a subclass off "
@@ -565,7 +595,16 @@ public <R extends HasMetadata> R resourcePatch(R resource, UnaryOperator<R> upda
565595
*/
566596
public <R extends HasMetadata> R resourcePatch(
567597
R resource, UnaryOperator<R> updateOperation, ManagedInformerEventSource<R, P, ?> ies) {
568-
return ies.eventFilteringUpdateAndCacheResource(resource, updateOperation);
598+
return resourcePatch(resource, updateOperation, ies, (Options) null);
599+
}
600+
601+
public <R extends HasMetadata> R resourcePatch(
602+
R resource,
603+
UnaryOperator<R> updateOperation,
604+
ManagedInformerEventSource<R, P, ?> ies,
605+
Options options) {
606+
return ies.eventFilteringUpdateAndCacheResource(
607+
resource, updateOperation, options != null && options.forceEventFiltering());
569608
}
570609

571610
/**
@@ -754,4 +793,11 @@ public P addFinalizerWithSSA(String finalizerName) {
754793
e);
755794
}
756795
}
796+
797+
@Experimental("This API might change")
798+
/**
799+
* Force filtering only if it is made sure that the update not results on a no-op change. See
800+
* details here: https://github.com/operator-framework/java-operator-sdk/pull/3484
801+
*/
802+
public record Options(boolean forceEventFiltering) {}
757803
}

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/KubernetesDependentResource.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import io.javaoperatorsdk.operator.api.reconciler.Context;
3131
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
3232
import io.javaoperatorsdk.operator.api.reconciler.Ignore;
33+
import io.javaoperatorsdk.operator.api.reconciler.ResourceOperations;
3334
import io.javaoperatorsdk.operator.api.reconciler.dependent.GarbageCollected;
3435
import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ConfiguredDependentResource;
3536
import io.javaoperatorsdk.operator.processing.GroupVersionKind;
@@ -90,7 +91,6 @@ public R create(R desired, P primary, Context<P> context) {
9091
desired.getClass(),
9192
ResourceID.fromResource(desired),
9293
ssa);
93-
9494
return ssa
9595
? context.resourceOperations().serverSideApply(desired, eventSource().orElse(null))
9696
: context.resourceOperations().create(desired, eventSource().orElse(null));
@@ -114,7 +114,10 @@ public R update(R actual, R desired, P primary, Context<P> context) {
114114
ssa);
115115
if (ssa) {
116116
updatedResource =
117-
context.resourceOperations().serverSideApply(desired, eventSource().orElse(null));
117+
context
118+
.resourceOperations()
119+
.serverSideApply(
120+
desired, eventSource().orElse(null), new ResourceOperations.Options(true));
118121
} else {
119122
var updatedActual = GenericResourceUpdater.updateResource(actual, desired, context);
120123
updatedResource =

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSource.java‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
4040
import io.javaoperatorsdk.operator.api.config.Informable;
4141
import io.javaoperatorsdk.operator.api.config.NamespaceChangeable;
42+
import io.javaoperatorsdk.operator.api.reconciler.Experimental;
4243
import io.javaoperatorsdk.operator.api.reconciler.dependent.RecentOperationCacheFiller;
4344
import io.javaoperatorsdk.operator.health.InformerHealthIndicator;
4445
import io.javaoperatorsdk.operator.health.InformerWrappingEventSourceHealthIndicator;
@@ -88,14 +89,23 @@ public void changeNamespaces(Set<String> namespaces) {
8889
}
8990
}
9091

92+
@Experimental(
93+
"Internal API. Use ResourceOperation not this directly, this API might change in the future")
94+
public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator<R> updateMethod) {
95+
return eventFilteringUpdateAndCacheResource(resourceToUpdate, updateMethod, false);
96+
}
97+
9198
/**
9299
* Updates the resource and makes sure that the response is available for the next reconciliation.
93100
* Also makes sure that the even produced by this update is filtered, thus does not trigger the
94101
* reconciliation.
95102
*/
103+
@Experimental(
104+
"Internal API. Use ResourceOperation not this directly, this API might change in the future")
96105
@SuppressWarnings("unchecked")
97-
public R eventFilteringUpdateAndCacheResource(R resourceToUpdate, UnaryOperator<R> updateMethod) {
98-
if (resourceToUpdate.getMetadata().getResourceVersion() == null) {
106+
public R eventFilteringUpdateAndCacheResource(
107+
R resourceToUpdate, UnaryOperator<R> updateMethod, boolean forceUpdateFilter) {
108+
if (resourceToUpdate.getMetadata().getResourceVersion() == null && !forceUpdateFilter) {
99109
log.debug("No resourceVersion set. Skipping event filtering.");
100110
return updateAndCacheResource(resourceToUpdate, updateMethod);
101111
}

‎operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/reconciler/ResourceOperationsTest.java‎

Lines changed: 16 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,7 @@ void addsFinalizer() {
8484

8585
// Mock successful finalizer addition
8686
when(controllerEventSource.eventFilteringUpdateAndCacheResource(
87-
any(), any(UnaryOperator.class)))
87+
any(), any(UnaryOperator.class), anyBoolean()))
8888
.thenAnswer(
8989
invocation -> {
9090
var res = TestUtils.testCustomResource1();
@@ -99,7 +99,7 @@ void addsFinalizer() {
9999
assertThat(result.hasFinalizer(FINALIZER_NAME)).isTrue();
100100
assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2");
101101
verify(controllerEventSource, times(1))
102-
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
102+
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
103103
}
104104

105105
@Test
@@ -111,7 +111,7 @@ void addsFinalizerWithSSA() {
111111

112112
// Mock successful SSA finalizer addition
113113
when(controllerEventSource.eventFilteringUpdateAndCacheResource(
114-
any(), any(UnaryOperator.class)))
114+
any(), any(UnaryOperator.class), anyBoolean()))
115115
.thenAnswer(
116116
invocation -> {
117117
var res = TestUtils.testCustomResource1();
@@ -126,7 +126,7 @@ void addsFinalizerWithSSA() {
126126
assertThat(result.hasFinalizer(FINALIZER_NAME)).isTrue();
127127
assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2");
128128
verify(controllerEventSource, times(1))
129-
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
129+
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
130130
}
131131

132132
@Test
@@ -139,7 +139,7 @@ void removesFinalizer() {
139139

140140
// Mock successful finalizer removal
141141
when(controllerEventSource.eventFilteringUpdateAndCacheResource(
142-
any(), any(UnaryOperator.class)))
142+
any(), any(UnaryOperator.class), anyBoolean()))
143143
.thenAnswer(
144144
invocation -> {
145145
var res = TestUtils.testCustomResource1();
@@ -154,7 +154,7 @@ void removesFinalizer() {
154154
assertThat(result.hasFinalizer(FINALIZER_NAME)).isFalse();
155155
assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2");
156156
verify(controllerEventSource, times(1))
157-
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
157+
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
158158
}
159159

160160
@Test
@@ -166,7 +166,7 @@ void retriesAddingFinalizerWithoutSSA() {
166166

167167
// First call throws conflict, second succeeds
168168
when(controllerEventSource.eventFilteringUpdateAndCacheResource(
169-
any(), any(UnaryOperator.class)))
169+
any(), any(UnaryOperator.class), anyBoolean()))
170170
.thenThrow(new KubernetesClientException("Conflict", 409, null))
171171
.thenAnswer(
172172
invocation -> {
@@ -184,7 +184,7 @@ void retriesAddingFinalizerWithoutSSA() {
184184
assertThat(result).isNotNull();
185185
assertThat(result.hasFinalizer(FINALIZER_NAME)).isTrue();
186186
verify(controllerEventSource, times(2))
187-
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
187+
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
188188
verify(resourceOp, times(1)).get();
189189
}
190190

@@ -198,7 +198,7 @@ void nullResourceIsGracefullyHandledOnFinalizerRemovalRetry() {
198198

199199
// First call throws conflict
200200
when(controllerEventSource.eventFilteringUpdateAndCacheResource(
201-
any(), any(UnaryOperator.class)))
201+
any(), any(UnaryOperator.class), anyBoolean()))
202202
.thenThrow(new KubernetesClientException("Conflict", 409, null));
203203

204204
// Return null on retry (resource was deleted)
@@ -207,7 +207,7 @@ void nullResourceIsGracefullyHandledOnFinalizerRemovalRetry() {
207207
resourceOperations.removeFinalizer(FINALIZER_NAME);
208208

209209
verify(controllerEventSource, times(1))
210-
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
210+
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
211211
verify(resourceOp, times(1)).get();
212212
}
213213

@@ -221,7 +221,7 @@ void retriesFinalizerRemovalWithFreshResource() {
221221

222222
// First call throws unprocessable (422), second succeeds
223223
when(controllerEventSource.eventFilteringUpdateAndCacheResource(
224-
any(), any(UnaryOperator.class)))
224+
any(), any(UnaryOperator.class), anyBoolean()))
225225
.thenThrow(new KubernetesClientException("Unprocessable", 422, null))
226226
.thenAnswer(
227227
invocation -> {
@@ -243,7 +243,7 @@ void retriesFinalizerRemovalWithFreshResource() {
243243
assertThat(result.getMetadata().getResourceVersion()).isEqualTo("3");
244244
assertThat(result.hasFinalizer(FINALIZER_NAME)).isFalse();
245245
verify(controllerEventSource, times(2))
246-
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
246+
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
247247
verify(resourceOp, times(1)).get();
248248
}
249249

@@ -261,15 +261,16 @@ void resourcePatchWithSingleEventSource() {
261261
when(context.eventSourceRetriever()).thenReturn(eventSourceRetriever);
262262
when(eventSourceRetriever.getEventSourcesFor(TestCustomResource.class))
263263
.thenReturn(List.of(managedEventSource));
264-
when(managedEventSource.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class)))
264+
when(managedEventSource.eventFilteringUpdateAndCacheResource(
265+
any(), any(UnaryOperator.class), anyBoolean()))
265266
.thenReturn(updatedResource);
266267

267268
var result = resourceOperations.resourcePatch(resource, UnaryOperator.identity());
268269

269270
assertThat(result).isNotNull();
270271
assertThat(result.getMetadata().getResourceVersion()).isEqualTo("2");
271272
verify(managedEventSource, times(1))
272-
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
273+
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
273274
}
274275

275276
@Test
@@ -303,7 +304,7 @@ void resourcePatchUsesFirstEventSourceIfMultipleEventSourcesPresent() {
303304
resourceOperations.resourcePatch(resource, UnaryOperator.identity());
304305

305306
verify(eventSource1, times(1))
306-
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class));
307+
.eventFilteringUpdateAndCacheResource(any(), any(UnaryOperator.class), anyBoolean());
307308
}
308309

309310
@Test

‎operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -799,6 +799,28 @@ void skipsEventFilteringWhenResourceVersionIsNull() {
799799
verify(eventHandlerMock, never()).handleEvent(any());
800800
}
801801

802+
@Test
803+
void forceUpdateFilterOpensFilterWindowEvenWhenResourceVersionIsNull() {
804+
// A write without a resourceVersion (e.g. an SSA finalizer add, which uses no optimistic
805+
// locking) would normally skip event filtering. forceUpdateFilter=true forces filtering
806+
// anyway so the resulting own event is correlated and does not trigger a spurious
807+
// reconciliation.
808+
var resourceToUpdate = testDeployment();
809+
resourceToUpdate.getMetadata().setResourceVersion(null);
810+
var updated = deploymentWithResourceVersion(3);
811+
when(temporaryResourceCache.doneEventFilterModify(any())).thenReturn(Optional.empty());
812+
813+
var result =
814+
informerEventSource.eventFilteringUpdateAndCacheResource(
815+
resourceToUpdate, r -> updated, true);
816+
817+
assertThat(result).isSameAs(updated);
818+
verify(temporaryResourceCache, times(1)).startEventFilteringModify(any());
819+
verify(temporaryResourceCache, times(1)).doneEventFilterModify(any());
820+
verify(temporaryResourceCache, times(1)).putResource(updated);
821+
verify(eventHandlerMock, never()).handleEvent(any());
822+
}
823+
802824
private PrimaryToSecondaryIndex<Deployment> injectIndexMock() throws Exception {
803825
@SuppressWarnings("unchecked")
804826
PrimaryToSecondaryIndex<Deployment> indexMock = mock(PrimaryToSecondaryIndex.class);

0 commit comments

Comments
 (0)