diff --git a/ctl/common.go b/ctl/common.go index 7ba2df2f7..558cdece3 100644 --- a/ctl/common.go +++ b/ctl/common.go @@ -2,6 +2,11 @@ package ctl import ( + "net" + "time" + + gohttp "net/http" + "github.com/molecula/featurebase/v2/http" "github.com/molecula/featurebase/v2/logger" "github.com/molecula/featurebase/v2/server" @@ -25,6 +30,14 @@ func SetTLSConfig(flags *pflag.FlagSet, prefix string, certificatePath *string, flags.BoolVarP(enableClientVerification, prefix+"tls.enable-client-verification", "", false, "Enable TLS certificate client verification for incoming connections") } +// default dial timeout is 30s for some reason which makes testing +// failures/retries really awkward. I don't think we need it that +// high, so I set it to 1s here... let's see what happens. +func clientOptions(client *gohttp.Client, dialer *net.Dialer) *gohttp.Client { + dialer.Timeout = time.Second * 1 + return client +} + // commandClient returns a pilosa.InternalHTTPClient for the command func commandClient(cmd CommandWithTLSSupport, opts ...http.InternalClientOption) (*http.InternalClient, error) { tls := cmd.TLSConfiguration() @@ -32,7 +45,7 @@ func commandClient(cmd CommandWithTLSSupport, opts ...http.InternalClientOption) if err != nil { return nil, errors.Wrap(err, "getting tls config") } - client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig), opts...) + client, err := http.NewInternalClient(cmd.TLSHost(), http.GetHTTPClient(tlsConfig, clientOptions), opts...) if err != nil { return nil, errors.Wrap(err, "getting internal client") } diff --git a/ctl/restore.go b/ctl/restore.go index f7d669f09..b7f1fb2dc 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "io" + "math" "net/http" "os" "path/filepath" @@ -137,8 +138,7 @@ func (cmd *RestoreCommand) restoreSchema(ctx context.Context, primary *topology. if len(existingSchema) == 0 { cmd.Logger().Printf("Load Schema") url := primary.URI.Path("/schema") - client := retryablehttp.NewClient() - client.RetryWaitMax = cmd.RetryPeriod + client := cmd.newClient() _, err = client.Post(url, "application/json", f) } else { schema := &pilosa.Schema{} @@ -185,6 +185,27 @@ func retryWith400(ctx context.Context, resp *http.Response, err error) (bool, er return retryablehttp.DefaultRetryPolicy(ctx, resp, err) } +// This logic is taken from featurebase/http/client.go If this logic +// is not the same as what's there, that could be a problem. Ideally +// all network calls from restore would go through the client and this +// would not longer be needed. +func (cmd *RestoreCommand) newClient() *retryablehttp.Client { + min := time.Millisecond * 100 + + // do some math to figure out how many attempts we need to get our + // total sleep time close to the period + attempts := math.Log2(float64(cmd.RetryPeriod)) - math.Log2(float64(min)) + attempts += 0.3 // mmmm, fudge + if attempts < 1 { + attempts = 1 + } + client := retryablehttp.NewClient() + client.RetryWaitMin = min + client.RetryMax = int(attempts) + client.CheckRetry = retryWith400 + return client +} + func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology.Node) error { logger := cmd.Logger() @@ -200,9 +221,7 @@ func (cmd *RestoreCommand) restoreIDAlloc(ctx context.Context, primary *topology logger.Printf("Load idalloc") url := primary.URI.Path("/internal/idalloc/restore") - client := retryablehttp.NewClient() - client.RetryWaitMax = cmd.RetryPeriod - client.CheckRetry = retryWith400 + client := cmd.newClient() _, err = client.Post(url, "application/octet-stream", f) return err } @@ -279,9 +298,7 @@ func (cmd *RestoreCommand) restoreShard(ctx context.Context, filename string) er req = req.WithContext(ctx) req.Header.Set("Content-Type", "application/octet-stream") - client := retryablehttp.NewClient() - client.RetryWaitMax = cmd.RetryPeriod - client.CheckRetry = retryWith400 + client := cmd.newClient() resp, err := client.Do(req) if err != nil { return err diff --git a/http/client.go b/http/client.go index 222499794..8d3437bb5 100644 --- a/http/client.go +++ b/http/client.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "io/ioutil" + "math" "math/rand" "net/http" "net/url" @@ -64,12 +65,22 @@ 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 { +func WithClientRetryPeriod(period time.Duration) InternalClientOption { + min := time.Millisecond * 100 + + // do some math to figure out how many attempts we need to get our + // total sleep time close to the period + attempts := math.Log2(float64(period)) - math.Log2(float64(min)) + attempts += 0.3 // mmmm, fudge + if attempts < 1 { + attempts = 1 + } + fmt.Println("attempts: ", int(attempts)) return func(c *InternalClient) { - fmt.Println("client w/ retry policy", waitMax) rc := retryablehttp.NewClient() rc.HTTPClient = c.httpClient - rc.RetryWaitMax = waitMax + rc.RetryWaitMin = min + rc.RetryMax = int(attempts) rc.CheckRetry = retryWith400Policy c.retryableClient = rc } @@ -109,7 +120,6 @@ func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client, o } if ic.retryableClient == nil { - fmt.Println("no retry policy") rc := retryablehttp.NewClient() rc.HTTPClient = ic.httpClient rc.CheckRetry = noRetryPolicy diff --git a/http/handler.go b/http/handler.go index 05b65aa58..7ef5eda37 100644 --- a/http/handler.go +++ b/http/handler.go @@ -2593,14 +2593,17 @@ func (s queryValidationSpec) validate(query url.Values) error { return nil } -func GetHTTPClient(t *tls.Config) *http.Client { +type ClientOption func(client *http.Client, dialer *net.Dialer) *http.Client + +func GetHTTPClient(t *tls.Config, opts ...ClientOption) *http.Client { + dialer := &net.Dialer{ + Timeout: 30 * time.Second, + KeepAlive: 30 * time.Second, + DualStack: true, + } transport := &http.Transport{ - Proxy: http.ProxyFromEnvironment, - DialContext: (&net.Dialer{ - Timeout: 30 * time.Second, - KeepAlive: 30 * time.Second, - DualStack: true, - }).DialContext, + Proxy: http.ProxyFromEnvironment, + DialContext: dialer.DialContext, MaxIdleConns: 1000, MaxIdleConnsPerHost: 200, IdleConnTimeout: 90 * time.Second, @@ -2610,7 +2613,12 @@ func GetHTTPClient(t *tls.Config) *http.Client { if t != nil { transport.TLSClientConfig = t } - return &http.Client{Transport: transport} + + client := &http.Client{Transport: transport} + for _, opt := range opts { + client = opt(client, dialer) + } + return client } // handlePostImportAtomicRecord handles /import-atomic-record requests diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index af89abc2a..e23d10f8c 100644 --- a/internal/clustertests/cluster_test.go +++ b/internal/clustertests/cluster_test.go @@ -149,22 +149,23 @@ func TestClusterStuff(t *testing.T) { // 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 { + "featurebase", "backup", "--host=pilosa1:10101", fmt.Sprintf("--output=%s", tmpdir+"/backuptest2"), "--retry-period=200ms"); err != nil { t.Fatalf("sending second backup command: %v", err) } - time.Sleep(time.Millisecond * 5) // want the backup to get started, then fail + time.Sleep(time.Millisecond * 10) // 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_pilosa2_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) }