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

fix: wait for Source objects observed generation

Browse Source 
This ensures the command will wait for the object to report a Ready
Condition with an ObservedGeneration matching the Generation of the
resource. Ensuring that when a "create" is actually a mutation, it waits
instead of prematurely assuming the Source to be Ready.

Signed-off-by: Hidde Beydals <hello@hidde.co>
This commit is contained in:
Hidde Beydals
2022-03-23 12:15:31 +01:00
parent b01d3aeecd
commit 3edcd16b62
7 changed files with 77 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
cmd/flux/create_source_bucket.go
View File

@@ -30,6 +30,8 @@ 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"
"github.com/fluxcd/flux2/internal/flags"
@@ -235,3 +237,30 @@ 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
}
}

17
cmd/flux/create_source_git.go
View File

@@ -23,19 +23,21 @@ import (
"net/url"
"os"
"github.com/fluxcd/pkg/apis/meta"
sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
"github.com/manifoldco/promptui"
"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"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/yaml"
"github.com/fluxcd/pkg/apis/meta"
"github.com/fluxcd/pkg/runtime/conditions"
sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
"github.com/fluxcd/flux2/internal/flags"
"github.com/fluxcd/flux2/internal/utils"
"github.com/fluxcd/flux2/pkg/manifestgen/sourcesecret"
@@ -355,7 +357,14 @@ func isGitRepositoryReady(ctx context.Context, kubeClient client.Client,
return false, err
}
if c := apimeta.FindStatusCondition(gitRepository.Status.Conditions, meta.ReadyCondition); c != nil {
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

27
cmd/flux/create_source_git_test.go
View File

@@ -106,10 +106,11 @@ func TestCreateSourceGit(t *testing.T) {
assertGoldenFile("testdata/create_source_git/success.golden"),
func(repo *sourcev1.GitRepository) {
newCondition := metav1.Condition{
Type: meta.ReadyCondition,
Status: metav1.ConditionTrue,
Reason: sourcev1.GitOperationSucceedReason,
Message: "succeeded message",
Type: meta.ReadyCondition,
Status: metav1.ConditionTrue,
Reason: sourcev1.GitOperationSucceedReason,
Message: "succeeded message",
ObservedGeneration: repo.GetGeneration(),
}
apimeta.SetStatusCondition(&repo.Status.Conditions, newCondition)
repo.Status.Artifact = &sourcev1.Artifact{
@@ -123,10 +124,11 @@ func TestCreateSourceGit(t *testing.T) {
assertError("failed message"),
func(repo *sourcev1.GitRepository) {
newCondition := metav1.Condition{
Type: meta.ReadyCondition,
Status: metav1.ConditionFalse,
Reason: sourcev1.URLInvalidReason,
Message: "failed message",
Type: meta.ReadyCondition,
Status: metav1.ConditionFalse,
Reason: sourcev1.URLInvalidReason,
Message: "failed message",
ObservedGeneration: repo.GetGeneration(),
}
apimeta.SetStatusCondition(&repo.Status.Conditions, newCondition)
},
@@ -137,10 +139,11 @@ func TestCreateSourceGit(t *testing.T) {
func(repo *sourcev1.GitRepository) {
// Updated with no artifact
newCondition := metav1.Condition{
Type: meta.ReadyCondition,
Status: metav1.ConditionTrue,
Reason: sourcev1.GitOperationSucceedReason,
Message: "succeeded message",
Type: meta.ReadyCondition,
Status: metav1.ConditionTrue,
Reason: sourcev1.GitOperationSucceedReason,
Message: "succeeded message",
ObservedGeneration: repo.GetGeneration(),
}
apimeta.SetStatusCondition(&repo.Status.Conditions, newCondition)
},

14
cmd/flux/create_source_helm.go
View File

@@ -23,10 +23,10 @@ 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"
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,12 +242,14 @@ func isHelmRepositoryReady(ctx context.Context, kubeClient client.Client,
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 := 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
}
if c := apimeta.FindStatusCondition(helmRepository.Status.Conditions, meta.ReadyCondition); c != nil {
// Further check the Status
switch c.Status {
case metav1.ConditionTrue:
return true, nil

32
cmd/flux/reconcile_source_bucket.go
View File

@@ -17,17 +17,10 @@ limitations under the License.
package main
import (
"context"
"fmt"
"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"
"github.com/fluxcd/pkg/apis/meta"
sourcev1 "github.com/fluxcd/source-controller/api/v1beta2"
)
@@ -48,31 +41,6 @@ func init() {
reconcileSourceCmd.AddCommand(reconcileSourceBucketCmd)
}
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
}
// Confirm the state we are observing is for the current generation
if bucket.Generation != bucket.Status.ObservedGeneration {
return false, nil
}
if c := apimeta.FindStatusCondition(bucket.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
}
}
func (obj bucketAdapter) lastHandledReconcileRequest() string {
return obj.Status.GetLastHandledReconcileRequest()
}