Merge pull request #66840 from janetkuo/job-ttl

api/api-rules/violation_exceptions.list

@@ -126,6 +126,7 @@ API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,K
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,KubeControllerManagerConfiguration,ResourceQuotaController
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,KubeControllerManagerConfiguration,SAController
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,KubeControllerManagerConfiguration,ServiceController
+API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,KubeControllerManagerConfiguration,TTLAfterFinishedController
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,NamespaceControllerConfiguration,NamespaceSyncPeriod
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,NamespaceControllerConfiguration,ConcurrentNamespaceSyncs
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,NodeIPAMControllerConfiguration,ServiceCIDR
@@ -156,6 +157,7 @@ API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,S
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,SAControllerConfiguration,ConcurrentSATokenSyncs
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,SAControllerConfiguration,RootCAFile
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,ServiceControllerConfiguration,ConcurrentServiceSyncs
+API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,TTLAfterFinishedControllerConfiguration,ConcurrentTTLSyncs
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,VolumeConfiguration,EnableHostPathProvisioning
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,VolumeConfiguration,EnableDynamicProvisioning
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,VolumeConfiguration,PersistentVolumeRecyclerConfiguration

api/openapi-spec/swagger.json

@@ -79306,6 +79306,11 @@
      "template": {
       "description": "Describes the pod that will be created when executing a job. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/",
       "$ref": "#/definitions/io.k8s.api.core.v1.PodTemplateSpec"
+     },
+     "ttlSecondsAfterFinished": {
+      "description": "ttlSecondsAfterFinished limits the lifetime of a Job that has finished execution (either Complete or Failed). If this field is set, ttlSecondsAfterFinished after the Job finishes, it is eligible to be automatically deleted. When the Job is being deleted, its lifecycle guarantees (e.g. finalizers) will be honored. If this field is unset, the Job won't be automatically deleted. If this field is set to zero, the Job becomes eligible to be deleted immediately after it finishes. This field is alpha-level and is only honored by servers that enable the TTLAfterFinished feature.",
+      "type": "integer",
+      "format": "int32"
      }
     }
    },

api/swagger-spec/batch_v1.json

@@ -1536,6 +1536,11 @@
      "template": {
       "$ref": "v1.PodTemplateSpec",
       "description": "Describes the pod that will be created when executing a job. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/"
+     },
+     "ttlSecondsAfterFinished": {
+      "type": "integer",
+      "format": "int32",
+      "description": "ttlSecondsAfterFinished limits the lifetime of a Job that has finished execution (either Complete or Failed). If this field is set, ttlSecondsAfterFinished after the Job finishes, it is eligible to be automatically deleted. When the Job is being deleted, its lifecycle guarantees (e.g. finalizers) will be honored. If this field is unset, the Job won't be automatically deleted. If this field is set to zero, the Job becomes eligible to be deleted immediately after it finishes. This field is alpha-level and is only honored by servers that enable the TTLAfterFinished feature."
      }
     }
    },

api/swagger-spec/batch_v1beta1.json

@@ -1591,6 +1591,11 @@
      "template": {
       "$ref": "v1.PodTemplateSpec",
       "description": "Describes the pod that will be created when executing a job. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/"
+     },
+     "ttlSecondsAfterFinished": {
+      "type": "integer",
+      "format": "int32",
+      "description": "ttlSecondsAfterFinished limits the lifetime of a Job that has finished execution (either Complete or Failed). If this field is set, ttlSecondsAfterFinished after the Job finishes, it is eligible to be automatically deleted. When the Job is being deleted, its lifecycle guarantees (e.g. finalizers) will be honored. If this field is unset, the Job won't be automatically deleted. If this field is set to zero, the Job becomes eligible to be deleted immediately after it finishes. This field is alpha-level and is only honored by servers that enable the TTLAfterFinished feature."
      }
     }
    },

api/swagger-spec/batch_v2alpha1.json

@@ -1591,6 +1591,11 @@
      "template": {
       "$ref": "v1.PodTemplateSpec",
       "description": "Describes the pod that will be created when executing a job. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/"
+     },
+     "ttlSecondsAfterFinished": {
+      "type": "integer",
+      "format": "int32",
+      "description": "ttlSecondsAfterFinished limits the lifetime of a Job that has finished execution (either Complete or Failed). If this field is set, ttlSecondsAfterFinished after the Job finishes, it is eligible to be automatically deleted. When the Job is being deleted, its lifecycle guarantees (e.g. finalizers) will be honored. If this field is unset, the Job won't be automatically deleted. If this field is set to zero, the Job becomes eligible to be deleted immediately after it finishes. This field is alpha-level and is only honored by servers that enable the TTLAfterFinished feature."
      }
     }
    },

cmd/kube-controller-manager/app/BUILD

@@ -68,6 +68,7 @@ go_library(
         "//pkg/controller/serviceaccount:go_default_library",
         "//pkg/controller/statefulset:go_default_library",
         "//pkg/controller/ttl:go_default_library",
+        "//pkg/controller/ttlafterfinished:go_default_library",
         "//pkg/controller/volume/attachdetach:go_default_library",
         "//pkg/controller/volume/expand:go_default_library",
         "//pkg/controller/volume/persistentvolume:go_default_library",

cmd/kube-controller-manager/app/controllermanager.go

@@ -378,6 +378,7 @@ func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc
 	controllers["clusterrole-aggregation"] = startClusterRoleAggregrationController
 	controllers["pvc-protection"] = startPVCProtectionController
 	controllers["pv-protection"] = startPVProtectionController
+	controllers["ttl-after-finished"] = startTTLAfterFinishedController
 
 	return controllers
 }

cmd/kube-controller-manager/app/core.go

@@ -52,6 +52,7 @@ import (
 	servicecontroller "k8s.io/kubernetes/pkg/controller/service"
 	serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
 	ttlcontroller "k8s.io/kubernetes/pkg/controller/ttl"
+	"k8s.io/kubernetes/pkg/controller/ttlafterfinished"
 	"k8s.io/kubernetes/pkg/controller/volume/attachdetach"
 	"k8s.io/kubernetes/pkg/controller/volume/expand"
 	persistentvolumecontroller "k8s.io/kubernetes/pkg/controller/volume/persistentvolume"
@@ -417,3 +418,14 @@ func startPVProtectionController(ctx ControllerContext) (http.Handler, bool, err
 	).Run(1, ctx.Stop)
 	return nil, true, nil
 }
+
+func startTTLAfterFinishedController(ctx ControllerContext) (http.Handler, bool, error) {
+	if !utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished) {
+		return nil, false, nil
+	}
+	go ttlafterfinished.New(
+		ctx.InformerFactory.Batch().V1().Jobs(),
+		ctx.ClientBuilder.ClientOrDie("ttl-after-finished-controller"),
+	).Run(int(ctx.ComponentConfig.TTLAfterFinishedController.ConcurrentTTLSyncs), ctx.Stop)
+	return nil, true, nil
+}

cmd/kube-controller-manager/app/options/BUILD

@@ -28,6 +28,7 @@ go_library(
         "replicationcontroller.go",
         "resourcequotacontroller.go",
         "serviceaccountcontroller.go",
+        "ttlafterfinishedcontroller.go",
     ],
     importpath = "k8s.io/kubernetes/cmd/kube-controller-manager/app/options",
     deps = [

cmd/kube-controller-manager/app/options/options.go

@@ -77,6 +77,7 @@ type KubeControllerManagerOptions struct {
 	ReplicationController            *ReplicationControllerOptions
 	ResourceQuotaController          *ResourceQuotaControllerOptions
 	SAController                     *SAControllerOptions
+	TTLAfterFinishedController       *TTLAfterFinishedControllerOptions
 
 	SecureServing *apiserveroptions.SecureServingOptionsWithLoopback
 	// TODO: remove insecure serving mode
@@ -172,6 +173,9 @@ func NewKubeControllerManagerOptions() (*KubeControllerManagerOptions, error) {
 		ServiceController: &cmoptions.ServiceControllerOptions{
 			ConcurrentServiceSyncs: componentConfig.ServiceController.ConcurrentServiceSyncs,
 		},
+		TTLAfterFinishedController: &TTLAfterFinishedControllerOptions{
+			ConcurrentTTLSyncs: componentConfig.TTLAfterFinishedController.ConcurrentTTLSyncs,
+		},
 		SecureServing: apiserveroptions.NewSecureServingOptions().WithLoopback(),
 		InsecureServing: (&apiserveroptions.DeprecatedInsecureServingOptions{
 			BindAddress: net.ParseIP(componentConfig.Generic.Address),
@@ -251,6 +255,7 @@ func (s *KubeControllerManagerOptions) Flags(allControllers []string, disabledBy
 	s.ReplicationController.AddFlags(fss.FlagSet("replicationcontroller"))
 	s.ResourceQuotaController.AddFlags(fss.FlagSet("resourcequota controller"))
 	s.SAController.AddFlags(fss.FlagSet("serviceaccount controller"))
+	s.TTLAfterFinishedController.AddFlags(fss.FlagSet("ttl-after-finished controller"))
 
 	fs := fss.FlagSet("misc")
 	fs.StringVar(&s.Master, "master", s.Master, "The address of the Kubernetes API server (overrides any value in kubeconfig).")
@@ -328,6 +333,9 @@ func (s *KubeControllerManagerOptions) ApplyTo(c *kubecontrollerconfig.Config) e
 	if err := s.ServiceController.ApplyTo(&c.ComponentConfig.ServiceController); err != nil {
 		return err
 	}
+	if err := s.TTLAfterFinishedController.ApplyTo(&c.ComponentConfig.TTLAfterFinishedController); err != nil {
+		return err
+	}
 	if err := s.InsecureServing.ApplyTo(&c.InsecureServing, &c.LoopbackClientConfig); err != nil {
 		return err
 	}
@@ -376,6 +384,7 @@ func (s *KubeControllerManagerOptions) Validate(allControllers []string, disable
 	errs = append(errs, s.ResourceQuotaController.Validate()...)
 	errs = append(errs, s.SAController.Validate()...)
 	errs = append(errs, s.ServiceController.Validate()...)
+	errs = append(errs, s.TTLAfterFinishedController.Validate()...)
 	errs = append(errs, s.SecureServing.Validate()...)
 	errs = append(errs, s.InsecureServing.Validate()...)
 	errs = append(errs, s.Authentication.Validate()...)

cmd/kube-controller-manager/app/options/options_test.go

@@ -116,6 +116,7 @@ func TestAddFlags(t *testing.T) {
 		"--cert-dir=/a/b/c",
 		"--bind-address=192.168.4.21",
 		"--secure-port=10001",
+		"--concurrent-ttl-after-finished-syncs=8",
 	}
 	fs.Parse(args)
 	// Sort GCIgnoredResources because it's built from a map, which means the
@@ -255,6 +256,9 @@ func TestAddFlags(t *testing.T) {
 		ServiceController: &cmoptions.ServiceControllerOptions{
 			ConcurrentServiceSyncs: 2,
 		},
+		TTLAfterFinishedController: &TTLAfterFinishedControllerOptions{
+			ConcurrentTTLSyncs: 8,
+		},
 		SecureServing: (&apiserveroptions.SecureServingOptions{
 			BindPort:    10001,
 			BindAddress: net.ParseIP("192.168.4.21"),

cmd/kube-controller-manager/app/options/ttlafterfinishedcontroller.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package options
+
+import (
+	"github.com/spf13/pflag"
+
+	kubectrlmgrconfig "k8s.io/kubernetes/pkg/controller/apis/config"
+)
+
+// TTLAfterFinishedControllerOptions holds the TTLAfterFinishedController options.
+type TTLAfterFinishedControllerOptions struct {
+	ConcurrentTTLSyncs int32
+}
+
+// AddFlags adds flags related to TTLAfterFinishedController for controller manager to the specified FlagSet.
+func (o *TTLAfterFinishedControllerOptions) AddFlags(fs *pflag.FlagSet) {
+	if o == nil {
+		return
+	}
+
+	fs.Int32Var(&o.ConcurrentTTLSyncs, "concurrent-ttl-after-finished-syncs", o.ConcurrentTTLSyncs, "The number of TTL-after-finished controller workers that are allowed to sync concurrently.")
+}
+
+// ApplyTo fills up TTLAfterFinishedController config with options.
+func (o *TTLAfterFinishedControllerOptions) ApplyTo(cfg *kubectrlmgrconfig.TTLAfterFinishedControllerConfiguration) error {
+	if o == nil {
+		return nil
+	}
+
+	cfg.ConcurrentTTLSyncs = o.ConcurrentTTLSyncs
+
+	return nil
+}
+
+// Validate checks validation of TTLAfterFinishedControllerOptions.
+func (o *TTLAfterFinishedControllerOptions) Validate() []error {
+	if o == nil {
+		return nil
+	}
+
+	errs := []error{}
+	return errs
+}

docs/api-reference/batch/v1/definitions.html

@@ -2505,6 +2505,13 @@ When an object is created, the system will populate this list with the current s
 <td class="tableblock halign-left valign-top"><p class="tableblock"><a href="#_v1_podtemplatespec">v1.PodTemplateSpec</a></p></td>
 <td class="tableblock halign-left valign-top"></td>
 </tr>
+<tr>
+<td class="tableblock halign-left valign-top"><p class="tableblock">ttlSecondsAfterFinished</p></td>
+<td class="tableblock halign-left valign-top"><p class="tableblock">ttlSecondsAfterFinished limits the lifetime of a Job that has finished execution (either Complete or Failed). If this field is set, ttlSecondsAfterFinished after the Job finishes, it is eligible to be automatically deleted. When the Job is being deleted, its lifecycle guarantees (e.g. finalizers) will be honored. If this field is unset, the Job won&#8217;t be automatically deleted. If this field is set to zero, the Job becomes eligible to be deleted immediately after it finishes. This field is alpha-level and is only honored by servers that enable the TTLAfterFinished feature.</p></td>
+<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
+<td class="tableblock halign-left valign-top"><p class="tableblock">integer (int32)</p></td>
+<td class="tableblock halign-left valign-top"></td>
+</tr>
 </tbody>
 </table>
 

docs/api-reference/batch/v1beta1/definitions.html

@@ -2539,6 +2539,13 @@ When an object is created, the system will populate this list with the current s
 <td class="tableblock halign-left valign-top"><p class="tableblock"><a href="#_v1_podtemplatespec">v1.PodTemplateSpec</a></p></td>
 <td class="tableblock halign-left valign-top"></td>
 </tr>
+<tr>
+<td class="tableblock halign-left valign-top"><p class="tableblock">ttlSecondsAfterFinished</p></td>
+<td class="tableblock halign-left valign-top"><p class="tableblock">ttlSecondsAfterFinished limits the lifetime of a Job that has finished execution (either Complete or Failed). If this field is set, ttlSecondsAfterFinished after the Job finishes, it is eligible to be automatically deleted. When the Job is being deleted, its lifecycle guarantees (e.g. finalizers) will be honored. If this field is unset, the Job won&#8217;t be automatically deleted. If this field is set to zero, the Job becomes eligible to be deleted immediately after it finishes. This field is alpha-level and is only honored by servers that enable the TTLAfterFinished feature.</p></td>
+<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
+<td class="tableblock halign-left valign-top"><p class="tableblock">integer (int32)</p></td>
+<td class="tableblock halign-left valign-top"></td>
+</tr>
 </tbody>
 </table>
 

docs/api-reference/batch/v2alpha1/definitions.html

@@ -2512,6 +2512,13 @@ When an object is created, the system will populate this list with the current s
 <td class="tableblock halign-left valign-top"><p class="tableblock"><a href="#_v1_podtemplatespec">v1.PodTemplateSpec</a></p></td>
 <td class="tableblock halign-left valign-top"></td>
 </tr>
+<tr>
+<td class="tableblock halign-left valign-top"><p class="tableblock">ttlSecondsAfterFinished</p></td>
+<td class="tableblock halign-left valign-top"><p class="tableblock">ttlSecondsAfterFinished limits the lifetime of a Job that has finished execution (either Complete or Failed). If this field is set, ttlSecondsAfterFinished after the Job finishes, it is eligible to be automatically deleted. When the Job is being deleted, its lifecycle guarantees (e.g. finalizers) will be honored. If this field is unset, the Job won&#8217;t be automatically deleted. If this field is set to zero, the Job becomes eligible to be deleted immediately after it finishes. This field is alpha-level and is only honored by servers that enable the TTLAfterFinished feature.</p></td>
+<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
+<td class="tableblock halign-left valign-top"><p class="tableblock">integer (int32)</p></td>
+<td class="tableblock halign-left valign-top"></td>
+</tr>
 </tbody>
 </table>
 

pkg/apis/batch/types.go

@@ -138,6 +138,18 @@ type JobSpec struct {
 
 	// Describes the pod that will be created when executing a job.
 	Template api.PodTemplateSpec
+
+	// ttlSecondsAfterFinished limits the lifetime of a Job that has finished
+	// execution (either Complete or Failed). If this field is set,
+	// ttlSecondsAfterFinished after the Job finishes, it is eligible to be
+	// automatically deleted. When the Job is being deleted, its lifecycle
+	// guarantees (e.g. finalizers) will be honored. If this field is unset,
+	// the Job won't be automatically deleted. If this field is set to zero,
+	// the Job becomes eligible to be deleted immediately after it finishes.
+	// This field is alpha-level and is only honored by servers that enable the
+	// TTLAfterFinished feature.
+	// +optional
+	TTLSecondsAfterFinished *int32
 }
 
 // JobStatus represents the current state of a Job.

pkg/apis/batch/v1/conversion.go

@@ -54,6 +54,7 @@ func Convert_batch_JobSpec_To_v1_JobSpec(in *batch.JobSpec, out *batchv1.JobSpec
 	out.Completions = in.Completions
 	out.ActiveDeadlineSeconds = in.ActiveDeadlineSeconds
 	out.BackoffLimit = in.BackoffLimit
+	out.TTLSecondsAfterFinished = in.TTLSecondsAfterFinished
 	out.Selector = in.Selector
 	if in.ManualSelector != nil {
 		out.ManualSelector = new(bool)
@@ -73,6 +74,7 @@ func Convert_v1_JobSpec_To_batch_JobSpec(in *batchv1.JobSpec, out *batch.JobSpec
 	out.Completions = in.Completions
 	out.ActiveDeadlineSeconds = in.ActiveDeadlineSeconds
 	out.BackoffLimit = in.BackoffLimit
+	out.TTLSecondsAfterFinished = in.TTLSecondsAfterFinished
 	out.Selector = in.Selector
 	if in.ManualSelector != nil {
 		out.ManualSelector = new(bool)

pkg/apis/batch/v1/zz_generated.conversion.go

@@ -217,6 +217,7 @@ func autoConvert_v1_JobSpec_To_batch_JobSpec(in *v1.JobSpec, out *batch.JobSpec,
 	if err := apiscorev1.Convert_v1_PodTemplateSpec_To_core_PodTemplateSpec(&in.Template, &out.Template, s); err != nil {
 		return err
 	}
+	out.TTLSecondsAfterFinished = (*int32)(unsafe.Pointer(in.TTLSecondsAfterFinished))
 	return nil
 }
 
@@ -230,6 +231,7 @@ func autoConvert_batch_JobSpec_To_v1_JobSpec(in *batch.JobSpec, out *v1.JobSpec,
 	if err := apiscorev1.Convert_core_PodTemplateSpec_To_v1_PodTemplateSpec(&in.Template, &out.Template, s); err != nil {
 		return err
 	}
+	out.TTLSecondsAfterFinished = (*int32)(unsafe.Pointer(in.TTLSecondsAfterFinished))
 	return nil
 }
 

pkg/apis/batch/validation/BUILD

@@ -14,11 +14,13 @@ go_library(
         "//pkg/apis/batch:go_default_library",
         "//pkg/apis/core:go_default_library",
         "//pkg/apis/core/validation:go_default_library",
+        "//pkg/features:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/validation:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/validation:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/validation/field:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/github.com/robfig/cron:go_default_library",
     ],
 )
@@ -30,8 +32,10 @@ go_test(
     deps = [
         "//pkg/apis/batch:go_default_library",
         "//pkg/apis/core:go_default_library",
+        "//pkg/features:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
     ],
 )
 

pkg/apis/batch/validation/validation.go

@@ -24,9 +24,11 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	apimachineryvalidation "k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/apis/batch"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	apivalidation "k8s.io/kubernetes/pkg/apis/core/validation"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 // TODO: generalize for other controller objects that will follow the same pattern, such as ReplicaSet and DaemonSet, and
@@ -117,6 +119,14 @@ func validateJobSpec(spec *batch.JobSpec, fldPath *field.Path) field.ErrorList {
 	if spec.BackoffLimit != nil {
 		allErrs = append(allErrs, apivalidation.ValidateNonnegativeField(int64(*spec.BackoffLimit), fldPath.Child("backoffLimit"))...)
 	}
+	if utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished) {
+		// normal validation for TTLSecondsAfterFinished
+		if spec.TTLSecondsAfterFinished != nil {
+			allErrs = append(allErrs, apivalidation.ValidateNonnegativeField(int64(*spec.TTLSecondsAfterFinished), fldPath.Child("ttlSecondsAfterFinished"))...)
+		}
+	} else if spec.TTLSecondsAfterFinished != nil {
+		allErrs = append(allErrs, field.Forbidden(fldPath.Child("ttlSecondsAfterFinished"), "disabled by feature-gate"))
+	}
 
 	allErrs = append(allErrs, apivalidation.ValidatePodTemplateSpec(&spec.Template, fldPath.Child("template"))...)
 	if spec.Template.Spec.RestartPolicy != api.RestartPolicyOnFailure &&

pkg/apis/batch/validation/validation_test.go

@@ -17,13 +17,16 @@ limitations under the License.
 package validation
 
 import (
+	"fmt"
 	"strings"
 	"testing"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/apis/batch"
 	api "k8s.io/kubernetes/pkg/apis/core"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 func getValidManualSelector() *metav1.LabelSelector {
@@ -64,7 +67,21 @@ func getValidPodTemplateSpecForGenerated(selector *metav1.LabelSelector) api.Pod
 	}
 }
 
+func featureToggle(feature utilfeature.Feature) []string {
+	enabled := fmt.Sprintf("%s=%t", feature, true)
+	disabled := fmt.Sprintf("%s=%t", feature, false)
+	return []string{enabled, disabled}
+}
+
 func TestValidateJob(t *testing.T) {
+	ttlEnabled := utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished)
+	defer func() {
+		err := utilfeature.DefaultFeatureGate.Set(fmt.Sprintf("%s=%t", features.TTLAfterFinished, ttlEnabled))
+		if err != nil {
+			t.Fatalf("Failed to set feature gate for %s: %v", features.TTLAfterFinished, err)
+		}
+	}()
+
 	validManualSelector := getValidManualSelector()
 	validPodTemplateSpecForManual := getValidPodTemplateSpecForManual(validManualSelector)
 	validGeneratedSelector := getValidGeneratedSelector()
@@ -214,15 +231,39 @@ func TestValidateJob(t *testing.T) {
 		},
 	}
 
-	for k, v := range errorCases {
-		errs := ValidateJob(&v)
-		if len(errs) == 0 {
-			t.Errorf("expected failure for %s", k)
+	for _, setFeature := range featureToggle(features.TTLAfterFinished) {
+		// Set error cases based on if TTLAfterFinished feature is enabled or not
+		if err := utilfeature.DefaultFeatureGate.Set(setFeature); err != nil {
+			t.Fatalf("Failed to set feature gate for %s: %v", features.TTLAfterFinished, err)
+		}
+		ttlCase := "spec.ttlSecondsAfterFinished:must be greater than or equal to 0"
+		if utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished) {
+			errorCases[ttlCase] = batch.Job{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "myjob",
+					Namespace: metav1.NamespaceDefault,
+					UID:       types.UID("1a2b3c"),
+				},
+				Spec: batch.JobSpec{
+					TTLSecondsAfterFinished: &negative,
+					Selector:                validGeneratedSelector,
+					Template:                validPodTemplateSpecForGenerated,
+				},
+			}
 		} else {
-			s := strings.Split(k, ":")
-			err := errs[0]
-			if err.Field != s[0] || !strings.Contains(err.Error(), s[1]) {
-				t.Errorf("unexpected error: %v, expected: %s", err, k)
+			delete(errorCases, ttlCase)
+		}
+
+		for k, v := range errorCases {
+			errs := ValidateJob(&v)
+			if len(errs) == 0 {
+				t.Errorf("expected failure for %s", k)
+			} else {
+				s := strings.Split(k, ":")
+				err := errs[0]
+				if err.Field != s[0] || !strings.Contains(err.Error(), s[1]) {
+					t.Errorf("unexpected error: %v, expected: %s", err, k)
+				}
 			}
 		}
 	}
@@ -584,6 +625,25 @@ func TestValidateCronJob(t *testing.T) {
 			},
 		},
 	}
+	if utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished) {
+		errorCases["spec.jobTemplate.spec.ttlSecondsAfterFinished:must be greater than or equal to 0"] = batch.CronJob{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "mycronjob",
+				Namespace: metav1.NamespaceDefault,
+				UID:       types.UID("1a2b3c"),
+			},
+			Spec: batch.CronJobSpec{
+				Schedule:          "* * * * ?",
+				ConcurrencyPolicy: batch.AllowConcurrent,
+				JobTemplate: batch.JobTemplateSpec{
+					Spec: batch.JobSpec{
+						TTLSecondsAfterFinished: &negative,
+						Template:                validPodTemplateSpec,
+					},
+				},
+			},
+		}
+	}
 
 	for k, v := range errorCases {
 		errs := ValidateCronJob(&v)

pkg/apis/batch/zz_generated.deepcopy.go

@@ -262,6 +262,11 @@ func (in *JobSpec) DeepCopyInto(out *JobSpec) {
 		**out = **in
 	}
 	in.Template.DeepCopyInto(&out.Template)
+	if in.TTLSecondsAfterFinished != nil {
+		in, out := &in.TTLSecondsAfterFinished, &out.TTLSecondsAfterFinished
+		*out = new(int32)
+		**out = **in
+	}
 	return
 }
 

pkg/controller/BUILD

@@ -131,6 +131,7 @@ filegroup(
         "//pkg/controller/statefulset:all-srcs",
         "//pkg/controller/testutil:all-srcs",
         "//pkg/controller/ttl:all-srcs",
+        "//pkg/controller/ttlafterfinished:all-srcs",
         "//pkg/controller/util/node:all-srcs",
         "//pkg/controller/volume/attachdetach:all-srcs",
         "//pkg/controller/volume/events:all-srcs",

pkg/controller/apis/config/types.go

@@ -96,6 +96,9 @@ type KubeControllerManagerConfiguration struct {
 	// ServiceControllerConfiguration holds configuration for ServiceController
 	// related features.
 	ServiceController ServiceControllerConfiguration
+	// TTLAfterFinishedControllerConfiguration holds configuration for
+	// TTLAfterFinishedController related features.
+	TTLAfterFinishedController TTLAfterFinishedControllerConfiguration
 }
 
 // GenericControllerManagerConfiguration holds configuration for a generic controller-manager
@@ -438,3 +441,10 @@ type PersistentVolumeRecyclerConfiguration struct {
 	// in a multi-node cluster.
 	IncrementTimeoutHostPath int32
 }
+
+// TTLAfterFinishedControllerConfiguration contains elements describing TTLAfterFinishedController.
+type TTLAfterFinishedControllerConfiguration struct {
+	// concurrentTTLSyncs is the number of TTL-after-finished collector workers that are
+	// allowed to sync concurrently.
+	ConcurrentTTLSyncs int32
+}

pkg/controller/apis/config/v1alpha1/defaults.go

@@ -48,6 +48,9 @@ func SetDefaults_KubeControllerManagerConfiguration(obj *kubectrlmgrconfigv1alph
 	if obj.SAController.ConcurrentSATokenSyncs == 0 {
 		obj.SAController.ConcurrentSATokenSyncs = 5
 	}
+	if obj.TTLAfterFinishedController.ConcurrentTTLSyncs <= 0 {
+		obj.TTLAfterFinishedController.ConcurrentTTLSyncs = 5
+	}
 
 	// These defaults override the recommended defaults from the apimachineryconfigv1alpha1 package that are applied automatically
 	// These client-connection defaults are specific to the kube-controller-manager

pkg/controller/apis/config/v1alpha1/zz_generated.conversion.go

@@ -289,6 +289,16 @@ func RegisterConversions(s *runtime.Scheme) error {
 	}); err != nil {
 		return err
 	}
+	if err := s.AddGeneratedConversionFunc((*v1alpha1.TTLAfterFinishedControllerConfiguration)(nil), (*config.TTLAfterFinishedControllerConfiguration)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_v1alpha1_TTLAfterFinishedControllerConfiguration_To_config_TTLAfterFinishedControllerConfiguration(a.(*v1alpha1.TTLAfterFinishedControllerConfiguration), b.(*config.TTLAfterFinishedControllerConfiguration), scope)
+	}); err != nil {
+		return err
+	}
+	if err := s.AddGeneratedConversionFunc((*config.TTLAfterFinishedControllerConfiguration)(nil), (*v1alpha1.TTLAfterFinishedControllerConfiguration)(nil), func(a, b interface{}, scope conversion.Scope) error {
+		return Convert_config_TTLAfterFinishedControllerConfiguration_To_v1alpha1_TTLAfterFinishedControllerConfiguration(a.(*config.TTLAfterFinishedControllerConfiguration), b.(*v1alpha1.TTLAfterFinishedControllerConfiguration), scope)
+	}); err != nil {
+		return err
+	}
 	if err := s.AddGeneratedConversionFunc((*v1alpha1.VolumeConfiguration)(nil), (*config.VolumeConfiguration)(nil), func(a, b interface{}, scope conversion.Scope) error {
 		return Convert_v1alpha1_VolumeConfiguration_To_config_VolumeConfiguration(a.(*v1alpha1.VolumeConfiguration), b.(*config.VolumeConfiguration), scope)
 	}); err != nil {
@@ -734,6 +744,9 @@ func autoConvert_v1alpha1_KubeControllerManagerConfiguration_To_config_KubeContr
 	if err := Convert_v1alpha1_ServiceControllerConfiguration_To_config_ServiceControllerConfiguration(&in.ServiceController, &out.ServiceController, s); err != nil {
 		return err
 	}
+	if err := Convert_v1alpha1_TTLAfterFinishedControllerConfiguration_To_config_TTLAfterFinishedControllerConfiguration(&in.TTLAfterFinishedController, &out.TTLAfterFinishedController, s); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -806,6 +819,9 @@ func autoConvert_config_KubeControllerManagerConfiguration_To_v1alpha1_KubeContr
 	if err := Convert_config_ServiceControllerConfiguration_To_v1alpha1_ServiceControllerConfiguration(&in.ServiceController, &out.ServiceController, s); err != nil {
 		return err
 	}
+	if err := Convert_config_TTLAfterFinishedControllerConfiguration_To_v1alpha1_TTLAfterFinishedControllerConfiguration(&in.TTLAfterFinishedController, &out.TTLAfterFinishedController, s); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -1070,6 +1086,26 @@ func autoConvert_config_ServiceControllerConfiguration_To_v1alpha1_ServiceContro
 	return nil
 }
 
+func autoConvert_v1alpha1_TTLAfterFinishedControllerConfiguration_To_config_TTLAfterFinishedControllerConfiguration(in *v1alpha1.TTLAfterFinishedControllerConfiguration, out *config.TTLAfterFinishedControllerConfiguration, s conversion.Scope) error {
+	out.ConcurrentTTLSyncs = in.ConcurrentTTLSyncs
+	return nil
+}
+
+// Convert_v1alpha1_TTLAfterFinishedControllerConfiguration_To_config_TTLAfterFinishedControllerConfiguration is an autogenerated conversion function.
+func Convert_v1alpha1_TTLAfterFinishedControllerConfiguration_To_config_TTLAfterFinishedControllerConfiguration(in *v1alpha1.TTLAfterFinishedControllerConfiguration, out *config.TTLAfterFinishedControllerConfiguration, s conversion.Scope) error {
+	return autoConvert_v1alpha1_TTLAfterFinishedControllerConfiguration_To_config_TTLAfterFinishedControllerConfiguration(in, out, s)
+}
+
+func autoConvert_config_TTLAfterFinishedControllerConfiguration_To_v1alpha1_TTLAfterFinishedControllerConfiguration(in *config.TTLAfterFinishedControllerConfiguration, out *v1alpha1.TTLAfterFinishedControllerConfiguration, s conversion.Scope) error {
+	out.ConcurrentTTLSyncs = in.ConcurrentTTLSyncs
+	return nil
+}
+
+// Convert_config_TTLAfterFinishedControllerConfiguration_To_v1alpha1_TTLAfterFinishedControllerConfiguration is an autogenerated conversion function.
+func Convert_config_TTLAfterFinishedControllerConfiguration_To_v1alpha1_TTLAfterFinishedControllerConfiguration(in *config.TTLAfterFinishedControllerConfiguration, out *v1alpha1.TTLAfterFinishedControllerConfiguration, s conversion.Scope) error {
+	return autoConvert_config_TTLAfterFinishedControllerConfiguration_To_v1alpha1_TTLAfterFinishedControllerConfiguration(in, out, s)
+}
+
 func autoConvert_v1alpha1_VolumeConfiguration_To_config_VolumeConfiguration(in *v1alpha1.VolumeConfiguration, out *config.VolumeConfiguration, s conversion.Scope) error {
 	if err := v1.Convert_Pointer_bool_To_bool(&in.EnableHostPathProvisioning, &out.EnableHostPathProvisioning, s); err != nil {
 		return err

pkg/controller/apis/config/zz_generated.deepcopy.go

@@ -285,6 +285,7 @@ func (in *KubeControllerManagerConfiguration) DeepCopyInto(out *KubeControllerMa
 	out.ResourceQuotaController = in.ResourceQuotaController
 	out.SAController = in.SAController
 	out.ServiceController = in.ServiceController
+	out.TTLAfterFinishedController = in.TTLAfterFinishedController
 	return
 }
 
@@ -490,6 +491,22 @@ func (in *ServiceControllerConfiguration) DeepCopy() *ServiceControllerConfigura
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TTLAfterFinishedControllerConfiguration) DeepCopyInto(out *TTLAfterFinishedControllerConfiguration) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TTLAfterFinishedControllerConfiguration.
+func (in *TTLAfterFinishedControllerConfiguration) DeepCopy() *TTLAfterFinishedControllerConfiguration {
+	if in == nil {
+		return nil
+	}
+	out := new(TTLAfterFinishedControllerConfiguration)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *VolumeConfiguration) DeepCopyInto(out *VolumeConfiguration) {
 	*out = *in
 	out.PersistentVolumeRecyclerConfiguration = in.PersistentVolumeRecyclerConfiguration

pkg/controller/ttlafterfinished/BUILD

+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["ttlafterfinished_controller.go"],
+    importpath = "k8s.io/kubernetes/pkg/controller/ttlafterfinished",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/controller:go_default_library",
+        "//pkg/controller/job:go_default_library",
+        "//pkg/kubectl/scheme:go_default_library",
+        "//pkg/util/metrics:go_default_library",
+        "//staging/src/k8s.io/api/batch/v1:go_default_library",
+        "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//staging/src/k8s.io/client-go/informers/batch/v1:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
+        "//staging/src/k8s.io/client-go/listers/batch/v1:go_default_library",
+        "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
+        "//staging/src/k8s.io/client-go/tools/record:go_default_library",
+        "//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["ttlafterfinished_controller_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//staging/src/k8s.io/api/batch/v1:go_default_library",
+        "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

pkg/controller/ttlafterfinished/ttlafterfinished_controller_test.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package ttlafterfinished
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	batch "k8s.io/api/batch/v1"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func newJob(completionTime, failedTime metav1.Time, ttl *int32) *batch.Job {
+	j := &batch.Job{
+		TypeMeta: metav1.TypeMeta{Kind: "Job"},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "foobar",
+			Namespace: metav1.NamespaceDefault,
+		},
+		Spec: batch.JobSpec{
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{"foo": "bar"},
+			},
+			Template: v1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						"foo": "bar",
+					},
+				},
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{
+						{Image: "foo/bar"},
+					},
+				},
+			},
+		},
+	}
+
+	if !completionTime.IsZero() {
+		c := batch.JobCondition{Type: batch.JobComplete, Status: v1.ConditionTrue, LastTransitionTime: completionTime}
+		j.Status.Conditions = append(j.Status.Conditions, c)
+	}
+
+	if !failedTime.IsZero() {
+		c := batch.JobCondition{Type: batch.JobFailed, Status: v1.ConditionTrue, LastTransitionTime: failedTime}
+		j.Status.Conditions = append(j.Status.Conditions, c)
+	}
+
+	if ttl != nil {
+		j.Spec.TTLSecondsAfterFinished = ttl
+	}
+
+	return j
+}
+
+func durationPointer(n int) *time.Duration {
+	s := time.Duration(n) * time.Second
+	return &s
+}
+
+func int32Ptr(n int32) *int32 {
+	return &n
+}
+
+func TestTimeLeft(t *testing.T) {
+	now := metav1.Now()
+
+	testCases := []struct {
+		name             string
+		completionTime   metav1.Time
+		failedTime       metav1.Time
+		ttl              *int32
+		since            *time.Time
+		expectErr        bool
+		expectErrStr     string
+		expectedTimeLeft *time.Duration
+	}{
+		{
+			name:         "Error case: Job unfinished",
+			ttl:          int32Ptr(100),
+			since:        &now.Time,
+			expectErr:    true,
+			expectErrStr: "should not be cleaned up",
+		},
+		{
+			name:           "Error case: Job completed now, no TTL",
+			completionTime: now,
+			since:          &now.Time,
+			expectErr:      true,
+			expectErrStr:   "should not be cleaned up",
+		},
+		{
+			name:             "Job completed now, 0s TTL",
+			completionTime:   now,
+			ttl:              int32Ptr(0),
+			since:            &now.Time,
+			expectedTimeLeft: durationPointer(0),
+		},
+		{
+			name:             "Job completed now, 10s TTL",
+			completionTime:   now,
+			ttl:              int32Ptr(10),
+			since:            &now.Time,
+			expectedTimeLeft: durationPointer(10),
+		},
+		{
+			name:             "Job completed 10s ago, 15s TTL",
+			completionTime:   metav1.NewTime(now.Add(-10 * time.Second)),
+			ttl:              int32Ptr(15),
+			since:            &now.Time,
+			expectedTimeLeft: durationPointer(5),
+		},
+		{
+			name:         "Error case: Job failed now, no TTL",
+			failedTime:   now,
+			since:        &now.Time,
+			expectErr:    true,
+			expectErrStr: "should not be cleaned up",
+		},
+		{
+			name:             "Job failed now, 0s TTL",
+			failedTime:       now,
+			ttl:              int32Ptr(0),
+			since:            &now.Time,
+			expectedTimeLeft: durationPointer(0),
+		},
+		{
+			name:             "Job failed now, 10s TTL",
+			failedTime:       now,
+			ttl:              int32Ptr(10),
+			since:            &now.Time,
+			expectedTimeLeft: durationPointer(10),
+		},
+		{
+			name:             "Job failed 10s ago, 15s TTL",
+			failedTime:       metav1.NewTime(now.Add(-10 * time.Second)),
+			ttl:              int32Ptr(15),
+			since:            &now.Time,
+			expectedTimeLeft: durationPointer(5),
+		},
+	}
+
+	for _, tc := range testCases {
+		job := newJob(tc.completionTime, tc.failedTime, tc.ttl)
+		gotTimeLeft, gotErr := timeLeft(job, tc.since)
+		if tc.expectErr != (gotErr != nil) {
+			t.Errorf("%s: expected error is %t, got %t, error: %v", tc.name, tc.expectErr, gotErr != nil, gotErr)
+		}
+		if tc.expectErr && len(tc.expectErrStr) == 0 {
+			t.Errorf("%s: invalid test setup; error message must not be empty for error cases", tc.name)
+		}
+		if tc.expectErr && !strings.Contains(gotErr.Error(), tc.expectErrStr) {
+			t.Errorf("%s: expected error message contains %q, got %v", tc.name, tc.expectErrStr, gotErr)
+		}
+		if !tc.expectErr {
+			if *gotTimeLeft != *tc.expectedTimeLeft {
+				t.Errorf("%s: expected time left %v, got %v", tc.name, tc.expectedTimeLeft, gotTimeLeft)
+			}
+		}
+	}
+}

pkg/features/kube_features.go

@@ -381,6 +381,12 @@ const (
 	//
 	// Enables control over ProcMountType for containers.
 	ProcMountType utilfeature.Feature = "ProcMountType"
+
+	// owner: @janetkuo
+	// alpha: v1.12
+	//
+	// Allow TTL controller to clean up Pods and Jobs after they finish.
+	TTLAfterFinished utilfeature.Feature = "TTLAfterFinished"
 )
 
 func init() {
@@ -445,6 +451,7 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
 	SCTPSupport:                                 {Default: false, PreRelease: utilfeature.Alpha},
 	VolumeSnapshotDataSource:                    {Default: false, PreRelease: utilfeature.Alpha},
 	ProcMountType:                               {Default: false, PreRelease: utilfeature.Alpha},
+	TTLAfterFinished:                            {Default: false, PreRelease: utilfeature.Alpha},
 
 	// inherited features from generic apiserver, relisted here to get a conflict if it is changed
 	// unintentionally on either side:

pkg/registry/batch/job/BUILD

@@ -18,6 +18,7 @@ go_library(
         "//pkg/api/pod:go_default_library",
         "//pkg/apis/batch:go_default_library",
         "//pkg/apis/batch/validation:go_default_library",
+        "//pkg/features:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/fields:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
@@ -27,6 +28,7 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/registry/rest:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/names:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
     ],
 )
 
@@ -39,10 +41,12 @@ go_test(
         "//pkg/api/testing:go_default_library",
         "//pkg/apis/batch:go_default_library",
         "//pkg/apis/core:go_default_library",
+        "//pkg/features:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/request:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/registry/rest:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
     ],
 )
 

pkg/registry/batch/job/strategy.go

@@ -30,10 +30,12 @@ import (
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/apiserver/pkg/storage"
 	"k8s.io/apiserver/pkg/storage/names"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	"k8s.io/kubernetes/pkg/api/pod"
 	"k8s.io/kubernetes/pkg/apis/batch"
 	"k8s.io/kubernetes/pkg/apis/batch/validation"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 // jobStrategy implements verification logic for Replication Controllers.
@@ -61,6 +63,10 @@ func (jobStrategy) PrepareForCreate(ctx context.Context, obj runtime.Object) {
 	job := obj.(*batch.Job)
 	job.Status = batch.JobStatus{}
 
+	if !utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished) {
+		job.Spec.TTLSecondsAfterFinished = nil
+	}
+
 	pod.DropDisabledAlphaFields(&job.Spec.Template.Spec)
 }
 
@@ -70,6 +76,11 @@ func (jobStrategy) PrepareForUpdate(ctx context.Context, obj, old runtime.Object
 	oldJob := old.(*batch.Job)
 	newJob.Status = oldJob.Status
 
+	if !utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished) {
+		newJob.Spec.TTLSecondsAfterFinished = nil
+		oldJob.Spec.TTLSecondsAfterFinished = nil
+	}
+
 	pod.DropDisabledAlphaFields(&newJob.Spec.Template.Spec)
 	pod.DropDisabledAlphaFields(&oldJob.Spec.Template.Spec)
 }

pkg/registry/batch/job/strategy_test.go

@@ -24,19 +24,24 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/api/testapi"
 	apitesting "k8s.io/kubernetes/pkg/api/testing"
 	"k8s.io/kubernetes/pkg/apis/batch"
 	api "k8s.io/kubernetes/pkg/apis/core"
+	"k8s.io/kubernetes/pkg/features"
 )
 
 func newBool(a bool) *bool {
-	r := new(bool)
-	*r = a
-	return r
+	return &a
+}
+
+func newInt32(i int32) *int32 {
+	return &i
 }
 
 func TestJobStrategy(t *testing.T) {
+	ttlEnabled := utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished)
 	ctx := genericapirequest.NewDefaultContext()
 	if !Strategy.NamespaceScoped() {
 		t.Errorf("Job must be namespace scoped")
@@ -64,9 +69,10 @@ func TestJobStrategy(t *testing.T) {
 			Namespace: metav1.NamespaceDefault,
 		},
 		Spec: batch.JobSpec{
-			Selector:       validSelector,
-			Template:       validPodTemplateSpec,
-			ManualSelector: newBool(true),
+			Selector:                validSelector,
+			Template:                validPodTemplateSpec,
+			TTLSecondsAfterFinished: newInt32(0), // Set TTL
+			ManualSelector:          newBool(true),
 		},
 		Status: batch.JobStatus{
 			Active: 11,
@@ -81,11 +87,21 @@ func TestJobStrategy(t *testing.T) {
 	if len(errs) != 0 {
 		t.Errorf("Unexpected error validating %v", errs)
 	}
+	if ttlEnabled && job.Spec.TTLSecondsAfterFinished == nil {
+		// When the TTL feature is enabled, the TTL field can be set
+		t.Errorf("Job should allow setting .spec.ttlSecondsAfterFinished when %v feature is enabled", features.TTLAfterFinished)
+	}
+	if !ttlEnabled && job.Spec.TTLSecondsAfterFinished != nil {
+		// When the TTL feature is disabled, the TTL field cannot be set
+		t.Errorf("Job should not allow setting .spec.ttlSecondsAfterFinished when %v feature is disabled", features.TTLAfterFinished)
+	}
+
 	parallelism := int32(10)
 	updatedJob := &batch.Job{
 		ObjectMeta: metav1.ObjectMeta{Name: "bar", ResourceVersion: "4"},
 		Spec: batch.JobSpec{
-			Parallelism: &parallelism,
+			Parallelism:             &parallelism,
+			TTLSecondsAfterFinished: newInt32(1), // Update TTL
 		},
 		Status: batch.JobStatus{
 			Active: 11,
@@ -101,6 +117,9 @@ func TestJobStrategy(t *testing.T) {
 	if len(errs) == 0 {
 		t.Errorf("Expected a validation error")
 	}
+	if ttlEnabled != (job.Spec.TTLSecondsAfterFinished != nil || updatedJob.Spec.TTLSecondsAfterFinished != nil) {
+		t.Errorf("Job should only allow updating .spec.ttlSecondsAfterFinished when %v feature is enabled", features.TTLAfterFinished)
+	}
 
 	// Make sure we correctly implement the interface.
 	// Otherwise a typo could silently change the default.

plugin/pkg/auth/authorizer/rbac/bootstrappolicy/controller_policy.go

@@ -340,6 +340,15 @@ func buildControllerRoles() ([]rbacv1.ClusterRole, []rbacv1.ClusterRoleBinding)
 			eventsRule(),
 		},
 	})
+	if utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished) {
+		addControllerRole(&controllerRoles, &controllerRoleBindings, rbacv1.ClusterRole{
+			ObjectMeta: metav1.ObjectMeta{Name: saRolePrefix + "ttl-after-finished-controller"},
+			Rules: []rbacv1.PolicyRule{
+				rbacv1helpers.NewRule("get", "list", "watch", "delete").Groups(batchGroup).Resources("jobs").RuleOrDie(),
+				eventsRule(),
+			},
+		})
+	}
 
 	return controllerRoles, controllerRoleBindings
 }

staging/src/k8s.io/api/batch/v1/generated.proto

@@ -134,6 +134,18 @@ message JobSpec {
   // Describes the pod that will be created when executing a job.
   // More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/
   optional k8s.io.api.core.v1.PodTemplateSpec template = 6;
+
+  // ttlSecondsAfterFinished limits the lifetime of a Job that has finished
+  // execution (either Complete or Failed). If this field is set,
+  // ttlSecondsAfterFinished after the Job finishes, it is eligible to be
+  // automatically deleted. When the Job is being deleted, its lifecycle
+  // guarantees (e.g. finalizers) will be honored. If this field is unset,
+  // the Job won't be automatically deleted. If this field is set to zero,
+  // the Job becomes eligible to be deleted immediately after it finishes.
+  // This field is alpha-level and is only honored by servers that enable the
+  // TTLAfterFinished feature.
+  // +optional
+  optional int32 ttlSecondsAfterFinished = 8;
 }
 
 // JobStatus represents the current state of a Job.

staging/src/k8s.io/api/batch/v1/types.go

@@ -114,6 +114,18 @@ type JobSpec struct {
 	// Describes the pod that will be created when executing a job.
 	// More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/
 	Template v1.PodTemplateSpec `json:"template" protobuf:"bytes,6,opt,name=template"`
+
+	// ttlSecondsAfterFinished limits the lifetime of a Job that has finished
+	// execution (either Complete or Failed). If this field is set,
+	// ttlSecondsAfterFinished after the Job finishes, it is eligible to be
+	// automatically deleted. When the Job is being deleted, its lifecycle
+	// guarantees (e.g. finalizers) will be honored. If this field is unset,
+	// the Job won't be automatically deleted. If this field is set to zero,
+	// the Job becomes eligible to be deleted immediately after it finishes.
+	// This field is alpha-level and is only honored by servers that enable the
+	// TTLAfterFinished feature.
+	// +optional
+	TTLSecondsAfterFinished *int32 `json:"ttlSecondsAfterFinished,omitempty" protobuf:"varint,8,opt,name=ttlSecondsAfterFinished"`
 }
 
 // JobStatus represents the current state of a Job.

staging/src/k8s.io/api/batch/v1/types_swagger_doc_generated.go

@@ -63,14 +63,15 @@ func (JobList) SwaggerDoc() map[string]string {
 }
 
 var map_JobSpec = map[string]string{
-	"":                      "JobSpec describes how the job execution will look like.",
-	"parallelism":           "Specifies the maximum desired number of pods the job should run at any given time. The actual number of pods running in steady state will be less than this number when ((.spec.completions - .status.successful) < .spec.parallelism), i.e. when the work left to do is less than max parallelism. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/",
-	"completions":           "Specifies the desired number of successfully finished pods the job should be run with.  Setting to nil means that the success of any pod signals the success of all pods, and allows parallelism to have any positive value.  Setting to 1 means that parallelism is limited to 1 and the success of that pod signals the success of the job. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/",
-	"activeDeadlineSeconds": "Specifies the duration in seconds relative to the startTime that the job may be active before the system tries to terminate it; value must be positive integer",
-	"backoffLimit":          "Specifies the number of retries before marking this job failed. Defaults to 6",
-	"selector":              "A label query over pods that should match the pod count. Normally, the system sets this field for you. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#label-selectors",
-	"manualSelector":        "manualSelector controls generation of pod labels and pod selectors. Leave `manualSelector` unset unless you are certain what you are doing. When false or unset, the system pick labels unique to this job and appends those labels to the pod template.  When true, the user is responsible for picking unique labels and specifying the selector.  Failure to pick a unique label may cause this and other jobs to not function correctly.  However, You may see `manualSelector=true` in jobs that were created with the old `extensions/v1beta1` API. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/#specifying-your-own-pod-selector",
-	"template":              "Describes the pod that will be created when executing a job. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/",
+	"":                        "JobSpec describes how the job execution will look like.",
+	"parallelism":             "Specifies the maximum desired number of pods the job should run at any given time. The actual number of pods running in steady state will be less than this number when ((.spec.completions - .status.successful) < .spec.parallelism), i.e. when the work left to do is less than max parallelism. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/",
+	"completions":             "Specifies the desired number of successfully finished pods the job should be run with.  Setting to nil means that the success of any pod signals the success of all pods, and allows parallelism to have any positive value.  Setting to 1 means that parallelism is limited to 1 and the success of that pod signals the success of the job. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/",
+	"activeDeadlineSeconds":   "Specifies the duration in seconds relative to the startTime that the job may be active before the system tries to terminate it; value must be positive integer",
+	"backoffLimit":            "Specifies the number of retries before marking this job failed. Defaults to 6",
+	"selector":                "A label query over pods that should match the pod count. Normally, the system sets this field for you. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#label-selectors",
+	"manualSelector":          "manualSelector controls generation of pod labels and pod selectors. Leave `manualSelector` unset unless you are certain what you are doing. When false or unset, the system pick labels unique to this job and appends those labels to the pod template.  When true, the user is responsible for picking unique labels and specifying the selector.  Failure to pick a unique label may cause this and other jobs to not function correctly.  However, You may see `manualSelector=true` in jobs that were created with the old `extensions/v1beta1` API. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/#specifying-your-own-pod-selector",
+	"template":                "Describes the pod that will be created when executing a job. More info: https://kubernetes.io/docs/concepts/workloads/controllers/jobs-run-to-completion/",
+	"ttlSecondsAfterFinished": "ttlSecondsAfterFinished limits the lifetime of a Job that has finished execution (either Complete or Failed). If this field is set, ttlSecondsAfterFinished after the Job finishes, it is eligible to be automatically deleted. When the Job is being deleted, its lifecycle guarantees (e.g. finalizers) will be honored. If this field is unset, the Job won't be automatically deleted. If this field is set to zero, the Job becomes eligible to be deleted immediately after it finishes. This field is alpha-level and is only honored by servers that enable the TTLAfterFinished feature.",
 }
 
 func (JobSpec) SwaggerDoc() map[string]string {

staging/src/k8s.io/api/batch/v1/zz_generated.deepcopy.go

@@ -138,6 +138,11 @@ func (in *JobSpec) DeepCopyInto(out *JobSpec) {
 		**out = **in
 	}
 	in.Template.DeepCopyInto(&out.Template)
+	if in.TTLSecondsAfterFinished != nil {
+		in, out := &in.TTLSecondsAfterFinished, &out.TTLSecondsAfterFinished
+		*out = new(int32)
+		**out = **in
+	}
 	return
 }
 

staging/src/k8s.io/kube-controller-manager/config/v1alpha1/types.go

@@ -145,6 +145,9 @@ type KubeControllerManagerConfiguration struct {
 	// ServiceControllerConfiguration holds configuration for ServiceController
 	// related features.
 	ServiceController ServiceControllerConfiguration
+	// TTLAfterFinishedControllerConfiguration holds configuration for
+	// TTLAfterFinishedController related features.
+	TTLAfterFinishedController TTLAfterFinishedControllerConfiguration
 }
 
 // GenericControllerManagerConfiguration holds configuration for a generic controller-manager.
@@ -438,3 +441,10 @@ type ServiceControllerConfiguration struct {
 	// management, but more CPU (and network) load.
 	ConcurrentServiceSyncs int32
 }
+
+// TTLAfterFinishedControllerConfiguration contains elements describing TTLAfterFinishedController.
+type TTLAfterFinishedControllerConfiguration struct {
+	// concurrentTTLSyncs is the number of TTL-after-finished collector workers that are
+	// allowed to sync concurrently.
+	ConcurrentTTLSyncs int32
+}

staging/src/k8s.io/kube-controller-manager/config/v1alpha1/zz_generated.deepcopy.go

@@ -300,6 +300,7 @@ func (in *KubeControllerManagerConfiguration) DeepCopyInto(out *KubeControllerMa
 	out.ResourceQuotaController = in.ResourceQuotaController
 	out.SAController = in.SAController
 	out.ServiceController = in.ServiceController
+	out.TTLAfterFinishedController = in.TTLAfterFinishedController
 	return
 }
 
@@ -510,6 +511,22 @@ func (in *ServiceControllerConfiguration) DeepCopy() *ServiceControllerConfigura
 }
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TTLAfterFinishedControllerConfiguration) DeepCopyInto(out *TTLAfterFinishedControllerConfiguration) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TTLAfterFinishedControllerConfiguration.
+func (in *TTLAfterFinishedControllerConfiguration) DeepCopy() *TTLAfterFinishedControllerConfiguration {
+	if in == nil {
+		return nil
+	}
+	out := new(TTLAfterFinishedControllerConfiguration)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *VolumeConfiguration) DeepCopyInto(out *VolumeConfiguration) {
 	*out = *in
 	if in.EnableHostPathProvisioning != nil {

test/e2e/apps/job.go

@@ -44,7 +44,7 @@ var _ = SIGDescribe("Job", func() {
 		Expect(err).NotTo(HaveOccurred())
 
 		By("Ensuring job reaches completions")
-		err = framework.WaitForJobFinish(f.ClientSet, f.Namespace.Name, job.Name, completions)
+		err = framework.WaitForJobComplete(f.ClientSet, f.Namespace.Name, job.Name, completions)
 		Expect(err).NotTo(HaveOccurred())
 	})
 
@@ -63,7 +63,7 @@ var _ = SIGDescribe("Job", func() {
 		Expect(err).NotTo(HaveOccurred())
 
 		By("Ensuring job reaches completions")
-		err = framework.WaitForJobFinish(f.ClientSet, f.Namespace.Name, job.Name, completions)
+		err = framework.WaitForJobComplete(f.ClientSet, f.Namespace.Name, job.Name, completions)
 		Expect(err).NotTo(HaveOccurred())
 	})
 
@@ -84,7 +84,7 @@ var _ = SIGDescribe("Job", func() {
 		Expect(err).NotTo(HaveOccurred())
 
 		By("Ensuring job reaches completions")
-		err = framework.WaitForJobFinish(f.ClientSet, f.Namespace.Name, job.Name, *job.Spec.Completions)
+		err = framework.WaitForJobComplete(f.ClientSet, f.Namespace.Name, job.Name, *job.Spec.Completions)
 		Expect(err).NotTo(HaveOccurred())
 	})
 

test/e2e/auth/metadata_concealment.go

@@ -58,7 +58,7 @@ var _ = SIGDescribe("Metadata Concealment", func() {
 		Expect(err).NotTo(HaveOccurred())
 
 		By("Ensuring job reaches completions")
-		err = framework.WaitForJobFinish(f.ClientSet, f.Namespace.Name, job.Name, int32(1))
+		err = framework.WaitForJobComplete(f.ClientSet, f.Namespace.Name, job.Name, int32(1))
 		Expect(err).NotTo(HaveOccurred())
 	})
 })

test/e2e/common/BUILD

@@ -34,6 +34,7 @@ go_library(
         "secrets_volume.go",
         "security_context.go",
         "sysctl.go",
+        "ttlafterfinished.go",
         "util.go",
         "volumes.go",
     ],
@@ -43,13 +44,16 @@ go_library(
         "//pkg/apis/core:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/conditions:go_default_library",
+        "//pkg/features:go_default_library",
         "//pkg/kubelet:go_default_library",
         "//pkg/kubelet/apis:go_default_library",
         "//pkg/kubelet/images:go_default_library",
         "//pkg/kubelet/sysctl:go_default_library",
         "//pkg/security/apparmor:go_default_library",
+        "//pkg/util/slice:go_default_library",
         "//pkg/util/version:go_default_library",
         "//staging/src/k8s.io/api/autoscaling/v1:go_default_library",
+        "//staging/src/k8s.io/api/batch/v1:go_default_library",
         "//staging/src/k8s.io/api/coordination/v1beta1:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
@@ -65,6 +69,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/util/uuid:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/scale:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",

test/e2e/common/ttlafterfinished.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package common
+
+import (
+	"time"
+
+	batch "k8s.io/api/batch/v1"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/util/slice"
+	"k8s.io/kubernetes/test/e2e/framework"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+const dummyFinalizer = "k8s.io/dummy-finalizer"
+
+var _ = framework.KubeDescribe("TTLAfterFinished", func() {
+	f := framework.NewDefaultFramework("ttlafterfinished")
+
+	It("Job should be deleted once it finishes after TTL seconds [Feature:TTLAfterFinished]", func() {
+		if !utilfeature.DefaultFeatureGate.Enabled(features.TTLAfterFinished) {
+			framework.Skipf("Skip because %s feature is not enabled; run test with --feature-gates=%s=true", features.TTLAfterFinished, features.TTLAfterFinished)
+		}
+		testFinishedJob(f)
+	})
+})
+
+func cleanupJob(f *framework.Framework, job *batch.Job) {
+	ns := f.Namespace.Name
+	c := f.ClientSet
+
+	framework.Logf("Remove the Job's dummy finalizer; the Job should be deleted cascadingly")
+	removeFinalizerFunc := func(j *batch.Job) {
+		j.ObjectMeta.Finalizers = slice.RemoveString(j.ObjectMeta.Finalizers, dummyFinalizer, nil)
+	}
+	_, err := framework.UpdateJobWithRetries(c, ns, job.Name, removeFinalizerFunc)
+	Expect(err).NotTo(HaveOccurred())
+	framework.WaitForJobGone(c, ns, job.Name, wait.ForeverTestTimeout)
+
+	err = framework.WaitForAllJobPodsGone(c, ns, job.Name)
+	Expect(err).NotTo(HaveOccurred())
+}
+
+func testFinishedJob(f *framework.Framework) {
+	ns := f.Namespace.Name
+	c := f.ClientSet
+
+	parallelism := int32(1)
+	completions := int32(1)
+	backoffLimit := int32(2)
+	ttl := int32(10)
+
+	job := framework.NewTestJob("randomlySucceedOrFail", "rand-non-local", v1.RestartPolicyNever, parallelism, completions, nil, backoffLimit)
+	job.Spec.TTLSecondsAfterFinished = &ttl
+	job.ObjectMeta.Finalizers = []string{dummyFinalizer}
+	defer cleanupJob(f, job)
+
+	framework.Logf("Create a Job %s/%s with TTL", job.Namespace, job.Name)
+	job, err := framework.CreateJob(c, ns, job)
+	Expect(err).NotTo(HaveOccurred())
+
+	framework.Logf("Wait for the Job to finish")
+	err = framework.WaitForJobFinish(c, ns, job.Name)
+	Expect(err).NotTo(HaveOccurred())
+
+	framework.Logf("Wait for TTL after finished controller to delete the Job")
+	err = framework.WaitForJobDeleting(c, ns, job.Name)
+	Expect(err).NotTo(HaveOccurred())
+
+	framework.Logf("Check Job's deletionTimestamp and compare with the time when the Job finished")
+	job, err = framework.GetJob(c, ns, job.Name)
+	Expect(err).NotTo(HaveOccurred())
+	finishTime := framework.JobFinishTime(job)
+	finishTimeUTC := finishTime.UTC()
+	Expect(finishTime.IsZero()).NotTo(BeTrue())
+
+	deleteAtUTC := job.ObjectMeta.DeletionTimestamp.UTC()
+	Expect(deleteAtUTC).NotTo(BeNil())
+
+	expireAtUTC := finishTimeUTC.Add(time.Duration(ttl) * time.Second)
+	Expect(deleteAtUTC.Before(expireAtUTC)).To(BeFalse())
+}

test/e2e/framework/BUILD

@@ -59,6 +59,7 @@ go_library(
         "//pkg/cloudprovider/providers/gce:go_default_library",
         "//pkg/controller:go_default_library",
         "//pkg/controller/deployment/util:go_default_library",
+        "//pkg/controller/job:go_default_library",
         "//pkg/controller/nodelifecycle:go_default_library",
         "//pkg/controller/service:go_default_library",
         "//pkg/features:go_default_library",

test/e2e/framework/jobs_util.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/wait"
 	clientset "k8s.io/client-go/kubernetes"
+	jobutil "k8s.io/kubernetes/pkg/controller/job"
 )
 
 const (
@@ -181,8 +182,8 @@ func WaitForAllJobPodsRunning(c clientset.Interface, ns, jobName string, paralle
 	})
 }
 
-// WaitForJobFinish uses c to wait for compeletions to complete for the Job jobName in namespace ns.
-func WaitForJobFinish(c clientset.Interface, ns, jobName string, completions int32) error {
+// WaitForJobComplete uses c to wait for compeletions to complete for the Job jobName in namespace ns.
+func WaitForJobComplete(c clientset.Interface, ns, jobName string, completions int32) error {
 	return wait.Poll(Poll, JobTimeout, func() (bool, error) {
 		curr, err := c.BatchV1().Jobs(ns).Get(jobName, metav1.GetOptions{})
 		if err != nil {
@@ -192,6 +193,17 @@ func WaitForJobFinish(c clientset.Interface, ns, jobName string, completions int
 	})
 }
 
+// WaitForJobFinish uses c to wait for the Job jobName in namespace ns to finish (either Failed or Complete).
+func WaitForJobFinish(c clientset.Interface, ns, jobName string) error {
+	return wait.PollImmediate(Poll, JobTimeout, func() (bool, error) {
+		curr, err := c.BatchV1().Jobs(ns).Get(jobName, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		return jobutil.IsJobFinished(curr), nil
+	})
+}
+
 // WaitForJobFailure uses c to wait for up to timeout for the Job named jobName in namespace ns to fail.
 func WaitForJobFailure(c clientset.Interface, ns, jobName string, timeout time.Duration, reason string) error {
 	return wait.Poll(Poll, timeout, func() (bool, error) {
@@ -239,6 +251,18 @@ func CheckForAllJobPodsRunning(c clientset.Interface, ns, jobName string, parall
 	return count == parallelism, nil
 }
 
+// WaitForAllJobPodsRunning wait for all pods for the Job named jobName in namespace ns
+// to be deleted.
+func WaitForAllJobPodsGone(c clientset.Interface, ns, jobName string) error {
+	return wait.PollImmediate(Poll, JobTimeout, func() (bool, error) {
+		pods, err := GetJobPods(c, ns, jobName)
+		if err != nil {
+			return false, err
+		}
+		return len(pods.Items) == 0, nil
+	})
+}
+
 func newBool(val bool) *bool {
 	p := new(bool)
 	*p = val
@@ -250,7 +274,7 @@ type updateJobFunc func(*batch.Job)
 func UpdateJobWithRetries(c clientset.Interface, namespace, name string, applyUpdate updateJobFunc) (job *batch.Job, err error) {
 	jobs := c.BatchV1().Jobs(namespace)
 	var updateErr error
-	pollErr := wait.PollImmediate(10*time.Millisecond, 1*time.Minute, func() (bool, error) {
+	pollErr := wait.PollImmediate(Poll, JobTimeout, func() (bool, error) {
 		if job, err = jobs.Get(name, metav1.GetOptions{}); err != nil {
 			return false, err
 		}
@@ -268,3 +292,25 @@ func UpdateJobWithRetries(c clientset.Interface, namespace, name string, applyUp
 	}
 	return job, pollErr
 }
+
+// WaitForJobDeleting uses c to wait for the Job jobName in namespace ns to have
+// a non-nil deletionTimestamp (i.e. being deleted).
+func WaitForJobDeleting(c clientset.Interface, ns, jobName string) error {
+	return wait.PollImmediate(Poll, JobTimeout, func() (bool, error) {
+		curr, err := c.BatchV1().Jobs(ns).Get(jobName, metav1.GetOptions{})
+		if err != nil {
+			return false, err
+		}
+		return curr.ObjectMeta.DeletionTimestamp != nil, nil
+	})
+}
+
+func JobFinishTime(finishedJob *batch.Job) metav1.Time {
+	var finishTime metav1.Time
+	for _, c := range finishedJob.Status.Conditions {
+		if (c.Type == batch.JobComplete || c.Type == batch.JobFailed) && c.Status == v1.ConditionTrue {
+			return c.LastTransitionTime
+		}
+	}
+	return finishTime
+}