feat: garbage-collect terminated SparkApplications via operator default TTL

charts/spark-operator-chart/README.md

@@ -94,7 +94,7 @@ See [helm uninstall](https://helm.sh/docs/helm/helm_uninstall) for command docum
 | hook.labels | object | `{}` | Extra labels for the Helm hook Job pod. |
 | hook.annotations | object | `{}` | Extra annotations for the Helm hook Job pod. |
 | controller.replicas | int | `1` | Number of replicas of controller. |
-| controller.featureGates | list | `[{"enabled":false,"name":"PartialRestart"},{"enabled":false,"name":"LoadSparkDefaults"},{"enabled":false,"name":"RestSubmitter"}]` | Feature gates to enable or disable specific features. |
+| controller.featureGates | list | `[{"enabled":false,"name":"PartialRestart"},{"enabled":false,"name":"LoadSparkDefaults"},{"enabled":false,"name":"RestSubmitter"},{"enabled":false,"name":"DefaultTimeToLive"}]` | Feature gates to enable or disable specific features. |
 | controller.revisionHistoryLimit | int | `10` | The number of old history to retain to allow rollback. |
 | controller.leaderElection.enable | bool | `true` | Specifies whether to enable leader election for controller. |
 | controller.leaderElection.leaseDuration | string | `"15s"` | Leader election lease duration. |
@@ -105,6 +105,7 @@ See [helm uninstall](https://helm.sh/docs/helm/helm_uninstall) for command docum
 | controller.logEncoder | string | `"console"` | Configure the encoder of logging, can be one of `console` or `json`. |
 | controller.driverPodCreationGracePeriod | string | `"10s"` | Grace period after a successful spark-submit when driver pod not found errors will be retried. Useful if the driver pod can take some time to be created. |
 | controller.maxTrackedExecutorPerApp | int | `1000` | Specifies the maximum number of Executor pods that can be tracked by the controller per SparkApplication. |
+| controller.defaultTimeToLiveSeconds | int | `0` | Default Time-To-Live (in seconds) applied to terminated SparkApplications that do not set spec.timeToLiveSeconds. Requires the DefaultTimeToLive feature gate to be enabled. 0 (default) disables it. |
 | controller.driverPodDisruptionBudget | object | `{"enable":false}` | Driver PDB feature gate. When true, the controller creates a PodDisruptionBudget for each SparkApplication that sets spec.driverPodDisruptionBudget=true. Default false. |
 | controller.kubeAPIQPS | int | `20` | Maximum QPS to the API server from the controller client. |
 | controller.kubeAPIBurst | int | `30` | Maximum burst for throttle from the controller client. |

charts/spark-operator-chart/templates/controller/deployment.yaml

@@ -132,6 +132,9 @@ spec:
         {{- if .Values.controller.maxTrackedExecutorPerApp }}
         - --max-tracked-executor-per-app={{ .Values.controller.maxTrackedExecutorPerApp }}
         {{- end }}
+        {{- if gt (int .Values.controller.defaultTimeToLiveSeconds) 0 }}
+        - --default-time-to-live-seconds={{ .Values.controller.defaultTimeToLiveSeconds }}
+        {{- end }}
         {{- if .Values.controller.driverPodDisruptionBudget.enable }}
         - --enable-driver-pdb=true
         {{- end }}

charts/spark-operator-chart/tests/controller/deployment_test.yaml

@@ -177,6 +177,33 @@ tests:
           path: spec.template.spec.containers[?(@.name=="spark-operator-controller")].args
           content: --controller-threads=30
 
+  - it: Should contain `--default-time-to-live-seconds` arg if `controller.defaultTimeToLiveSeconds` is set
+    set:
+      controller:
+        defaultTimeToLiveSeconds: 86400
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[?(@.name=="spark-operator-controller")].args
+          content: --default-time-to-live-seconds=86400
+
+  - it: Should not contain `--default-time-to-live-seconds` arg when `controller.defaultTimeToLiveSeconds` is 0
+    set:
+      controller:
+        defaultTimeToLiveSeconds: 0
+    asserts:
+      - notContains:
+          path: spec.template.spec.containers[?(@.name=="spark-operator-controller")].args
+          content: --default-time-to-live-seconds=0
+
+  - it: Should not contain `--default-time-to-live-seconds` arg when `controller.defaultTimeToLiveSeconds` is negative
+    set:
+      controller:
+        defaultTimeToLiveSeconds: -1
+    asserts:
+      - notContains:
+          path: spec.template.spec.containers[?(@.name=="spark-operator-controller")].args
+          content: --default-time-to-live-seconds=-1
+
   - it: Should contain `--enable-ui-service` arg if `controller.uiService.enable` is set to `true`
     set:
       controller:

charts/spark-operator-chart/values.yaml

@@ -95,6 +95,8 @@ controller:
     enabled: false
   - name: RestSubmitter
     enabled: false
+  - name: DefaultTimeToLive
+    enabled: false
 
   # -- The number of old history to retain to allow rollback.
   revisionHistoryLimit: 10
@@ -124,6 +126,11 @@ controller:
   # -- Specifies the maximum number of Executor pods that can be tracked by the controller per SparkApplication.
   maxTrackedExecutorPerApp: 1000
 
+  # -- Default Time-To-Live (in seconds) applied to terminated SparkApplications that do not
+  # set spec.timeToLiveSeconds. Requires the DefaultTimeToLive feature gate to be enabled.
+  # 0 (default) disables it.
+  defaultTimeToLiveSeconds: 0
+
   # -- Driver PDB feature gate. When true, the controller creates a PodDisruptionBudget for each SparkApplication that sets spec.driverPodDisruptionBudget=true. Default false.
   driverPodDisruptionBudget:
     enable: false

cmd/operator/controller/start.go

@@ -85,6 +85,7 @@ var (
 	controllerThreads        int
 	cacheSyncTimeout         time.Duration
 	maxTrackedExecutorPerApp int
+	defaultTimeToLiveSeconds int64
 
 	// Driver PDB feature gate. When enabled, the controller creates a
 	// PodDisruptionBudget for each SparkApplication that sets
@@ -205,6 +206,10 @@ func NewStartCommand() *cobra.Command {
 	command.Flags().StringVar(&namespaceSelector, "namespace-selector", "", "Label selector for namespaces to watch (e.g., 'spark-operator=enabled,env in (prod,staging)'). Namespaces matching this selector will be watched in addition to those specified via --namespaces. Requires ClusterRole permission to list and watch namespaces.")
 	command.Flags().DurationVar(&cacheSyncTimeout, "cache-sync-timeout", 30*time.Second, "Informer cache sync timeout.")
 	command.Flags().IntVar(&maxTrackedExecutorPerApp, "max-tracked-executor-per-app", 1000, "The maximum number of tracked executors per SparkApplication.")
+	command.Flags().Int64Var(&defaultTimeToLiveSeconds, "default-time-to-live-seconds", 0,
+		"Default Time-To-Live in seconds applied to terminated SparkApplications that do "+
+			"not set spec.timeToLiveSeconds. Requires the DefaultTimeToLive feature gate. "+
+			"0 (default) or negative disables it.")
 	command.Flags().BoolVar(&enableDriverPDB, "enable-driver-pdb", false,
 		"Enable creation of a PodDisruptionBudget for Spark driver pods. "+
 			"Each SparkApplication must additionally opt in via "+
@@ -528,6 +533,7 @@ func newSparkApplicationReconcilerOptions() sparkapplication.Options {
 		SparkExecutorMetrics:         sparkExecutorMetrics,
 		MaxTrackedExecutorPerApp:     maxTrackedExecutorPerApp,
 		EnableDriverPDB:              enableDriverPDB,
+		DefaultTimeToLiveSeconds:     defaultTimeToLiveSeconds,
 	}
 	if enableBatchScheduler {
 		options.KubeSchedulerNames = kubeSchedulerNames

internal/controller/sparkapplication/controller.go

@@ -78,6 +78,12 @@ type Options struct {
 	// When false, the reconciler will not create or delete a PDB regardless of
 	// the SparkApplication spec. Defaults to false.
 	EnableDriverPDB bool
+
+	// DefaultTimeToLiveSeconds is the operator-wide default TTL (in seconds) applied to
+	// terminated SparkApplications that do not set spec.timeToLiveSeconds. Only honored
+	// when the DefaultTimeToLive feature gate is enabled and the value is > 0. Applied as
+	// a runtime fallback for the cleanup decision; the object's spec is never modified.
+	DefaultTimeToLiveSeconds int64
 }
 
 // Reconciler reconciles a SparkApplication object.
@@ -714,7 +720,13 @@ func (r *Reconciler) reconcileTerminatedSparkApplication(ctx context.Context, re
 		return ctrl.Result{}, nil
 	}
 
-	if util.IsExpired(app) {
+	effectiveTTLSeconds, usedDefault := util.EffectiveTimeToLiveSeconds(app, r.options.DefaultTimeToLiveSeconds)
+	if usedDefault {
+		logger.Info("Applying operator default TTL to terminated SparkApplication",
+			"ttlSeconds", *effectiveTTLSeconds, "state", app.Status.AppState.State)
+	}
+
+	if util.IsExpiredWithTTL(app, effectiveTTLSeconds) {
 		logger.Info("Deleting expired SparkApplication", "state", app.Status.AppState.State)
 		if err := r.client.Delete(ctx, app); err != nil {
 			return ctrl.Result{Requeue: true}, err
@@ -735,14 +747,14 @@ func (r *Reconciler) reconcileTerminatedSparkApplication(ctx context.Context, re
 		return ctrl.Result{Requeue: true}, err
 	}
 
-	// If termination time or TTL is not set, will not requeue this application.
-	if app.Status.TerminationTime.IsZero() || app.Spec.TimeToLiveSeconds == nil || *app.Spec.TimeToLiveSeconds <= 0 {
+	// If termination time or effective TTL is not set, will not requeue this application.
+	if app.Status.TerminationTime.IsZero() || effectiveTTLSeconds == nil || *effectiveTTLSeconds <= 0 {
 		return ctrl.Result{}, nil
 	}
 
 	// Otherwise, requeue the application for subsequent deletion.
 	now := time.Now()
-	ttl := time.Duration(*app.Spec.TimeToLiveSeconds) * time.Second
+	ttl := time.Duration(*effectiveTTLSeconds) * time.Second
 	survival := now.Sub(app.Status.TerminationTime.Time)
 
 	// If survival time is greater than TTL, requeue the application immediately.

internal/controller/sparkapplication/controller_test.go

@@ -38,6 +38,7 @@ import (
 	"github.com/kubeflow/spark-operator/v2/api/v1beta2"
 	"github.com/kubeflow/spark-operator/v2/internal/controller/sparkapplication"
 	"github.com/kubeflow/spark-operator/v2/pkg/common"
+	"github.com/kubeflow/spark-operator/v2/pkg/features"
 	"github.com/kubeflow/spark-operator/v2/pkg/util"
 )
 
@@ -535,6 +536,63 @@ var _ = Describe("SparkApplication Controller", func() {
 		})
 	})
 
+	Context("When reconciling a terminated SparkApplication with the operator default TTL", func() {
+		ctx := context.Background()
+		appName := "test-default-ttl"
+		appNamespace := "default"
+		key := types.NamespacedName{Name: appName, Namespace: appNamespace}
+
+		BeforeEach(func() {
+			Expect(features.SetEnable(features.DefaultTimeToLive, true)).To(Succeed())
+			DeferCleanup(func() {
+				Expect(features.SetEnable(features.DefaultTimeToLive, false)).To(Succeed())
+			})
+
+			// Create the SparkApplication if it does not already exist.
+			app := &v1beta2.SparkApplication{}
+			err := k8sClient.Get(ctx, key, app)
+			if errors.IsNotFound(err) {
+				app = &v1beta2.SparkApplication{
+					ObjectMeta: metav1.ObjectMeta{Name: appName, Namespace: appNamespace},
+					Spec: v1beta2.SparkApplicationSpec{
+						MainApplicationFile: ptr.To("local:///dummy.jar"),
+					},
+				}
+				v1beta2.SetSparkApplicationDefaults(app)
+				Expect(k8sClient.Create(ctx, app)).To(Succeed())
+			} else {
+				Expect(err).NotTo(HaveOccurred())
+			}
+
+			// Unconditionally drive the app into a terminated state whose TTL has already
+			// elapsed, so the precondition holds regardless of any pre-existing object.
+			Expect(k8sClient.Get(ctx, key, app)).To(Succeed())
+			app.Status.AppState.State = v1beta2.ApplicationStateCompleted
+			app.Status.TerminationTime = metav1.NewTime(time.Now().Add(-2 * time.Minute))
+			Expect(k8sClient.Status().Update(ctx, app)).To(Succeed())
+		})
+
+		AfterEach(func() {
+			app := &v1beta2.SparkApplication{}
+			Expect(errors.IsNotFound(k8sClient.Get(ctx, key, app))).To(BeTrue())
+		})
+
+		It("Should delete a terminated app past the operator default TTL when it has no spec TTL", func() {
+			reconciler := sparkapplication.NewReconciler(
+				nil,
+				k8sClient.Scheme(),
+				k8sClient,
+				nil,
+				nil,
+				&sparkapplication.SparkSubmitter{},
+				sparkapplication.Options{Namespaces: []string{appNamespace}, DefaultTimeToLiveSeconds: 60},
+			)
+			result, err := reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: key})
+			Expect(err).NotTo(HaveOccurred())
+			Expect(result.Requeue).To(BeFalse())
+		})
+	})
+
 	Context("When reconciling a failed SparkApplication", func() {
 		ctx := context.Background()
 		appName := "test"

pkg/features/features.go

@@ -49,6 +49,15 @@ const (
 	// owner: @venkomirisetti
 	// alpha: v2.6.0
 	RestSubmitter featuregate.Feature = "RestSubmitter"
+
+	// DefaultTimeToLive enables applying an operator-configured default TTL to
+	// terminated SparkApplications that do not set spec.timeToLiveSeconds. The
+	// value is used only for the cleanup decision (a runtime fallback) and never
+	// mutates the object.
+	//
+	// owner: @dineshkumar181094
+	// alpha: v2.6.0
+	DefaultTimeToLive featuregate.Feature = "DefaultTimeToLive"
 )
 
 // To add a new feature gate, follow these steps:
@@ -91,6 +100,8 @@ var defaultFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{
 	LoadSparkDefaults: {Default: false, PreRelease: featuregate.Alpha},
 
 	RestSubmitter: {Default: false, PreRelease: featuregate.Alpha},
+
+	DefaultTimeToLive: {Default: false, PreRelease: featuregate.Alpha},
 }
 
 // SetFeatureGateDuringTest sets the specified feature gate to the specified value during a test.

pkg/features/features_test.go

@@ -61,6 +61,14 @@ func TestSetEnableWithRegisteredFeature(t *testing.T) {
 	assert.False(t, Enabled(testFeature))
 }
 
+func TestDefaultTimeToLiveGateRegisteredAndOffByDefault(t *testing.T) {
+	// The gate must be registered (init ran) and default to disabled.
+	assert.False(t, Enabled(DefaultTimeToLive))
+
+	SetFeatureGateDuringTest(t, DefaultTimeToLive, true)
+	assert.True(t, Enabled(DefaultTimeToLive))
+}
+
 func TestSetFeatureGateDuringTestHelper(t *testing.T) {
 	// Register a test feature for this test
 	testFeature := featuregate.Feature("TestFeatureForDuringTest")

pkg/util/sparkapplication.go

@@ -32,6 +32,7 @@ import (
 
 	"github.com/kubeflow/spark-operator/v2/api/v1beta2"
 	"github.com/kubeflow/spark-operator/v2/pkg/common"
+	"github.com/kubeflow/spark-operator/v2/pkg/features"
 )
 
 // GetDriverPodName returns name of the driver pod of the given spark application.
@@ -60,14 +61,26 @@ func IsTerminated(app *v1beta2.SparkApplication) bool {
 		app.Status.AppState.State == v1beta2.ApplicationStateFailed
 }
 
-// IsExpired returns whether the given SparkApplication is expired.
+// IsExpired returns whether the given SparkApplication is expired according to its
+// own spec.timeToLiveSeconds.
 func IsExpired(app *v1beta2.SparkApplication) bool {
+	return IsExpiredWithTTL(app, app.Spec.TimeToLiveSeconds)
+}
+
+// IsExpiredWithTTL returns whether the given terminated SparkApplication has outlived
+// the provided TTL. The TTL may originate from the user's spec or from an
+// operator-configured default (see EffectiveTimeToLiveSeconds).
+//
+//   - A nil ttlSeconds means no TTL is defined: the application never expires.
+//   - A non-nil ttlSeconds that is <= 0 means the application expires immediately
+//     once it has a termination time.
+func IsExpiredWithTTL(app *v1beta2.SparkApplication, ttlSeconds *int64) bool {
 	// The application has no TTL defined and will never expire.
-	if app.Spec.TimeToLiveSeconds == nil {
+	if ttlSeconds == nil {
 		return false
 	}
 
-	ttl := time.Duration(*app.Spec.TimeToLiveSeconds) * time.Second
+	ttl := time.Duration(*ttlSeconds) * time.Second
 	now := time.Now()
 	if !app.Status.TerminationTime.IsZero() && now.Sub(app.Status.TerminationTime.Time) > ttl {
 		return true
@@ -76,6 +89,29 @@ func IsExpired(app *v1beta2.SparkApplication) bool {
 	return false
 }
 
+// EffectiveTimeToLiveSeconds returns the TTL (in seconds) that should govern cleanup
+// of the given SparkApplication, and whether the operator default (rather than the
+// user spec) was the source.
+//
+// Resolution order:
+//   - the user's spec.timeToLiveSeconds whenever it is set (a value <= 0 is an
+//     explicit request to expire immediately and still wins), else
+//   - the operator default when the DefaultTimeToLive feature gate is enabled and
+//     defaultSeconds > 0, else
+//   - nil, meaning "never expire".
+//
+// It never mutates the SparkApplication.
+func EffectiveTimeToLiveSeconds(app *v1beta2.SparkApplication, defaultSeconds int64) (*int64, bool) {
+	if app.Spec.TimeToLiveSeconds != nil {
+		return app.Spec.TimeToLiveSeconds, false
+	}
+	if features.Enabled(features.DefaultTimeToLive) && defaultSeconds > 0 {
+		seconds := defaultSeconds
+		return &seconds, true
+	}
+	return nil, false
+}
+
 // IsDriverRunning returns whether the driver pod of the given SparkApplication is running.
 func IsDriverRunning(app *v1beta2.SparkApplication) bool {
 	return app.Status.AppState.State == v1beta2.ApplicationStateRunning