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

Merge branch 'master' into index-delete-shard

Browse source
This commit is contained in:
tgruben 2021-06-25 11:26:31 -05:00 • committed by GitHub
commit b15b11462a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 60 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.circleci/config.yml
View file

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

3
etcd/embed.go
View file

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

43
etcd/leasedkv.go
View file

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

4
executor.go
View file

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

5
executor_test.go
View file

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

2
go.mod
View file

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

4
go.sum
View file

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

1
pql/ast.go
View file

@ -481,6 +481,7 @@ var callInfoByFunc = map[string]callInfo{
allowUnknown: false,
prototypes: map[string]interface{}{
"field": "",
"_field": "",
"filter": nil,
"nth": nil,
},

18
test/disco.go
View file

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