feat: reconciliation timeout

Signed-off-by: Attila Mészáros <a_meszaros@apple.com>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/BaseConfigurationService.java

@@ -311,6 +311,19 @@ private <P extends HasMetadata> ResolvedControllerConfiguration<P> controllerCon
       timeUnit = reconciliationInterval.timeUnit();
     }
 
+    final var reconciliationTimeoutAnnotation =
+        valueOrDefaultFromAnnotation(
+            annotation,
+            io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration
+                ::reconciliationTimeout,
+            "reconciliationTimeout");
+    long timeoutInterval = -1;
+    TimeUnit timeoutTimeUnit = null;
+    if (reconciliationTimeoutAnnotation != null && reconciliationTimeoutAnnotation.interval() > 0) {
+      timeoutInterval = reconciliationTimeoutAnnotation.interval();
+      timeoutTimeUnit = reconciliationTimeoutAnnotation.timeUnit();
+    }
+
     var fieldManager =
         valueOrDefaultFromAnnotation(
             annotation,
@@ -334,6 +347,7 @@ private <P extends HasMetadata> ResolvedControllerConfiguration<P> controllerCon
         retry,
         rateLimiter,
         ResolvedControllerConfiguration.getMaxReconciliationInterval(interval, timeUnit),
+        ResolvedControllerConfiguration.getReconciliationTimeout(timeoutInterval, timeoutTimeUnit),
         valueOrDefaultFromAnnotation(
             annotation,
             io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration::finalizerName,

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfiguration.java

@@ -121,4 +121,8 @@ default boolean triggerReconcilerOnAllEvent() {
   default boolean triggerReconcilerOnAllEvents() {
     return false;
   }
+
+  default Optional<Duration> reconciliationTimeout() {
+    return Optional.empty();
+  }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java

@@ -43,6 +43,7 @@ public class ControllerConfigurationOverrider<R extends HasMetadata> {
   private RateLimiter rateLimiter;
   private String fieldManager;
   private Duration reconciliationMaxInterval;
+  private Duration reconciliationTimeout;
   private Map<DependentResourceSpec, Object> configurations;
   private final InformerConfiguration<R>.Builder config;
   private boolean triggerReconcilerOnAllEvents;
@@ -54,6 +55,7 @@ private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     this.config = InformerConfiguration.builder(informerConfig);
     this.retry = original.getRetry();
     this.reconciliationMaxInterval = original.maxReconciliationInterval().orElse(null);
+    this.reconciliationTimeout = original.reconciliationTimeout().orElse(null);
     this.original = original;
     this.rateLimiter = original.getRateLimiter();
     this.name = original.getName();
@@ -140,6 +142,12 @@ public ControllerConfigurationOverrider<R> withReconciliationMaxInterval(
     return this;
   }
 
+  public ControllerConfigurationOverrider<R> withReconciliationTimeout(
+      Duration reconciliationTimeout) {
+    this.reconciliationTimeout = reconciliationTimeout;
+    return this;
+  }
+
   public ControllerConfigurationOverrider<R> withOnAddFilter(OnAddFilter<R> onAddFilter) {
     config.withOnAddFilter(onAddFilter);
     return this;
@@ -225,6 +233,7 @@ public ControllerConfiguration<R> build() {
         retry,
         rateLimiter,
         reconciliationMaxInterval,
+        reconciliationTimeout,
         finalizer,
         configurations,
         fieldManager,

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ResolvedControllerConfiguration.java

@@ -40,6 +40,7 @@ public class ResolvedControllerConfiguration<P extends HasMetadata>
   private final Retry retry;
   private final RateLimiter rateLimiter;
   private final Duration maxReconciliationInterval;
+  private final Duration reconciliationTimeout;
   private final String finalizer;
   private final Map<DependentResourceSpec, Object> configurations;
   private final ConfigurationService configurationService;
@@ -55,6 +56,7 @@ public ResolvedControllerConfiguration(ControllerConfiguration<P> other) {
         other.getRetry(),
         other.getRateLimiter(),
         other.maxReconciliationInterval().orElse(null),
+        other.reconciliationTimeout().orElse(null),
         other.getFinalizerName(),
         Collections.emptyMap(),
         other.fieldManager(),
@@ -71,6 +73,7 @@ public ResolvedControllerConfiguration(
       Retry retry,
       RateLimiter rateLimiter,
       Duration maxReconciliationInterval,
+      Duration reconciliationTimeout,
       String finalizer,
       Map<DependentResourceSpec, Object> configurations,
       String fieldManager,
@@ -85,6 +88,7 @@ public ResolvedControllerConfiguration(
         retry,
         rateLimiter,
         maxReconciliationInterval,
+        reconciliationTimeout,
         finalizer,
         configurations,
         fieldManager,
@@ -101,6 +105,7 @@ protected ResolvedControllerConfiguration(
       Retry retry,
       RateLimiter rateLimiter,
       Duration maxReconciliationInterval,
+      Duration reconciliationTimeout,
       String finalizer,
       Map<DependentResourceSpec, Object> configurations,
       String fieldManager,
@@ -115,6 +120,7 @@ protected ResolvedControllerConfiguration(
     this.retry = ensureRetry(retry);
     this.rateLimiter = ensureRateLimiter(rateLimiter);
     this.maxReconciliationInterval = maxReconciliationInterval;
+    this.reconciliationTimeout = reconciliationTimeout;
     this.configurations = configurations != null ? configurations : Collections.emptyMap();
     this.finalizer =
         ControllerConfiguration.ensureValidFinalizerName(finalizer, getResourceTypeName());
@@ -137,6 +143,7 @@ protected ResolvedControllerConfiguration(
         null,
         null,
         null,
+        null,
         configurationService,
         InformerConfiguration.builder(resourceClass).buildForController(),
         false);
@@ -151,6 +158,10 @@ public static Duration getMaxReconciliationInterval(long interval, TimeUnit time
     return interval > 0 ? Duration.of(interval, timeUnit.toChronoUnit()) : null;
   }
 
+  public static Duration getReconciliationTimeout(long interval, TimeUnit timeUnit) {
+    return interval > 0 ? Duration.of(interval, timeUnit.toChronoUnit()) : null;
+  }
+
   public static String getAssociatedReconcilerClassName(
       Class<? extends Reconciler> reconcilerClass) {
     return reconcilerClass.getCanonicalName();
@@ -208,6 +219,11 @@ public Optional<Duration> maxReconciliationInterval() {
     return Optional.ofNullable(maxReconciliationInterval);
   }
 
+  @Override
+  public Optional<Duration> reconciliationTimeout() {
+    return Optional.ofNullable(reconciliationTimeout);
+  }
+
   @Override
   public ConfigurationService getConfigurationService() {
     return configurationService;

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

@@ -38,6 +38,7 @@ public final class Constants {
   public static final long NO_LONG_VALUE_SET = -1L;
 
   public static final long NO_MAX_RECONCILIATION_INTERVAL = -1L;
+  public static final long NO_RECONCILIATION_TIMEOUT = -1L;
 
   public static final String RESOURCE_GVK_KEY = "josdk.resource.gvk";
   public static final String CONTROLLER_NAME = "controller.name";

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

@@ -105,4 +105,14 @@ MaxReconciliationInterval maxReconciliationInterval() default
    * documentation for further details.
    */
   boolean triggerReconcilerOnAllEvents() default false;
+
+  /**
+   * Optional configuration of the maximum duration a single reconciliation is allowed to run. If
+   * the reconciliation takes longer than this timeout, the reconciliation thread will be
+   * interrupted and the result will be treated as an error, triggering retry logic.
+   *
+   * @return the reconciliation timeout configuration
+   */
+  ReconciliationTimeout reconciliationTimeout() default
+      @ReconciliationTimeout(interval = Constants.NO_RECONCILIATION_TIMEOUT);
 }

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

@@ -20,14 +20,21 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.ReconciliationTimeoutException;
 import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.monitoring.Metrics;
@@ -59,7 +66,12 @@ public class EventProcessor<P extends HasMetadata> implements EventHandler, Life
   private final RateLimiter<? extends RateLimitState> rateLimiter;
   private final ResourceStateManager resourceStateManager = new ResourceStateManager();
   private final Map<String, Object> metricsMetadata;
+  private final Duration reconciliationTimeout;
+  private final Map<ResourceID, Future<?>> runningFutures = new ConcurrentHashMap<>();
+  private final Map<ResourceID, ScheduledFuture<?>> timeoutFutures = new ConcurrentHashMap<>();
+  private final Set<ResourceID> timedOutResources = ConcurrentHashMap.newKeySet();
   private ExecutorService executor;
+  private ScheduledExecutorService timeoutScheduler;
 
   public EventProcessor(
       EventSourceManager<P> eventSourceManager, ConfigurationService configurationService) {
@@ -100,6 +112,8 @@ private EventProcessor(
     this.metrics = metrics != null ? metrics : Metrics.NOOP;
     this.eventSourceManager = eventSourceManager;
     this.rateLimiter = controllerConfiguration.getRateLimiter();
+    this.reconciliationTimeout =
+        (Duration) controllerConfiguration.reconciliationTimeout().orElse(null);
 
     metricsMetadata =
         Optional.ofNullable(eventSourceManager.getController())
@@ -182,7 +196,16 @@ private void submitReconciliationExecution(ResourceState state) {
         state.unMarkEventReceived(triggerOnAllEvents());
         metrics.reconciliationSubmitted(latest, state.getRetry(), metricsMetadata);
         log.debug("Executing events for custom resource. Scope: {}", executionScope);
-        executor.execute(new ReconcilerExecutor(resourceID, executionScope));
+        Future<?> future = executor.submit(new ReconcilerExecutor(resourceID, executionScope));
+        if (reconciliationTimeout != null) {
+          runningFutures.put(resourceID, future);
+          var timeoutFuture =
+              timeoutScheduler.schedule(
+                  () -> handleReconciliationTimeout(resourceID, future),
+                  reconciliationTimeout.toMillis(),
+                  TimeUnit.MILLISECONDS);
+          timeoutFutures.put(resourceID, timeoutFuture);
+        }
       } else {
         log.debug(
             "Skipping executing controller. Controller in execution: {}. Latest"
@@ -450,17 +473,22 @@ private boolean isRetryConfigured() {
   @Override
   public synchronized void stop() {
     this.running = false;
+    timeoutFutures.values().forEach(f -> f.cancel(false));
+    timeoutFutures.clear();
+    timedOutResources.clear();
+    runningFutures.clear();
   }
 
   @Override
   public synchronized void start() throws OperatorException {
     log.debug("Starting event processor: {}", this);
     // on restart new executor service is created and needs to be set here
-    executor =
-        controllerConfiguration
-            .getConfigurationService()
-            .getExecutorServiceManager()
-            .reconcileExecutorService();
+    var executorServiceManager =
+        controllerConfiguration.getConfigurationService().getExecutorServiceManager();
+    executor = executorServiceManager.reconcileExecutorService();
+    if (reconciliationTimeout != null) {
+      timeoutScheduler = executorServiceManager.scheduledExecutorService();
+    }
     this.running = true;
     handleAlreadyMarkedEvents();
   }
@@ -476,6 +504,29 @@ private void handleAlreadyMarkedEvents() {
     }
   }
 
+  private void handleReconciliationTimeout(ResourceID resourceID, Future<?> future) {
+    log.warn("Reconciliation timed out for resource: {}", resourceID);
+    timedOutResources.add(resourceID);
+    future.cancel(true);
+  }
+
+  private void cancelTimeoutIfActive(ResourceID resourceID) {
+    if (reconciliationTimeout != null) {
+      runningFutures.remove(resourceID);
+      var scheduledTimeout = timeoutFutures.remove(resourceID);
+      if (scheduledTimeout != null) {
+        scheduledTimeout.cancel(false);
+      }
+    }
+  }
+
+  private static boolean isInterruptedException(Throwable e) {
+    if (e instanceof InterruptedException) {
+      return true;
+    }
+    return e.getCause() != null && e.getCause() instanceof InterruptedException;
+  }
+
   private class ReconcilerExecutor implements Runnable {
     private final ExecutionScope<P> executionScope;
     private final ResourceID resourceID;
@@ -531,10 +582,33 @@ public void run() {
         MDCUtils.addResourceInfo(executionScope.getResource());
         metrics.reconciliationStarted(executionScope.getResource(), metricsMetadata);
         thread.setName("ReconcilerExecutor-" + controllerName() + "-" + thread.getId());
-        PostExecutionControl<P> postExecutionControl =
-            reconciliationDispatcher.handleExecution(executionScope);
+
+        PostExecutionControl<P> postExecutionControl;
+        try {
+          postExecutionControl = reconciliationDispatcher.handleExecution(executionScope);
+        } catch (Exception e) {
+          if (timedOutResources.remove(resourceID) || isInterruptedException(e)) {
+            log.warn("Reconciliation timed out for: {}", executionScope);
+            postExecutionControl =
+                PostExecutionControl.exceptionDuringExecution(
+                    new ReconciliationTimeoutException(reconciliationTimeout));
+          } else {
+            throw e;
+          }
+        }
+
+        // Check if this reconciliation was timed out (e.g. timeout fired just as execution
+        // finished)
+        if (timedOutResources.remove(resourceID)) {
+          log.warn("Reconciliation timed out for: {}", executionScope);
+          postExecutionControl =
+              PostExecutionControl.exceptionDuringExecution(
+                  new ReconciliationTimeoutException(reconciliationTimeout));
+        }
+
         eventProcessingFinished(executionScope, postExecutionControl);
       } finally {
+        cancelTimeoutIfActive(resourceID);
         metrics.reconciliationFinished(
             executionScope.getResource(), executionScope.getRetryInfo(), metricsMetadata);
         // restore original name