Merge pull request #484 from fluxcd/source-reconcile-at

Utilize LastHandledReconcileAt for source reconcile commands

cmd/flux/create_source_helm.go

@@ -23,9 +23,11 @@ import (
 	"net/url"
 	"os"
 
+	"github.com/fluxcd/pkg/apis/meta"
 	"github.com/spf13/cobra"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -242,3 +244,28 @@ func upsertHelmRepository(ctx context.Context, kubeClient client.Client,
 	logger.Successf("source updated")
 	return namespacedName, nil
 }
+
+func isHelmRepositoryReady(ctx context.Context, kubeClient client.Client,
+	namespacedName types.NamespacedName, helmRepository *sourcev1.HelmRepository) wait.ConditionFunc {
+	return func() (bool, error) {
+		err := kubeClient.Get(ctx, namespacedName, helmRepository)
+		if err != nil {
+			return false, err
+		}
+
+		// Confirm the state we are observing is for the current generation
+		if helmRepository.Generation != helmRepository.Status.ObservedGeneration {
+			return false, nil
+		}
+
+		if c := apimeta.FindStatusCondition(helmRepository.Status.Conditions, meta.ReadyCondition); c != nil {
+			switch c.Status {
+			case metav1.ConditionTrue:
+				return true, nil
+			case metav1.ConditionFalse:
+				return false, fmt.Errorf(c.Message)
+			}
+		}
+		return false, nil
+	}
+}

cmd/flux/reconcile_source_bucket.go

@@ -21,8 +21,10 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/fluxcd/flux2/internal/utils"
 	"github.com/fluxcd/pkg/apis/meta"
+	"k8s.io/client-go/util/retry"
+
+	"github.com/fluxcd/flux2/internal/utils"
 
 	"github.com/spf13/cobra"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
@@ -66,35 +68,30 @@ func reconcileSourceBucketCmdRun(cmd *cobra.Command, args []string) error {
 		Namespace: namespace,
 		Name:      name,
 	}
-
-	logger.Actionf("annotating Bucket source %s in %s namespace", name, namespace)
 	var bucket sourcev1.Bucket
 	err = kubeClient.Get(ctx, namespacedName, &bucket)
 	if err != nil {
 		return err
 	}
 
-	if bucket.Annotations == nil {
+	lastHandledReconcileAt := bucket.Status.LastHandledReconcileAt
-		bucket.Annotations = map[string]string{
+	logger.Actionf("annotating Bucket source %s in %s namespace", name, namespace)
-			meta.ReconcileAtAnnotation: time.Now().Format(time.RFC3339Nano),
+	if err := requestBucketReconciliation(ctx, kubeClient, namespacedName, &bucket); err != nil {
-		}
-	} else {
-		bucket.Annotations[meta.ReconcileAtAnnotation] = time.Now().Format(time.RFC3339Nano)
-	}
-	if err := kubeClient.Update(ctx, &bucket); err != nil {
 		return err
 	}
 	logger.Successf("Bucket source annotated")
 
 	logger.Waitingf("waiting for Bucket source reconciliation")
-	if err := wait.PollImmediate(pollInterval, timeout,
+	if err := wait.PollImmediate(
-		isBucketReady(ctx, kubeClient, namespacedName, &bucket)); err != nil {
+		pollInterval, timeout,
+		bucketReconciliationHandled(ctx, kubeClient, namespacedName, &bucket, lastHandledReconcileAt),
+	); err != nil {
 		return err
 	}
 	logger.Successf("Bucket source reconciliation completed")
 
-	if bucket.Status.Artifact == nil {
+	if apimeta.IsStatusConditionFalse(bucket.Status.Conditions, meta.ReadyCondition) {
-		return fmt.Errorf("Bucket source reconciliation completed but no artifact was found")
+		return fmt.Errorf("Bucket source reconciliation failed")
 	}
 	logger.Successf("fetched revision %s", bucket.Status.Artifact.Revision)
 	return nil
@@ -124,3 +121,31 @@ func isBucketReady(ctx context.Context, kubeClient client.Client,
 		return false, nil
 	}
 }
+
+func bucketReconciliationHandled(ctx context.Context, kubeClient client.Client,
+	namespacedName types.NamespacedName, bucket *sourcev1.Bucket, lastHandledReconcileAt string) wait.ConditionFunc {
+	return func() (bool, error) {
+		err := kubeClient.Get(ctx, namespacedName, bucket)
+		if err != nil {
+			return false, err
+		}
+		return bucket.Status.LastHandledReconcileAt != lastHandledReconcileAt, nil
+	}
+}
+
+func requestBucketReconciliation(ctx context.Context, kubeClient client.Client,
+	namespacedName types.NamespacedName, bucket *sourcev1.Bucket) error {
+	return retry.RetryOnConflict(retry.DefaultBackoff, func() (err error) {
+		if err := kubeClient.Get(ctx, namespacedName, bucket); err != nil {
+			return err
+		}
+		if bucket.Annotations == nil {
+			bucket.Annotations = map[string]string{
+				meta.ReconcileAtAnnotation: time.Now().Format(time.RFC3339Nano),
+			}
+		} else {
+			bucket.Annotations[meta.ReconcileAtAnnotation] = time.Now().Format(time.RFC3339Nano)
+		}
+		return kubeClient.Update(ctx, bucket)
+	})
+}

cmd/flux/reconcile_source_git.go

@@ -23,6 +23,9 @@ import (
 
 	"github.com/fluxcd/flux2/internal/utils"
 	"github.com/fluxcd/pkg/apis/meta"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/client-go/util/retry"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/spf13/cobra"
 	"k8s.io/apimachinery/pkg/types"
@@ -63,36 +66,57 @@ func reconcileSourceGitCmdRun(cmd *cobra.Command, args []string) error {
 		Namespace: namespace,
 		Name:      name,
 	}
-
+	var repository sourcev1.GitRepository
-	logger.Actionf("annotating GitRepository source %s in %s namespace", name, namespace)
+	err = kubeClient.Get(ctx, namespacedName, &repository)
-	var gitRepository sourcev1.GitRepository
-	err = kubeClient.Get(ctx, namespacedName, &gitRepository)
 	if err != nil {
 		return err
 	}
 
-	if gitRepository.Annotations == nil {
+	logger.Actionf("annotating GitRepository source %s in %s namespace", name, namespace)
-		gitRepository.Annotations = map[string]string{
+	if err := requestGitRepositoryReconciliation(ctx, kubeClient, namespacedName, &repository); err != nil {
-			meta.ReconcileAtAnnotation: time.Now().Format(time.RFC3339Nano),
-		}
-	} else {
-		gitRepository.Annotations[meta.ReconcileAtAnnotation] = time.Now().Format(time.RFC3339Nano)
-	}
-	if err := kubeClient.Update(ctx, &gitRepository); err != nil {
 		return err
 	}
 	logger.Successf("GitRepository source annotated")
 
+	lastHandledReconcileAt := repository.Status.LastHandledReconcileAt
 	logger.Waitingf("waiting for GitRepository source reconciliation")
 	if err := wait.PollImmediate(pollInterval, timeout,
-		isGitRepositoryReady(ctx, kubeClient, namespacedName, &gitRepository)); err != nil {
+		gitRepositoryReconciliationHandled(ctx, kubeClient, namespacedName, &repository, lastHandledReconcileAt)); err != nil {
 		return err
 	}
 	logger.Successf("GitRepository source reconciliation completed")
 
-	if gitRepository.Status.Artifact == nil {
+	if apimeta.IsStatusConditionFalse(repository.Status.Conditions, meta.ReadyCondition) {
-		return fmt.Errorf("GitRepository source reconciliation completed but no artifact was found")
+		return fmt.Errorf("GitRepository source reconciliation failed")
 	}
-	logger.Successf("fetched revision %s", gitRepository.Status.Artifact.Revision)
+	logger.Successf("fetched revision %s", repository.Status.Artifact.Revision)
 	return nil
 }
+
+func gitRepositoryReconciliationHandled(ctx context.Context, kubeClient client.Client,
+	namespacedName types.NamespacedName, repository *sourcev1.GitRepository, lastHandledReconcileAt string) wait.ConditionFunc {
+	return func() (bool, error) {
+		err := kubeClient.Get(ctx, namespacedName, repository)
+		if err != nil {
+			return false, err
+		}
+		return repository.Status.LastHandledReconcileAt != lastHandledReconcileAt, nil
+	}
+}
+
+func requestGitRepositoryReconciliation(ctx context.Context, kubeClient client.Client,
+	namespacedName types.NamespacedName, repository *sourcev1.GitRepository) error {
+	return retry.RetryOnConflict(retry.DefaultBackoff, func() (err error) {
+		if err := kubeClient.Get(ctx, namespacedName, repository); err != nil {
+			return err
+		}
+		if repository.Annotations == nil {
+			repository.Annotations = map[string]string{
+				meta.ReconcileAtAnnotation: time.Now().Format(time.RFC3339Nano),
+			}
+		} else {
+			repository.Annotations[meta.ReconcileAtAnnotation] = time.Now().Format(time.RFC3339Nano)
+		}
+		return kubeClient.Update(ctx, repository)
+	})
+}

cmd/flux/reconcile_source_helm.go

@@ -21,12 +21,13 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/fluxcd/flux2/internal/utils"
 	"github.com/fluxcd/pkg/apis/meta"
+	"k8s.io/client-go/util/retry"
+
+	"github.com/fluxcd/flux2/internal/utils"
 
 	"github.com/spf13/cobra"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -66,61 +67,57 @@ func reconcileSourceHelmCmdRun(cmd *cobra.Command, args []string) error {
 		Namespace: namespace,
 		Name:      name,
 	}
-
+	var repository sourcev1.HelmRepository
-	logger.Actionf("annotating HelmRepository source %s in %s namespace", name, namespace)
+	err = kubeClient.Get(ctx, namespacedName, &repository)
-	var helmRepository sourcev1.HelmRepository
-	err = kubeClient.Get(ctx, namespacedName, &helmRepository)
 	if err != nil {
 		return err
 	}
 
-	if helmRepository.Annotations == nil {
+	logger.Actionf("annotating HelmRepository source %s in %s namespace", name, namespace)
-		helmRepository.Annotations = map[string]string{
+	if err := requestHelmRepositoryReconciliation(ctx, kubeClient, namespacedName, &repository); err != nil {
-			meta.ReconcileAtAnnotation: time.Now().Format(time.RFC3339Nano),
-		}
-	} else {
-		helmRepository.Annotations[meta.ReconcileAtAnnotation] = time.Now().Format(time.RFC3339Nano)
-	}
-	if err := kubeClient.Update(ctx, &helmRepository); err != nil {
 		return err
 	}
 	logger.Successf("HelmRepository source annotated")
 
+	lastHandledReconcileAt := repository.Status.LastHandledReconcileAt
 	logger.Waitingf("waiting for HelmRepository source reconciliation")
 	if err := wait.PollImmediate(pollInterval, timeout,
-		isHelmRepositoryReady(ctx, kubeClient, namespacedName, &helmRepository)); err != nil {
+		helmRepositoryReconciliationHandled(ctx, kubeClient, namespacedName, &repository, lastHandledReconcileAt)); err != nil {
 		return err
 	}
 	logger.Successf("HelmRepository source reconciliation completed")
 
-	if helmRepository.Status.Artifact == nil {
+	if apimeta.IsStatusConditionFalse(repository.Status.Conditions, meta.ReadyCondition) {
-		return fmt.Errorf("HelmRepository source reconciliation completed but no artifact was found")
+		return fmt.Errorf("HelmRepository source reconciliation failed")
 	}
-	logger.Successf("fetched revision %s", helmRepository.Status.Artifact.Revision)
+	logger.Successf("fetched revision %s", repository.Status.Artifact.Revision)
 	return nil
 }
 
-func isHelmRepositoryReady(ctx context.Context, kubeClient client.Client,
+func helmRepositoryReconciliationHandled(ctx context.Context, kubeClient client.Client,
-	namespacedName types.NamespacedName, helmRepository *sourcev1.HelmRepository) wait.ConditionFunc {
+	namespacedName types.NamespacedName, repository *sourcev1.HelmRepository, lastHandledReconcileAt string) wait.ConditionFunc {
 	return func() (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, helmRepository)
+		err := kubeClient.Get(ctx, namespacedName, repository)
 		if err != nil {
 			return false, err
 		}
-
+		return repository.Status.LastHandledReconcileAt != lastHandledReconcileAt, nil
-		// Confirm the state we are observing is for the current generation
+	}
-		if helmRepository.Generation != helmRepository.Status.ObservedGeneration {
-			return false, nil
 }
 
-		if c := apimeta.FindStatusCondition(helmRepository.Status.Conditions, meta.ReadyCondition); c != nil {
+func requestHelmRepositoryReconciliation(ctx context.Context, kubeClient client.Client,
-			switch c.Status {
+	namespacedName types.NamespacedName, repository *sourcev1.HelmRepository) error {
-			case metav1.ConditionTrue:
+	return retry.RetryOnConflict(retry.DefaultBackoff, func() (err error) {
-				return true, nil
+		if err := kubeClient.Get(ctx, namespacedName, repository); err != nil {
-			case metav1.ConditionFalse:
+			return err
-				return false, fmt.Errorf(c.Message)
 		}
+		if repository.Annotations == nil {
+			repository.Annotations = map[string]string{
+				meta.ReconcileAtAnnotation: time.Now().Format(time.RFC3339Nano),
 			}
-		return false, nil
+		} else {
+			repository.Annotations[meta.ReconcileAtAnnotation] = time.Now().Format(time.RFC3339Nano)
 		}
+		return kubeClient.Update(ctx, repository)
+	})
 }