From 01103b26f02fc07445969f7d88358b7c9ced5091 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 21 Jun 2021 11:12:09 -0500 Subject: [PATCH 1/5] make etcd bootstrap timeout configurable It turns out that it's desireable to be able to configure the bootstrap timeout for etcd, because during startup, we end up delaying that long (N-1) times in series during each cluster creation, which is pointless when we're starting the whole cluster. Reduces test runtime by several minutes. --- etcd/embed.go | 3 +++ go.mod | 2 +- go.sum | 4 ++-- test/disco.go | 18 ++++++++++-------- 4 files changed, 16 insertions(+), 11 deletions(-) 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/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/test/disco.go b/test/disco.go index fad3783b9..ac19940d5 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: 12, + LPeerSocket: []*net.TCPListener{peerListener}, + LClientSocket: []*net.TCPListener{clientListener}, + BootstrapTimeout: 50 * time.Millisecond, } peerUrls[i] = fmt.Sprintf("%s=%s", name, peerURL) } From 5898c58a80a62c9311ec5dcf0305fa7f1d55edd7 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 21 Jun 2021 11:15:27 -0500 Subject: [PATCH 2/5] try to force TMPDIR to be the ram disk CircleCI is providing a ramdisk as /mnt/ramdisk. Using the ramdisk instead of local storage makes fsync operations essentially free, which removes (some of) the frequent multi-second delays we see during runs otherwise. --- .circleci/config.yml | 2 ++ 1 file changed, 2 insertions(+) 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 From 13dd357dddee0251be19238afb18aa0f1aa69644 Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 21 Jun 2021 11:16:40 -0500 Subject: [PATCH 3/5] bump heartbeatTTL to ludicrous value (60s) for testing After continuing to see weird test failures, did some more careful testing, discovered that CI can pause a machine entirely for up to 29 seconds or so very rarely, and 5-10 seconds quite frequently, which causes cascading heartbeat failures and so on. Remove those. --- test/disco.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/disco.go b/test/disco.go index ac19940d5..8ace1f96e 100644 --- a/test/disco.go +++ b/test/disco.go @@ -107,7 +107,7 @@ func GetPortsGenConfigs(tb testing.TB, nodes []*Command) error { AClientURL: clientURL, LPeerURL: peerURL, APeerURL: peerURL, - HeartbeatTTL: 12, + HeartbeatTTL: 60, LPeerSocket: []*net.TCPListener{peerListener}, LClientSocket: []*net.TCPListener{clientListener}, BootstrapTimeout: 50 * time.Millisecond, From 0e1d448baa998e3b1c28b07746463c0ae9d4e8ba Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 21 Jun 2021 12:23:15 -0500 Subject: [PATCH 4/5] explicitly revoke lease on shutdown If we explicitly shut a node down, we don't want everyone else thinking it's up for the next 5 seconds. Worse, in CI, we have random long delays (10+ seconds) with no CPU activity at all, so we have to set the TTL longer there. Which makes any test checking for responsive detection of a node going down take even longer. So! We revoke leases on our way down, and this makes the tests not take so long, and allows us to have a reasonable timeout on the test, while having a completely unreasonable HeartbeatTTL to make CI stop breaking randomly. --- etcd/leasedkv.go | 43 ++++++++++++++++++++++++++++++++++--------- 1 file changed, 34 insertions(+), 9 deletions(-) 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. From 357caf68c3fbe73ab3a37c1f6a9593a1c22a042e Mon Sep 17 00:00:00 2001 From: Mahesh Arumugam Date: Thu, 24 Jun 2021 15:10:05 -0700 Subject: [PATCH 5/5] Fix percentile query: field is mandatory (should not crash), fieldnames can be unquoted --- executor.go | 4 ++-- executor_test.go | 5 +++++ pql/ast.go | 1 + 3 files changed, 8 insertions(+), 2 deletions(-) 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/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, },