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/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/backup.go b/cmd/backup.go index 5761b6d57..28712123d 100644 --- a/cmd/backup.go +++ b/cmd/backup.go @@ -23,11 +23,13 @@ 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.") + 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 e9af62d24..bc9271d0e 100644 --- a/cmd/restore.go +++ b/cmd/restore.go @@ -25,6 +25,8 @@ 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.") + 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/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/backup.go b/ctl/backup.go index 0e8a257f1..302041dfe 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -10,11 +10,13 @@ import ( "io/ioutil" "os" "path/filepath" + "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" ) @@ -37,6 +39,12 @@ 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"` + + // Host:port on which to listen for pprof. + Pprof string `json:"pprof"` + // Reusable client. client pilosa.InternalClient @@ -51,11 +59,20 @@ func NewBackupCommand(stdin io.Reader, stdout, stderr io.Writer) *BackupCommand return &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") @@ -70,7 +87,7 @@ func (cmd *BackupCommand) Run(ctx context.Context) (err error) { } // 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) } @@ -262,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)) + 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/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/ctl/restore.go b/ctl/restore.go index 373581d5d..f7d669f09 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -5,7 +5,6 @@ import ( "context" "crypto/tls" "encoding/json" - "errors" "fmt" "io" "net/http" @@ -13,10 +12,15 @@ 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" + "github.com/pkg/errors" "golang.org/x/sync/errgroup" ) @@ -29,6 +33,13 @@ 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"` + + // Host:port on which to listen for pprof. + Pprof string `json:"pprof"` + // Reusable client. client pilosa.InternalClient @@ -41,13 +52,20 @@ 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, + 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 == "" { @@ -62,7 +80,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 +137,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{} @@ -159,6 +178,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() @@ -174,7 +200,9 @@ 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 + client.CheckRetry = retryWith400 _, err = client.Post(url, "application/octet-stream", f) return err } @@ -244,14 +272,16 @@ 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 + client.CheckRetry = retryWith400 resp, err := client.Do(req) if err != nil { return err @@ -319,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 } @@ -380,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/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/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= diff --git a/http/client.go b/http/client.go index 70c0639ec..222499794 100644 --- a/http/client.go +++ b/http/client.go @@ -11,15 +11,18 @@ import ( "math/rand" "net/http" "net/url" + "os" "path" "sort" "strconv" "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" + "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,8 +34,11 @@ type InternalClient struct { defaultURI *pnet.URI serializer pilosa.Serializer + log logger.Logger + // 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 } @@ -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,66 @@ 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(waitMax time.Duration) InternalClientOption { + return func(c *InternalClient) { + fmt.Println("client w/ retry policy", waitMax) + rc := retryablehttp.NewClient() + rc.HTTPClient = c.httpClient + rc.RetryWaitMax = waitMax + rc.CheckRetry = retryWith400Policy + c.retryableClient = rc + } +} + +func WithClientLogger(log logger.Logger) InternalClientOption { + return func(c *InternalClient) { + c.log = log + } +} + +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, serializer: proto.Serializer{}, httpClient: remoteClient, + log: logger.NewStandardLogger(os.Stderr), } + + for _, opt := range opts { + opt(ic) + } + + if ic.retryableClient == nil { + fmt.Println("no retry policy") + rc := retryablehttp.NewClient() + rc.HTTPClient = ic.httpClient + rc.CheckRetry = noRetryPolicy + ic.retryableClient = rc + } + return ic } // MaxShardByIndex returns the number of shards on a server by index. @@ -1717,19 +1773,36 @@ func giveRawResponse(b bool) executeRequestOption { } } +type nopCloser struct { + *bytes.Reader +} + +func (n nopCloser) Close() error { + return nil +} + // 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.httpClient.Do(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() @@ -2009,7 +2082,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() @@ -2026,14 +2099,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 } @@ -2041,7 +2114,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() @@ -2058,14 +2131,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 } 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 } diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index 2fca2f1ab..af89abc2a 100644 --- a/internal/clustertests/cluster_test.go +++ b/internal/clustertests/cluster_test.go @@ -3,6 +3,8 @@ package clustertest import ( "context" + "fmt" + "net/http" "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,88 @@ 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 := 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) + } + 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) + } + + 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. + 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") }