From 8486efaa79034398d8a27b38ed568d1145bae23f Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Mon, 13 Dec 2021 12:27:44 -0600 Subject: [PATCH] add exponential retry logic to internal http client, use in backup --- cmd/backup.go | 11 ++++---- ctl/backup.go | 9 +++++-- ctl/common.go | 4 +-- http/client.go | 68 ++++++++++++++++++++++++++++++++++++++++++++++---- 4 files changed, 78 insertions(+), 14 deletions(-) diff --git a/cmd/backup.go b/cmd/backup.go index 5761b6d57..133f0dbce 100644 --- a/cmd/backup.go +++ b/cmd/backup.go @@ -23,11 +23,12 @@ Backs up a FeatureBase server to a local, tar-formatted snapshot file. } flags := ccmd.Flags() - flags.StringVarP(&cmd.OutputDir, "output", "o", "", "output dir to write to") - flags.BoolVar(&cmd.NoSync, "no-sync", false, "disable file sync") - flags.IntVar(&cmd.Concurrency, "concurrency", cmd.Concurrency, "number of concurrent backup goroutines") - flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.") - flags.StringVar(&cmd.Index, "index", "", "index to backup, default backs up all indexes. ") + flags.StringVarP(&cmd.OutputDir, "output", "o", "", "Output directory to write to.") + flags.BoolVar(&cmd.NoSync, "no-sync", false, "Disable file sync") + flags.IntVar(&cmd.Concurrency, "concurrency", cmd.Concurrency, "Number of concurrent backup goroutines.") + flags.StringVar(&cmd.Host, "host", "localhost:10101", "The address (host:port) of FeatureBase (HTTP).") + flags.StringVar(&cmd.Index, "index", "", "Index to backup, default backs up all indexes. ") + flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.") ctl.SetTLSConfig(flags, "", &cmd.TLS.CertificatePath, &cmd.TLS.CertificateKeyPath, &cmd.TLS.CACertPath, &cmd.TLS.SkipVerify, &cmd.TLS.EnableClientVerification) return ccmd } diff --git a/ctl/backup.go b/ctl/backup.go index 0e8a257f1..8ba57c278 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -10,6 +10,7 @@ import ( "io/ioutil" "os" "path/filepath" + "time" pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/http" @@ -37,6 +38,9 @@ type BackupCommand struct { // nolint: maligned // Number of concurrent backup goroutines running at a time. Concurrency int + // Amount of time after first failed request to continue retrying. + RetryPeriod time.Duration `json:"retry-period"` + // Reusable client. client pilosa.InternalClient @@ -51,6 +55,7 @@ func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand return &BackupCommand{ CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), Concurrency: 1, + RetryPeriod: time.Minute, } } @@ -70,7 +75,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) { } // Create a client to the server. - client, err := commandClient(cmd) + client, err := commandClient(cmd, http.WithClientRetryPeriod(cmd.RetryPeriod)) if err != nil { return fmt.Errorf("creating client: %w", err) } @@ -262,7 +267,7 @@ func (cmd *BackupCommand) backupShardNode(ctx context.Context, indexName string, logger := cmd.Logger() logger.Printf("backing up shard: index=%q id=%d", indexName, shard) - client := http.NewInternalClientFromURI(&node.URI, http.GetHTTPClient(cmd.tlsConfig)) + client := http.NewInternalClientFromURI(&node.URI, http.GetHTTPClient(cmd.tlsConfig), http.WithClientRetryPeriod(cmd.RetryPeriod)) rc, err := client.ShardReader(ctx, indexName, shard) if err != nil { return fmt.Errorf("fetching shard reader: %w", err) diff --git a/ctl/common.go b/ctl/common.go index c23b42f4c..7ba2df2f7 100644 --- a/ctl/common.go +++ b/ctl/common.go @@ -26,13 +26,13 @@ func SetTLSConfig(flags *pflag.FlagSet, prefix string, certificatePath *string, } // commandClient returns a pilosa.InternalHTTPClient for the command -func commandClient(cmd CommandWithTLSSupport) (*http.InternalClient, error) { +func commandClient(cmd CommandWithTLSSupport, opts ...http.InternalClientOption) (*http.InternalClient, error) { tls := cmd.TLSConfiguration() tlsConfig, err := server.GetTLSConfig(&tls, cmd.Logger()) if err != nil { return nil, errors.Wrap(err, "getting tls config") } - client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig)) + client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig), opts...) if err != nil { return nil, errors.Wrap(err, "getting internal client") } diff --git a/http/client.go b/http/client.go index 70c0639ec..8590a97fa 100644 --- a/http/client.go +++ b/http/client.go @@ -11,6 +11,7 @@ import ( "math/rand" "net/http" "net/url" + "os" "path" "sort" "strconv" @@ -20,6 +21,7 @@ import ( pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/encoding/proto" "github.com/molecula/featurebase/v2/ingest" + "github.com/molecula/featurebase/v2/logger" pnet "github.com/molecula/featurebase/v2/net" "github.com/molecula/featurebase/v2/topology" "github.com/molecula/featurebase/v2/tracing" @@ -31,6 +33,10 @@ type InternalClient struct { defaultURI *pnet.URI serializer pilosa.Serializer + log logger.Logger + + retryPeriod time.Duration + // The client to use for HTTP communication. httpClient *http.Client // the local node's API, used for operations that we can short-circuit that way @@ -40,7 +46,7 @@ type InternalClient struct { // NewInternalClient returns a new instance of InternalClient to connect to host. // If api is non-nil, the client uses it for some same-host operations instead // of going through http. -func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient, error) { +func NewInternalClient(host string, remoteClient *http.Client, opts ...InternalClientOption) (*InternalClient, error) { if host == "" { return nil, pilosa.ErrHostRequired } @@ -50,16 +56,38 @@ func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient, return nil, errors.Wrap(err, "getting URI") } - client := NewInternalClientFromURI(uri, remoteClient) + client := NewInternalClientFromURI(uri, remoteClient, opts...) return client, nil } -func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client) *InternalClient { - return &InternalClient{ +type InternalClientOption func(c *InternalClient) + +// WithClientRetryPeriod is the max amount of total time the client will +// retry failed requests using exponential backoff. +func WithClientRetryPeriod(period time.Duration) InternalClientOption { + return func(c *InternalClient) { + c.retryPeriod = period + } +} + +func WithClientLogger(log logger.Logger) InternalClientOption { + return func(c *InternalClient) { + c.log = log + } +} + +func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, opts ...InternalClientOption) *InternalClient { + ic := &InternalClient{ defaultURI: defaultURI, serializer: proto.Serializer{}, httpClient: remoteClient, + log: logger.NewStandardLogger(os.Stderr), } + + for _, opt := range opts { + opt(ic) + } + return ic } // MaxShardByIndex returns the number of shards on a server by index. @@ -1717,6 +1745,36 @@ func giveRawResponse(b bool) executeRequestOption { } } +func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error) { + sleepDuration := time.Second + resp, err := c.httpClient.Do(req) + // start timer after first request, so if retryPeriod > 0 we + // pretty much always do at least one retry + start := time.Now() + for ; ; resp, err = c.httpClient.Do(req) { + if err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300 { + if time.Since(start) > c.retryPeriod { + break + } + if err != nil { + c.log.Printf("retrying request due to error: '%v'", err) + } else { + if bod, readErr := ioutil.ReadAll(resp.Body); readErr != nil { + c.log.Printf("retrying request due to status: %d, error reading body: '%v', body: '%s'", resp.StatusCode, readErr, bod) + } else { + c.log.Printf("retrying request due to status: %d, body: '%s'", resp.StatusCode, bod) + } + } + + time.Sleep(sleepDuration) + sleepDuration *= 2 + } else { + break + } + } + return resp, err +} + // executeRequest executes the given request and checks the Response. For // responses with non-2XX status, the body is read and closed, and an error is // returned. If the error is nil, the caller must ensure that the response body @@ -1729,7 +1787,7 @@ func (c *InternalClient) executeRequest(req *http.Request, opts ...executeReques tracing.GlobalTracer.InjectHTTPHeaders(req) req.Close = false - resp, err := c.httpClient.Do(req) + resp, err := c.doWithRetry(req) if err != nil { if resp != nil { resp.Body.Close()