scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 00:55:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1807 from molecula/backup-http-retry

Browse source 
add exponential retry logic to internal http client, use in backup and restore
This commit is contained in:
Matthew Jaffee 2021-12-22 13:00:24 -06:00 • committed by GitHub
commit f1b525fc74
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 370 additions and 94 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.circleci/config.yml
View file

@ -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

8
Makefile
View file

@ -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:

14
client.go
View file

@ -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
}

12
cmd/backup.go
View file

@ -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
}

2
cmd/restore.go
View file

@ -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,

12
cmd/slurp/slurp.go
View file

@ -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
}

23
ctl/backup.go
View file

@ -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)

4
ctl/common.go
View file

@ -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")
}

58
ctl/restore.go
View file

@ -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
}

56
ctl/util.go Normal file
View file

@ -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
}

1
go.mod
View file

@ -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

5
go.sum
View file

@ -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=

101
http/client.go
View file

@ -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
}

2
http/handler.go
View file

@ -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
}

157
internal/clustertests/cluster_test.go
View file

@ -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) {

7
internal/clustertests/pause_node_test.go
View file

@ -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")
}