From ebade6b42da0e954127b68265ecf62d9cbe67efd Mon Sep 17 00:00:00 2001 From: Brad Davidson Date: Thu, 16 Jul 2026 19:35:24 +0000 Subject: [PATCH 1/2] Fix old job replacement The controller would never replace jobs created by older versions of helm-controller that hadn't been suspended. We now reenqueue replacement until the job has no running pods, and suspend the job to cleanly stop pods before deleting it. There are also new tests to validate replacement of old failed and successful jobs. Signed-off-by: Brad Davidson (cherry picked from commit cbea780440a2b830dde70146cfab923108bd0d1d) Signed-off-by: Brad Davidson --- pkg/controllers/chart/chart.go | 84 ++++++++++++++------- scripts/e2e | 2 +- test/framework/controller.go | 24 ++++-- test/framework/framework.go | 48 ++++++++++++ test/suite/helm_test.go | 129 +++++++++++++++++++++++++++++++++ 5 files changed, 252 insertions(+), 35 deletions(-) diff --git a/pkg/controllers/chart/chart.go b/pkg/controllers/chart/chart.go index 10ee2471..55d6a203 100644 --- a/pkg/controllers/chart/chart.go +++ b/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 } diff --git a/scripts/e2e b/scripts/e2e index 78a49d4d..be360dc6 100755 --- a/scripts/e2e +++ b/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 diff --git a/test/framework/controller.go b/test/framework/controller.go index fcf2ed95..59b64371 100644 --- a/test/framework/controller.go +++ b/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 { diff --git a/test/framework/framework.go b/test/framework/framework.go index eb07f251..cfc3adfc 100644 --- a/test/framework/framework.go +++ b/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") diff --git a/test/suite/helm_test.go b/test/suite/helm_test.go index b338d644..b18c8347 100644 --- a/test/suite/helm_test.go +++ b/test/suite/helm_test.go @@ -1067,4 +1067,133 @@ var _ = Describe("HelmChart Controller Tests", Ordered, func() { Eventually(framework.ListConfigMapReleases, 120*time.Second, 5*time.Second).WithArguments(chart).Should(HaveLen(0)) }) }) + + Context("When an old active job exists", func() { + var ( + err error + chart *v1.HelmChart + ) + BeforeAll(func() { + chart = &v1.HelmChart{ObjectMeta: metav1.ObjectMeta{Namespace: framework.Namespace, Name: "traefik-example"}, Status: v1.HelmChartStatus{JobName: "helm-install-traefik-example"}} + Expect(framework.NewShellJob(chart, "sh", "-c", "sleep 3600; exit 0")).ToNot(BeNil()) + Eventually(framework.GetJob, 30*time.Second, 5*time.Second).WithArguments(chart).Should(HaveField("Status.Active", BeNumerically("==", 1))) + + chart = framework.NewHelmChart("traefik-example", + "stable/traefik", + "1.86.1", + "v3", + "", + "metrics:\n prometheus:\n enabled: true\nkubernetes:\n ingressEndpoint:\n useDefaultPublishedService: true\nimage: docker.io/rancher/library-traefik\n", + map[string]intstr.IntOrString{ + "rbac.enabled": { + Type: intstr.String, + StrVal: "true", + }, + "ssl.enabled": { + Type: intstr.String, + StrVal: "true", + }, + }) + chart, err = framework.CreateHelmChart(chart, framework.Namespace) + Expect(err).ToNot(HaveOccurred()) + }) + + It("Should create a release for the chart", func() { + Eventually(framework.ListSecretReleases, 120*time.Second, 5*time.Second).WithArguments(chart).Should(HaveLen(1)) + }) + + AfterAll(func() { + err = framework.DeleteHelmChart(chart.Name, chart.Namespace) + Expect(err).ToNot(HaveOccurred()) + + Eventually(getHelmChartIgnoreNotFound, 120*time.Second, 5*time.Second).WithArguments(chart.Name, chart.Namespace).Should(BeNil()) + Eventually(framework.ListSecretReleases, 120*time.Second, 5*time.Second).WithArguments(chart).Should(HaveLen(0)) + }) + }) + + Context("When an old successful job exists", func() { + var ( + err error + chart *v1.HelmChart + ) + BeforeAll(func() { + chart = &v1.HelmChart{ObjectMeta: metav1.ObjectMeta{Namespace: framework.Namespace, Name: "traefik-example"}, Status: v1.HelmChartStatus{JobName: "helm-install-traefik-example"}} + Expect(framework.NewShellJob(chart, "sh", "-c", "sleep 5; exit 0")).ToNot(BeNil()) + Eventually(framework.GetJob, 30*time.Second, 5*time.Second).WithArguments(chart).Should(HaveField("Status.Conditions", ContainElement(HaveField("Type", batchv1.JobComplete)))) + + chart = framework.NewHelmChart("traefik-example", + "stable/traefik", + "1.86.1", + "v3", + "", + "metrics:\n prometheus:\n enabled: true\nkubernetes:\n ingressEndpoint:\n useDefaultPublishedService: true\nimage: docker.io/rancher/library-traefik\n", + map[string]intstr.IntOrString{ + "rbac.enabled": { + Type: intstr.String, + StrVal: "true", + }, + "ssl.enabled": { + Type: intstr.String, + StrVal: "true", + }, + }) + chart, err = framework.CreateHelmChart(chart, framework.Namespace) + Expect(err).ToNot(HaveOccurred()) + }) + + It("Should create a release for the chart", func() { + Eventually(framework.ListSecretReleases, 120*time.Second, 5*time.Second).WithArguments(chart).Should(HaveLen(1)) + }) + + AfterAll(func() { + err = framework.DeleteHelmChart(chart.Name, chart.Namespace) + Expect(err).ToNot(HaveOccurred()) + + Eventually(getHelmChartIgnoreNotFound, 120*time.Second, 5*time.Second).WithArguments(chart.Name, chart.Namespace).Should(BeNil()) + Eventually(framework.ListSecretReleases, 120*time.Second, 5*time.Second).WithArguments(chart).Should(HaveLen(0)) + }) + }) + + Context("When an old failed job exists", func() { + var ( + err error + chart *v1.HelmChart + ) + BeforeAll(func() { + chart = &v1.HelmChart{ObjectMeta: metav1.ObjectMeta{Namespace: framework.Namespace, Name: "traefik-example"}, Status: v1.HelmChartStatus{JobName: "helm-install-traefik-example"}} + Expect(framework.NewShellJob(chart, "sh", "-c", "sleep 5; exit 1")).ToNot(BeNil()) + Eventually(framework.GetJob, 30*time.Second, 5*time.Second).WithArguments(chart).Should(HaveField("Status.Conditions", ContainElement(HaveField("Type", batchv1.JobFailed)))) + + chart = framework.NewHelmChart("traefik-example", + "stable/traefik", + "1.86.1", + "v3", + "", + "metrics:\n prometheus:\n enabled: true\nkubernetes:\n ingressEndpoint:\n useDefaultPublishedService: true\nimage: docker.io/rancher/library-traefik\n", + map[string]intstr.IntOrString{ + "rbac.enabled": { + Type: intstr.String, + StrVal: "true", + }, + "ssl.enabled": { + Type: intstr.String, + StrVal: "true", + }, + }) + chart, err = framework.CreateHelmChart(chart, framework.Namespace) + Expect(err).ToNot(HaveOccurred()) + }) + + It("Should create a release for the chart", func() { + Eventually(framework.ListSecretReleases, 120*time.Second, 5*time.Second).WithArguments(chart).Should(HaveLen(1)) + }) + + AfterAll(func() { + err = framework.DeleteHelmChart(chart.Name, chart.Namespace) + Expect(err).ToNot(HaveOccurred()) + + Eventually(getHelmChartIgnoreNotFound, 120*time.Second, 5*time.Second).WithArguments(chart.Name, chart.Namespace).Should(BeNil()) + Eventually(framework.ListSecretReleases, 120*time.Second, 5*time.Second).WithArguments(chart).Should(HaveLen(0)) + }) + }) }) From e701a5703fbaeffe4868b1a0e779a5b373ba67f3 Mon Sep 17 00:00:00 2001 From: Brad Davidson Date: Thu, 16 Jul 2026 20:17:01 +0000 Subject: [PATCH 2/2] Run all go commands with cache Signed-off-by: Brad Davidson (cherry picked from commit f6dd056340ba75296e1463d78f4deaa254e0039f) Signed-off-by: Brad Davidson --- Dockerfile | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/Dockerfile b/Dockerfile index 013a59f6..ad129b33 100644 --- a/Dockerfile +++ b/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