From 6626b97ac1e348f2ce9ab90c44ca460f99e73655 Mon Sep 17 00:00:00 2001 From: Nsanjayboruds Date: Thu, 9 Jul 2026 11:30:35 +0530 Subject: [PATCH] refactor: replace WaitNext with retry.Do in piped components Signed-off-by: Nsanjayboruds --- pkg/app/piped/cmd/piped/piped.go | 10 ++-- pkg/app/piped/controller/controller.go | 42 ++++++++--------- pkg/app/piped/controller/planner.go | 33 +++++++------ pkg/app/piped/controller/scheduler.go | 46 ++++++++++--------- pkg/app/piped/executor/lambda/lambda.go | 12 ++--- .../piped/platformprovider/lambda/client.go | 17 +++---- pkg/app/piped/trigger/cache.go | 18 ++++---- pkg/app/piped/trigger/deployment.go | 12 ++--- pkg/backoff/backoff.go | 6 +-- pkg/backoff/backoff_test.go | 14 +++--- 10 files changed, 105 insertions(+), 105 deletions(-) diff --git a/pkg/app/piped/cmd/piped/piped.go b/pkg/app/piped/cmd/piped/piped.go index 86909fab15..6f3005427b 100644 --- a/pkg/app/piped/cmd/piped/piped.go +++ b/pkg/app/piped/cmd/piped/piped.go @@ -834,19 +834,21 @@ func (p *piped) sendPipedMeta(ctx context.Context, client pipedservice.Client, c } } - for retry.WaitNext(ctx) { - if res, err := client.ReportPipedMeta(ctx, req); err == nil { + _, err = retry.Do(ctx, func() (interface{}, error) { + res, err := client.ReportPipedMeta(ctx, req) + if err == nil { cfg.Name = res.Name if cfg.WebAddress == "" { cfg.WebAddress = res.WebBaseUrl } - return nil + return nil, nil } logger.Warn("failed to report piped meta to control-plane, wait to the next retry", zap.Int("calls", retry.Calls()), zap.Error(err), ) - } + return nil, err + }) return err } diff --git a/pkg/app/piped/controller/controller.go b/pkg/app/piped/controller/controller.go index 2d1f72af1d..2fad655c06 100644 --- a/pkg/app/piped/controller/controller.go +++ b/pkg/app/piped/controller/controller.go @@ -674,8 +674,6 @@ func (c *controller) startNewScheduler(ctx context.Context, d *model.Deployment) func (c *controller) getMostRecentlySuccessfulDeployment(ctx context.Context, applicationID string) (*model.ApplicationDeploymentReference, error) { var ( - err error - resp *pipedservice.GetApplicationMostRecentDeploymentResponse retry = pipedservice.NewRetry(3) req = &pipedservice.GetApplicationMostRecentDeploymentRequest{ ApplicationId: applicationID, @@ -683,15 +681,17 @@ func (c *controller) getMostRecentlySuccessfulDeployment(ctx context.Context, ap } ) - for retry.WaitNext(ctx) { - if resp, err = c.apiClient.GetApplicationMostRecentDeployment(ctx, req); err == nil { - return resp.Deployment, nil - } - if !pipedservice.Retriable(err) { - return nil, err + d, err := retry.Do(ctx, func() (interface{}, error) { + resp, err := c.apiClient.GetApplicationMostRecentDeployment(ctx, req) + if err != nil { + return nil, pipedservice.NewRetriableErr(err) } + return resp.Deployment, nil + }) + if err != nil { + return nil, err } - return nil, err + return d.(*model.ApplicationDeploymentReference), nil } func (c *controller) shouldStartPlanningDeployment(ctx context.Context, d *model.Deployment) (plannable, cancel bool, cancelReason string, err error) { @@ -715,7 +715,6 @@ func (c *controller) shouldStartPlanningDeployment(ctx context.Context, d *model func (c *controller) cancelDeployment(ctx context.Context, d *model.Deployment, reason string) error { var ( - err error req = &pipedservice.ReportDeploymentCompletedRequest{ DeploymentId: d.Id, Status: model.DeploymentStatus_DEPLOYMENT_CANCELLED, @@ -728,12 +727,13 @@ func (c *controller) cancelDeployment(ctx context.Context, d *model.Deployment, retry = pipedservice.NewRetry(10) ) - for retry.WaitNext(ctx) { - if _, err = c.apiClient.ReportDeploymentCompleted(ctx, req); err == nil { - return nil + _, err := retry.Do(ctx, func() (interface{}, error) { + _, err := c.apiClient.ReportDeploymentCompleted(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report deployment status to control-plane: %v", err) } - err = fmt.Errorf("failed to report deployment status to control-plane: %v", err) - } + return nil, nil + }) return err } @@ -749,7 +749,6 @@ func (l appLiveResourceLister) ListKubernetesResources() ([]provider.Manifest, b func reportApplicationDeployingStatus(ctx context.Context, c apiClient, appID string, deploying bool) error { var ( - err error retry = pipedservice.NewRetry(10) req = &pipedservice.ReportApplicationDeployingStatusRequest{ ApplicationId: appID, @@ -757,11 +756,12 @@ func reportApplicationDeployingStatus(ctx context.Context, c apiClient, appID st } ) - for retry.WaitNext(ctx) { - if _, err = c.ReportApplicationDeployingStatus(ctx, req); err == nil { - return nil + _, err := retry.Do(ctx, func() (interface{}, error) { + _, err := c.ReportApplicationDeployingStatus(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report application deploying status to control-plane: %w", err) } - err = fmt.Errorf("failed to report application deploying status to control-plane: %w", err) - } + return nil, nil + }) return err } diff --git a/pkg/app/piped/controller/planner.go b/pkg/app/piped/controller/planner.go index 5bbf2f1b36..c16c4b8b78 100644 --- a/pkg/app/piped/controller/planner.go +++ b/pkg/app/piped/controller/planner.go @@ -273,12 +273,13 @@ func (p *planner) reportDeploymentPlanned(ctx context.Context, out pln.Output) e }) }() - for retry.WaitNext(ctx) { - if _, err = p.apiClient.ReportDeploymentPlanned(ctx, req); err == nil { - return nil + _, err = retry.Do(ctx, func() (interface{}, error) { + _, err := p.apiClient.ReportDeploymentPlanned(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report deployment status to control-plane: %v", err) } - err = fmt.Errorf("failed to report deployment status to control-plane: %v", err) - } + return nil, nil + }) if err != nil { p.logger.Error("failed to mark deployment to be planned", zap.Error(err)) @@ -319,12 +320,13 @@ func (p *planner) reportDeploymentFailed(ctx context.Context, reason string) err }) }() - for retry.WaitNext(ctx) { - if _, err = p.apiClient.ReportDeploymentCompleted(ctx, req); err == nil { - return nil + _, err = retry.Do(ctx, func() (interface{}, error) { + _, err := p.apiClient.ReportDeploymentCompleted(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report deployment status to control-plane: %v", err) } - err = fmt.Errorf("failed to report deployment status to control-plane: %v", err) - } + return nil, nil + }) if err != nil { p.logger.Error("failed to mark deployment to be failed", zap.Error(err)) @@ -365,12 +367,13 @@ func (p *planner) reportDeploymentCancelled(ctx context.Context, commander, reas }) }() - for retry.WaitNext(ctx) { - if _, err = p.apiClient.ReportDeploymentCompleted(ctx, req); err == nil { - return nil + _, err = retry.Do(ctx, func() (interface{}, error) { + _, err := p.apiClient.ReportDeploymentCompleted(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report deployment status to control-plane: %v", err) } - err = fmt.Errorf("failed to report deployment status to control-plane: %v", err) - } + return nil, nil + }) if err != nil { p.logger.Error("failed to mark deployment to be cancelled", zap.Error(err)) diff --git a/pkg/app/piped/controller/scheduler.go b/pkg/app/piped/controller/scheduler.go index 6c5b888803..ed48930e3a 100644 --- a/pkg/app/piped/controller/scheduler.go +++ b/pkg/app/piped/controller/scheduler.go @@ -678,14 +678,13 @@ func (s *scheduler) reportStageStatus(ctx context.Context, stageID string, statu // Update stage status at local. s.stageStatuses[stageID] = status - // Update stage status on the remote. - for retry.WaitNext(ctx) { - _, err = s.apiClient.ReportStageStatusChanged(ctx, req) - if err == nil { - break + _, err = retry.Do(ctx, func() (interface{}, error) { + _, err := s.apiClient.ReportStageStatusChanged(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report stage status to control-plane: %v", err) } - err = fmt.Errorf("failed to report stage status to control-plane: %v", err) - } + return nil, nil + }) return err } @@ -704,12 +703,13 @@ func (s *scheduler) reportDeploymentStatusChanged(ctx context.Context, status mo ) // Update deployment status on remote. - for retry.WaitNext(ctx) { - if _, err = s.apiClient.ReportDeploymentStatusChanged(ctx, req); err == nil { - return nil + _, err = retry.Do(ctx, func() (interface{}, error) { + _, err := s.apiClient.ReportDeploymentStatusChanged(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report deployment status to control-plane: %v", err) } - err = fmt.Errorf("failed to report deployment status to control-plane: %v", err) - } + return nil, nil + }) return err } @@ -782,12 +782,13 @@ func (s *scheduler) reportDeploymentCompleted(ctx context.Context, status model. }() // Update deployment status on remote. - for retry.WaitNext(ctx) { - if _, err = s.apiClient.ReportDeploymentCompleted(ctx, req); err == nil { - return nil + _, err = retry.Do(ctx, func() (interface{}, error) { + _, err := s.apiClient.ReportDeploymentCompleted(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report deployment status to control-plane: %w", err) } - err = fmt.Errorf("failed to report deployment status to control-plane: %w", err) - } + return nil, nil + }) return err } @@ -825,12 +826,13 @@ func (s *scheduler) reportMostRecentlySuccessfulDeployment(ctx context.Context) retry = pipedservice.NewRetry(10) ) - for retry.WaitNext(ctx) { - if _, err = s.apiClient.ReportApplicationMostRecentDeployment(ctx, req); err == nil { - return nil + _, err = retry.Do(ctx, func() (interface{}, error) { + _, err := s.apiClient.ReportApplicationMostRecentDeployment(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report most recent successful deployment: %w", err) } - err = fmt.Errorf("failed to report most recent successful deployment: %w", err) - } + return nil, nil + }) return err } diff --git a/pkg/app/piped/executor/lambda/lambda.go b/pkg/app/piped/executor/lambda/lambda.go index 886f8a57d6..69edd9da9e 100644 --- a/pkg/app/piped/executor/lambda/lambda.go +++ b/pkg/app/piped/executor/lambda/lambda.go @@ -308,9 +308,8 @@ func build(ctx context.Context, in *executor.Input, client provider.Client, fm p in.LogPersister.Info("Waiting to update lambda function in progress...") retry := backoff.NewRetry(provider.RequestRetryTime, backoff.NewConstant(provider.RetryIntervalDuration)) - publishFunctionSucceed := false startWaitingStamp := time.Now() - for retry.WaitNext(ctx) { + _, err = retry.Do(ctx, func() (interface{}, error) { // Commit version for applied Lambda function. // Note: via the current docs of [Lambda.PublishVersion](https://docs.aws.amazon.com/sdk-for-go/api/service/lambda/#Lambda.PublishVersion) // AWS Lambda doesn't publish a version if the function's configuration and code haven't changed since the last version. @@ -318,12 +317,11 @@ func build(ctx context.Context, in *executor.Input, client provider.Client, fm p version, err = client.PublishFunction(ctx, fm) if err != nil { in.Logger.Error("Failed publish new version for Lambda function") - } else { - publishFunctionSucceed = true - break + return nil, err } - } - if !publishFunctionSucceed { + return nil, nil + }) + if err != nil { in.LogPersister.Errorf("Failed to commit new version for Lambda function %s: %v", fm.Spec.Name, err) return } diff --git a/pkg/app/piped/platformprovider/lambda/client.go b/pkg/app/piped/platformprovider/lambda/client.go index ddaa86723a..f9bca30a14 100644 --- a/pkg/app/piped/platformprovider/lambda/client.go +++ b/pkg/app/piped/platformprovider/lambda/client.go @@ -284,9 +284,7 @@ func (c *client) UpdateFunctionFromSource(ctx context.Context, fm FunctionManife func (c *client) updateFunctionConfiguration(ctx context.Context, fm FunctionManifest) error { retry := backoff.NewRetry(RequestRetryTime, backoff.NewConstant(RetryIntervalDuration)) - updateFunctionConfigurationSucceed := false - var err error - for retry.WaitNext(ctx) { + _, err := retry.Do(ctx, func() (interface{}, error) { configInput := &lambda.UpdateFunctionConfigurationInput{ FunctionName: aws.String(fm.Spec.Name), Role: aws.String(fm.Spec.Role), @@ -314,16 +312,15 @@ func (c *client) updateFunctionConfiguration(ctx context.Context, fm FunctionMan SubnetIds: fm.Spec.VPCConfig.SubnetIDs, } } - _, err = c.client.UpdateFunctionConfiguration(ctx, configInput) + _, err := c.client.UpdateFunctionConfiguration(ctx, configInput) if err != nil { c.logger.Error("Failed to update function configuration") - } else { - updateFunctionConfigurationSucceed = true - break + return nil, fmt.Errorf("failed to update configuration for Lambda function %s: %w", fm.Spec.Name, err) } - } - if !updateFunctionConfigurationSucceed { - return fmt.Errorf("failed to update configuration for Lambda function %s: %w", fm.Spec.Name, err) + return nil, nil + }) + if err != nil { + return err } // Wait until function updated successfully. diff --git a/pkg/app/piped/trigger/cache.go b/pkg/app/piped/trigger/cache.go index 8ae227060a..b768d838cb 100644 --- a/pkg/app/piped/trigger/cache.go +++ b/pkg/app/piped/trigger/cache.go @@ -58,8 +58,6 @@ func (s *lastTriggeredCommitStore) Put(applicationID, commit string) error { func (s *lastTriggeredCommitStore) getLastTriggeredDeployment(ctx context.Context, applicationID string) (*model.ApplicationDeploymentReference, error) { var ( - err error - resp *pipedservice.GetApplicationMostRecentDeploymentResponse retry = pipedservice.NewRetry(3) req = &pipedservice.GetApplicationMostRecentDeploymentRequest{ ApplicationId: applicationID, @@ -67,13 +65,15 @@ func (s *lastTriggeredCommitStore) getLastTriggeredDeployment(ctx context.Contex } ) - for retry.WaitNext(ctx) { - if resp, err = s.apiClient.GetApplicationMostRecentDeployment(ctx, req); err == nil { - return resp.Deployment, nil - } - if !pipedservice.Retriable(err) { - return nil, err + d, err := retry.Do(ctx, func() (interface{}, error) { + resp, err := s.apiClient.GetApplicationMostRecentDeployment(ctx, req) + if err != nil { + return nil, pipedservice.NewRetriableErr(err) } + return resp.Deployment, nil + }) + if err != nil { + return nil, err } - return nil, err + return d.(*model.ApplicationDeploymentReference), nil } diff --git a/pkg/app/piped/trigger/deployment.go b/pkg/app/piped/trigger/deployment.go index beea39cefe..5d8b6a7763 100644 --- a/pkg/app/piped/trigger/deployment.go +++ b/pkg/app/piped/trigger/deployment.go @@ -113,7 +113,6 @@ func buildDeployment( func reportMostRecentlyTriggeredDeployment(ctx context.Context, client apiClient, d *model.Deployment) error { var ( - err error req = &pipedservice.ReportApplicationMostRecentDeploymentRequest{ ApplicationId: d.ApplicationId, Status: model.DeploymentStatus_DEPLOYMENT_PENDING, @@ -129,11 +128,12 @@ func reportMostRecentlyTriggeredDeployment(ctx context.Context, client apiClient retry = pipedservice.NewRetry(10) ) - for retry.WaitNext(ctx) { - if _, err = client.ReportApplicationMostRecentDeployment(ctx, req); err == nil { - return nil + _, err := retry.Do(ctx, func() (interface{}, error) { + _, err := client.ReportApplicationMostRecentDeployment(ctx, req) + if err != nil { + return nil, fmt.Errorf("failed to report most recent successful deployment: %w", err) } - err = fmt.Errorf("failed to report most recent successful deployment: %w", err) - } + return nil, nil + }) return err } diff --git a/pkg/backoff/backoff.go b/pkg/backoff/backoff.go index 122e7ab2f5..d13548ed33 100644 --- a/pkg/backoff/backoff.go +++ b/pkg/backoff/backoff.go @@ -44,7 +44,6 @@ type Backoff interface { type Retry interface { Do(ctx context.Context, operation func() (interface{}, error)) (interface{}, error) - WaitNext(ctx context.Context) bool Calls() int } @@ -62,8 +61,7 @@ type retry struct { backoff Backoff } -// TODO: Find all using of WaitNext and replace by Do to avoid panic. -func (r *retry) WaitNext(ctx context.Context) bool { +func (r *retry) waitNext(ctx context.Context) bool { defer func() { r.calls++ }() @@ -98,7 +96,7 @@ func (r *retry) WaitNext(ctx context.Context) bool { func (r *retry) Do(ctx context.Context, operation func() (interface{}, error)) (interface{}, error) { var err error - for r.WaitNext(ctx) { + for r.waitNext(ctx) { var data interface{} data, err = operation() if err == nil { diff --git a/pkg/backoff/backoff_test.go b/pkg/backoff/backoff_test.go index ad8f55a54c..cc9295b98c 100644 --- a/pkg/backoff/backoff_test.go +++ b/pkg/backoff/backoff_test.go @@ -23,28 +23,28 @@ import ( "github.com/stretchr/testify/assert" ) -func TestWaitNext(t *testing.T) { +func TestWaitNextInternal(t *testing.T) { var ( bo = NewConstant(time.Millisecond) - r = NewRetry(10, bo) + r = NewRetry(10, bo).(*retry) ctx, cancel = context.WithCancel(context.TODO()) ) - ok := r.WaitNext(ctx) + ok := r.waitNext(ctx) assert.Equal(t, true, ok) cancel() - ok = r.WaitNext(ctx) + ok = r.waitNext(ctx) assert.Equal(t, false, ok) } -func TestWaitNextCancel(t *testing.T) { +func TestWaitNextCancelInternal(t *testing.T) { var ( bo = NewConstant(time.Minute) - r = NewRetry(3, bo) + r = NewRetry(3, bo).(*retry) ctx, cancel = context.WithCancel(context.TODO()) ) cancel() - ok := r.WaitNext(ctx) + ok := r.waitNext(ctx) assert.Equal(t, false, ok) }