From 8486efaa79034398d8a27b38ed568d1145bae23f Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Mon, 13 Dec 2021 12:27:44 -0600 Subject: [PATCH 01/10] 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() From cdf4bc4c88272ca61dd9e51aee19e9298b5ab11d Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Mon, 13 Dec 2021 16:43:49 -0600 Subject: [PATCH 02/10] add clustertests testing backup's retry --- .circleci/config.yml | 2 +- Makefile | 8 +- http/client.go | 32 +++--- internal/clustertests/cluster_test.go | 132 ++++++++++++++++------- internal/clustertests/pause_node_test.go | 7 +- 5 files changed, 118 insertions(+), 63 deletions(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index c3e1a43e4..f18532e49 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -151,7 +151,7 @@ jobs: - checkout-plus - skip-if-root-unchanged - setup_remote_docker - - run: make clustertests-build + - run: make clustertests release: executor: name: golang diff --git a/Makefile b/Makefile index a24d8c40a..e33c92b56 100644 --- a/Makefile +++ b/Makefile @@ -149,12 +149,10 @@ DOCKER_COMPOSE=internal/clustertests/docker-compose.yml clustertests: vendor docker-compose -f $(DOCKER_COMPOSE) down docker-compose -f $(DOCKER_COMPOSE) build - docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1 + docker-compose -f $(DOCKER_COMPOSE) up -d pilosa1 pilosa2 pilosa3 + docker-compose -f $(DOCKER_COMPOSE) run client1 + docker-compose -f $(DOCKER_COMPOSE) down -# Like clustertests, but rebuilds all images. -clustertests-build: vendor - docker-compose -f $(DOCKER_COMPOSE) down -v - docker-compose -f $(DOCKER_COMPOSE) up --exit-code-from=client1 --build # Install Pilosa install: diff --git a/http/client.go b/http/client.go index 8590a97fa..bb08478d6 100644 --- a/http/client.go +++ b/http/client.go @@ -1751,26 +1751,22 @@ func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error) // 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 { + for ; err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300; resp, err = c.httpClient.Do(req) { + 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 } return resp, err } diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index 2fca2f1ab..00967ac07 100644 --- a/internal/clustertests/cluster_test.go +++ b/internal/clustertests/cluster_test.go @@ -3,6 +3,8 @@ package clustertest import ( "context" + "fmt" + "io/ioutil" "os" "os/exec" "testing" @@ -30,56 +32,53 @@ func TestClusterStuff(t *testing.T) { t.Fatalf("getting client: %v", err) } - t.Run("long pause", func(t *testing.T) { - err := cli1.CreateIndex(context.Background(), "testidx", pilosa.IndexOptions{}) - if err != nil { - t.Fatalf("creating index: %v", err) - } - err = cli1.CreateFieldWithOptions(context.Background(), "testidx", "testf", pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: 100}) - if err != nil { - t.Fatalf("creating field: %v", err) - } + if err := cli1.CreateIndex(context.Background(), "testidx", pilosa.IndexOptions{}); err != nil { + t.Fatalf("creating index: %v", err) + } + if err := cli1.CreateFieldWithOptions(context.Background(), "testidx", "testf", pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked, CacheSize: 100}); err != nil { + t.Fatalf("creating field: %v", err) + } - req := &pilosa.ImportRequest{ - Index: "testidx", - Field: "testf", - } - req.ColumnIDs = make([]uint64, 10) - req.RowIDs = make([]uint64, 10) + req := &pilosa.ImportRequest{ + Index: "testidx", + Field: "testf", + } + req.ColumnIDs = make([]uint64, 10) + req.RowIDs = make([]uint64, 10) - for i := 0; i < 1000; i++ { - req.RowIDs[i%10] = 0 - req.ColumnIDs[i%10] = uint64((i/10)*pilosa.ShardWidth + i%10) - req.Shard = uint64(i / 10) - if i%10 == 9 { - err = cli1.Import(context.Background(), nil, req, &pilosa.ImportOptions{}) - if err != nil { - t.Fatalf("importing: %v", err) - } - } - } - - // Check query results from each node. - for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} { - r, err := cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"}) + for i := 0; i < 1000; i++ { + req.RowIDs[i%10] = 0 + req.ColumnIDs[i%10] = uint64((i/10)*pilosa.ShardWidth + i%10) + req.Shard = uint64(i / 10) + if i%10 == 9 { + err = cli1.Import(context.Background(), nil, req, &pilosa.ImportOptions{}) if err != nil { - t.Fatalf("count querying pilosa%d: %v", i, err) - } - if r.Results[0].(uint64) != 1000 { - t.Fatalf("count on pilosa%d after import is %d", i, r.Results[0].(uint64)) + t.Fatalf("importing: %v", err) } } + } + + // Check query results from each node. + for i, cli := range []*picli.InternalClient{cli1, cli2, cli3} { + r, err := cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"}) + if err != nil { + t.Fatalf("count querying pilosa%d: %v", i, err) + } + if r.Results[0].(uint64) != 1000 { + t.Fatalf("count on pilosa%d after import is %d", i, r.Results[0].(uint64)) + } + } + t.Run("long pause", func(t *testing.T) { pcmd := exec.Command("/pumba", "pause", "clustertests_pilosa3_1", "--duration", "10s") pcmd.Stdout = os.Stdout pcmd.Stderr = os.Stderr t.Log("pausing pilosa3 for 10s") - err = pcmd.Start() - if err != nil { + + if err := pcmd.Start(); err != nil { t.Fatalf("starting pumba command: %v", err) } - err = pcmd.Wait() - if err != nil { + if err := pcmd.Wait(); err != nil { t.Fatalf("waiting on pumba pause cmd: %v", err) } @@ -98,6 +97,63 @@ func TestClusterStuff(t *testing.T) { } } }) + + t.Run("backup", func(t *testing.T) { + // do backup with node 1 down, but restart it after a few seconds + if err := sendCmd("docker", "stop", "clustertests_pilosa1_1"); err != nil { + t.Fatalf("sending stop command: %v", err) + } + var backupCmd *exec.Cmd + tmpdir, err := ioutil.TempDir("", "") + if err != nil { + t.Fatalf("getting tmp dir: %v", err) + } + if backupCmd, err = startCmd( + "featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest")); err != nil { + t.Fatalf("sending backup command: %v", err) + } + time.Sleep(time.Second * 5) + if err = sendCmd("docker", "start", "clustertests_pilosa1_1"); err != nil { + t.Fatalf("sending start command: %v", err) + } + + if err = backupCmd.Wait(); err != nil { + t.Fatalf("waiting on backup to finish: %v", err) + } + + // now do backup with all nodes down and too short a timeout + // so it fails. Has be to be all 3 because the cluster has + // replicas=3 and the backup command will retry on replicas. + if err = sendCmd("docker", "stop", "clustertests_pilosa2_1"); err != nil { + t.Fatalf("sending stop command: %v", err) + } + if backupCmd, err = startCmd( + "featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest2"), "--retry-period=0.5s"); err != nil { + t.Fatalf("sending second backup command: %v", err) + } + time.Sleep(time.Millisecond * 5) // want the backup to get started, then fail + if err = sendCmd("docker", "stop", "clustertests_pilosa1_1"); err != nil { + t.Fatalf("sending stop command: %v", err) + } + if err = sendCmd("docker", "stop", "clustertests_pilosa3_1"); err != nil { + t.Fatalf("sending stop command: %v", err) + } + + time.Sleep(time.Second * 5) + if err = sendCmd("docker", "start", "clustertests_pilosa1_1"); err != nil { + t.Fatalf("sending start command: %v", err) + } + if err = sendCmd("docker", "start", "clustertests_pilosa2_1"); err != nil { + t.Fatalf("sending start command: %v", err) + } + if err = sendCmd("docker", "start", "clustertests_pilosa3_1"); err != nil { + t.Fatalf("sending start command: %v", err) + } + if err = backupCmd.Wait(); err == nil { + t.Fatal("backup command should have errored but didn't") + } + + }) } func waitForStatus(t *testing.T, stator func(context.Context) (string, error), status string, n int, sleep time.Duration) { diff --git a/internal/clustertests/pause_node_test.go b/internal/clustertests/pause_node_test.go index 1cfa81417..256078a90 100644 --- a/internal/clustertests/pause_node_test.go +++ b/internal/clustertests/pause_node_test.go @@ -23,11 +23,16 @@ import ( "github.com/pkg/errors" ) -func sendCmd(cmd string, args ...string) error { +func startCmd(cmd string, args ...string) (*exec.Cmd, error) { pcmd := exec.Command(cmd, args...) pcmd.Stdout = os.Stdout pcmd.Stderr = os.Stderr err := pcmd.Start() + return pcmd, err +} + +func sendCmd(cmd string, args ...string) error { + pcmd, err := startCmd(cmd, args...) if err != nil { return errors.Wrap(err, "starting cmd") } From 3105a24542a6334608ee7bf82a59ecf66ab99a83 Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Mon, 20 Dec 2021 17:32:52 -0600 Subject: [PATCH 03/10] rewind Body on retry this is really not ideal, and there are libraries for this kind of thing, but I'd have to figure out how to make the libraries work with everywhere we're already creating stdlib http clients. --- http/client.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/http/client.go b/http/client.go index bb08478d6..0e2b1b428 100644 --- a/http/client.go +++ b/http/client.go @@ -1745,13 +1745,33 @@ func giveRawResponse(b bool) executeRequestOption { } } +type nopCloser struct { + *bytes.Reader +} + +func (n nopCloser) Close() error { + return nil +} + func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error) { sleepDuration := time.Second + newBody := nopCloser{} + if req.Body != nil { + bod, err := ioutil.ReadAll(req.Body) + if err != nil { + return nil, errors.Wrap(err, "reading body") + } + newBody.Reader = bytes.NewReader(bod) + req.Body = newBody + } 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 ; err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300; resp, err = c.httpClient.Do(req) { + if newBody.Reader != nil { + newBody.Seek(0, io.SeekStart) + } if time.Since(start) > c.retryPeriod { break } From f676fbfc5151c68478c46c9f18da97ede2ac2967 Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Mon, 20 Dec 2021 17:42:22 -0600 Subject: [PATCH 04/10] add retryability to restore command --- cmd/restore.go | 1 + ctl/restore.go | 21 ++++++++++++++++----- go.mod | 1 + go.sum | 5 +++++ 4 files changed, 23 insertions(+), 5 deletions(-) diff --git a/cmd/restore.go b/cmd/restore.go index e9af62d24..71fb5ec7d 100644 --- a/cmd/restore.go +++ b/cmd/restore.go @@ -25,6 +25,7 @@ The Restore command will take a backup archive and restore it to a new, clean cl flags.StringVarP(&cmd.Path, "source", "s", "", "backup file; specify '-' to restore from stdin tar stream") flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.") flags.IntVar(&cmd.Concurrency, "concurrency", 1, "number of concurrent uploads") + 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, diff --git a/ctl/restore.go b/ctl/restore.go index 373581d5d..0804c3fa2 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -13,8 +13,12 @@ import ( "path/filepath" "strconv" "strings" + "time" + + "github.com/hashicorp/go-retryablehttp" pilosa "github.com/molecula/featurebase/v2" + fb_http "github.com/molecula/featurebase/v2/http" "github.com/molecula/featurebase/v2/server" "github.com/molecula/featurebase/v2/topology" "golang.org/x/sync/errgroup" @@ -29,6 +33,10 @@ type RestoreCommand struct { // Filepath to the backup file. Path string + + // Amount of time after first failed request to continue retrying. + RetryPeriod time.Duration `json:"retry-period"` + // Reusable client. client pilosa.InternalClient @@ -62,7 +70,7 @@ func (cmd *RestoreCommand) Run(ctx context.Context) (err error) { return fmt.Errorf("parsing tls config: %w", err) } // Create a client to the server. - client, err := commandClient(cmd) + client, err := commandClient(cmd, fb_http.WithClientRetryPeriod(cmd.RetryPeriod)) if err != nil { return fmt.Errorf("creating client: %w", err) } @@ -119,7 +127,8 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology. if len(existingSchema) == 0 { cmd.Logger().Printf("Load Schema") url := primary.URI.Path("/schema") - var client http.Client + client := retryablehttp.NewClient() + client.RetryWaitMax = cmd.RetryPeriod _, err = client.Post(url, "application/json", f) } else { schema := &pilosa.Schema{} @@ -174,7 +183,8 @@ func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology logger.Printf("Load idalloc") url := primary.URI.Path("/internal/idalloc/restore") - var client http.Client + client := retryablehttp.NewClient() + client.RetryWaitMax = cmd.RetryPeriod _, err = client.Post(url, "application/octet-stream", f) return err } @@ -244,14 +254,15 @@ func (cmd *RestoreCommand) restoreShard(ctx context.Context, filename string) er defer f.Close() url := node.URI.Path(fmt.Sprintf("/internal/restore/%v/%v", indexName, shard)) - req, err := http.NewRequest("POST", url, f) + req, err := retryablehttp.NewRequest("POST", url, f) if err != nil { return err } req = req.WithContext(ctx) req.Header.Set("Content-Type", "application/octet-stream") - var client http.Client + client := retryablehttp.NewClient() + client.RetryWaitMax = cmd.RetryPeriod resp, err := client.Do(req) if err != nil { return err diff --git a/go.mod b/go.mod index b66980688..58b7e60c2 100644 --- a/go.mod +++ b/go.mod @@ -25,6 +25,7 @@ require ( github.com/gopherjs/gopherjs v0.0.0-20200217142428-fce0ec30dd00 // indirect github.com/gorilla/handlers v1.3.0 github.com/gorilla/mux v1.7.0 + github.com/hashicorp/go-retryablehttp v0.7.0 github.com/improbable-eng/grpc-web v0.13.0 github.com/lib/pq v1.8.0 github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b diff --git a/go.sum b/go.sum index 4224f965d..a60699bb5 100644 --- a/go.sum +++ b/go.sum @@ -186,10 +186,15 @@ github.com/grpc-ecosystem/grpc-gateway v1.9.5/go.mod h1:vNeuVxBJEsws4ogUvrchl83t github.com/hashicorp/consul/api v1.1.0/go.mod h1:VmuI/Lkw1nC05EYQWNKwWGbkg+FbDBtguAZLlVdkD9Q= github.com/hashicorp/consul/sdk v0.1.1/go.mod h1:VKf9jXwCTEY1QZP2MOLRhb5i/I/ssyNV1vwHyQBF0x8= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= +github.com/hashicorp/go-cleanhttp v0.5.1 h1:dH3aiDG9Jvb5r5+bYHsikaOUIpcM0xvgMXVoDkXMzJM= github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80= +github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI= +github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ= github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60= github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM= github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk= +github.com/hashicorp/go-retryablehttp v0.7.0 h1:eu1EI/mbirUgP5C8hVsTNaGZreBDlYiwC1FZWkvQPQ4= +github.com/hashicorp/go-retryablehttp v0.7.0/go.mod h1:vAew36LZh98gCBJNLH42IQ1ER/9wtLZZ8meHqQvEYWY= github.com/hashicorp/go-rootcerts v1.0.0/go.mod h1:K6zTfqpRlCUIjkwsN4Z+hiSfzSTQa6eBIzfwKfwNnHU= github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU= github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4= From d3b9193c8d2cf29b135910b0990490d42c86b95f Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Mon, 20 Dec 2021 21:48:45 -0600 Subject: [PATCH 05/10] try to fix data race with http lib WARNING: DATA RACE Write at 0x00c008121e80 by goroutine 235: bytes.(*Reader).WriteTo() /usr/local/go/src/bytes/reader.go:139 +0x45 github.com/molecula/featurebase/v2/http.nopCloser.WriteTo() :1 +0x5d io.copyBuffer() /usr/local/go/src/io/io.go:391 +0x482 io.Copy() /usr/local/go/src/io/io.go:368 +0x78 net/http.(*transferWriter).doBodyCopy() /usr/local/go/src/net/http/transfer.go:400 +0x2f net/http.(*transferWriter).writeBody() /usr/local/go/src/net/http/transfer.go:364 +0xc9a net/http.(*Request).write() /usr/local/go/src/net/http/request.go:682 +0x887 net/http.(*persistConn).writeLoop() /usr/local/go/src/net/http/transport.go:2343 +0x349 Previous write at 0x00c008121e80 by goroutine 192: bytes.(*Reader).Seek() /usr/local/go/src/bytes/reader.go:118 +0x824 github.com/molecula/featurebase/v2/http.(*InternalClient).doWithRetry() /go/src/github.com/molecula/featurebase/http/client.go:1773 +0x86d github.com/molecula/featurebase/v2/http.(*InternalClient).executeRequest() /go/src/github.com/molecula/featurebase/http/client.go:1806 +0x15b github.com/molecula/featurebase/v2/http.(*InternalClient).CreateIndex() /go/src/github.com/molecula/featurebase/http/client.go:433 +0xbf8 github.com/molecula/featurebase/v2/server_test.TestMain_Set_Quick.func1() /go/src/github.com/molecula/featurebase/server/server_test.go:64 +0x624 testing.tRunner() /usr/local/go/src/testing/testing.go:1123 +0x202 Goroutine 235 (running) created at: net/http.(*Transport).dialConn() /usr/local/go/src/net/http/transport.go:1709 +0xc30 net/http.(*Transport).dialConnFor() /usr/local/go/src/net/http/transport.go:1421 +0x151 Goroutine 192 (running) created at: testing.(*T).Run() /usr/local/go/src/testing/testing.go:1168 +0x5bb github.com/molecula/featurebase/v2/server_test.TestMain_Set_Quick() /go/src/github.com/molecula/featurebase/server/server_test.go:45 +0x116 testing.tRunner() /usr/local/go/src/testing/testing.go:1123 +0x202 --- http/client.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/http/client.go b/http/client.go index 0e2b1b428..c32dd86ae 100644 --- a/http/client.go +++ b/http/client.go @@ -1755,22 +1755,22 @@ func (n nopCloser) Close() error { func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error) { sleepDuration := time.Second - newBody := nopCloser{} + var bod []byte + var err error if req.Body != nil { - bod, err := ioutil.ReadAll(req.Body) + bod, err = ioutil.ReadAll(req.Body) if err != nil { return nil, errors.Wrap(err, "reading body") } - newBody.Reader = bytes.NewReader(bod) - req.Body = newBody + req.Body = nopCloser{bytes.NewReader(bod)} } 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 ; err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300; resp, err = c.httpClient.Do(req) { - if newBody.Reader != nil { - newBody.Seek(0, io.SeekStart) + if req.Body != nil { + req.Body = nopCloser{bytes.NewReader(bod)} // can't seek due to races with http lib internals } if time.Since(start) > c.retryPeriod { break From 2bce39644554b3a7eee8ec6af23511ebd55f64d2 Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Mon, 20 Dec 2021 22:16:06 -0600 Subject: [PATCH 06/10] add retry restore test and custom retry policy --- ctl/restore.go | 10 ++++++++ internal/clustertests/cluster_test.go | 35 +++++++++++++++++++++++---- 2 files changed, 40 insertions(+), 5 deletions(-) diff --git a/ctl/restore.go b/ctl/restore.go index 0804c3fa2..cf99619cb 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -49,6 +49,7 @@ type RestoreCommand struct { func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreCommand { return &RestoreCommand{ CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), + RetryPeriod: time.Second * 30, Concurrency: 1, } } @@ -168,6 +169,13 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology. return err } +func RetryWith400(ctx context.Context, resp *http.Response, err error) (bool, error) { + if resp != nil && resp.StatusCode > 400 { // we have some dumb status codes + return true, nil + } + return retryablehttp.DefaultRetryPolicy(ctx, resp, err) +} + func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology.Node) error { logger := cmd.Logger() @@ -185,6 +193,7 @@ func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology client := retryablehttp.NewClient() client.RetryWaitMax = cmd.RetryPeriod + client.CheckRetry = RetryWith400 _, err = client.Post(url, "application/octet-stream", f) return err } @@ -263,6 +272,7 @@ func (cmd *RestoreCommand) restoreShard(ctx context.Context, filename string) er client := retryablehttp.NewClient() client.RetryWaitMax = cmd.RetryPeriod + client.CheckRetry = RetryWith400 resp, err := client.Do(req) if err != nil { return err diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index 00967ac07..af89abc2a 100644 --- a/internal/clustertests/cluster_test.go +++ b/internal/clustertests/cluster_test.go @@ -4,7 +4,7 @@ package clustertest import ( "context" "fmt" - "io/ioutil" + "net/http" "os" "os/exec" "testing" @@ -104,10 +104,7 @@ func TestClusterStuff(t *testing.T) { t.Fatalf("sending stop command: %v", err) } var backupCmd *exec.Cmd - tmpdir, err := ioutil.TempDir("", "") - if err != nil { - t.Fatalf("getting tmp dir: %v", err) - } + tmpdir := t.TempDir() if backupCmd, err = startCmd( "featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest")); err != nil { t.Fatalf("sending backup command: %v", err) @@ -121,6 +118,34 @@ func TestClusterStuff(t *testing.T) { t.Fatalf("waiting on backup to finish: %v", err) } + fmt.Println("STARTING RESTORE") + + client := http.Client{} + if req, err := http.NewRequest(http.MethodDelete, "http://pilosa1:10101/index/testidx", nil); err != nil { + t.Fatalf("getting req: %v", err) + } else if resp, err := client.Do(req); err != nil { + t.Fatalf("doing request: %v", err) + } else if resp.StatusCode >= 400 { + t.Fatalf("bad response: %v", resp) + } + + var restoreCmd *exec.Cmd + if restoreCmd, err = startCmd("featurebase", "restore", "-s", tmpdir+"/backuptest", "--host", "pilosa1:10101"); err != nil { + t.Fatalf("starting restore: %v", err) + } + time.Sleep(time.Millisecond * 50) + if err = sendCmd("docker", "stop", "clustertests_pilosa2_1"); err != nil { + t.Fatalf("sending stop command: %v", err) + } + + time.Sleep(time.Second * 10) + if err = sendCmd("docker", "start", "clustertests_pilosa2_1"); err != nil { + t.Fatalf("sending stop command: %v", err) + } + if err := restoreCmd.Wait(); err != nil { + t.Fatalf("restore failed: %v", err) + } + // now do backup with all nodes down and too short a timeout // so it fails. Has be to be all 3 because the cluster has // replicas=3 and the backup command will retry on replicas. From cde3f6b5ea5744135410f300e92acc895912f756 Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Tue, 21 Dec 2021 15:34:13 -0600 Subject: [PATCH 07/10] add profiling to backup/restore --- cmd/backup.go | 1 + cmd/restore.go | 1 + ctl/backup.go | 18 +++++++++++++--- ctl/restore.go | 11 +++++++++- ctl/util.go | 56 +++++++++++++++++++++++++++++++++++++++++++++++++ http/handler.go | 2 +- 6 files changed, 84 insertions(+), 5 deletions(-) create mode 100644 ctl/util.go diff --git a/cmd/backup.go b/cmd/backup.go index 133f0dbce..28712123d 100644 --- a/cmd/backup.go +++ b/cmd/backup.go @@ -29,6 +29,7 @@ Backs up a FeatureBase server to a local, tar-formatted snapshot file. 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.") + flags.StringVar(&cmd.Pprof, "pprof", cmd.Pprof, "host:port to listen for profiling requests at /debug/pprof and /debug/fgprof.") ctl.SetTLSConfig(flags, "", &cmd.TLS.CertificatePath, &cmd.TLS.CertificateKeyPath, &cmd.TLS.CACertPath, &cmd.TLS.SkipVerify, &cmd.TLS.EnableClientVerification) return ccmd } diff --git a/cmd/restore.go b/cmd/restore.go index 71fb5ec7d..bc9271d0e 100644 --- a/cmd/restore.go +++ b/cmd/restore.go @@ -26,6 +26,7 @@ The Restore command will take a backup archive and restore it to a new, clean cl flags.StringVar(&cmd.Host, "host", "localhost:10101", "host:port of FeatureBase.") flags.IntVar(&cmd.Concurrency, "concurrency", 1, "number of concurrent uploads") flags.DurationVar(&cmd.RetryPeriod, "retry-period", cmd.RetryPeriod, "Length of time after HTTP request failure to continue retrying request.") + flags.StringVar(&cmd.Pprof, "pprof", cmd.Pprof, "host:port to listen for profiling requests at /debug/pprof and /debug/fgprof.") ctl.SetTLSConfig( flags, "", &cmd.TLS.CertificatePath, diff --git a/ctl/backup.go b/ctl/backup.go index 8ba57c278..302041dfe 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -13,9 +13,10 @@ import ( "time" pilosa "github.com/molecula/featurebase/v2" - "github.com/molecula/featurebase/v2/http" + fb_http "github.com/molecula/featurebase/v2/http" "github.com/molecula/featurebase/v2/server" "github.com/molecula/featurebase/v2/topology" + "github.com/pkg/errors" "golang.org/x/sync/errgroup" ) @@ -41,6 +42,9 @@ type BackupCommand struct { // nolint: maligned // Amount of time after first failed request to continue retrying. RetryPeriod time.Duration `json:"retry-period"` + // Host:port on which to listen for pprof. + Pprof string `json:"pprof"` + // Reusable client. client pilosa.InternalClient @@ -56,11 +60,19 @@ func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), Concurrency: 1, RetryPeriod: time.Minute, + Pprof: "localhost:43809", } } // Run executes the main program execution. func (cmd *BackupCommand) Run(ctx context.Context) (err error) { + logger := cmd.Logger() + close, err := startProfilingServer(cmd.Pprof, logger) + if err != nil { + return errors.Wrap(err, "starting profiling server") + } + defer close() + // Validate arguments. if cmd.OutputDir == "" { return fmt.Errorf("-o flag required") @@ -75,7 +87,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) { } // Create a client to the server. - client, err := commandClient(cmd, http.WithClientRetryPeriod(cmd.RetryPeriod)) + client, err := commandClient(cmd, fb_http.WithClientRetryPeriod(cmd.RetryPeriod)) if err != nil { return fmt.Errorf("creating client: %w", err) } @@ -267,7 +279,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), http.WithClientRetryPeriod(cmd.RetryPeriod)) + client := fb_http.NewInternalClientFromURI(&node.URI, fb_http.GetHTTPClient(cmd.tlsConfig), fb_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/restore.go b/ctl/restore.go index cf99619cb..589eb8138 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -5,7 +5,6 @@ import ( "context" "crypto/tls" "encoding/json" - "errors" "fmt" "io" "net/http" @@ -21,6 +20,7 @@ import ( fb_http "github.com/molecula/featurebase/v2/http" "github.com/molecula/featurebase/v2/server" "github.com/molecula/featurebase/v2/topology" + "github.com/pkg/errors" "golang.org/x/sync/errgroup" ) @@ -37,6 +37,9 @@ type RestoreCommand struct { // Amount of time after first failed request to continue retrying. RetryPeriod time.Duration `json:"retry-period"` + // Host:port on which to listen for pprof. + Pprof string `json:"pprof"` + // Reusable client. client pilosa.InternalClient @@ -51,12 +54,18 @@ func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreComman CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), RetryPeriod: time.Second * 30, Concurrency: 1, + Pprof: "localhost:43809", } } // Run executes the restore. func (cmd *RestoreCommand) Run(ctx context.Context) (err error) { logger := cmd.Logger() + close, err := startProfilingServer(cmd.Pprof, logger) + if err != nil { + return errors.Wrap(err, "starting profiling server") + } + defer close() // Validate arguments. if cmd.Path == "" { diff --git a/ctl/util.go b/ctl/util.go new file mode 100644 index 000000000..60ac082c9 --- /dev/null +++ b/ctl/util.go @@ -0,0 +1,56 @@ +package ctl + +import ( + "context" + "net" + "net/http" + "net/http/pprof" + "runtime" + "time" + + "github.com/felixge/fgprof" + "github.com/molecula/featurebase/v2/logger" + "github.com/pkg/errors" +) + +// startProfilingServer starts a server which handles /debug/pprof and +// /debug/fgprof for use in utilities we might want to profile but +// wouldn't otherwise be running an http server. Caller should call +// the returned close function before exiting to release resources. +func startProfilingServer(addr string, logger logger.Logger) (close func() error, err error) { + if addr == "" { + return func() error { return nil }, nil + } + + sm := http.NewServeMux() + sm.Handle("/debug/fgprof", fgprof.Handler()) + sm.HandleFunc("/debug/pprof/", pprof.Index) + sm.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) + sm.HandleFunc("/debug/pprof/profile", pprof.Profile) + sm.HandleFunc("/debug/pprof/symbol", pprof.Symbol) + sm.HandleFunc("/debug/pprof/trace", pprof.Trace) + s := &http.Server{ + Addr: addr, + Handler: sm, + } + runtime.SetBlockProfileRate(10000000) // 1 sample per 10 ms + runtime.SetMutexProfileFraction(100) // 1% sampling + ln, err := net.Listen("tcp", addr) + if err != nil { + return nil, err + } + go func() { + logger.Printf("Listening for /debug/pprof/ and /debug/fgprof on '%s'", addr) + logger.Printf("%v", s.Serve(ln)) + }() + + return func() error { + ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) + defer cancel() + err := s.Shutdown(ctx) + if err != nil { + return errors.Wrap(err, "shutting down profiling server") + } + return s.Close() + }, nil +} diff --git a/http/handler.go b/http/handler.go index 9e626babf..958f29662 100644 --- a/http/handler.go +++ b/http/handler.go @@ -3348,7 +3348,7 @@ func (h *Handler) handlePostRestore(w http.ResponseWriter, r *http.Request) { //validate shard for this node err = h.api.RestoreShard(ctx, indexName, shard, r.Body) if err != nil { - http.Error(w, fmt.Sprintf("failed to restore shared %v %v err:%v", indexName, shard, err), http.StatusBadRequest) + http.Error(w, fmt.Sprintf("failed to restore shard %v %v err:%v", indexName, shard, err), http.StatusBadRequest) return } From 640ba45129f1df002a2df378fda4ecda50c6aa06 Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Wed, 22 Dec 2021 10:48:10 -0600 Subject: [PATCH 08/10] use retryableHTTP in client, fix memory usage of restore instead of awkwardly reading an entire file into a buffer, we use retryablehttp's reader func to open the file fresh if we need to retry, so a small fixed-size buffer can be used internally for copying the contents onto the network. --- client.go | 14 +++++-- cmd/slurp/slurp.go | 12 ++++-- ctl/restore.go | 22 ++++------ http/client.go | 101 ++++++++++++++++++++++----------------------- 4 files changed, 76 insertions(+), 73 deletions(-) diff --git a/client.go b/client.go index fe91eb124..75741fcd3 100644 --- a/client.go +++ b/client.go @@ -80,8 +80,14 @@ type InternalClient interface { GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]NodeUsage, error) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error) - ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error - ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error + // ImportFieldKeys and ImportIndexKeys are mainly used when + // restoring a backup. They take a readerFunc which returns a + // reader rather than taking an io.Reader directly to allow for + // efficient retries (rather than reading the entire request body + // into a buffer and reusing it). Reader returned from the func + // must be properly closed by the implementation. + ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error + ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error // SetInternalAPI tells the client the API it should use for internal/loopback ops // where applicable. @@ -277,11 +283,11 @@ func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error) { return nil, nil } -func (c nopInternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error { +func (c nopInternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error { return nil } -func (c nopInternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error { +func (c nopInternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error { return nil } diff --git a/cmd/slurp/slurp.go b/cmd/slurp/slurp.go index 446db53b1..800a5a773 100644 --- a/cmd/slurp/slurp.go +++ b/cmd/slurp/slurp.go @@ -92,8 +92,10 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error { byteData, err := ioutil.ReadAll(tr) vprint.PanicOn(err) - br := bytes.NewReader(byteData) - err = r.client.ImportFieldKeys(context.Background(), uri, index, fieldName, false, br) + readerFunc := func() (io.Reader, error) { + return bytes.NewReader(byteData), nil + } + err = r.client.ImportFieldKeys(context.Background(), uri, index, fieldName, false, readerFunc) if err != nil { return err } @@ -106,9 +108,11 @@ func (r *stateMachine) NewHeader(h *tar.Header, tr *tar.Reader) error { } byteData, err := ioutil.ReadAll(tr) vprint.PanicOn(err) + readerFunc := func() (io.Reader, error) { + return bytes.NewReader(byteData), nil + } - br := bytes.NewReader(byteData) - err = r.client.ImportIndexKeys(context.Background(), uri, index, int(partition), false, br) + err = r.client.ImportIndexKeys(context.Background(), uri, index, int(partition), false, readerFunc) if err != nil { return err } diff --git a/ctl/restore.go b/ctl/restore.go index 589eb8138..d0c0b7ec7 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -178,7 +178,7 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology. return err } -func RetryWith400(ctx context.Context, resp *http.Response, err error) (bool, error) { +func retryWith400(ctx context.Context, resp *http.Response, err error) (bool, error) { if resp != nil && resp.StatusCode > 400 { // we have some dumb status codes return true, nil } @@ -202,7 +202,7 @@ func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology client := retryablehttp.NewClient() client.RetryWaitMax = cmd.RetryPeriod - client.CheckRetry = RetryWith400 + client.CheckRetry = retryWith400 _, err = client.Post(url, "application/octet-stream", f) return err } @@ -281,7 +281,7 @@ func (cmd *RestoreCommand) restoreShard(ctx context.Context, filename string) er client := retryablehttp.NewClient() client.RetryWaitMax = cmd.RetryPeriod - client.CheckRetry = RetryWith400 + client.CheckRetry = retryWith400 resp, err := client.Do(req) if err != nil { return err @@ -349,13 +349,11 @@ func (cmd *RestoreCommand) restoreIndexTranslationFile(ctx context.Context, file for _, node := range nodes { if err := func() error { - f, err := os.Open(filename) - if err != nil { - return err + readerFunc := func() (io.Reader, error) { + return os.Open(filename) // gets used as an HTTP request body and closed by http library } - defer f.Close() - return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, f) + return cmd.client.ImportIndexKeys(ctx, &node.URI, indexName, partitionID, false, readerFunc) }(); err != nil { return err } @@ -410,13 +408,11 @@ func (cmd *RestoreCommand) restoreFieldTranslationFile(ctx context.Context, node for _, node := range nodes { if err := func() error { - f, err := os.Open(filename) - if err != nil { - return err + readerFunc := func() (io.Reader, error) { + return os.Open(filename) } - defer f.Close() - return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, f) + return cmd.client.ImportFieldKeys(ctx, &node.URI, indexName, fieldName, false, readerFunc) }(); err != nil { return err } diff --git a/http/client.go b/http/client.go index c32dd86ae..a3466adac 100644 --- a/http/client.go +++ b/http/client.go @@ -18,6 +18,7 @@ import ( "strings" "time" + "github.com/hashicorp/go-retryablehttp" pilosa "github.com/molecula/featurebase/v2" "github.com/molecula/featurebase/v2/encoding/proto" "github.com/molecula/featurebase/v2/ingest" @@ -35,10 +36,9 @@ type InternalClient struct { log logger.Logger - retryPeriod time.Duration - // The client to use for HTTP communication. - httpClient *http.Client + httpClient *http.Client + retryableClient *retryablehttp.Client // the local node's API, used for operations that we can short-circuit that way api *pilosa.API } @@ -64,9 +64,13 @@ 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 { +func WithClientRetryPeriod(waitMax time.Duration) InternalClientOption { return func(c *InternalClient) { - c.retryPeriod = period + c.retryableClient = &retryablehttp.Client{ + HTTPClient: c.httpClient, + RetryWaitMax: waitMax, + CheckRetry: retryWith400Policy, + } } } @@ -76,6 +80,21 @@ func WithClientLogger(log logger.Logger) InternalClientOption { } } +func noRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool, error) { + return false, nil +} + +// retryWith400Policy wraps retryablehttp's default retry policy to +// also retry on 4XX errors which *should* be client errors and +// therefore useless to retry, but we have some incorrect status codes. +// TODO: fix the incorrect status codes so we can get rid of this. +func retryWith400Policy(ctx context.Context, resp *http.Response, err error) (bool, error) { + if resp != nil && resp.StatusCode > 400 { + return true, nil + } + return retryablehttp.DefaultRetryPolicy(ctx, resp, err) +} + func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, opts ...InternalClientOption) *InternalClient { ic := &InternalClient{ defaultURI: defaultURI, @@ -87,6 +106,13 @@ func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, o for _, opt := range opts { opt(ic) } + + if ic.retryableClient == nil { + ic.retryableClient = &retryablehttp.Client{ + HTTPClient: ic.httpClient, + CheckRetry: noRetryPolicy, + } + } return ic } @@ -1753,57 +1779,28 @@ func (n nopCloser) Close() error { return nil } -func (c *InternalClient) doWithRetry(req *http.Request) (*http.Response, error) { - sleepDuration := time.Second - var bod []byte - var err error - if req.Body != nil { - bod, err = ioutil.ReadAll(req.Body) - if err != nil { - return nil, errors.Wrap(err, "reading body") - } - req.Body = nopCloser{bytes.NewReader(bod)} - } - 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 ; err != nil || resp.StatusCode < 200 || resp.StatusCode >= 300; resp, err = c.httpClient.Do(req) { - if req.Body != nil { - req.Body = nopCloser{bytes.NewReader(bod)} // can't seek due to races with http lib internals - } - 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 - } - 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 // is closed. func (c *InternalClient) executeRequest(req *http.Request, opts ...executeRequestOption) (*http.Response, error) { + return c.executeRetryableRequest(&retryablehttp.Request{Request: req}, opts...) +} + +func (c *InternalClient) executeRetryableRequest(req *retryablehttp.Request, opts ...executeRequestOption) (*http.Response, error) { + tracing.GlobalTracer.InjectHTTPHeaders(req.Request) + req.Close = false eo := &executeOpts{} for _, opt := range opts { opt(eo) } - tracing.GlobalTracer.InjectHTTPHeaders(req) - req.Close = false - resp, err := c.doWithRetry(req) + resp, err := c.retryableClient.Do(req) + + return c.handleResponse(req.Request, eo, resp, err) +} + +func (c *InternalClient) handleResponse(req *http.Request, eo *executeOpts, resp *http.Response, err error) (*http.Response, error) { if err != nil { if resp != nil { resp.Body.Close() @@ -2083,7 +2080,7 @@ func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, return resp.Body, nil } -func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error { +func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, readerFunc func() (io.Reader, error)) error { span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportIndexKeys") defer span.Finish() @@ -2100,14 +2097,14 @@ func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, ind url := fmt.Sprintf("%s/internal/translate/index/%s/%d", uri, index, partitionID) // Generate HTTP request. - httpReq, err := http.NewRequest("POST", url, rddbdata) + httpReq, err := retryablehttp.NewRequest("POST", url, readerFunc) if err != nil { return errors.Wrap(err, "creating request") } httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version) // Execute request against the host. - resp, err := c.executeRequest(httpReq.WithContext(ctx)) + resp, err := c.executeRetryableRequest(httpReq.WithContext(ctx)) if err != nil { return err } @@ -2115,7 +2112,7 @@ func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, ind return nil } -func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error { +func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, readerFunc func() (io.Reader, error)) error { span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportFieldKeys") defer span.Finish() @@ -2132,14 +2129,14 @@ func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, ind url := fmt.Sprintf("%s/internal/translate/field/%s/%s", uri, index, field) // Generate HTTP request. - httpReq, err := http.NewRequest("POST", url, rddbdata) + httpReq, err := retryablehttp.NewRequest("POST", url, readerFunc) if err != nil { return errors.Wrap(err, "creating request") } httpReq.Header.Set("User-Agent", "pilosa/"+pilosa.Version) // Execute request against the host. - resp, err := c.executeRequest(httpReq.WithContext(ctx)) + resp, err := c.executeRetryableRequest(httpReq.WithContext(ctx)) if err != nil { return err } From ea59f14d5025138a91b6bfff154c8f9fa83dd43a Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Wed, 22 Dec 2021 11:21:11 -0600 Subject: [PATCH 09/10] must use retryablehttp.NewClient to get defaults otherwise it won't actually retry :( --- http/client.go | 20 +++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/http/client.go b/http/client.go index a3466adac..7f7374d4d 100644 --- a/http/client.go +++ b/http/client.go @@ -66,11 +66,12 @@ type InternalClientOption func(c *InternalClient) // retry failed requests using exponential backoff. func WithClientRetryPeriod(waitMax time.Duration) InternalClientOption { return func(c *InternalClient) { - c.retryableClient = &retryablehttp.Client{ - HTTPClient: c.httpClient, - RetryWaitMax: waitMax, - CheckRetry: retryWith400Policy, - } + fmt.Println("client w/ retry policy", waitMax) + rc := retryablehttp.NewClient() + rc.HTTPClient = c.httpClient + rc.RetryWaitMax = waitMax + rc.CheckRetry = retryWith400Policy + c.retryableClient = rc } } @@ -108,10 +109,11 @@ func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, o } if ic.retryableClient == nil { - ic.retryableClient = &retryablehttp.Client{ - HTTPClient: ic.httpClient, - CheckRetry: noRetryPolicy, - } + fmt.Println("no retry policy") + rc := retryablehttp.NewClient() + rc.HTTPClient = ic.httpClient + rc.CheckRetry = noRetryPolicy + ic.retryableClient = rc } return ic } From 295fab4892d81336b71c49a6d0a4ec6e790e6e7a Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Wed, 22 Dec 2021 12:21:11 -0600 Subject: [PATCH 10/10] retry on >= 400, not just greater. good catch --- ctl/restore.go | 2 +- http/client.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/ctl/restore.go b/ctl/restore.go index d0c0b7ec7..f7d669f09 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -179,7 +179,7 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology. } func retryWith400(ctx context.Context, resp *http.Response, err error) (bool, error) { - if resp != nil && resp.StatusCode > 400 { // we have some dumb status codes + if resp != nil && resp.StatusCode >= 400 { // we have some dumb status codes return true, nil } return retryablehttp.DefaultRetryPolicy(ctx, resp, err) diff --git a/http/client.go b/http/client.go index 7f7374d4d..222499794 100644 --- a/http/client.go +++ b/http/client.go @@ -90,7 +90,7 @@ func noRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool, e // therefore useless to retry, but we have some incorrect status codes. // TODO: fix the incorrect status codes so we can get rid of this. func retryWith400Policy(ctx context.Context, resp *http.Response, err error) (bool, error) { - if resp != nil && resp.StatusCode > 400 { + if resp != nil && resp.StatusCode >= 400 { return true, nil } return retryablehttp.DefaultRetryPolicy(ctx, resp, err)