Revision 6d8052314b9e01447d1bb2030bd762e0baf2d433 authored by Joe Betz on 11 October 2019, 17:23:13 UTC, committed by Joe Betz on 11 October 2019, 17:23:13 UTC
1 parent 5e4d852
v3_grpc_test.go
// Copyright 2016 The etcd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package integration
import (
"bytes"
"context"
"fmt"
"io/ioutil"
"math/rand"
"os"
"reflect"
"testing"
"time"
"github.com/coreos/etcd/clientv3"
"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
"github.com/coreos/etcd/pkg/testutil"
"github.com/coreos/etcd/pkg/transport"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
// TestV3PutOverwrite puts a key with the v3 api to a random cluster member,
// overwrites it, then checks that the change was applied.
func TestV3PutOverwrite(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
key := []byte("foo")
reqput := &pb.PutRequest{Key: key, Value: []byte("bar"), PrevKv: true}
respput, err := kvc.Put(context.TODO(), reqput)
if err != nil {
t.Fatalf("couldn't put key (%v)", err)
}
// overwrite
reqput.Value = []byte("baz")
respput2, err := kvc.Put(context.TODO(), reqput)
if err != nil {
t.Fatalf("couldn't put key (%v)", err)
}
if respput2.Header.Revision <= respput.Header.Revision {
t.Fatalf("expected newer revision on overwrite, got %v <= %v",
respput2.Header.Revision, respput.Header.Revision)
}
if pkv := respput2.PrevKv; pkv == nil || string(pkv.Value) != "bar" {
t.Fatalf("expected PrevKv=bar, got response %+v", respput2)
}
reqrange := &pb.RangeRequest{Key: key}
resprange, err := kvc.Range(context.TODO(), reqrange)
if err != nil {
t.Fatalf("couldn't get key (%v)", err)
}
if len(resprange.Kvs) != 1 {
t.Fatalf("expected 1 key, got %v", len(resprange.Kvs))
}
kv := resprange.Kvs[0]
if kv.ModRevision <= kv.CreateRevision {
t.Errorf("expected modRev > createRev, got %d <= %d",
kv.ModRevision, kv.CreateRevision)
}
if !reflect.DeepEqual(reqput.Value, kv.Value) {
t.Errorf("expected value %v, got %v", reqput.Value, kv.Value)
}
}
// TestPutRestart checks if a put after an unrelated member restart succeeds
func TestV3PutRestart(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
defer clus.Terminate(t)
kvIdx := rand.Intn(3)
kvc := toGRPC(clus.Client(kvIdx)).KV
stopIdx := kvIdx
for stopIdx == kvIdx {
stopIdx = rand.Intn(3)
}
clus.clients[stopIdx].Close()
clus.Members[stopIdx].Stop(t)
clus.Members[stopIdx].Restart(t)
c, cerr := NewClientV3(clus.Members[stopIdx])
if cerr != nil {
t.Fatalf("cannot create client: %v", cerr)
}
clus.clients[stopIdx] = c
ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
defer cancel()
reqput := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
_, err := kvc.Put(ctx, reqput)
if err != nil && err == ctx.Err() {
t.Fatalf("expected grpc error, got local ctx error (%v)", err)
}
}
// TestV3CompactCurrentRev ensures keys are present when compacting on current revision.
func TestV3CompactCurrentRev(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
preq := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
for i := 0; i < 3; i++ {
if _, err := kvc.Put(context.Background(), preq); err != nil {
t.Fatalf("couldn't put key (%v)", err)
}
}
// get key to add to proxy cache, if any
if _, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")}); err != nil {
t.Fatal(err)
}
// compact on current revision
_, err := kvc.Compact(context.Background(), &pb.CompactionRequest{Revision: 4})
if err != nil {
t.Fatalf("couldn't compact kv space (%v)", err)
}
// key still exists when linearized?
_, err = kvc.Range(context.Background(), &pb.RangeRequest{Key: []byte("foo")})
if err != nil {
t.Fatalf("couldn't get key after compaction (%v)", err)
}
// key still exists when serialized?
_, err = kvc.Range(context.Background(), &pb.RangeRequest{Key: []byte("foo"), Serializable: true})
if err != nil {
t.Fatalf("couldn't get serialized key after compaction (%v)", err)
}
}
// TestV3HashKV ensures that multiple calls of HashKV on same node return same hash and compact rev.
func TestV3HashKV(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
mvc := toGRPC(clus.RandClient()).Maintenance
for i := 0; i < 10; i++ {
resp, err := kvc.Put(context.Background(), &pb.PutRequest{Key: []byte("foo"), Value: []byte(fmt.Sprintf("bar%d", i))})
if err != nil {
t.Fatal(err)
}
rev := resp.Header.Revision
hresp, err := mvc.HashKV(context.Background(), &pb.HashKVRequest{})
if err != nil {
t.Fatal(err)
}
if rev != hresp.Header.Revision {
t.Fatalf("Put rev %v != HashKV rev %v", rev, hresp.Header.Revision)
}
prevHash := hresp.Hash
prevCompactRev := hresp.CompactRevision
for i := 0; i < 10; i++ {
hresp, err := mvc.HashKV(context.Background(), &pb.HashKVRequest{})
if err != nil {
t.Fatal(err)
}
if rev != hresp.Header.Revision {
t.Fatalf("Put rev %v != HashKV rev %v", rev, hresp.Header.Revision)
}
if prevHash != hresp.Hash {
t.Fatalf("prevHash %v != Hash %v", prevHash, hresp.Hash)
}
if prevCompactRev != hresp.CompactRevision {
t.Fatalf("prevCompactRev %v != CompactRevision %v", prevHash, hresp.Hash)
}
prevHash = hresp.Hash
prevCompactRev = hresp.CompactRevision
}
}
}
func TestV3TxnTooManyOps(t *testing.T) {
defer testutil.AfterTest(t)
maxTxnOps := uint(128)
clus := NewClusterV3(t, &ClusterConfig{Size: 3, MaxTxnOps: maxTxnOps})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
// unique keys
i := new(int)
keyf := func() []byte {
*i++
return []byte(fmt.Sprintf("key-%d", i))
}
addCompareOps := func(txn *pb.TxnRequest) {
txn.Compare = append(txn.Compare,
&pb.Compare{
Result: pb.Compare_GREATER,
Target: pb.Compare_CREATE,
Key: keyf(),
})
}
addSuccessOps := func(txn *pb.TxnRequest) {
txn.Success = append(txn.Success,
&pb.RequestOp{
Request: &pb.RequestOp_RequestPut{
RequestPut: &pb.PutRequest{
Key: keyf(),
Value: []byte("bar"),
},
},
})
}
addFailureOps := func(txn *pb.TxnRequest) {
txn.Failure = append(txn.Failure,
&pb.RequestOp{
Request: &pb.RequestOp_RequestPut{
RequestPut: &pb.PutRequest{
Key: keyf(),
Value: []byte("bar"),
},
},
})
}
addTxnOps := func(txn *pb.TxnRequest) {
newTxn := &pb.TxnRequest{}
addSuccessOps(newTxn)
txn.Success = append(txn.Success,
&pb.RequestOp{Request: &pb.RequestOp_RequestTxn{
RequestTxn: newTxn,
},
},
)
}
tests := []func(txn *pb.TxnRequest){
addCompareOps,
addSuccessOps,
addFailureOps,
addTxnOps,
}
for i, tt := range tests {
txn := &pb.TxnRequest{}
for j := 0; j < int(maxTxnOps+1); j++ {
tt(txn)
}
_, err := kvc.Txn(context.Background(), txn)
if !eqErrGRPC(err, rpctypes.ErrGRPCTooManyOps) {
t.Errorf("#%d: err = %v, want %v", i, err, rpctypes.ErrGRPCTooManyOps)
}
}
}
func TestV3TxnDuplicateKeys(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
defer clus.Terminate(t)
putreq := &pb.RequestOp{Request: &pb.RequestOp_RequestPut{RequestPut: &pb.PutRequest{Key: []byte("abc"), Value: []byte("def")}}}
delKeyReq := &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{
RequestDeleteRange: &pb.DeleteRangeRequest{
Key: []byte("abc"),
},
},
}
delInRangeReq := &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{
RequestDeleteRange: &pb.DeleteRangeRequest{
Key: []byte("a"), RangeEnd: []byte("b"),
},
},
}
delOutOfRangeReq := &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{
RequestDeleteRange: &pb.DeleteRangeRequest{
Key: []byte("abb"), RangeEnd: []byte("abc"),
},
},
}
txnDelReq := &pb.RequestOp{Request: &pb.RequestOp_RequestTxn{
RequestTxn: &pb.TxnRequest{Success: []*pb.RequestOp{delInRangeReq}},
},
}
txnDelReqTwoSide := &pb.RequestOp{Request: &pb.RequestOp_RequestTxn{
RequestTxn: &pb.TxnRequest{
Success: []*pb.RequestOp{delInRangeReq},
Failure: []*pb.RequestOp{delInRangeReq}},
},
}
txnPutReq := &pb.RequestOp{Request: &pb.RequestOp_RequestTxn{
RequestTxn: &pb.TxnRequest{Success: []*pb.RequestOp{putreq}},
},
}
txnPutReqTwoSide := &pb.RequestOp{Request: &pb.RequestOp_RequestTxn{
RequestTxn: &pb.TxnRequest{
Success: []*pb.RequestOp{putreq},
Failure: []*pb.RequestOp{putreq}},
},
}
kvc := toGRPC(clus.RandClient()).KV
tests := []struct {
txnSuccess []*pb.RequestOp
werr error
}{
{
txnSuccess: []*pb.RequestOp{putreq, putreq},
werr: rpctypes.ErrGRPCDuplicateKey,
},
{
txnSuccess: []*pb.RequestOp{putreq, delKeyReq},
werr: rpctypes.ErrGRPCDuplicateKey,
},
{
txnSuccess: []*pb.RequestOp{putreq, delInRangeReq},
werr: rpctypes.ErrGRPCDuplicateKey,
},
// Then(Put(a), Then(Del(a)))
{
txnSuccess: []*pb.RequestOp{putreq, txnDelReq},
werr: rpctypes.ErrGRPCDuplicateKey,
},
// Then(Del(a), Then(Put(a)))
{
txnSuccess: []*pb.RequestOp{delInRangeReq, txnPutReq},
werr: rpctypes.ErrGRPCDuplicateKey,
},
// Then((Then(Put(a)), Else(Put(a))), (Then(Put(a)), Else(Put(a)))
{
txnSuccess: []*pb.RequestOp{txnPutReqTwoSide, txnPutReqTwoSide},
werr: rpctypes.ErrGRPCDuplicateKey,
},
// Then(Del(x), (Then(Put(a)), Else(Put(a))))
{
txnSuccess: []*pb.RequestOp{delOutOfRangeReq, txnPutReqTwoSide},
werr: nil,
},
// Then(Then(Del(a)), (Then(Del(a)), Else(Del(a))))
{
txnSuccess: []*pb.RequestOp{txnDelReq, txnDelReqTwoSide},
werr: nil,
},
{
txnSuccess: []*pb.RequestOp{delKeyReq, delInRangeReq, delKeyReq, delInRangeReq},
werr: nil,
},
{
txnSuccess: []*pb.RequestOp{putreq, delOutOfRangeReq},
werr: nil,
},
}
for i, tt := range tests {
txn := &pb.TxnRequest{Success: tt.txnSuccess}
_, err := kvc.Txn(context.Background(), txn)
if !eqErrGRPC(err, tt.werr) {
t.Errorf("#%d: err = %v, want %v", i, err, tt.werr)
}
}
}
// Testv3TxnRevision tests that the transaction header revision is set as expected.
func TestV3TxnRevision(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
pr := &pb.PutRequest{Key: []byte("abc"), Value: []byte("def")}
presp, err := kvc.Put(context.TODO(), pr)
if err != nil {
t.Fatal(err)
}
txnget := &pb.RequestOp{Request: &pb.RequestOp_RequestRange{RequestRange: &pb.RangeRequest{Key: []byte("abc")}}}
txn := &pb.TxnRequest{Success: []*pb.RequestOp{txnget}}
tresp, err := kvc.Txn(context.TODO(), txn)
if err != nil {
t.Fatal(err)
}
// did not update revision
if presp.Header.Revision != tresp.Header.Revision {
t.Fatalf("got rev %d, wanted rev %d", tresp.Header.Revision, presp.Header.Revision)
}
txndr := &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{RequestDeleteRange: &pb.DeleteRangeRequest{Key: []byte("def")}}}
txn = &pb.TxnRequest{Success: []*pb.RequestOp{txndr}}
tresp, err = kvc.Txn(context.TODO(), txn)
if err != nil {
t.Fatal(err)
}
// did not update revision
if presp.Header.Revision != tresp.Header.Revision {
t.Fatalf("got rev %d, wanted rev %d", tresp.Header.Revision, presp.Header.Revision)
}
txnput := &pb.RequestOp{Request: &pb.RequestOp_RequestPut{RequestPut: &pb.PutRequest{Key: []byte("abc"), Value: []byte("123")}}}
txn = &pb.TxnRequest{Success: []*pb.RequestOp{txnput}}
tresp, err = kvc.Txn(context.TODO(), txn)
if err != nil {
t.Fatal(err)
}
// updated revision
if tresp.Header.Revision != presp.Header.Revision+1 {
t.Fatalf("got rev %d, wanted rev %d", tresp.Header.Revision, presp.Header.Revision+1)
}
}
// Testv3TxnCmpHeaderRev tests that the txn header revision is set as expected
// when compared to the Succeeded field in the txn response.
func TestV3TxnCmpHeaderRev(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
for i := 0; i < 10; i++ {
// Concurrently put a key with a txn comparing on it.
revc := make(chan int64, 1)
go func() {
defer close(revc)
pr := &pb.PutRequest{Key: []byte("k"), Value: []byte("v")}
presp, err := kvc.Put(context.TODO(), pr)
if err != nil {
t.Fatal(err)
}
revc <- presp.Header.Revision
}()
// The read-only txn uses the optimized readindex server path.
txnget := &pb.RequestOp{Request: &pb.RequestOp_RequestRange{
RequestRange: &pb.RangeRequest{Key: []byte("k")}}}
txn := &pb.TxnRequest{Success: []*pb.RequestOp{txnget}}
// i = 0 /\ Succeeded => put followed txn
cmp := &pb.Compare{
Result: pb.Compare_EQUAL,
Target: pb.Compare_VERSION,
Key: []byte("k"),
TargetUnion: &pb.Compare_Version{Version: int64(i)},
}
txn.Compare = append(txn.Compare, cmp)
tresp, err := kvc.Txn(context.TODO(), txn)
if err != nil {
t.Fatal(err)
}
prev := <-revc
// put followed txn; should eval to false
if prev > tresp.Header.Revision && !tresp.Succeeded {
t.Errorf("#%d: got else but put rev %d followed txn rev (%+v)", i, prev, tresp)
}
// txn follows put; should eval to true
if tresp.Header.Revision >= prev && tresp.Succeeded {
t.Errorf("#%d: got then but put rev %d preceded txn (%+v)", i, prev, tresp)
}
}
}
// TestV3TxnRangeCompare tests range comparisons in txns
func TestV3TxnRangeCompare(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)
// put keys, named by expected revision
for _, k := range []string{"/a/2", "/a/3", "/a/4", "/f/5"} {
if _, err := clus.Client(0).Put(context.TODO(), k, "x"); err != nil {
t.Fatal(err)
}
}
tests := []struct {
cmp pb.Compare
wSuccess bool
}{
{
// >= /a/; all create revs fit
pb.Compare{
Key: []byte("/a/"),
RangeEnd: []byte{0},
Target: pb.Compare_CREATE,
Result: pb.Compare_LESS,
TargetUnion: &pb.Compare_CreateRevision{6},
},
true,
},
{
// >= /a/; one create rev doesn't fit
pb.Compare{
Key: []byte("/a/"),
RangeEnd: []byte{0},
Target: pb.Compare_CREATE,
Result: pb.Compare_LESS,
TargetUnion: &pb.Compare_CreateRevision{5},
},
false,
},
{
// prefix /a/*; all create revs fit
pb.Compare{
Key: []byte("/a/"),
RangeEnd: []byte("/a0"),
Target: pb.Compare_CREATE,
Result: pb.Compare_LESS,
TargetUnion: &pb.Compare_CreateRevision{5},
},
true,
},
{
// prefix /a/*; one create rev doesn't fit
pb.Compare{
Key: []byte("/a/"),
RangeEnd: []byte("/a0"),
Target: pb.Compare_CREATE,
Result: pb.Compare_LESS,
TargetUnion: &pb.Compare_CreateRevision{4},
},
false,
},
{
// does not exist, does not succeed
pb.Compare{
Key: []byte("/b/"),
RangeEnd: []byte("/b0"),
Target: pb.Compare_VALUE,
Result: pb.Compare_EQUAL,
TargetUnion: &pb.Compare_Value{[]byte("x")},
},
false,
},
{
// all keys are leased
pb.Compare{
Key: []byte("/a/"),
RangeEnd: []byte("/a0"),
Target: pb.Compare_LEASE,
Result: pb.Compare_GREATER,
TargetUnion: &pb.Compare_Lease{0},
},
false,
},
{
// no keys are leased
pb.Compare{
Key: []byte("/a/"),
RangeEnd: []byte("/a0"),
Target: pb.Compare_LEASE,
Result: pb.Compare_EQUAL,
TargetUnion: &pb.Compare_Lease{0},
},
true,
},
}
kvc := toGRPC(clus.Client(0)).KV
for i, tt := range tests {
txn := &pb.TxnRequest{}
txn.Compare = append(txn.Compare, &tt.cmp)
tresp, err := kvc.Txn(context.TODO(), txn)
if err != nil {
t.Fatal(err)
}
if tt.wSuccess != tresp.Succeeded {
t.Errorf("#%d: expected %v, got %v", i, tt.wSuccess, tresp.Succeeded)
}
}
}
// TestV3TxnNested tests nested txns follow paths as expected.
func TestV3TxnNestedPath(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
cmpTrue := &pb.Compare{
Result: pb.Compare_EQUAL,
Target: pb.Compare_VERSION,
Key: []byte("k"),
TargetUnion: &pb.Compare_Version{Version: int64(0)},
}
cmpFalse := &pb.Compare{
Result: pb.Compare_EQUAL,
Target: pb.Compare_VERSION,
Key: []byte("k"),
TargetUnion: &pb.Compare_Version{Version: int64(1)},
}
// generate random path to eval txns
topTxn := &pb.TxnRequest{}
txn := topTxn
txnPath := make([]bool, 10)
for i := range txnPath {
nextTxn := &pb.TxnRequest{}
op := &pb.RequestOp{Request: &pb.RequestOp_RequestTxn{RequestTxn: nextTxn}}
txnPath[i] = rand.Intn(2) == 0
if txnPath[i] {
txn.Compare = append(txn.Compare, cmpTrue)
txn.Success = append(txn.Success, op)
} else {
txn.Compare = append(txn.Compare, cmpFalse)
txn.Failure = append(txn.Failure, op)
}
txn = nextTxn
}
tresp, err := kvc.Txn(context.TODO(), topTxn)
if err != nil {
t.Fatal(err)
}
curTxnResp := tresp
for i := range txnPath {
if curTxnResp.Succeeded != txnPath[i] {
t.Fatalf("expected path %+v, got response %+v", txnPath, *tresp)
}
curTxnResp = curTxnResp.Responses[0].Response.(*pb.ResponseOp_ResponseTxn).ResponseTxn
}
}
// TestV3PutIgnoreValue ensures that writes with ignore_value overwrites with previous key-value pair.
func TestV3PutIgnoreValue(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
key, val := []byte("foo"), []byte("bar")
putReq := pb.PutRequest{Key: key, Value: val}
// create lease
lc := toGRPC(clus.RandClient()).Lease
lresp, err := lc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: 30})
if err != nil {
t.Fatal(err)
}
if lresp.Error != "" {
t.Fatal(lresp.Error)
}
tests := []struct {
putFunc func() error
putErr error
wleaseID int64
}{
{ // put failure for non-existent key
func() error {
preq := putReq
preq.IgnoreValue = true
_, err := kvc.Put(context.TODO(), &preq)
return err
},
rpctypes.ErrGRPCKeyNotFound,
0,
},
{ // txn failure for non-existent key
func() error {
preq := putReq
preq.Value = nil
preq.IgnoreValue = true
txn := &pb.TxnRequest{}
txn.Success = append(txn.Success, &pb.RequestOp{
Request: &pb.RequestOp_RequestPut{RequestPut: &preq}})
_, err := kvc.Txn(context.TODO(), txn)
return err
},
rpctypes.ErrGRPCKeyNotFound,
0,
},
{ // put success
func() error {
_, err := kvc.Put(context.TODO(), &putReq)
return err
},
nil,
0,
},
{ // txn success, attach lease
func() error {
preq := putReq
preq.Value = nil
preq.Lease = lresp.ID
preq.IgnoreValue = true
txn := &pb.TxnRequest{}
txn.Success = append(txn.Success, &pb.RequestOp{
Request: &pb.RequestOp_RequestPut{RequestPut: &preq}})
_, err := kvc.Txn(context.TODO(), txn)
return err
},
nil,
lresp.ID,
},
{ // non-empty value with ignore_value should error
func() error {
preq := putReq
preq.IgnoreValue = true
_, err := kvc.Put(context.TODO(), &preq)
return err
},
rpctypes.ErrGRPCValueProvided,
0,
},
{ // overwrite with previous value, ensure no prev-kv is returned and lease is detached
func() error {
preq := putReq
preq.Value = nil
preq.IgnoreValue = true
presp, err := kvc.Put(context.TODO(), &preq)
if err != nil {
return err
}
if presp.PrevKv != nil && len(presp.PrevKv.Key) != 0 {
return fmt.Errorf("unexexpected previous key-value %v", presp.PrevKv)
}
return nil
},
nil,
0,
},
{ // revoke lease, ensure detached key doesn't get deleted
func() error {
_, err := lc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
return err
},
nil,
0,
},
}
for i, tt := range tests {
if err := tt.putFunc(); !eqErrGRPC(err, tt.putErr) {
t.Fatalf("#%d: err expected %v, got %v", i, tt.putErr, err)
}
if tt.putErr != nil {
continue
}
rr, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: key})
if err != nil {
t.Fatalf("#%d: %v", i, err)
}
if len(rr.Kvs) != 1 {
t.Fatalf("#%d: len(rr.KVs) expected 1, got %d", i, len(rr.Kvs))
}
if !bytes.Equal(rr.Kvs[0].Value, val) {
t.Fatalf("#%d: value expected %q, got %q", i, val, rr.Kvs[0].Value)
}
if rr.Kvs[0].Lease != tt.wleaseID {
t.Fatalf("#%d: lease ID expected %d, got %d", i, tt.wleaseID, rr.Kvs[0].Lease)
}
}
}
// TestV3PutIgnoreLease ensures that writes with ignore_lease uses previous lease for the key overwrites.
func TestV3PutIgnoreLease(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
// create lease
lc := toGRPC(clus.RandClient()).Lease
lresp, err := lc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: 30})
if err != nil {
t.Fatal(err)
}
if lresp.Error != "" {
t.Fatal(lresp.Error)
}
key, val, val1 := []byte("zoo"), []byte("bar"), []byte("bar1")
putReq := pb.PutRequest{Key: key, Value: val}
tests := []struct {
putFunc func() error
putErr error
wleaseID int64
wvalue []byte
}{
{ // put failure for non-existent key
func() error {
preq := putReq
preq.IgnoreLease = true
_, err := kvc.Put(context.TODO(), &preq)
return err
},
rpctypes.ErrGRPCKeyNotFound,
0,
nil,
},
{ // txn failure for non-existent key
func() error {
preq := putReq
preq.IgnoreLease = true
txn := &pb.TxnRequest{}
txn.Success = append(txn.Success, &pb.RequestOp{
Request: &pb.RequestOp_RequestPut{RequestPut: &preq}})
_, err := kvc.Txn(context.TODO(), txn)
return err
},
rpctypes.ErrGRPCKeyNotFound,
0,
nil,
},
{ // put success
func() error {
preq := putReq
preq.Lease = lresp.ID
_, err := kvc.Put(context.TODO(), &preq)
return err
},
nil,
lresp.ID,
val,
},
{ // txn success, modify value using 'ignore_lease' and ensure lease is not detached
func() error {
preq := putReq
preq.Value = val1
preq.IgnoreLease = true
txn := &pb.TxnRequest{}
txn.Success = append(txn.Success, &pb.RequestOp{
Request: &pb.RequestOp_RequestPut{RequestPut: &preq}})
_, err := kvc.Txn(context.TODO(), txn)
return err
},
nil,
lresp.ID,
val1,
},
{ // non-empty lease with ignore_lease should error
func() error {
preq := putReq
preq.Lease = lresp.ID
preq.IgnoreLease = true
_, err := kvc.Put(context.TODO(), &preq)
return err
},
rpctypes.ErrGRPCLeaseProvided,
0,
nil,
},
{ // overwrite with previous value, ensure no prev-kv is returned and lease is detached
func() error {
presp, err := kvc.Put(context.TODO(), &putReq)
if err != nil {
return err
}
if presp.PrevKv != nil && len(presp.PrevKv.Key) != 0 {
return fmt.Errorf("unexexpected previous key-value %v", presp.PrevKv)
}
return nil
},
nil,
0,
val,
},
{ // revoke lease, ensure detached key doesn't get deleted
func() error {
_, err := lc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
return err
},
nil,
0,
val,
},
}
for i, tt := range tests {
if err := tt.putFunc(); !eqErrGRPC(err, tt.putErr) {
t.Fatalf("#%d: err expected %v, got %v", i, tt.putErr, err)
}
if tt.putErr != nil {
continue
}
rr, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: key})
if err != nil {
t.Fatalf("#%d: %v", i, err)
}
if len(rr.Kvs) != 1 {
t.Fatalf("#%d: len(rr.KVs) expected 1, got %d", i, len(rr.Kvs))
}
if !bytes.Equal(rr.Kvs[0].Value, tt.wvalue) {
t.Fatalf("#%d: value expected %q, got %q", i, val, rr.Kvs[0].Value)
}
if rr.Kvs[0].Lease != tt.wleaseID {
t.Fatalf("#%d: lease ID expected %d, got %d", i, tt.wleaseID, rr.Kvs[0].Lease)
}
}
}
// TestV3PutMissingLease ensures that a Put on a key with a bogus lease fails.
func TestV3PutMissingLease(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
key := []byte("foo")
preq := &pb.PutRequest{Key: key, Lease: 123456}
tests := []func(){
// put case
func() {
if presp, err := kvc.Put(context.TODO(), preq); err == nil {
t.Errorf("succeeded put key. req: %v. resp: %v", preq, presp)
}
},
// txn success case
func() {
txn := &pb.TxnRequest{}
txn.Success = append(txn.Success, &pb.RequestOp{
Request: &pb.RequestOp_RequestPut{
RequestPut: preq}})
if tresp, err := kvc.Txn(context.TODO(), txn); err == nil {
t.Errorf("succeeded txn success. req: %v. resp: %v", txn, tresp)
}
},
// txn failure case
func() {
txn := &pb.TxnRequest{}
txn.Failure = append(txn.Failure, &pb.RequestOp{
Request: &pb.RequestOp_RequestPut{
RequestPut: preq}})
cmp := &pb.Compare{
Result: pb.Compare_GREATER,
Target: pb.Compare_CREATE,
Key: []byte("bar"),
}
txn.Compare = append(txn.Compare, cmp)
if tresp, err := kvc.Txn(context.TODO(), txn); err == nil {
t.Errorf("succeeded txn failure. req: %v. resp: %v", txn, tresp)
}
},
// ignore bad lease in failure on success txn
func() {
txn := &pb.TxnRequest{}
rreq := &pb.RangeRequest{Key: []byte("bar")}
txn.Success = append(txn.Success, &pb.RequestOp{
Request: &pb.RequestOp_RequestRange{
RequestRange: rreq}})
txn.Failure = append(txn.Failure, &pb.RequestOp{
Request: &pb.RequestOp_RequestPut{
RequestPut: preq}})
if tresp, err := kvc.Txn(context.TODO(), txn); err != nil {
t.Errorf("failed good txn. req: %v. resp: %v", txn, tresp)
}
},
}
for i, f := range tests {
f()
// key shouldn't have been stored
rreq := &pb.RangeRequest{Key: key}
rresp, err := kvc.Range(context.TODO(), rreq)
if err != nil {
t.Errorf("#%d. could not rangereq (%v)", i, err)
} else if len(rresp.Kvs) != 0 {
t.Errorf("#%d. expected no keys, got %v", i, rresp)
}
}
}
// TestV3DeleteRange tests various edge cases in the DeleteRange API.
func TestV3DeleteRange(t *testing.T) {
defer testutil.AfterTest(t)
tests := []struct {
keySet []string
begin string
end string
prevKV bool
wantSet [][]byte
deleted int64
}{
// delete middle
{
[]string{"foo", "foo/abc", "fop"},
"foo/", "fop", false,
[][]byte{[]byte("foo"), []byte("fop")}, 1,
},
// no delete
{
[]string{"foo", "foo/abc", "fop"},
"foo/", "foo/", false,
[][]byte{[]byte("foo"), []byte("foo/abc"), []byte("fop")}, 0,
},
// delete first
{
[]string{"foo", "foo/abc", "fop"},
"fo", "fop", false,
[][]byte{[]byte("fop")}, 2,
},
// delete tail
{
[]string{"foo", "foo/abc", "fop"},
"foo/", "fos", false,
[][]byte{[]byte("foo")}, 2,
},
// delete exact
{
[]string{"foo", "foo/abc", "fop"},
"foo/abc", "", false,
[][]byte{[]byte("foo"), []byte("fop")}, 1,
},
// delete none, [x,x)
{
[]string{"foo"},
"foo", "foo", false,
[][]byte{[]byte("foo")}, 0,
},
// delete middle with preserveKVs set
{
[]string{"foo", "foo/abc", "fop"},
"foo/", "fop", true,
[][]byte{[]byte("foo"), []byte("fop")}, 1,
},
}
for i, tt := range tests {
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
kvc := toGRPC(clus.RandClient()).KV
ks := tt.keySet
for j := range ks {
reqput := &pb.PutRequest{Key: []byte(ks[j]), Value: []byte{}}
_, err := kvc.Put(context.TODO(), reqput)
if err != nil {
t.Fatalf("couldn't put key (%v)", err)
}
}
dreq := &pb.DeleteRangeRequest{
Key: []byte(tt.begin),
RangeEnd: []byte(tt.end),
PrevKv: tt.prevKV,
}
dresp, err := kvc.DeleteRange(context.TODO(), dreq)
if err != nil {
t.Fatalf("couldn't delete range on test %d (%v)", i, err)
}
if tt.deleted != dresp.Deleted {
t.Errorf("expected %d on test %v, got %d", tt.deleted, i, dresp.Deleted)
}
if tt.prevKV {
if len(dresp.PrevKvs) != int(dresp.Deleted) {
t.Errorf("preserve %d keys, want %d", len(dresp.PrevKvs), dresp.Deleted)
}
}
rreq := &pb.RangeRequest{Key: []byte{0x0}, RangeEnd: []byte{0xff}}
rresp, err := kvc.Range(context.TODO(), rreq)
if err != nil {
t.Errorf("couldn't get range on test %v (%v)", i, err)
}
if dresp.Header.Revision != rresp.Header.Revision {
t.Errorf("expected revision %v, got %v",
dresp.Header.Revision, rresp.Header.Revision)
}
keys := [][]byte{}
for j := range rresp.Kvs {
keys = append(keys, rresp.Kvs[j].Key)
}
if !reflect.DeepEqual(tt.wantSet, keys) {
t.Errorf("expected %v on test %v, got %v", tt.wantSet, i, keys)
}
// can't defer because tcp ports will be in use
clus.Terminate(t)
}
}
// TestV3TxnInvalidRange tests that invalid ranges are rejected in txns.
func TestV3TxnInvalidRange(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
preq := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
for i := 0; i < 3; i++ {
_, err := kvc.Put(context.Background(), preq)
if err != nil {
t.Fatalf("couldn't put key (%v)", err)
}
}
_, err := kvc.Compact(context.Background(), &pb.CompactionRequest{Revision: 2})
if err != nil {
t.Fatalf("couldn't compact kv space (%v)", err)
}
// future rev
txn := &pb.TxnRequest{}
txn.Success = append(txn.Success, &pb.RequestOp{
Request: &pb.RequestOp_RequestPut{
RequestPut: preq}})
rreq := &pb.RangeRequest{Key: []byte("foo"), Revision: 100}
txn.Success = append(txn.Success, &pb.RequestOp{
Request: &pb.RequestOp_RequestRange{
RequestRange: rreq}})
if _, err := kvc.Txn(context.TODO(), txn); !eqErrGRPC(err, rpctypes.ErrGRPCFutureRev) {
t.Errorf("err = %v, want %v", err, rpctypes.ErrGRPCFutureRev)
}
// compacted rev
tv, _ := txn.Success[1].Request.(*pb.RequestOp_RequestRange)
tv.RequestRange.Revision = 1
if _, err := kvc.Txn(context.TODO(), txn); !eqErrGRPC(err, rpctypes.ErrGRPCCompacted) {
t.Errorf("err = %v, want %v", err, rpctypes.ErrGRPCCompacted)
}
}
func TestV3TooLargeRequest(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
defer clus.Terminate(t)
kvc := toGRPC(clus.RandClient()).KV
// 2MB request value
largeV := make([]byte, 2*1024*1024)
preq := &pb.PutRequest{Key: []byte("foo"), Value: largeV}
_, err := kvc.Put(context.Background(), preq)
if !eqErrGRPC(err, rpctypes.ErrGRPCRequestTooLarge) {
t.Errorf("err = %v, want %v", err, rpctypes.ErrGRPCRequestTooLarge)
}
}
// TestV3Hash tests hash.
func TestV3Hash(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
defer clus.Terminate(t)
cli := clus.RandClient()
kvc := toGRPC(cli).KV
m := toGRPC(cli).Maintenance
preq := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
for i := 0; i < 3; i++ {
_, err := kvc.Put(context.Background(), preq)
if err != nil {
t.Fatalf("couldn't put key (%v)", err)
}
}
resp, err := m.Hash(context.Background(), &pb.HashRequest{})
if err != nil || resp.Hash == 0 {
t.Fatalf("couldn't hash (%v, hash %d)", err, resp.Hash)
}
}
// TestV3HashRestart ensures that hash stays the same after restart.
func TestV3HashRestart(t *testing.T) {
defer testutil.AfterTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)
cli := clus.RandClient()
resp, err := toGRPC(cli).Maintenance.Hash(context.Background(), &pb.HashRequest{})
if err != nil || resp.Hash == 0 {
t.Fatalf("couldn't hash (%v, hash %d)", err, resp.Hash)
}
hash1 := resp.Hash
clus.Members[0].Stop(t)
clus.Members[0].Restart(t)
clus.waitLeader(t, clus.Members)
kvc := toGRPC(clus.Client(0)).KV
waitForRestart(t, kvc)
cli = clus.RandClient()
resp, err = toGRPC(cli).Maintenance.Hash(context.Background(), &pb.HashRequest{})
if err != nil || resp.Hash == 0 {
t.Fatalf("couldn't hash (%v, hash %d)", err, resp.Hash)
}
hash2 := resp.Hash
if hash1 != hash2 {
t.Fatalf("hash expected %d, got %d", hash1, hash2)
}
}
// TestV3StorageQuotaAPI tests the V3 server respects quotas at the API layer
func TestV3StorageQuotaAPI(t *testing.T) {
defer testutil.AfterTest(t)
quotasize := int64(16 * os.Getpagesize())
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
// Set a quota on one node
clus.Members[0].QuotaBackendBytes = quotasize
clus.Members[0].Stop(t)
clus.Members[0].Restart(t)
defer clus.Terminate(t)
kvc := toGRPC(clus.Client(0)).KV
waitForRestart(t, kvc)
key := []byte("abc")
// test small put that fits in quota
smallbuf := make([]byte, 512)
if _, err := kvc.Put(context.TODO(), &pb.PutRequest{Key: key, Value: smallbuf}); err != nil {
t.Fatal(err)
}
// test big put
bigbuf := make([]byte, quotasize)
_, err := kvc.Put(context.TODO(), &pb.PutRequest{Key: key, Value: bigbuf})
if !eqErrGRPC(err, rpctypes.ErrGRPCNoSpace) {
t.Fatalf("big put got %v, expected %v", err, rpctypes.ErrGRPCNoSpace)
}
// test big txn
puttxn := &pb.RequestOp{
Request: &pb.RequestOp_RequestPut{
RequestPut: &pb.PutRequest{
Key: key,
Value: bigbuf,
},
},
}
txnreq := &pb.TxnRequest{}
txnreq.Success = append(txnreq.Success, puttxn)
_, txnerr := kvc.Txn(context.TODO(), txnreq)
if !eqErrGRPC(txnerr, rpctypes.ErrGRPCNoSpace) {
t.Fatalf("big txn got %v, expected %v", err, rpctypes.ErrGRPCNoSpace)
}
}
func TestV3RangeRequest(t *testing.T) {
defer testutil.AfterTest(t)
tests := []struct {
putKeys []string
reqs []pb.RangeRequest
wresps [][]string
wmores []bool
}{
// single key
{
[]string{"foo", "bar"},
[]pb.RangeRequest{
// exists
{Key: []byte("foo")},
// doesn't exist
{Key: []byte("baz")},
},
[][]string{
{"foo"},
{},
},
[]bool{false, false},
},
// multi-key
{
[]string{"a", "b", "c", "d", "e"},
[]pb.RangeRequest{
// all in range
{Key: []byte("a"), RangeEnd: []byte("z")},
// [b, d)
{Key: []byte("b"), RangeEnd: []byte("d")},
// out of range
{Key: []byte("f"), RangeEnd: []byte("z")},
// [c,c) = empty
{Key: []byte("c"), RangeEnd: []byte("c")},
// [d, b) = empty
{Key: []byte("d"), RangeEnd: []byte("b")},
// ["\0", "\0") => all in range
{Key: []byte{0}, RangeEnd: []byte{0}},
},
[][]string{
{"a", "b", "c", "d", "e"},
{"b", "c"},
{},
{},
{},
{"a", "b", "c", "d", "e"},
},
[]bool{false, false, false, false, false, false},
},
// revision
{
[]string{"a", "b", "c", "d", "e"},
[]pb.RangeRequest{
{Key: []byte("a"), RangeEnd: []byte("z"), Revision: 0},
{Key: []byte("a"), RangeEnd: []byte("z"), Revision: 1},
{Key: []byte("a"), RangeEnd: []byte("z"), Revision: 2},
{Key: []byte("a"), RangeEnd: []byte("z"), Revision: 3},
},
[][]string{
{"a", "b", "c", "d", "e"},
{},
{"a"},
{"a", "b"},
},
[]bool{false, false, false, false},
},
// limit
{
[]string{"foo", "bar"},
[]pb.RangeRequest{
// more
{Key: []byte("a"), RangeEnd: []byte("z"), Limit: 1},
// no more
{Key: []byte("a"), RangeEnd: []byte("z"), Limit: 2},
},
[][]string{
{"bar"},
{"bar", "foo"},
},
[]bool{true, false},
},
// sort
{
[]string{"b", "a", "c", "d", "c"},
[]pb.RangeRequest{
{
Key: []byte("a"), RangeEnd: []byte("z"),
Limit: 1,
SortOrder: pb.RangeRequest_ASCEND,
SortTarget: pb.RangeRequest_KEY,
},
{
Key: []byte("a"), RangeEnd: []byte("z"),
Limit: 1,
SortOrder: pb.RangeRequest_DESCEND,
SortTarget: pb.RangeRequest_KEY,
},
{
Key: []byte("a"), RangeEnd: []byte("z"),
Limit: 1,
SortOrder: pb.RangeRequest_ASCEND,
SortTarget: pb.RangeRequest_CREATE,
},
{
Key: []byte("a"), RangeEnd: []byte("z"),
Limit: 1,
SortOrder: pb.RangeRequest_DESCEND,
SortTarget: pb.RangeRequest_MOD,
},
{
Key: []byte("z"), RangeEnd: []byte("z"),
Limit: 1,
SortOrder: pb.RangeRequest_DESCEND,
SortTarget: pb.RangeRequest_CREATE,
},
{ // sort ASCEND by default
Key: []byte("a"), RangeEnd: []byte("z"),
Limit: 10,
SortOrder: pb.RangeRequest_NONE,
SortTarget: pb.RangeRequest_CREATE,
},
},
[][]string{
{"a"},
{"d"},
{"b"},
{"c"},
{},
{"b", "a", "c", "d"},
},
[]bool{true, true, true, true, false, false},
},
// min/max mod rev
{
[]string{"rev2", "rev3", "rev4", "rev5", "rev6"},
[]pb.RangeRequest{
{
Key: []byte{0}, RangeEnd: []byte{0},
MinModRevision: 3,
},
{
Key: []byte{0}, RangeEnd: []byte{0},
MaxModRevision: 3,
},
{
Key: []byte{0}, RangeEnd: []byte{0},
MinModRevision: 3,
MaxModRevision: 5,
},
{
Key: []byte{0}, RangeEnd: []byte{0},
MaxModRevision: 10,
},
},
[][]string{
{"rev3", "rev4", "rev5", "rev6"},
{"rev2", "rev3"},
{"rev3", "rev4", "rev5"},
{"rev2", "rev3", "rev4", "rev5", "rev6"},
},
[]bool{false, false, false, false},
},
// min/max create rev
{
[]string{"rev2", "rev3", "rev2", "rev2", "rev6", "rev3"},
[]pb.RangeRequest{
{
Key: []byte{0}, RangeEnd: []byte{0},
MinCreateRevision: 3,
},
{
Key: []byte{0}, RangeEnd: []byte{0},
MaxCreateRevision: 3,
},
{
Key: []byte{0}, RangeEnd: []byte{0},
MinCreateRevision: 3,
MaxCreateRevision: 5,
},
{
Key: []byte{0}, RangeEnd: []byte{0},
MaxCreateRevision: 10,
},
},
[][]string{
{"rev3", "rev6"},
{"rev2", "rev3"},
{"rev3"},
{"rev2", "rev3", "rev6"},
},
[]bool{false, false, false, false},
},
}
for i, tt := range tests {
clus := NewClusterV3(t, &ClusterConfig{Size: 3})
for _, k := range tt.putKeys {
kvc := toGRPC(clus.RandClient()).KV
req := &pb.PutRequest{Key: []byte(k), Value: []byte("bar")}
if _, err := kvc.Put(context.TODO(), req); err != nil {
t.Fatalf("#%d: couldn't put key (%v)", i, err)
}
}
for j, req := range tt.reqs {
kvc := toGRPC(clus.RandClient()).KV
resp, err := kvc.Range(context.TODO(), &req)
if err != nil {
t.Errorf("#%d.%d: Range error: %v", i, j, err)
continue
}
if len(resp.Kvs) != len(tt.wresps[j]) {
t.Errorf("#%d.%d: bad len(resp.Kvs). got = %d, want = %d, ", i, j, len(resp.Kvs), len(tt.wresps[j]))
continue
}
for k, wKey := range tt.wresps[j] {
respKey := string(resp.Kvs[k].Key)
if respKey != wKey {
t.Errorf("#%d.%d: key[%d]. got = %v, want = %v, ", i, j, k, respKey, wKey)
}
}
if resp.More != tt.wmores[j] {
t.Errorf("#%d.%d: bad more. got = %v, want = %v, ", i, j, resp.More, tt.wmores[j])
}
wrev := int64(len(tt.putKeys) + 1)
if resp.Header.Revision != wrev {
t.Errorf("#%d.%d: bad header revision. got = %d. want = %d", i, j, resp.Header.Revision, wrev)
}
}
clus.Terminate(t)
}
}
func newClusterV3NoClients(t *testing.T, cfg *ClusterConfig) *ClusterV3 {
cfg.UseGRPC = true
clus := &ClusterV3{cluster: NewClusterByConfig(t, cfg)}
clus.Launch(t)
return clus
}
// TestTLSGRPCRejectInsecureClient checks that connection is rejected if server is TLS but not client.
func TestTLSGRPCRejectInsecureClient(t *testing.T) {
defer testutil.AfterTest(t)
cfg := ClusterConfig{Size: 3, ClientTLS: &testTLSInfo}
clus := newClusterV3NoClients(t, &cfg)
defer clus.Terminate(t)
// nil out TLS field so client will use an insecure connection
clus.Members[0].ClientTLSInfo = nil
client, err := NewClientV3(clus.Members[0])
if err != nil && err != context.DeadlineExceeded {
t.Fatalf("unexpected error (%v)", err)
} else if client == nil {
// Ideally, no client would be returned. However, grpc will
// return a connection without trying to handshake first so
// the connection appears OK.
return
}
defer client.Close()
donec := make(chan error, 1)
go func() {
ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
reqput := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
_, perr := toGRPC(client).KV.Put(ctx, reqput)
cancel()
donec <- perr
}()
if perr := <-donec; perr == nil {
t.Fatalf("expected client error on put")
}
}
// TestTLSGRPCRejectSecureClient checks that connection is rejected if client is TLS but not server.
func TestTLSGRPCRejectSecureClient(t *testing.T) {
defer testutil.AfterTest(t)
cfg := ClusterConfig{Size: 3}
clus := newClusterV3NoClients(t, &cfg)
defer clus.Terminate(t)
clus.Members[0].ClientTLSInfo = &testTLSInfo
client, err := NewClientV3(clus.Members[0])
if client != nil || err == nil {
t.Fatalf("expected no client")
} else if err != context.DeadlineExceeded {
t.Fatalf("unexpected error (%v)", err)
}
}
// TestTLSGRPCAcceptSecureAll checks that connection is accepted if both client and server are TLS
func TestTLSGRPCAcceptSecureAll(t *testing.T) {
defer testutil.AfterTest(t)
cfg := ClusterConfig{Size: 3, ClientTLS: &testTLSInfo}
clus := newClusterV3NoClients(t, &cfg)
defer clus.Terminate(t)
client, err := NewClientV3(clus.Members[0])
if err != nil {
t.Fatalf("expected tls client (%v)", err)
}
defer client.Close()
reqput := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
if _, err := toGRPC(client).KV.Put(context.TODO(), reqput); err != nil {
t.Fatalf("unexpected error on put over tls (%v)", err)
}
}
// TestTLSReloadAtomicReplace ensures server reloads expired/valid certs
// when all certs are atomically replaced by directory renaming.
// And expects server to reject client requests, and vice versa.
func TestTLSReloadAtomicReplace(t *testing.T) {
tmpDir, err := ioutil.TempDir(os.TempDir(), "fixtures-tmp")
if err != nil {
t.Fatal(err)
}
os.RemoveAll(tmpDir)
defer os.RemoveAll(tmpDir)
certsDir, err := ioutil.TempDir(os.TempDir(), "fixtures-to-load")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(certsDir)
certsDirExp, err := ioutil.TempDir(os.TempDir(), "fixtures-expired")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(certsDirExp)
cloneFunc := func() transport.TLSInfo {
tlsInfo, terr := copyTLSFiles(testTLSInfo, certsDir)
if terr != nil {
t.Fatal(terr)
}
if _, err = copyTLSFiles(testTLSInfoExpired, certsDirExp); err != nil {
t.Fatal(err)
}
return tlsInfo
}
replaceFunc := func() {
if err = os.Rename(certsDir, tmpDir); err != nil {
t.Fatal(err)
}
if err = os.Rename(certsDirExp, certsDir); err != nil {
t.Fatal(err)
}
// after rename,
// 'certsDir' contains expired certs
// 'tmpDir' contains valid certs
// 'certsDirExp' does not exist
}
revertFunc := func() {
if err = os.Rename(tmpDir, certsDirExp); err != nil {
t.Fatal(err)
}
if err = os.Rename(certsDir, tmpDir); err != nil {
t.Fatal(err)
}
if err = os.Rename(certsDirExp, certsDir); err != nil {
t.Fatal(err)
}
}
testTLSReload(t, cloneFunc, replaceFunc, revertFunc)
}
// TestTLSReloadCopy ensures server reloads expired/valid certs
// when new certs are copied over, one by one. And expects server
// to reject client requests, and vice versa.
func TestTLSReloadCopy(t *testing.T) {
certsDir, err := ioutil.TempDir(os.TempDir(), "fixtures-to-load")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(certsDir)
cloneFunc := func() transport.TLSInfo {
tlsInfo, terr := copyTLSFiles(testTLSInfo, certsDir)
if terr != nil {
t.Fatal(terr)
}
return tlsInfo
}
replaceFunc := func() {
if _, err = copyTLSFiles(testTLSInfoExpired, certsDir); err != nil {
t.Fatal(err)
}
}
revertFunc := func() {
if _, err = copyTLSFiles(testTLSInfo, certsDir); err != nil {
t.Fatal(err)
}
}
testTLSReload(t, cloneFunc, replaceFunc, revertFunc)
}
func testTLSReload(t *testing.T, cloneFunc func() transport.TLSInfo, replaceFunc func(), revertFunc func()) {
defer testutil.AfterTest(t)
// 1. separate copies for TLS assets modification
tlsInfo := cloneFunc()
// 2. start cluster with valid certs
clus := NewClusterV3(t, &ClusterConfig{Size: 1, PeerTLS: &tlsInfo, ClientTLS: &tlsInfo})
defer clus.Terminate(t)
// 3. concurrent client dialing while certs become expired
errc := make(chan error, 1)
go func() {
for {
cc, err := tlsInfo.ClientConfig()
if err != nil {
// errors in 'go/src/crypto/tls/tls.go'
// tls: private key does not match public key
// tls: failed to find any PEM data in key input
// tls: failed to find any PEM data in certificate input
// Or 'does not exist', 'not found', etc
t.Log(err)
continue
}
cli, cerr := clientv3.New(clientv3.Config{
Endpoints: []string{clus.Members[0].GRPCAddr()},
DialTimeout: time.Second,
DialOptions: []grpc.DialOption{grpc.WithBlock()},
TLS: cc,
})
if cerr != nil {
errc <- cerr
return
}
cli.Close()
}
}()
// 4. replace certs with expired ones
replaceFunc()
// 5. expect dial time-out when loading expired certs
select {
case gerr := <-errc:
if gerr != context.DeadlineExceeded {
t.Fatalf("expected %v, got %v", context.DeadlineExceeded, gerr)
}
case <-time.After(5 * time.Second):
t.Fatal("failed to receive dial timeout error")
}
// 6. replace expired certs back with valid ones
revertFunc()
// 7. new requests should trigger listener to reload valid certs
tls, terr := tlsInfo.ClientConfig()
if terr != nil {
t.Fatal(terr)
}
cl, cerr := clientv3.New(clientv3.Config{
Endpoints: []string{clus.Members[0].GRPCAddr()},
DialTimeout: 5 * time.Second,
DialOptions: []grpc.DialOption{grpc.WithBlock()},
TLS: tls,
})
if cerr != nil {
t.Fatalf("expected no error, got %v", cerr)
}
cl.Close()
}
func TestGRPCRequireLeader(t *testing.T) {
defer testutil.AfterTest(t)
cfg := ClusterConfig{Size: 3}
clus := newClusterV3NoClients(t, &cfg)
defer clus.Terminate(t)
clus.Members[1].Stop(t)
clus.Members[2].Stop(t)
client, err := NewClientV3(clus.Members[0])
if err != nil {
t.Fatalf("cannot create client: %v", err)
}
defer client.Close()
// wait for election timeout, then member[0] will not have a leader.
time.Sleep(time.Duration(3*electionTicks) * tickDuration)
md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
ctx := metadata.NewOutgoingContext(context.Background(), md)
reqput := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
if _, err := toGRPC(client).KV.Put(ctx, reqput); rpctypes.ErrorDesc(err) != rpctypes.ErrNoLeader.Error() {
t.Errorf("err = %v, want %v", err, rpctypes.ErrNoLeader)
}
}
func TestGRPCStreamRequireLeader(t *testing.T) {
defer testutil.AfterTest(t)
cfg := ClusterConfig{Size: 3}
clus := newClusterV3NoClients(t, &cfg)
defer clus.Terminate(t)
client, err := NewClientV3(clus.Members[0])
if err != nil {
t.Fatalf("failed to create client (%v)", err)
}
defer client.Close()
wAPI := toGRPC(client).Watch
md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
ctx := metadata.NewOutgoingContext(context.Background(), md)
wStream, err := wAPI.Watch(ctx)
if err != nil {
t.Fatalf("wAPI.Watch error: %v", err)
}
clus.Members[1].Stop(t)
clus.Members[2].Stop(t)
// existing stream should be rejected
_, err = wStream.Recv()
if rpctypes.ErrorDesc(err) != rpctypes.ErrNoLeader.Error() {
t.Errorf("err = %v, want %v", err, rpctypes.ErrNoLeader)
}
// new stream should also be rejected
wStream, err = wAPI.Watch(ctx)
if err != nil {
t.Fatalf("wAPI.Watch error: %v", err)
}
_, err = wStream.Recv()
if rpctypes.ErrorDesc(err) != rpctypes.ErrNoLeader.Error() {
t.Errorf("err = %v, want %v", err, rpctypes.ErrNoLeader)
}
clus.Members[1].Restart(t)
clus.Members[2].Restart(t)
clus.waitLeader(t, clus.Members)
time.Sleep(time.Duration(2*electionTicks) * tickDuration)
// new stream should also be OK now after we restarted the other members
wStream, err = wAPI.Watch(ctx)
if err != nil {
t.Fatalf("wAPI.Watch error: %v", err)
}
wreq := &pb.WatchRequest{
RequestUnion: &pb.WatchRequest_CreateRequest{
CreateRequest: &pb.WatchCreateRequest{Key: []byte("foo")},
},
}
err = wStream.Send(wreq)
if err != nil {
t.Errorf("err = %v, want nil", err)
}
}
// TestV3LargeRequests ensures that configurable MaxRequestBytes works as intended.
func TestV3LargeRequests(t *testing.T) {
defer testutil.AfterTest(t)
tests := []struct {
maxRequestBytes uint
valueSize int
expectError error
}{
// don't set to 0. use 0 as the default.
{1, 1024, rpctypes.ErrGRPCRequestTooLarge},
{10 * 1024 * 1024, 9 * 1024 * 1024, nil},
{10 * 1024 * 1024, 10 * 1024 * 1024, rpctypes.ErrGRPCRequestTooLarge},
{10 * 1024 * 1024, 10*1024*1024 + 5, rpctypes.ErrGRPCRequestTooLarge},
}
for i, test := range tests {
clus := NewClusterV3(t, &ClusterConfig{Size: 1, MaxRequestBytes: test.maxRequestBytes})
kvcli := toGRPC(clus.Client(0)).KV
reqput := &pb.PutRequest{Key: []byte("foo"), Value: make([]byte, test.valueSize)}
_, err := kvcli.Put(context.TODO(), reqput)
if !eqErrGRPC(err, test.expectError) {
t.Errorf("#%d: expected error %v, got %v", i, test.expectError, err)
}
// request went through, expect large response back from server
if test.expectError == nil {
reqget := &pb.RangeRequest{Key: []byte("foo")}
// limit receive call size with original value + gRPC overhead bytes
_, err = kvcli.Range(context.TODO(), reqget, grpc.MaxCallRecvMsgSize(test.valueSize+512*1024))
if err != nil {
t.Errorf("#%d: range expected no error, got %v", i, err)
}
}
clus.Terminate(t)
}
}
func eqErrGRPC(err1 error, err2 error) bool {
return !(err1 == nil && err2 != nil) || err1.Error() == err2.Error()
}
// waitForRestart tries a range request until the client's server responds.
// This is mainly a stop-gap function until grpcproxy's KVClient adapter
// (and by extension, clientv3) supports grpc.CallOption pass-through so
// FailFast=false works with Put.
func waitForRestart(t *testing.T, kvc pb.KVClient) {
req := &pb.RangeRequest{Key: []byte("_"), Serializable: true}
// TODO: Remove retry loop once the new grpc load balancer provides retry.
var err error
for i := 0; i < 10; i++ {
if _, err = kvc.Range(context.TODO(), req, grpc.FailFast(false)); err != nil {
if status, ok := status.FromError(err); ok && status.Code() == codes.Unavailable {
time.Sleep(time.Millisecond * 250)
} else {
t.Fatal(err)
}
}
}
if err != nil {
t.Fatalf("timed out waiting for restart: %v", err)
}
}
Computing file changes ...
