Skip to content

Commit f4828e9

Browse files
csviricsviri
authored
fix: max reconciliation interval applies after retry exhaustion (#1491)
1 parent 4771d82 commit f4828e9

File tree

8 files changed

+191
-46
lines changed

8 files changed

+191
-46
lines changed

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java

Lines changed: 35 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import io.javaoperatorsdk.operator.OperatorException;
1515
import io.javaoperatorsdk.operator.api.config.ConfigurationService;
1616
import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
17+
import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
1718
import io.javaoperatorsdk.operator.api.config.ExecutorServiceManager;
1819
import io.javaoperatorsdk.operator.api.monitoring.Metrics;
1920
import io.javaoperatorsdk.operator.api.reconciler.Constants;
@@ -37,71 +38,64 @@ public class EventProcessor<R extends HasMetadata> implements EventHandler, Life
3738
private static final long MINIMAL_RATE_LIMIT_RESCHEDULE_DURATION = 50;
3839

3940
private volatile boolean running;
41+
private final ControllerConfiguration<?> controllerConfiguration;
4042
private final ReconciliationDispatcher<R> reconciliationDispatcher;
4143
private final Retry retry;
4244
private final ExecutorService executor;
43-
private final String controllerName;
4445
private final Metrics metrics;
4546
private final Cache<R> cache;
4647
private final EventSourceManager<R> eventSourceManager;
4748
private final RateLimiter<? extends RateLimitState> rateLimiter;
4849
private final ResourceStateManager resourceStateManager = new ResourceStateManager();
4950
private final Map<String, Object> metricsMetadata;
5051

52+
5153
public EventProcessor(EventSourceManager<R> eventSourceManager) {
5254
this(
55+
eventSourceManager.getController().getConfiguration(),
5356
eventSourceManager.getControllerResourceEventSource(),
5457
ExecutorServiceManager.instance().executorService(),
55-
eventSourceManager.getController().getConfiguration().getName(),
5658
new ReconciliationDispatcher<>(eventSourceManager.getController()),
57-
eventSourceManager.getController().getConfiguration().getRetry(),
5859
ConfigurationServiceProvider.instance().getMetrics(),
59-
eventSourceManager.getController().getConfiguration().getRateLimiter(),
6060
eventSourceManager);
6161
}
6262

6363
@SuppressWarnings("rawtypes")
6464
EventProcessor(
65+
ControllerConfiguration controllerConfiguration,
6566
ReconciliationDispatcher<R> reconciliationDispatcher,
6667
EventSourceManager<R> eventSourceManager,
67-
String relatedControllerName,
68-
Retry retry,
69-
RateLimiter rateLimiter,
7068
Metrics metrics) {
7169
this(
70+
controllerConfiguration,
7271
eventSourceManager.getControllerResourceEventSource(),
7372
null,
74-
relatedControllerName,
7573
reconciliationDispatcher,
76-
retry,
7774
metrics,
78-
rateLimiter,
7975
eventSourceManager);
8076
}
8177

8278
@SuppressWarnings({"rawtypes", "unchecked"})
8379
private EventProcessor(
80+
ControllerConfiguration controllerConfiguration,
8481
Cache<R> cache,
8582
ExecutorService executor,
86-
String relatedControllerName,
8783
ReconciliationDispatcher<R> reconciliationDispatcher,
88-
Retry retry,
8984
Metrics metrics,
90-
RateLimiter rateLimiter,
9185
EventSourceManager<R> eventSourceManager) {
86+
this.controllerConfiguration = controllerConfiguration;
9287
this.running = false;
9388
this.executor =
9489
executor == null
9590
? new ScheduledThreadPoolExecutor(
9691
ConfigurationService.DEFAULT_RECONCILIATION_THREADS_NUMBER)
9792
: executor;
98-
this.controllerName = relatedControllerName;
9993
this.reconciliationDispatcher = reconciliationDispatcher;
100-
this.retry = retry;
94+
this.retry = controllerConfiguration.getRetry();
10195
this.cache = cache;
10296
this.metrics = metrics != null ? metrics : Metrics.NOOP;
10397
this.eventSourceManager = eventSourceManager;
104-
this.rateLimiter = rateLimiter;
98+
this.rateLimiter = controllerConfiguration.getRateLimiter();
10599

106100
metricsMetadata = Optional.ofNullable(eventSourceManager.getController())
107101
.map(Controller::getAssociatedGroupVersionKind)
@@ -272,18 +266,31 @@ synchronized void eventProcessingFinished(
272266
reScheduleExecutionIfInstructed(postExecutionControl, executionScope.getResource());
273267
}
274268
}
275-
276269
}
277270

278271
private void reScheduleExecutionIfInstructed(
279272
PostExecutionControl<R> postExecutionControl, R customResource) {
273+
280274
postExecutionControl
281275
.getReScheduleDelay()
282-
.ifPresent(delay -> {
276+
.ifPresentOrElse(delay -> {
283277
var resourceID = ResourceID.fromResource(customResource);
284278
log.debug("ReScheduling event for resource: {} with delay: {}",
285279
resourceID, delay);
286280
retryEventSource().scheduleOnce(resourceID, delay);
281+
}, () -> scheduleExecutionForMaxReconciliationInterval(customResource));
282+
}
283+
284+
private void scheduleExecutionForMaxReconciliationInterval(R customResource) {
285+
this.controllerConfiguration
286+
.maxReconciliationInterval()
287+
.ifPresent(m -> {
288+
var resourceID = ResourceID.fromResource(customResource);
289+
var delay = m.toMillis();
290+
log.debug("ReScheduling event for resource because for max reconciliation interval: " +
291+
"{} with delay: {}",
292+
resourceID, delay);
293+
retryEventSource().scheduleOnce(resourceID, delay);
287294
});
288295
}
289296

@@ -319,7 +326,10 @@ private void handleRetryOnException(
319326
metrics.failedReconciliation(resourceID, exception, metricsMetadata);
320327
retryEventSource().scheduleOnce(resourceID, delay);
321328
},
322-
() -> log.error("Exhausted retries for {}", executionScope));
329+
() -> {
330+
log.error("Exhausted retries for {}", executionScope);
331+
scheduleExecutionForMaxReconciliationInterval(executionScope.getResource());
332+
});
323333
}
324334

325335
private void cleanupOnSuccessfulExecution(ExecutionScope<R> executionScope) {
@@ -390,7 +400,7 @@ public void run() {
390400
final var name = thread.getName();
391401
try {
392402
MDCUtils.addResourceInfo(executionScope.getResource());
393-
thread.setName("ReconcilerExecutor-" + controllerName + "-" + thread.getId());
403+
thread.setName("ReconcilerExecutor-" + controllerName() + "-" + thread.getId());
394404
PostExecutionControl<R> postExecutionControl =
395405
reconciliationDispatcher.handleExecution(executionScope);
396406
eventProcessingFinished(executionScope, postExecutionControl);
@@ -403,10 +413,14 @@ public void run() {
403413

404414
@Override
405415
public String toString() {
406-
return controllerName + " -> " + executionScope;
416+
return controllerName() + " -> " + executionScope;
407417
}
408418
}
409419

420+
private String controllerName() {
421+
return controllerConfiguration.getName();
422+
}
423+
410424
public synchronized boolean isUnderProcessing(ResourceID resourceID) {
411425
return isControllerUnderExecution(resourceStateManager.getOrCreate(resourceID));
412426
}

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcher.java

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -264,9 +264,7 @@ private PostExecutionControl<P> createPostExecutionControl(P updatedCustomResour
264264
private void updatePostExecutionControlWithReschedule(
265265
PostExecutionControl<P> postExecutionControl,
266266
BaseControl<?> baseControl) {
267-
baseControl.getScheduleDelay().ifPresentOrElse(postExecutionControl::withReSchedule,
268-
() -> controller.getConfiguration().maxReconciliationInterval()
269-
.ifPresent(m -> postExecutionControl.withReSchedule(m.toMillis())));
267+
baseControl.getScheduleDelay().ifPresent(postExecutionControl::withReSchedule);
270268
}
271269

272270

‎operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java

Lines changed: 56 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import org.slf4j.LoggerFactory;
1616

1717
import io.fabric8.kubernetes.api.model.HasMetadata;
18+
import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
1819
import io.javaoperatorsdk.operator.api.config.RetryConfiguration;
1920
import io.javaoperatorsdk.operator.api.monitoring.Metrics;
2021
import io.javaoperatorsdk.operator.processing.event.rate.LinearRateLimiter;
@@ -25,6 +26,8 @@
2526
import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent;
2627
import io.javaoperatorsdk.operator.processing.event.source.timer.TimerEventSource;
2728
import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
29+
import io.javaoperatorsdk.operator.processing.retry.Retry;
30+
import io.javaoperatorsdk.operator.processing.retry.RetryExecution;
2831
import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
2932

3033
import static io.javaoperatorsdk.operator.TestUtils.markForDeletion;
@@ -70,12 +73,15 @@ void setup() {
7073
when(eventSourceManagerMock.getControllerResourceEventSource())
7174
.thenReturn(controllerResourceEventSourceMock);
7275
eventProcessor =
73-
spy(new EventProcessor(reconciliationDispatcherMock, eventSourceManagerMock, "Test", null,
74-
rateLimiterMock, null));
76+
spy(new EventProcessor(controllerConfiguration(null, rateLimiterMock),
77+
reconciliationDispatcherMock,
78+
eventSourceManagerMock, null));
7579
eventProcessor.start();
7680
eventProcessorWithRetry =
77-
spy(new EventProcessor(reconciliationDispatcherMock, eventSourceManagerMock, "Test",
78-
GenericRetry.defaultLimitedExponentialRetry(), rateLimiterMock, null));
81+
spy(new EventProcessor(
82+
controllerConfiguration(GenericRetry.defaultLimitedExponentialRetry(),
83+
rateLimiterMock),
84+
reconciliationDispatcherMock, eventSourceManagerMock, null));
7985
eventProcessorWithRetry.start();
8086
when(eventProcessor.retryEventSource()).thenReturn(retryTimerEventSourceMock);
8187
when(eventProcessorWithRetry.retryEventSource()).thenReturn(retryTimerEventSourceMock);
@@ -258,8 +264,9 @@ void cancelScheduleOnceEventsOnSuccessfulExecution() {
258264
void startProcessedMarkedEventReceivedBefore() {
259265
var crID = new ResourceID("test-cr", TEST_NAMESPACE);
260266
eventProcessor =
261-
spy(new EventProcessor(reconciliationDispatcherMock, eventSourceManagerMock, "Test", null,
262-
LinearRateLimiter.deactivatedRateLimiter(),
267+
spy(new EventProcessor(controllerConfiguration(null,
268+
LinearRateLimiter.deactivatedRateLimiter()), reconciliationDispatcherMock,
269+
eventSourceManagerMock,
263270
metricsMock));
264271
when(controllerResourceEventSourceMock.get(eq(crID)))
265272
.thenReturn(Optional.of(testCustomResource()));
@@ -367,6 +374,40 @@ void rateLimitsReconciliationSubmission() {
367374
verify(retryTimerEventSourceMock, times(1)).scheduleOnce((ResourceID) any(), anyLong());
368375
}
369376

377+
@Test
378+
void schedulesRetryForMarReconciliationInterval() {
379+
TestCustomResource customResource = testCustomResource();
380+
ExecutionScope executionScope = new ExecutionScope(customResource, null);
381+
PostExecutionControl postExecutionControl =
382+
PostExecutionControl.defaultDispatch();
383+
384+
eventProcessorWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
385+
386+
verify(retryTimerEventSourceMock, times(1)).scheduleOnce((ResourceID) any(), anyLong());
387+
}
388+
389+
@Test
390+
void schedulesRetryForMarReconciliationIntervalIfRetryExhausted() {
391+
RetryExecution mockRetryExecution = mock(RetryExecution.class);
392+
when(mockRetryExecution.nextDelay()).thenReturn(Optional.empty());
393+
Retry retry = mock(Retry.class);
394+
when(retry.initExecution()).thenReturn(mockRetryExecution);
395+
eventProcessorWithRetry =
396+
spy(new EventProcessor(controllerConfiguration(retry,
397+
LinearRateLimiter.deactivatedRateLimiter()), reconciliationDispatcherMock,
398+
eventSourceManagerMock,
399+
metricsMock));
400+
eventProcessorWithRetry.start();
401+
ExecutionScope executionScope = new ExecutionScope(testCustomResource(), null);
402+
PostExecutionControl postExecutionControl =
403+
PostExecutionControl.exceptionDuringExecution(new RuntimeException());
404+
when(eventProcessorWithRetry.retryEventSource()).thenReturn(retryTimerEventSourceMock);
405+
406+
eventProcessorWithRetry.eventProcessingFinished(executionScope, postExecutionControl);
407+
408+
verify(retryTimerEventSourceMock, times(1)).scheduleOnce((ResourceID) any(), anyLong());
409+
}
410+
370411
private ResourceID eventAlreadyUnderProcessing() {
371412
when(reconciliationDispatcherMock.handleExecution(any()))
372413
.then(
@@ -407,4 +448,13 @@ private void overrideData(ResourceID id, HasMetadata applyTo) {
407448
applyTo.getMetadata().setNamespace(id.getNamespace().orElse(null));
408449
}
409450

451+
ControllerConfiguration controllerConfiguration(Retry retry, RateLimiter rateLimiter) {
452+
ControllerConfiguration res = mock(ControllerConfiguration.class);
453+
when(res.getName()).thenReturn("Test");
454+
when(res.getRetry()).thenReturn(retry);
455+
when(res.getRateLimiter()).thenReturn(rateLimiter);
456+
when(res.maxReconciliationInterval()).thenReturn(Optional.of(Duration.ofMillis(1000)));
457+
return res;
458+
}
459+
410460
}

‎operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java

Lines changed: 0 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -600,19 +600,6 @@ void errorStatusHandlerCanPatchResource() {
600600
any(), any());
601601
}
602602

603-
@Test
604-
void schedulesReconciliationIfMaxDelayIsSet() {
605-
testCustomResource.addFinalizer(DEFAULT_FINALIZER);
606-
607-
reconciler.reconcile = (r, c) -> UpdateControl.noUpdate();
608-
609-
PostExecutionControl control =
610-
reconciliationDispatcher.handleExecution(executionScopeWithCREvent(testCustomResource));
611-
612-
assertThat(control.getReScheduleDelay()).isPresent()
613-
.hasValue(TimeUnit.HOURS.toMillis(RECONCILIATION_MAX_INTERVAL));
614-
}
615-
616603
@Test
617604
void canSkipSchedulingMaxDelayIf() {
618605
testCustomResource.addFinalizer(DEFAULT_FINALIZER);

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/MaxIntervalAfterRetryIT.java

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
package io.javaoperatorsdk.operator;
2+
3+
import java.util.concurrent.TimeUnit;
4+
5+
import org.junit.jupiter.api.Test;
6+
import org.junit.jupiter.api.extension.RegisterExtension;
7+
8+
import io.fabric8.kubernetes.api.model.ObjectMeta;
9+
import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension;
10+
import io.javaoperatorsdk.operator.sample.maxintervalafterretry.MaxIntervalAfterRetryTestCustomResource;
11+
import io.javaoperatorsdk.operator.sample.maxintervalafterretry.MaxIntervalAfterRetryTestReconciler;
12+
13+
import static org.assertj.core.api.Assertions.assertThat;
14+
import static org.awaitility.Awaitility.await;
15+
16+
class MaxIntervalAfterRetryIT {
17+
18+
@RegisterExtension
19+
LocallyRunOperatorExtension operator =
20+
LocallyRunOperatorExtension.builder()
21+
.withReconciler(new MaxIntervalAfterRetryTestReconciler()).build();
22+
23+
@Test
24+
void reconciliationTriggeredBasedOnMaxInterval() {
25+
MaxIntervalAfterRetryTestCustomResource cr = createTestResource();
26+
27+
operator.create(cr);
28+
29+
await()
30+
.pollInterval(50, TimeUnit.MILLISECONDS)
31+
.atMost(1, TimeUnit.SECONDS)
32+
.untilAsserted(
33+
() -> assertThat(operator.getReconcilerOfType(MaxIntervalAfterRetryTestReconciler.class)
34+
.getNumberOfExecutions()).isGreaterThan(5));
35+
}
36+
37+
private MaxIntervalAfterRetryTestCustomResource createTestResource() {
38+
MaxIntervalAfterRetryTestCustomResource cr = new MaxIntervalAfterRetryTestCustomResource();
39+
cr.setMetadata(new ObjectMeta());
40+
cr.getMetadata().setName("maxintervalretrytest1");
41+
return cr;
42+
}
43+
}

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/MaxIntervalIT.java

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import io.javaoperatorsdk.operator.sample.maxinterval.MaxIntervalTestCustomResource;
1111
import io.javaoperatorsdk.operator.sample.maxinterval.MaxIntervalTestReconciler;
1212

13+
import static org.assertj.core.api.Assertions.assertThat;
1314
import static org.awaitility.Awaitility.await;
1415

1516
class MaxIntervalIT {
@@ -27,9 +28,10 @@ void reconciliationTriggeredBasedOnMaxInterval() {
2728
await()
2829
.pollInterval(50, TimeUnit.MILLISECONDS)
2930
.atMost(500, TimeUnit.MILLISECONDS)
30-
.until(
31-
() -> ((MaxIntervalTestReconciler) operator.getFirstReconciler())
32-
.getNumberOfExecutions() > 3);
31+
.untilAsserted(
32+
() -> assertThat(operator.getReconcilerOfType(MaxIntervalTestReconciler.class)
33+
.getNumberOfExecutions())
34+
.isGreaterThan(3));
3335
}
3436

3537
private MaxIntervalTestCustomResource createTestResource() {

‎operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/maxintervalafterretry/MaxIntervalAfterRetryTestCustomResource.java

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package io.javaoperatorsdk.operator.sample.maxintervalafterretry;
2+
3+
import io.fabric8.kubernetes.api.model.Namespaced;
4+
import io.fabric8.kubernetes.client.CustomResource;
5+
import io.fabric8.kubernetes.model.annotation.Group;
6+
import io.fabric8.kubernetes.model.annotation.ShortNames;
7+
import io.fabric8.kubernetes.model.annotation.Version;
8+
9+
@Group("sample.javaoperatorsdk")
10+
@Version("v1")
11+
@ShortNames("mir")
12+
public class MaxIntervalAfterRetryTestCustomResource
13+
extends CustomResource<Void, Void>
14+
implements Namespaced {
15+
}

0 commit comments

Comments
 (0)