WIP: pkg/cvo/updatepayload: Drop the Job controller for release-manifests downloads

Previously, the CVO launched a Job and waited for it to complete to
get manifests for an incoming release payload.  But the Job controller
doesn't bubble up details about why the pod has trouble
(e.g. Init:SignatureValidationFailed), so to get those details, we
need direct access to the Pod.  The Job controller doesn't seem like
it's adding much value here, so we're dropping it and monitoring the
Pod ourselves.

Author: wking

lib/resourcebuilder/batch.go

@@ -7,32 +7,10 @@ import (
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
 	batchclientv1 "k8s.io/client-go/kubernetes/typed/batch/v1"
 	"k8s.io/klog/v2"
 )
 
-// WaitForJobCompletion waits for job to complete.
-func WaitForJobCompletion(ctx context.Context, client batchclientv1.JobsGetter, job *batchv1.Job) error {
-	return wait.PollUntilContextCancel(ctx, defaultObjectPollInterval, true, func(localCtx context.Context) (bool, error) {
-		j, err := client.Jobs(job.Namespace).Get(localCtx, job.Name, metav1.GetOptions{})
-		if err != nil {
-			return false, fmt.Errorf("error getting Job %s: %v", job.Name, err)
-		}
-
-		if done, err := checkJobHealth(localCtx, j); err != nil && done {
-			return false, err
-		} else if err != nil {
-			klog.Error(err)
-			return false, nil
-		} else if !done {
-			klog.V(2).Infof("Job %s in namespace %s is not ready, continuing to wait.", job.ObjectMeta.Name, job.ObjectMeta.Namespace)
-			return false, nil
-		}
-		return true, nil
-	})
-}
-
 func (b *builder) checkJobHealth(ctx context.Context, job *batchv1.Job) error {
 	if b.mode == InitializingMode {
 		return nil

pkg/cvo/updatepayload.go

@@ -13,12 +13,12 @@ import (
 
 	"github.com/pkg/errors"
 
-	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	randutil "k8s.io/apimachinery/pkg/util/rand"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
@@ -162,9 +162,9 @@ func (r *payloadRetriever) targetUpdatePayloadDir(ctx context.Context, update co
 	payloadHash := base64.RawURLEncoding.EncodeToString(hash.Sum(nil))
 	tdir := filepath.Join(r.workingDir, payloadHash)
 
-	// Prune older jobs and directories while gracefully handling errors.
-	if err := r.pruneJobs(ctx, 0); err != nil {
-		klog.Warningf("failed to prune jobs: %v", err)
+	// Prune older pods and directories while gracefully handling errors.
+	if err := r.prunePods(ctx); err != nil {
+		klog.Warningf("failed to prune pods: %v", err)
 	}
 
 	if err := payload.ValidateDirectory(tdir); os.IsNotExist(err) {
@@ -217,123 +217,132 @@ func (r *payloadRetriever) fetchUpdatePayloadToDir(ctx context.Context, dir stri
 		return container
 	}
 
-	job := &batchv1.Job{
+	pod := &v1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
 			Namespace: namespace,
+			Labels: map[string]string{
+				"k8s-app": "retrieve-openshift-release",
+			},
 		},
-		Spec: batchv1.JobSpec{
+		Spec: corev1.PodSpec{
 			ActiveDeadlineSeconds: deadline,
-			Template: corev1.PodTemplateSpec{
-				Spec: corev1.PodSpec{
-					InitContainers: []corev1.Container{
-						setContainerDefaults(corev1.Container{
-							Name:       "cleanup",
-							Command:    []string{"sh", "-c", "rm -fR ./*"},
-							WorkingDir: baseDir,
-						}),
-						setContainerDefaults(corev1.Container{
-							Name:    "make-temporary-directory",
-							Command: []string{"mkdir", tmpDir},
-						}),
-						setContainerDefaults(corev1.Container{
-							Name: "move-operator-manifests-to-temporary-directory",
-							Command: []string{
-								"mv",
-								filepath.Join(payload.DefaultPayloadDir, payload.CVOManifestDir),
-								filepath.Join(tmpDir, payload.CVOManifestDir),
-							},
-						}),
-						setContainerDefaults(corev1.Container{
-							Name: "move-release-manifests-to-temporary-directory",
-							Command: []string{
-								"mv",
-								filepath.Join(payload.DefaultPayloadDir, payload.ReleaseManifestDir),
-								filepath.Join(tmpDir, payload.ReleaseManifestDir),
-							},
-						}),
+			InitContainers: []corev1.Container{
+				setContainerDefaults(corev1.Container{
+					Name:       "cleanup",
+					Command:    []string{"sh", "-c", "rm -fR ./*"},
+					WorkingDir: baseDir,
+				}),
+				setContainerDefaults(corev1.Container{
+					Name:    "make-temporary-directory",
+					Command: []string{"mkdir", tmpDir},
+				}),
+				setContainerDefaults(corev1.Container{
+					Name: "move-operator-manifests-to-temporary-directory",
+					Command: []string{
+						"mv",
+						filepath.Join(payload.DefaultPayloadDir, payload.CVOManifestDir),
+						filepath.Join(tmpDir, payload.CVOManifestDir),
 					},
-					Containers: []corev1.Container{
-						setContainerDefaults(corev1.Container{
-							Name:    "rename-to-final-location",
-							Command: []string{"mv", tmpDir, dir},
-						}),
+				}),
+				setContainerDefaults(corev1.Container{
+					Name: "move-release-manifests-to-temporary-directory",
+					Command: []string{
+						"mv",
+						filepath.Join(payload.DefaultPayloadDir, payload.ReleaseManifestDir),
+						filepath.Join(tmpDir, payload.ReleaseManifestDir),
 					},
-					Volumes: []corev1.Volume{{
-						Name: "payloads",
-						VolumeSource: corev1.VolumeSource{
-							HostPath: &corev1.HostPathVolumeSource{
-								Path: targetUpdatePayloadsDir,
-							},
-						},
-					}},
-					NodeName: nodename,
-					NodeSelector: map[string]string{
-						nodeSelectorKey: "",
+				}),
+			},
+			Containers: []corev1.Container{
+				setContainerDefaults(corev1.Container{
+					Name:    "rename-to-final-location",
+					Command: []string{"mv", tmpDir, dir},
+				}),
+			},
+			Volumes: []corev1.Volume{{
+				Name: "payloads",
+				VolumeSource: corev1.VolumeSource{
+					HostPath: &corev1.HostPathVolumeSource{
+						Path: targetUpdatePayloadsDir,
 					},
-					PriorityClassName: "openshift-user-critical",
-					Tolerations: []corev1.Toleration{{
-						Key: nodeSelectorKey,
-					}},
-					RestartPolicy: corev1.RestartPolicyOnFailure,
 				},
+			}},
+			NodeName: nodename,
+			NodeSelector: map[string]string{
+				nodeSelectorKey: "",
 			},
+			PriorityClassName: "openshift-user-critical",
+			Tolerations: []corev1.Toleration{{
+				Key: nodeSelectorKey,
+			}},
+			RestartPolicy: corev1.RestartPolicyOnFailure,
 		},
-	}
+	},
 
-	if _, err := r.kubeClient.BatchV1().Jobs(job.Namespace).Create(ctx, job, metav1.CreateOptions{}); err != nil {
+	if _, err := r.kubeClient.V1().Pods(pod.Namespace).Create(ctx, pod, metav1.CreateOptions{}); err != nil {
 		return err
 	}
-	return resourcebuilder.WaitForJobCompletion(ctx, r.kubeClient.BatchV1(), job)
+
+	return wait.PollUntilContextCancel(ctx, 3 * time.Second, true, func(localCtx context.Context) (bool, error) {
+		p, err := r.kubeClient.V1().Pods(pod.Namespace).Get(localCtx, pod.Name, metav1.GetOptions{})
+		if err != nil {
+			klog.Warningf("unable to get OpenShift release retrieval pod: %v", err)
+			return false, nil
+		}
+ 
+		if p.Status.Phase != "", {
+			return false, err
+		} else if err != nil {
+			klog.Error(err)
+			return false, nil
+		} else if !done {
+			klog.V(2).Infof("Job %s in namespace %s is not ready, continuing to wait.", job.ObjectMeta.Name, job.ObjectMeta.Namespace)
+			return false, nil
+		}
+		return true, nil
+	})
 }
 
-// pruneJobs deletes the older, finished jobs in the namespace.
-// retain - the number of newest jobs to keep.
-func (r *payloadRetriever) pruneJobs(ctx context.Context, retain int) error {
+// prunePods deletes the older, finished pods in the namespace.
+func (r *payloadRetriever) prunePods(ctx context.Context) error {
+	var errs []error
+
+	// begin transitional job pruning, in case any dangled from earlier versions
 	jobs, err := r.kubeClient.BatchV1().Jobs(r.namespace).List(ctx, metav1.ListOptions{})
 	if err != nil {
-		return err
-	}
-	if len(jobs.Items) <= retain {
-		return nil
+		errs = append(errs, err)
 	}
 
-	// Select jobs to be deleted
-	var deleteJobs []batchv1.Job
 	for _, job := range jobs.Items {
-		switch {
-		// Ignore jobs not beginning with operatorName
-		case !strings.HasPrefix(job.Name, r.operatorName+"-"):
-			break
-		default:
-			deleteJobs = append(deleteJobs, job)
+		if !strings.HasPrefix(job.Name, r.operatorName+"-") {
+			// Ignore jobs not beginning with operatorName
+			continue
+		}
+		err := r.kubeClient.BatchV1().Jobs(r.namespace).Delete(ctx, job.Name, metav1.DeleteOptions{})
+		if err != nil {
+			errs = append(errs, errors.Wrapf(err, "failed to delete job %v", job.Name))
 		}
 	}
-	if len(deleteJobs) <= retain {
-		return nil
-	}
+	// end transitional job pruning
 
-	// Sort jobs by StartTime to determine the newest. nil StartTime is assumed newest.
-	sort.Slice(deleteJobs, func(i, j int) bool {
-		if deleteJobs[i].Status.StartTime == nil {
-			return false
-		}
-		if deleteJobs[j].Status.StartTime == nil {
-			return true
-		}
-		return deleteJobs[i].Status.StartTime.Before(deleteJobs[j].Status.StartTime)
+	pods, err := r.kubeClient.BatchV1().Pods(r.namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: "k8s-app=retrieve-openshift-release",
 	})
+	if err != nil {
+		errs = append(errs, err)
+	}
 
-	var errs []error
-	for _, job := range deleteJobs[:len(deleteJobs)-retain] {
-		err := r.kubeClient.BatchV1().Jobs(r.namespace).Delete(ctx, job.Name, metav1.DeleteOptions{})
+	for _, pod := pods.Items {
+		err := r.kubeClient.V1().Pods(r.namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{})
 		if err != nil {
-			errs = append(errs, errors.Wrapf(err, "failed to delete job %v", job.Name))
+			errs = append(errs, errors.Wrapf(err, "failed to delete pod %v", pod.Name))
 		}
 	}
+
 	agg := utilerrors.NewAggregate(errs)
 	if agg != nil {
-		return fmt.Errorf("error deleting jobs: %v", agg.Error())
+		return fmt.Errorf("error deleting pods: %v", agg.Error())
 	}
 	return nil
 }