Mirrors/flux2
1
0
Fork 0
mirror of synced 2026-02-06 19:05:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

misc: address wait.PollImmediate deprecation

Browse Source 
Signed-off-by: Hidde Beydals <hidde@hhh.computer>
This commit is contained in:
Hidde Beydals
2023-10-13 10:44:54 +02:00
parent 22134b1233
commit 3fdb292381
18 changed files with 61 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/flux/create.go
View File

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

9
cmd/flux/create_alert.go
View File

@@ -132,8 +132,8 @@ func createAlertCmdRun(cmd *cobra.Command, args []string) error {
} }
logger.Waitingf("waiting for Alert reconciliation") logger.Waitingf("waiting for Alert reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isAlertReady(ctx, kubeClient, namespacedName, &alert)); err != nil { isAlertReady(kubeClient, namespacedName, &alert)); err != nil {
return err return err
} }
logger.Successf("Alert %s is ready", name) logger.Successf("Alert %s is ready", name)
@@ -171,9 +171,8 @@ func upsertAlert(ctx context.Context, kubeClient client.Client,
return namespacedName, nil return namespacedName, nil
} }
func isAlertReady(ctx context.Context, kubeClient client.Client, func isAlertReady(kubeClient client.Client, namespacedName types.NamespacedName, alert *notificationv1b2.Alert) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, alert *notificationv1b2.Alert) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, alert) err := kubeClient.Get(ctx, namespacedName, alert)
if err != nil { if err != nil {
return false, err return false, err

9
cmd/flux/create_alertprovider.go
View File

@@ -127,8 +127,8 @@ func createAlertProviderCmdRun(cmd *cobra.Command, args []string) error {
} }
logger.Waitingf("waiting for Provider reconciliation") logger.Waitingf("waiting for Provider reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isAlertProviderReady(ctx, kubeClient, namespacedName, &provider)); err != nil { isAlertProviderReady(kubeClient, namespacedName, &provider)); err != nil {
return err return err
} }
@@ -168,9 +168,8 @@ func upsertAlertProvider(ctx context.Context, kubeClient client.Client,
return namespacedName, nil return namespacedName, nil
} }
func isAlertProviderReady(ctx context.Context, kubeClient client.Client, func isAlertProviderReady(kubeClient client.Client, namespacedName types.NamespacedName, provider *notificationv1.Provider) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, provider *notificationv1.Provider) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, provider) err := kubeClient.Get(ctx, namespacedName, provider)
if err != nil { if err != nil {
return false, err return false, err

9
cmd/flux/create_helmrelease.go
View File

@@ -303,8 +303,8 @@ func createHelmReleaseCmdRun(cmd *cobra.Command, args []string) error {
} }
logger.Waitingf("waiting for HelmRelease reconciliation") logger.Waitingf("waiting for HelmRelease reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isHelmReleaseReady(ctx, kubeClient, namespacedName, &helmRelease)); err != nil { isHelmReleaseReady(kubeClient, namespacedName, &helmRelease)); err != nil {
return err return err
} }
logger.Successf("HelmRelease %s is ready", name) logger.Successf("HelmRelease %s is ready", name)
@@ -344,9 +344,8 @@ func upsertHelmRelease(ctx context.Context, kubeClient client.Client,
return namespacedName, nil return namespacedName, nil
} }
func isHelmReleaseReady(ctx context.Context, kubeClient client.Client, func isHelmReleaseReady(kubeClient client.Client, namespacedName types.NamespacedName, helmRelease *helmv2.HelmRelease) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, helmRelease *helmv2.HelmRelease) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, helmRelease) err := kubeClient.Get(ctx, namespacedName, helmRelease)
if err != nil { if err != nil {
return false, err return false, err

9
cmd/flux/create_kustomization.go
View File

@@ -263,8 +263,8 @@ func createKsCmdRun(cmd *cobra.Command, args []string) error {
} }
logger.Waitingf("waiting for Kustomization reconciliation") logger.Waitingf("waiting for Kustomization reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isKustomizationReady(ctx, kubeClient, namespacedName, &kustomization)); err != nil { isKustomizationReady(kubeClient, namespacedName, &kustomization)); err != nil {
return err return err
} }
logger.Successf("Kustomization %s is ready", name) logger.Successf("Kustomization %s is ready", name)
@@ -304,9 +304,8 @@ func upsertKustomization(ctx context.Context, kubeClient client.Client,
return namespacedName, nil return namespacedName, nil
} }
func isKustomizationReady(ctx context.Context, kubeClient client.Client, func isKustomizationReady(kubeClient client.Client, namespacedName types.NamespacedName, kustomization *kustomizev1.Kustomization) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, kustomization *kustomizev1.Kustomization) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, kustomization) err := kubeClient.Get(ctx, namespacedName, kustomization)
if err != nil { if err != nil {
return false, err return false, err

9
cmd/flux/create_receiver.go
View File

@@ -139,8 +139,8 @@ func createReceiverCmdRun(cmd *cobra.Command, args []string) error {
} }
logger.Waitingf("waiting for Receiver reconciliation") logger.Waitingf("waiting for Receiver reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isReceiverReady(ctx, kubeClient, namespacedName, &receiver)); err != nil { isReceiverReady(kubeClient, namespacedName, &receiver)); err != nil {
return err return err
} }
logger.Successf("Receiver %s is ready", name) logger.Successf("Receiver %s is ready", name)
@@ -180,9 +180,8 @@ func upsertReceiver(ctx context.Context, kubeClient client.Client,
return namespacedName, nil return namespacedName, nil
} }
func isReceiverReady(ctx context.Context, kubeClient client.Client, func isReceiverReady(kubeClient client.Client, namespacedName types.NamespacedName, receiver *notificationv1.Receiver) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, receiver *notificationv1.Receiver) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, receiver) err := kubeClient.Get(ctx, namespacedName, receiver)
if err != nil { if err != nil {
return false, err return false, err

9
cmd/flux/create_source_bucket.go
View File

@@ -204,8 +204,8 @@ func createSourceBucketCmdRun(cmd *cobra.Command, args []string) error {
} }
logger.Waitingf("waiting for Bucket source reconciliation") logger.Waitingf("waiting for Bucket source reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isBucketReady(ctx, kubeClient, namespacedName, bucket)); err != nil { isBucketReady(kubeClient, namespacedName, bucket)); err != nil {
return err return err
} }
logger.Successf("Bucket source reconciliation completed") logger.Successf("Bucket source reconciliation completed")
@@ -248,9 +248,8 @@ func upsertBucket(ctx context.Context, kubeClient client.Client,
return namespacedName, nil return namespacedName, nil
} }
func isBucketReady(ctx context.Context, kubeClient client.Client, func isBucketReady(kubeClient client.Client, namespacedName types.NamespacedName, bucket *sourcev1.Bucket) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, bucket *sourcev1.Bucket) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, bucket) err := kubeClient.Get(ctx, namespacedName, bucket)
if err != nil { if err != nil {
return false, err return false, err

9
cmd/flux/create_source_git.go
View File

@@ -325,8 +325,8 @@ func createSourceGitCmdRun(cmd *cobra.Command, args []string) error {
} }
logger.Waitingf("waiting for GitRepository source reconciliation") logger.Waitingf("waiting for GitRepository source reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isGitRepositoryReady(ctx, kubeClient, namespacedName, &gitRepository)); err != nil { isGitRepositoryReady(kubeClient, namespacedName, &gitRepository)); err != nil {
return err return err
} }
logger.Successf("GitRepository source reconciliation completed") logger.Successf("GitRepository source reconciliation completed")
@@ -369,9 +369,8 @@ func upsertGitRepository(ctx context.Context, kubeClient client.Client,
return namespacedName, nil return namespacedName, nil
} }
func isGitRepositoryReady(ctx context.Context, kubeClient client.Client, func isGitRepositoryReady(kubeClient client.Client, namespacedName types.NamespacedName, gitRepository *sourcev1.GitRepository) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, gitRepository *sourcev1.GitRepository) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, gitRepository) err := kubeClient.Get(ctx, namespacedName, gitRepository)
if err != nil { if err != nil {
return false, err return false, err

9
cmd/flux/create_source_helm.go
View File

@@ -231,8 +231,8 @@ func createSourceHelmCmdRun(cmd *cobra.Command, args []string) error {
} }
logger.Waitingf("waiting for HelmRepository source reconciliation") logger.Waitingf("waiting for HelmRepository source reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isHelmRepositoryReady(ctx, kubeClient, namespacedName, helmRepository)); err != nil { isHelmRepositoryReady(kubeClient, namespacedName, helmRepository)); err != nil {
return err return err
} }
logger.Successf("HelmRepository source reconciliation completed") logger.Successf("HelmRepository source reconciliation completed")
@@ -280,9 +280,8 @@ func upsertHelmRepository(ctx context.Context, kubeClient client.Client,
return namespacedName, nil return namespacedName, nil
} }
func isHelmRepositoryReady(ctx context.Context, kubeClient client.Client, func isHelmRepositoryReady(kubeClient client.Client, namespacedName types.NamespacedName, helmRepository *sourcev1.HelmRepository) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, helmRepository *sourcev1.HelmRepository) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, helmRepository) err := kubeClient.Get(ctx, namespacedName, helmRepository)
if err != nil { if err != nil {
return false, err return false, err

9
cmd/flux/create_source_oci.go
View File

@@ -192,8 +192,8 @@ func createSourceOCIRepositoryCmdRun(cmd *cobra.Command, args []string) error {
} }
logger.Waitingf("waiting for OCIRepository reconciliation") logger.Waitingf("waiting for OCIRepository reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isOCIRepositoryReady(ctx, kubeClient, namespacedName, repository)); err != nil { isOCIRepositoryReady(kubeClient, namespacedName, repository)); err != nil {
return err return err
} }
logger.Successf("OCIRepository reconciliation completed") logger.Successf("OCIRepository reconciliation completed")
@@ -236,9 +236,8 @@ func upsertOCIRepository(ctx context.Context, kubeClient client.Client,
return namespacedName, nil return namespacedName, nil
} }
func isOCIRepositoryReady(ctx context.Context, kubeClient client.Client, func isOCIRepositoryReady(kubeClient client.Client, namespacedName types.NamespacedName, ociRepository *sourcev1.OCIRepository) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, ociRepository *sourcev1.OCIRepository) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, ociRepository) err := kubeClient.Get(ctx, namespacedName, ociRepository)
if err != nil { if err != nil {
return false, err return false, err

18
cmd/flux/reconcile.go
View File

@@ -113,8 +113,8 @@ func (reconcile reconcileCommand) run(cmd *cobra.Command, args []string) error {
logger.Successf("%s annotated", reconcile.kind) logger.Successf("%s annotated", reconcile.kind)
if reconcile.kind == notificationv1b2.AlertKind || reconcile.kind == notificationv1.ReceiverKind { if reconcile.kind == notificationv1b2.AlertKind || reconcile.kind == notificationv1.ReceiverKind {
if err = wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err = wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isReconcileReady(ctx, kubeClient, namespacedName, reconcile.object)); err != nil { isReconcileReady(kubeClient, namespacedName, reconcile.object)); err != nil {
return err return err
} }
@@ -124,8 +124,8 @@ func (reconcile reconcileCommand) run(cmd *cobra.Command, args []string) error {
lastHandledReconcileAt := reconcile.object.lastHandledReconcileRequest() lastHandledReconcileAt := reconcile.object.lastHandledReconcileRequest()
logger.Waitingf("waiting for %s reconciliation", reconcile.kind) logger.Waitingf("waiting for %s reconciliation", reconcile.kind)
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
reconciliationHandled(ctx, kubeClient, namespacedName, reconcile.object, lastHandledReconcileAt)); err != nil { reconciliationHandled(kubeClient, namespacedName, reconcile.object, lastHandledReconcileAt)); err != nil {
return err return err
} }
readyCond := apimeta.FindStatusCondition(reconcilableConditions(reconcile.object), meta.ReadyCondition) readyCond := apimeta.FindStatusCondition(reconcilableConditions(reconcile.object), meta.ReadyCondition)
@@ -140,9 +140,8 @@ func (reconcile reconcileCommand) run(cmd *cobra.Command, args []string) error {
return nil return nil
} }
func reconciliationHandled(ctx context.Context, kubeClient client.Client, func reconciliationHandled(kubeClient client.Client, namespacedName types.NamespacedName, obj reconcilable, lastHandledReconcileAt string) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, obj reconcilable, lastHandledReconcileAt string) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, obj.asClientObject()) err := kubeClient.Get(ctx, namespacedName, obj.asClientObject())
if err != nil { if err != nil {
return false, err return false, err
@@ -176,9 +175,8 @@ func requestReconciliation(ctx context.Context, kubeClient client.Client,
}) })
} }
func isReconcileReady(ctx context.Context, kubeClient client.Client, func isReconcileReady(kubeClient client.Client, namespacedName types.NamespacedName, obj reconcilable) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, obj reconcilable) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, obj.asClientObject()) err := kubeClient.Get(ctx, namespacedName, obj.asClientObject())
if err != nil { if err != nil {
return false, err return false, err

4
cmd/flux/reconcile_alertprovider.go
View File

@@ -84,8 +84,8 @@ func reconcileAlertProviderCmdRun(cmd *cobra.Command, args []string) error {
logger.Successf("Provider annotated") logger.Successf("Provider annotated")
logger.Waitingf("waiting for reconciliation") logger.Waitingf("waiting for reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isAlertProviderReady(ctx, kubeClient, namespacedName, &alertProvider)); err != nil { isAlertProviderReady(kubeClient, namespacedName, &alertProvider)); err != nil {
return err return err
} }
logger.Successf("Provider reconciliation completed") logger.Successf("Provider reconciliation completed")

4
cmd/flux/reconcile_receiver.go
View File

@@ -88,8 +88,8 @@ func reconcileReceiverCmdRun(cmd *cobra.Command, args []string) error {
logger.Successf("Receiver annotated") logger.Successf("Receiver annotated")
logger.Waitingf("waiting for Receiver reconciliation") logger.Waitingf("waiting for Receiver reconciliation")
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isReceiverReady(ctx, kubeClient, namespacedName, &receiver)); err != nil { isReceiverReady(kubeClient, namespacedName, &receiver)); err != nil {
return err return err
} }

4
cmd/flux/reconcile_with_source.go
View File

@@ -82,8 +82,8 @@ func (reconcile reconcileWithSourceCommand) run(cmd *cobra.Command, args []strin
logger.Successf("%s annotated", reconcile.kind) logger.Successf("%s annotated", reconcile.kind)
logger.Waitingf("waiting for %s reconciliation", reconcile.kind) logger.Waitingf("waiting for %s reconciliation", reconcile.kind)
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
reconciliationHandled(ctx, kubeClient, namespacedName, reconcile.object, lastHandledReconcileAt)); err != nil { reconciliationHandled(kubeClient, namespacedName, reconcile.object, lastHandledReconcileAt)); err != nil {
return err return err
} }

4
cmd/flux/resume.go
View File

@@ -212,8 +212,8 @@ func (resume resumeCommand) reconcile(ctx context.Context, res resumable) reconc
logger.Waitingf("waiting for %s reconciliation", resume.kind) logger.Waitingf("waiting for %s reconciliation", resume.kind)
if err := wait.PollImmediate(rootArgs.pollInterval, rootArgs.timeout, if err := wait.PollUntilContextTimeout(ctx, rootArgs.pollInterval, rootArgs.timeout, true,
isReady(ctx, resume.client, namespacedName, res)); err != nil { isReady(resume.client, namespacedName, res)); err != nil {
return reconcileResponse{ return reconcileResponse{
resumable: res, resumable: res,
err: err, err: err,

5
cmd/flux/status.go
View File

@@ -56,9 +56,8 @@ func statusableConditions(object statusable) []metav1.Condition {
return []metav1.Condition{} return []metav1.Condition{}
} }
func isReady(ctx context.Context, kubeClient client.Client, func isReady(kubeClient client.Client, namespacedName types.NamespacedName, object statusable) wait.ConditionWithContextFunc {
namespacedName types.NamespacedName, object statusable) wait.ConditionFunc { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
err := kubeClient.Get(ctx, namespacedName, object.asClientObject()) err := kubeClient.Get(ctx, namespacedName, object.asClientObject())
if err != nil { if err != nil {
return false, err return false, err

7
pkg/bootstrap/bootstrap.go
View File

@@ -20,6 +20,7 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"k8s.io/apimachinery/pkg/util/wait"
"strings" "strings"
"time" "time"
@@ -172,10 +173,8 @@ func kustomizationPathDiffers(ctx context.Context, kube client.Client, objKey cl
return k.Spec.Path, nil return k.Spec.Path, nil
} }
func kustomizationReconciled(ctx context.Context, kube client.Client, objKey client.ObjectKey, func kustomizationReconciled(kube client.Client, objKey client.ObjectKey, kustomization *kustomizev1.Kustomization, expectRevision string) wait.ConditionWithContextFunc {
kustomization *kustomizev1.Kustomization, expectRevision string) func() (bool, error) { return func(ctx context.Context) (bool, error) {
return func() (bool, error) {
if err := kube.Get(ctx, objKey, kustomization); err != nil { if err := kube.Get(ctx, objKey, kustomization); err != nil {
return false, err return false, err
} }

5
pkg/bootstrap/bootstrap_plain_git.go
View File

@@ -401,9 +401,8 @@ func (b *PlainGitBootstrapper) ReportKustomizationHealth(ctx context.Context, op
expectRevision := fmt.Sprintf("%s@%s", options.Branch, git.Hash(head).Digest()) expectRevision := fmt.Sprintf("%s@%s", options.Branch, git.Hash(head).Digest())
var k kustomizev1.Kustomization var k kustomizev1.Kustomization
if err := wait.PollImmediate(pollInterval, timeout, kustomizationReconciled( if err := wait.PollUntilContextTimeout(ctx, pollInterval, timeout, true,
ctx, b.kube, objKey, &k, expectRevision), kustomizationReconciled(b.kube, objKey, &k, expectRevision)); err != nil {
); err != nil {
b.logger.Failuref(err.Error()) b.logger.Failuref(err.Error())
return err return err
} }