mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
etcdserver: add timeout for processing v3 request
This commit is contained in:
parent
8df52dc6fa
commit
6ee5f9c677
@ -30,6 +30,9 @@ const (
|
|||||||
// accept large request which might block raft stream. User
|
// accept large request which might block raft stream. User
|
||||||
// specify a large value might end up with shooting in the foot.
|
// specify a large value might end up with shooting in the foot.
|
||||||
maxRequestBytes = 1.5 * 1024 * 1024
|
maxRequestBytes = 1.5 * 1024 * 1024
|
||||||
|
|
||||||
|
// max timeout for waiting a v3 request to go through raft.
|
||||||
|
maxV3RequestTimeout = 5 * time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
type RaftKV interface {
|
type RaftKV interface {
|
||||||
@ -283,14 +286,17 @@ func (s *EtcdServer) processInternalRaftRequest(ctx context.Context, r pb.Intern
|
|||||||
|
|
||||||
ch := s.w.Register(r.ID)
|
ch := s.w.Register(r.ID)
|
||||||
|
|
||||||
s.r.Propose(ctx, data)
|
cctx, cancel := context.WithTimeout(ctx, maxV3RequestTimeout)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
s.r.Propose(cctx, data)
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case x := <-ch:
|
case x := <-ch:
|
||||||
return x.(*applyResult), nil
|
return x.(*applyResult), nil
|
||||||
case <-ctx.Done():
|
case <-cctx.Done():
|
||||||
s.w.Trigger(r.ID, nil) // GC wait
|
s.w.Trigger(r.ID, nil) // GC wait
|
||||||
return nil, ctx.Err()
|
return nil, cctx.Err()
|
||||||
case <-s.done:
|
case <-s.done:
|
||||||
return nil, ErrStopped
|
return nil, ErrStopped
|
||||||
}
|
}
|
||||||
|
@ -16,6 +16,7 @@ package integration
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"math/rand"
|
||||||
"reflect"
|
"reflect"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@ -74,6 +75,41 @@ func TestV3PutOverwrite(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestPutRestart checks if a put after an unrelated member restart succeeds
|
||||||
|
func TestV3PutRestart(t *testing.T) {
|
||||||
|
// this test might block for 5 seconds, make it parallel to speed up the test.
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
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.Fatal(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.
|
// TestV3CompactCurrentRev ensures keys are present when compacting on current revision.
|
||||||
func TestV3CompactCurrentRev(t *testing.T) {
|
func TestV3CompactCurrentRev(t *testing.T) {
|
||||||
defer testutil.AfterTest(t)
|
defer testutil.AfterTest(t)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user