mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
Remove code used to make v2 proposals
Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>
This commit is contained in:
Marek Siarkowicz
parent
b4fd31f254
commit
dd7a4d28a8
@ -140,8 +140,6 @@ type ServerV2 interface {
|
|||||||
Server
|
Server
|
||||||
Leader() types.ID
|
Leader() types.ID
|
||||||
|
|
||||||
// Do takes a V2 request and attempts to fulfill it, returning a Response.
|
|
||||||
Do(ctx context.Context, r pb.Request) (Response, error)
|
|
||||||
ClientCertAuthEnabled() bool
|
ClientCertAuthEnabled() bool
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@ -62,123 +62,6 @@ import (
|
|||||||
"go.etcd.io/raft/v3/raftpb"
|
"go.etcd.io/raft/v3/raftpb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestDoLocalAction tests requests which do not need to go through raft to be applied,
|
|
||||||
// and are served through local data.
|
|
||||||
func TestDoLocalAction(t *testing.T) {
|
|
||||||
tests := []struct {
|
|
||||||
req pb.Request
|
|
||||||
|
|
||||||
wresp Response
|
|
||||||
werr error
|
|
||||||
wactions []testutil.Action
|
|
||||||
}{
|
|
||||||
{
|
|
||||||
pb.Request{Method: "GET", ID: 1, Wait: true},
|
|
||||||
Response{Watcher: v2store.NewNopWatcher()}, nil, []testutil.Action{{Name: "Watch"}},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
pb.Request{Method: "GET", ID: 1},
|
|
||||||
Response{Event: &v2store.Event{}}, nil,
|
|
||||||
[]testutil.Action{
|
|
||||||
{
|
|
||||||
Name: "Get",
|
|
||||||
Params: []any{"", false, false},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
pb.Request{Method: "HEAD", ID: 1},
|
|
||||||
Response{Event: &v2store.Event{}}, nil,
|
|
||||||
[]testutil.Action{
|
|
||||||
{
|
|
||||||
Name: "Get",
|
|
||||||
Params: []any{"", false, false},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
pb.Request{Method: "BADMETHOD", ID: 1},
|
|
||||||
Response{}, errors.ErrUnknownMethod, []testutil.Action{},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
for i, tt := range tests {
|
|
||||||
st := mockstore.NewRecorder()
|
|
||||||
srv := &EtcdServer{
|
|
||||||
lgMu: new(sync.RWMutex),
|
|
||||||
lg: zaptest.NewLogger(t),
|
|
||||||
v2store: st,
|
|
||||||
reqIDGen: idutil.NewGenerator(0, time.Time{}),
|
|
||||||
}
|
|
||||||
resp, err := srv.Do(context.Background(), tt.req)
|
|
||||||
|
|
||||||
if err != tt.werr {
|
|
||||||
t.Fatalf("#%d: err = %+v, want %+v", i, err, tt.werr)
|
|
||||||
}
|
|
||||||
if !reflect.DeepEqual(resp, tt.wresp) {
|
|
||||||
t.Errorf("#%d: resp = %+v, want %+v", i, resp, tt.wresp)
|
|
||||||
}
|
|
||||||
gaction := st.Action()
|
|
||||||
if !reflect.DeepEqual(gaction, tt.wactions) {
|
|
||||||
t.Errorf("#%d: action = %+v, want %+v", i, gaction, tt.wactions)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestDoBadLocalAction tests server requests which do not need to go through consensus,
|
|
||||||
// and return errors when they fetch from local data.
|
|
||||||
func TestDoBadLocalAction(t *testing.T) {
|
|
||||||
storeErr := fmt.Errorf("bah")
|
|
||||||
tests := []struct {
|
|
||||||
req pb.Request
|
|
||||||
|
|
||||||
wactions []testutil.Action
|
|
||||||
}{
|
|
||||||
{
|
|
||||||
pb.Request{Method: "GET", ID: 1, Wait: true},
|
|
||||||
[]testutil.Action{{Name: "Watch"}},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
pb.Request{Method: "GET", ID: 1},
|
|
||||||
[]testutil.Action{
|
|
||||||
{
|
|
||||||
Name: "Get",
|
|
||||||
Params: []any{"", false, false},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
pb.Request{Method: "HEAD", ID: 1},
|
|
||||||
[]testutil.Action{
|
|
||||||
{
|
|
||||||
Name: "Get",
|
|
||||||
Params: []any{"", false, false},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
for i, tt := range tests {
|
|
||||||
st := mockstore.NewErrRecorder(storeErr)
|
|
||||||
srv := &EtcdServer{
|
|
||||||
lgMu: new(sync.RWMutex),
|
|
||||||
lg: zaptest.NewLogger(t),
|
|
||||||
v2store: st,
|
|
||||||
reqIDGen: idutil.NewGenerator(0, time.Time{}),
|
|
||||||
}
|
|
||||||
resp, err := srv.Do(context.Background(), tt.req)
|
|
||||||
|
|
||||||
if err != storeErr {
|
|
||||||
t.Fatalf("#%d: err = %+v, want %+v", i, err, storeErr)
|
|
||||||
}
|
|
||||||
if !reflect.DeepEqual(resp, Response{}) {
|
|
||||||
t.Errorf("#%d: resp = %+v, want %+v", i, resp, Response{})
|
|
||||||
}
|
|
||||||
gaction := st.Action()
|
|
||||||
if !reflect.DeepEqual(gaction, tt.wactions) {
|
|
||||||
t.Errorf("#%d: action = %+v, want %+v", i, gaction, tt.wactions)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestApplyRepeat tests that server handles repeat raft messages gracefully
|
// TestApplyRepeat tests that server handles repeat raft messages gracefully
|
||||||
func TestApplyRepeat(t *testing.T) {
|
func TestApplyRepeat(t *testing.T) {
|
||||||
lg := zaptest.NewLogger(t)
|
lg := zaptest.NewLogger(t)
|
||||||
@ -795,115 +678,6 @@ func TestApplyMultiConfChangeShouldStop(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestDoProposal(t *testing.T) {
|
|
||||||
tests := []pb.Request{
|
|
||||||
{Method: "POST", ID: 1},
|
|
||||||
{Method: "PUT", ID: 1},
|
|
||||||
{Method: "DELETE", ID: 1},
|
|
||||||
{Method: "GET", ID: 1, Quorum: true},
|
|
||||||
}
|
|
||||||
for i, tt := range tests {
|
|
||||||
st := mockstore.NewRecorder()
|
|
||||||
r := newRaftNode(raftNodeConfig{
|
|
||||||
lg: zaptest.NewLogger(t),
|
|
||||||
Node: newNodeCommitter(),
|
|
||||||
storage: mockstorage.NewStorageRecorder(""),
|
|
||||||
raftStorage: raft.NewMemoryStorage(),
|
|
||||||
transport: newNopTransporter(),
|
|
||||||
})
|
|
||||||
srv := &EtcdServer{
|
|
||||||
lgMu: new(sync.RWMutex),
|
|
||||||
lg: zaptest.NewLogger(t),
|
|
||||||
Cfg: config.ServerConfig{Logger: zaptest.NewLogger(t), TickMs: 1, SnapshotCatchUpEntries: DefaultSnapshotCatchUpEntries},
|
|
||||||
r: *r,
|
|
||||||
v2store: st,
|
|
||||||
reqIDGen: idutil.NewGenerator(0, time.Time{}),
|
|
||||||
SyncTicker: &time.Ticker{},
|
|
||||||
consistIndex: cindex.NewFakeConsistentIndex(0),
|
|
||||||
}
|
|
||||||
srv.applyV2 = &applierV2store{store: srv.v2store, cluster: srv.cluster}
|
|
||||||
srv.start()
|
|
||||||
resp, err := srv.Do(context.Background(), tt)
|
|
||||||
srv.Stop()
|
|
||||||
|
|
||||||
action := st.Action()
|
|
||||||
if len(action) != 1 {
|
|
||||||
t.Errorf("#%d: len(action) = %d, want 1", i, len(action))
|
|
||||||
}
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("#%d: err = %v, want nil", i, err)
|
|
||||||
}
|
|
||||||
// resp.Index is set in Do() based on the raft state; may either be 0 or 1
|
|
||||||
wresp := Response{Event: &v2store.Event{}, Index: resp.Index}
|
|
||||||
if !reflect.DeepEqual(resp, wresp) {
|
|
||||||
t.Errorf("#%d: resp = %v, want %v", i, resp, wresp)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDoProposalCancelled(t *testing.T) {
|
|
||||||
wt := mockwait.NewRecorder()
|
|
||||||
srv := &EtcdServer{
|
|
||||||
lgMu: new(sync.RWMutex),
|
|
||||||
lg: zaptest.NewLogger(t),
|
|
||||||
Cfg: config.ServerConfig{Logger: zaptest.NewLogger(t), TickMs: 1, SnapshotCatchUpEntries: DefaultSnapshotCatchUpEntries},
|
|
||||||
r: *newRaftNode(raftNodeConfig{Node: newNodeNop()}),
|
|
||||||
w: wt,
|
|
||||||
reqIDGen: idutil.NewGenerator(0, time.Time{}),
|
|
||||||
}
|
|
||||||
srv.applyV2 = &applierV2store{store: srv.v2store, cluster: srv.cluster}
|
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
cancel()
|
|
||||||
_, err := srv.Do(ctx, pb.Request{Method: "PUT"})
|
|
||||||
|
|
||||||
if err != errors.ErrCanceled {
|
|
||||||
t.Fatalf("err = %v, want %v", err, errors.ErrCanceled)
|
|
||||||
}
|
|
||||||
w := []testutil.Action{{Name: "Register"}, {Name: "Trigger"}}
|
|
||||||
if !reflect.DeepEqual(wt.Action(), w) {
|
|
||||||
t.Errorf("wt.action = %+v, want %+v", wt.Action(), w)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDoProposalTimeout(t *testing.T) {
|
|
||||||
srv := &EtcdServer{
|
|
||||||
lgMu: new(sync.RWMutex),
|
|
||||||
lg: zaptest.NewLogger(t),
|
|
||||||
Cfg: config.ServerConfig{Logger: zaptest.NewLogger(t), TickMs: 1, SnapshotCatchUpEntries: DefaultSnapshotCatchUpEntries},
|
|
||||||
r: *newRaftNode(raftNodeConfig{Node: newNodeNop()}),
|
|
||||||
w: mockwait.NewNop(),
|
|
||||||
reqIDGen: idutil.NewGenerator(0, time.Time{}),
|
|
||||||
}
|
|
||||||
srv.applyV2 = &applierV2store{store: srv.v2store, cluster: srv.cluster}
|
|
||||||
|
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), 0)
|
|
||||||
_, err := srv.Do(ctx, pb.Request{Method: "PUT"})
|
|
||||||
cancel()
|
|
||||||
if err != errors.ErrTimeout {
|
|
||||||
t.Fatalf("err = %v, want %v", err, errors.ErrTimeout)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDoProposalStopped(t *testing.T) {
|
|
||||||
srv := &EtcdServer{
|
|
||||||
lgMu: new(sync.RWMutex),
|
|
||||||
lg: zaptest.NewLogger(t),
|
|
||||||
Cfg: config.ServerConfig{Logger: zaptest.NewLogger(t), TickMs: 1, SnapshotCatchUpEntries: DefaultSnapshotCatchUpEntries},
|
|
||||||
r: *newRaftNode(raftNodeConfig{lg: zaptest.NewLogger(t), Node: newNodeNop()}),
|
|
||||||
w: mockwait.NewNop(),
|
|
||||||
reqIDGen: idutil.NewGenerator(0, time.Time{}),
|
|
||||||
}
|
|
||||||
srv.applyV2 = &applierV2store{store: srv.v2store, cluster: srv.cluster}
|
|
||||||
|
|
||||||
srv.stopping = make(chan struct{})
|
|
||||||
close(srv.stopping)
|
|
||||||
_, err := srv.Do(context.Background(), pb.Request{Method: "PUT", ID: 1})
|
|
||||||
if err != errors.ErrStopped {
|
|
||||||
t.Errorf("err = %v, want %v", err, errors.ErrStopped)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestSync tests sync 1. is nonblocking 2. proposes SYNC request.
|
// TestSync tests sync 1. is nonblocking 2. proposes SYNC request.
|
||||||
func TestSync(t *testing.T) {
|
func TestSync(t *testing.T) {
|
||||||
n := newNodeRecorder()
|
n := newNodeRecorder()
|
||||||
@ -1190,73 +964,6 @@ func TestSnapshotOrdering(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestTriggerSnap for Applied > SnapshotCount should trigger a SaveSnap event
|
|
||||||
func TestTriggerSnap(t *testing.T) {
|
|
||||||
be, tmpPath := betesting.NewDefaultTmpBackend(t)
|
|
||||||
defer func() {
|
|
||||||
os.RemoveAll(tmpPath)
|
|
||||||
}()
|
|
||||||
|
|
||||||
snapc := 10
|
|
||||||
st := mockstore.NewRecorder()
|
|
||||||
p := mockstorage.NewStorageRecorderStream("")
|
|
||||||
r := newRaftNode(raftNodeConfig{
|
|
||||||
lg: zaptest.NewLogger(t),
|
|
||||||
Node: newNodeCommitter(),
|
|
||||||
raftStorage: raft.NewMemoryStorage(),
|
|
||||||
storage: p,
|
|
||||||
transport: newNopTransporter(),
|
|
||||||
})
|
|
||||||
srv := &EtcdServer{
|
|
||||||
lgMu: new(sync.RWMutex),
|
|
||||||
lg: zaptest.NewLogger(t),
|
|
||||||
Cfg: config.ServerConfig{Logger: zaptest.NewLogger(t), TickMs: 1, SnapshotCount: uint64(snapc), SnapshotCatchUpEntries: DefaultSnapshotCatchUpEntries},
|
|
||||||
r: *r,
|
|
||||||
v2store: st,
|
|
||||||
reqIDGen: idutil.NewGenerator(0, time.Time{}),
|
|
||||||
SyncTicker: &time.Ticker{},
|
|
||||||
consistIndex: cindex.NewConsistentIndex(be),
|
|
||||||
}
|
|
||||||
srv.applyV2 = &applierV2store{store: srv.v2store, cluster: srv.cluster}
|
|
||||||
|
|
||||||
srv.kv = mvcc.New(zaptest.NewLogger(t), be, &lease.FakeLessor{}, mvcc.StoreConfig{})
|
|
||||||
srv.be = be
|
|
||||||
|
|
||||||
cl := membership.NewCluster(zaptest.NewLogger(t))
|
|
||||||
srv.cluster = cl
|
|
||||||
|
|
||||||
srv.start()
|
|
||||||
|
|
||||||
donec := make(chan struct{})
|
|
||||||
go func() {
|
|
||||||
defer close(donec)
|
|
||||||
wcnt := 3 + snapc
|
|
||||||
gaction, _ := p.Wait(wcnt)
|
|
||||||
|
|
||||||
// each operation is recorded as a Save
|
|
||||||
// (SnapshotCount+1) * Puts + SaveSnap = (SnapshotCount+1) * Save + SaveSnap + Release
|
|
||||||
if len(gaction) != wcnt {
|
|
||||||
t.Logf("gaction: %v", gaction)
|
|
||||||
t.Errorf("len(action) = %d, want %d", len(gaction), wcnt)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if !reflect.DeepEqual(gaction[wcnt-2], testutil.Action{Name: "SaveSnap"}) {
|
|
||||||
t.Errorf("action = %s, want SaveSnap", gaction[wcnt-2])
|
|
||||||
}
|
|
||||||
|
|
||||||
if !reflect.DeepEqual(gaction[wcnt-1], testutil.Action{Name: "Release"}) {
|
|
||||||
t.Errorf("action = %s, want Release", gaction[wcnt-1])
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
for i := 0; i < snapc+1; i++ {
|
|
||||||
srv.Do(context.Background(), pb.Request{Method: "PUT"})
|
|
||||||
}
|
|
||||||
|
|
||||||
<-donec
|
|
||||||
srv.Stop()
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestConcurrentApplyAndSnapshotV3 will send out snapshots concurrently with
|
// TestConcurrentApplyAndSnapshotV3 will send out snapshots concurrently with
|
||||||
// proposals.
|
// proposals.
|
||||||
func TestConcurrentApplyAndSnapshotV3(t *testing.T) {
|
func TestConcurrentApplyAndSnapshotV3(t *testing.T) {
|
||||||
@ -1866,25 +1573,6 @@ func (n *nodeConfChangeCommitterRecorder) ApplyConfChange(conf raftpb.ConfChange
|
|||||||
return &raftpb.ConfState{}
|
return &raftpb.ConfState{}
|
||||||
}
|
}
|
||||||
|
|
||||||
// nodeCommitter commits proposed data immediately.
|
|
||||||
type nodeCommitter struct {
|
|
||||||
readyNode
|
|
||||||
index uint64
|
|
||||||
}
|
|
||||||
|
|
||||||
func newNodeCommitter() raft.Node {
|
|
||||||
return &nodeCommitter{*newNopReadyNode(), 0}
|
|
||||||
}
|
|
||||||
func (n *nodeCommitter) Propose(ctx context.Context, data []byte) error {
|
|
||||||
n.index++
|
|
||||||
ents := []raftpb.Entry{{Index: n.index, Data: data}}
|
|
||||||
n.readyc <- raft.Ready{
|
|
||||||
Entries: ents,
|
|
||||||
CommittedEntries: ents,
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func newTestCluster(t testing.TB) *membership.RaftCluster {
|
func newTestCluster(t testing.TB) *membership.RaftCluster {
|
||||||
return membership.NewCluster(zaptest.NewLogger(t))
|
return membership.NewCluster(zaptest.NewLogger(t))
|
||||||
}
|
}
|
||||||
|
|||||||
@ -15,152 +15,11 @@
|
|||||||
package etcdserver
|
package etcdserver
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
pb "go.etcd.io/etcd/api/v3/etcdserverpb"
|
pb "go.etcd.io/etcd/api/v3/etcdserverpb"
|
||||||
"go.etcd.io/etcd/server/v3/etcdserver/api/membership"
|
|
||||||
"go.etcd.io/etcd/server/v3/etcdserver/api/v2store"
|
|
||||||
"go.etcd.io/etcd/server/v3/etcdserver/errors"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
type RequestV2 pb.Request
|
type RequestV2 pb.Request
|
||||||
|
|
||||||
type RequestV2Handler interface {
|
|
||||||
Post(ctx context.Context, r *RequestV2) (Response, error)
|
|
||||||
Put(ctx context.Context, r *RequestV2) (Response, error)
|
|
||||||
Delete(ctx context.Context, r *RequestV2) (Response, error)
|
|
||||||
QGet(ctx context.Context, r *RequestV2) (Response, error)
|
|
||||||
Get(ctx context.Context, r *RequestV2) (Response, error)
|
|
||||||
Head(ctx context.Context, r *RequestV2) (Response, error)
|
|
||||||
}
|
|
||||||
|
|
||||||
type reqV2HandlerEtcdServer struct {
|
|
||||||
reqV2HandlerStore
|
|
||||||
s *EtcdServer
|
|
||||||
}
|
|
||||||
|
|
||||||
type reqV2HandlerStore struct {
|
|
||||||
store v2store.Store
|
|
||||||
applier ApplierV2
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewStoreRequestV2Handler(s v2store.Store, applier ApplierV2) RequestV2Handler {
|
|
||||||
return &reqV2HandlerStore{s, applier}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerStore) Post(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
return a.applier.Post(r), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerStore) Put(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
return a.applier.Put(r, membership.ApplyBoth), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerStore) Delete(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
return a.applier.Delete(r), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerStore) QGet(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
return a.applier.QGet(r), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerStore) Get(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
if r.Wait {
|
|
||||||
wc, err := a.store.Watch(r.Path, r.Recursive, r.Stream, r.Since)
|
|
||||||
return Response{Watcher: wc}, err
|
|
||||||
}
|
|
||||||
ev, err := a.store.Get(r.Path, r.Recursive, r.Sorted)
|
|
||||||
return Response{Event: ev}, err
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerStore) Head(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
ev, err := a.store.Get(r.Path, r.Recursive, r.Sorted)
|
|
||||||
return Response{Event: ev}, err
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerEtcdServer) Post(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
return a.processRaftRequest(ctx, r)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerEtcdServer) Put(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
return a.processRaftRequest(ctx, r)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerEtcdServer) Delete(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
return a.processRaftRequest(ctx, r)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerEtcdServer) QGet(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
return a.processRaftRequest(ctx, r)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *reqV2HandlerEtcdServer) processRaftRequest(ctx context.Context, r *RequestV2) (Response, error) {
|
|
||||||
data, err := ((*pb.Request)(r)).Marshal()
|
|
||||||
if err != nil {
|
|
||||||
return Response{}, err
|
|
||||||
}
|
|
||||||
ch := a.s.w.Register(r.ID)
|
|
||||||
|
|
||||||
start := time.Now()
|
|
||||||
a.s.r.Propose(ctx, data)
|
|
||||||
proposalsPending.Inc()
|
|
||||||
defer proposalsPending.Dec()
|
|
||||||
|
|
||||||
select {
|
|
||||||
case x := <-ch:
|
|
||||||
resp := x.(Response)
|
|
||||||
return resp, resp.Err
|
|
||||||
case <-ctx.Done():
|
|
||||||
proposalsFailed.Inc()
|
|
||||||
a.s.w.Trigger(r.ID, nil) // GC wait
|
|
||||||
return Response{}, a.s.parseProposeCtxErr(ctx.Err(), start)
|
|
||||||
case <-a.s.stopping:
|
|
||||||
}
|
|
||||||
return Response{}, errors.ErrStopped
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *EtcdServer) Do(ctx context.Context, r pb.Request) (Response, error) {
|
|
||||||
r.ID = s.reqIDGen.Next()
|
|
||||||
h := &reqV2HandlerEtcdServer{
|
|
||||||
reqV2HandlerStore: reqV2HandlerStore{
|
|
||||||
store: s.v2store,
|
|
||||||
applier: s.applyV2,
|
|
||||||
},
|
|
||||||
s: s,
|
|
||||||
}
|
|
||||||
rp := &r
|
|
||||||
resp, err := ((*RequestV2)(rp)).Handle(ctx, h)
|
|
||||||
resp.Term, resp.Index = s.Term(), s.CommittedIndex()
|
|
||||||
return resp, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Handle interprets r and performs an operation on s.store according to r.Method
|
|
||||||
// and other fields. If r.Method is "POST", "PUT", "DELETE", or a "GET" with
|
|
||||||
// Quorum == true, r will be sent through consensus before performing its
|
|
||||||
// respective operation. Do will block until an action is performed or there is
|
|
||||||
// an error.
|
|
||||||
func (r *RequestV2) Handle(ctx context.Context, v2api RequestV2Handler) (Response, error) {
|
|
||||||
if r.Method == "GET" && r.Quorum {
|
|
||||||
r.Method = "QGET"
|
|
||||||
}
|
|
||||||
switch r.Method {
|
|
||||||
case "POST":
|
|
||||||
return v2api.Post(ctx, r)
|
|
||||||
case "PUT":
|
|
||||||
return v2api.Put(ctx, r)
|
|
||||||
case "DELETE":
|
|
||||||
return v2api.Delete(ctx, r)
|
|
||||||
case "QGET":
|
|
||||||
return v2api.QGet(ctx, r)
|
|
||||||
case "GET":
|
|
||||||
return v2api.Get(ctx, r)
|
|
||||||
case "HEAD":
|
|
||||||
return v2api.Head(ctx, r)
|
|
||||||
}
|
|
||||||
return Response{}, errors.ErrUnknownMethod
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *RequestV2) String() string {
|
func (r *RequestV2) String() string {
|
||||||
rpb := pb.Request(*r)
|
rpb := pb.Request(*r)
|
||||||
return rpb.String()
|
return rpb.String()
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user