Check readiness of Flux kinds using kstatus

cmd/flux/create.go

@@ -132,7 +132,7 @@ func (names apiType) upsertAndWait(object upsertWaitable, mutate func() error) e
 
 	logger.Waitingf("waiting for %s reconciliation", names.kind)
 	if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
-		isReady(kubeClient, namespacedName, object)); err != nil {
+		isObjectReadyConditionFunc(kubeClient, namespacedName, object.asClientObject())); err != nil {
 		return err
 	}
 	logger.Successf("%s reconciliation completed", names.kind)

cmd/flux/create_alert.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/spf13/cobra"
 	"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"
@@ -133,7 +132,7 @@ func createAlertCmdRun(cmd *cobra.Command, args []string) error {
 
 	logger.Waitingf("waiting for Alert reconciliation")
 	if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
-		isAlertReady(kubeClient, namespacedName, &alert)); err != nil {
+		isObjectReadyConditionFunc(kubeClient, namespacedName, &alert)); err != nil {
 		return err
 	}
 	logger.Successf("Alert %s is ready", name)
@@ -170,22 +169,3 @@ func upsertAlert(ctx context.Context, kubeClient client.Client,
 	logger.Successf("Alert updated")
 	return namespacedName, nil
 }
-
-func isAlertReady(kubeClient client.Client, namespacedName types.NamespacedName, alert *notificationv1b2.Alert) wait.ConditionWithContextFunc {
-	return func(ctx context.Context) (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, alert)
-		if err != nil {
-			return false, err
-		}
-
-		if c := apimeta.FindStatusCondition(alert.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/create_alertprovider.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/spf13/cobra"
 	"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"
@@ -128,7 +127,7 @@ func createAlertProviderCmdRun(cmd *cobra.Command, args []string) error {
 
 	logger.Waitingf("waiting for Provider reconciliation")
 	if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
-		isAlertProviderReady(kubeClient, namespacedName, &provider)); err != nil {
+		isObjectReadyConditionFunc(kubeClient, namespacedName, &provider)); err != nil {
 		return err
 	}
 
@@ -167,22 +166,3 @@ func upsertAlertProvider(ctx context.Context, kubeClient client.Client,
 	logger.Successf("Provider updated")
 	return namespacedName, nil
 }
-
-func isAlertProviderReady(kubeClient client.Client, namespacedName types.NamespacedName, provider *notificationv1.Provider) wait.ConditionWithContextFunc {
-	return func(ctx context.Context) (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, provider)
-		if err != nil {
-			return false, err
-		}
-
-		if c := apimeta.FindStatusCondition(provider.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/create_helmrelease.go

@@ -32,7 +32,6 @@ import (
 	"github.com/spf13/cobra"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/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"
@@ -304,7 +303,7 @@ func createHelmReleaseCmdRun(cmd *cobra.Command, args []string) error {
 
 	logger.Waitingf("waiting for HelmRelease reconciliation")
 	if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
-		isHelmReleaseReady(kubeClient, namespacedName, &helmRelease)); err != nil {
+		isObjectReadyConditionFunc(kubeClient, namespacedName, &helmRelease)); err != nil {
 		return err
 	}
 	logger.Successf("HelmRelease %s is ready", name)
@@ -344,22 +343,6 @@ func upsertHelmRelease(ctx context.Context, kubeClient client.Client,
 	return namespacedName, nil
 }
 
-func isHelmReleaseReady(kubeClient client.Client, namespacedName types.NamespacedName, helmRelease *helmv2.HelmRelease) wait.ConditionWithContextFunc {
-	return func(ctx context.Context) (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, helmRelease)
-		if err != nil {
-			return false, err
-		}
-
-		// Confirm the state we are observing is for the current generation
-		if helmRelease.Generation != helmRelease.Status.ObservedGeneration {
-			return false, nil
-		}
-
-		return apimeta.IsStatusConditionTrue(helmRelease.Status.Conditions, meta.ReadyCondition), nil
-	}
-}
-
 func validateStrategy(input string) bool {
 	allowedStrategy := []string{"Revision", "ChartVersion"}
 

cmd/flux/create_kustomization.go

@@ -24,7 +24,6 @@ import (
 
 	"github.com/spf13/cobra"
 	"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"
@@ -264,7 +263,7 @@ func createKsCmdRun(cmd *cobra.Command, args []string) error {
 
 	logger.Waitingf("waiting for Kustomization reconciliation")
 	if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
-		isKustomizationReady(kubeClient, namespacedName, &kustomization)); err != nil {
+		isObjectReadyConditionFunc(kubeClient, namespacedName, &kustomization)); err != nil {
 		return err
 	}
 	logger.Successf("Kustomization %s is ready", name)
@@ -303,27 +302,3 @@ func upsertKustomization(ctx context.Context, kubeClient client.Client,
 	logger.Successf("Kustomization updated")
 	return namespacedName, nil
 }
-
-func isKustomizationReady(kubeClient client.Client, namespacedName types.NamespacedName, kustomization *kustomizev1.Kustomization) wait.ConditionWithContextFunc {
-	return func(ctx context.Context) (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, kustomization)
-		if err != nil {
-			return false, err
-		}
-
-		// Confirm the state we are observing is for the current generation
-		if kustomization.Generation != kustomization.Status.ObservedGeneration {
-			return false, nil
-		}
-
-		if c := apimeta.FindStatusCondition(kustomization.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/create_receiver.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/spf13/cobra"
 	"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"
@@ -140,7 +139,7 @@ func createReceiverCmdRun(cmd *cobra.Command, args []string) error {
 
 	logger.Waitingf("waiting for Receiver reconciliation")
 	if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
-		isReceiverReady(kubeClient, namespacedName, &receiver)); err != nil {
+		isObjectReadyConditionFunc(kubeClient, namespacedName, &receiver)); err != nil {
 		return err
 	}
 	logger.Successf("Receiver %s is ready", name)
@@ -179,22 +178,3 @@ func upsertReceiver(ctx context.Context, kubeClient client.Client,
 	logger.Successf("Receiver updated")
 	return namespacedName, nil
 }
-
-func isReceiverReady(kubeClient client.Client, namespacedName types.NamespacedName, receiver *notificationv1.Receiver) wait.ConditionWithContextFunc {
-	return func(ctx context.Context) (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, receiver)
-		if err != nil {
-			return false, err
-		}
-
-		if c := apimeta.FindStatusCondition(receiver.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/create_source_bucket.go

@@ -31,7 +31,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/fluxcd/pkg/apis/meta"
-	"github.com/fluxcd/pkg/runtime/conditions"
 
 	sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
 
@@ -205,7 +204,7 @@ func createSourceBucketCmdRun(cmd *cobra.Command, args []string) error {
 
 	logger.Waitingf("waiting for Bucket source reconciliation")
 	if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
-		isBucketReady(kubeClient, namespacedName, bucket)); err != nil {
+		isObjectReadyConditionFunc(kubeClient, namespacedName, bucket)); err != nil {
 		return err
 	}
 	logger.Successf("Bucket source reconciliation completed")
@@ -247,29 +246,3 @@ func upsertBucket(ctx context.Context, kubeClient client.Client,
 	logger.Successf("Bucket source updated")
 	return namespacedName, nil
 }
-
-func isBucketReady(kubeClient client.Client, namespacedName types.NamespacedName, bucket *sourcev1.Bucket) wait.ConditionWithContextFunc {
-	return func(ctx context.Context) (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, bucket)
-		if err != nil {
-			return false, err
-		}
-
-		if c := conditions.Get(bucket, meta.ReadyCondition); c != nil {
-			// Confirm the Ready condition we are observing is for the
-			// current generation
-			if c.ObservedGeneration != bucket.GetGeneration() {
-				return false, nil
-			}
-
-			// Further check the Status
-			switch c.Status {
-			case metav1.ConditionTrue:
-				return true, nil
-			case metav1.ConditionFalse:
-				return false, fmt.Errorf(c.Message)
-			}
-		}
-		return false, nil
-	}
-}

cmd/flux/create_source_git.go

@@ -35,7 +35,6 @@ import (
 	"sigs.k8s.io/yaml"
 
 	"github.com/fluxcd/pkg/apis/meta"
-	"github.com/fluxcd/pkg/runtime/conditions"
 
 	sourcev1 "github.com/fluxcd/source-controller/api/v1"
 
@@ -326,7 +325,7 @@ func createSourceGitCmdRun(cmd *cobra.Command, args []string) error {
 
 	logger.Waitingf("waiting for GitRepository source reconciliation")
 	if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
-		isGitRepositoryReady(kubeClient, namespacedName, &gitRepository)); err != nil {
+		isObjectReadyConditionFunc(kubeClient, namespacedName, &gitRepository)); err != nil {
 		return err
 	}
 	logger.Successf("GitRepository source reconciliation completed")
@@ -368,29 +367,3 @@ func upsertGitRepository(ctx context.Context, kubeClient client.Client,
 	logger.Successf("GitRepository source updated")
 	return namespacedName, nil
 }
-
-func isGitRepositoryReady(kubeClient client.Client, namespacedName types.NamespacedName, gitRepository *sourcev1.GitRepository) wait.ConditionWithContextFunc {
-	return func(ctx context.Context) (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, gitRepository)
-		if err != nil {
-			return false, err
-		}
-
-		if c := conditions.Get(gitRepository, meta.ReadyCondition); c != nil {
-			// Confirm the Ready condition we are observing is for the
-			// current generation
-			if c.ObservedGeneration != gitRepository.GetGeneration() {
-				return false, nil
-			}
-
-			// Further check the Status
-			switch c.Status {
-			case metav1.ConditionTrue:
-				return true, nil
-			case metav1.ConditionFalse:
-				return false, fmt.Errorf(c.Message)
-			}
-		}
-		return false, nil
-	}
-}

cmd/flux/create_source_git_test.go

@@ -181,12 +181,21 @@ func TestCreateSourceGit(t *testing.T) {
 						Time: time.Now(),
 					},
 				}
+				repo.Status.ObservedGeneration = repo.GetGeneration()
 			},
 		}, {
 			"Failed",
 			command,
 			assertError("failed message"),
 			func(repo *sourcev1.GitRepository) {
+				stalledCondition := metav1.Condition{
+					Type:               meta.StalledCondition,
+					Status:             metav1.ConditionTrue,
+					Reason:             sourcev1.URLInvalidReason,
+					Message:            "failed message",
+					ObservedGeneration: repo.GetGeneration(),
+				}
+				apimeta.SetStatusCondition(&repo.Status.Conditions, stalledCondition)
 				newCondition := metav1.Condition{
 					Type:               meta.ReadyCondition,
 					Status:             metav1.ConditionFalse,
@@ -195,6 +204,7 @@ func TestCreateSourceGit(t *testing.T) {
 					ObservedGeneration: repo.GetGeneration(),
 				}
 				apimeta.SetStatusCondition(&repo.Status.Conditions, newCondition)
+				repo.Status.ObservedGeneration = repo.GetGeneration()
 			},
 		}, {
 			"NoArtifact",
@@ -210,6 +220,7 @@ func TestCreateSourceGit(t *testing.T) {
 					ObservedGeneration: repo.GetGeneration(),
 				}
 				apimeta.SetStatusCondition(&repo.Status.Conditions, newCondition)
+				repo.Status.ObservedGeneration = repo.GetGeneration()
 			},
 		},
 	}

cmd/flux/create_source_helm.go

@@ -23,7 +23,6 @@ import (
 	"os"
 
 	"github.com/fluxcd/pkg/apis/meta"
-	"github.com/fluxcd/pkg/runtime/conditions"
 	"github.com/spf13/cobra"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -232,7 +231,7 @@ func createSourceHelmCmdRun(cmd *cobra.Command, args []string) error {
 
 	logger.Waitingf("waiting for HelmRepository source reconciliation")
 	if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
-		isHelmRepositoryReady(kubeClient, namespacedName, helmRepository)); err != nil {
+		isObjectReadyConditionFunc(kubeClient, namespacedName, helmRepository)); err != nil {
 		return err
 	}
 	logger.Successf("HelmRepository source reconciliation completed")
@@ -279,29 +278,3 @@ func upsertHelmRepository(ctx context.Context, kubeClient client.Client,
 	logger.Successf("source updated")
 	return namespacedName, nil
 }
-
-func isHelmRepositoryReady(kubeClient client.Client, namespacedName types.NamespacedName, helmRepository *sourcev1.HelmRepository) wait.ConditionWithContextFunc {
-	return func(ctx context.Context) (bool, error) {
-		err := kubeClient.Get(ctx, namespacedName, helmRepository)
-		if err != nil {
-			return false, err
-		}
-
-		if c := conditions.Get(helmRepository, meta.ReadyCondition); c != nil {
-			// Confirm the Ready condition we are observing is for the
-			// current generation
-			if c.ObservedGeneration != helmRepository.GetGeneration() {
-				return false, nil
-			}
-
-			// Further check the Status
-			switch c.Status {
-			case metav1.ConditionTrue:
-				return true, nil
-			case metav1.ConditionFalse:
-				return false, fmt.Errorf(c.Message)
-			}
-		}
-		return false, nil
-	}
-}