diff --git a/pkg/clients/kube/kube.go b/pkg/clients/kube/kube.go index 73e50392..a168000f 100644 --- a/pkg/clients/kube/kube.go +++ b/pkg/clients/kube/kube.go @@ -104,6 +104,9 @@ func (c *Client) GetJobLogs(ctx context.Context, jobName string) (string, error) return string(logs), nil } +// errJobIncomplete is a sentinel error to indicate a job is still in progress +var errJobIncomplete = fmt.Errorf("job incomplete") + // WaitForJobCompletion waits for a job to complete (succeed or fail). It's context-aware, // so set ctx to something like `context.WithTimeout(ctx, 10*time.Second)` to set a timeout. func (c *Client) WaitForJobCompletion(ctx context.Context, jobName string) error { @@ -113,22 +116,49 @@ func (c *Client) WaitForJobCompletion(ctx context.Context, jobName string) error } defer watcher.Stop() + // Add a ticker to periodically check job status directly as a fallback + // This handles cases where watch events are delayed, missing, or incomplete + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + + checkJobCompletion := func(job *batchv1.Job) error { + // Rely on Kubernetes' own logic via job conditions + jobStatus := getJobStatus(job) + if jobStatus == batchv1.JobComplete || jobStatus == batchv1.JobSuccessCriteriaMet { + return nil + } + if jobStatus == batchv1.JobFailed || jobStatus == batchv1.JobFailureTarget { + return fmt.Errorf("job %s failed", jobName) + } + + // Job not yet complete + return errJobIncomplete + } + for { select { case <-ctx.Done(): return fmt.Errorf("context cancelled while waiting for job %s to complete", jobName) - case event := <-watcher.ResultChan(): - job, ok := event.Object.(*batchv1.Job) + case <-ticker.C: + // Polling fallback: directly query job status + currentJob, err := c.GetJob(ctx, jobName) + if err != nil { + continue + } + if err := checkJobCompletion(currentJob); err != errJobIncomplete { + return err + } + case event, ok := <-watcher.ResultChan(): if !ok { + // Watch channel closed, fall back to polling continue } - - jobStatus := getJobStatus(job) - if jobStatus == batchv1.JobComplete || jobStatus == batchv1.JobSuccessCriteriaMet { - return nil + job, isJob := event.Object.(*batchv1.Job) + if !isJob { + continue } - if jobStatus == batchv1.JobFailed || jobStatus == batchv1.JobFailureTarget { - return fmt.Errorf("job %s failed", jobName) + if err := checkJobCompletion(job); err != errJobIncomplete { + return err } } } diff --git a/pkg/clients/kube/kube_test.go b/pkg/clients/kube/kube_test.go index f276a89b..20e6071a 100644 --- a/pkg/clients/kube/kube_test.go +++ b/pkg/clients/kube/kube_test.go @@ -3,7 +3,9 @@ package kube import ( "context" "reflect" + "strings" "testing" + "time" batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" @@ -309,3 +311,121 @@ func getMockedJob(name string) *batchv1.Job { }, } } + +func TestClient_WaitForJobCompletion(t *testing.T) { + tests := []struct { + name string + jobName string + setupJob func(*fak8s.Clientset) error + wantErr bool + errContains string + timeoutSeconds int + }{ + { + name: "job completes with JobComplete condition", + jobName: "test-job-condition-complete", + setupJob: func(clientset *fak8s.Clientset) error { + job := getMockedJob("test-job-condition-complete") + job.Status.Conditions = []batchv1.JobCondition{ + { + Type: batchv1.JobComplete, + Status: corev1.ConditionTrue, + }, + } + _, err := clientset.BatchV1().Jobs(testNamespace).Create(context.Background(), job, metav1.CreateOptions{}) + return err + }, + wantErr: false, + timeoutSeconds: 15, + }, + { + name: "job fails with JobFailed condition", + jobName: "test-job-condition-failed", + setupJob: func(clientset *fak8s.Clientset) error { + job := getMockedJob("test-job-condition-failed") + job.Status.Conditions = []batchv1.JobCondition{ + { + Type: batchv1.JobFailed, + Status: corev1.ConditionTrue, + }, + } + _, err := clientset.BatchV1().Jobs(testNamespace).Create(context.Background(), job, metav1.CreateOptions{}) + return err + }, + wantErr: true, + errContains: "failed", + timeoutSeconds: 15, + }, + { + name: "job completes with JobSuccessCriteriaMet condition", + jobName: "test-job-success-criteria-met", + setupJob: func(clientset *fak8s.Clientset) error { + job := getMockedJob("test-job-success-criteria-met") + job.Status.Conditions = []batchv1.JobCondition{ + { + Type: batchv1.JobSuccessCriteriaMet, + Status: corev1.ConditionTrue, + }, + } + _, err := clientset.BatchV1().Jobs(testNamespace).Create(context.Background(), job, metav1.CreateOptions{}) + return err + }, + wantErr: false, + timeoutSeconds: 15, + }, + { + name: "job fails with JobFailureTarget condition", + jobName: "test-job-failure-target", + setupJob: func(clientset *fak8s.Clientset) error { + job := getMockedJob("test-job-failure-target") + job.Status.Conditions = []batchv1.JobCondition{ + { + Type: batchv1.JobFailureTarget, + Status: corev1.ConditionTrue, + }, + } + _, err := clientset.BatchV1().Jobs(testNamespace).Create(context.Background(), job, metav1.CreateOptions{}) + return err + }, + wantErr: true, + errContains: "failed", + timeoutSeconds: 15, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Create a new mock clientset for each test + mockedClientset := fak8s.NewClientset() + setupJobMockingReactors(mockedClientset) + + // Setup the job + if err := tt.setupJob(mockedClientset); err != nil { + t.Fatalf("Failed to setup job: %v", err) + } + + c := &Client{ + clientset: mockedClientset, + namespace: testNamespace, + } + + // Create context with timeout + ctx, cancel := context.WithTimeout(context.Background(), time.Duration(tt.timeoutSeconds)*time.Second) + defer cancel() + + err := c.WaitForJobCompletion(ctx, tt.jobName) + + if tt.wantErr { + if err == nil { + t.Errorf("WaitForJobCompletion() expected error but got nil") + } else if tt.errContains != "" && !strings.Contains(err.Error(), tt.errContains) { + t.Errorf("WaitForJobCompletion() error = %v, want error containing %v", err, tt.errContains) + } + } else { + if err != nil { + t.Errorf("WaitForJobCompletion() unexpected error = %v", err) + } + } + }) + } +}