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

.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();
+    }
+  }
 }