[release/0.16] Fix old job replacement

Dockerfile

@@ -5,13 +5,11 @@ RUN apk add --no-cache bash git gcc musl-dev
 WORKDIR /src
 COPY . .
 
-RUN GOPROXY=direct go install sigs.k8s.io/controller-tools/cmd/controller-gen@257e3a04698a16ea834c49f457de7704474f9a74
-RUN GOPROXY=direct go install github.com/elastic/crd-ref-docs@7de989285647936ac62ea1ff8887e25e0056bc58
-
-RUN go generate ./...
-
 RUN --mount=type=cache,id=gomod,target=/go/pkg/mod \
     --mount=type=cache,id=gobuild,target=/root/.cache/go-build \
+    GOPROXY=direct go install sigs.k8s.io/controller-tools/cmd/controller-gen@257e3a04698a16ea834c49f457de7704474f9a74; \
+    GOPROXY=direct go install github.com/elastic/crd-ref-docs@7de989285647936ac62ea1ff8887e25e0056bc58; \
+    go generate ./...; \
     ./scripts/build
 
 FROM scratch AS binary
@@ -35,13 +33,17 @@ RUN if [ "${ARCH}" != "arm" ]; then \
     tar --strip-components=1 -xzf golangci-lint.tar.gz -C /usr/local/bin golangci-lint-${GOLANGCI_VERSION#v}-linux-${ARCH}/golangci-lint && \
     rm -f /tmp/golangci-lint.tar.gz; \
     fi
-RUN if [ "${ARCH}" = "amd64" ]; then \
-    go install sigs.k8s.io/kustomize/kustomize/v5@1155ccbe3c1fc56cbbd82847899f86c7b824e005; \
+RUN --mount=type=cache,id=gomod,target=/go/pkg/mod \
+    --mount=type=cache,id=gobuild,target=/root/.cache/go-build \
+    if [ "${ARCH}" = "amd64" ]; then \
+      go install sigs.k8s.io/kustomize/kustomize/v5@1155ccbe3c1fc56cbbd82847899f86c7b824e005; \
     fi
 
 WORKDIR /src
 COPY go.mod go.sum pkg/ main.go ./
-RUN go mod download
+RUN --mount=type=cache,id=gomod,target=/go/pkg/mod \
+    --mount=type=cache,id=gobuild,target=/root/.cache/go-build \
+    go mod download
 COPY . .
 
 FROM dev AS package

pkg/controllers/chart/chart.go

@@ -214,23 +214,28 @@ func (c *Controller) reconcileJob(_, newObj runtime.Object) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-	// old object is sourced from wrangler applied annotation, not current state.
-	// use cached object to check current conditions.
+	// Old object is sourced from wrangler applied annotation, not current state.
+	// Instead, we use the cached object to check current conditions.
 	// ref: https://github.com/rancher/wrangler/blob/v3.7.0/pkg/apply/reconcilers.go#L136-L137
 	oldJob, err := c.jobCache.Get(newJob.Namespace, newJob.Name)
 	if err != nil {
 		return false, err
 	}
-	hasSuspendedCondition := false
-	// to avoid replacing a job before the job controller has synced it, wait for
-	// job controller to add the Suspended condition before replacing
-	for _, condition := range oldJob.Status.Conditions {
-		if condition.Type == batch.JobSuspended {
-			hasSuspendedCondition = true
-		}
+	// To avoid racing, we want to avoid creating and deleting jobs faster than
+	// the controller can keep up. The addition of at least one condition
+	// indicates that the controller has observed and synced the job at least
+	// once.
+	if len(oldJob.Status.Conditions) == 0 {
+		return false, errors.New("wait for job controller sync before replace")
+	}
+	// To avoid replacing a job while old pods are still running, we wait for
+	// there to be no active or terminating pods before deleting.
+	podCount := oldJob.Status.Active
+	if oldJob.Status.Terminating != nil {
+		podCount += *oldJob.Status.Terminating
 	}
-	if !hasSuspendedCondition {
-		return false, errors.New("wait for job controller to add conditions before replacing")
+	if podCount != 0 {
+		return false, errors.New("wait for pods to terminate before replace")
 	}
 	return false, apply.ErrReplace
 }
@@ -341,8 +346,8 @@ func (c *Controller) OnChange(chart *v1.HelmChart, chartStatus v1.HelmChartStatu
 	// handler will be run again when the job controller updates the job to mark the
 	// job as resumed.
 	if c.jobReady(chart) {
-		if err := c.resumeJob(chart); err != nil {
-			return nil, chartStatus, err
+		if err := c.setJobSuspended(chart, false); err != nil {
+			return nil, chartStatus, fmt.Errorf("failed to resume job: %w", err)
 		}
 		return nil, chartStatus, generic.ErrSkip
 	}
@@ -381,6 +386,11 @@ func (c *Controller) OnChange(chart *v1.HelmChart, chartStatus v1.HelmChartStatu
 		},
 	}
 
+	// Suspend the current job before apply attempts to delete and recreate it.
+	// The job may not exist, or may have already finished, or already be suspend, so
+	// we don't care about whether or not this succeeds.
+	_ = c.setJobSuspended(chart, true)
+
 	// emit an event to indicate that this Helm chart is being applied
 	annotations := map[string]string{KeyConfigHash: job.Spec.Template.ObjectMeta.Annotations[KeyConfigHash]}
 	c.recorder.AnnotatedEventf(chart, annotations, corev1.EventTypeNormal, "ApplyJob", "Applying HelmChart from %s using Job %s/%s ", chartSource(chart), job.Namespace, job.Name)
@@ -402,12 +412,12 @@ func (c *Controller) OnRemove(key string, chart *v1.HelmChart) (*v1.HelmChart, e
 		return nil, nil
 	}
 
-	// If the job is ready (has the Suspended condition), resume the job, and
-	// return ErrSkip. This handler will be run again when the job controller
-	// updates the job to mark the job as resumed.
+	// If the job is ready (has the Suspended condition and has never been
+	// started), resume the job, and return ErrSkip. This handler will be run
+	// again when the job controller updates the job to mark the job as resumed.
 	if c.jobReady(chart) {
-		if err := c.resumeJob(chart); err != nil {
-			return nil, err
+		if err := c.setJobSuspended(chart, false); err != nil {
+			return nil, fmt.Errorf("failed to resume job: %w", err)
 		}
 		return nil, generic.ErrSkip
 	}
@@ -436,6 +446,11 @@ func (c *Controller) OnRemove(key string, chart *v1.HelmChart) (*v1.HelmChart, e
 		return nil, err
 	}
 
+	// Suspend the current job before apply attempts to delete and recreate it.
+	// The job may not exist, or may have already finished, or already be suspend, so
+	// we don't care about whether or not this succeeds.
+	_ = c.setJobSuspended(chart, true)
+
 	c.recorder.Eventf(chart, corev1.EventTypeNormal, "RemoveJob", "Uninstalling HelmChart using Job %s/%s ", job.Namespace, job.Name)
 
 	if chart.Status.JobName != job.Name {
@@ -683,12 +698,16 @@ func (c *Controller) jobComplete(chart *v1.HelmChart) bool {
 	return false
 }
 
-// jobReady returns true if the job is suspended and the Job controller has
-// added a True Suspended condition to the job for the given chart. The
-// addition of this condition indicates that the controller has observed and
-// synced the job at least once.
+// jobReady returns true if the job is suspended, has never been started, and
+// the Job controller has added a True Suspended condition to the job for the
+// given chart. The addition of this condition indicates that the controller
+// has observed and synced the job at least once.
+// We create jobs suspended, and look for generation == 1 to indicate that the
+// job has not potentially previously run; we cannot look at status.startTime as
+// this is cleared when the job is suspended.
 func (c *Controller) jobReady(chart *v1.HelmChart) bool {
-	if job, _ := c.jobs.Cache().Get(chart.Namespace, jobName(chart)); job != nil && job.Spec.Suspend != nil && *job.Spec.Suspend {
+	job, _ := c.jobs.Cache().Get(chart.Namespace, jobName(chart))
+	if job != nil && job.Generation == 1 && job.Spec.Suspend != nil && *job.Spec.Suspend {
 		for _, condition := range job.Status.Conditions {
 			if condition.Type == batch.JobSuspended {
 				return condition.Status == corev1.ConditionTrue
@@ -698,11 +717,22 @@ func (c *Controller) jobReady(chart *v1.HelmChart) bool {
 	return false
 }
 
-// resumeJob patches the job for the given chart to set spec.suspended = false
-func (c *Controller) resumeJob(chart *v1.HelmChart) error {
+// setJobSuspended patches the job for the given chart to set spec.suspend.
+// We use a JSONPatch that tests that the value is not already set to the
+// desired state, so the Patch will return an error if the change is a no-op or
+// the job does not exist, which will prevent spurious events from being
+// emitted.
+func (c *Controller) setJobSuspended(chart *v1.HelmChart, suspend bool) error {
 	name := jobName(chart)
-	c.recorder.Eventf(chart, corev1.EventTypeNormal, "ResumeJob", "Resuming synced Job %s/%s ", chart.Namespace, name)
-	_, err := c.jobs.Patch(chart.Namespace, name, types.MergePatchType, []byte(`{"spec":{"suspend":false}}`))
+	b := fmt.Appendf(nil, `[{"op":"test","path":"/spec/suspend","value":%t},{"op":"replace","path":"/spec/suspend","value":%t}]`, !suspend, suspend)
+	_, err := c.jobs.Patch(chart.Namespace, name, types.JSONPatchType, b)
+	if err == nil {
+		if suspend {
+			c.recorder.Eventf(chart, corev1.EventTypeNormal, "SuspendJob", "Suspended Job %s/%s for delete", chart.Namespace, name)
+		} else {
+			c.recorder.Eventf(chart, corev1.EventTypeNormal, "ResumeJob", "Resumed synced Job %s/%s", chart.Namespace, name)
+		}
+	}
 	return err
 }
 

scripts/e2e

@@ -43,4 +43,4 @@ setup_k8s
 export KUBECONFIG=$PWD/kube_config_cluster.yml
 export HELM_CONTROLLER_IMAGE=$(cat bin/helm-controller-image.txt)
 load_helm_image
-go test -cover -tags=test -v ./test/...
+go test ./test/suite/ -test.v -ginkgo.vv -ginkgo.fail-fast

test/framework/controller.go

@@ -19,26 +19,27 @@ import (
 
 func (f *Framework) setupController(ctx context.Context) error {
 	_, err := f.ClientSet.CoreV1().Namespaces().Create(ctx, f.getNS(), metav1.CreateOptions{})
-	if err != nil {
+	if err != nil && !errors.IsAlreadyExists(err) {
 		return err
 	}
 
 	_, err = f.ClientSet.RbacV1().ClusterRoles().Create(ctx, f.getCr(), metav1.CreateOptions{})
-	if err != nil {
+	if err != nil && !errors.IsAlreadyExists(err) {
 		return err
 	}
 
 	_, err = f.ClientSet.RbacV1().ClusterRoleBindings().Create(ctx, f.getCrb(), metav1.CreateOptions{})
-	if err != nil {
+	if err != nil && !errors.IsAlreadyExists(err) {
 		return err
 	}
 
-	if err := crd.BatchCreateCRDs(ctx, f.ClientExt.ApiextensionsV1().CustomResourceDefinitions(), nil, time.Minute, f.crds); err != nil {
+	err = crd.BatchCreateCRDs(ctx, f.ClientExt.ApiextensionsV1().CustomResourceDefinitions(), nil, time.Minute, f.crds)
+	if err != nil && !errors.IsAlreadyExists(err) {
 		return err
 	}
 
 	_, err = f.ClientSet.CoreV1().ServiceAccounts(f.Namespace).Create(ctx, f.getSa(), metav1.CreateOptions{})
-	if err != nil {
+	if err != nil && !errors.IsAlreadyExists(err) {
 		return err
 	}
 
@@ -59,8 +60,17 @@ func (f *Framework) setupController(ctx context.Context) error {
 		return err
 	}
 
-	_, err = f.ClientSet.AppsV1().Deployments(f.Namespace).Create(context.TODO(), f.getDeployment(), metav1.CreateOptions{})
-	return err
+	if _, err := f.ClientSet.AppsV1().Deployments(f.Namespace).Create(context.TODO(), f.getDeployment(), metav1.CreateOptions{}); err != nil {
+		if errors.IsAlreadyExists(err) {
+			if _, err := f.ClientSet.AppsV1().Deployments(f.Namespace).Update(context.TODO(), f.getDeployment(), metav1.UpdateOptions{}); err != nil {
+				return err
+			}
+		} else {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func (f *Framework) getNS() *corev1.Namespace {

test/framework/framework.go

@@ -13,8 +13,10 @@ import (
 
 	"k8s.io/apimachinery/pkg/util/json"
 	"k8s.io/client-go/util/retry"
+	"k8s.io/utils/ptr"
 
 	v1 "github.com/k3s-io/helm-controller/pkg/apis/helm.cattle.io/v1"
+	chartcontroller "github.com/k3s-io/helm-controller/pkg/controllers/chart"
 	"github.com/k3s-io/helm-controller/pkg/controllers/common"
 	"github.com/k3s-io/helm-controller/pkg/controllers/extjson"
 	helmcrd "github.com/k3s-io/helm-controller/pkg/crds"
@@ -331,6 +333,52 @@ func (f *Framework) ListChartPods(chart *v1.HelmChart, appName string) ([]corev1
 	return podList.Items, nil
 }
 
+func (f *Framework) NewShellJob(chart *v1.HelmChart, command ...string) (*batchv1.Job, error) {
+	if chart.Status.JobName == "" {
+		return nil, errors.New("job name must be populated")
+	}
+	j := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: chart.Namespace,
+			Name:      chart.Status.JobName,
+			Labels: map[string]string{
+				chartcontroller.LabelChartName: chart.Name,
+			},
+		},
+		Spec: batchv1.JobSpec{
+			BackoffLimit: ptr.To(int32(0)),
+			Parallelism:  ptr.To(int32(1)),
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						chartcontroller.LabelChartName: chart.Name,
+					},
+				},
+				Spec: corev1.PodSpec{
+					RestartPolicy:                corev1.RestartPolicyOnFailure,
+					AutomountServiceAccountToken: ptr.To(false),
+					NodeSelector: map[string]string{
+						corev1.LabelOSStable: "linux",
+					},
+					Containers: []corev1.Container{
+						{
+							Name:            "shell",
+							Command:         command,
+							Image:           "docker.io/library/busybox:1.37.0",
+							ImagePullPolicy: corev1.PullIfNotPresent,
+						},
+					},
+				},
+			},
+		},
+	}
+	r, err := f.ClientSet.BatchV1().Jobs(chart.Namespace).Create(context.TODO(), j, metav1.CreateOptions{})
+	if err != nil {
+		return nil, err
+	}
+	return r, nil
+}
+
 func (f *Framework) GetJob(chart *v1.HelmChart) (*batchv1.Job, error) {
 	if chart.Status.JobName == "" {
 		return nil, errors.New("waiting for job name to be populated")