Add Job.status.uncountedTerminatedPods for Job tracking

pkg/apis/batch/types.go

@@ -18,9 +18,18 @@ package batch
 
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	api "k8s.io/kubernetes/pkg/apis/core"
 )
 
+// JobTrackingFinalizer is a finalizer for Job's pods. It prevents them from
+// being deleted before being accounted in the Job status.
+// The apiserver and job controller use this string as a Job annotation, to
+// mark Jobs that are being tracked using pod finalizers. Two releases after
+// the JobTrackingWithFinalizers graduates to GA, JobTrackingFinalizer will
+// no longer be used as a Job annotation.
+const JobTrackingFinalizer = "batch.kubernetes.io/job-tracking"
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // Job represents the configuration of a single job.
@@ -255,6 +264,38 @@ type JobStatus struct {
 	// represented as "1,3-5,7".
 	// +optional
 	CompletedIndexes string
+
+	// UncountedTerminatedPods holds the UIDs of Pods that have terminated but
+	// the job controller hasn't yet accounted for in the status counters.
+	//
+	// The job controller creates pods with a finalizer. When a pod terminates
+	// (succeeded or failed), the controller does three steps to account for it
+	// in the job status:
+	// (1) Add the pod UID to the corresponding array in this field.
+	// (2) Remove the pod finalizer.
+	// (3) Remove the pod UID from the array while increasing the corresponding
+	//     counter.
+	//
+	// This field is alpha-level. The job controller only makes use of this field
+	// when the feature gate PodTrackingWithFinalizers is enabled.
+	// Old jobs might not be tracked using this field, in which case the field
+	// remains null.
+	// +optional
+	UncountedTerminatedPods *UncountedTerminatedPods
+}
+
+// UncountedTerminatedPods holds UIDs of Pods that have terminated but haven't
+// been accounted in Job status counters.
+type UncountedTerminatedPods struct {
+	// Succeeded holds UIDs of succeeded Pods.
+	// +listType=set
+	// +optional
+	Succeeded []types.UID
+
+	// Failed holds UIDs of failed Pods.
+	// +listType=set
+	// +optional
+	Failed []types.UID
 }
 
 // JobConditionType is a valid value for JobCondition.Type

pkg/apis/batch/validation/validation.go

@@ -20,10 +20,10 @@ import (
 	"fmt"
 
 	"github.com/robfig/cron/v3"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	unversionedvalidation "k8s.io/apimachinery/pkg/apis/meta/v1/validation"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/sets"
 	apimachineryvalidation "k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/kubernetes/pkg/apis/batch"
@@ -83,11 +83,14 @@ func ValidateGeneratedSelector(obj *batch.Job) field.ErrorList {
 }
 
 // ValidateJob validates a Job and returns an ErrorList with any errors.
-func ValidateJob(job *batch.Job, opts apivalidation.PodValidationOptions) field.ErrorList {
+func ValidateJob(job *batch.Job, opts JobValidationOptions) field.ErrorList {
 	// Jobs and rcs have the same name validation
 	allErrs := apivalidation.ValidateObjectMeta(&job.ObjectMeta, true, apivalidation.ValidateReplicationControllerName, field.NewPath("metadata"))
 	allErrs = append(allErrs, ValidateGeneratedSelector(job)...)
-	allErrs = append(allErrs, ValidateJobSpec(&job.Spec, field.NewPath("spec"), opts)...)
+	allErrs = append(allErrs, ValidateJobSpec(&job.Spec, field.NewPath("spec"), opts.PodValidationOptions)...)
+	if !opts.AllowTrackingAnnotation && hasJobTrackingAnnotation(job) {
+		allErrs = append(allErrs, field.Forbidden(field.NewPath("metadata").Child("annotations").Key(batch.JobTrackingFinalizer), "cannot add this annotation"))
+	}
 	if job.Spec.CompletionMode != nil && *job.Spec.CompletionMode == batch.IndexedCompletion && job.Spec.Completions != nil && *job.Spec.Completions > 0 {
 		// For indexed job, the job controller appends a suffix (`-$INDEX`)
 		// to the pod hostname when indexed job create pods.
@@ -101,6 +104,14 @@ func ValidateJob(job *batch.Job, opts apivalidation.PodValidationOptions) field.
 	return allErrs
 }
 
+func hasJobTrackingAnnotation(job *batch.Job) bool {
+	if job.Annotations == nil {
+		return false
+	}
+	_, ok := job.Annotations[batch.JobTrackingFinalizer]
+	return ok
+}
+
 // ValidateJobSpec validates a JobSpec and returns an ErrorList with any errors.
 func ValidateJobSpec(spec *batch.JobSpec, fldPath *field.Path, opts apivalidation.PodValidationOptions) field.ErrorList {
 	allErrs := validateJobSpec(spec, fldPath, opts)
@@ -169,12 +180,36 @@ func validateJobSpec(spec *batch.JobSpec, fldPath *field.Path, opts apivalidatio
 	return allErrs
 }
 
-// ValidateJobStatus validates a JobStatus and returns an ErrorList with any errors.
-func ValidateJobStatus(status *batch.JobStatus, fldPath *field.Path) field.ErrorList {
+// validateJobStatus validates a JobStatus and returns an ErrorList with any errors.
+func validateJobStatus(status *batch.JobStatus, fldPath *field.Path) field.ErrorList {
 	allErrs := field.ErrorList{}
 	allErrs = append(allErrs, apivalidation.ValidateNonnegativeField(int64(status.Active), fldPath.Child("active"))...)
 	allErrs = append(allErrs, apivalidation.ValidateNonnegativeField(int64(status.Succeeded), fldPath.Child("succeeded"))...)
 	allErrs = append(allErrs, apivalidation.ValidateNonnegativeField(int64(status.Failed), fldPath.Child("failed"))...)
+	if status.UncountedTerminatedPods != nil {
+		path := fldPath.Child("uncountedTerminatedPods")
+		seen := sets.NewString()
+		for i, k := range status.UncountedTerminatedPods.Succeeded {
+			p := path.Child("succeeded").Index(i)
+			if k == "" {
+				allErrs = append(allErrs, field.Invalid(p, k, "must not be empty"))
+			} else if seen.Has(string(k)) {
+				allErrs = append(allErrs, field.Duplicate(p, k))
+			} else {
+				seen.Insert(string(k))
+			}
+		}
+		for i, k := range status.UncountedTerminatedPods.Failed {
+			p := path.Child("failed").Index(i)
+			if k == "" {
+				allErrs = append(allErrs, field.Invalid(p, k, "must not be empty"))
+			} else if seen.Has(string(k)) {
+				allErrs = append(allErrs, field.Duplicate(p, k))
+			} else {
+				seen.Insert(string(k))
+			}
+		}
+	}
 	return allErrs
 }
 
@@ -206,7 +241,7 @@ func ValidateJobSpecUpdate(spec, oldSpec batch.JobSpec, fldPath *field.Path, opt
 // ValidateJobStatusUpdate validates an update to a JobStatus and returns an ErrorList with any errors.
 func ValidateJobStatusUpdate(status, oldStatus batch.JobStatus) field.ErrorList {
 	allErrs := field.ErrorList{}
-	allErrs = append(allErrs, ValidateJobStatus(&status, field.NewPath("status"))...)
+	allErrs = append(allErrs, validateJobStatus(&status, field.NewPath("status"))...)
 	return allErrs
 }
 
@@ -306,3 +341,9 @@ func ValidateJobTemplateSpec(spec *batch.JobTemplateSpec, fldPath *field.Path, o
 	}
 	return allErrs
 }
+
+type JobValidationOptions struct {
+	apivalidation.PodValidationOptions
+	// Allow Job to have the annotation batch.kubernetes.io/job-tracking
+	AllowTrackingAnnotation bool
+}