etcd/etcdserver/server_test.go

1395 lines
37 KiB
Go
Raw Normal View History

// Copyright 2015 CoreOS, Inc.
//
// 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.
2014-08-26 05:39:02 +04:00
package etcdserver
2014-08-29 03:41:42 +04:00
import (
2014-09-27 01:18:59 +04:00
"encoding/json"
"fmt"
"io/ioutil"
"log"
"net/http"
"os"
2014-10-22 04:54:21 +04:00
"path"
2014-08-29 03:41:42 +04:00
"reflect"
"strconv"
2014-08-29 03:41:42 +04:00
"testing"
2014-08-30 02:32:41 +04:00
"time"
2014-09-03 08:36:14 +04:00
"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
2014-09-03 08:36:14 +04:00
pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
"github.com/coreos/etcd/pkg/idutil"
"github.com/coreos/etcd/pkg/pbutil"
2014-10-27 19:46:37 +03:00
"github.com/coreos/etcd/pkg/testutil"
"github.com/coreos/etcd/pkg/types"
2014-08-29 03:41:42 +04:00
"github.com/coreos/etcd/raft"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/store"
)
// 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
2015-01-02 22:21:23 +03:00
wactions []testutil.Action
}{
{
pb.Request{Method: "GET", ID: 1, Wait: true},
2015-01-02 22:21:23 +03:00
Response{Watcher: &nopWatcher{}}, nil, []testutil.Action{{Name: "Watch"}},
},
{
pb.Request{Method: "GET", ID: 1},
Response{Event: &store.Event{}}, nil,
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Get",
Params: []interface{}{"", false, false},
},
},
},
{
pb.Request{Method: "HEAD", ID: 1},
Response{Event: &store.Event{}}, nil,
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Get",
Params: []interface{}{"", false, false},
},
},
},
{
pb.Request{Method: "BADMETHOD", ID: 1},
2015-01-02 22:21:23 +03:00
Response{}, ErrUnknownMethod, []testutil.Action{},
},
}
for i, tt := range tests {
st := &storeRecorder{}
2014-12-16 03:07:47 +03:00
srv := &EtcdServer{
store: st,
reqIDGen: idutil.NewGenerator(0, time.Time{}),
}
resp, err := srv.Do(context.TODO(), 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)
}
}
}
2014-09-18 13:04:16 +04:00
// 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
2015-01-02 22:21:23 +03:00
wactions []testutil.Action
2014-09-18 13:04:16 +04:00
}{
{
pb.Request{Method: "GET", ID: 1, Wait: true},
2015-01-02 22:21:23 +03:00
[]testutil.Action{{Name: "Watch"}},
2014-09-18 13:04:16 +04:00
},
{
pb.Request{Method: "GET", ID: 1},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Get",
Params: []interface{}{"", false, false},
},
},
2014-09-18 13:04:16 +04:00
},
{
pb.Request{Method: "HEAD", ID: 1},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Get",
Params: []interface{}{"", false, false},
},
},
},
2014-09-18 13:04:16 +04:00
}
for i, tt := range tests {
st := &errStoreRecorder{err: storeErr}
2014-12-16 03:07:47 +03:00
srv := &EtcdServer{
store: st,
reqIDGen: idutil.NewGenerator(0, time.Time{}),
}
2014-09-18 13:04:16 +04:00
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)
2014-09-18 13:04:16 +04:00
}
}
}
func TestApplyRequest(t *testing.T) {
tests := []struct {
req pb.Request
wresp Response
2015-01-02 22:21:23 +03:00
wactions []testutil.Action
}{
// POST ==> Create
{
pb.Request{Method: "POST", ID: 1},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Create",
Params: []interface{}{"", false, "", true, time.Time{}},
},
},
},
2014-09-25 02:46:40 +04:00
// POST ==> Create, with expiration
{
pb.Request{Method: "POST", ID: 1, Expiration: 1337},
2014-09-25 02:46:40 +04:00
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Create",
Params: []interface{}{"", false, "", true, time.Unix(0, 1337)},
2014-09-25 02:46:40 +04:00
},
},
},
// POST ==> Create, with dir
{
pb.Request{Method: "POST", ID: 1, Dir: true},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Create",
Params: []interface{}{"", true, "", true, time.Time{}},
},
},
},
// PUT ==> Set
{
pb.Request{Method: "PUT", ID: 1},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Set",
Params: []interface{}{"", false, "", time.Time{}},
},
},
},
// PUT ==> Set, with dir
{
pb.Request{Method: "PUT", ID: 1, Dir: true},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Set",
Params: []interface{}{"", true, "", time.Time{}},
},
},
},
// PUT with PrevExist=true ==> Update
{
2014-12-31 01:51:26 +03:00
pb.Request{Method: "PUT", ID: 1, PrevExist: pbutil.Boolp(true)},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Update",
Params: []interface{}{"", "", time.Time{}},
},
},
},
// PUT with PrevExist=false ==> Create
{
2014-12-31 01:51:26 +03:00
pb.Request{Method: "PUT", ID: 1, PrevExist: pbutil.Boolp(false)},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Create",
Params: []interface{}{"", false, "", false, time.Time{}},
},
},
},
// PUT with PrevExist=true *and* PrevIndex set ==> Update
// TODO(jonboulle): is this expected?!
{
2014-12-31 01:51:26 +03:00
pb.Request{Method: "PUT", ID: 1, PrevExist: pbutil.Boolp(true), PrevIndex: 1},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Update",
Params: []interface{}{"", "", time.Time{}},
},
},
},
// PUT with PrevExist=false *and* PrevIndex set ==> Create
// TODO(jonboulle): is this expected?!
{
2014-12-31 01:51:26 +03:00
pb.Request{Method: "PUT", ID: 1, PrevExist: pbutil.Boolp(false), PrevIndex: 1},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Create",
Params: []interface{}{"", false, "", false, time.Time{}},
},
},
},
// PUT with PrevIndex set ==> CompareAndSwap
{
pb.Request{Method: "PUT", ID: 1, PrevIndex: 1},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "CompareAndSwap",
Params: []interface{}{"", "", uint64(1), "", time.Time{}},
},
},
},
// PUT with PrevValue set ==> CompareAndSwap
{
pb.Request{Method: "PUT", ID: 1, PrevValue: "bar"},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "CompareAndSwap",
Params: []interface{}{"", "bar", uint64(0), "", time.Time{}},
},
},
},
// PUT with PrevIndex and PrevValue set ==> CompareAndSwap
{
pb.Request{Method: "PUT", ID: 1, PrevIndex: 1, PrevValue: "bar"},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "CompareAndSwap",
Params: []interface{}{"", "bar", uint64(1), "", time.Time{}},
},
},
},
// DELETE ==> Delete
{
pb.Request{Method: "DELETE", ID: 1},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Delete",
Params: []interface{}{"", false, false},
},
},
},
// DELETE with PrevIndex set ==> CompareAndDelete
{
pb.Request{Method: "DELETE", ID: 1, PrevIndex: 1},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "CompareAndDelete",
Params: []interface{}{"", "", uint64(1)},
},
},
},
// DELETE with PrevValue set ==> CompareAndDelete
{
pb.Request{Method: "DELETE", ID: 1, PrevValue: "bar"},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "CompareAndDelete",
Params: []interface{}{"", "bar", uint64(0)},
},
},
},
// DELETE with PrevIndex *and* PrevValue set ==> CompareAndDelete
{
pb.Request{Method: "DELETE", ID: 1, PrevIndex: 5, PrevValue: "bar"},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "CompareAndDelete",
Params: []interface{}{"", "bar", uint64(5)},
},
},
},
// QGET ==> Get
{
pb.Request{Method: "QGET", ID: 1},
Response{Event: &store.Event{}},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "Get",
Params: []interface{}{"", false, false},
},
},
},
// SYNC ==> DeleteExpiredKeys
2014-09-16 04:35:02 +04:00
{
pb.Request{Method: "SYNC", ID: 1},
Response{},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "DeleteExpiredKeys",
Params: []interface{}{time.Unix(0, 0)},
},
},
2014-09-16 04:35:02 +04:00
},
{
pb.Request{Method: "SYNC", ID: 1, Time: 12345},
Response{},
2015-01-02 22:21:23 +03:00
[]testutil.Action{
{
Name: "DeleteExpiredKeys",
Params: []interface{}{time.Unix(0, 12345)},
},
},
},
// Unknown method - error
{
pb.Request{Method: "BADMETHOD", ID: 1},
Response{err: ErrUnknownMethod},
2015-01-02 22:21:23 +03:00
[]testutil.Action{},
},
}
for i, tt := range tests {
st := &storeRecorder{}
srv := &EtcdServer{store: st}
resp := srv.applyRequest(tt.req)
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)
}
}
}
func TestApplyRequestOnAdminMemberAttributes(t *testing.T) {
cl := newTestCluster([]*Member{{ID: 1}})
srv := &EtcdServer{
store: &storeRecorder{},
Cluster: cl,
}
req := pb.Request{
Method: "PUT",
ID: 1,
Path: path.Join(storeMembersPrefix, strconv.FormatUint(1, 16), attributesSuffix),
Val: `{"Name":"abc","ClientURLs":["http://127.0.0.1:4001"]}`,
}
srv.applyRequest(req)
w := Attributes{Name: "abc", ClientURLs: []string{"http://127.0.0.1:4001"}}
if g := cl.Member(1).Attributes; !reflect.DeepEqual(g, w) {
t.Errorf("attributes = %v, want %v", g, w)
}
}
func TestApplyConfChangeError(t *testing.T) {
cl := newCluster("")
cl.SetStore(store.New())
cl.SetTransport(&nopTransporter{})
for i := 1; i <= 4; i++ {
cl.AddMember(&Member{ID: types.ID(i)}, uint64(i))
}
cl.RemoveMember(4, 5)
tests := []struct {
cc raftpb.ConfChange
werr error
}{
{
raftpb.ConfChange{
Type: raftpb.ConfChangeAddNode,
NodeID: 4,
},
ErrIDRemoved,
},
{
raftpb.ConfChange{
Type: raftpb.ConfChangeUpdateNode,
NodeID: 4,
},
ErrIDRemoved,
},
{
raftpb.ConfChange{
Type: raftpb.ConfChangeAddNode,
NodeID: 1,
},
ErrIDExists,
},
{
raftpb.ConfChange{
Type: raftpb.ConfChangeRemoveNode,
NodeID: 5,
},
ErrIDNotFound,
},
}
for i, tt := range tests {
n := &nodeRecorder{}
srv := &EtcdServer{
r: raftNode{Node: n},
Cluster: cl,
}
_, err := srv.applyConfChange(tt.cc, nil, 10)
if err != tt.werr {
t.Errorf("#%d: applyConfChange error = %v, want %v", i, err, tt.werr)
}
cc := raftpb.ConfChange{Type: tt.cc.Type, NodeID: raft.None}
2015-01-02 22:21:23 +03:00
w := []testutil.Action{
{
2015-01-02 22:21:23 +03:00
Name: "ApplyConfChange",
Params: []interface{}{cc},
},
}
if g := n.Action(); !reflect.DeepEqual(g, w) {
t.Errorf("#%d: action = %+v, want %+v", i, g, w)
}
}
}
2014-11-14 10:03:34 +03:00
func TestApplyConfChangeShouldStop(t *testing.T) {
cl := newCluster("")
cl.SetStore(store.New())
cl.SetTransport(&nopTransporter{})
2014-11-14 10:03:34 +03:00
for i := 1; i <= 3; i++ {
cl.AddMember(&Member{ID: types.ID(i)}, uint64(i))
2014-11-14 10:03:34 +03:00
}
srv := &EtcdServer{
id: 1,
r: raftNode{
Node: &nodeRecorder{},
transport: &nopTransporter{},
},
Cluster: cl,
2014-11-14 10:03:34 +03:00
}
cc := raftpb.ConfChange{
Type: raftpb.ConfChangeRemoveNode,
NodeID: 2,
}
// remove non-local member
shouldStop, err := srv.applyConfChange(cc, &raftpb.ConfState{}, 10)
2014-11-14 10:03:34 +03:00
if err != nil {
t.Fatalf("unexpected error %v", err)
}
if shouldStop != false {
t.Errorf("shouldStop = %t, want %t", shouldStop, false)
}
// remove local member
cc.NodeID = 1
shouldStop, err = srv.applyConfChange(cc, &raftpb.ConfState{}, 10)
2014-11-14 10:03:34 +03:00
if err != nil {
t.Fatalf("unexpected error %v", err)
}
if shouldStop != true {
t.Errorf("shouldStop = %t, want %t", shouldStop, true)
}
}
func TestDoProposal(t *testing.T) {
tests := []pb.Request{
pb.Request{Method: "POST", ID: 1},
pb.Request{Method: "PUT", ID: 1},
pb.Request{Method: "DELETE", ID: 1},
pb.Request{Method: "GET", ID: 1, Quorum: true},
}
for i, tt := range tests {
st := &storeRecorder{}
srv := &EtcdServer{
r: raftNode{
Node: newNodeCommitter(),
storage: &storageRecorder{},
raftStorage: raft.NewMemoryStorage(),
transport: &nopTransporter{},
},
store: st,
reqIDGen: idutil.NewGenerator(0, time.Time{}),
}
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)
}
wresp := Response{Event: &store.Event{}}
if !reflect.DeepEqual(resp, wresp) {
t.Errorf("#%d: resp = %v, want %v", i, resp, wresp)
}
}
}
func TestDoProposalCancelled(t *testing.T) {
wait := &waitRecorder{}
srv := &EtcdServer{
r: raftNode{Node: &nodeRecorder{}},
w: wait,
reqIDGen: idutil.NewGenerator(0, time.Time{}),
}
ctx, cancel := context.WithCancel(context.Background())
cancel()
_, err := srv.Do(ctx, pb.Request{Method: "PUT"})
if err != ErrCanceled {
t.Fatalf("err = %v, want %v", err, ErrCanceled)
}
2015-01-02 22:21:23 +03:00
w := []testutil.Action{{Name: "Register"}, {Name: "Trigger"}}
if !reflect.DeepEqual(wait.action, w) {
t.Errorf("wait.action = %+v, want %+v", wait.action, w)
}
}
func TestDoProposalTimeout(t *testing.T) {
srv := &EtcdServer{
r: raftNode{Node: &nodeRecorder{}},
2014-12-16 03:07:47 +03:00
w: &waitRecorder{},
reqIDGen: idutil.NewGenerator(0, time.Time{}),
}
ctx, _ := context.WithTimeout(context.Background(), 0)
_, err := srv.Do(ctx, pb.Request{Method: "PUT"})
if err != ErrTimeout {
t.Fatalf("err = %v, want %v", err, ErrTimeout)
}
}
func TestDoProposalStopped(t *testing.T) {
srv := &EtcdServer{
r: raftNode{Node: &nodeRecorder{}},
w: &waitRecorder{},
reqIDGen: idutil.NewGenerator(0, time.Time{}),
}
srv.done = make(chan struct{})
close(srv.done)
_, err := srv.Do(context.Background(), pb.Request{Method: "PUT", ID: 1})
if err != ErrStopped {
t.Errorf("err = %v, want %v", err, ErrStopped)
}
}
// TestSync tests sync 1. is nonblocking 2. proposes SYNC request.
2014-09-16 04:35:02 +04:00
func TestSync(t *testing.T) {
n := &nodeRecorder{}
2014-09-16 04:35:02 +04:00
srv := &EtcdServer{
r: raftNode{Node: n},
2014-12-16 03:07:47 +03:00
reqIDGen: idutil.NewGenerator(0, time.Time{}),
2014-09-16 04:35:02 +04:00
}
// check that sync is non-blocking
timer := time.AfterFunc(time.Second, func() {
t.Fatalf("sync should be non-blocking but did not return after 1s!")
})
srv.sync(10 * time.Second)
timer.Stop()
2014-10-27 19:46:37 +03:00
testutil.ForceGosched()
action := n.Action()
if len(action) != 1 {
t.Fatalf("len(action) = %d, want 1", len(action))
}
2015-01-02 22:21:23 +03:00
if action[0].Name != "Propose" {
t.Fatalf("action = %s, want Propose", action[0].Name)
2014-09-16 04:35:02 +04:00
}
2015-01-02 22:21:23 +03:00
data := action[0].Params[0].([]byte)
var r pb.Request
if err := r.Unmarshal(data); err != nil {
t.Fatalf("unmarshal request error: %v", err)
2014-09-16 04:35:02 +04:00
}
if r.Method != "SYNC" {
t.Errorf("method = %s, want SYNC", r.Method)
2014-09-16 04:35:02 +04:00
}
}
// TestSyncTimeout tests the case that sync 1. is non-blocking 2. cancel request
// after timeout
func TestSyncTimeout(t *testing.T) {
n := &nodeProposalBlockerRecorder{}
2014-09-16 04:35:02 +04:00
srv := &EtcdServer{
r: raftNode{Node: n},
2014-12-16 03:07:47 +03:00
reqIDGen: idutil.NewGenerator(0, time.Time{}),
2014-09-16 04:35:02 +04:00
}
// check that sync is non-blocking
timer := time.AfterFunc(time.Second, func() {
t.Fatalf("sync should be non-blocking but did not return after 1s!")
})
srv.sync(0)
timer.Stop()
2014-09-16 04:35:02 +04:00
// give time for goroutine in sync to cancel
2014-10-27 19:46:37 +03:00
testutil.ForceGosched()
2015-01-02 22:21:23 +03:00
w := []testutil.Action{{Name: "Propose blocked"}}
if g := n.Action(); !reflect.DeepEqual(g, w) {
t.Errorf("action = %v, want %v", g, w)
2014-09-16 04:35:02 +04:00
}
}
// TODO: TestNoSyncWhenNoLeader
// TestSyncTrigger tests that the server proposes a SYNC request when its sync timer ticks
func TestSyncTrigger(t *testing.T) {
n := newReadyNode()
st := make(chan time.Time, 1)
2014-09-16 04:35:02 +04:00
srv := &EtcdServer{
r: raftNode{
Node: n,
raftStorage: raft.NewMemoryStorage(),
transport: &nopTransporter{},
storage: &storageRecorder{},
},
store: &storeRecorder{},
SyncTicker: st,
reqIDGen: idutil.NewGenerator(0, time.Time{}),
2014-09-16 04:35:02 +04:00
}
srv.start()
defer srv.Stop()
// trigger the server to become a leader and accept sync requests
n.readyc <- raft.Ready{
SoftState: &raft.SoftState{
RaftState: raft.StateLeader,
},
}
// trigger a sync request
st <- time.Time{}
testutil.ForceGosched()
action := n.Action()
if len(action) != 1 {
t.Fatalf("len(action) = %d, want 1", len(action))
}
2015-01-02 22:21:23 +03:00
if action[0].Name != "Propose" {
t.Fatalf("action = %s, want Propose", action[0].Name)
}
2015-01-02 22:21:23 +03:00
data := action[0].Params[0].([]byte)
var req pb.Request
if err := req.Unmarshal(data); err != nil {
t.Fatalf("error unmarshalling data: %v", err)
2014-09-16 04:35:02 +04:00
}
if req.Method != "SYNC" {
t.Fatalf("unexpected proposed request: %#v", req.Method)
2014-09-16 04:35:02 +04:00
}
}
2014-09-17 05:18:45 +04:00
// snapshot should snapshot the store and cut the persistent
func TestSnapshot(t *testing.T) {
s := raft.NewMemoryStorage()
s.Append([]raftpb.Entry{{Index: 1}})
2014-09-17 05:18:45 +04:00
st := &storeRecorder{}
p := &storageRecorder{}
srv := &EtcdServer{
r: raftNode{
Node: &nodeRecorder{},
raftStorage: s,
storage: p,
},
store: st,
2014-09-17 05:18:45 +04:00
}
srv.snapshot(1, raftpb.ConfState{Nodes: []uint64{1}})
testutil.ForceGosched()
gaction := st.Action()
if len(gaction) != 2 {
t.Fatalf("len(action) = %d, want 1", len(gaction))
2014-09-17 05:18:45 +04:00
}
if !reflect.DeepEqual(gaction[0], testutil.Action{Name: "Clone"}) {
t.Errorf("action = %s, want Clone", gaction[0])
}
if !reflect.DeepEqual(gaction[1], testutil.Action{Name: "SaveNoCopy"}) {
t.Errorf("action = %s, want SaveNoCopy", gaction[1])
2014-09-17 05:18:45 +04:00
}
gaction = p.Action()
if len(gaction) != 1 {
t.Fatalf("len(action) = %d, want 1", len(gaction))
2014-09-17 05:18:45 +04:00
}
if !reflect.DeepEqual(gaction[0], testutil.Action{Name: "SaveSnap"}) {
t.Errorf("action = %s, want SaveSnap", gaction[0])
2014-09-17 05:18:45 +04:00
}
}
// Applied > SnapCount should trigger a SaveSnap event
func TestTriggerSnap(t *testing.T) {
snapc := 10
2014-09-17 05:18:45 +04:00
st := &storeRecorder{}
p := &storageRecorder{}
srv := &EtcdServer{
snapCount: uint64(snapc),
r: raftNode{
Node: newNodeCommitter(),
raftStorage: raft.NewMemoryStorage(),
storage: p,
transport: &nopTransporter{},
},
store: st,
reqIDGen: idutil.NewGenerator(0, time.Time{}),
2014-09-17 05:18:45 +04:00
}
srv.start()
for i := 0; i < snapc+1; i++ {
srv.Do(context.Background(), pb.Request{Method: "PUT"})
2014-09-17 05:18:45 +04:00
}
srv.Stop()
2014-09-17 05:18:45 +04:00
gaction := p.Action()
2014-09-17 05:18:45 +04:00
// each operation is recorded as a Save
// (SnapCount+1) * Puts + SaveSnap = (SnapCount+1) * Save + SaveSnap
wcnt := 2 + snapc
2014-10-11 15:20:14 +04:00
if len(gaction) != wcnt {
t.Fatalf("len(action) = %d, want %d", len(gaction), wcnt)
2014-09-17 05:18:45 +04:00
}
if !reflect.DeepEqual(gaction[wcnt-1], testutil.Action{Name: "SaveSnap"}) {
2014-10-11 15:20:14 +04:00
t.Errorf("action = %s, want SaveSnap", gaction[wcnt-1])
2014-09-17 05:18:45 +04:00
}
}
2014-09-19 10:13:08 +04:00
// TestRecvSnapshot tests when it receives a snapshot from raft leader,
// it should trigger storage.SaveSnap and also store.Recover.
func TestRecvSnapshot(t *testing.T) {
n := newReadyNode()
2014-09-19 10:13:08 +04:00
st := &storeRecorder{}
p := &storageRecorder{}
cl := newCluster("abc")
cl.SetStore(store.New())
cl.SetTransport(&nopTransporter{})
2014-09-19 10:13:08 +04:00
s := &EtcdServer{
r: raftNode{
Node: n,
transport: &nopTransporter{},
storage: p,
raftStorage: raft.NewMemoryStorage(),
},
store: st,
Cluster: cl,
2014-09-19 10:13:08 +04:00
}
s.start()
n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 1}}}
2014-09-19 10:13:08 +04:00
// make goroutines move forward to receive snapshot
2014-10-27 19:46:37 +03:00
testutil.ForceGosched()
2014-09-19 10:13:08 +04:00
s.Stop()
2015-01-02 22:21:23 +03:00
wactions := []testutil.Action{{Name: "Recovery"}}
if g := st.Action(); !reflect.DeepEqual(g, wactions) {
t.Errorf("store action = %v, want %v", g, wactions)
2014-09-19 10:13:08 +04:00
}
2015-01-02 22:21:23 +03:00
wactions = []testutil.Action{{Name: "SaveSnap"}, {Name: "Save"}}
if g := p.Action(); !reflect.DeepEqual(g, wactions) {
t.Errorf("storage action = %v, want %v", g, wactions)
2014-09-19 10:13:08 +04:00
}
}
// TestApplySnapshotAndCommittedEntries tests that server applies snapshot
// first and then committed entries.
func TestApplySnapshotAndCommittedEntries(t *testing.T) {
n := newReadyNode()
st := &storeRecorder{}
cl := newCluster("abc")
cl.SetStore(store.New())
cl.SetTransport(&nopTransporter{})
storage := raft.NewMemoryStorage()
s := &EtcdServer{
r: raftNode{
Node: n,
storage: &storageRecorder{},
raftStorage: storage,
transport: &nopTransporter{},
},
store: st,
Cluster: cl,
}
s.start()
req := &pb.Request{Method: "QGET"}
n.readyc <- raft.Ready{
Snapshot: raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 1}},
CommittedEntries: []raftpb.Entry{
{Index: 2, Data: pbutil.MustMarshal(req)},
},
}
// make goroutines move forward to receive snapshot
testutil.ForceGosched()
s.Stop()
actions := st.Action()
if len(actions) != 2 {
t.Fatalf("len(action) = %d, want 2", len(actions))
}
2015-01-02 22:21:23 +03:00
if actions[0].Name != "Recovery" {
t.Errorf("actions[0] = %s, want %s", actions[0].Name, "Recovery")
}
2015-01-02 22:21:23 +03:00
if actions[1].Name != "Get" {
t.Errorf("actions[1] = %s, want %s", actions[1].Name, "Get")
}
}
// TestAddMember tests AddMember can propose and perform node addition.
func TestAddMember(t *testing.T) {
2014-09-23 23:02:44 +04:00
n := newNodeConfChangeCommitterRecorder()
n.readyc <- raft.Ready{
2014-12-09 08:17:04 +03:00
SoftState: &raft.SoftState{RaftState: raft.StateLeader},
}
cl := newTestCluster(nil)
st := store.New()
cl.SetStore(st)
cl.SetTransport(&nopTransporter{})
s := &EtcdServer{
r: raftNode{
Node: n,
raftStorage: raft.NewMemoryStorage(),
storage: &storageRecorder{},
transport: &nopTransporter{},
},
store: st,
Cluster: cl,
reqIDGen: idutil.NewGenerator(0, time.Time{}),
}
s.start()
m := Member{ID: 1234, RaftAttributes: RaftAttributes{PeerURLs: []string{"foo"}}}
err := s.AddMember(context.TODO(), m)
gaction := n.Action()
s.Stop()
if err != nil {
t.Fatalf("AddMember error: %v", err)
}
2015-01-02 22:21:23 +03:00
wactions := []testutil.Action{{Name: "ProposeConfChange:ConfChangeAddNode"}, {Name: "ApplyConfChange:ConfChangeAddNode"}}
if !reflect.DeepEqual(gaction, wactions) {
t.Errorf("action = %v, want %v", gaction, wactions)
}
if cl.Member(1234) == nil {
t.Errorf("member with id 1234 is not added")
}
}
// TestRemoveMember tests RemoveMember can propose and perform node removal.
func TestRemoveMember(t *testing.T) {
2014-09-23 23:02:44 +04:00
n := newNodeConfChangeCommitterRecorder()
n.readyc <- raft.Ready{
2014-12-09 08:17:04 +03:00
SoftState: &raft.SoftState{RaftState: raft.StateLeader},
}
cl := newTestCluster(nil)
st := store.New()
cl.SetStore(store.New())
cl.SetTransport(&nopTransporter{})
s := &EtcdServer{
r: raftNode{
Node: n,
raftStorage: raft.NewMemoryStorage(),
storage: &storageRecorder{},
transport: &nopTransporter{},
},
store: st,
Cluster: cl,
reqIDGen: idutil.NewGenerator(0, time.Time{}),
}
s.start()
s.AddMember(context.TODO(), Member{ID: 1234})
err := s.RemoveMember(context.TODO(), 1234)
gaction := n.Action()
s.Stop()
if err != nil {
t.Fatalf("RemoveMember error: %v", err)
}
wactions := []testutil.Action{
{Name: "ProposeConfChange:ConfChangeAddNode"},
{Name: "ApplyConfChange:ConfChangeAddNode"},
{Name: "ProposeConfChange:ConfChangeRemoveNode"},
{Name: "ApplyConfChange:ConfChangeRemoveNode"},
}
if !reflect.DeepEqual(gaction, wactions) {
t.Errorf("action = %v, want %v", gaction, wactions)
}
if cl.Member(1234) != nil {
t.Errorf("member with id 1234 is not removed")
}
}
// TestUpdateMember tests RemoveMember can propose and perform node update.
func TestUpdateMember(t *testing.T) {
n := newNodeConfChangeCommitterRecorder()
n.readyc <- raft.Ready{
2014-12-09 08:17:04 +03:00
SoftState: &raft.SoftState{RaftState: raft.StateLeader},
}
cl := newTestCluster(nil)
st := store.New()
cl.SetStore(st)
cl.SetTransport(&nopTransporter{})
s := &EtcdServer{
r: raftNode{
Node: n,
raftStorage: raft.NewMemoryStorage(),
storage: &storageRecorder{},
transport: &nopTransporter{},
},
store: st,
Cluster: cl,
reqIDGen: idutil.NewGenerator(0, time.Time{}),
}
s.start()
s.AddMember(context.TODO(), Member{ID: 1234})
wm := Member{ID: 1234, RaftAttributes: RaftAttributes{PeerURLs: []string{"http://127.0.0.1:1"}}}
err := s.UpdateMember(context.TODO(), wm)
gaction := n.Action()
s.Stop()
if err != nil {
t.Fatalf("UpdateMember error: %v", err)
}
wactions := []testutil.Action{
{Name: "ProposeConfChange:ConfChangeAddNode"},
{Name: "ApplyConfChange:ConfChangeAddNode"},
{Name: "ProposeConfChange:ConfChangeUpdateNode"},
{Name: "ApplyConfChange:ConfChangeUpdateNode"},
}
if !reflect.DeepEqual(gaction, wactions) {
t.Errorf("action = %v, want %v", gaction, wactions)
}
if !reflect.DeepEqual(cl.Member(1234), &wm) {
t.Errorf("member = %v, want %v", cl.Member(1234), &wm)
}
}
// TODO: test server could stop itself when being removed
2014-09-27 01:18:59 +04:00
func TestPublish(t *testing.T) {
n := &nodeRecorder{}
2014-09-27 01:18:59 +04:00
ch := make(chan interface{}, 1)
// simulate that request has gone through consensus
ch <- Response{}
w := &waitWithResponse{ch: ch}
srv := &EtcdServer{
id: 1,
r: raftNode{Node: n},
attributes: Attributes{Name: "node1", ClientURLs: []string{"http://a", "http://b"}},
2014-10-26 07:37:13 +03:00
Cluster: &Cluster{},
w: w,
2014-12-16 03:07:47 +03:00
reqIDGen: idutil.NewGenerator(0, time.Time{}),
2014-09-27 01:18:59 +04:00
}
srv.publish(time.Hour)
2014-09-27 01:18:59 +04:00
action := n.Action()
if len(action) != 1 {
t.Fatalf("len(action) = %d, want 1", len(action))
2014-09-27 01:18:59 +04:00
}
2015-01-02 22:21:23 +03:00
if action[0].Name != "Propose" {
t.Fatalf("action = %s, want Propose", action[0].Name)
}
2015-01-02 22:21:23 +03:00
data := action[0].Params[0].([]byte)
2014-09-27 01:18:59 +04:00
var r pb.Request
if err := r.Unmarshal(data); err != nil {
2014-09-27 01:18:59 +04:00
t.Fatalf("unmarshal request error: %v", err)
}
if r.Method != "PUT" {
t.Errorf("method = %s, want PUT", r.Method)
}
wm := Member{ID: 1, Attributes: Attributes{Name: "node1", ClientURLs: []string{"http://a", "http://b"}}}
if wpath := path.Join(memberStoreKey(wm.ID), attributesSuffix); r.Path != wpath {
t.Errorf("path = %s, want %s", r.Path, wpath)
2014-09-27 01:18:59 +04:00
}
var gattr Attributes
if err := json.Unmarshal([]byte(r.Val), &gattr); err != nil {
2014-09-27 01:18:59 +04:00
t.Fatalf("unmarshal val error: %v", err)
}
if !reflect.DeepEqual(gattr, wm.Attributes) {
t.Errorf("member = %v, want %v", gattr, wm.Attributes)
2014-09-27 01:18:59 +04:00
}
}
// TestPublishStopped tests that publish will be stopped if server is stopped.
func TestPublishStopped(t *testing.T) {
srv := &EtcdServer{
r: raftNode{
Node: &nodeRecorder{},
transport: &nopTransporter{},
},
Cluster: &Cluster{},
w: &waitRecorder{},
done: make(chan struct{}),
stop: make(chan struct{}),
reqIDGen: idutil.NewGenerator(0, time.Time{}),
2014-09-27 01:18:59 +04:00
}
close(srv.done)
srv.publish(time.Hour)
2014-09-27 01:18:59 +04:00
}
// TestPublishRetry tests that publish will keep retry until success.
func TestPublishRetry(t *testing.T) {
log.SetOutput(ioutil.Discard)
defer log.SetOutput(os.Stderr)
2014-09-27 01:18:59 +04:00
n := &nodeRecorder{}
srv := &EtcdServer{
r: raftNode{Node: n},
2014-12-16 03:07:47 +03:00
w: &waitRecorder{},
done: make(chan struct{}),
reqIDGen: idutil.NewGenerator(0, time.Time{}),
2014-09-27 01:18:59 +04:00
}
time.AfterFunc(500*time.Microsecond, func() { close(srv.done) })
srv.publish(10 * time.Nanosecond)
2014-09-27 01:18:59 +04:00
action := n.Action()
// multiple Proposes
if cnt := len(action); cnt < 2 {
t.Errorf("len(action) = %d, want >= 2", cnt)
2014-09-27 01:18:59 +04:00
}
}
2014-11-14 01:16:48 +03:00
func TestStopNotify(t *testing.T) {
s := &EtcdServer{
stop: make(chan struct{}),
done: make(chan struct{}),
}
go func() {
<-s.stop
close(s.done)
}()
notifier := s.StopNotify()
select {
case <-notifier:
t.Fatalf("received unexpected stop notification")
default:
}
s.Stop()
select {
case <-notifier:
default:
t.Fatalf("cannot receive stop notification")
}
}
func TestGetOtherPeerURLs(t *testing.T) {
tests := []struct {
membs []*Member
self string
wurls []string
}{
{
[]*Member{
newTestMember(1, []string{"http://10.0.0.1"}, "a", nil),
},
"a",
[]string{},
},
{
[]*Member{
newTestMember(1, []string{"http://10.0.0.1"}, "a", nil),
newTestMember(2, []string{"http://10.0.0.2"}, "b", nil),
newTestMember(3, []string{"http://10.0.0.3"}, "c", nil),
},
"a",
[]string{"http://10.0.0.2", "http://10.0.0.3"},
},
{
[]*Member{
newTestMember(1, []string{"http://10.0.0.1"}, "a", nil),
newTestMember(3, []string{"http://10.0.0.3"}, "c", nil),
newTestMember(2, []string{"http://10.0.0.2"}, "b", nil),
},
"a",
[]string{"http://10.0.0.2", "http://10.0.0.3"},
},
}
for i, tt := range tests {
cl := NewClusterFromMembers("", types.ID(0), tt.membs)
urls := getRemotePeerURLs(cl, tt.self)
if !reflect.DeepEqual(urls, tt.wurls) {
t.Errorf("#%d: urls = %+v, want %+v", i, urls, tt.wurls)
}
}
}
2015-01-02 21:56:09 +03:00
// storeRecorder records all the methods it receives.
// storeRecorder DOES NOT work as a actual store.
// It always returns invaild empty response and no error.
2015-01-02 22:21:23 +03:00
type storeRecorder struct{ testutil.Recorder }
2014-09-17 05:18:45 +04:00
func (s *storeRecorder) Version() int { return 0 }
func (s *storeRecorder) Index() uint64 { return 0 }
func (s *storeRecorder) Get(path string, recursive, sorted bool) (*store.Event, error) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{
Name: "Get",
Params: []interface{}{path, recursive, sorted},
})
return &store.Event{}, nil
}
func (s *storeRecorder) Set(path string, dir bool, val string, expr time.Time) (*store.Event, error) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{
Name: "Set",
Params: []interface{}{path, dir, val, expr},
})
return &store.Event{}, nil
}
func (s *storeRecorder) Update(path, val string, expr time.Time) (*store.Event, error) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{
Name: "Update",
Params: []interface{}{path, val, expr},
})
return &store.Event{}, nil
}
func (s *storeRecorder) Create(path string, dir bool, val string, uniq bool, exp time.Time) (*store.Event, error) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{
Name: "Create",
Params: []interface{}{path, dir, val, uniq, exp},
})
return &store.Event{}, nil
}
func (s *storeRecorder) CompareAndSwap(path, prevVal string, prevIdx uint64, val string, expr time.Time) (*store.Event, error) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{
Name: "CompareAndSwap",
Params: []interface{}{path, prevVal, prevIdx, val, expr},
})
return &store.Event{}, nil
}
func (s *storeRecorder) Delete(path string, dir, recursive bool) (*store.Event, error) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{
Name: "Delete",
Params: []interface{}{path, dir, recursive},
})
return &store.Event{}, nil
}
func (s *storeRecorder) CompareAndDelete(path, prevVal string, prevIdx uint64) (*store.Event, error) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{
Name: "CompareAndDelete",
Params: []interface{}{path, prevVal, prevIdx},
})
return &store.Event{}, nil
}
func (s *storeRecorder) Watch(_ string, _, _ bool, _ uint64) (store.Watcher, error) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{Name: "Watch"})
2015-01-02 21:56:09 +03:00
return &nopWatcher{}, nil
}
2014-09-17 05:18:45 +04:00
func (s *storeRecorder) Save() ([]byte, error) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{Name: "Save"})
2014-09-17 05:18:45 +04:00
return nil, nil
}
2014-09-19 10:13:08 +04:00
func (s *storeRecorder) Recovery(b []byte) error {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{Name: "Recovery"})
2014-09-19 10:13:08 +04:00
return nil
}
func (s *storeRecorder) SaveNoCopy() ([]byte, error) {
s.Record(testutil.Action{Name: "SaveNoCopy"})
return nil, nil
}
func (s *storeRecorder) Clone() store.Store {
s.Record(testutil.Action{Name: "Clone"})
return s
}
2014-10-17 11:04:22 +04:00
func (s *storeRecorder) JsonStats() []byte { return nil }
2014-09-16 04:35:02 +04:00
func (s *storeRecorder) DeleteExpiredKeys(cutoff time.Time) {
2015-01-02 22:21:23 +03:00
s.Record(testutil.Action{
Name: "DeleteExpiredKeys",
Params: []interface{}{cutoff},
})
2014-09-16 04:35:02 +04:00
}
2015-01-02 21:56:09 +03:00
type nopWatcher struct{}
2015-01-02 21:56:09 +03:00
func (w *nopWatcher) EventChan() chan *store.Event { return nil }
func (w *nopWatcher) StartIndex() uint64 { return 0 }
func (w *nopWatcher) Remove() {}
2015-01-02 21:56:09 +03:00
// errStoreRecorder is a storeRecorder, but returns the given error on
// Get, Watch methods.
2014-09-18 13:04:16 +04:00
type errStoreRecorder struct {
storeRecorder
err error
}
2015-01-02 21:56:09 +03:00
func (s *errStoreRecorder) Get(path string, recursive, sorted bool) (*store.Event, error) {
s.storeRecorder.Get(path, recursive, sorted)
2014-09-18 13:04:16 +04:00
return nil, s.err
}
2015-01-02 21:56:09 +03:00
func (s *errStoreRecorder) Watch(path string, recursive, sorted bool, index uint64) (store.Watcher, error) {
s.storeRecorder.Watch(path, recursive, sorted, index)
2014-09-18 13:04:16 +04:00
return nil, s.err
}
type waitRecorder struct {
2015-01-02 22:21:23 +03:00
action []testutil.Action
}
2014-10-12 04:35:58 +04:00
func (w *waitRecorder) Register(id uint64) <-chan interface{} {
2015-01-02 22:21:23 +03:00
w.action = append(w.action, testutil.Action{Name: "Register"})
return nil
}
2014-10-12 04:35:58 +04:00
func (w *waitRecorder) Trigger(id uint64, x interface{}) {
2015-01-02 22:21:23 +03:00
w.action = append(w.action, testutil.Action{Name: "Trigger"})
}
type waitWithResponse struct {
ch <-chan interface{}
}
func (w *waitWithResponse) Register(id uint64) <-chan interface{} {
return w.ch
}
func (w *waitWithResponse) Trigger(id uint64, x interface{}) {}
2014-09-17 05:18:45 +04:00
2015-01-02 22:21:23 +03:00
type storageRecorder struct{ testutil.Recorder }
2014-09-17 05:18:45 +04:00
func (p *storageRecorder) Save(st raftpb.HardState, ents []raftpb.Entry) error {
2015-01-02 22:21:23 +03:00
p.Record(testutil.Action{Name: "Save"})
return nil
2014-09-17 05:18:45 +04:00
}
func (p *storageRecorder) SaveSnap(st raftpb.Snapshot) error {
if !raft.IsEmptySnap(st) {
2015-01-02 22:21:23 +03:00
p.Record(testutil.Action{Name: "SaveSnap"})
2014-09-17 05:18:45 +04:00
}
return nil
2014-09-17 05:18:45 +04:00
}
2014-12-03 10:20:21 +03:00
func (p *storageRecorder) Close() error { return nil }
2014-09-18 01:38:10 +04:00
2015-01-02 22:21:23 +03:00
type nodeRecorder struct{ testutil.Recorder }
2015-01-02 22:21:23 +03:00
func (n *nodeRecorder) Tick() { n.Record(testutil.Action{Name: "Tick"}) }
func (n *nodeRecorder) Campaign(ctx context.Context) error {
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "Campaign"})
return nil
}
func (n *nodeRecorder) Propose(ctx context.Context, data []byte) error {
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "Propose", Params: []interface{}{data}})
return nil
}
2014-09-23 23:02:44 +04:00
func (n *nodeRecorder) ProposeConfChange(ctx context.Context, conf raftpb.ConfChange) error {
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "ProposeConfChange"})
return nil
}
func (n *nodeRecorder) Step(ctx context.Context, msg raftpb.Message) error {
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "Step"})
return nil
}
func (n *nodeRecorder) Status() raft.Status { return raft.Status{} }
func (n *nodeRecorder) Ready() <-chan raft.Ready { return nil }
func (n *nodeRecorder) Advance() {}
func (n *nodeRecorder) ApplyConfChange(conf raftpb.ConfChange) *raftpb.ConfState {
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "ApplyConfChange", Params: []interface{}{conf}})
2014-12-09 08:17:04 +03:00
return &raftpb.ConfState{}
}
func (n *nodeRecorder) Stop() {
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "Stop"})
}
2015-02-05 01:41:14 +03:00
func (n *nodeRecorder) ReportUnreachable(id uint64) {}
func (n *nodeRecorder) ReportSnapshot(id uint64, status raft.SnapshotStatus) {}
2014-10-08 15:58:53 +04:00
func (n *nodeRecorder) Compact(index uint64, nodes []uint64, d []byte) {
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "Compact"})
}
type nodeProposalBlockerRecorder struct {
nodeRecorder
}
func (n *nodeProposalBlockerRecorder) Propose(ctx context.Context, data []byte) error {
<-ctx.Done()
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "Propose blocked"})
return nil
2014-09-18 01:38:10 +04:00
}
2014-09-23 23:02:44 +04:00
type nodeConfChangeCommitterRecorder struct {
nodeRecorder
readyc chan raft.Ready
index uint64
}
2014-09-23 23:02:44 +04:00
func newNodeConfChangeCommitterRecorder() *nodeConfChangeCommitterRecorder {
readyc := make(chan raft.Ready, 1)
2014-09-23 23:02:44 +04:00
return &nodeConfChangeCommitterRecorder{readyc: readyc}
}
2014-09-23 23:02:44 +04:00
func (n *nodeConfChangeCommitterRecorder) ProposeConfChange(ctx context.Context, conf raftpb.ConfChange) error {
data, err := conf.Marshal()
if err != nil {
return err
}
n.index++
n.readyc <- raft.Ready{CommittedEntries: []raftpb.Entry{{Index: n.index, Type: raftpb.EntryConfChange, Data: data}}}
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "ProposeConfChange:" + conf.Type.String()})
return nil
}
2014-09-23 23:02:44 +04:00
func (n *nodeConfChangeCommitterRecorder) Ready() <-chan raft.Ready {
return n.readyc
}
func (n *nodeConfChangeCommitterRecorder) ApplyConfChange(conf raftpb.ConfChange) *raftpb.ConfState {
2015-01-02 22:21:23 +03:00
n.Record(testutil.Action{Name: "ApplyConfChange:" + conf.Type.String()})
2014-12-09 08:17:04 +03:00
return &raftpb.ConfState{}
}
2014-09-27 01:18:59 +04:00
// nodeCommitter commits proposed data immediately.
type nodeCommitter struct {
nodeRecorder
readyc chan raft.Ready
index uint64
2014-09-27 01:18:59 +04:00
}
func newNodeCommitter() *nodeCommitter {
readyc := make(chan raft.Ready, 1)
return &nodeCommitter{readyc: readyc}
2014-09-27 01:18:59 +04:00
}
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 (n *nodeCommitter) Ready() <-chan raft.Ready {
return n.readyc
}
type readyNode struct {
nodeRecorder
readyc chan raft.Ready
}
func newReadyNode() *readyNode {
readyc := make(chan raft.Ready, 1)
return &readyNode{readyc: readyc}
}
func (n *readyNode) Ready() <-chan raft.Ready { return n.readyc }
type nopTransporter struct{}
func (s *nopTransporter) Handler() http.Handler { return nil }
func (s *nopTransporter) Send(m []raftpb.Message) {}
func (s *nopTransporter) AddPeer(id types.ID, us []string) {}
func (s *nopTransporter) RemovePeer(id types.ID) {}
func (s *nopTransporter) RemoveAllPeers() {}
func (s *nopTransporter) UpdatePeer(id types.ID, us []string) {}
func (s *nopTransporter) Stop() {}
func (s *nopTransporter) Pause() {}
func (s *nopTransporter) Resume() {}