Skip to content

🐛 e2e: only retry creating objects that failed #13265

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nojnhuh wants to merge 1 commit into
base: main
Choose a base branch
from
+137 −12
Open

🐛 e2e: only retry creating objects that failed #13265

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion test/e2e/scale.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -689,7 +689,7 @@ func getClusterCreateAndWaitFn(input clusterctl.ApplyCustomClusterTemplateAndWai
WaitForClusterIntervals: input.WaitForClusterIntervals,
WaitForControlPlaneIntervals: input.WaitForControlPlaneIntervals,
WaitForMachineDeployments: input.WaitForMachineDeployments,
CreateOrUpdateOpts: input.CreateOrUpdateOpts,
CreateOpts: input.CreateOpts,
PreWaitForCluster: input.PreWaitForCluster,
PostMachinesProvisioned: input.PostMachinesProvisioned,
ControlPlaneWaiters: input.ControlPlaneWaiters,
Expand Down
132 changes: 127 additions & 5 deletions test/framework/cluster_proxy.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,10 @@ type ClusterProxy interface {
// GetLogCollector returns the machine log collector for the Kubernetes cluster.
GetLogCollector() ClusterLogCollector

// Create creates objects using the clusterProxy client.
// It will return an error if any object already exists.
Create(ctx context.Context, resources []byte, options ...CreateOption) error

// CreateOrUpdate creates or updates objects using the clusterProxy client
CreateOrUpdate(ctx context.Context, resources []byte, options ...CreateOrUpdateOption) error

Expand All @@ -105,11 +109,42 @@ type ClusterProxy interface {
Dispose(context.Context)
}

// createConfig contains options for use with Create.
type createConfig struct {
labelSelector labels.Selector
createOpts []client.CreateOption
pollTimeout, pollInterval time.Duration
}

// CreateOption is a configuration option supplied to Create.
type CreateOption func(*createConfig)

// CreateWithLabelSelector allows definition of the LabelSelector to be used in Create.
func CreateWithLabelSelector(labelSelector labels.Selector) CreateOption {
return func(c *createConfig) {
c.labelSelector = labelSelector
}
}

// CreateWithCreateOpts allows definition of the Create options to be used in resource Create.
func CreateWithCreateOpts(createOpts ...client.CreateOption) CreateOption {
return func(c *createConfig) {
c.createOpts = createOpts
}
}

// CreateWithPolling enables retries over the specified interval.
func CreateWithPolling(pollTimeout, pollInterval time.Duration) CreateOption {
return func(c *createConfig) {
c.pollTimeout = pollTimeout
c.pollInterval = pollInterval
}
}

// createOrUpdateConfig contains options for use with CreateOrUpdate.
type createOrUpdateConfig struct {
labelSelector labels.Selector
createOpts []client.CreateOption
updateOpts []client.UpdateOption
createConfig
updateOpts []client.UpdateOption
}

// CreateOrUpdateOption is a configuration option supplied to CreateOrUpdate.
Expand All @@ -136,6 +171,14 @@ func WithUpdateOpts(updateOpts ...client.UpdateOption) CreateOrUpdateOption {
}
}

// WithPolling enables retries over the specified interval.
func WithPolling(pollTimeout, pollInterval time.Duration) CreateOrUpdateOption {
return func(c *createOrUpdateConfig) {
c.pollTimeout = pollTimeout
c.pollInterval = pollInterval
}
}

// ClusterLogCollector defines an object that can collect logs from a machine.
type ClusterLogCollector interface {
// CollectMachineLog collects log from a machine.
Expand Down Expand Up @@ -306,6 +349,56 @@ func (p *clusterProxy) GetCache(ctx context.Context) cache.Cache {
return p.cache
}

// Create creates objects using the clusterProxy client.
// It will return an error if any object already exists.
// Defaults to use FieldValidation: strict, which can be overwritten with CreateOptions.
func (p *clusterProxy) Create(ctx context.Context, resources []byte, opts ...CreateOption) error {
Expect(ctx).NotTo(BeNil(), "ctx is required for CreateOrUpdate")
Expect(resources).NotTo(BeNil(), "resources is required for CreateOrUpdate")
labelSelector := labels.Everything()
config := &createConfig{}
for _, opt := range opts {
opt(config)
}
if config.labelSelector != nil {
labelSelector = config.labelSelector
}
// Prepending field validation strict so that it is used per default, but can still be overwritten.
config.createOpts = append([]client.CreateOption{client.FieldValidation("Strict")}, config.createOpts...)
objs, err := yaml.ToUnstructured(resources)
if err != nil {
return err
}

retryDisabled := config.pollTimeout == 0 && config.pollInterval == 0
var retErrs []error
for _, o := range objs {
labels := labels.Set(o.GetLabels())
if labelSelector.Matches(labels) {
var err error
if retryDisabled {
err = p.GetClient().Create(ctx, &o, config.createOpts...)
} else {
err = wait.PollUntilContextTimeout(ctx, config.pollInterval, config.pollTimeout, true /*immediate*/, func(ctx context.Context) (bool, error) {
if err := p.GetClient().Create(ctx, &o, config.createOpts...); err != nil {
if apierrors.IsAlreadyExists(err) {
// Retrying won't help. Abort early.
return false, fmt.Errorf("create %s %s %s: %v", o.GetAPIVersion(), o.GetKind(), klog.KObj(&o), err)
}
log.Logf("error creating %s %s %s, will retry: %v", o.GetAPIVersion(), o.GetKind(), klog.KObj(&o), err)
return false, nil
}
return true, nil
})
}
if err != nil {
retErrs = append(retErrs, err)
}
}
}
return kerrors.NewAggregate(retErrs)
}

// CreateOrUpdate creates or updates objects using the clusterProxy client.
// Defaults to use FieldValidation: strict, which can be overwritten with CreateOrUpdateOptions.
func (p *clusterProxy) CreateOrUpdate(ctx context.Context, resources []byte, opts ...CreateOrUpdateOption) error {
Expand All @@ -327,6 +420,7 @@ func (p *clusterProxy) CreateOrUpdate(ctx context.Context, resources []byte, opt
return err
}

retryDisabled := config.pollTimeout == 0 && config.pollInterval == 0
existingObject := &unstructured.Unstructured{}
var retErrs []error
for _, o := range objs {
Expand All @@ -341,15 +435,43 @@ func (p *clusterProxy) CreateOrUpdate(ctx context.Context, resources []byte, opt
if err := p.GetClient().Get(ctx, objectKey, existingObject); err != nil {
// Expected error -- if the object does not exist, create it
if apierrors.IsNotFound(err) {
if err := p.GetClient().Create(ctx, &o, config.createOpts...); err != nil {
var err error
if retryDisabled {
err = p.GetClient().Create(ctx, &o, config.createOpts...)
} else {
err = wait.PollUntilContextTimeout(ctx, config.pollInterval, config.pollTimeout, true /*immediate*/, func(ctx context.Context) (bool, error) {
if err := p.GetClient().Create(ctx, &o, config.createOpts...); err != nil {
if apierrors.IsAlreadyExists(err) {
// Retrying won't help. Abort early.
return false, fmt.Errorf("create %s %s %s: %v", o.GetAPIVersion(), o.GetKind(), klog.KObj(&o), err)
}
log.Logf("error creating %s %s %s, will retry: %v", o.GetAPIVersion(), o.GetKind(), klog.KObj(&o), err)
return false, nil
}
return true, nil
})
}
if err != nil {
retErrs = append(retErrs, err)
}
} else {
retErrs = append(retErrs, err)
}
} else {
o.SetResourceVersion(existingObject.GetResourceVersion())
if err := p.GetClient().Update(ctx, &o, config.updateOpts...); err != nil {
var err error
if retryDisabled {
err = p.GetClient().Update(ctx, &o, config.updateOpts...)
} else {
err = wait.PollUntilContextTimeout(ctx, config.pollInterval, config.pollTimeout, true /*immediate*/, func(ctx context.Context) (bool, error) {
if err := p.GetClient().Update(ctx, &o, config.updateOpts...); err != nil {
log.Logf("error creating %s %s %s, will retry: %v", o.GetAPIVersion(), o.GetKind(), klog.KObj(&o), err)
return false, nil
}
return true, nil
})
}
if err != nil {
retErrs = append(retErrs, err)
}
}
Expand Down
15 changes: 9 additions & 6 deletions test/framework/clusterctl/clusterctl_helpers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,7 +260,7 @@ type ApplyClusterTemplateAndWaitInput struct {
WaitForControlPlaneIntervals []interface{}
WaitForMachineDeployments []interface{}
WaitForMachinePools []interface{}
CreateOrUpdateOpts []framework.CreateOrUpdateOption // options to be passed to CreateOrUpdate function config
CreateOpts []framework.CreateOption // options to be passed to Create function config
PreWaitForCluster func()
PostMachinesProvisioned func()
ControlPlaneWaiters
Expand Down Expand Up @@ -372,7 +372,7 @@ func ApplyClusterTemplateAndWait(ctx context.Context, input ApplyClusterTemplate
WaitForControlPlaneIntervals: input.WaitForControlPlaneIntervals,
WaitForMachineDeployments: input.WaitForMachineDeployments,
WaitForMachinePools: input.WaitForMachinePools,
CreateOrUpdateOpts: input.CreateOrUpdateOpts,
CreateOpts: input.CreateOpts,
PreWaitForCluster: input.PreWaitForCluster,
PostMachinesProvisioned: input.PostMachinesProvisioned,
ControlPlaneWaiters: input.ControlPlaneWaiters,
Expand All @@ -391,7 +391,7 @@ type ApplyCustomClusterTemplateAndWaitInput struct {
WaitForControlPlaneIntervals []interface{}
WaitForMachineDeployments []interface{}
WaitForMachinePools []interface{}
CreateOrUpdateOpts []framework.CreateOrUpdateOption // options to be passed to CreateOrUpdate function config
CreateOpts []framework.CreateOption // options to be passed to Create function config
PreWaitForCluster func()
PostMachinesProvisioned func()
ControlPlaneWaiters
Expand Down Expand Up @@ -425,9 +425,12 @@ func ApplyCustomClusterTemplateAndWait(ctx context.Context, input ApplyCustomClu
}

log.Logf("Applying the cluster template yaml of cluster %s", klog.KRef(input.Namespace, input.ClusterName))
Eventually(func() error {
return input.ClusterProxy.CreateOrUpdate(ctx, input.CustomTemplateYAML, input.CreateOrUpdateOpts...)
}, 1*time.Minute).Should(Succeed(), "Failed to apply the cluster template")
createOpts := []framework.CreateOption{
// Set default polling. Can be overridden by users.
framework.CreateWithPolling(1*time.Minute, 250*time.Millisecond),
}
createOpts = append(createOpts, input.CreateOpts...)
Expect(input.ClusterProxy.Create(ctx, input.CustomTemplateYAML, createOpts...)).To(Succeed(), "Failed to apply the cluster template")

// Once we applied the cluster template we can run PreWaitForCluster.
// Note: This can e.g. be used to verify the BeforeClusterCreate lifecycle hook is executed
Expand Down
Loading