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

Address PR comments

Browse source
This commit is contained in:
Kuba Podgórski 2021-03-03 16:11:24 +01:00
parent a14baf8c15
commit fa293ba6c3
2 changed files with 33 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
etcd/embed.go
View file

@ -264,13 +264,17 @@ func (e *Etcd) nodeState(ctx context.Context, peerID string) (disco.NodeState, e
return disco.NodeStateResizing, nil
}
kvs := resp.Responses[0].GetResponseRange().Kvs
if len(kvs) > 1 {
return disco.NodeStateUnknown, disco.ErrTooManyResults
if len(resp.Responses) == 0 {
return disco.NodeStateUnknown, disco.ErrNoResults
}
kvs := resp.Responses[0].GetResponseRange().Kvs
if len(kvs) == 0 {
return disco.NodeStateUnknown, disco.ErrNoResults
}
if len(kvs) > 1 {
return disco.NodeStateUnknown, disco.ErrTooManyResults
}
return disco.NodeState(kvs[0].Value), nil
}
@ -698,12 +702,11 @@ func (e *Etcd) getKeyBytes(ctx context.Context, key string) ([]byte, error) {
return nil, err
}
kvs := resp.Responses[0].GetResponseRange().Kvs
if !resp.Succeeded {
return nil, errors.New("tx failed")
if len(resp.Responses) == 0 {
return nil, errors.New("key does not exist")
}
kvs := resp.Responses[0].GetResponseRange().Kvs
if len(kvs) == 0 {
return nil, errors.New("key does not exist")
}
@ -711,24 +714,28 @@ func (e *Etcd) getKeyBytes(ctx context.Context, key string) ([]byte, error) {
return kvs[0].Value, nil
}
func (e *Etcd) getKeyWithPrefix(ctx context.Context, key string) ([]string, [][]byte, error) {
func (e *Etcd) getKeyWithPrefix(ctx context.Context, key string) (keys []string, values [][]byte, err error) {
resp, err := e.cli.Txn(ctx).
Then(clientv3.OpGet(key, clientv3.WithPrefix())).
Commit()
if err != nil {
return nil, nil, err
}
if len(resp.Responses) == 0 {
return nil, nil, errors.New("key does not exist")
}
kvs := resp.Responses[0].GetResponseRange().Kvs
if len(kvs) == 0 {
return nil, nil, nil
}
var (
keys []string
values [][]byte
)
for _, kv := range kvs {
keys = append(keys, string(kv.Key))
values = append(values, kv.Value)
keys = make([]string, len(kvs))
values = make([][]byte, len(kvs))
for i, kv := range kvs {
keys[i] = string(kv.Key)
values[i] = kv.Value
}
return keys, values, nil
@ -746,6 +753,9 @@ func (e *Etcd) keyExists(ctx context.Context, key string) (bool, error) {
return false, nil
}
if len(resp.Responses) == 0 {
return false, nil
}
return resp.Responses[0].GetResponseRange().Count > 0, nil
}

10
internal/clustertests/cluster_test.go
View file

@ -22,6 +22,7 @@ import (
"time"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/disco"
picli "github.com/pilosa/pilosa/v2/http"
)
@ -90,7 +91,7 @@ func TestClusterStuff(t *testing.T) {
}
t.Log("done with pause, waiting for stability")
waitForStatus(t, cli1, "NORMAL", 30, time.Second)
waitForStatus(t, cli1, string(disco.ClusterStateNormal), 30, time.Second)
t.Log("done waiting for stability")
// Check query results from each node.
@ -112,9 +113,9 @@ func waitForStatus(t *testing.T, c *picli.InternalClient, status string, n int,
for i := 0; i < n; i++ {
s, err := c.Status(context.TODO())
if err != nil {
t.Logf("Status (%d/%d): %v (sleep: %s)\\n", i, n, err, sleep.String())
t.Logf("Status (try %d/%d): %v (retrying in %s)", i, n, err, sleep.String())
} else {
t.Logf("Status (%d/%d): %s (sleep: %s)\n", i, n, s, sleep.String())
t.Logf("Status (try %d/%d): %s (retrying in %s)", i, n, s, sleep.String())
}
if s == status {
return
@ -127,6 +128,7 @@ func waitForStatus(t *testing.T, c *picli.InternalClient, status string, n int,
t.Fatalf("querying status: %v", err)
}
if status != s {
t.Fatalf("waited %d %v for status: %v, got: %v", n, sleep, status, s)
waited := time.Duration(n) * sleep
t.Fatalf("waited %s for status: %s, got: %s", waited.String(), status, s)
}
}