Use conditions rather than pod counters to wait for jobs

Signed-off-by: Ben Lewis <blewis@auditboard.com>

pkg/kube/ready.go

@@ -18,7 +18,6 @@ package kube // import "helm.sh/helm/v3/pkg/kube"
 
 import (
 	"context"
-	"fmt"
 
 	appsv1 "k8s.io/api/apps/v1"
 	appsv1beta1 "k8s.io/api/apps/v1beta1"
@@ -35,6 +34,8 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/scheme"
 
+	"github.com/pkg/errors"
+
 	deploymentutil "helm.sh/helm/v3/internal/third_party/k8s.io/kubernetes/deployment/util"
 )
 
@@ -110,7 +111,9 @@ func (c *ReadyChecker) IsReady(ctx context.Context, v *resource.Info) (bool, err
 				return false, err
 			}
 			ready, err := c.jobReady(job)
-			return ready, err
+			if !ready {
+				return false, err
+			}
 		}
 	case *appsv1.Deployment, *appsv1beta1.Deployment, *appsv1beta2.Deployment, *extensionsv1beta1.Deployment:
 		currentDeployment, err := c.client.AppsV1().Deployments(v.Namespace).Get(ctx, v.Name, metav1.GetOptions{})
@@ -226,16 +229,14 @@ func (c *ReadyChecker) isPodReady(pod *corev1.Pod) bool {
 }
 
 func (c *ReadyChecker) jobReady(job *batchv1.Job) (bool, error) {
-	if job.Status.Failed > *job.Spec.BackoffLimit {
-		c.log("Job is failed: %s/%s", job.GetNamespace(), job.GetName())
-		// If a job is failed, it can't recover, so throw an error
-		return false, fmt.Errorf("job is failed: %s/%s", job.GetNamespace(), job.GetName())
-	}
-	if job.Spec.Completions != nil && job.Status.Succeeded < *job.Spec.Completions {
-		c.log("Job is not completed: %s/%s", job.GetNamespace(), job.GetName())
-		return false, nil
+	for _, condition := range job.Status.Conditions {
+		if condition.Type == batchv1.JobComplete && condition.Status == "True" {
+			return true, nil
+		} else if condition.Type == batchv1.JobFailed && condition.Status == "True" {
+			return false, errors.Errorf("job failed: %s", condition.Reason)
+		}
 	}
-	return true, nil
+	return false, nil
 }
 
 func (c *ReadyChecker) serviceReady(s *corev1.Service) bool {

pkg/kube/ready_test.go

@@ -241,51 +241,22 @@ func Test_ReadyChecker_jobReady(t *testing.T) {
 	}{
 		{
 			name:    "job is completed",
-			args:    args{job: newJob("foo", 1, intToInt32(1), 1, 0)},
+			args:    args{job: newJob("foo", intToInt32(1), true, false)},
 			want:    true,
 			wantErr: false,
 		},
 		{
 			name:    "job is incomplete",
-			args:    args{job: newJob("foo", 1, intToInt32(1), 0, 0)},
+			args:    args{job: newJob("foo", intToInt32(1), false, false)},
 			want:    false,
 			wantErr: false,
 		},
 		{
-			name:    "job is failed but within BackoffLimit",
-			args:    args{job: newJob("foo", 1, intToInt32(1), 0, 1)},
-			want:    false,
-			wantErr: false,
-		},
-		{
-			name:    "job is completed with retry",
-			args:    args{job: newJob("foo", 1, intToInt32(1), 1, 1)},
-			want:    true,
-			wantErr: false,
-		},
-		{
-			name:    "job is failed and beyond BackoffLimit",
-			args:    args{job: newJob("foo", 1, intToInt32(1), 0, 2)},
+			name:    "job is failed",
+			args:    args{job: newJob("foo", intToInt32(1), false, true)},
 			want:    false,
 			wantErr: true,
 		},
-		{
-			name:    "job is completed single run",
-			args:    args{job: newJob("foo", 0, intToInt32(1), 1, 0)},
-			want:    true,
-			wantErr: false,
-		},
-		{
-			name:    "job is failed single run",
-			args:    args{job: newJob("foo", 0, intToInt32(1), 0, 1)},
-			want:    false,
-			wantErr: true,
-		},
-		{
-			name: "job with null completions",
-			args: args{job: newJob("foo", 0, nil, 1, 0)},
-			want: true,
-		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -498,14 +469,27 @@ func newPersistentVolumeClaim(name string, phase corev1.PersistentVolumeClaimPha
 	}
 }
 
-func newJob(name string, backoffLimit int, completions *int32, succeeded int, failed int) *batchv1.Job {
+func newJob(name string, completions *int32, succeeded bool, failed bool) *batchv1.Job {
+	var conditions []batchv1.JobCondition
+	if succeeded {
+		conditions = append(conditions, batchv1.JobCondition{
+			Type:   batchv1.JobComplete,
+			Status: "True",
+		})
+	}
+	if failed {
+		conditions = append(conditions, batchv1.JobCondition{
+			Type:   batchv1.JobFailed,
+			Status: "True",
+		})
+	}
 	return &batchv1.Job{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
 			Namespace: defaultNamespace,
 		},
 		Spec: batchv1.JobSpec{
-			BackoffLimit: intToInt32(backoffLimit),
+			BackoffLimit: intToInt32(1),
 			Completions:  completions,
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
@@ -522,8 +506,7 @@ func newJob(name string, backoffLimit int, completions *int32, succeeded int, fa
 			},
 		},
 		Status: batchv1.JobStatus{
-			Succeeded: int32(succeeded),
-			Failed:    int32(failed),
+			Conditions: conditions,
 		},
 	}
 }