[SPARK-52648] Add support for maximal retain duration for Spark application resources

### What changes were proposed in this pull request?

This PR adds support for configuring the maximal retain duration for Spark apps. Working with the resourceRetainPolicy, it enhances the garbage collection mechanism.

### Why are the changes needed?

Current resourceRetainPolicy provides flexibility for retain Spark app resources after its terminated. Introducing maximal retain duration would add one protection layer to avoid terminated resources (pods, config maps .etc) from taking quota in cluster.

### Does this PR introduce _any_ user-facing change?

New configurable field `spec.applicationTolerations.resourceRetainDurationMillis` added to SparkApplication CRD

### How was this patch tested?

CIs - including new unit test and e2e scenario

### Was this patch authored or co-authored using generative AI tooling?

No

Closes #268 from jiangzho/retentionDuration.

Authored-by: Zhou JIANG <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: jiangzho

.github/workflows/build_and_test.yml

@@ -82,6 +82,7 @@ jobs:
           - spark-versions
           - python
           - state-transition
+          - resource-retain-duration
           - watched-namespaces
         exclude:
           - mode: dynamic
@@ -90,6 +91,8 @@ jobs:
             test-group: python
           - mode: dynamic
             test-group: state-transition
+          - mode: dynamic
+            test-group: resource-retain-duration
           - mode: static
             test-group: watched-namespaces
     steps:

build-tools/helm/spark-kubernetes-operator/crds/sparkapplications.spark.apache.org-v1.yaml

@@ -17411,6 +17411,9 @@ spec:
                         minExecutors:
                           type: integer
                       type: object
+                    resourceRetainDurationMillis:
+                      default: -1
+                      type: integer
                     resourceRetainPolicy:
                       enum:
                         - Always

docs/spark_custom_resources.md

@@ -289,12 +289,17 @@ On the other hand, when developing an application, it's possible to configure
 ```yaml
 applicationTolerations:
   # Acceptable values are 'Always', 'OnFailure', 'Never'
+  # Setting this to 'OnFailure' would retain secondary resources if and only if the app fails
   resourceRetainPolicy: OnFailure
+  # Secondary resources would be garbage collected 10 minutes after app termination 
+  resourceRetainDurationMillis: 600000
 ```
 
 to avoid operator attempt to delete driver pod and driver resources if app fails. Similarly,
 if resourceRetainPolicy is set to `Always`, operator would not delete driver resources
-when app ends. Note that this applies only to operator-created resources (driver pod, SparkConf
+when app ends. They would be by default kept with the same lifecycle as the App. It's also
+possible to configure `resourceRetainDurationMillis` to define the maximal retain duration for
+these resources. Note that this applies only to operator-created resources (driver pod, SparkConf
 configmap .etc). You may also want to tune `spark.kubernetes.driver.service.deleteOnTermination`
 and `spark.kubernetes.executor.deleteOnTermination` to control the behavior of driver-created
 resources.

spark-operator-api/src/main/java/org/apache/spark/k8s/operator/Constants.java

@@ -63,6 +63,8 @@ public class Constants {
           + "observed status for details.";
   public static final String APP_CANCELLED_MESSAGE =
       "Spark application has been shutdown as requested.";
+  public static final String APP_EXCEEDED_RETAIN_DURATION_MESSAGE =
+      "Spark application resources released after exceeding the configured retain duration.";
   public static final String DRIVER_UNEXPECTED_REMOVED_MESSAGE =
       "Driver removed. This could caused by 'exit' called in driver process with non-zero "
           + "code, involuntary disruptions or unintentional destroy behavior, check "

spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationTolerations.java

@@ -19,13 +19,18 @@
 
 package org.apache.spark.k8s.operator.spec;
 
+import java.time.Instant;
+
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonInclude;
+import io.fabric8.generator.annotation.Default;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
+import org.apache.spark.k8s.operator.status.ApplicationState;
+
 /** Toleration settings for a Spark application. */
 @Data
 @NoArgsConstructor
@@ -43,4 +48,40 @@ public class ApplicationTolerations {
   @Builder.Default protected ExecutorInstanceConfig instanceConfig = new ExecutorInstanceConfig();
 
   @Builder.Default protected ResourceRetainPolicy resourceRetainPolicy = ResourceRetainPolicy.Never;
+
+  /**
+   * Time-to-live in milliseconds for secondary resources of SparkApplications after termination. If
+   * set to a negative value, secondary resources could be retained with the same lifecycle as the
+   * application according to the retain policy.
+   */
+  @Default("-1")
+  @Builder.Default
+  protected Long resourceRetainDurationMillis = -1L;
+
+  /**
+   * Check whether a terminated application has exceeded the resource retain duration at the
+   * provided instant
+   *
+   * @param lastObservedState last observed state of the application
+   * @return true if the app has terminated and resource retain duration is configured to a positive
+   *     value and the app is not within retain duration; false otherwise.
+   */
+  public boolean exceedRetainDurationAtInstant(
+      ApplicationState lastObservedState, Instant instant) {
+    return lastObservedState != null
+        && lastObservedState.getCurrentStateSummary().isTerminated()
+        && resourceRetainDurationMillis > 0L
+        && Instant.parse(lastObservedState.getLastTransitionTime())
+            .plusMillis(resourceRetainDurationMillis)
+            .isBefore(instant);
+  }
+
+  /**
+   * Indicates whether the reconciler need to perform retain duration check
+   *
+   * @return true `resourceRetainDurationMillis` is set to non-negative value
+   */
+  public boolean isRetainDurationEnabled() {
+    return resourceRetainDurationMillis >= 0L;
+  }
 }

spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/reconcilesteps/AppCleanUpStep.java

@@ -30,7 +30,6 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.Pod;
-import lombok.AllArgsConstructor;
 import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -47,16 +46,22 @@
 import org.apache.spark.k8s.operator.status.ApplicationStatus;
 import org.apache.spark.k8s.operator.utils.ReconcilerUtils;
 import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
+import org.apache.spark.k8s.operator.utils.SparkAppStatusUtils;
 
 /**
  * Cleanup all secondary resources when application is deleted, or at the end of each attempt.
  * Update Application status to indicate whether another attempt would be made.
  */
-@AllArgsConstructor
 @NoArgsConstructor
 @Slf4j
 public class AppCleanUpStep extends AppReconcileStep {
   private Supplier<ApplicationState> onDemandCleanUpReason;
+  private String stateUpdateMessage;
+
+  public AppCleanUpStep(Supplier<ApplicationState> onDemandCleanUpReason) {
+    super();
+    this.onDemandCleanUpReason = onDemandCleanUpReason;
+  }
 
   /**
    * Cleanup secondary resources for an application if needed and updates application status
@@ -88,42 +93,37 @@ public ReconcileProgress reconcile(
     ApplicationStatus currentStatus = application.getStatus();
     ApplicationState currentState = currentStatus.getCurrentState();
     ApplicationTolerations tolerations = application.getSpec().getApplicationTolerations();
-    if (ApplicationStateSummary.ResourceReleased.equals(currentState.getCurrentStateSummary())) {
-      statusRecorder.removeCachedStatus(application);
-      return ReconcileProgress.completeAndNoRequeue();
-    }
-    String stateMessage = null;
-    if (isOnDemandCleanup()) {
-      log.info("Cleaning up application resources on demand");
-    } else {
-      if (ApplicationStateSummary.TerminatedWithoutReleaseResources.equals(
-          currentState.getCurrentStateSummary())) {
-        statusRecorder.removeCachedStatus(application);
-        return ReconcileProgress.completeAndNoRequeue();
-      } else if (currentState.getCurrentStateSummary().isStopping()) {
-        if (retainReleaseResourceForPolicyAndState(
-            tolerations.getResourceRetainPolicy(), currentState)) {
-          if (tolerations.getRestartConfig() != null
-              && !RestartPolicy.Never.equals(tolerations.getRestartConfig().getRestartPolicy())) {
-            stateMessage =
-                "Application is configured to restart, resources created in current "
-                    + "attempt would be force released.";
-            log.warn(stateMessage);
-          } else {
-            ApplicationState terminationState =
-                new ApplicationState(
-                    ApplicationStateSummary.TerminatedWithoutReleaseResources,
-                    "Application is terminated without releasing resources as configured.");
-            long requeueAfterMillis =
-                tolerations.getApplicationTimeoutConfig().getTerminationRequeuePeriodMillis();
-            return appendStateAndRequeueAfter(
-                context, statusRecorder, terminationState, Duration.ofMillis(requeueAfterMillis));
-          }
+    if (currentState.getCurrentStateSummary().isTerminated()) {
+      Optional<ReconcileProgress> terminatedAppProgress =
+          checkEarlyExitForTerminatedApp(application, statusRecorder);
+      if (terminatedAppProgress.isPresent()) {
+        return terminatedAppProgress.get();
+      }
+    } else if (isOnDemandCleanup()) {
+      log.info("Releasing secondary resources for application on demand.");
+    } else if (currentState.getCurrentStateSummary().isStopping()) {
+      if (retainReleaseResourceForPolicyAndState(
+          tolerations.getResourceRetainPolicy(), currentState)) {
+        if (tolerations.getRestartConfig() != null
+            && !RestartPolicy.Never.equals(tolerations.getRestartConfig().getRestartPolicy())) {
+          stateUpdateMessage =
+              "Application is configured to restart, resources created in current "
+                  + "attempt would be force released.";
+          log.warn(stateUpdateMessage);
+        } else {
+          ApplicationState terminationState =
+              new ApplicationState(
+                  ApplicationStateSummary.TerminatedWithoutReleaseResources,
+                  "Application is terminated without releasing resources as configured.");
+          long requeueAfterMillis =
+              tolerations.getApplicationTimeoutConfig().getTerminationRequeuePeriodMillis();
+          return appendStateAndRequeueAfter(
+              context, statusRecorder, terminationState, Duration.ofMillis(requeueAfterMillis));
         }
-      } else {
-        log.debug("Clean up is not expected for app, proceeding to next step.");
-        return ReconcileProgress.proceed();
       }
+    } else {
+      log.debug("Clean up is not expected for app, proceeding to next step.");
+      return ReconcileProgress.proceed();
     }
 
     List<HasMetadata> resourcesToRemove = new ArrayList<>();
@@ -159,8 +159,8 @@ public ReconcileProgress reconcile(
     ApplicationStatus updatedStatus;
     if (onDemandCleanUpReason != null) {
       ApplicationState state = onDemandCleanUpReason.get();
-      if (StringUtils.isNotEmpty(stateMessage)) {
-        state.setMessage(stateMessage);
+      if (StringUtils.isNotEmpty(stateUpdateMessage)) {
+        state.setMessage(stateUpdateMessage);
       }
       long requeueAfterMillis =
           tolerations.getApplicationTimeoutConfig().getTerminationRequeuePeriodMillis();
@@ -171,7 +171,7 @@ public ReconcileProgress reconcile(
           currentStatus.terminateOrRestart(
               tolerations.getRestartConfig(),
               tolerations.getResourceRetainPolicy(),
-              stateMessage,
+              stateUpdateMessage,
               SparkOperatorConf.TRIM_ATTEMPT_STATE_TRANSITION_HISTORY.getValue());
       long requeueAfterMillis =
           tolerations.getApplicationTimeoutConfig().getTerminationRequeuePeriodMillis();
@@ -184,6 +184,41 @@ public ReconcileProgress reconcile(
     }
   }
 
+  protected Optional<ReconcileProgress> checkEarlyExitForTerminatedApp(
+      final SparkApplication application, final SparkAppStatusRecorder statusRecorder) {
+    ApplicationStatus currentStatus = application.getStatus();
+    ApplicationState currentState = currentStatus.getCurrentState();
+    ApplicationTolerations tolerations = application.getSpec().getApplicationTolerations();
+    if (ApplicationStateSummary.ResourceReleased.equals(currentState.getCurrentStateSummary())) {
+      statusRecorder.removeCachedStatus(application);
+      return Optional.of(ReconcileProgress.completeAndNoRequeue());
+    }
+    if (isOnDemandCleanup()) {
+      return Optional.empty();
+    }
+    if (ApplicationStateSummary.TerminatedWithoutReleaseResources.equals(
+        currentState.getCurrentStateSummary())) {
+      if (tolerations.isRetainDurationEnabled()) {
+        Instant now = Instant.now();
+        if (tolerations.exceedRetainDurationAtInstant(currentState, now)) {
+          onDemandCleanUpReason = SparkAppStatusUtils::appExceededRetainDuration;
+          return Optional.empty();
+        } else {
+          Duration nextCheckDuration =
+              Duration.between(
+                  Instant.now(),
+                  Instant.parse(currentState.getLastTransitionTime())
+                      .plusMillis(tolerations.getResourceRetainDurationMillis()));
+          return Optional.of(ReconcileProgress.completeAndRequeueAfter(nextCheckDuration));
+        }
+      } else {
+        statusRecorder.removeCachedStatus(application);
+        return Optional.of(ReconcileProgress.completeAndNoRequeue());
+      }
+    }
+    return Optional.empty();
+  }
+
   protected boolean isOnDemandCleanup() {
     return onDemandCleanUpReason != null;
   }

spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/SparkAppStatusUtils.java

@@ -61,6 +61,11 @@ public static ApplicationState appCancelled() {
         ApplicationStateSummary.ResourceReleased, Constants.APP_CANCELLED_MESSAGE);
   }
 
+  public static ApplicationState appExceededRetainDuration() {
+    return new ApplicationState(
+        ApplicationStateSummary.ResourceReleased, Constants.APP_EXCEEDED_RETAIN_DURATION_MESSAGE);
+  }
+
   public static boolean hasReachedState(
       SparkApplication application, ApplicationState stateToCheck) {
     return isValidApplicationStatus(application)

spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java

@@ -116,4 +116,61 @@ void testCleanupRunningApp() {
       assertTrue(deleteControl.isRemoveFinalizer());
     }
   }
+
+  @SuppressWarnings("PMD.UnusedLocalVariable")
+  @Test
+  void testCleanupAppTerminatedWithoutReleaseResources() {
+    try (MockedConstruction<SparkAppContext> mockAppContext =
+            mockConstruction(
+                SparkAppContext.class,
+                (mock, context) -> {
+                  when(mock.getResource()).thenReturn(app);
+                  when(mock.getClient()).thenReturn(mockClient);
+                  when(mock.getDriverPod()).thenReturn(Optional.of(mockDriver));
+                  when(mock.getDriverPodSpec()).thenReturn(mockDriver);
+                  when(mock.getDriverPreResourcesSpec()).thenReturn(Collections.emptyList());
+                  when(mock.getDriverResourcesSpec()).thenReturn(Collections.emptyList());
+                });
+        MockedStatic<ReconcilerUtils> utils = Mockito.mockStatic(ReconcilerUtils.class)) {
+      // delete app
+      app.setStatus(
+          app.getStatus()
+              .appendNewState(
+                  new ApplicationState(
+                      ApplicationStateSummary.TerminatedWithoutReleaseResources, "")));
+      DeleteControl deleteControl = reconciler.cleanup(app, mockContext);
+      assertFalse(deleteControl.isRemoveFinalizer());
+      utils.verify(() -> ReconcilerUtils.deleteResourceIfExists(mockClient, mockDriver, false));
+      assertEquals(
+          ApplicationStateSummary.ResourceReleased,
+          app.getStatus().getCurrentState().getCurrentStateSummary());
+
+      // proceed delete for terminated app
+      deleteControl = reconciler.cleanup(app, mockContext);
+      assertTrue(deleteControl.isRemoveFinalizer());
+    }
+  }
+
+  @SuppressWarnings("PMD.UnusedLocalVariable")
+  @Test
+  void testCleanupAppTerminatedResourceReleased() {
+    try (MockedConstruction<SparkAppContext> mockAppContext =
+            mockConstruction(
+                SparkAppContext.class,
+                (mock, context) -> {
+                  when(mock.getResource()).thenReturn(app);
+                  when(mock.getClient()).thenReturn(mockClient);
+                  when(mock.getDriverPreResourcesSpec()).thenReturn(Collections.emptyList());
+                  when(mock.getDriverResourcesSpec()).thenReturn(Collections.emptyList());
+                });
+        MockedStatic<ReconcilerUtils> utils = Mockito.mockStatic(ReconcilerUtils.class)) {
+      // delete app
+      app.setStatus(
+          app.getStatus()
+              .appendNewState(new ApplicationState(ApplicationStateSummary.ResourceReleased, "")));
+      DeleteControl deleteControl = reconciler.cleanup(app, mockContext);
+      assertTrue(deleteControl.isRemoveFinalizer());
+      utils.verifyNoInteractions();
+    }
+  }
 }