diff --git a/.circleci/config.yml b/.circleci/config.yml index 14c7a3367..000155a28 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -126,6 +126,8 @@ jobs: name: golang version: << parameters.golang_version >> resource_class: << parameters.resource_class >> + environment: + TMPDIR: /mnt/ramdisk steps: - checkout-plus - skip-if-root-unchanged diff --git a/etcd/embed.go b/etcd/embed.go index 3f3ec75a5..25800c4de 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -61,6 +61,8 @@ type Options struct { LPeerSocket []*net.TCPListener LClientSocket []*net.TCPListener + + BootstrapTimeout time.Duration } var ( @@ -224,6 +226,7 @@ func parseOptions(opt Options) *embed.Config { cfg.Name = opt.Name cfg.Dir = opt.Dir cfg.InitialClusterToken = opt.ClusterName + cfg.BootstrapTimeout = opt.BootstrapTimeout cfg.LCUrls = types.MustNewURLs([]string{opt.LClientURL}) if opt.AClientURL != "" { cfg.ACUrls = types.MustNewURLs([]string{opt.AClientURL}) diff --git a/etcd/leasedkv.go b/etcd/leasedkv.go index 4356821c3..515c70ac4 100644 --- a/etcd/leasedkv.go +++ b/etcd/leasedkv.go @@ -30,8 +30,10 @@ import ( // It will try to renew the lease at any cost after losing it. // It will recreate the previous existing value for the key again. type leasedKV struct { - e *Etcd - cancel context.CancelFunc + e *Etcd + cancel context.CancelFunc + done <-chan struct{} + leaseID clientv3.LeaseID key string ttlSeconds int64 @@ -70,8 +72,8 @@ func (l *leasedKV) create(initValue string) (<-chan *clientv3.LeaseKeepAliveResp if l.cancel != nil { l.cancel() } - l.cancel = cancel + l.done = ctx.Done() var leaseResp *clientv3.LeaseGrantResponse err := l.e.retryClient(func(cli *clientv3.Client) (err error) { @@ -81,10 +83,11 @@ func (l *leasedKV) create(initValue string) (<-chan *clientv3.LeaseKeepAliveResp if err != nil { return nil, errors.Wrap(err, "creating a lease") } + l.leaseID = leaseResp.ID err = l.e.retryClient(func(cli *clientv3.Client) (err error) { _, err = cli.Txn(ctx). - Then(clientv3.OpPut(l.key, initValue, clientv3.WithLease(leaseResp.ID))). + Then(clientv3.OpPut(l.key, initValue, clientv3.WithLease(l.leaseID))). Commit() return err }) @@ -94,7 +97,7 @@ func (l *leasedKV) create(initValue string) (<-chan *clientv3.LeaseKeepAliveResp var kaChan <-chan *clientv3.LeaseKeepAliveResponse err = l.e.retryClient(func(cli *clientv3.Client) (err error) { - kaChan, err = cli.KeepAlive(ctx, leaseResp.ID) + kaChan, err = cli.KeepAlive(ctx, l.leaseID) return err }) if err != nil { @@ -108,8 +111,11 @@ func (l *leasedKV) create(initValue string) (<-chan *clientv3.LeaseKeepAliveResp func (l *leasedKV) consumeLease(ch <-chan *clientv3.LeaseKeepAliveResponse) { for { - _, ok := <-ch - if !ok { + select { + case _, ok := <-ch: + if ok { + continue + } l.mu.Lock() if l.stopped { @@ -134,6 +140,9 @@ func (l *leasedKV) consumeLease(ch <-chan *clientv3.LeaseKeepAliveResponse) { log.Printf("lease %q recreated after a problem", l.key) l.mu.Unlock() return + case <-l.done: + // don't recreate lease. + return } } } @@ -143,12 +152,28 @@ func (l *leasedKV) consumeLease(ch <-chan *clientv3.LeaseKeepAliveResponse) { func (l *leasedKV) Stop() { l.mu.Lock() defer l.mu.Unlock() - + l.stopped = true if l.cancel != nil { l.cancel() } + // low-effort attempt to cancel existing lease. if the cluster is + // shutting down, we don't want this to take long. + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + err := l.e.retryClient(func(cli *clientv3.Client) (err error) { + _, err = cli.Revoke(ctx, l.leaseID) + return err + }) + // if retryClient succeeds, just to be thorough, we'll cancel that + // context. + cancel() + if err != nil { + // It turns out that this will almost always report a + // failure because since we're shutting things down, + // the cluster as a whole may not be able to process responses. + // So this is a low-interest message usually. + l.e.logger.Debugf("revoking lease during shutdown: %v", err) + } - l.stopped = true } // Set will change the specific value for this key. diff --git a/executor.go b/executor.go index 15d6fb571..449bd69ee 100644 --- a/executor.go +++ b/executor.go @@ -1239,10 +1239,10 @@ func (e *executor) executePercentile(ctx context.Context, qcx *Qcx, index string } // get field - if fieldArg := c.Args["field"]; fieldArg == "" { + fieldName, err := c.FirstStringArg("field", "_field") + if err != nil { return ValCount{}, errors.New("Percentile(): field required") } - fieldName, _, _ := c.StringArg("field") // filter call for min & max var filterCall *pql.Call diff --git a/executor_test.go b/executor_test.go index 9427914b2..856c2b600 100644 --- a/executor_test.go +++ b/executor_test.go @@ -7026,6 +7026,11 @@ func variousQueriesOnPercentiles(t *testing.T, c *test.Cluster) { query: query, csvVerifier: fmt.Sprintf("%d,1\n", expectedPercentile), }) + query2 := fmt.Sprintf(`Percentile(field=net_worth, filter=Row(val="foo"), nth=%d)`, nth) + tests = append(tests, testCase{ + query: query2, + csvVerifier: fmt.Sprintf("%d,1\n", expectedPercentile), + }) } for i, tst := range tests { diff --git a/go.mod b/go.mod index 7af634a50..3eb1fdbe6 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/pilosa/pilosa/v2 -replace go.etcd.io/etcd => github.com/molecula/etcd v0.0.0-20210115113447-5d28bda617d2 +replace go.etcd.io/etcd => github.com/molecula/etcd v0.0.0-20210621160528-2cd93f1df0e7 require ( github.com/CAFxX/gcnotifier v0.0.0-20190112062741-224a280d589d diff --git a/go.sum b/go.sum index 296acadcb..4755dfe03 100644 --- a/go.sum +++ b/go.sum @@ -226,8 +226,8 @@ github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b h1:cZADDaNYM7xn/nklO3g198JerGQjadFuA0ofxBJgK0Y= github.com/molecula/apophenia v0.0.0-20190827192002-68b7a14a478b/go.mod h1:uXd1BiH7xLmgkhVmspdJLENv6uGWrTL/MQX2TN7Yz9s= -github.com/molecula/etcd v0.0.0-20210115113447-5d28bda617d2 h1:pkzCVLSrFQGVQv3raVGJw6aJCJdIZC/z59tUsSU1Zws= -github.com/molecula/etcd v0.0.0-20210115113447-5d28bda617d2/go.mod h1:1X1h4BZ44WjM0LJof1gKKLap1OA4RsicGCDRtACTkLI= +github.com/molecula/etcd v0.0.0-20210621160528-2cd93f1df0e7 h1:hufElvtCighE0G2VFJYDGWCY8JlmCWZ1FmXvlf25yUQ= +github.com/molecula/etcd v0.0.0-20210621160528-2cd93f1df0e7/go.mod h1:1X1h4BZ44WjM0LJof1gKKLap1OA4RsicGCDRtACTkLI= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223 h1:F9x/1yl3T2AeKLr2AMdilSD8+f9bvMnNN8VS5iDtovc= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs= diff --git a/pql/ast.go b/pql/ast.go index cb1b78b9e..dddf72812 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -481,6 +481,7 @@ var callInfoByFunc = map[string]callInfo{ allowUnknown: false, prototypes: map[string]interface{}{ "field": "", + "_field": "", "filter": nil, "nth": nil, }, diff --git a/test/disco.go b/test/disco.go index fad3783b9..8ace1f96e 100644 --- a/test/disco.go +++ b/test/disco.go @@ -19,6 +19,7 @@ import ( "net" "strings" "testing" + "time" "github.com/pilosa/pilosa/v2/etcd" "github.com/pilosa/pilosa/v2/server" @@ -101,14 +102,15 @@ func GetPortsGenConfigs(tb testing.TB, nodes []*Command) error { config.BindGRPC = grpcUrl config.GRPCListener = grpcListener config.Etcd = etcd.Options{ - Dir: discoDir, - LClientURL: clientURL, - AClientURL: clientURL, - LPeerURL: peerURL, - APeerURL: peerURL, - HeartbeatTTL: 12, - LPeerSocket: []*net.TCPListener{peerListener}, - LClientSocket: []*net.TCPListener{clientListener}, + Dir: discoDir, + LClientURL: clientURL, + AClientURL: clientURL, + LPeerURL: peerURL, + APeerURL: peerURL, + HeartbeatTTL: 60, + LPeerSocket: []*net.TCPListener{peerListener}, + LClientSocket: []*net.TCPListener{clientListener}, + BootstrapTimeout: 50 * time.Millisecond, } peerUrls[i] = fmt.Sprintf("%s=%s", name, peerURL) }