2016-05-13 06:49:28 +03:00
|
|
|
// Copyright 2015 The etcd Authors
|
2015-01-25 06:19:16 +03:00
|
|
|
//
|
|
|
|
// 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-11-16 21:21:05 +03:00
|
|
|
|
|
|
|
package rafthttp
|
|
|
|
|
|
|
|
import (
|
|
|
|
"errors"
|
2015-10-22 22:06:47 +03:00
|
|
|
"fmt"
|
2015-06-02 21:46:36 +03:00
|
|
|
"io"
|
2014-11-16 21:21:05 +03:00
|
|
|
"io/ioutil"
|
|
|
|
"net/http"
|
|
|
|
"sync"
|
|
|
|
"testing"
|
2015-06-02 10:38:26 +03:00
|
|
|
"time"
|
2014-11-16 21:21:05 +03:00
|
|
|
|
|
|
|
"github.com/coreos/etcd/etcdserver/stats"
|
|
|
|
"github.com/coreos/etcd/pkg/testutil"
|
|
|
|
"github.com/coreos/etcd/pkg/types"
|
2014-11-18 03:40:52 +03:00
|
|
|
"github.com/coreos/etcd/raft/raftpb"
|
2015-06-02 21:46:36 +03:00
|
|
|
"github.com/coreos/etcd/version"
|
2014-11-16 21:21:05 +03:00
|
|
|
)
|
|
|
|
|
2015-02-05 20:40:46 +03:00
|
|
|
// TestPipelineSend tests that pipeline could send data using roundtripper
|
2014-11-16 21:21:05 +03:00
|
|
|
// and increase success count in stats.
|
2015-02-05 20:40:46 +03:00
|
|
|
func TestPipelineSend(t *testing.T) {
|
2016-08-18 23:59:38 +03:00
|
|
|
tr := &roundTripperRecorder{rec: testutil.NewRecorderStream()}
|
2015-04-06 18:38:13 +03:00
|
|
|
picker := mustNewURLPicker(t, []string{"http://localhost:2380"})
|
2016-02-05 09:05:59 +03:00
|
|
|
tp := &Transport{pipelineRt: tr}
|
2016-05-31 02:17:13 +03:00
|
|
|
p := startTestPipeline(tp, picker)
|
2014-11-16 21:21:05 +03:00
|
|
|
|
2015-02-15 19:44:34 +03:00
|
|
|
p.msgc <- raftpb.Message{Type: raftpb.MsgApp}
|
2016-08-18 23:59:38 +03:00
|
|
|
tr.rec.Wait(1)
|
2015-02-05 20:40:46 +03:00
|
|
|
p.stop()
|
2016-05-31 02:17:13 +03:00
|
|
|
if p.followerStats.Counts.Success != 1 {
|
|
|
|
t.Errorf("success = %d, want 1", p.followerStats.Counts.Success)
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-10-22 22:06:47 +03:00
|
|
|
// TestPipelineKeepSendingWhenPostError tests that pipeline can keep
|
|
|
|
// sending messages if previous messages meet post error.
|
|
|
|
func TestPipelineKeepSendingWhenPostError(t *testing.T) {
|
2016-04-20 00:11:38 +03:00
|
|
|
tr := &respRoundTripper{rec: testutil.NewRecorderStream(), err: fmt.Errorf("roundtrip error")}
|
2015-10-22 22:06:47 +03:00
|
|
|
picker := mustNewURLPicker(t, []string{"http://localhost:2380"})
|
2016-02-05 09:05:59 +03:00
|
|
|
tp := &Transport{pipelineRt: tr}
|
2016-05-31 02:17:13 +03:00
|
|
|
p := startTestPipeline(tp, picker)
|
2016-04-20 00:11:38 +03:00
|
|
|
defer p.stop()
|
2015-10-22 22:06:47 +03:00
|
|
|
|
|
|
|
for i := 0; i < 50; i++ {
|
|
|
|
p.msgc <- raftpb.Message{Type: raftpb.MsgApp}
|
|
|
|
}
|
|
|
|
|
2016-04-20 00:11:38 +03:00
|
|
|
_, err := tr.rec.Wait(50)
|
|
|
|
if err != nil {
|
|
|
|
t.Errorf("unexpected wait error %v", err)
|
2015-10-22 22:06:47 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-06-23 02:06:57 +03:00
|
|
|
func TestPipelineExceedMaximumServing(t *testing.T) {
|
2016-09-13 13:45:48 +03:00
|
|
|
rt := newRoundTripperBlocker()
|
2015-04-06 18:38:13 +03:00
|
|
|
picker := mustNewURLPicker(t, []string{"http://localhost:2380"})
|
2016-09-13 13:45:48 +03:00
|
|
|
tp := &Transport{pipelineRt: rt}
|
2016-05-31 02:17:13 +03:00
|
|
|
p := startTestPipeline(tp, picker)
|
|
|
|
defer p.stop()
|
2014-11-16 21:21:05 +03:00
|
|
|
|
|
|
|
// keep the sender busy and make the buffer full
|
|
|
|
// nothing can go out as we block the sender
|
2015-02-05 20:40:46 +03:00
|
|
|
for i := 0; i < connPerPipeline+pipelineBufSize; i++ {
|
2015-02-15 19:44:34 +03:00
|
|
|
select {
|
|
|
|
case p.msgc <- raftpb.Message{}:
|
2016-12-30 20:46:16 +03:00
|
|
|
case <-time.After(time.Second):
|
2015-02-15 19:44:34 +03:00
|
|
|
t.Errorf("failed to send out message")
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// try to send a data when we are sure the buffer is full
|
2015-02-15 19:44:34 +03:00
|
|
|
select {
|
|
|
|
case p.msgc <- raftpb.Message{}:
|
|
|
|
t.Errorf("unexpected message sendout")
|
|
|
|
default:
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
// unblock the senders and force them to send out the data
|
2016-09-13 13:45:48 +03:00
|
|
|
rt.unblock()
|
2014-11-16 21:21:05 +03:00
|
|
|
|
|
|
|
// It could send new data after previous ones succeed
|
2015-02-15 19:44:34 +03:00
|
|
|
select {
|
|
|
|
case p.msgc <- raftpb.Message{}:
|
2016-09-13 13:45:48 +03:00
|
|
|
case <-time.After(time.Second):
|
2015-02-15 19:44:34 +03:00
|
|
|
t.Errorf("failed to send out message")
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-02-05 20:40:46 +03:00
|
|
|
// TestPipelineSendFailed tests that when send func meets the post error,
|
2014-11-16 21:21:05 +03:00
|
|
|
// it increases fail count in stats.
|
2015-02-05 20:40:46 +03:00
|
|
|
func TestPipelineSendFailed(t *testing.T) {
|
2015-04-06 18:38:13 +03:00
|
|
|
picker := mustNewURLPicker(t, []string{"http://localhost:2380"})
|
2016-08-18 23:59:38 +03:00
|
|
|
rt := newRespRoundTripper(0, errors.New("blah"))
|
|
|
|
rt.rec = testutil.NewRecorderStream()
|
|
|
|
tp := &Transport{pipelineRt: rt}
|
2016-05-31 02:17:13 +03:00
|
|
|
p := startTestPipeline(tp, picker)
|
2014-11-16 21:21:05 +03:00
|
|
|
|
2015-02-15 19:44:34 +03:00
|
|
|
p.msgc <- raftpb.Message{Type: raftpb.MsgApp}
|
2016-08-18 23:59:38 +03:00
|
|
|
if _, err := rt.rec.Wait(1); err != nil {
|
|
|
|
t.Fatal(err)
|
|
|
|
}
|
|
|
|
|
2015-02-05 20:40:46 +03:00
|
|
|
p.stop()
|
2014-11-16 21:21:05 +03:00
|
|
|
|
2016-05-31 02:17:13 +03:00
|
|
|
if p.followerStats.Counts.Fail != 1 {
|
|
|
|
t.Errorf("fail = %d, want 1", p.followerStats.Counts.Fail)
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-02-05 20:40:46 +03:00
|
|
|
func TestPipelinePost(t *testing.T) {
|
2016-08-18 23:59:38 +03:00
|
|
|
tr := &roundTripperRecorder{rec: &testutil.RecorderBuffered{}}
|
2015-04-06 18:38:13 +03:00
|
|
|
picker := mustNewURLPicker(t, []string{"http://localhost:2380"})
|
2016-05-31 02:17:13 +03:00
|
|
|
tp := &Transport{ClusterID: types.ID(1), pipelineRt: tr}
|
|
|
|
p := startTestPipeline(tp, picker)
|
2015-02-15 19:44:34 +03:00
|
|
|
if err := p.post([]byte("some data")); err != nil {
|
2016-01-14 12:28:29 +03:00
|
|
|
t.Fatalf("unexpected post error: %v", err)
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
2016-08-18 23:59:38 +03:00
|
|
|
act, err := tr.rec.Wait(1)
|
|
|
|
if err != nil {
|
|
|
|
t.Fatal(err)
|
|
|
|
}
|
2015-02-05 20:40:46 +03:00
|
|
|
p.stop()
|
2014-11-16 21:21:05 +03:00
|
|
|
|
2016-08-18 23:59:38 +03:00
|
|
|
req := act[0].Params[0].(*http.Request)
|
|
|
|
|
|
|
|
if g := req.Method; g != "POST" {
|
2014-11-16 21:21:05 +03:00
|
|
|
t.Errorf("method = %s, want %s", g, "POST")
|
|
|
|
}
|
2016-08-18 23:59:38 +03:00
|
|
|
if g := req.URL.String(); g != "http://localhost:2380/raft" {
|
2015-04-06 18:38:13 +03:00
|
|
|
t.Errorf("url = %s, want %s", g, "http://localhost:2380/raft")
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
2016-08-18 23:59:38 +03:00
|
|
|
if g := req.Header.Get("Content-Type"); g != "application/protobuf" {
|
2014-11-16 21:21:05 +03:00
|
|
|
t.Errorf("content type = %s, want %s", g, "application/protobuf")
|
|
|
|
}
|
2016-08-18 23:59:38 +03:00
|
|
|
if g := req.Header.Get("X-Server-Version"); g != version.Version {
|
2015-06-02 21:46:36 +03:00
|
|
|
t.Errorf("version = %s, want %s", g, version.Version)
|
|
|
|
}
|
2016-08-18 23:59:38 +03:00
|
|
|
if g := req.Header.Get("X-Min-Cluster-Version"); g != version.MinClusterVersion {
|
2015-06-02 21:46:36 +03:00
|
|
|
t.Errorf("min version = %s, want %s", g, version.MinClusterVersion)
|
|
|
|
}
|
2016-08-18 23:59:38 +03:00
|
|
|
if g := req.Header.Get("X-Etcd-Cluster-ID"); g != "1" {
|
2014-11-16 21:21:05 +03:00
|
|
|
t.Errorf("cluster id = %s, want %s", g, "1")
|
|
|
|
}
|
2016-08-18 23:59:38 +03:00
|
|
|
b, err := ioutil.ReadAll(req.Body)
|
2014-11-16 21:21:05 +03:00
|
|
|
if err != nil {
|
|
|
|
t.Fatalf("unexpected ReadAll error: %v", err)
|
|
|
|
}
|
|
|
|
if string(b) != "some data" {
|
|
|
|
t.Errorf("body = %s, want %s", b, "some data")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-02-05 20:40:46 +03:00
|
|
|
func TestPipelinePostBad(t *testing.T) {
|
2014-11-16 21:21:05 +03:00
|
|
|
tests := []struct {
|
|
|
|
u string
|
|
|
|
code int
|
|
|
|
err error
|
|
|
|
}{
|
|
|
|
// RoundTrip returns error
|
2015-04-06 18:38:13 +03:00
|
|
|
{"http://localhost:2380", 0, errors.New("blah")},
|
2014-11-16 21:21:05 +03:00
|
|
|
// unexpected response status code
|
2015-04-06 18:38:13 +03:00
|
|
|
{"http://localhost:2380", http.StatusOK, nil},
|
|
|
|
{"http://localhost:2380", http.StatusCreated, nil},
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
for i, tt := range tests {
|
2015-02-27 18:54:06 +03:00
|
|
|
picker := mustNewURLPicker(t, []string{tt.u})
|
2016-02-05 09:05:59 +03:00
|
|
|
tp := &Transport{pipelineRt: newRespRoundTripper(tt.code, tt.err)}
|
2016-05-31 02:17:13 +03:00
|
|
|
p := startTestPipeline(tp, picker)
|
2015-02-15 19:44:34 +03:00
|
|
|
err := p.post([]byte("some data"))
|
2015-02-05 20:40:46 +03:00
|
|
|
p.stop()
|
2014-11-16 21:21:05 +03:00
|
|
|
|
|
|
|
if err == nil {
|
|
|
|
t.Errorf("#%d: err = nil, want not nil", i)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-02-05 20:40:46 +03:00
|
|
|
func TestPipelinePostErrorc(t *testing.T) {
|
2014-11-16 21:21:05 +03:00
|
|
|
tests := []struct {
|
|
|
|
u string
|
|
|
|
code int
|
|
|
|
err error
|
|
|
|
}{
|
2015-04-06 18:38:13 +03:00
|
|
|
{"http://localhost:2380", http.StatusForbidden, nil},
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
for i, tt := range tests {
|
2015-02-27 18:54:06 +03:00
|
|
|
picker := mustNewURLPicker(t, []string{tt.u})
|
2016-02-05 09:05:59 +03:00
|
|
|
tp := &Transport{pipelineRt: newRespRoundTripper(tt.code, tt.err)}
|
2016-05-31 02:17:13 +03:00
|
|
|
p := startTestPipeline(tp, picker)
|
2015-02-15 19:44:34 +03:00
|
|
|
p.post([]byte("some data"))
|
2015-02-05 20:40:46 +03:00
|
|
|
p.stop()
|
2014-11-16 21:21:05 +03:00
|
|
|
select {
|
2016-05-31 02:17:13 +03:00
|
|
|
case <-p.errorc:
|
2014-11-16 21:21:05 +03:00
|
|
|
default:
|
2015-01-03 07:00:29 +03:00
|
|
|
t.Fatalf("#%d: cannot receive from errorc", i)
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-06-02 10:38:26 +03:00
|
|
|
func TestStopBlockedPipeline(t *testing.T) {
|
|
|
|
picker := mustNewURLPicker(t, []string{"http://localhost:2380"})
|
2016-02-05 09:05:59 +03:00
|
|
|
tp := &Transport{pipelineRt: newRoundTripperBlocker()}
|
2016-05-31 02:17:13 +03:00
|
|
|
p := startTestPipeline(tp, picker)
|
2015-06-02 10:38:26 +03:00
|
|
|
// send many messages that most of them will be blocked in buffer
|
|
|
|
for i := 0; i < connPerPipeline*10; i++ {
|
|
|
|
p.msgc <- raftpb.Message{}
|
|
|
|
}
|
|
|
|
|
|
|
|
done := make(chan struct{})
|
|
|
|
go func() {
|
|
|
|
p.stop()
|
|
|
|
done <- struct{}{}
|
|
|
|
}()
|
|
|
|
select {
|
|
|
|
case <-done:
|
|
|
|
case <-time.After(time.Second):
|
|
|
|
t.Fatalf("failed to stop pipeline in 1s")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-11-16 21:21:05 +03:00
|
|
|
type roundTripperBlocker struct {
|
2015-06-11 19:16:23 +03:00
|
|
|
unblockc chan struct{}
|
|
|
|
mu sync.Mutex
|
|
|
|
cancel map[*http.Request]chan struct{}
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func newRoundTripperBlocker() *roundTripperBlocker {
|
2015-06-11 19:16:23 +03:00
|
|
|
return &roundTripperBlocker{
|
|
|
|
unblockc: make(chan struct{}),
|
|
|
|
cancel: make(map[*http.Request]chan struct{}),
|
|
|
|
}
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
2015-10-27 01:02:50 +03:00
|
|
|
|
2014-11-16 21:21:05 +03:00
|
|
|
func (t *roundTripperBlocker) unblock() {
|
2015-06-11 19:16:23 +03:00
|
|
|
close(t.unblockc)
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
2015-10-27 01:02:50 +03:00
|
|
|
|
2015-06-02 10:38:26 +03:00
|
|
|
func (t *roundTripperBlocker) CancelRequest(req *http.Request) {
|
|
|
|
t.mu.Lock()
|
|
|
|
defer t.mu.Unlock()
|
2015-06-11 19:16:23 +03:00
|
|
|
if c, ok := t.cancel[req]; ok {
|
|
|
|
c <- struct{}{}
|
|
|
|
delete(t.cancel, req)
|
2015-06-02 10:38:26 +03:00
|
|
|
}
|
|
|
|
}
|
2014-11-16 21:21:05 +03:00
|
|
|
|
|
|
|
type respRoundTripper struct {
|
2016-04-20 00:11:38 +03:00
|
|
|
mu sync.Mutex
|
|
|
|
rec testutil.Recorder
|
2015-10-22 22:06:47 +03:00
|
|
|
|
2015-05-07 22:40:22 +03:00
|
|
|
code int
|
|
|
|
header http.Header
|
|
|
|
err error
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func newRespRoundTripper(code int, err error) *respRoundTripper {
|
|
|
|
return &respRoundTripper{code: code, err: err}
|
|
|
|
}
|
|
|
|
func (t *respRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
|
2015-10-22 22:06:47 +03:00
|
|
|
t.mu.Lock()
|
|
|
|
defer t.mu.Unlock()
|
2016-04-20 00:11:38 +03:00
|
|
|
if t.rec != nil {
|
|
|
|
t.rec.Record(testutil.Action{Name: "req", Params: []interface{}{req}})
|
|
|
|
}
|
2015-05-07 22:40:22 +03:00
|
|
|
return &http.Response{StatusCode: t.code, Header: t.header, Body: &nopReadCloser{}}, t.err
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
type roundTripperRecorder struct {
|
2016-08-18 23:59:38 +03:00
|
|
|
rec testutil.Recorder
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (t *roundTripperRecorder) RoundTrip(req *http.Request) (*http.Response, error) {
|
2016-08-18 23:59:38 +03:00
|
|
|
if t.rec != nil {
|
|
|
|
t.rec.Record(testutil.Action{Name: "req", Params: []interface{}{req}})
|
|
|
|
}
|
2014-11-16 21:21:05 +03:00
|
|
|
return &http.Response{StatusCode: http.StatusNoContent, Body: &nopReadCloser{}}, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
type nopReadCloser struct{}
|
|
|
|
|
2015-06-02 21:46:36 +03:00
|
|
|
func (n *nopReadCloser) Read(p []byte) (int, error) { return 0, io.EOF }
|
2014-11-16 21:21:05 +03:00
|
|
|
func (n *nopReadCloser) Close() error { return nil }
|
2016-05-31 02:17:13 +03:00
|
|
|
|
|
|
|
func startTestPipeline(tr *Transport, picker *urlPicker) *pipeline {
|
|
|
|
p := &pipeline{
|
2016-06-02 08:12:47 +03:00
|
|
|
peerID: types.ID(1),
|
|
|
|
tr: tr,
|
|
|
|
picker: picker,
|
2016-05-31 02:17:13 +03:00
|
|
|
status: newPeerStatus(types.ID(1)),
|
|
|
|
raft: &fakeRaft{},
|
|
|
|
followerStats: &stats.FollowerStats{},
|
|
|
|
errorc: make(chan error, 1),
|
|
|
|
}
|
|
|
|
p.start()
|
|
|
|
return p
|
|
|
|
}
|