check readiness using kstatus

This change replaces all the many functions and ways of calculating
readiness of objects into one unified way that uses kstatus.Compute() to
check if the object is in progress or current. Only the objects that are
current are considered to be ready. This takes advantage of the kstatus
compatibility of Flux's APIs and also makes sure that they remain
kstatus compatible.

The new isObjectReady() function is also aware of static/statusless
objects and knows how to check their readiness using kstatus. This
prepares the CLI for the upcoming static API objects.

All the is*Ready() functions for specific objects have been removed.

This change doesn't affect any of the existing tests results.

Introduce suspend and resume subcommands for alert-provider.

Signed-off-by: Sunny <darkowlzz@protonmail.com>

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.PollImmediate(rootArgs.pollInterval, rootArgs.timeout,
- isReady(ctx, kubeClient, namespacedName, object)); err != nil {
+ isObjectReadyConditionFunc(ctx, 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.PollImmediate(rootArgs.pollInterval, rootArgs.timeout,
- isAlertReady(ctx, kubeClient, namespacedName, &alert)); err != nil {
+ isObjectReadyConditionFunc(ctx, kubeClient, namespacedName, &alert)); err != nil {
  return err
  }
  logger.Successf("Alert %s is ready", name)
@@ -170,23 +169,3 @@ func upsertAlert(ctx context.Context, kubeClient client.Client,
  logger.Successf("Alert updated")
  return namespacedName, nil
 }
-
-func isAlertReady(ctx context.Context, kubeClient client.Client,
- namespacedName types.NamespacedName, alert *notificationv1b2.Alert) wait.ConditionFunc {
- return func() (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.PollImmediate(rootArgs.pollInterval, rootArgs.timeout,
- isAlertProviderReady(ctx, kubeClient, namespacedName, &provider)); err != nil {
+ isObjectReadyConditionFunc(ctx, kubeClient, namespacedName, &provider)); err != nil {
  return err
  }
 
@@ -167,23 +166,3 @@ func upsertAlertProvider(ctx context.Context, kubeClient client.Client,
  logger.Successf("Provider updated")
  return namespacedName, nil
 }
-
-func isAlertProviderReady(ctx context.Context, kubeClient client.Client,
- namespacedName types.NamespacedName, provider *notificationv1.Provider) wait.ConditionFunc {
- return func() (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.PollImmediate(rootArgs.pollInterval, rootArgs.timeout,
- isHelmReleaseReady(ctx, kubeClient, namespacedName, &helmRelease)); err != nil {
+ isObjectReadyConditionFunc(ctx, kubeClient, namespacedName, &helmRelease)); err != nil {
  return err
  }
  logger.Successf("HelmRelease %s is ready", name)
@@ -344,23 +343,6 @@ func upsertHelmRelease(ctx context.Context, kubeClient client.Client,
  return namespacedName, nil
 }
 
-func isHelmReleaseReady(ctx context.Context, kubeClient client.Client,
- namespacedName types.NamespacedName, helmRelease *helmv2.HelmRelease) wait.ConditionFunc {
- return func() (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.PollImmediate(rootArgs.pollInterval, rootArgs.timeout,
- isKustomizationReady(ctx, kubeClient, namespacedName, &kustomization)); err != nil {
+ isObjectReadyConditionFunc(ctx, kubeClient, namespacedName, &kustomization)); err != nil {
  return err
  }
  logger.Successf("Kustomization %s is ready", name)
@@ -303,28 +302,3 @@ func upsertKustomization(ctx context.Context, kubeClient client.Client,
  logger.Successf("Kustomization updated")
  return namespacedName, nil
 }
-
-func isKustomizationReady(ctx context.Context, kubeClient client.Client,
- namespacedName types.NamespacedName, kustomization *kustomizev1.Kustomization) wait.ConditionFunc {
- return func() (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.PollImmediate(rootArgs.pollInterval, rootArgs.timeout,
- isReceiverReady(ctx, kubeClient, namespacedName, &receiver)); err != nil {
+ isObjectReadyConditionFunc(ctx, kubeClient, namespacedName, &receiver)); err != nil {
  return err
  }
  logger.Successf("Receiver %s is ready", name)
@@ -179,23 +178,3 @@ func upsertReceiver(ctx context.Context, kubeClient client.Client,
  logger.Successf("Receiver updated")
  return namespacedName, nil
 }
-
-func isReceiverReady(ctx context.Context, kubeClient client.Client,
- namespacedName types.NamespacedName, receiver *notificationv1.Receiver) wait.ConditionFunc {
- return func() (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.PollImmediate(rootArgs.pollInterval, rootArgs.timeout,
- isBucketReady(ctx, kubeClient, namespacedName, bucket)); err != nil {
+ isObjectReadyConditionFunc(ctx, kubeClient, namespacedName, bucket)); err != nil {
  return err
  }
  logger.Successf("Bucket source reconciliation completed")
@@ -247,30 +246,3 @@ func upsertBucket(ctx context.Context, kubeClient client.Client,
  logger.Successf("Bucket source updated")
  return namespacedName, nil
 }
-
-func isBucketReady(ctx context.Context, kubeClient client.Client,
- namespacedName types.NamespacedName, bucket *sourcev1.Bucket) wait.ConditionFunc {
- return func() (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.PollImmediate(rootArgs.pollInterval, rootArgs.timeout,
- isGitRepositoryReady(ctx, kubeClient, namespacedName, &gitRepository)); err != nil {
+ isObjectReadyConditionFunc(ctx, kubeClient, namespacedName, &gitRepository)); err != nil {
  return err
  }
  logger.Successf("GitRepository source reconciliation completed")
@@ -368,30 +367,3 @@ func upsertGitRepository(ctx context.Context, kubeClient client.Client,
  logger.Successf("GitRepository source updated")
  return namespacedName, nil
 }
-
-func isGitRepositoryReady(ctx context.Context, kubeClient client.Client,
- namespacedName types.NamespacedName, gitRepository *sourcev1.GitRepository) wait.ConditionFunc {
- return func() (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()
  },
  },
  }