Skip to content
Open
Show file tree
Hide file tree
Changes from 6 commits
Commits
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
3 changes: 2 additions & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ toolchain go1.26.5
require (
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-20260709200747-435963d16310.1
buf.build/go/protovalidate v1.2.0
github.com/cenkalti/backoff/v4 v4.3.0
github.com/cenkalti/backoff/v7 v7.0.0
github.com/fsnotify/fsnotify v1.10.1
github.com/gabriel-vasile/mimetype v1.4.13
github.com/go-resty/resty/v2 v2.17.2
Expand Down Expand Up @@ -134,6 +134,7 @@ require (
github.com/bytedance/gopkg v0.1.3 // indirect
github.com/bytedance/sonic v1.15.0 // indirect
github.com/bytedance/sonic/loader v0.5.0 // indirect
github.com/cenkalti/backoff/v4 v4.3.0 // indirect
github.com/cenkalti/backoff/v5 v5.0.3 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/cloudwego/base64x v0.1.6 // indirect
Expand Down
2 changes: 2 additions & 0 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,8 @@ github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK3
github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM=
github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw=
github.com/cenkalti/backoff/v7 v7.0.0 h1:ZP+QAaaOnVUHo+ufFpZ835hbT3x2fy+h2lecVEosZ6A=
github.com/cenkalti/backoff/v7 v7.0.0/go.mod h1:qcKBGwsu4hpxHtQ8tWYsQ+ifzx2+sS+Xx/3jfe30lI8=
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
Expand Down
63 changes: 46 additions & 17 deletions internal/backoff/backoff.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,9 @@ package backoff

import (
"context"
"time"

"github.com/cenkalti/backoff/v4"
"github.com/cenkalti/backoff/v7"
"github.com/nginx/agent/v3/internal/config"
)

Expand Down Expand Up @@ -46,14 +47,51 @@ import (
// 10 19.210 {stop}
//
// Information from https://pkg.go.dev/github.com/cenkalti/backoff/v4#section-readme
// RetryOptions builds a slice of backoff.RetryOption from the config.BackOff
// settings and the context. Call sites can pass the returned slice to backoff.Retry as
// variadic options.
//
// If the context has a deadline, the elapsed time limit is set to the remaining
// time until that deadline (if it's less restrictive than the config-based
// MaxElapsedTime). This mirrors the v4 API behavior where WithContext would
// respect the context timeout. If the context has no deadline, the config-based
// MaxElapsedTime is used.
func RetryOptions(ctx context.Context, backoffSettings *config.BackOff) []backoff.RetryOption {
eb := backoff.NewExponentialBackOff()
eb.InitialInterval = backoffSettings.InitialInterval
eb.MaxInterval = backoffSettings.MaxInterval
eb.RandomizationFactor = backoffSettings.RandomizationFactor
eb.Multiplier = backoffSettings.Multiplier

maxElapsedTime := backoffSettings.MaxElapsedTime
if deadline, ok := ctx.Deadline(); ok {
timeUntilDeadline := time.Until(deadline)
// Use the smaller of the two timeouts, but only if deadline is positive
if timeUntilDeadline > 0 && (maxElapsedTime == 0 || timeUntilDeadline < maxElapsedTime) {
maxElapsedTime = timeUntilDeadline
}
}

return []backoff.RetryOption{
backoff.WithBackOff(eb),
backoff.WithMaxElapsedTime(maxElapsedTime),
}
}

// WaitUntil retries a no-result operation until it succeeds, a permanent
// error is returned, or the retry options elapse. It adapts the operation to
// the generic backoff.Retry API.
func WaitUntil(
ctx context.Context,
backoffSettings *config.BackOff,
operation backoff.Operation,
operation func() error,
) error {
backoffWithContext := Context(ctx, backoffSettings)
retryOpts := RetryOptions(ctx, backoffSettings)
_, err := backoff.Retry(ctx, func() (struct{}, error) {
return struct{}{}, operation()
}, retryOpts...)

return backoff.Retry(operation, backoffWithContext)
return err
}

// WaitUntilWithData Implementation of backoff operations that increases the back off period for each retry
Expand All @@ -63,21 +101,12 @@ func WaitUntil(
func WaitUntilWithData[T any](
ctx context.Context,
backoffSettings *config.BackOff,
operation backoff.OperationWithData[T],
operation backoff.Operation[T],
) (T, error) {
backoffWithContext := Context(ctx, backoffSettings)

return backoff.RetryWithData(operation, backoffWithContext)
return backoff.Retry[T](ctx, operation, RetryOptions(ctx, backoffSettings)...)
}

//nolint:ireturn // must return an interface
func Context(ctx context.Context, backoffSettings *config.BackOff) backoff.BackOffContext {
eb := backoff.NewExponentialBackOff()
eb.InitialInterval = backoffSettings.InitialInterval
eb.MaxInterval = backoffSettings.MaxInterval
eb.MaxElapsedTime = backoffSettings.MaxElapsedTime
eb.RandomizationFactor = backoffSettings.RandomizationFactor
eb.Multiplier = backoffSettings.Multiplier

return backoff.WithContext(eb, ctx)
func Context(ctx context.Context, backoffSettings *config.BackOff) []backoff.RetryOption {
Comment thread
karensantana marked this conversation as resolved.
Outdated
return RetryOptions(ctx, backoffSettings)
}
5 changes: 2 additions & 3 deletions internal/backoff/backoff_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@ import (
"testing"
"time"

"github.com/cenkalti/backoff/v4"
"github.com/nginx/agent/v3/internal/config"
"github.com/stretchr/testify/assert"
)
Expand All @@ -21,7 +20,7 @@ func TestWaitUntil(t *testing.T) {
invocations := 0
tests := []struct {
context context.Context
operation backoff.Operation
operation func() error
name string
initialInterval time.Duration
maxInterval time.Duration
Expand Down Expand Up @@ -106,7 +105,7 @@ func TestWaitUntilWithData(t *testing.T) {
invocations := -1
tests := []struct {
context context.Context
operation backoff.OperationWithData[int]
operation func() (int, error)
name string
initialInterval time.Duration
maxInterval time.Duration
Expand Down
31 changes: 17 additions & 14 deletions internal/command/command_service.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ import (
"sync"
"sync/atomic"

"github.com/cenkalti/backoff/v4"
"github.com/cenkalti/backoff/v7"

mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
"github.com/nginx/agent/v3/internal/config"
Expand Down Expand Up @@ -116,9 +116,10 @@ func (cs *CommandService) UpdateDataPlaneStatus(
return response, nil
}

response, err := backoff.RetryWithData(
response, err := backoff.Retry(
backOffCtx,
sendDataPlaneStatus,
backoffHelpers.Context(backOffCtx, cfg.Client.Backoff),
backoffHelpers.RetryOptions(backOffCtx, cfg.Client.Backoff)...,
)
if err != nil {
return err
Expand Down Expand Up @@ -152,9 +153,10 @@ func (cs *CommandService) UpdateDataPlaneHealth(ctx context.Context, instanceHea
backOffCtx, backoffCancel := context.WithTimeout(ctx, cfg.Client.Backoff.MaxElapsedTime)
defer backoffCancel()

response, err := backoff.RetryWithData(
response, err := backoff.Retry(
backOffCtx,
cs.dataPlaneHealthCallback(ctx, request),
backoffHelpers.Context(backOffCtx, cfg.Client.Backoff),
backoffHelpers.RetryOptions(backOffCtx, cfg.Client.Backoff)...,
)
if err != nil {
return err
Expand All @@ -167,7 +169,6 @@ func (cs *CommandService) UpdateDataPlaneHealth(ctx context.Context, instanceHea

func (cs *CommandService) SendDataPlaneResponse(ctx context.Context, response *mpi.DataPlaneResponse) error {
slog.DebugContext(ctx, "Sending data plane response", "response", response)

cfg := cs.config()
backOffCtx, backoffCancel := context.WithTimeout(ctx, cfg.Client.Backoff.MaxElapsedTime)
defer backoffCancel()
Expand All @@ -177,9 +178,10 @@ func (cs *CommandService) SendDataPlaneResponse(ctx context.Context, response *m
return err
}

return backoff.Retry(
return backoffHelpers.WaitUntil(
backOffCtx,
cfg.Client.Backoff,
cs.sendDataPlaneResponseCallback(ctx, response),
backoffHelpers.Context(backOffCtx, cfg.Client.Backoff),
)
}

Expand Down Expand Up @@ -209,7 +211,7 @@ func (cs *CommandService) Subscribe(ctx context.Context) {
case <-ctx.Done():
return
default:
retryError := backoff.Retry(cs.receiveCallback(ctx), backoffHelpers.Context(ctx, commonSettings))
retryError := backoffHelpers.WaitUntil(ctx, commonSettings, cs.receiveCallback(ctx))
if retryError != nil {
slog.WarnContext(ctx, "Failed to receive messages from subscribe stream", "error", retryError)
}
Expand Down Expand Up @@ -281,9 +283,10 @@ func (cs *CommandService) createConnectionCall(ctx context.Context) (*mpi.Create
}

slog.DebugContext(ctx, "Sending create connection request", "request", request)
resp, err := backoff.RetryWithData(
resp, err := backoff.Retry(
ctx,
cs.connectCallback(ctx, request),
backoffHelpers.Context(ctx, commonSettings),
backoffHelpers.RetryOptions(ctx, commonSettings)...,
)
if err != nil {
cs.isConnected.Store(false)
Expand Down Expand Up @@ -378,12 +381,12 @@ func (cs *CommandService) handleConfigApplyResponse(
}

slog.DebugContext(ctx, "Sending data plane response for queued config apply request", "response", newResponse)

backOffCtx, backoffCancel := context.WithTimeout(ctx, cfg.Client.Backoff.MaxElapsedTime)

err := backoff.Retry(
err := backoffHelpers.WaitUntil(
backOffCtx,
cfg.Client.Backoff,
cs.sendDataPlaneResponseCallback(ctx, newResponse),
backoffHelpers.Context(backOffCtx, cfg.Client.Backoff),
)
backoffCancel()

Expand Down
22 changes: 12 additions & 10 deletions internal/file/file_service_operator.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ import (
"sync"
"sync/atomic"

"github.com/cenkalti/backoff/v4"
"github.com/cenkalti/backoff/v7"
mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
backoffHelpers "github.com/nginx/agent/v3/internal/backoff"
"github.com/nginx/agent/v3/internal/config"
Expand Down Expand Up @@ -105,9 +105,10 @@ func (fso *FileServiceOperator) File(
return response, nil
}

getFileResp, getFileErr := backoff.RetryWithData(
getFileResp, getFileErr := backoff.Retry(
backOffCtx,
getFile,
backoffHelpers.Context(backOffCtx, fso.agentConfig.Client.Backoff),
backoffHelpers.RetryOptions(backOffCtx, fso.agentConfig.Client.Backoff)...,
)

if getFileErr != nil {
Expand Down Expand Up @@ -215,9 +216,10 @@ func (fso *FileServiceOperator) UpdateOverview(
return response, nil
}

response, err := backoff.RetryWithData(
response, err := backoff.Retry(
newCtx,
sendUpdateOverview,
backoffHelpers.Context(newCtx, backoffSettings),
backoffHelpers.RetryOptions(newCtx, backoffSettings)...,
)
if err != nil {
return err
Expand Down Expand Up @@ -399,7 +401,6 @@ func (fso *FileServiceOperator) sendUpdateFileRequest(
},
MessageMeta: messageMeta,
}

backOffCtx, backoffCancel := context.WithTimeout(ctx, fso.agentConfig.Client.Backoff.MaxElapsedTime)
defer backoffCancel()

Expand Down Expand Up @@ -431,9 +432,10 @@ func (fso *FileServiceOperator) sendUpdateFileRequest(
return response, nil
}

response, err := backoff.RetryWithData(
response, err := backoff.Retry(
backOffCtx,
sendUpdateFile,
backoffHelpers.Context(backOffCtx, fso.agentConfig.Client.Backoff),
backoffHelpers.RetryOptions(backOffCtx, fso.agentConfig.Client.Backoff)...,
)
if err != nil {
return err
Expand Down Expand Up @@ -522,7 +524,7 @@ func (fso *FileServiceOperator) sendUpdateFileStreamHeader(
return nil
}

return backoff.Retry(sendUpdateFileHeader, backoffHelpers.Context(backOffCtx, fso.agentConfig.Client.Backoff))
return backoffHelpers.WaitUntil(backOffCtx, fso.agentConfig.Client.Backoff, sendUpdateFileHeader)
}

func (fso *FileServiceOperator) sendFileUpdateStreamChunks(
Expand Down Expand Up @@ -616,7 +618,7 @@ func (fso *FileServiceOperator) sendFileUpdateStreamChunk(
return nil
}

return backoff.Retry(sendUpdateFileChunk, backoffHelpers.Context(backOffCtx, fso.agentConfig.Client.Backoff))
return backoffHelpers.WaitUntil(backOffCtx, fso.agentConfig.Client.Backoff, sendUpdateFileChunk)
}

func (fso *FileServiceOperator) setupIdentifiers(ctx context.Context, iteration int) (context.Context, string) {
Expand Down
3 changes: 2 additions & 1 deletion internal/file/file_service_operator_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import (
"testing"
"time"

"github.com/cenkalti/backoff/v7"
mpi "github.com/nginx/agent/v3/api/grpc/mpi/v1"
"github.com/nginx/agent/v3/api/grpc/mpi/v1/v1fakes"
"github.com/nginx/agent/v3/pkg/files"
Expand Down Expand Up @@ -127,7 +128,7 @@ func TestFileServiceOperator_UpdateOverview_NoConnection(t *testing.T) {
},
}, filePath, 0)

assert.ErrorIs(t, err, context.DeadlineExceeded)
assert.ErrorIs(t, err, backoff.ErrMaxElapsedTime)
}

func TestFileManagerService_UpdateFile(t *testing.T) {
Expand Down
2 changes: 1 addition & 1 deletion internal/grpc/grpc.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ import (

"github.com/nginx/agent/v3/internal/datasource/file"

"github.com/cenkalti/backoff/v4"
"github.com/cenkalti/backoff/v7"
grpcRetry "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/retry"

"buf.build/go/protovalidate"
Expand Down
4 changes: 2 additions & 2 deletions internal/grpc/grpc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ import (
"fmt"
"testing"

"github.com/cenkalti/backoff/v4"
"github.com/cenkalti/backoff/v7"
"github.com/nginx/agent/v3/test/stub"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
Expand Down Expand Up @@ -391,7 +391,7 @@ func Test_ValidateGrpcError(t *testing.T) {
require.ErrorIs(t, TestError{}, result)
}
result = ValidateGrpcError(status.Errorf(codes.InvalidArgument, "error"))
require.ErrorIs(t, &backoff.PermanentError{}, result)
require.ErrorIs(t, result, backoff.ErrPermanent)
}

func Test_transportCredentials(t *testing.T) {
Expand Down
9 changes: 7 additions & 2 deletions internal/nginx/nginx_instance_operator.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,8 @@ import (
"path/filepath"
"time"

"github.com/nginx/agent/v3/internal/backoff"
"github.com/cenkalti/backoff/v7"
backoffHelpers "github.com/nginx/agent/v3/internal/backoff"
"github.com/nginx/agent/v3/pkg/nginxprocess"

"github.com/nginx/agent/v3/pkg/host/exec"
Expand Down Expand Up @@ -146,7 +147,7 @@ func (i *NginxInstanceOperator) checkWorkers(ctx context.Context, instanceID str
slog.DebugContext(ctx, "Found parent process ID, checking NGINX worker processes have reloaded",
"process_id", newPid)

err := backoff.WaitUntil(ctx, backoffSettings, func() error {
err := backoffHelpers.WaitUntil(ctx, backoffSettings, func() error {
currentWorkers := i.nginxProcessOperator.NginxWorkerProcesses(ctx, newPid)
if len(currentWorkers) == 0 {
return errors.New("waiting for NGINX worker processes")
Expand All @@ -160,6 +161,10 @@ func (i *NginxInstanceOperator) checkWorkers(ctx context.Context, instanceID str

return fmt.Errorf("waiting for NGINX worker to be newer than %v", createdTime)
})

if re := backoff.AsRetryError(err); re != nil {
err = re.LastErr
}
if err != nil {
slog.WarnContext(
ctx,
Expand Down
Loading
Loading