Merge branch 'master' into fb901

.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

.gitlab/.gitlab-ci.yml

@@ -13,6 +13,7 @@ include:
 variables:
   GOVERSION: "1.16.10"
 
+
 stages:
   - lint
   - test

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:

api.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/molecula/featurebase/v2/disco"
 	"github.com/molecula/featurebase/v2/ingest"
+	"github.com/molecula/featurebase/v2/rbf"
 
 	//"github.com/molecula/featurebase/v2/pg"
 	"github.com/molecula/featurebase/v2/pql"
@@ -130,9 +131,16 @@ func (api *API) SetAPIOptions(opts ...apiOption) error {
 var validAPIMethods = map[disco.ClusterState]map[apiMethod]struct{}{
 	disco.ClusterStateStarting: methodsCommon,
 	disco.ClusterStateNormal:   appendMap(methodsCommon, methodsNormal),
-	disco.ClusterStateDegraded: appendMap(methodsCommon, methodsDegraded),
+	// Ideally, this would be just `appendMap(methodsCommon, methodsDegraded)`,
+	// but in an attempt to reduce the influence that state (determined by etcd)
+	// has on a node under load, this is set to effectively allow all requests
+	// in a DEGRADED state.
+	disco.ClusterStateDegraded: appendMap(methodsCommon, methodsNormal),
 	disco.ClusterStateResizing: appendMap(methodsCommon, methodsResizing),
-	disco.ClusterStateDown:     methodsCommon,
+	// Ideally, this would be just `methodsCommon`, but in an attempt to reduce
+	// the influence that state (determined by etcd) has on a node under load,
+	// this is set to effectively allow all requests in a DOWN state.
+	disco.ClusterStateDown: appendMap(methodsCommon, methodsNormal),
 }
 
 func appendMap(a, b map[apiMethod]struct{}) map[apiMethod]struct{} {
@@ -3156,6 +3164,21 @@ func (api *API) Plan(ctx context.Context, q string) (*Stmt, error) {
 	return api.server.PlanSQL(ctx, q)
 }
 
+func (api *API) RBFDebugInfo() map[string]*rbf.DebugInfo {
+	infos := make(map[string]*rbf.DebugInfo)
+
+	for key, dbShard := range api.holder.Txf().dbPerShard.Flatmap {
+		wrapper, ok := dbShard.W.(*RbfDBWrapper)
+		if !ok {
+			continue
+		}
+
+		skey := fmt.Sprintf("%s/%d", key.index, key.shard)
+		infos[skey] = wrapper.db.DebugInfo()
+	}
+	return infos
+}
+
 type serverInfo struct {
 	ShardWidth       uint64 `json:"shardWidth"`
 	ReplicaN         int    `json:"replicaN"`

api_test.go

@@ -1415,3 +1415,25 @@ func TestVariousApiTranslateCalls(t *testing.T) {
 		*/
 	}
 }
+
+func TestAPI_RBFDebugInfo(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c := test.MustRunCluster(t, 1,
+		[]server.CommandOption{
+			server.OptCommandServerOptions(
+				pilosa.OptServerNodeID("node0"),
+				pilosa.OptServerClusterHasher(&offsetModHasher{}),
+				pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
+			)},
+	)
+	defer c.Close()
+
+	coord := c.GetPrimary()
+
+	if _, err := coord.API.CreateIndex(ctx, "i", pilosa.IndexOptions{}); err != nil {
+		t.Fatal(err)
+	} else if infos := coord.API.RBFDebugInfo(); infos == nil {
+		t.Fatal("expected info")
+	}
+}

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
 }
 

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
 }

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,

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
 			}

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)

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,14 +30,22 @@ 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) (*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, clientOptions), opts...)
 	if err != nil {
 		return nil, errors.Wrap(err, "getting internal client")
 	}

ctl/restore.go

@@ -5,18 +5,23 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net/http"
 	"os"
 	"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 +34,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 +53,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 +81,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 +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")
-		var client http.Client
+		client := cmd.newClient()
 		_, err = client.Post(url, "application/json", f)
 	} else {
 		schema := &pilosa.Schema{}
@@ -159,6 +178,34 @@ 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)
+}
+
+// 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()
 
@@ -174,7 +221,7 @@ 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 := cmd.newClient()
 	_, err = client.Post(url, "application/octet-stream", f)
 	return err
 }
@@ -244,14 +291,14 @@ 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 := cmd.newClient()
 		resp, err := client.Do(req)
 		if err != nil {
 			return err
@@ -319,13 +366,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 +425,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
 		}

ctl/server.go

@@ -45,7 +45,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
 
 	// Etcd
 	// Etcd.Name used Config.Name for its value.
-	// Etcd.Dir defaults to a directory under the pilosa data directory.
+	flags.StringVar(&srv.Config.Etcd.Dir, "etcd.dir", srv.Config.Etcd.Dir, "Directory to store etcd data files. If not provided, a directory will be created under the main data-dir directory.")
 	// Etcd.ClusterName uses Cluster.Name for its value
 	flags.StringVar(&srv.Config.Etcd.LClientURL, "etcd.listen-client-address", srv.Config.Etcd.LClientURL, "Listen client address.")
 	flags.StringVar(&srv.Config.Etcd.AClientURL, "etcd.advertise-client-address", srv.Config.Etcd.AClientURL, "Advertise client address. If not provided, uses the listen client address.")

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

executor.go

@@ -179,11 +179,18 @@ func newExecutor(opts ...executorOption) *executor {
 
 func (e *executor) addWorker() {
 	e.workersWG.Add(1)
-	atomic.AddInt64(&e.currentWorkers, 1)
+	n := atomic.AddInt64(&e.currentWorkers, 1)
+	if e.Holder != nil {
+		e.Holder.Stats.Gauge("worker_total", float64(n), 0)
+	}
+
 	go func() {
 		defer e.workersWG.Done()
 		e.worker(e.work)
-		atomic.AddInt64(&e.currentWorkers, -1)
+		n := atomic.AddInt64(&e.currentWorkers, -1)
+		if e.Holder != nil {
+			e.Holder.Stats.Gauge("worker_total", float64(n), 0)
+		}
 	}()
 }
 
@@ -204,6 +211,14 @@ func (e *executor) Close() error {
 	return nil
 }
 
+// InitStats initializes stats counters. Must be called after Holder set.
+func (e *executor) InitStats() {
+	if e.Holder != nil {
+		e.Holder.Stats.Count("job_total", 0, 0)
+		e.Holder.Stats.Gauge("worker_total", float64(atomic.LoadInt64(&e.currentWorkers)), 0)
+	}
+}
+
 // Execute executes a PQL query.
 func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) (QueryResponse, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.Execute")
@@ -587,6 +602,11 @@ func (e *executor) execute(ctx context.Context, qcx *Qcx, index string, q *pql.Q
 			return nil, err
 		}
 
+		if vc, ok := v.(ValCount); ok {
+			vc.cleanup()
+			v = vc
+		}
+
 		results = append(results, v)
 		// Some Calls can have significant data associated with them
 		// that gets generated during processing, such as Precomputed
@@ -597,6 +617,22 @@ func (e *executor) execute(ctx context.Context, qcx *Qcx, index string, q *pql.Q
 	return results, nil
 }
 
+// cleanup removes the integer value (Val) from the ValCount if one of
+// the other fields is in use.
+//
+// ValCounts are normally holding data which is stored as a BSI
+// (integer) under the hood. Sometimes it's convenient to be able to
+// compare the underlying integer values rather than their
+// interpretation as decimal, timestamp, etc, so the lower level
+// functions may return both integer and the interpreted value, but we
+// don't want to pass that all the way back to the client, so we
+// remove it here.
+func (vc *ValCount) cleanup() {
+	if vc.Val != 0 && (vc.FloatVal != 0 || !vc.TimestampVal.IsZero() || vc.DecimalVal != nil) {
+		vc.Val = 0
+	}
+}
+
 // preprocessQuery expands any calls that need preprocessing.
 func (e *executor) preprocessQuery(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (*pql.Call, error) {
 	switch c.Name {
@@ -1261,6 +1297,10 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string
 	if err != nil {
 		return ValCount{}, errors.New("Percentile(): field required")
 	}
+	field := e.Holder.Field(index, fieldName)
+	if field == nil {
+		return ValCount{}, ErrFieldNotFound
+	}
 
 	// filter call for min & max
 	var filterCall *pql.Call
@@ -1281,7 +1321,7 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string
 		return ValCount{}, errors.Wrap(err, "executing Min call for Percentile")
 	}
 	if nthFloat == 0.0 {
-		return ValCount{Val: minVal.Val, Count: minVal.Count}, nil
+		return minVal, nil
 	}
 
 	// get max
@@ -1348,11 +1388,11 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string
 		} else if leftCountWeighted < rightCount {
 			min = possibleNthVal + 1
 		} else {
-			return ValCount{Val: possibleNthVal, Count: 1}, nil
+			return field.valCountize(possibleNthVal, 1, nil)
 		}
 	}
 
-	return ValCount{Val: min, Count: 1}, nil
+	return field.valCountize(min, 1, nil)
 
 }
 
@@ -5989,6 +6029,7 @@ type job struct {
 func (e *executor) worker(work chan job) {
 	for j := range work {
 		atomic.AddUint64(&e.workCounter, 1)
+		e.Holder.Stats.Count("job_total", 1, 0)
 		if j.idleHands {
 			return
 		}
@@ -8130,6 +8171,8 @@ func getScaledInt(f *Field, v interface{}) (int64, error) {
 		switch tv := v.(type) {
 		case time.Time:
 			value = tv.UnixNano() / TimeUnitNanos(f.options.TimeUnit)
+		case int64:
+			value = tv
 		default:
 			return 0, errors.Errorf("unexpected timestamp value type %T, val %v", tv, tv)
 		}

executor_internal_test.go

@@ -489,3 +489,18 @@ func TestExecutorSafeCopyDistinctTimestamp(t *testing.T) {
 		t.Fatalf("Did not copy results. got %+v, want %+v", copied.Results, response.Results)
 	}
 }
+
+func TestGetScaledInt(t *testing.T) {
+	f := OpenField(t, OptFieldTypeTimestamp(time.Now(), "ms"))
+	defer f.Close()
+	// check that fields with type timestamp return the int64 passed in to getScaledInt with nil err
+	v := time.Now().Unix()
+	res, err := getScaledInt(f.Field, v)
+	if err != nil {
+		t.Errorf("got error %v, expected nil", err)
+	}
+	if !reflect.DeepEqual(res, v) {
+		t.Errorf("expected %v, got %v", v, res)
+	}
+
+}

field.go

@@ -1388,18 +1388,7 @@ func (f *Field) MaxForShard(tx Tx, shard uint64, filter *Row) (ValCount, error)
 		return ValCount{}, errors.Wrap(err, "calling fragment.max")
 	}
 
-	valCount := ValCount{Count: int64(cnt)}
-
-	if f.Options().Type == FieldTypeDecimal {
-		dec := pql.NewDecimal(max+bsig.Base, bsig.Scale)
-		valCount.DecimalVal = &dec
-	} else if f.Options().Type == FieldTypeTimestamp {
-		valCount.TimestampVal = time.Unix(0, (max+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC()
-	} else {
-		valCount.Val = max + bsig.Base
-	}
-
-	return valCount, nil
+	return f.valCountize(max, cnt, bsig)
 }
 
 // MinForShard returns the minimum value which appears in this shard
@@ -1434,17 +1423,32 @@ func (f *Field) MinForShard(tx Tx, shard uint64, filter *Row) (ValCount, error)
 		return ValCount{}, errors.Wrap(err, "calling fragment.min")
 	}
 
+	return f.valCountize(min, cnt, bsig)
+}
+
+// valCountize takes the "raw" value and count we get from the
+// fragment and calculates the cooked values for this field
+// (timestamping, decimaling, or just adding in the base). It always
+// includes the int64 "Val\" value to make comparisons easier in the
+// executor (at time of writing, Percentile takes advantage of this,
+// but we might be able to simplify logic in other places as well).
+func (f *Field) valCountize(val int64, cnt uint64, bsig *bsiGroup) (ValCount, error) {
+	if bsig == nil {
+		bsig = f.bsiGroup(f.name)
+		if bsig == nil {
+			return ValCount{}, ErrBSIGroupNotFound
+		}
+
+	}
 	valCount := ValCount{Count: int64(cnt)}
 
 	if f.Options().Type == FieldTypeDecimal {
-		dec := pql.NewDecimal(min+bsig.Base, bsig.Scale)
+		dec := pql.NewDecimal(val+bsig.Base, bsig.Scale)
 		valCount.DecimalVal = &dec
 	} else if f.Options().Type == FieldTypeTimestamp {
-		valCount.TimestampVal = time.Unix(0, (min+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC()
-	} else {
-		valCount.Val = min + bsig.Base
+		valCount.TimestampVal = time.Unix(0, (val+bsig.Base)*TimeUnitNanos(f.options.TimeUnit)).UTC()
 	}
-
+	valCount.Val = val + bsig.Base
 	return valCount, nil
 }
 

field_internal_test.go

@@ -182,6 +182,23 @@ func TestBSIGroup_BaseValue(t *testing.T) {
 	})
 }
 
+func TestField_ValCountize(t *testing.T) {
+	f := OpenField(t, OptFieldTypeDefault())
+	defer f.Close()
+	// check that you get an empty val count and err
+	// BSIGroupNotFound on nil bsig from
+	// f.bsiGroup(f.name)
+	f.bsiGroups = []*bsiGroup{}
+	v, err := f.valCountize(42, 42, nil)
+	if !reflect.DeepEqual(v, ValCount{}) {
+		t.Errorf("expected %v, got %v", ValCount{}, v)
+	}
+	if err != ErrBSIGroupNotFound {
+		t.Errorf("expected %v, got %v", ErrBSIGroupNotFound, err)
+	}
+
+}
+
 // Ensure field can open and retrieve a view.
 func TestField_DeleteView(t *testing.T) {
 	f := OpenField(t, OptFieldTypeDefault())
@@ -748,29 +765,29 @@ func TestDecimalField_MinMaxForShard(t *testing.T) {
 			name:      "single",
 			columnIDs: []uint64{1},
 			values:    []float64{10.1},
-			expMax:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
-			expMin:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
+			expMax:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
+			expMin:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
 		},
 		{
 			name:      "twovals",
 			columnIDs: []uint64{1, 2},
 			values:    []float64{10.1, 20.2},
-			expMax:    ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 1},
-			expMin:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
+			expMax:    ValCount{Val: 20200, DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 1},
+			expMin:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 1},
 		},
 		{
 			name:      "multiplecounts",
 			columnIDs: []uint64{1, 2, 3, 4, 5},
 			values:    []float64{10.1, 20.2, 10.1, 10.1, 20.2},
-			expMax:    ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
-			expMin:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
+			expMax:    ValCount{Val: 20200, DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
+			expMin:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
 		},
 		{
 			name:      "middlevals",
 			columnIDs: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
 			values:    []float64{10.1, 20.2, 10.1, 10.1, 20.2, 11, 12, 11, 13, 11},
-			expMax:    ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
-			expMin:    ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
+			expMax:    ValCount{Val: 20200, DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
+			expMin:    ValCount{Val: 10100, DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
 		},
 	} {
 		t.Run(test.name+strconv.Itoa(i), func(t *testing.T) {

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

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=

http/client.go

@@ -8,18 +8,22 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math"
 	"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 +35,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 +47,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 +57,75 @@ 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 {
+	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) {
+		rc := retryablehttp.NewClient()
+		rc.HTTPClient = c.httpClient
+		rc.RetryWaitMin = min
+		rc.RetryMax = int(attempts)
+		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 {
+		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 +1783,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 +2092,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 +2109,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 +2124,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 +2141,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
 	}

http/handler.go

@@ -441,6 +441,9 @@ func newRouter(handler *Handler) http.Handler {
 	router.HandleFunc("/internal/idalloc/data", handler.handleIDAllocData).Methods("GET").Name("IDAllocData")
 
 	router.HandleFunc("/internal/restore/{index}/{shardID}", handler.handlePostRestore).Methods("POST").Name("Restore")
+
+	router.HandleFunc("/internal/debug/rbf", handler.handleGetInternalDebugRBFJSON).Methods("GET").Name("GetInternalDebugRBFJSON")
+
 	// endpoints for collecting cpu profiles from a chosen begin point to
 	// when the client wants to stop. Used for profiling imports that
 	// could be long or short.
@@ -2064,6 +2067,18 @@ func validateProtobufHeader(r *http.Request) (error string, code int) {
 	return
 }
 
+// handleGetInternalDebugRBFJSON handles /internal/debug/rbf requests.
+func (h *Handler) handleGetInternalDebugRBFJSON(w http.ResponseWriter, r *http.Request) {
+	buf, err := json.MarshalIndent(h.api.RBFDebugInfo(), "", "  ")
+	if err != nil {
+		http.Error(w, "marshal json: "+err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.Write(buf)
+}
+
 // handleGetMetricsJSON handles /metrics.json requests, translating text metrics results to more consumable JSON.
 func (h *Handler) handleGetMetricsJSON(w http.ResponseWriter, r *http.Request) {
 	if !validHeaderAcceptJSON(r.Header) {
@@ -2578,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,
@@ -2595,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
@@ -3348,7 +3371,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
 	}
 

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,89 @@ 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 backupCmd, err = startCmd(
+			"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 * 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)
+		}
+		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) {

internal/clustertests/docker-compose.yml

@@ -9,6 +9,7 @@ services:
       - "33455:10101"
     environment:
       - PILOSA_NAME=pilosa1
+      - PILOSA_ETCD_DIR=/root/.etcd
       - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
       - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa1:10201
       - PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301
@@ -28,6 +29,7 @@ services:
       - "33456:10101"
     environment:
       - PILOSA_NAME=pilosa2
+      - PILOSA_ETCD_DIR=/root/.etcd
       - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
       - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa2:10201
       - PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301
@@ -47,6 +49,7 @@ services:
       - "33457:10101"
     environment:
       - PILOSA_NAME=pilosa3
+      - PILOSA_ETCD_DIR=/root/.etcd
       - PILOSA_ETCD_LISTEN_CLIENT_ADDRESS=http://0.0.0.0:10201
       - PILOSA_ETCD_ADVERTISE_CLIENT_ADDRESS=http://pilosa3:10201
       - PILOSA_ETCD_LISTEN_PEER_ADDRESS=http://0.0.0.0:10301

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

rbf/db.go

@@ -7,6 +7,8 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"runtime/debug"
+	"sort"
 	"sync"
 	"syscall"
 
@@ -637,6 +639,7 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {
 		pageMap:     db.pageMap,
 		walPageN:    db.walPageN,
 		writable:    writable,
+		stack:       debug.Stack(), // DEBUG
 
 		DeleteEmptyContainer: true,
 	}
@@ -815,6 +818,20 @@ func (db *DB) getCursor(tx *Tx) *Cursor {
 	return c
 }
 
+func (db *DB) DebugInfo() *DebugInfo {
+	info := &DebugInfo{Path: db.Path}
+	for tx := range db.txs {
+		info.Txs = append(info.Txs, tx.DebugInfo())
+	}
+	sort.Slice(info.Txs, func(i, j int) bool { return info.Txs[i].Ptr < info.Txs[j].Ptr })
+	return info
+}
+
+type DebugInfo struct {
+	Path string         `json:"path"`
+	Txs  []*TxDebugInfo `json:"txs"`
+}
+
 // Shared pool for in-memory database pages.
 // These are used before being flushed to disk.
 var pagePool = &sync.Pool{

rbf/db_test.go

@@ -339,6 +339,21 @@ func TestDB_MultiTx(t *testing.T) {
 	}
 }
 
+func TestDB_DebugInfo(t *testing.T) {
+	db := MustOpenDB(t)
+	defer MustCloseDB(t, db)
+
+	tx := MustBegin(t, db, true)
+	defer tx.Rollback()
+
+	info := db.DebugInfo()
+	if got, want := info.Path, db.Path; got != want {
+		t.Fatalf("Path=%q, want %q", got, want)
+	} else if got, want := len(info.Txs), 1; got != want {
+		t.Fatalf("len(Txs)=%d, want %d", got, want)
+	}
+}
+
 // premake pool of random values
 const randPool = (1 << 18)
 

rbf/tx.go

@@ -65,6 +65,9 @@ type Tx struct {
 	// manages to trigger a *deallocation* (which I don't think should be
 	// happening), we'll process that one after the current list is processed.
 	pendingFreelistAdds []uint32
+
+	// DEBUG
+	stack []byte
 }
 
 func (tx *Tx) DBPath() string {
@@ -2042,6 +2045,20 @@ func (tx *Tx) GetSortedFieldViewList() (fvs []txkey.FieldView, _ error) {
 	return
 }
 
+func (tx *Tx) DebugInfo() *TxDebugInfo {
+	return &TxDebugInfo{
+		Ptr:      fmt.Sprintf("%p", tx),
+		Writable: tx.writable,
+		Stack:    string(tx.stack),
+	}
+}
+
+type TxDebugInfo struct {
+	Ptr      string `json:"ptr"`
+	Writable bool   `json:"writable"`
+	Stack    string `json:"stack,omitempty"`
+}
+
 // SnapshotReader returns a reader that provides a snapshot for the current database state.
 func (tx *Tx) SnapshotReader() (io.Reader, error) {
 	if tx.db == nil {

server.go

@@ -506,6 +506,10 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	s.holder.schemator = s.schemator
 	s.holder.sharder = s.sharder
 	s.holder.serializer = s.serializer
+
+	// Initial stats must be invoked after the executor obtains reference to the holder.
+	s.executor.InitStats()
+
 	return s, nil
 }
 

server/server_test.go

@@ -17,7 +17,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/molecula/featurebase/v2"
+	pilosa "github.com/molecula/featurebase/v2"
 	"github.com/molecula/featurebase/v2/disco"
 	"github.com/molecula/featurebase/v2/http"
 	"github.com/molecula/featurebase/v2/pql"
@@ -26,6 +26,7 @@ import (
 	"github.com/molecula/featurebase/v2/test"
 	"github.com/molecula/featurebase/v2/testhook"
 	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -504,6 +505,30 @@ func TestClusteringNodesReplica1(t *testing.T) {
 		t.Fatalf("starting cluster: %v", err)
 	}
 
+	indexName := "idx"
+	fieldName := "fld"
+
+	// Create the schema.
+	if _, err := cluster.GetPrimary().API.CreateIndex(context.Background(), indexName, pilosa.IndexOptions{}); err != nil {
+		t.Fatalf("creating index: %v", err)
+	}
+	if _, err := cluster.GetPrimary().API.CreateField(context.Background(), indexName, fieldName); err != nil {
+		t.Fatalf("creating field: %v", err)
+	}
+
+	// Set some columns across shards to ensure that the Row query will require
+	// data from all nodes.
+	data := []string{}
+	for rowID := 1; rowID < 2; rowID++ {
+		for columnID := 1; columnID < 10; columnID++ {
+			data = append(data, fmt.Sprintf(`Set(%d, %s=%d)`, columnID*pilosa.ShardWidth, fieldName, rowID))
+		}
+	}
+	if _, err := cluster.GetPrimary().Query(t, indexName, "", strings.Join(data, "")); err != nil {
+		t.Fatalf("setting columns: %v", err)
+	}
+
+	// Shut down a node.
 	if err := cluster.GetNonPrimary().Command.Close(); err != nil {
 		t.Fatalf("closing third node: %v", err)
 	}
@@ -513,7 +538,12 @@ func TestClusteringNodesReplica1(t *testing.T) {
 	}
 
 	// confirm that cluster stops accepting queries after one node closes
-	if _, err := cluster.GetPrimary().API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
+	qry := &pilosa.QueryRequest{
+		Index: "idx",
+		Query: fmt.Sprintf("Row(%s=1)", fieldName),
+	}
+
+	if _, err := cluster.GetPrimary().API.Query(context.Background(), qry); !strings.Contains(err.Error(), "shard unavailable") {
 		t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
 	}
 }
@@ -540,8 +570,34 @@ func TestClusteringNodesReplica2(t *testing.T) {
 	}
 	defer cluster.Close()
 
+	indexName := "idx"
+	fieldName := "fld"
+
 	coord, others := cluster.GetPrimary(), cluster.GetNonPrimaries()
 
+	// Create the schema.
+	if _, err := coord.API.CreateIndex(context.Background(), indexName, pilosa.IndexOptions{}); err != nil {
+		t.Fatalf("creating index: %v", err)
+	}
+	if _, err := coord.API.CreateField(context.Background(), indexName, fieldName); err != nil {
+		t.Fatalf("creating field: %v", err)
+	}
+
+	// Set some columns across shards to ensure that the Row query will require
+	// data from all nodes.
+	data := []string{}
+	cols := []uint64{}
+	for rowID := 1; rowID < 2; rowID++ {
+		for columnID := 1; columnID < 30; columnID++ {
+			col := uint64(columnID * pilosa.ShardWidth)
+			cols = append(cols, col)
+			data = append(data, fmt.Sprintf(`Set(%d, %s=%d)`, col, fieldName, rowID))
+		}
+	}
+	if _, err := coord.Query(t, indexName, "", strings.Join(data, "")); err != nil {
+		t.Fatalf("setting columns: %v", err)
+	}
+
 	if err := others[0].Close(); err != nil {
 		t.Fatalf("closing third node: %v", err)
 	}
@@ -569,8 +625,30 @@ func TestClusteringNodesReplica2(t *testing.T) {
 		t.Fatalf("after closing second server: %v", err)
 	}
 
-	if _, err := coord.API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state DOWN") {
-		t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
+	qry := &pilosa.QueryRequest{
+		Index: "idx",
+		Query: fmt.Sprintf("Row(%s=1)", fieldName),
+	}
+
+	// Because we no longer block queries when the cluster is in state DOWN,
+	// there are cases where a DOWN cluster can still respond to a query. In
+	// that case, we want the test to pass. But if the unavailable node(s) cause
+	// the query to result in an error, we check that it's the error we expect.
+	resp, err := coord.API.Query(context.Background(), qry)
+	if err != nil {
+		if !strings.Contains(err.Error(), "shard unavailable") {
+			t.Fatalf("got unexpected error querying an incomplete cluster: %v", err)
+		}
+	} else {
+		if len(resp.Results) == 0 {
+			t.Fatal("got no results")
+		}
+
+		row, ok := resp.Results[0].(*pilosa.Row)
+		if !ok {
+			t.Fatalf("expected a *pilosa.Row, but got %T", resp.Results[0])
+		}
+		require.Equal(t, row.Columns(), cols)
 	}
 }