googleapis / googleapis/google-cloud-go

bigquery/storage/managedwriter: custom gax.Retryer

Open
#5,094 7 comments 0 reactions 1 assignee Claimed by @shollyman View on GitHub
api: bigquery priority: p3 type: feature request
Dominant language
Go
Stars
4.5k
Forks
1.6k
Avg merge
1d 13h
Merged PRs (30d)
109

Description

**Is your feature request related to a problem? Please describe.**

The BQ ManagedWriter for the Storage API uses a `gax.Retryer` under the hood.
It does so for 2 purposes:
- when (re)-opening an underlying stream connection
- when appending data

In both cases it is not possible for me to define what Retryer to use. This means that I cannot define
my own conditions to retry. One severe consequence is that currently it is not possible to put a maximum deadline for the
back-off algorithm, and so in theory this can block forever.

**Describe the solution you'd like**

Be able to define my own retryer. For the AppendRows command this could be as obvious as allowing us to define `gax.CallOptions`s ourselves. This solution wouldn't allow us to define what to do for the re(connect) use-case though. So either the latter is handled separately with an initial config, or the initial config is to be used for both use cases.

**Describe alternatives you've considered**

- do not use the managed writer: this is possible by using the apiv1 package directly,
but feels a lot like reinventing the wheel to me;
- accept that I cannot define the Retryer myself: this is possible, but makes it that in worst case
it could be retrying forever;
- fork the managed writer code: this is what I've done now and allows me to use a custom retryer;

---

**Extra**

Here is the code of the retryer I developed myself and use in my forked ManagedWriter.
Note that I am not an expert in either GRPC or the google cloud Go API. I wrote this Retryer
based on some short experience with it as well as the documentation sources.

```go
// Retryer is a retryer inspired by other community back-off implementations,
// in order to not have another dependency added to this library,
// while still being able to rely on existing retry-related google code of
// dependencies already required by this library for its core functionality
type Retryer struct {
backoff gax.Backoff
retries int
maxRetries int
startTime time.Time
maxRetryDeadlineOffset time.Duration
deadlineCtx context.Context
cancelDeadlineCtx func()
}

// compile-time interface compliance
var _ gax.Retryer = (*Retryer)(nil)

func NewRetryer(ctx context.Context, maxRetries int, initialRetryDelay time.Duration, maxRetryDeadlineOffset time.Duration, retryDelayMultiplier float64) *Retryer {
startTime := time.Now()
deadlineCtx, cancelDeadlineCtx := context.WithDeadline(ctx, startTime.Add(maxRetryDeadlineOffset))
return &Retryer{
backoff: gax.Backoff{
Initial: initialRetryDelay,
Max: maxRetryDeadlineOffset,
Multiplier: retryDelayMultiplier,
},
maxRetries: maxRetries,
startTime: startTime,
maxRetryDeadlineOffset: maxRetryDeadlineOffset,
deadlineCtx: deadlineCtx,
cancelDeadlineCtx: cancelDeadlineCtx,
}
}

// RetryOp retries the operation
func (r *Retryer) RetryOp(op func(context.Context) error) error {
defer r.cancelDeadlineCtx()
for {
err := op(r.deadlineCtx)
if err == nil {
return nil
}
pause, ok := r.Retry(err)
if !ok {
// retryer wishes not to retry, return early
return err
}
// we'll retry, but first will sleep
time.Sleep(pause)
}
}

// Retry implements gax::Retryer::Retry
func (r *Retryer) Retry(err error) (pause time.Duration, shouldRetry bool) {
defer func() {
if !shouldRetry {
r.cancelDeadlineCtx()
}
}()
if err == nil {
// no error returned, no need to retry
return 0, false
}
if errors.Is(r.deadlineCtx.Err(), context.Canceled) {
// if parent ctx is done or the deadline has been reached,
// no retry is possible any longer either
return 0, false
}
if r.retries >= r.maxRetries {
// no longer need to retry,
// already exhausted our retry attempts
return 0, false
}
// do not retry in case we have a non-retryable GRPC error
st, ok := status.FromError(err)
if !ok {
// do not retry, as this either means we have an unexpected kind of error,
// or it means we had no error at all
return false
}
switch st.Code() {
case codes.Unavailable, codes.FailedPrecondition, codes.ResourceExhausted, codes.DataLoss:
// do retry
default:
return false
}
// correct the Max time, as to stay as close as possible to our max elapsed retry time
elapsedTime := time.Since(r.startTime)
r.backoff.Max = r.maxRetryDeadlineOffset - elapsedTime
// retry with the pause time indicated by the gax BackOff algorithm
r.retries += 1
return r.backoff.Pause(), true
}
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.