Changes job handling in gitopts controller

Changes the job handling so jobs are deleted after they succeed.
Jobs are created when they are needed, this avoids having always a minimum of 1 job per
`gitrepo` resource.

It also deletes the Owns(job) statement from the reconciler setup as we're already setting
the controller reference when creating the job and the Owns statement makes extra reconciler
calls that are not needed.

Signed-off-by: Xavi Garcia <[email protected]>

Author: 0xavi0

integrationtests/gitjob/controller/controller_test.go

@@ -21,6 +21,7 @@ import (
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/rancher/wrangler/pkg/name"
 	"github.com/rancher/wrangler/v3/pkg/genericcondition"
 
 	"github.com/rancher/fleet/integrationtests/utils"
@@ -183,7 +184,7 @@ var _ = Describe("GitJob controller", func() {
 							TypeMeta:      metav1.TypeMeta{Kind: "GitRepo"},
 						})
 					g.Expect(events).ToNot(BeNil())
-					g.Expect(len(events.Items)).To(Equal(2))
+					g.Expect(len(events.Items)).To(Equal(3))
 					g.Expect(events.Items[0].Reason).To(Equal("GotNewCommit"))
 					g.Expect(events.Items[0].Message).To(Equal("9ca3a0ad308ed8bffa6602572e2a1343af9c3d2e"))
 					g.Expect(events.Items[0].Type).To(Equal("Normal"))
@@ -192,7 +193,17 @@ var _ = Describe("GitJob controller", func() {
 					g.Expect(events.Items[1].Message).To(Equal("GitJob was created"))
 					g.Expect(events.Items[1].Type).To(Equal("Normal"))
 					g.Expect(events.Items[1].Source.Component).To(Equal("gitjob-controller"))
+					g.Expect(events.Items[2].Reason).To(Equal("JobDeleted"))
+					g.Expect(events.Items[2].Message).To(Equal("job deletion triggered because job succedded"))
+					g.Expect(events.Items[2].Type).To(Equal("Normal"))
+					g.Expect(events.Items[2].Source.Component).To(Equal("gitjob-controller"))
 				}).Should(Succeed())
+
+				// job should not be present
+				Consistently(func() bool {
+					err := k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job)
+					return errors.IsNotFound(err)
+				}, 10*time.Second, 1*time.Second).Should(BeTrue())
 			})
 		})
 
@@ -312,13 +323,6 @@ var _ = Describe("GitJob controller", func() {
 					g.Expect(checkCondition(&gitRepo, "Ready", corev1.ConditionTrue, "")).To(BeTrue())
 					g.Expect(checkCondition(&gitRepo, "Accepted", corev1.ConditionTrue, "")).To(BeTrue())
 				}).Should(Succeed())
-
-				By("verifying that the job is deleted if Spec.Generation changed")
-				Expect(simulateIncreaseGitRepoGeneration(gitRepo)).ToNot(HaveOccurred())
-				Eventually(func() bool {
-					jobName = names.SafeConcatName(gitRepoName, names.Hex(repo+commit, 5))
-					return errors.IsNotFound(k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job))
-				}).Should(BeTrue())
 			})
 		})
 	})
@@ -373,9 +377,51 @@ var _ = Describe("GitJob controller", func() {
 				jobName = names.SafeConcatName(gitRepoName, names.Hex(repo+commit, 5))
 				return k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job)
 			}).Should(Not(HaveOccurred()))
+			// simulate job was successful
+			Eventually(func() error {
+				err := k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job)
+				// We could be checking this when the job is still not created
+				Expect(client.IgnoreNotFound(err)).ToNot(HaveOccurred())
+				job.Status.Succeeded = 1
+				job.Status.Conditions = []batchv1.JobCondition{
+					{
+						Type:   "Complete",
+						Status: "True",
+					},
+				}
+				return k8sClient.Status().Update(ctx, &job)
+			}).Should(Not(HaveOccurred()))
+			// wait until the job has finished
+			Eventually(func() bool {
+				jobName = name.SafeConcatName(gitRepoName, name.Hex(repo+commit, 5))
+				err := k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job)
+				return errors.IsNotFound(err)
+			}).Should(BeTrue())
+
 			// store the generation value to compare against later
 			generationValue = gitRepo.Spec.ForceSyncGeneration
 			Expect(simulateIncreaseForceSyncGeneration(gitRepo)).ToNot(HaveOccurred())
+			// simulate job was successful
+			Eventually(func() error {
+				jobName = name.SafeConcatName(gitRepoName, name.Hex(repo+commit, 5))
+				err := k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job)
+				// We could be checking this when the job is still not created
+				Expect(client.IgnoreNotFound(err)).ToNot(HaveOccurred())
+				job.Status.Succeeded = 1
+				job.Status.Conditions = []batchv1.JobCondition{
+					{
+						Type:   "Complete",
+						Status: "True",
+					},
+				}
+				return k8sClient.Status().Update(ctx, &job)
+			}).Should(Not(HaveOccurred()))
+			// wait until the job has finished
+			Eventually(func() bool {
+				jobName = name.SafeConcatName(gitRepoName, name.Hex(repo+commit, 5))
+				err := k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job)
+				return errors.IsNotFound(err)
+			}).Should(BeTrue())
 		})
 		BeforeEach(func() {
 			expectedCommit = commit
@@ -414,7 +460,7 @@ var _ = Describe("GitJob controller", func() {
 				g.Expect(events.Items[1].Type).To(Equal("Normal"))
 				g.Expect(events.Items[1].Source.Component).To(Equal("gitjob-controller"))
 				g.Expect(events.Items[2].Reason).To(Equal("JobDeleted"))
-				g.Expect(events.Items[2].Message).To(Equal("job deletion triggered because of ForceUpdateGeneration"))
+				g.Expect(events.Items[2].Message).To(Equal("job deletion force-deletion-fa0eb triggered because job succedded"))
 				g.Expect(events.Items[2].Type).To(Equal("Normal"))
 				g.Expect(events.Items[2].Source.Component).To(Equal("gitjob-controller"))
 			}).Should(Succeed())
@@ -446,6 +492,26 @@ var _ = Describe("GitJob controller", func() {
 				jobName = names.SafeConcatName(gitRepoName, names.Hex(repo+commit, 5))
 				return k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job)
 			}).Should(Not(HaveOccurred()))
+			// simulate job was successful
+			Eventually(func() error {
+				err := k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job)
+				// We could be checking this when the job is still not created
+				Expect(client.IgnoreNotFound(err)).ToNot(HaveOccurred())
+				job.Status.Succeeded = 1
+				job.Status.Conditions = []batchv1.JobCondition{
+					{
+						Type:   "Complete",
+						Status: "True",
+					},
+				}
+				return k8sClient.Status().Update(ctx, &job)
+			}).Should(Not(HaveOccurred()))
+			// wait until the job has finished
+			Eventually(func() bool {
+				jobName = name.SafeConcatName(gitRepoName, name.Hex(repo+commit, 5))
+				err := k8sClient.Get(ctx, types.NamespacedName{Name: jobName, Namespace: gitRepoNamespace}, &job)
+				return errors.IsNotFound(err)
+			}).Should(BeTrue())
 
 			// change a gitrepo field, this will change the Generation field. This simulates changing fleet apply parameters.
 			Expect(retry.RetryOnConflict(retry.DefaultRetry, func() error {

internal/cmd/controller/gitops/reconciler/gitjob_controller.go

@@ -102,7 +102,6 @@ func (r *GitJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
 				),
 			),
 		).
-		Owns(&batchv1.Job{}).
 		Watches(
 			// Fan out from bundle to gitrepo
 			&v1alpha1.Bundle{},
@@ -261,7 +260,8 @@ func (r *GitJobReconciler) manageGitJob(ctx context.Context, logger logr.Logger,
 			}
 		}
 
-		if gitrepo.Status.Commit != "" {
+		if r.shouldCreateJob(gitrepo, oldCommit) {
+			r.updateGenerationValuesIfNeeded(gitrepo)
 			if err := r.validateExternalSecretExist(ctx, gitrepo); err != nil {
 				r.Recorder.Event(gitrepo, fleetevent.Warning, "FailedValidatingSecret", err.Error())
 				return result(repoPolled, gitrepo), grutil.UpdateErrorStatus(ctx, r.Client, name, gitrepo.Status, err)
@@ -281,7 +281,7 @@ func (r *GitJobReconciler) manageGitJob(ctx context.Context, logger logr.Logger,
 			}
 			r.Recorder.Event(gitrepo, fleetevent.Normal, "Created", "GitJob was created")
 		}
-	} else if gitrepo.Status.Commit != "" {
+	} else if gitrepo.Status.Commit != "" && gitrepo.Status.Commit == oldCommit {
 		if err = r.deleteJobIfNeeded(ctx, gitrepo, &job); err != nil {
 			return result(repoPolled, gitrepo), fmt.Errorf("error deleting git job: %w", err)
 		}
@@ -330,6 +330,37 @@ func (r *GitJobReconciler) cleanupGitRepo(ctx context.Context, logger logr.Logge
 	return nil
 }
 
+// shouldCreateJob checks if the conditions to create a new job are met.
+// It checks for all the conditions so, in case more than one is met, it sets all the
+// values related in one single reconciler loop
+func (r *GitJobReconciler) shouldCreateJob(gitrepo *v1alpha1.GitRepo, oldCommit string) bool {
+	if gitrepo.Status.Commit != "" && gitrepo.Status.Commit != oldCommit {
+		return true
+	}
+
+	if gitrepo.Spec.ForceSyncGeneration != gitrepo.Status.UpdateGeneration {
+		return true
+	}
+
+	// k8s Jobs are immutable. Recreate the job if the GitRepo Spec has changed.
+	// Avoid deleting the job twice
+	if generationChanged(gitrepo) {
+		return true
+	}
+
+	return false
+}
+
+func (r *GitJobReconciler) updateGenerationValuesIfNeeded(gitrepo *v1alpha1.GitRepo) {
+	if gitrepo.Spec.ForceSyncGeneration != gitrepo.Status.UpdateGeneration {
+		gitrepo.Status.UpdateGeneration = gitrepo.Spec.ForceSyncGeneration
+	}
+
+	if generationChanged(gitrepo) {
+		gitrepo.Status.ObservedGeneration = gitrepo.Generation
+	}
+}
+
 func (r *GitJobReconciler) addGitRepoFinalizer(ctx context.Context, nsName types.NamespacedName) error {
 	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
 		gitrepo := &v1alpha1.GitRepo{}
@@ -444,32 +475,14 @@ func (r *GitJobReconciler) createJob(ctx context.Context, gitRepo *v1alpha1.GitR
 
 func (r *GitJobReconciler) deleteJobIfNeeded(ctx context.Context, gitRepo *v1alpha1.GitRepo, job *batchv1.Job) error {
 	logger := log.FromContext(ctx)
-	jobDeleted := false
-	jobDeletedMessage := ""
-	// if force delete is set, delete the job to make sure a new job is created
-	if gitRepo.Spec.ForceSyncGeneration != gitRepo.Status.UpdateGeneration {
-		gitRepo.Status.UpdateGeneration = gitRepo.Spec.ForceSyncGeneration
-		jobDeletedMessage = "job deletion triggered because of ForceUpdateGeneration"
-		logger.Info(jobDeletedMessage)
-		if err := r.Delete(ctx, job, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil && !errors.IsNotFound(err) {
-			return err
-		}
-		jobDeleted = true
-	}
 
-	// k8s Jobs are immutable. Recreate the job if the GitRepo Spec has changed.
-	// Avoid deleting the job twice
-	if !jobDeleted && generationChanged(gitRepo) {
-		jobDeletedMessage = "job deletion triggered because of generation change"
-		gitRepo.Status.ObservedGeneration = gitRepo.Generation
+	// check if the job finished and was successful
+	if job.Status.Succeeded == 1 {
+		jobDeletedMessage := fmt.Sprintf("job deletion %s triggered because job succedded", job.Name)
 		logger.Info(jobDeletedMessage)
 		if err := r.Delete(ctx, job, client.PropagationPolicy(metav1.DeletePropagationBackground)); err != nil && !errors.IsNotFound(err) {
 			return err
 		}
-		jobDeleted = true
-	}
-
-	if jobDeleted {
 		r.Recorder.Event(gitRepo, fleetevent.Normal, "JobDeleted", jobDeletedMessage)
 	}