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 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)) + }) + }) })