Repository: kubernetes-sigs/kueue

Issue:
Optional garbage collection of finished Workloads
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**:

I would like to have an manager option to delete workload resources as soon as (or with a ttl) the scheduled Job is finished.

**Why is this needed**:

 I changed a configuration to retain more history of job executions of a cronjob, and the memory consumption of the kueue-manager more than doubled:

![Bildschirmfoto 2024-01-19 um 11 57 50](https://github.com/kubernetes-sigs/kueue/assets/2535856/c6ff8083-f69e-4268-b24a-edf7a16f3fc7)

**Completion requirements**:

This enhancement requires the following artifacts:

- [ ] Design doc
- [ ] API change
- [ ] Docs update

The artifacts should be linked in subsequent comments.


Hints from issue comments:
@woehrl01: Not sure if this is related, but I also found out that after that change of keeping history ( and not having a ttl on jobs), Kueue stopped working and showing an insane amount of admitted workloads (using v0.5.2)


![Bildschirmfoto 2024-01-19 um 12 16 28](https://github.com/kubernetes-sigs/kueue/assets/2535856/6b733219-9ba5-4dbc-9260-a24f6c509292)

Deleting all the succeeded jobs by hand recovered that.
@woehrl01: I guess the admitted workload bug this is fixed by #1654. It would be still nice to remove the workload resource all together.@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle rotten

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@alculquicondor: Are you asking to just delete the Workload but keep the parent Job (or Job CRD)?@woehrl01: @alculquicondor yes. That was the idea. The workload will be deleted eventually, but this would free up etcd storage until ttl of the job crd has been reached. @alculquicondor: /retitle Optional garbage collection of finished Workloads

🤔 maybe we can also do this for orphan Workloads #1789@mwysokin: /assign@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.

This bot triages issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Reopen this issue with `/reopen`
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/close not-planned

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-ci-robot: @k8s-triage-robot: Closing this issue, marking it as "Not Planned".

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/1618#issuecomment-2263301502):

>The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.
>
>This bot triages issues according to the following rules:
>- After 90d of inactivity, `lifecycle/stale` is applied
>- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
>- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed
>
>You can:
>- Reopen this issue with `/reopen`
>- Mark this issue as fresh with `/remove-lifecycle rotten`
>- Offer to help out with [Issue Triage][1]
>
>Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).
>
>/close not-planned
>
>[1]: https://www.kubernetes.dev/docs/guide/issue-triage/


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@kannon92: /reopen@k8s-ci-robot: @kannon92: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/1618#issuecomment-2263856566):

>/reopen


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@kannon92: Opening due to the work seems to be in flight.@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.

This bot triages issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Reopen this issue with `/reopen`
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/close not-planned

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-ci-robot: @k8s-triage-robot: Closing this issue, marking it as "Not Planned".

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/1618#issuecomment-2323035097):

>The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.
>
>This bot triages issues according to the following rules:
>- After 90d of inactivity, `lifecycle/stale` is applied
>- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
>- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed
>
>You can:
>- Reopen this issue with `/reopen`
>- Mark this issue as fresh with `/remove-lifecycle rotten`
>- Offer to help out with [Issue Triage][1]
>
>Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).
>
>/close not-planned
>
>[1]: https://www.kubernetes.dev/docs/guide/issue-triage/


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@woehrl01: /reopen@k8s-ci-robot: @woehrl01: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/kueue/issues/1618#issuecomment-2323038437):

>/reopen


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@woehrl01: /remove-lifecycle rotten@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle rotten

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@mimowo: /remove-lifecycle rotten
cc @mwysokin would you like to continue working on this?@mykysha: /assign@k8s-ci-robot: You must be a member of the [kubernetes-sigs/kueue-maintainers](https://github.com/orgs/kubernetes-sigs/teams/kueue-maintainers/members) GitHub team to add status labels. If you believe you should be able to issue the /status command, please contact your Kueue Maintainers and have them propose you as an additional delegate for this responsibility.

Solution patch:
diff --git a/apis/config/v1beta1/configuration_types.go b/apis/config/v1beta1/configuration_types.go
index 89b045148..23ce6fd8d 100644
--- a/apis/config/v1beta1/configuration_types.go
+++ b/apis/config/v1beta1/configuration_types.go
@@ -506,6 +506,14 @@ type ObjectRetentionPolicies struct {
 
 // WorkloadRetentionPolicy defines the policies for when Workloads should be deleted.
 type WorkloadRetentionPolicy struct {
+	// AfterFinished is the duration to wait after a Workload finishes
+	// before deleting it.
+	// A duration of 0 will delete immediately.
+	// A nil value disables automatic deletion.
+	// Represented using metav1.Duration (e.g. "10m", "1h30m").
+	// +optional
+	AfterFinished *metav1.Duration `json:"afterFinished,omitempty"`
+
 	// AfterDeactivatedByKueue is the duration to wait after *any* Kueue-managed Workload
 	// (such as a Job, JobSet, or other custom workload types) has been marked
 	// as deactivated by Kueue before automatically deleting it.
diff --git a/apis/config/v1beta1/zz_generated.deepcopy.go b/apis/config/v1beta1/zz_generated.deepcopy.go
index ba96d6e59..293cb2c28 100644
--- a/apis/config/v1beta1/zz_generated.deepcopy.go
+++ b/apis/config/v1beta1/zz_generated.deepcopy.go
@@ -596,6 +596,11 @@ func (in *WaitForPodsReady) DeepCopy() *WaitForPodsReady {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *WorkloadRetentionPolicy) DeepCopyInto(out *WorkloadRetentionPolicy) {
 	*out = *in
+	if in.AfterFinished != nil {
+		in, out := &in.AfterFinished, &out.AfterFinished
+		*out = new(v1.Duration)
+		**out = **in
+	}
 	if in.AfterDeactivatedByKueue != nil {
 		in, out := &in.AfterDeactivatedByKueue, &out.AfterDeactivatedByKueue
 		*out = new(v1.Duration)
diff --git a/charts/kueue/values.yaml b/charts/kueue/values.yaml
index c47109e88..090fccf55 100644
--- a/charts/kueue/values.yaml
+++ b/charts/kueue/values.yaml
@@ -132,6 +132,7 @@ managerConfig:
     #     example.com/accelerator-gpc: 4
     #objectRetentionPolicies:
     #  workloads:
+    #    afterFinished: null # null indicates infinite retention, 0s means no retention at all
     #    afterDeactivatedByKueue: null # null indicates infinite retention, 0s means no retention at all
 # ports definition for metricsService and webhookService.
 metricsService:
diff --git a/config/components/manager/controller_manager_config.yaml b/config/components/manager/controller_manager_config.yaml
index dc24e28a9..799fffa60 100644
--- a/config/components/manager/controller_manager_config.yaml
+++ b/config/components/manager/controller_manager_config.yaml
@@ -74,4 +74,5 @@ integrations:
 #      example.com/accelerator-gpc: 4
 #objectRetentionPolicies:
 #  workloads:
+#    afterFinished: null # null indicates infinite retention, 0s means no retention at all
 #    afterDeactivatedByKueue: null # null indicates infinite retention, 0s means no retention at all
diff --git a/pkg/config/validation.go b/pkg/config/validation.go
index 5e1249b1d..fffc2b5c6 100644
--- a/pkg/config/validation.go
+++ b/pkg/config/validation.go
@@ -414,6 +414,10 @@ func validateObjectRetentionPolicies(c *configapi.Configuration) field.ErrorList
 	if rr == nil || rr.Workloads == nil {
 		return allErrs
 	}
+	if rr.Workloads.AfterFinished != nil && rr.Workloads.AfterFinished.Duration < 0 {
+		allErrs = append(allErrs, field.Invalid(objectRetentionPoliciesWorkloadsPath.Child("afterFinished"),
+			c.ObjectRetentionPolicies.Workloads.AfterFinished, apimachineryvalidation.IsNegativeErrorMsg))
+	}
 	if rr.Workloads.AfterDeactivatedByKueue != nil && rr.Workloads.AfterDeactivatedByKueue.Duration < 0 {
 		allErrs = append(allErrs, field.Invalid(objectRetentionPoliciesWorkloadsPath.Child("afterDeactivatedByKueue"),
 			c.ObjectRetentionPolicies.Workloads.AfterDeactivatedByKueue.Duration.String(), apimachineryvalidation.IsNegativeErrorMsg))
diff --git a/pkg/controller/core/core.go b/pkg/controller/core/core.go
index 3647d00d3..397df17ea 100644
--- a/pkg/controller/core/core.go
+++ b/pkg/controller/core/core.go
@@ -86,6 +86,7 @@ func SetupControllers(mgr ctrl.Manager, qManager *queue.Manager, cc *cache.Cache
 		mgr.GetEventRecorderFor(constants.WorkloadControllerName),
 		WithWorkloadUpdateWatchers(qRec, cqRec),
 		WithWaitForPodsReady(waitForPodsReady(cfg.WaitForPodsReady)),
+		WithWorkloadRetention(workloadRetention(cfg.ObjectRetentionPolicies)),
 	).SetupWithManager(mgr, cfg); err != nil {
 		return "Workload", err
 	}
@@ -112,6 +113,16 @@ func waitForPodsReady(cfg *configapi.WaitForPodsReady) *waitForPodsReadyConfig {
 	return &result
 }
 
+func workloadRetention(cfg *configapi.ObjectRetentionPolicies) *workloadRetentionConfig {
+	if cfg == nil || cfg.Workloads == nil || cfg.Workloads.AfterFinished == nil {
+		return nil
+	}
+
+	return &workloadRetentionConfig{
+		afterFinished: &cfg.Workloads.AfterFinished.Duration,
+	}
+}
+
 func queueVisibilityUpdateInterval(cfg *configapi.Configuration) time.Duration {
 	if cfg.QueueVisibility != nil {
 		return time.Duration(cfg.QueueVisibility.UpdateIntervalSeconds) * time.Second
diff --git a/pkg/controller/core/workload_controller.go b/pkg/controller/core/workload_controller.go
index 215db21b3..8a379a3d5 100644
--- a/pkg/controller/core/workload_controller.go
+++ b/pkg/controller/core/workload_controller.go
@@ -73,9 +73,14 @@ type waitForPodsReadyConfig struct {
 	requeuingBackoffJitter      float64
 }
 
+type workloadRetentionConfig struct {
+	afterFinished *time.Duration
+}
+
 type options struct {
-	watchers               []WorkloadUpdateWatcher
-	waitForPodsReadyConfig *waitForPodsReadyConfig
+	watchers                []WorkloadUpdateWatcher
+	waitForPodsReadyConfig  *waitForPodsReadyConfig
+	workloadRetentionConfig *workloadRetentionConfig
 }
 
 // Option configures the reconciler.
@@ -95,6 +100,13 @@ func WithWorkloadUpdateWatchers(value ...WorkloadUpdateWatcher) Option {
 	}
 }
 
+// WithWorkloadRetention allows to specify retention for workload resources
+func WithWorkloadRetention(value *workloadRetentionConfig) Option {
+	return func(o *options) {
+		o.workloadRetentionConfig = value
+	}
+}
+
 var defaultOptions = options{}
 
 type WorkloadUpdateWatcher interface {
@@ -103,14 +115,15 @@ type WorkloadUpdateWatcher interface {
 
 // WorkloadReconciler reconciles a Workload object
 type WorkloadReconciler struct {
-	log              logr.Logger
-	queues           *queue.Manager
-	cache            *cache.Cache
-	client           client.Client
-	watchers         []WorkloadUpdateWatcher
-	waitForPodsReady *waitForPodsReadyConfig
-	recorder         record.EventRecorder
-	clock            clock.Clock
+	log               logr.Logger
+	queues            *queue.Manager
+	cache             *cache.Cache
+	client            client.Client
+	watchers          []WorkloadUpdateWatcher
+	waitForPodsReady  *waitForPodsReadyConfig
+	recorder          record.EventRecorder
+	clock             clock.Clock
+	workloadRetention *workloadRetentionConfig
 }
 
 var _ reconcile.Reconciler = (*WorkloadReconciler)(nil)
@@ -123,14 +136,15 @@ func NewWorkloadReconciler(client client.Client, queues *queue.Manager, cache *c
 	}
 
 	return &WorkloadReconciler{
-		log:              ctrl.Log.WithName("workload-reconciler"),
-		client:           client,
-		queues:           queues,
-		cache:            cache,
-		watchers:         options.watchers,
-		waitForPodsReady: options.waitForPodsReadyConfig,
-		recorder:         recorder,
-		clock:            realClock,
+		log:               ctrl.Log.WithName("workload-reconciler"),
+		client:            client,
+		queues:            queues,
+		cache:             cache,
+		watchers:          options.watchers,
+		waitForPodsReady:  options.waitForPodsReadyConfig,
+		recorder:          recorder,
+		clock:             realClock,
+		workloadRetention: options.workloadRetentionConfig,
 	}
 }
 
@@ -152,10 +166,29 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	log.V(2).Info("Reconcile Workload")
 
 	if len(wl.OwnerReferences) == 0 && !wl.DeletionTimestamp.IsZero() {
+		// manual deletion triggered by the user
 		return ctrl.Result{}, workload.RemoveFinalizer(ctx, r.client, &wl)
 	}
 
-	if apimeta.IsStatusConditionTrue(wl.Status.Conditions, kueue.WorkloadFinished) {
+	finishedCond := apimeta.FindStatusCondition(wl.Status.Conditions, kueue.WorkloadFinished)
+	if finishedCond != nil && finishedCond.Status == metav1.ConditionTrue {
+		if !features.Enabled(features.ObjectRetentionPolicies) || r.workloadRetention == nil || r.workloadRetention.afterFinished == nil {
+			return ctrl.Result{}, nil
+		}
+
+		now := r.clock.Now()
+		expirationTime := finishedCond.LastTransitionTime.Add(*r.workloadRetention.afterFinished)
+		if now.Before(expirationTime) {
+			remainingTime := expirationTime.Sub(now)
+			log.V(3).Info("Requeueing workload for deletion after retention period", "remainingTime", remainingTime)
+			return ctrl.Result{RequeueAfter: remainingTime}, nil
+		}
+
+		log.V(2).Info("Deleting workload because it has finished and the retention period has elapsed", "retention", *r.workloadRetention.afterFinished)
+		if err := r.client.Delete(ctx, &wl); err != nil {
+			return ctrl.Result{}, client.IgnoreNotFound(err)
+		}
+		r.recorder.Eventf(&wl, corev1.EventTypeNormal, "Deleted", "Deleted finished workload due to elapsed retention")
 		return ctrl.Result{}, nil
 	}
 
diff --git a/site/content/en/docs/reference/kueue-config.v1beta1.md b/site/content/en/docs/reference/kueue-config.v1beta1.md
index ad8c36a40..2e8a7adeb 100644
--- a/site/content/en/docs/reference/kueue-config.v1beta1.md
+++ b/site/content/en/docs/reference/kueue-config.v1beta1.md
@@ -1055,6 +1055,17 @@ If not set, there is no timeout.</p>
 <tbody>
     
   
+<tr><td><code>afterFinished</code><br/>
+<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.28/#duration-v1-meta"><code>k8s.io/apimachinery/pkg/apis/meta/v1.Duration</code></a>
+</td>
+<td>
+   <p>AfterFinished is the duration to wait after a Workload finishes
+before deleting it.
+A duration of 0 will delete immediately.
+A nil value disables automatic deletion.
+Represented using metav1.Duration (e.g. &quot;10m&quot;, &quot;1h30m&quot;).</p>
+</td>
+</tr>
 <tr><td><code>afterDeactivatedByKueue</code><br/>
 <a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.28/#duration-v1-meta"><code>k8s.io/apimachinery/pkg/apis/meta/v1.Duration</code></a>
 </td>



Test patch:
diff --git a/apis/config/v1beta1/defaults_test.go b/apis/config/v1beta1/defaults_test.go
index 0cb033ff7..ecf85c93b 100644
--- a/apis/config/v1beta1/defaults_test.go
+++ b/apis/config/v1beta1/defaults_test.go
@@ -600,6 +600,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				},
 				ObjectRetentionPolicies: &ObjectRetentionPolicies{
 					Workloads: &WorkloadRetentionPolicy{
+						AfterFinished:           &metav1.Duration{Duration: 30 * time.Minute},
 						AfterDeactivatedByKueue: &metav1.Duration{Duration: 30 * time.Minute},
 					},
 				},
@@ -617,6 +618,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				ObjectRetentionPolicies: &ObjectRetentionPolicies{
 					Workloads: &WorkloadRetentionPolicy{
+						AfterFinished:           &metav1.Duration{Duration: 30 * time.Minute},
 						AfterDeactivatedByKueue: &metav1.Duration{Duration: 30 * time.Minute},
 					},
 				},
diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go
index 6d1c60e69..42ef53023 100644
--- a/pkg/config/config_test.go
+++ b/pkg/config/config_test.go
@@ -334,6 +334,7 @@ kind: Configuration
 namespace: kueue-system
 objectRetentionPolicies:
   workloads: 
+    afterFinished: 30m
     afterDeactivatedByKueue: 30m
 `), os.FileMode(0600)); err != nil {
 		t.Fatal(err)
@@ -960,6 +961,7 @@ objectRetentionPolicies:
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				ObjectRetentionPolicies: &configapi.ObjectRetentionPolicies{
 					Workloads: &configapi.WorkloadRetentionPolicy{
+						AfterFinished:           &metav1.Duration{Duration: 30 * time.Minute},
 						AfterDeactivatedByKueue: &metav1.Duration{Duration: 30 * time.Minute},
 					},
 				},
diff --git a/pkg/config/validation_test.go b/pkg/config/validation_test.go
index 9a292dd8a..7138a0abe 100644
--- a/pkg/config/validation_test.go
+++ b/pkg/config/validation_test.go
@@ -797,6 +797,42 @@ func TestValidate(t *testing.T) {
 				},
 			},
 		},
+		"negative afterFinished in .objectRetentionPolicies.workloads": {
+			cfg: &configapi.Configuration{
+				Integrations: defaultIntegrations,
+				ObjectRetentionPolicies: &configapi.ObjectRetentionPolicies{
+					Workloads: &configapi.WorkloadRetentionPolicy{
+						AfterFinished: ptr.To(metav1.Duration{Duration: -1}),
+					},
+				},
+			},
+			wantErr: field.ErrorList{
+				&field.Error{
+					Type:  field.ErrorTypeInvalid,
+					Field: "objectRetentionPolicies.workloads.afterFinished",
+				},
+			},
+		},
+		"zero afterFinished in .objectRetentionPolicies.workloads": {
+			cfg: &configapi.Configuration{
+				Integrations: defaultIntegrations,
+				ObjectRetentionPolicies: &configapi.ObjectRetentionPolicies{
+					Workloads: &configapi.WorkloadRetentionPolicy{
+						AfterFinished: ptr.To(metav1.Duration{Duration: 0}),
+					},
+				},
+			},
+		},
+		"positive afterFinished in .objectRetentionPolicies.workloads": {
+			cfg: &configapi.Configuration{
+				Integrations: defaultIntegrations,
+				ObjectRetentionPolicies: &configapi.ObjectRetentionPolicies{
+					Workloads: &configapi.WorkloadRetentionPolicy{
+						AfterFinished: ptr.To(metav1.Duration{Duration: 1}),
+					},
+				},
+			},
+		},
 		"negative afterDeactivatedByKueue in .objectRetentionPolicies.workloads": {
 			cfg: &configapi.Configuration{
 				Integrations: defaultIntegrations,
diff --git a/pkg/controller/core/workload_controller_test.go b/pkg/controller/core/workload_controller_test.go
index bd85707fc..9c1796232 100644
--- a/pkg/controller/core/workload_controller_test.go
+++ b/pkg/controller/core/workload_controller_test.go
@@ -38,8 +38,10 @@ import (
 
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/cache"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/queue"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	"sigs.k8s.io/kueue/test/util"
 )
 
 func TestAdmittedNotReadyWorkload(t *testing.T) {
@@ -395,6 +397,8 @@ func TestReconcile(t *testing.T) {
 	fakeClock := testingclock.NewFakeClock(testStartTime)
 
 	cases := map[string]struct {
+		enableObjectRetentionPolicies bool
+
 		workload       *kueue.Workload
 		cq             *kueue.ClusterQueue
 		lq             *kueue.LocalQueue
@@ -1696,9 +1700,111 @@ func TestReconcile(t *testing.T) {
 				},
 			},
 		},
+		"shouldn't delete the workload because, object retention not configured": {
+			enableObjectRetentionPolicies: true,
+			workload: utiltesting.MakeWorkload("wl", "ns").
+				Condition(metav1.Condition{
+					Type:   kueue.WorkloadFinished,
+					Status: metav1.ConditionTrue,
+				}).
+				Obj(),
+			reconcilerOpts: []Option{
+				WithWorkloadRetention(nil),
+			},
+			wantWorkload: utiltesting.MakeWorkload("wl", "ns").
+				Condition(metav1.Condition{
+					Type:   kueue.WorkloadFinished,
+					Status: metav1.ConditionTrue,
+				}).
+				Obj(),
+			wantError: nil,
+		},
+		"shouldn't try to delete the workload (no event emitted) because it is already being deleted by kubernetes, object retention configured": {
+			enableObjectRetentionPolicies: true,
+			workload: utiltesting.MakeWorkload("wl", "ns").
+				Condition(metav1.Condition{
+					Type:   kueue.WorkloadFinished,
+					Status: metav1.ConditionTrue,
+					LastTransitionTime: metav1.Time{
+						Time: testStartTime,
+					},
+				}).
+				DeletionTimestamp(testStartTime).
+				Finalizers(kueue.ResourceInUseFinalizerName).
+				Obj(),
+			reconcilerOpts: []Option{
+				WithWorkloadRetention(
+					&workloadRetentionConfig{
+						afterFinished: ptr.To(util.LongTimeout),
+					},
+				),
+			},
+			wantWorkload: nil,
+			wantError:    nil,
+		},
+		"shouldn't try to delete the workload because the retention period hasn't elapsed yet, object retention configured": {
+			enableObjectRetentionPolicies: true,
+			workload: utiltesting.MakeWorkload("wl", "ns").
+				Condition(metav1.Condition{
+					Type:               kueue.WorkloadFinished,
+					Status:             metav1.ConditionTrue,
+					LastTransitionTime: metav1.NewTime(testStartTime.Add(-util.Timeout)),
+				}).
+				Obj(),
+			reconcilerOpts: []Option{
+				WithWorkloadRetention(
+					&workloadRetentionConfig{
+						afterFinished: ptr.To(util.LongTimeout),
+					},
+				),
+			},
+			wantResult: reconcile.Result{
+				RequeueAfter: util.LongTimeout - util.Timeout,
+			},
+			wantWorkload: utiltesting.MakeWorkload("wl", "ns").
+				Condition(metav1.Condition{
+					Type:               kueue.WorkloadFinished,
+					Status:             metav1.ConditionTrue,
+					LastTransitionTime: metav1.NewTime(testStartTime.Add(-util.Timeout)),
+				}).
+				Obj(),
+			wantError: nil,
+		},
+		"should delete the workload because the retention period has elapsed, object retention configured": {
+			enableObjectRetentionPolicies: true,
+			workload: utiltesting.MakeWorkload("wl", "ns").
+				Condition(metav1.Condition{
+					Type:               kueue.WorkloadFinished,
+					Status:             metav1.ConditionTrue,
+					LastTransitionTime: metav1.NewTime(testStartTime.Add(-2 * util.LongTimeout)),
+				}).
+				Obj(),
+			reconcilerOpts: []Option{
+				WithWorkloadRetention(
+					&workloadRetentionConfig{
+						afterFinished: ptr.To(util.LongTimeout),
+					},
+				),
+			},
+			wantWorkload: nil,
+			wantEvents: []utiltesting.EventRecord{
+				{
+					Key: types.NamespacedName{
+						Namespace: "ns",
+						Name:      "wl",
+					},
+					EventType: corev1.EventTypeNormal,
+					Reason:    "Deleted",
+					Message:   "Deleted finished workload due to elapsed retention",
+				},
+			},
+			wantError: nil,
+		},
 	}
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
+			features.SetFeatureGateDuringTest(t, features.ObjectRetentionPolicies, tc.enableObjectRetentionPolicies)
+
 			objs := []client.Object{tc.workload}
 			clientBuilder := utiltesting.NewClientBuilder().WithObjects(objs...).WithStatusSubresource(objs...).WithInterceptorFuncs(interceptor.Funcs{SubResourcePatch: utiltesting.TreatSSAAsStrategicMerge})
 			cl := clientBuilder.Build()
diff --git a/test/e2e/customconfigs/objectretentionpolicies_test.go b/test/e2e/customconfigs/objectretentionpolicies_test.go
index fecba1d83..def4fa560 100644
--- a/test/e2e/customconfigs/objectretentionpolicies_test.go
+++ b/test/e2e/customconfigs/objectretentionpolicies_test.go
@@ -48,6 +48,7 @@ var _ = ginkgo.Describe("ObjectRetentionPolicies with TinyTimeout", ginkgo.Order
 			cfg.FeatureGates = map[string]bool{string(features.ObjectRetentionPolicies): true}
 			cfg.ObjectRetentionPolicies = &configapi.ObjectRetentionPolicies{
 				Workloads: &configapi.WorkloadRetentionPolicy{
+					AfterFinished:           &metav1.Duration{Duration: util.TinyTimeout},
 					AfterDeactivatedByKueue: &metav1.Duration{Duration: util.TinyTimeout},
 				},
 			}
@@ -75,6 +76,44 @@ var _ = ginkgo.Describe("ObjectRetentionPolicies with TinyTimeout", ginkgo.Order
 		util.ExpectObjectToBeDeleted(ctx, k8sClient, rf, true)
 	})
 
+	ginkgo.When("workload has finished", func() {
+		ginkgo.It("should delete the Workload", func() {
+			job := testingjob.MakeJob("job", ns.Name).
+				Image(util.E2eTestAgnHostImage, util.BehaviorExitFast).
+				Queue(kueue.LocalQueueName(lq.Name)).
+				RequestAndLimit(corev1.ResourceCPU, "1").
+				Obj()
+			ginkgo.By("Creating a Job", func() {
+				util.MustCreate(ctx, k8sClient, job)
+			})
+
+			wlKey := types.NamespacedName{
+				Namespace: job.Namespace,
+				Name:      workloadjob.GetWorkloadNameForJob(job.Name, job.UID),
+			}
+			wl := &kueue.Workload{}
+
+			ginkgo.By("Waiting for the Workload to be created", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking that the Workload is deleted after it is finished", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, wlKey, wl)).To(testing.BeNotFoundError())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Checking that the Job is not deleted", func() {
+				createdJob := &batchv1.Job{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(job), createdJob)).To(gomega.Succeed())
+				}, util.ConsistentDuration, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
+
 	ginkgo.When("manually deactivating a Workload", func() {
 		ginkgo.It("shouldn't delete the Job or the Workload", func() {
 			job := testingjob.MakeJob("job", ns.Name).
diff --git a/test/integration/singlecluster/controller/core/clusterqueue_controller_test.go b/test/integration/singlecluster/controller/core/clusterqueue_controller_test.go
index c7aa1ad54..8556a3156 100644
--- a/test/integration/singlecluster/controller/core/clusterqueue_controller_test.go
+++ b/test/integration/singlecluster/controller/core/clusterqueue_controller_test.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/metrics"
@@ -941,7 +942,19 @@ var _ = ginkgo.Describe("ClusterQueue controller with queue visibility is enable
 		ginkgo.By("Enabling queue visibility feature", func() {
 			gomega.Expect(features.SetEnable(features.QueueVisibility, true)).To(gomega.Succeed())
 		})
-		fwk.StartManager(ctx, cfg, managerSetup)
+		fwk.StartManager(
+			ctx, cfg,
+			managerAndControllerSetup(
+				&config.Configuration{
+					QueueVisibility: &config.QueueVisibility{
+						UpdateIntervalSeconds: 2,
+						ClusterQueues: &config.ClusterQueueVisibility{
+							MaxCount: 3,
+						},
+					},
+				},
+			),
+		)
 	})
 
 	ginkgo.AfterAll(func() {
diff --git a/test/integration/singlecluster/controller/core/suite_test.go b/test/integration/singlecluster/controller/core/suite_test.go
index f223e10ee..599bcb806 100644
--- a/test/integration/singlecluster/controller/core/suite_test.go
+++ b/test/integration/singlecluster/controller/core/suite_test.go
@@ -64,26 +64,29 @@ var _ = ginkgo.AfterSuite(func() {
 })
 
 func managerSetup(ctx context.Context, mgr manager.Manager) {
-	err := indexer.Setup(ctx, mgr.GetFieldIndexer())
-	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+	managerAndControllerSetup(nil)(ctx, mgr)
+}
 
-	failedWebhook, err := webhooks.Setup(mgr)
-	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
+func managerAndControllerSetup(controllersCfg *config.Configuration) framework.ManagerSetup {
+	return func(ctx context.Context, mgr manager.Manager) {
+		err := indexer.Setup(ctx, mgr.GetFieldIndexer())
+		gomega.Expect(err).NotTo(gomega.HaveOccurred())
 
-	controllersCfg := &config.Configuration{}
-	mgr.GetScheme().Default(controllersCfg)
+		failedWebhook, err := webhooks.Setup(mgr)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "webhook", failedWebhook)
 
-	controllersCfg.Metrics.EnableClusterQueueResources = true
-	controllersCfg.QueueVisibility = &config.QueueVisibility{
-		UpdateIntervalSeconds: 2,
-		ClusterQueues: &config.ClusterQueueVisibility{
-			MaxCount: 3,
-		},
-	}
+		if controllersCfg == nil {
+			controllersCfg = &config.Configuration{}
+		}
+
+		mgr.GetScheme().Default(controllersCfg)
 
-	cCache := cache.New(mgr.GetClient())
-	queues := queue.NewManager(mgr.GetClient(), cCache)
+		controllersCfg.Metrics.EnableClusterQueueResources = true
 
-	failedCtrl, err := core.SetupControllers(mgr, queues, cCache, controllersCfg)
-	gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
+		cCache := cache.New(mgr.GetClient())
+		queues := queue.NewManager(mgr.GetClient(), cCache)
+
+		failedCtrl, err := core.SetupControllers(mgr, queues, cCache, controllersCfg)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred(), "controller", failedCtrl)
+	}
 }
diff --git a/test/integration/singlecluster/controller/core/workload_controller_test.go b/test/integration/singlecluster/controller/core/workload_controller_test.go
index 5821aa389..5c02a53df 100644
--- a/test/integration/singlecluster/controller/core/workload_controller_test.go
+++ b/test/integration/singlecluster/controller/core/workload_controller_test.go
@@ -29,8 +29,10 @@ import (
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	config "sigs.k8s.io/kueue/apis/config/v1beta1"
 	kueue "sigs.k8s.io/kueue/apis/kueue/v1beta1"
 	"sigs.k8s.io/kueue/pkg/constants"
+	"sigs.k8s.io/kueue/pkg/features"
 	"sigs.k8s.io/kueue/pkg/util/slices"
 	"sigs.k8s.io/kueue/pkg/util/testing"
 	"sigs.k8s.io/kueue/pkg/workload"
@@ -544,3 +546,187 @@ var _ = ginkgo.Describe("Workload controller", ginkgo.Ordered, ginkgo.ContinueOn
 		})
 	})
 })
+
+var _ = ginkgo.Describe("Workload controller with resource retention", ginkgo.Ordered, ginkgo.ContinueOnFailure, func() {
+	ginkgo.When("manager is setup with tiny retention period", func() {
+		var (
+			ns              *corev1.Namespace
+			createdWorkload kueue.Workload
+			localQueue      *kueue.LocalQueue
+			clusterQueue    *kueue.ClusterQueue
+			flavor          *kueue.ResourceFlavor
+		)
+
+		ginkgo.BeforeAll(func() {
+			fwk.StartManager(
+				ctx, cfg,
+				managerAndControllerSetup(
+					&config.Configuration{
+						ObjectRetentionPolicies: &config.ObjectRetentionPolicies{
+							Workloads: &config.WorkloadRetentionPolicy{
+								AfterFinished: &metav1.Duration{
+									Duration: util.TinyTimeout,
+								},
+							},
+						},
+					},
+				),
+			)
+
+			gomega.Expect(features.SetEnable(features.ObjectRetentionPolicies, true)).To(gomega.Succeed())
+		})
+
+		ginkgo.AfterAll(func() {
+			fwk.StopManager(ctx)
+		})
+
+		ginkgo.BeforeEach(func() {
+			ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "core-workload-")
+			flavor = testing.MakeResourceFlavor(flavorOnDemand).Obj()
+			gomega.Expect(k8sClient.Create(ctx, flavor)).Should(gomega.Succeed())
+			clusterQueue = testing.MakeClusterQueue("cq").
+				ResourceGroup(*testing.MakeFlavorQuotas(flavorOnDemand).
+					Resource(corev1.ResourceCPU, "1").Obj()).
+				Obj()
+			localQueue = testing.MakeLocalQueue("q", ns.Name).ClusterQueue("cq").Obj()
+			gomega.Expect(k8sClient.Create(ctx, localQueue)).To(gomega.Succeed())
+		})
+
+		ginkgo.AfterEach(func() {
+			gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, flavor, true)
+		})
+
+		ginkgo.It("should delete the workload after retention period elapses", func() {
+			var (
+				wl    *kueue.Workload
+				wlKey client.ObjectKey
+			)
+
+			ginkgo.By("creating a workload", func() {
+				wl = testing.MakeWorkload("wl-to-expire", ns.Name).Queue("q").Obj()
+				wlKey = client.ObjectKeyFromObject(wl)
+				gomega.Expect(k8sClient.Create(ctx, wl)).To(gomega.Succeed())
+			})
+
+			ginkgo.By("simulating workload admission", func() {
+				admission := testing.MakeAdmission("cq").Obj()
+				gomega.Expect(k8sClient.Get(ctx, wlKey, &createdWorkload)).To(gomega.Succeed())
+				gomega.Expect(util.SetQuotaReservation(ctx, k8sClient, &createdWorkload, admission)).Should(gomega.Succeed())
+				util.SyncAdmittedConditionForWorkloads(ctx, k8sClient, &createdWorkload)
+			})
+
+			ginkgo.By("marking workload as finished", func() {
+				gomega.Expect(k8sClient.Get(ctx, wlKey, &createdWorkload)).To(gomega.Succeed())
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdWorkload.Status.Conditions = append(createdWorkload.Status.Conditions, metav1.Condition{
+						Type:               kueue.WorkloadFinished,
+						Status:             metav1.ConditionTrue,
+						LastTransitionTime: metav1.Now(),
+						Reason:             "FinishedByTest",
+						Message:            "Finished for testing purposes",
+					})
+					g.Expect(k8sClient.Status().Update(ctx, &createdWorkload)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("workload should be deleted after the retention period", func() {
+				gomega.Eventually(func() error {
+					return k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), &createdWorkload)
+				}, util.Timeout, util.Interval).ShouldNot(gomega.Succeed())
+			})
+		})
+	})
+
+	ginkgo.When("manager is setup with long retention period", func() {
+		var (
+			ns              *corev1.Namespace
+			createdWorkload kueue.Workload
+			localQueue      *kueue.LocalQueue
+			clusterQueue    *kueue.ClusterQueue
+			flavor          *kueue.ResourceFlavor
+		)
+
+		ginkgo.BeforeAll(func() {
+			fwk.StartManager(
+				ctx, cfg,
+				managerAndControllerSetup(
+					&config.Configuration{
+						ObjectRetentionPolicies: &config.ObjectRetentionPolicies{
+							Workloads: &config.WorkloadRetentionPolicy{
+								AfterFinished: &metav1.Duration{
+									Duration: util.LongTimeout,
+								},
+							},
+						},
+					},
+				),
+			)
+
+			gomega.Expect(features.SetEnable(features.ObjectRetentionPolicies, true)).To(gomega.Succeed())
+		})
+
+		ginkgo.AfterAll(func() {
+			fwk.StopManager(ctx)
+		})
+
+		ginkgo.BeforeEach(func() {
+			ns = util.CreateNamespaceFromPrefixWithLog(ctx, k8sClient, "core-workload-")
+			flavor = testing.MakeResourceFlavor(flavorOnDemand).Obj()
+			gomega.Expect(k8sClient.Create(ctx, flavor)).Should(gomega.Succeed())
+			clusterQueue = testing.MakeClusterQueue("cq").
+				ResourceGroup(*testing.MakeFlavorQuotas(flavorOnDemand).
+					Resource(corev1.ResourceCPU, "1").Obj()).
+				Obj()
+			localQueue = testing.MakeLocalQueue("q", ns.Name).ClusterQueue("cq").Obj()
+			gomega.Expect(k8sClient.Create(ctx, localQueue)).To(gomega.Succeed())
+		})
+
+		ginkgo.AfterEach(func() {
+			gomega.Expect(util.DeleteNamespace(ctx, k8sClient, ns)).To(gomega.Succeed())
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, clusterQueue, true)
+			util.ExpectObjectToBeDeleted(ctx, k8sClient, flavor, true)
+		})
+
+		ginkgo.It("should not delete the workload before retention period elapses", func() {
+			var (
+				wl    *kueue.Workload
+				wlKey client.ObjectKey
+			)
+
+			ginkgo.By("creating a workload", func() {
+				wl = testing.MakeWorkload("wl-to-stay", ns.Name).Queue("q").Obj()
+				wlKey = client.ObjectKeyFromObject(wl)
+				gomega.Expect(k8sClient.Create(ctx, wl)).To(gomega.Succeed())
+			})
+
+			ginkgo.By("simulating workload admission", func() {
+				admission := testing.MakeAdmission("cq").Obj()
+				gomega.Expect(k8sClient.Get(ctx, wlKey, &createdWorkload)).To(gomega.Succeed())
+				gomega.Expect(util.SetQuotaReservation(ctx, k8sClient, &createdWorkload, admission)).Should(gomega.Succeed())
+				util.SyncAdmittedConditionForWorkloads(ctx, k8sClient, &createdWorkload)
+			})
+
+			ginkgo.By("marking workload as finished", func() {
+				gomega.Expect(k8sClient.Get(ctx, wlKey, &createdWorkload)).To(gomega.Succeed())
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdWorkload.Status.Conditions = append(createdWorkload.Status.Conditions, metav1.Condition{
+						Type:               kueue.WorkloadFinished,
+						Status:             metav1.ConditionTrue,
+						LastTransitionTime: metav1.Now(),
+						Reason:             "FinishedByTest",
+						Message:            "Finished for testing purposes",
+					})
+					g.Expect(k8sClient.Status().Update(ctx, &createdWorkload)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("workload should not be deleted before the retention period", func() {
+				gomega.Consistently(func() error {
+					return k8sClient.Get(ctx, client.ObjectKeyFromObject(wl), &createdWorkload)
+				}, util.ShortTimeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
+	})
+})