feat: support for graceful shutdown based on configuration

Support for graceful shutdown based on configuration

Signed-off-by: 10000-ki <[email protected]>

Fix naming

Signed-off-by: 10000-ki <[email protected]>

Fix lint error

Signed-off-by: 10000-ki <[email protected]>

Fix lint error

Signed-off-by: 10000-ki <[email protected]>

Fix lint error

Signed-off-by: 10000-ki <[email protected]>

Fix test duration

Signed-off-by: 10000-ki <[email protected]>

Author: 10000-ki

docs/content/en/docs/patterns-and-best-practices/_index.md

@@ -120,3 +120,25 @@ might be a permission issue for some resources in another namespace.
 The `stopOnInformerErrorDuringStartup` has implication on [cache sync timeout](https://github.com/java-operator-sdk/java-operator-sdk/blob/114c4312c32b34688811df8dd7cea275878c9e73/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java#L177-L179)
 behavior. If true operator will stop on cache sync timeout. if `false`, after the timeout the controller will start
 reconcile resources even if one or more event source caches did not sync yet.  
+
+## Graceful Shutdown
+
+We can provide sufficient time for the reconciler to process and complete the currently ongoing events before shutting down. 
+There are two ways to support the graceful shutdown feature.
+
+The first method is to directly provide a timeout value to the `stop` method.
+
+```java
+final var operator = new Operator();
+
+operator.stop(Duration.ofSeconds(5));
+```
+
+The second method is to specify the timeout value by overriding the `ConfigurationService`.
+
+```java
+final var overridden = new ConfigurationServiceOverrider(config)
+    .withReconciliationTerminationTimeout(Duration.ofSeconds(5));
+
+final var operator = new Operator(overridden);
+```

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Operator.java

@@ -164,7 +164,9 @@ public void stop(Duration gracefulShutdownTimeout) throws OperatorException {
 
   @Override
   public void stop() throws OperatorException {
-    stop(Duration.ZERO);
+    Duration reconciliationTerminationTimeout =
+        configurationService.reconciliationTerminationTimeout();
+    stop(reconciliationTerminationTimeout);
   }
 
   /**

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

@@ -101,8 +101,8 @@ static ConfigurationService newOverriddenConfigurationService(
    *
    * @param reconciler the reconciler we want the configuration of
    * @param <R> the {@code CustomResource} type associated with the specified reconciler
-   * @return the {@link ControllerConfiguration} associated with the specified reconciler or {@code
-   * null} if no configuration exists for the reconciler
+   * @return the {@link ControllerConfiguration} associated with the specified reconciler or
+   *         {@code null} if no configuration exists for the reconciler
    */
   <R extends HasMetadata> ControllerConfiguration<R> getConfigurationFor(Reconciler<R> reconciler);
 
@@ -211,7 +211,7 @@ default int concurrentWorkflowExecutorThreads() {
 
   /**
    * Override to provide a custom {@link Metrics} implementation
-   * 
+   *
    * @return the {@link Metrics} implementation
    */
   default Metrics getMetrics() {
@@ -221,7 +221,7 @@ default Metrics getMetrics() {
   /**
    * Override to provide a custom {@link ExecutorService} implementation to change how threads
    * handle concurrent reconciliations
-   * 
+   *
    * @return the {@link ExecutorService} implementation to use for concurrent reconciliation
    *         processing
    */
@@ -232,7 +232,7 @@ default ExecutorService getExecutorService() {
   /**
    * Override to provide a custom {@link ExecutorService} implementation to change how dependent
    * workflows are processed in parallel
-   * 
+   *
    * @return the {@link ExecutorService} implementation to use for dependent workflow processing
    */
   default ExecutorService getWorkflowExecutorService() {
@@ -242,7 +242,7 @@ default ExecutorService getWorkflowExecutorService() {
   /**
    * Determines whether the associated Kubernetes client should be closed when the associated
    * {@link io.javaoperatorsdk.operator.Operator} is stopped.
-   * 
+   *
    * @return {@code true} if the Kubernetes should be closed on stop, {@code false} otherwise
    */
   default boolean closeClientOnStop() {
@@ -252,7 +252,7 @@ default boolean closeClientOnStop() {
   /**
    * Override to provide a custom {@link DependentResourceFactory} implementation to change how
    * {@link io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource} are instantiated
-   * 
+   *
    * @return the custom {@link DependentResourceFactory} implementation
    */
   @SuppressWarnings("rawtypes")
@@ -264,7 +264,7 @@ default DependentResourceFactory dependentResourceFactory() {
    * Retrieves the optional {@link LeaderElectionConfiguration} to specify how the associated
    * {@link io.javaoperatorsdk.operator.Operator} handles leader election to ensure only one
    * instance of the operator runs on the cluster at any given time
-   * 
+   *
    * @return the {@link LeaderElectionConfiguration}
    */
   default Optional<LeaderElectionConfiguration> getLeaderElectionConfiguration() {
@@ -299,6 +299,16 @@ default Duration cacheSyncTimeout() {
     return Duration.ofMinutes(2);
   }
 
+  /**
+   * This is the timeout value that allows the reconciliation threads to gracefully shut down. If no
+   * value is set, the default is immediate shutdown.
+   *
+   * @return The duration of time to wait before terminating the reconciliation threads
+   */
+  default Duration reconciliationTerminationTimeout() {
+    return Duration.ZERO;
+  }
+
   /**
    * Handler for an informer stop. Informer stops if there is a non-recoverable error. Like received
    * a resource that cannot be deserialized.
@@ -326,7 +336,7 @@ default Optional<InformerStoppedHandler> getInformerStoppedHandler() {
    * Override to provide a custom {@link ManagedWorkflowFactory} implementation to change how
    * {@link io.javaoperatorsdk.operator.processing.dependent.workflow.ManagedWorkflow} are
    * instantiated
-   * 
+   *
    * @return the custom {@link ManagedWorkflowFactory} implementation
    */
   @SuppressWarnings("rawtypes")
@@ -336,7 +346,7 @@ default ManagedWorkflowFactory getWorkflowFactory() {
 
   /**
    * Override to provide a custom {@link ExecutorServiceManager} implementation
-   * 
+   *
    * @return the custom {@link ExecutorServiceManager} implementation
    */
   default ExecutorServiceManager getExecutorServiceManager() {
@@ -353,9 +363,8 @@ default ExecutorServiceManager getExecutorServiceManager() {
    * SSA based create/update can be still used with the legacy matching, just overriding the match
    * method of Kubernetes Dependent Resource.
    *
-   * @since 4.4.0
-   *
    * @return if SSA should be used for dependent resources
+   * @since 4.4.0
    */
   default boolean ssaBasedCreateUpdateMatchForDependentResources() {
     return true;
@@ -383,9 +392,8 @@ default Set<Class<? extends HasMetadata>> defaultNonSSAResource() {
    * <p>
    * Disable this if you want to react to your own dependent resource updates
    *
-   * @since 4.5.0
-   *
    * @return if special annotation should be used for dependent resource to filter events
+   * @since 4.5.0
    */
   default boolean previousAnnotationForDependentResourcesEventFiltering() {
     return true;
@@ -400,9 +408,8 @@ default boolean previousAnnotationForDependentResourcesEventFiltering() {
    * logic, and you want to further minimize the amount of work done / updates issued by the
    * operator.
    *
-   * @since 4.5.0
-   *
    * @return if resource version should be parsed (as integer)
+   * @since 4.5.0
    */
   default boolean parseResourceVersionsForEventFilteringAndCaching() {
     return false;
@@ -415,8 +422,8 @@ default boolean parseResourceVersionsForEventFilteringAndCaching() {
    *
    * @return {@code true} if Server-Side Apply (SSA) should be used when patching the primary
    *         resources, {@code false} otherwise
-   * @since 5.0.0
    * @see ConfigurationServiceOverrider#withUseSSAToPatchPrimaryResource(boolean)
+   * @since 5.0.0
    */
   default boolean useSSAToPatchPrimaryResource() {
     return true;
@@ -427,18 +434,17 @@ default boolean useSSAToPatchPrimaryResource() {
    * Determines whether resources retrieved from caches such as via calls to
    * {@link Context#getSecondaryResource(Class)} should be defensively cloned first.
    * </p>
-   * 
+   *
    * <p>
    * Defensive cloning to prevent problematic cache modifications (modifying the resource would
    * otherwise modify the stored copy in the cache) was transparently done in previous JOSDK
    * versions. This might have performance consequences and, with the more prevalent use of
    * Server-Side Apply, where you should create a new copy of your resource with only modified
    * fields, such modifications of these resources are less likely to occur.
    * </p>
-   * 
+   *
    * @return {@code true} if resources should be defensively cloned before returning them from
    *         caches, {@code false} otherwise
-   * 
    * @since 5.0.0
    */
   default boolean cloneSecondaryResourcesWhenGettingFromCache() {

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

@@ -32,6 +32,7 @@ public class ConfigurationServiceOverrider {
   private InformerStoppedHandler informerStoppedHandler;
   private Boolean stopOnInformerErrorDuringStartup;
   private Duration cacheSyncTimeout;
+  private Duration reconciliationTerminationTimeout;
   private Boolean ssaBasedCreateUpdateMatchForDependentResources;
   private Set<Class<? extends HasMetadata>> defaultNonSSAResource;
   private Boolean previousAnnotationForDependentResources;
@@ -127,6 +128,12 @@ public ConfigurationServiceOverrider withCacheSyncTimeout(Duration cacheSyncTime
     return this;
   }
 
+  public ConfigurationServiceOverrider withReconciliationTerminationTimeout(
+      Duration reconciliationTerminationTimeout) {
+    this.reconciliationTerminationTimeout = reconciliationTerminationTimeout;
+    return this;
+  }
+
   public ConfigurationServiceOverrider withSSABasedCreateUpdateMatchForDependentResources(
       boolean value) {
     this.ssaBasedCreateUpdateMatchForDependentResources = value;
@@ -251,6 +258,12 @@ public Duration cacheSyncTimeout() {
         return overriddenValueOrDefault(cacheSyncTimeout, ConfigurationService::cacheSyncTimeout);
       }
 
+      @Override
+      public Duration reconciliationTerminationTimeout() {
+        return overriddenValueOrDefault(reconciliationTerminationTimeout,
+            ConfigurationService::reconciliationTerminationTimeout);
+      }
+
       @Override
       public boolean ssaBasedCreateUpdateMatchForDependentResources() {
         return overriddenValueOrDefault(ssaBasedCreateUpdateMatchForDependentResources,

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.api.config;
 
+import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.Executors;
 
@@ -63,6 +64,7 @@ public <R extends HasMetadata> R clone(R object) {
         .withLeaderElectionConfiguration(new LeaderElectionConfiguration("newLease", "newLeaseNS"))
         .withInformerStoppedHandler((informer, ex) -> {
         })
+        .withReconciliationTerminationTimeout(Duration.ofSeconds(30))
         .build();
 
     assertNotEquals(config.closeClientOnStop(), overridden.closeClientOnStop());
@@ -77,6 +79,8 @@ public <R extends HasMetadata> R clone(R object) {
         overridden.getLeaderElectionConfiguration());
     assertNotEquals(config.getInformerStoppedHandler(),
         overridden.getLeaderElectionConfiguration());
+    assertNotEquals(config.reconciliationTerminationTimeout(),
+        overridden.reconciliationTerminationTimeout());
   }
 
 }

operator-framework/src/test/java/io/javaoperatorsdk/operator/GracefulStopIT.java

@@ -28,7 +28,7 @@ public class GracefulStopIT {
           .build();
 
   @Test
-  void stopsGracefullyWIthTimeout() {
+  void stopsGracefullyWithTimeout() {
     testGracefulStop(TEST_1, RECONCILER_SLEEP, 2);
   }
 

operator-framework/src/test/java/io/javaoperatorsdk/operator/GracefulStopWithConfigurationIT.java

@@ -0,0 +1,74 @@
+package io.javaoperatorsdk.operator;
+
+import java.time.Duration;
+
+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 io.javaoperatorsdk.operator.sample.gracefulstop.GracefulStopTestCustomResource;
+import io.javaoperatorsdk.operator.sample.gracefulstop.GracefulStopTestCustomResourceSpec;
+import io.javaoperatorsdk.operator.sample.gracefulstop.GracefulStopTestReconciler;
+
+import static io.javaoperatorsdk.operator.sample.gracefulstop.GracefulStopTestReconciler.*;
+import static org.assertj.core.api.Assertions.*;
+import static org.awaitility.Awaitility.*;
+
+public class GracefulStopWithConfigurationIT {
+
+  public static final String TEST_1 = "test1";
+
+  @RegisterExtension
+  LocallyRunOperatorExtension operator =
+      LocallyRunOperatorExtension.builder()
+          .withConfigurationService(o -> o.withCloseClientOnStop(false)
+              .withReconciliationTerminationTimeout(Duration.ofMillis(RECONCILER_SLEEP)))
+          .withReconciler(new GracefulStopTestReconciler())
+          .build();
+
+  @Test
+  void stopsGracefullyWithTimeoutConfiguration() {
+    testGracefulStop(TEST_1, 2);
+  }
+
+  private void testGracefulStop(String resourceName, int expectedFinalGeneration) {
+    var testRes = operator.create(testResource(resourceName));
+    await().untilAsserted(() -> {
+      var r = operator.get(GracefulStopTestCustomResource.class, resourceName);
+      assertThat(r.getStatus()).isNotNull();
+      assertThat(r.getStatus().getObservedGeneration()).isEqualTo(1);
+      assertThat(operator.getReconcilerOfType(GracefulStopTestReconciler.class)
+          .getNumberOfExecutions()).isEqualTo(1);
+    });
+
+    testRes.getSpec().setValue(2);
+    operator.replace(testRes);
+
+    await().pollDelay(Duration.ofMillis(50)).untilAsserted(
+        () -> assertThat(operator.getReconcilerOfType(GracefulStopTestReconciler.class)
+            .getNumberOfExecutions()).isEqualTo(2));
+
+    operator.getOperator().stop();
+
+    await().untilAsserted(() -> {
+      var r = operator.get(GracefulStopTestCustomResource.class, resourceName);
+      assertThat(r.getStatus()).isNotNull();
+      assertThat(r.getStatus().getObservedGeneration()).isEqualTo(expectedFinalGeneration);
+    });
+  }
+
+  public GracefulStopTestCustomResource testResource(String name) {
+    GracefulStopTestCustomResource resource =
+        new GracefulStopTestCustomResource();
+    resource.setMetadata(
+        new ObjectMetaBuilder()
+            .withName(name)
+            .withNamespace(operator.getNamespace())
+            .build());
+    resource.setSpec(new GracefulStopTestCustomResourceSpec());
+    resource.getSpec().setValue(1);
+    return resource;
+  }
+
+}