Add vendor folder to git

This commit is contained in:
Lucas Käldström 2017-06-26 19:23:05 +03:00
parent 66cf5eaafb
commit 183585f56f
No known key found for this signature in database
GPG key ID: 600FEFBBD0D40D21
6916 changed files with 2629581 additions and 1 deletions

27
vendor/github.com/coreos/etcd/rafthttp/coder.go generated vendored Normal file
View file

@ -0,0 +1,27 @@
// Copyright 2015 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 rafthttp
import "github.com/coreos/etcd/raft/raftpb"
type encoder interface {
// encode encodes the given message to an output stream.
encode(m *raftpb.Message) error
}
type decoder interface {
// decode decodes the message from an input stream.
decode() (raftpb.Message, error)
}

16
vendor/github.com/coreos/etcd/rafthttp/doc.go generated vendored Normal file
View file

@ -0,0 +1,16 @@
// Copyright 2015 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 rafthttp implements HTTP transportation layer for etcd/raft pkg.
package rafthttp

View file

@ -0,0 +1,35 @@
// Copyright 2015 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 rafthttp
import (
"errors"
"net/http"
)
func (t *roundTripperBlocker) RoundTrip(req *http.Request) (*http.Response, error) {
c := make(chan struct{}, 1)
t.mu.Lock()
t.cancel[req] = c
t.mu.Unlock()
select {
case <-t.unblockc:
return &http.Response{StatusCode: http.StatusNoContent, Body: &nopReadCloser{}}, nil
case <-req.Cancel:
return nil, errors.New("request canceled")
case <-c:
return nil, errors.New("request canceled")
}
}

View file

@ -0,0 +1,180 @@
// Copyright 2015 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 rafthttp
import (
"net/http/httptest"
"reflect"
"testing"
"time"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft"
"github.com/coreos/etcd/raft/raftpb"
"golang.org/x/net/context"
)
func TestSendMessage(t *testing.T) {
// member 1
tr := &Transport{
ID: types.ID(1),
ClusterID: types.ID(1),
Raft: &fakeRaft{},
ServerStats: newServerStats(),
LeaderStats: stats.NewLeaderStats("1"),
}
tr.Start()
srv := httptest.NewServer(tr.Handler())
defer srv.Close()
// member 2
recvc := make(chan raftpb.Message, 1)
p := &fakeRaft{recvc: recvc}
tr2 := &Transport{
ID: types.ID(2),
ClusterID: types.ID(1),
Raft: p,
ServerStats: newServerStats(),
LeaderStats: stats.NewLeaderStats("2"),
}
tr2.Start()
srv2 := httptest.NewServer(tr2.Handler())
defer srv2.Close()
tr.AddPeer(types.ID(2), []string{srv2.URL})
defer tr.Stop()
tr2.AddPeer(types.ID(1), []string{srv.URL})
defer tr2.Stop()
if !waitStreamWorking(tr.Get(types.ID(2)).(*peer)) {
t.Fatalf("stream from 1 to 2 is not in work as expected")
}
data := []byte("some data")
tests := []raftpb.Message{
// these messages are set to send to itself, which facilitates testing.
{Type: raftpb.MsgProp, From: 1, To: 2, Entries: []raftpb.Entry{{Data: data}}},
{Type: raftpb.MsgApp, From: 1, To: 2, Term: 1, Index: 3, LogTerm: 0, Entries: []raftpb.Entry{{Index: 4, Term: 1, Data: data}}, Commit: 3},
{Type: raftpb.MsgAppResp, From: 1, To: 2, Term: 1, Index: 3},
{Type: raftpb.MsgVote, From: 1, To: 2, Term: 1, Index: 3, LogTerm: 0},
{Type: raftpb.MsgVoteResp, From: 1, To: 2, Term: 1},
{Type: raftpb.MsgSnap, From: 1, To: 2, Term: 1, Snapshot: raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 1000, Term: 1}, Data: data}},
{Type: raftpb.MsgHeartbeat, From: 1, To: 2, Term: 1, Commit: 3},
{Type: raftpb.MsgHeartbeatResp, From: 1, To: 2, Term: 1},
}
for i, tt := range tests {
tr.Send([]raftpb.Message{tt})
msg := <-recvc
if !reflect.DeepEqual(msg, tt) {
t.Errorf("#%d: msg = %+v, want %+v", i, msg, tt)
}
}
}
// TestSendMessageWhenStreamIsBroken tests that message can be sent to the
// remote in a limited time when all underlying connections are broken.
func TestSendMessageWhenStreamIsBroken(t *testing.T) {
// member 1
tr := &Transport{
ID: types.ID(1),
ClusterID: types.ID(1),
Raft: &fakeRaft{},
ServerStats: newServerStats(),
LeaderStats: stats.NewLeaderStats("1"),
}
tr.Start()
srv := httptest.NewServer(tr.Handler())
defer srv.Close()
// member 2
recvc := make(chan raftpb.Message, 1)
p := &fakeRaft{recvc: recvc}
tr2 := &Transport{
ID: types.ID(2),
ClusterID: types.ID(1),
Raft: p,
ServerStats: newServerStats(),
LeaderStats: stats.NewLeaderStats("2"),
}
tr2.Start()
srv2 := httptest.NewServer(tr2.Handler())
defer srv2.Close()
tr.AddPeer(types.ID(2), []string{srv2.URL})
defer tr.Stop()
tr2.AddPeer(types.ID(1), []string{srv.URL})
defer tr2.Stop()
if !waitStreamWorking(tr.Get(types.ID(2)).(*peer)) {
t.Fatalf("stream from 1 to 2 is not in work as expected")
}
// break the stream
srv.CloseClientConnections()
srv2.CloseClientConnections()
var n int
for {
select {
// TODO: remove this resend logic when we add retry logic into the code
case <-time.After(time.Millisecond):
n++
tr.Send([]raftpb.Message{{Type: raftpb.MsgHeartbeat, From: 1, To: 2, Term: 1, Commit: 3}})
case <-recvc:
if n > 50 {
t.Errorf("disconnection time = %dms, want < 50ms", n)
}
return
}
}
}
func newServerStats() *stats.ServerStats {
ss := &stats.ServerStats{}
ss.Initialize()
return ss
}
func waitStreamWorking(p *peer) bool {
for i := 0; i < 1000; i++ {
time.Sleep(time.Millisecond)
if _, ok := p.msgAppV2Writer.writec(); !ok {
continue
}
if _, ok := p.writer.writec(); !ok {
continue
}
return true
}
return false
}
type fakeRaft struct {
recvc chan<- raftpb.Message
err error
removedID uint64
}
func (p *fakeRaft) Process(ctx context.Context, m raftpb.Message) error {
select {
case p.recvc <- m:
default:
}
return p.err
}
func (p *fakeRaft) IsIDRemoved(id uint64) bool { return id == p.removedID }
func (p *fakeRaft) ReportUnreachable(id uint64) {}
func (p *fakeRaft) ReportSnapshot(id uint64, status raft.SnapshotStatus) {}

358
vendor/github.com/coreos/etcd/rafthttp/http.go generated vendored Normal file
View file

@ -0,0 +1,358 @@
// Copyright 2015 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 rafthttp
import (
"errors"
"fmt"
"io/ioutil"
"net/http"
"path"
"strings"
pioutil "github.com/coreos/etcd/pkg/ioutil"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/snap"
"github.com/coreos/etcd/version"
"golang.org/x/net/context"
)
const (
// connReadLimitByte limits the number of bytes
// a single read can read out.
//
// 64KB should be large enough for not causing
// throughput bottleneck as well as small enough
// for not causing a read timeout.
connReadLimitByte = 64 * 1024
)
var (
RaftPrefix = "/raft"
ProbingPrefix = path.Join(RaftPrefix, "probing")
RaftStreamPrefix = path.Join(RaftPrefix, "stream")
RaftSnapshotPrefix = path.Join(RaftPrefix, "snapshot")
errIncompatibleVersion = errors.New("incompatible version")
errClusterIDMismatch = errors.New("cluster ID mismatch")
)
type peerGetter interface {
Get(id types.ID) Peer
}
type writerToResponse interface {
WriteTo(w http.ResponseWriter)
}
type pipelineHandler struct {
tr Transporter
r Raft
cid types.ID
}
// newPipelineHandler returns a handler for handling raft messages
// from pipeline for RaftPrefix.
//
// The handler reads out the raft message from request body,
// and forwards it to the given raft state machine for processing.
func newPipelineHandler(tr Transporter, r Raft, cid types.ID) http.Handler {
return &pipelineHandler{
tr: tr,
r: r,
cid: cid,
}
}
func (h *pipelineHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != "POST" {
w.Header().Set("Allow", "POST")
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
return
}
w.Header().Set("X-Etcd-Cluster-ID", h.cid.String())
if err := checkClusterCompatibilityFromHeader(r.Header, h.cid); err != nil {
http.Error(w, err.Error(), http.StatusPreconditionFailed)
return
}
if from, err := types.IDFromString(r.Header.Get("X-Server-From")); err != nil {
if urls := r.Header.Get("X-PeerURLs"); urls != "" {
h.tr.AddRemote(from, strings.Split(urls, ","))
}
}
// Limit the data size that could be read from the request body, which ensures that read from
// connection will not time out accidentally due to possible blocking in underlying implementation.
limitedr := pioutil.NewLimitedBufferReader(r.Body, connReadLimitByte)
b, err := ioutil.ReadAll(limitedr)
if err != nil {
plog.Errorf("failed to read raft message (%v)", err)
http.Error(w, "error reading raft message", http.StatusBadRequest)
recvFailures.WithLabelValues(r.RemoteAddr).Inc()
return
}
var m raftpb.Message
if err := m.Unmarshal(b); err != nil {
plog.Errorf("failed to unmarshal raft message (%v)", err)
http.Error(w, "error unmarshaling raft message", http.StatusBadRequest)
recvFailures.WithLabelValues(r.RemoteAddr).Inc()
return
}
receivedBytes.WithLabelValues(types.ID(m.From).String()).Add(float64(len(b)))
if err := h.r.Process(context.TODO(), m); err != nil {
switch v := err.(type) {
case writerToResponse:
v.WriteTo(w)
default:
plog.Warningf("failed to process raft message (%v)", err)
http.Error(w, "error processing raft message", http.StatusInternalServerError)
w.(http.Flusher).Flush()
// disconnect the http stream
panic(err)
}
return
}
// Write StatusNoContent header after the message has been processed by
// raft, which facilitates the client to report MsgSnap status.
w.WriteHeader(http.StatusNoContent)
}
type snapshotHandler struct {
tr Transporter
r Raft
snapshotter *snap.Snapshotter
cid types.ID
}
func newSnapshotHandler(tr Transporter, r Raft, snapshotter *snap.Snapshotter, cid types.ID) http.Handler {
return &snapshotHandler{
tr: tr,
r: r,
snapshotter: snapshotter,
cid: cid,
}
}
// ServeHTTP serves HTTP request to receive and process snapshot message.
//
// If request sender dies without closing underlying TCP connection,
// the handler will keep waiting for the request body until TCP keepalive
// finds out that the connection is broken after several minutes.
// This is acceptable because
// 1. snapshot messages sent through other TCP connections could still be
// received and processed.
// 2. this case should happen rarely, so no further optimization is done.
func (h *snapshotHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != "POST" {
w.Header().Set("Allow", "POST")
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
return
}
w.Header().Set("X-Etcd-Cluster-ID", h.cid.String())
if err := checkClusterCompatibilityFromHeader(r.Header, h.cid); err != nil {
http.Error(w, err.Error(), http.StatusPreconditionFailed)
return
}
if from, err := types.IDFromString(r.Header.Get("X-Server-From")); err != nil {
if urls := r.Header.Get("X-PeerURLs"); urls != "" {
h.tr.AddRemote(from, strings.Split(urls, ","))
}
}
dec := &messageDecoder{r: r.Body}
m, err := dec.decode()
if err != nil {
msg := fmt.Sprintf("failed to decode raft message (%v)", err)
plog.Errorf(msg)
http.Error(w, msg, http.StatusBadRequest)
recvFailures.WithLabelValues(r.RemoteAddr).Inc()
return
}
receivedBytes.WithLabelValues(types.ID(m.From).String()).Add(float64(m.Size()))
if m.Type != raftpb.MsgSnap {
plog.Errorf("unexpected raft message type %s on snapshot path", m.Type)
http.Error(w, "wrong raft message type", http.StatusBadRequest)
return
}
plog.Infof("receiving database snapshot [index:%d, from %s] ...", m.Snapshot.Metadata.Index, types.ID(m.From))
// save incoming database snapshot.
n, err := h.snapshotter.SaveDBFrom(r.Body, m.Snapshot.Metadata.Index)
if err != nil {
msg := fmt.Sprintf("failed to save KV snapshot (%v)", err)
plog.Error(msg)
http.Error(w, msg, http.StatusInternalServerError)
return
}
receivedBytes.WithLabelValues(types.ID(m.From).String()).Add(float64(n))
plog.Infof("received and saved database snapshot [index: %d, from: %s] successfully", m.Snapshot.Metadata.Index, types.ID(m.From))
if err := h.r.Process(context.TODO(), m); err != nil {
switch v := err.(type) {
// Process may return writerToResponse error when doing some
// additional checks before calling raft.Node.Step.
case writerToResponse:
v.WriteTo(w)
default:
msg := fmt.Sprintf("failed to process raft message (%v)", err)
plog.Warningf(msg)
http.Error(w, msg, http.StatusInternalServerError)
}
return
}
// Write StatusNoContent header after the message has been processed by
// raft, which facilitates the client to report MsgSnap status.
w.WriteHeader(http.StatusNoContent)
}
type streamHandler struct {
tr *Transport
peerGetter peerGetter
r Raft
id types.ID
cid types.ID
}
func newStreamHandler(tr *Transport, pg peerGetter, r Raft, id, cid types.ID) http.Handler {
return &streamHandler{
tr: tr,
peerGetter: pg,
r: r,
id: id,
cid: cid,
}
}
func (h *streamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != "GET" {
w.Header().Set("Allow", "GET")
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
return
}
w.Header().Set("X-Server-Version", version.Version)
w.Header().Set("X-Etcd-Cluster-ID", h.cid.String())
if err := checkClusterCompatibilityFromHeader(r.Header, h.cid); err != nil {
http.Error(w, err.Error(), http.StatusPreconditionFailed)
return
}
var t streamType
switch path.Dir(r.URL.Path) {
case streamTypeMsgAppV2.endpoint():
t = streamTypeMsgAppV2
case streamTypeMessage.endpoint():
t = streamTypeMessage
default:
plog.Debugf("ignored unexpected streaming request path %s", r.URL.Path)
http.Error(w, "invalid path", http.StatusNotFound)
return
}
fromStr := path.Base(r.URL.Path)
from, err := types.IDFromString(fromStr)
if err != nil {
plog.Errorf("failed to parse from %s into ID (%v)", fromStr, err)
http.Error(w, "invalid from", http.StatusNotFound)
return
}
if h.r.IsIDRemoved(uint64(from)) {
plog.Warningf("rejected the stream from peer %s since it was removed", from)
http.Error(w, "removed member", http.StatusGone)
return
}
p := h.peerGetter.Get(from)
if p == nil {
// This may happen in following cases:
// 1. user starts a remote peer that belongs to a different cluster
// with the same cluster ID.
// 2. local etcd falls behind of the cluster, and cannot recognize
// the members that joined after its current progress.
if urls := r.Header.Get("X-PeerURLs"); urls != "" {
h.tr.AddRemote(from, strings.Split(urls, ","))
}
plog.Errorf("failed to find member %s in cluster %s", from, h.cid)
http.Error(w, "error sender not found", http.StatusNotFound)
return
}
wto := h.id.String()
if gto := r.Header.Get("X-Raft-To"); gto != wto {
plog.Errorf("streaming request ignored (ID mismatch got %s want %s)", gto, wto)
http.Error(w, "to field mismatch", http.StatusPreconditionFailed)
return
}
w.WriteHeader(http.StatusOK)
w.(http.Flusher).Flush()
c := newCloseNotifier()
conn := &outgoingConn{
t: t,
Writer: w,
Flusher: w.(http.Flusher),
Closer: c,
}
p.attachOutgoingConn(conn)
<-c.closeNotify()
}
// checkClusterCompatibilityFromHeader checks the cluster compatibility of
// the local member from the given header.
// It checks whether the version of local member is compatible with
// the versions in the header, and whether the cluster ID of local member
// matches the one in the header.
func checkClusterCompatibilityFromHeader(header http.Header, cid types.ID) error {
if err := checkVersionCompability(header.Get("X-Server-From"), serverVersion(header), minClusterVersion(header)); err != nil {
plog.Errorf("request version incompatibility (%v)", err)
return errIncompatibleVersion
}
if gcid := header.Get("X-Etcd-Cluster-ID"); gcid != cid.String() {
plog.Errorf("request cluster ID mismatch (got %s want %s)", gcid, cid)
return errClusterIDMismatch
}
return nil
}
type closeNotifier struct {
done chan struct{}
}
func newCloseNotifier() *closeNotifier {
return &closeNotifier{
done: make(chan struct{}),
}
}
func (n *closeNotifier) Close() error {
close(n.done)
return nil
}
func (n *closeNotifier) closeNotify() <-chan struct{} { return n.done }

392
vendor/github.com/coreos/etcd/rafthttp/http_test.go generated vendored Normal file
View file

@ -0,0 +1,392 @@
// Copyright 2015 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 rafthttp
import (
"bytes"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
"time"
"github.com/coreos/etcd/pkg/pbutil"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/snap"
"github.com/coreos/etcd/version"
)
func TestServeRaftPrefix(t *testing.T) {
testCases := []struct {
method string
body io.Reader
p Raft
clusterID string
wcode int
}{
{
// bad method
"GET",
bytes.NewReader(
pbutil.MustMarshal(&raftpb.Message{}),
),
&fakeRaft{},
"0",
http.StatusMethodNotAllowed,
},
{
// bad method
"PUT",
bytes.NewReader(
pbutil.MustMarshal(&raftpb.Message{}),
),
&fakeRaft{},
"0",
http.StatusMethodNotAllowed,
},
{
// bad method
"DELETE",
bytes.NewReader(
pbutil.MustMarshal(&raftpb.Message{}),
),
&fakeRaft{},
"0",
http.StatusMethodNotAllowed,
},
{
// bad request body
"POST",
&errReader{},
&fakeRaft{},
"0",
http.StatusBadRequest,
},
{
// bad request protobuf
"POST",
strings.NewReader("malformed garbage"),
&fakeRaft{},
"0",
http.StatusBadRequest,
},
{
// good request, wrong cluster ID
"POST",
bytes.NewReader(
pbutil.MustMarshal(&raftpb.Message{}),
),
&fakeRaft{},
"1",
http.StatusPreconditionFailed,
},
{
// good request, Processor failure
"POST",
bytes.NewReader(
pbutil.MustMarshal(&raftpb.Message{}),
),
&fakeRaft{
err: &resWriterToError{code: http.StatusForbidden},
},
"0",
http.StatusForbidden,
},
{
// good request, Processor failure
"POST",
bytes.NewReader(
pbutil.MustMarshal(&raftpb.Message{}),
),
&fakeRaft{
err: &resWriterToError{code: http.StatusInternalServerError},
},
"0",
http.StatusInternalServerError,
},
{
// good request, Processor failure
"POST",
bytes.NewReader(
pbutil.MustMarshal(&raftpb.Message{}),
),
&fakeRaft{err: errors.New("blah")},
"0",
http.StatusInternalServerError,
},
{
// good request
"POST",
bytes.NewReader(
pbutil.MustMarshal(&raftpb.Message{}),
),
&fakeRaft{},
"0",
http.StatusNoContent,
},
}
for i, tt := range testCases {
req, err := http.NewRequest(tt.method, "foo", tt.body)
if err != nil {
t.Fatalf("#%d: could not create request: %#v", i, err)
}
req.Header.Set("X-Etcd-Cluster-ID", tt.clusterID)
req.Header.Set("X-Server-Version", version.Version)
rw := httptest.NewRecorder()
h := newPipelineHandler(NewNopTransporter(), tt.p, types.ID(0))
// goroutine because the handler panics to disconnect on raft error
donec := make(chan struct{})
go func() {
defer func() {
recover()
close(donec)
}()
h.ServeHTTP(rw, req)
}()
<-donec
if rw.Code != tt.wcode {
t.Errorf("#%d: got code=%d, want %d", i, rw.Code, tt.wcode)
}
}
}
func TestServeRaftStreamPrefix(t *testing.T) {
tests := []struct {
path string
wtype streamType
}{
{
RaftStreamPrefix + "/message/1",
streamTypeMessage,
},
{
RaftStreamPrefix + "/msgapp/1",
streamTypeMsgAppV2,
},
}
for i, tt := range tests {
req, err := http.NewRequest("GET", "http://localhost:2380"+tt.path, nil)
if err != nil {
t.Fatalf("#%d: could not create request: %#v", i, err)
}
req.Header.Set("X-Etcd-Cluster-ID", "1")
req.Header.Set("X-Server-Version", version.Version)
req.Header.Set("X-Raft-To", "2")
peer := newFakePeer()
peerGetter := &fakePeerGetter{peers: map[types.ID]Peer{types.ID(1): peer}}
tr := &Transport{}
h := newStreamHandler(tr, peerGetter, &fakeRaft{}, types.ID(2), types.ID(1))
rw := httptest.NewRecorder()
go h.ServeHTTP(rw, req)
var conn *outgoingConn
select {
case conn = <-peer.connc:
case <-time.After(time.Second):
t.Fatalf("#%d: failed to attach outgoingConn", i)
}
if g := rw.Header().Get("X-Server-Version"); g != version.Version {
t.Errorf("#%d: X-Server-Version = %s, want %s", i, g, version.Version)
}
if conn.t != tt.wtype {
t.Errorf("#%d: type = %s, want %s", i, conn.t, tt.wtype)
}
conn.Close()
}
}
func TestServeRaftStreamPrefixBad(t *testing.T) {
removedID := uint64(5)
tests := []struct {
method string
path string
clusterID string
remote string
wcode int
}{
// bad method
{
"PUT",
RaftStreamPrefix + "/message/1",
"1",
"1",
http.StatusMethodNotAllowed,
},
// bad method
{
"POST",
RaftStreamPrefix + "/message/1",
"1",
"1",
http.StatusMethodNotAllowed,
},
// bad method
{
"DELETE",
RaftStreamPrefix + "/message/1",
"1",
"1",
http.StatusMethodNotAllowed,
},
// bad path
{
"GET",
RaftStreamPrefix + "/strange/1",
"1",
"1",
http.StatusNotFound,
},
// bad path
{
"GET",
RaftStreamPrefix + "/strange",
"1",
"1",
http.StatusNotFound,
},
// non-existent peer
{
"GET",
RaftStreamPrefix + "/message/2",
"1",
"1",
http.StatusNotFound,
},
// removed peer
{
"GET",
RaftStreamPrefix + "/message/" + fmt.Sprint(removedID),
"1",
"1",
http.StatusGone,
},
// wrong cluster ID
{
"GET",
RaftStreamPrefix + "/message/1",
"2",
"1",
http.StatusPreconditionFailed,
},
// wrong remote id
{
"GET",
RaftStreamPrefix + "/message/1",
"1",
"2",
http.StatusPreconditionFailed,
},
}
for i, tt := range tests {
req, err := http.NewRequest(tt.method, "http://localhost:2380"+tt.path, nil)
if err != nil {
t.Fatalf("#%d: could not create request: %#v", i, err)
}
req.Header.Set("X-Etcd-Cluster-ID", tt.clusterID)
req.Header.Set("X-Server-Version", version.Version)
req.Header.Set("X-Raft-To", tt.remote)
rw := httptest.NewRecorder()
tr := &Transport{}
peerGetter := &fakePeerGetter{peers: map[types.ID]Peer{types.ID(1): newFakePeer()}}
r := &fakeRaft{removedID: removedID}
h := newStreamHandler(tr, peerGetter, r, types.ID(1), types.ID(1))
h.ServeHTTP(rw, req)
if rw.Code != tt.wcode {
t.Errorf("#%d: code = %d, want %d", i, rw.Code, tt.wcode)
}
}
}
func TestCloseNotifier(t *testing.T) {
c := newCloseNotifier()
select {
case <-c.closeNotify():
t.Fatalf("received unexpected close notification")
default:
}
c.Close()
select {
case <-c.closeNotify():
default:
t.Fatalf("failed to get close notification")
}
}
// errReader implements io.Reader to facilitate a broken request.
type errReader struct{}
func (er *errReader) Read(_ []byte) (int, error) { return 0, errors.New("some error") }
type resWriterToError struct {
code int
}
func (e *resWriterToError) Error() string { return "" }
func (e *resWriterToError) WriteTo(w http.ResponseWriter) { w.WriteHeader(e.code) }
type fakePeerGetter struct {
peers map[types.ID]Peer
}
func (pg *fakePeerGetter) Get(id types.ID) Peer { return pg.peers[id] }
type fakePeer struct {
msgs []raftpb.Message
snapMsgs []snap.Message
peerURLs types.URLs
connc chan *outgoingConn
paused bool
}
func newFakePeer() *fakePeer {
fakeURL, _ := url.Parse("http://localhost")
return &fakePeer{
connc: make(chan *outgoingConn, 1),
peerURLs: types.URLs{*fakeURL},
}
}
func (pr *fakePeer) send(m raftpb.Message) {
if pr.paused {
return
}
pr.msgs = append(pr.msgs, m)
}
func (pr *fakePeer) sendSnap(m snap.Message) {
if pr.paused {
return
}
pr.snapMsgs = append(pr.snapMsgs, m)
}
func (pr *fakePeer) update(urls types.URLs) { pr.peerURLs = urls }
func (pr *fakePeer) attachOutgoingConn(conn *outgoingConn) { pr.connc <- conn }
func (pr *fakePeer) activeSince() time.Time { return time.Time{} }
func (pr *fakePeer) stop() {}
func (pr *fakePeer) Pause() { pr.paused = true }
func (pr *fakePeer) Resume() { pr.paused = false }

73
vendor/github.com/coreos/etcd/rafthttp/metrics.go generated vendored Normal file
View file

@ -0,0 +1,73 @@
// Copyright 2015 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 rafthttp
import "github.com/prometheus/client_golang/prometheus"
var (
sentBytes = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: "etcd",
Subsystem: "network",
Name: "peer_sent_bytes_total",
Help: "The total number of bytes sent to peers.",
},
[]string{"To"},
)
receivedBytes = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: "etcd",
Subsystem: "network",
Name: "peer_received_bytes_total",
Help: "The total number of bytes received from peers.",
},
[]string{"From"},
)
sentFailures = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: "etcd",
Subsystem: "network",
Name: "peer_sent_failures_total",
Help: "The total number of send failures from peers.",
},
[]string{"To"},
)
recvFailures = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: "etcd",
Subsystem: "network",
Name: "peer_received_failures_total",
Help: "The total number of receive failures from peers.",
},
[]string{"From"},
)
rtts = prometheus.NewHistogramVec(prometheus.HistogramOpts{
Namespace: "etcd",
Subsystem: "network",
Name: "peer_round_trip_time_seconds",
Help: "Round-Trip-Time histogram between peers.",
Buckets: prometheus.ExponentialBuckets(0.0001, 2, 14),
},
[]string{"To"},
)
)
func init() {
prometheus.MustRegister(sentBytes)
prometheus.MustRegister(receivedBytes)
prometheus.MustRegister(sentFailures)
prometheus.MustRegister(recvFailures)
prometheus.MustRegister(rtts)
}

64
vendor/github.com/coreos/etcd/rafthttp/msg_codec.go generated vendored Normal file
View file

@ -0,0 +1,64 @@
// Copyright 2015 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 rafthttp
import (
"encoding/binary"
"errors"
"io"
"github.com/coreos/etcd/pkg/pbutil"
"github.com/coreos/etcd/raft/raftpb"
)
// messageEncoder is a encoder that can encode all kinds of messages.
// It MUST be used with a paired messageDecoder.
type messageEncoder struct {
w io.Writer
}
func (enc *messageEncoder) encode(m *raftpb.Message) error {
if err := binary.Write(enc.w, binary.BigEndian, uint64(m.Size())); err != nil {
return err
}
_, err := enc.w.Write(pbutil.MustMarshal(m))
return err
}
// messageDecoder is a decoder that can decode all kinds of messages.
type messageDecoder struct {
r io.Reader
}
var (
readBytesLimit uint64 = 512 * 1024 * 1024 // 512 MB
ErrExceedSizeLimit = errors.New("rafthttp: error limit exceeded")
)
func (dec *messageDecoder) decode() (raftpb.Message, error) {
var m raftpb.Message
var l uint64
if err := binary.Read(dec.r, binary.BigEndian, &l); err != nil {
return m, err
}
if l > readBytesLimit {
return m, ErrExceedSizeLimit
}
buf := make([]byte, int(l))
if _, err := io.ReadFull(dec.r, buf); err != nil {
return m, err
}
return m, m.Unmarshal(buf)
}

View file

@ -0,0 +1,96 @@
// Copyright 2015 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 rafthttp
import (
"bytes"
"reflect"
"testing"
"github.com/coreos/etcd/raft/raftpb"
)
func TestMessage(t *testing.T) {
// Lower readBytesLimit to make test pass in restricted resources environment
originalLimit := readBytesLimit
readBytesLimit = 1000
defer func() {
readBytesLimit = originalLimit
}()
tests := []struct {
msg raftpb.Message
encodeErr error
decodeErr error
}{
{
raftpb.Message{
Type: raftpb.MsgApp,
From: 1,
To: 2,
Term: 1,
LogTerm: 1,
Index: 3,
Entries: []raftpb.Entry{{Term: 1, Index: 4}},
},
nil,
nil,
},
{
raftpb.Message{
Type: raftpb.MsgProp,
From: 1,
To: 2,
Entries: []raftpb.Entry{
{Data: []byte("some data")},
{Data: []byte("some data")},
{Data: []byte("some data")},
},
},
nil,
nil,
},
{
raftpb.Message{
Type: raftpb.MsgProp,
From: 1,
To: 2,
Entries: []raftpb.Entry{
{Data: bytes.Repeat([]byte("a"), int(readBytesLimit+10))},
},
},
nil,
ErrExceedSizeLimit,
},
}
for i, tt := range tests {
b := &bytes.Buffer{}
enc := &messageEncoder{w: b}
if err := enc.encode(&tt.msg); err != tt.encodeErr {
t.Errorf("#%d: encode message error expected %v, got %v", i, tt.encodeErr, err)
continue
}
dec := &messageDecoder{r: b}
m, err := dec.decode()
if err != tt.decodeErr {
t.Errorf("#%d: decode message error expected %v, got %v", i, tt.decodeErr, err)
continue
}
if err == nil {
if !reflect.DeepEqual(m, tt.msg) {
t.Errorf("#%d: message = %+v, want %+v", i, m, tt.msg)
}
}
}
}

View file

@ -0,0 +1,248 @@
// Copyright 2015 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 rafthttp
import (
"encoding/binary"
"fmt"
"io"
"time"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/pbutil"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
)
const (
msgTypeLinkHeartbeat uint8 = 0
msgTypeAppEntries uint8 = 1
msgTypeApp uint8 = 2
msgAppV2BufSize = 1024 * 1024
)
// msgappv2 stream sends three types of message: linkHeartbeatMessage,
// AppEntries and MsgApp. AppEntries is the MsgApp that is sent in
// replicate state in raft, whose index and term are fully predictable.
//
// Data format of linkHeartbeatMessage:
// | offset | bytes | description |
// +--------+-------+-------------+
// | 0 | 1 | \x00 |
//
// Data format of AppEntries:
// | offset | bytes | description |
// +--------+-------+-------------+
// | 0 | 1 | \x01 |
// | 1 | 8 | length of entries |
// | 9 | 8 | length of first entry |
// | 17 | n1 | first entry |
// ...
// | x | 8 | length of k-th entry data |
// | x+8 | nk | k-th entry data |
// | x+8+nk | 8 | commit index |
//
// Data format of MsgApp:
// | offset | bytes | description |
// +--------+-------+-------------+
// | 0 | 1 | \x02 |
// | 1 | 8 | length of encoded message |
// | 9 | n | encoded message |
type msgAppV2Encoder struct {
w io.Writer
fs *stats.FollowerStats
term uint64
index uint64
buf []byte
uint64buf []byte
uint8buf []byte
}
func newMsgAppV2Encoder(w io.Writer, fs *stats.FollowerStats) *msgAppV2Encoder {
return &msgAppV2Encoder{
w: w,
fs: fs,
buf: make([]byte, msgAppV2BufSize),
uint64buf: make([]byte, 8),
uint8buf: make([]byte, 1),
}
}
func (enc *msgAppV2Encoder) encode(m *raftpb.Message) error {
start := time.Now()
switch {
case isLinkHeartbeatMessage(m):
enc.uint8buf[0] = byte(msgTypeLinkHeartbeat)
if _, err := enc.w.Write(enc.uint8buf); err != nil {
return err
}
case enc.index == m.Index && enc.term == m.LogTerm && m.LogTerm == m.Term:
enc.uint8buf[0] = byte(msgTypeAppEntries)
if _, err := enc.w.Write(enc.uint8buf); err != nil {
return err
}
// write length of entries
binary.BigEndian.PutUint64(enc.uint64buf, uint64(len(m.Entries)))
if _, err := enc.w.Write(enc.uint64buf); err != nil {
return err
}
for i := 0; i < len(m.Entries); i++ {
// write length of entry
binary.BigEndian.PutUint64(enc.uint64buf, uint64(m.Entries[i].Size()))
if _, err := enc.w.Write(enc.uint64buf); err != nil {
return err
}
if n := m.Entries[i].Size(); n < msgAppV2BufSize {
if _, err := m.Entries[i].MarshalTo(enc.buf); err != nil {
return err
}
if _, err := enc.w.Write(enc.buf[:n]); err != nil {
return err
}
} else {
if _, err := enc.w.Write(pbutil.MustMarshal(&m.Entries[i])); err != nil {
return err
}
}
enc.index++
}
// write commit index
binary.BigEndian.PutUint64(enc.uint64buf, m.Commit)
if _, err := enc.w.Write(enc.uint64buf); err != nil {
return err
}
enc.fs.Succ(time.Since(start))
default:
if err := binary.Write(enc.w, binary.BigEndian, msgTypeApp); err != nil {
return err
}
// write size of message
if err := binary.Write(enc.w, binary.BigEndian, uint64(m.Size())); err != nil {
return err
}
// write message
if _, err := enc.w.Write(pbutil.MustMarshal(m)); err != nil {
return err
}
enc.term = m.Term
enc.index = m.Index
if l := len(m.Entries); l > 0 {
enc.index = m.Entries[l-1].Index
}
enc.fs.Succ(time.Since(start))
}
return nil
}
type msgAppV2Decoder struct {
r io.Reader
local, remote types.ID
term uint64
index uint64
buf []byte
uint64buf []byte
uint8buf []byte
}
func newMsgAppV2Decoder(r io.Reader, local, remote types.ID) *msgAppV2Decoder {
return &msgAppV2Decoder{
r: r,
local: local,
remote: remote,
buf: make([]byte, msgAppV2BufSize),
uint64buf: make([]byte, 8),
uint8buf: make([]byte, 1),
}
}
func (dec *msgAppV2Decoder) decode() (raftpb.Message, error) {
var (
m raftpb.Message
typ uint8
)
if _, err := io.ReadFull(dec.r, dec.uint8buf); err != nil {
return m, err
}
typ = uint8(dec.uint8buf[0])
switch typ {
case msgTypeLinkHeartbeat:
return linkHeartbeatMessage, nil
case msgTypeAppEntries:
m = raftpb.Message{
Type: raftpb.MsgApp,
From: uint64(dec.remote),
To: uint64(dec.local),
Term: dec.term,
LogTerm: dec.term,
Index: dec.index,
}
// decode entries
if _, err := io.ReadFull(dec.r, dec.uint64buf); err != nil {
return m, err
}
l := binary.BigEndian.Uint64(dec.uint64buf)
m.Entries = make([]raftpb.Entry, int(l))
for i := 0; i < int(l); i++ {
if _, err := io.ReadFull(dec.r, dec.uint64buf); err != nil {
return m, err
}
size := binary.BigEndian.Uint64(dec.uint64buf)
var buf []byte
if size < msgAppV2BufSize {
buf = dec.buf[:size]
if _, err := io.ReadFull(dec.r, buf); err != nil {
return m, err
}
} else {
buf = make([]byte, int(size))
if _, err := io.ReadFull(dec.r, buf); err != nil {
return m, err
}
}
dec.index++
// 1 alloc
pbutil.MustUnmarshal(&m.Entries[i], buf)
}
// decode commit index
if _, err := io.ReadFull(dec.r, dec.uint64buf); err != nil {
return m, err
}
m.Commit = binary.BigEndian.Uint64(dec.uint64buf)
case msgTypeApp:
var size uint64
if err := binary.Read(dec.r, binary.BigEndian, &size); err != nil {
return m, err
}
buf := make([]byte, int(size))
if _, err := io.ReadFull(dec.r, buf); err != nil {
return m, err
}
pbutil.MustUnmarshal(&m, buf)
dec.term = m.Term
dec.index = m.Index
if l := len(m.Entries); l > 0 {
dec.index = m.Entries[l-1].Index
}
default:
return m, fmt.Errorf("failed to parse type %d in msgappv2 stream", typ)
}
return m, nil
}

View file

@ -0,0 +1,123 @@
// Copyright 2015 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 rafthttp
import (
"bytes"
"reflect"
"testing"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
)
func TestMsgAppV2(t *testing.T) {
tests := []raftpb.Message{
linkHeartbeatMessage,
{
Type: raftpb.MsgApp,
From: 1,
To: 2,
Term: 1,
LogTerm: 1,
Index: 0,
Entries: []raftpb.Entry{
{Term: 1, Index: 1, Data: []byte("some data")},
{Term: 1, Index: 2, Data: []byte("some data")},
{Term: 1, Index: 3, Data: []byte("some data")},
},
},
// consecutive MsgApp
{
Type: raftpb.MsgApp,
From: 1,
To: 2,
Term: 1,
LogTerm: 1,
Index: 3,
Entries: []raftpb.Entry{
{Term: 1, Index: 4, Data: []byte("some data")},
},
},
linkHeartbeatMessage,
// consecutive MsgApp after linkHeartbeatMessage
{
Type: raftpb.MsgApp,
From: 1,
To: 2,
Term: 1,
LogTerm: 1,
Index: 4,
Entries: []raftpb.Entry{
{Term: 1, Index: 5, Data: []byte("some data")},
},
},
// MsgApp with higher term
{
Type: raftpb.MsgApp,
From: 1,
To: 2,
Term: 3,
LogTerm: 1,
Index: 5,
Entries: []raftpb.Entry{
{Term: 3, Index: 6, Data: []byte("some data")},
},
},
linkHeartbeatMessage,
// consecutive MsgApp
{
Type: raftpb.MsgApp,
From: 1,
To: 2,
Term: 3,
LogTerm: 2,
Index: 6,
Entries: []raftpb.Entry{
{Term: 3, Index: 7, Data: []byte("some data")},
},
},
// consecutive empty MsgApp
{
Type: raftpb.MsgApp,
From: 1,
To: 2,
Term: 3,
LogTerm: 2,
Index: 7,
Entries: nil,
},
linkHeartbeatMessage,
}
b := &bytes.Buffer{}
enc := newMsgAppV2Encoder(b, &stats.FollowerStats{})
dec := newMsgAppV2Decoder(b, types.ID(2), types.ID(1))
for i, tt := range tests {
if err := enc.encode(&tt); err != nil {
t.Errorf("#%d: unexpected encode message error: %v", i, err)
continue
}
m, err := dec.decode()
if err != nil {
t.Errorf("#%d: unexpected decode message error: %v", i, err)
continue
}
if !reflect.DeepEqual(m, tt) {
t.Errorf("#%d: message = %+v, want %+v", i, m, tt)
}
}
}

307
vendor/github.com/coreos/etcd/rafthttp/peer.go generated vendored Normal file
View file

@ -0,0 +1,307 @@
// Copyright 2015 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 rafthttp
import (
"sync"
"time"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/snap"
"golang.org/x/net/context"
)
const (
// ConnReadTimeout and ConnWriteTimeout are the i/o timeout set on each connection rafthttp pkg creates.
// A 5 seconds timeout is good enough for recycling bad connections. Or we have to wait for
// tcp keepalive failing to detect a bad connection, which is at minutes level.
// For long term streaming connections, rafthttp pkg sends application level linkHeartbeatMessage
// to keep the connection alive.
// For short term pipeline connections, the connection MUST be killed to avoid it being
// put back to http pkg connection pool.
ConnReadTimeout = 5 * time.Second
ConnWriteTimeout = 5 * time.Second
recvBufSize = 4096
// maxPendingProposals holds the proposals during one leader election process.
// Generally one leader election takes at most 1 sec. It should have
// 0-2 election conflicts, and each one takes 0.5 sec.
// We assume the number of concurrent proposers is smaller than 4096.
// One client blocks on its proposal for at least 1 sec, so 4096 is enough
// to hold all proposals.
maxPendingProposals = 4096
streamAppV2 = "streamMsgAppV2"
streamMsg = "streamMsg"
pipelineMsg = "pipeline"
sendSnap = "sendMsgSnap"
)
type Peer interface {
// send sends the message to the remote peer. The function is non-blocking
// and has no promise that the message will be received by the remote.
// When it fails to send message out, it will report the status to underlying
// raft.
send(m raftpb.Message)
// sendSnap sends the merged snapshot message to the remote peer. Its behavior
// is similar to send.
sendSnap(m snap.Message)
// update updates the urls of remote peer.
update(urls types.URLs)
// attachOutgoingConn attaches the outgoing connection to the peer for
// stream usage. After the call, the ownership of the outgoing
// connection hands over to the peer. The peer will close the connection
// when it is no longer used.
attachOutgoingConn(conn *outgoingConn)
// activeSince returns the time that the connection with the
// peer becomes active.
activeSince() time.Time
// stop performs any necessary finalization and terminates the peer
// elegantly.
stop()
}
// peer is the representative of a remote raft node. Local raft node sends
// messages to the remote through peer.
// Each peer has two underlying mechanisms to send out a message: stream and
// pipeline.
// A stream is a receiver initialized long-polling connection, which
// is always open to transfer messages. Besides general stream, peer also has
// a optimized stream for sending msgApp since msgApp accounts for large part
// of all messages. Only raft leader uses the optimized stream to send msgApp
// to the remote follower node.
// A pipeline is a series of http clients that send http requests to the remote.
// It is only used when the stream has not been established.
type peer struct {
// id of the remote raft peer node
id types.ID
r Raft
status *peerStatus
picker *urlPicker
msgAppV2Writer *streamWriter
writer *streamWriter
pipeline *pipeline
snapSender *snapshotSender // snapshot sender to send v3 snapshot messages
msgAppV2Reader *streamReader
msgAppReader *streamReader
recvc chan raftpb.Message
propc chan raftpb.Message
mu sync.Mutex
paused bool
cancel context.CancelFunc // cancel pending works in go routine created by peer.
stopc chan struct{}
}
func startPeer(transport *Transport, urls types.URLs, peerID types.ID, fs *stats.FollowerStats) *peer {
plog.Infof("starting peer %s...", peerID)
defer plog.Infof("started peer %s", peerID)
status := newPeerStatus(peerID)
picker := newURLPicker(urls)
errorc := transport.ErrorC
r := transport.Raft
pipeline := &pipeline{
peerID: peerID,
tr: transport,
picker: picker,
status: status,
followerStats: fs,
raft: r,
errorc: errorc,
}
pipeline.start()
p := &peer{
id: peerID,
r: r,
status: status,
picker: picker,
msgAppV2Writer: startStreamWriter(peerID, status, fs, r),
writer: startStreamWriter(peerID, status, fs, r),
pipeline: pipeline,
snapSender: newSnapshotSender(transport, picker, peerID, status),
recvc: make(chan raftpb.Message, recvBufSize),
propc: make(chan raftpb.Message, maxPendingProposals),
stopc: make(chan struct{}),
}
ctx, cancel := context.WithCancel(context.Background())
p.cancel = cancel
go func() {
for {
select {
case mm := <-p.recvc:
if err := r.Process(ctx, mm); err != nil {
plog.Warningf("failed to process raft message (%v)", err)
}
case <-p.stopc:
return
}
}
}()
// r.Process might block for processing proposal when there is no leader.
// Thus propc must be put into a separate routine with recvc to avoid blocking
// processing other raft messages.
go func() {
for {
select {
case mm := <-p.propc:
if err := r.Process(ctx, mm); err != nil {
plog.Warningf("failed to process raft message (%v)", err)
}
case <-p.stopc:
return
}
}
}()
p.msgAppV2Reader = &streamReader{
peerID: peerID,
typ: streamTypeMsgAppV2,
tr: transport,
picker: picker,
status: status,
recvc: p.recvc,
propc: p.propc,
}
p.msgAppReader = &streamReader{
peerID: peerID,
typ: streamTypeMessage,
tr: transport,
picker: picker,
status: status,
recvc: p.recvc,
propc: p.propc,
}
p.msgAppV2Reader.start()
p.msgAppReader.start()
return p
}
func (p *peer) send(m raftpb.Message) {
p.mu.Lock()
paused := p.paused
p.mu.Unlock()
if paused {
return
}
writec, name := p.pick(m)
select {
case writec <- m:
default:
p.r.ReportUnreachable(m.To)
if isMsgSnap(m) {
p.r.ReportSnapshot(m.To, raft.SnapshotFailure)
}
if p.status.isActive() {
plog.MergeWarningf("dropped internal raft message to %s since %s's sending buffer is full (bad/overloaded network)", p.id, name)
}
plog.Debugf("dropped %s to %s since %s's sending buffer is full", m.Type, p.id, name)
}
}
func (p *peer) sendSnap(m snap.Message) {
go p.snapSender.send(m)
}
func (p *peer) update(urls types.URLs) {
p.picker.update(urls)
}
func (p *peer) attachOutgoingConn(conn *outgoingConn) {
var ok bool
switch conn.t {
case streamTypeMsgAppV2:
ok = p.msgAppV2Writer.attach(conn)
case streamTypeMessage:
ok = p.writer.attach(conn)
default:
plog.Panicf("unhandled stream type %s", conn.t)
}
if !ok {
conn.Close()
}
}
func (p *peer) activeSince() time.Time { return p.status.activeSince() }
// Pause pauses the peer. The peer will simply drops all incoming
// messages without returning an error.
func (p *peer) Pause() {
p.mu.Lock()
defer p.mu.Unlock()
p.paused = true
p.msgAppReader.pause()
p.msgAppV2Reader.pause()
}
// Resume resumes a paused peer.
func (p *peer) Resume() {
p.mu.Lock()
defer p.mu.Unlock()
p.paused = false
p.msgAppReader.resume()
p.msgAppV2Reader.resume()
}
func (p *peer) stop() {
plog.Infof("stopping peer %s...", p.id)
defer plog.Infof("stopped peer %s", p.id)
close(p.stopc)
p.cancel()
p.msgAppV2Writer.stop()
p.writer.stop()
p.pipeline.stop()
p.snapSender.stop()
p.msgAppV2Reader.stop()
p.msgAppReader.stop()
}
// pick picks a chan for sending the given message. The picked chan and the picked chan
// string name are returned.
func (p *peer) pick(m raftpb.Message) (writec chan<- raftpb.Message, picked string) {
var ok bool
// Considering MsgSnap may have a big size, e.g., 1G, and will block
// stream for a long time, only use one of the N pipelines to send MsgSnap.
if isMsgSnap(m) {
return p.pipeline.msgc, pipelineMsg
} else if writec, ok = p.msgAppV2Writer.writec(); ok && isMsgApp(m) {
return writec, streamAppV2
} else if writec, ok = p.writer.writec(); ok {
return writec, streamMsg
}
return p.pipeline.msgc, pipelineMsg
}
func isMsgApp(m raftpb.Message) bool { return m.Type == raftpb.MsgApp }
func isMsgSnap(m raftpb.Message) bool { return m.Type == raftpb.MsgSnap }

77
vendor/github.com/coreos/etcd/rafthttp/peer_status.go generated vendored Normal file
View file

@ -0,0 +1,77 @@
// Copyright 2015 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 rafthttp
import (
"fmt"
"sync"
"time"
"github.com/coreos/etcd/pkg/types"
)
type failureType struct {
source string
action string
}
type peerStatus struct {
id types.ID
mu sync.Mutex // protect variables below
active bool
since time.Time
}
func newPeerStatus(id types.ID) *peerStatus {
return &peerStatus{
id: id,
}
}
func (s *peerStatus) activate() {
s.mu.Lock()
defer s.mu.Unlock()
if !s.active {
plog.Infof("peer %s became active", s.id)
s.active = true
s.since = time.Now()
}
}
func (s *peerStatus) deactivate(failure failureType, reason string) {
s.mu.Lock()
defer s.mu.Unlock()
msg := fmt.Sprintf("failed to %s %s on %s (%s)", failure.action, s.id, failure.source, reason)
if s.active {
plog.Errorf(msg)
plog.Infof("peer %s became inactive", s.id)
s.active = false
s.since = time.Time{}
return
}
plog.Debugf(msg)
}
func (s *peerStatus) isActive() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.active
}
func (s *peerStatus) activeSince() time.Time {
s.mu.Lock()
defer s.mu.Unlock()
return s.since
}

87
vendor/github.com/coreos/etcd/rafthttp/peer_test.go generated vendored Normal file
View file

@ -0,0 +1,87 @@
// Copyright 2015 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 rafthttp
import (
"testing"
"github.com/coreos/etcd/raft/raftpb"
)
func TestPeerPick(t *testing.T) {
tests := []struct {
msgappWorking bool
messageWorking bool
m raftpb.Message
wpicked string
}{
{
true, true,
raftpb.Message{Type: raftpb.MsgSnap},
pipelineMsg,
},
{
true, true,
raftpb.Message{Type: raftpb.MsgApp, Term: 1, LogTerm: 1},
streamAppV2,
},
{
true, true,
raftpb.Message{Type: raftpb.MsgProp},
streamMsg,
},
{
true, true,
raftpb.Message{Type: raftpb.MsgHeartbeat},
streamMsg,
},
{
false, true,
raftpb.Message{Type: raftpb.MsgApp, Term: 1, LogTerm: 1},
streamMsg,
},
{
false, false,
raftpb.Message{Type: raftpb.MsgApp, Term: 1, LogTerm: 1},
pipelineMsg,
},
{
false, false,
raftpb.Message{Type: raftpb.MsgProp},
pipelineMsg,
},
{
false, false,
raftpb.Message{Type: raftpb.MsgSnap},
pipelineMsg,
},
{
false, false,
raftpb.Message{Type: raftpb.MsgHeartbeat},
pipelineMsg,
},
}
for i, tt := range tests {
peer := &peer{
msgAppV2Writer: &streamWriter{working: tt.msgappWorking},
writer: &streamWriter{working: tt.messageWorking},
pipeline: &pipeline{},
}
_, picked := peer.pick(tt.m)
if picked != tt.wpicked {
t.Errorf("#%d: picked = %v, want %v", i, picked, tt.wpicked)
}
}
}

159
vendor/github.com/coreos/etcd/rafthttp/pipeline.go generated vendored Normal file
View file

@ -0,0 +1,159 @@
// Copyright 2015 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 rafthttp
import (
"bytes"
"errors"
"io/ioutil"
"sync"
"time"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/httputil"
"github.com/coreos/etcd/pkg/pbutil"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft"
"github.com/coreos/etcd/raft/raftpb"
)
const (
connPerPipeline = 4
// pipelineBufSize is the size of pipeline buffer, which helps hold the
// temporary network latency.
// The size ensures that pipeline does not drop messages when the network
// is out of work for less than 1 second in good path.
pipelineBufSize = 64
)
var errStopped = errors.New("stopped")
type pipeline struct {
peerID types.ID
tr *Transport
picker *urlPicker
status *peerStatus
raft Raft
errorc chan error
// deprecate when we depercate v2 API
followerStats *stats.FollowerStats
msgc chan raftpb.Message
// wait for the handling routines
wg sync.WaitGroup
stopc chan struct{}
}
func (p *pipeline) start() {
p.stopc = make(chan struct{})
p.msgc = make(chan raftpb.Message, pipelineBufSize)
p.wg.Add(connPerPipeline)
for i := 0; i < connPerPipeline; i++ {
go p.handle()
}
plog.Infof("started HTTP pipelining with peer %s", p.peerID)
}
func (p *pipeline) stop() {
close(p.stopc)
p.wg.Wait()
plog.Infof("stopped HTTP pipelining with peer %s", p.peerID)
}
func (p *pipeline) handle() {
defer p.wg.Done()
for {
select {
case m := <-p.msgc:
start := time.Now()
err := p.post(pbutil.MustMarshal(&m))
end := time.Now()
if err != nil {
p.status.deactivate(failureType{source: pipelineMsg, action: "write"}, err.Error())
if m.Type == raftpb.MsgApp && p.followerStats != nil {
p.followerStats.Fail()
}
p.raft.ReportUnreachable(m.To)
if isMsgSnap(m) {
p.raft.ReportSnapshot(m.To, raft.SnapshotFailure)
}
sentFailures.WithLabelValues(types.ID(m.To).String()).Inc()
continue
}
p.status.activate()
if m.Type == raftpb.MsgApp && p.followerStats != nil {
p.followerStats.Succ(end.Sub(start))
}
if isMsgSnap(m) {
p.raft.ReportSnapshot(m.To, raft.SnapshotFinish)
}
sentBytes.WithLabelValues(types.ID(m.To).String()).Add(float64(m.Size()))
case <-p.stopc:
return
}
}
}
// post POSTs a data payload to a url. Returns nil if the POST succeeds,
// error on any failure.
func (p *pipeline) post(data []byte) (err error) {
u := p.picker.pick()
req := createPostRequest(u, RaftPrefix, bytes.NewBuffer(data), "application/protobuf", p.tr.URLs, p.tr.ID, p.tr.ClusterID)
done := make(chan struct{}, 1)
cancel := httputil.RequestCanceler(req)
go func() {
select {
case <-done:
case <-p.stopc:
waitSchedule()
cancel()
}
}()
resp, err := p.tr.pipelineRt.RoundTrip(req)
done <- struct{}{}
if err != nil {
p.picker.unreachable(u)
return err
}
b, err := ioutil.ReadAll(resp.Body)
if err != nil {
p.picker.unreachable(u)
return err
}
resp.Body.Close()
err = checkPostResponse(resp, b, req, p.peerID)
if err != nil {
p.picker.unreachable(u)
// errMemberRemoved is a critical error since a removed member should
// always be stopped. So we use reportCriticalError to report it to errorc.
if err == errMemberRemoved {
reportCriticalError(err, p.errorc)
}
return err
}
return nil
}
// waitSchedule waits other goroutines to be scheduled for a while
func waitSchedule() { time.Sleep(time.Millisecond) }

311
vendor/github.com/coreos/etcd/rafthttp/pipeline_test.go generated vendored Normal file
View file

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

View file

@ -0,0 +1,67 @@
// Copyright 2015 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 rafthttp
import (
"time"
"github.com/xiang90/probing"
)
var (
// proberInterval must be shorter than read timeout.
// Or the connection will time-out.
proberInterval = ConnReadTimeout - time.Second
statusMonitoringInterval = 30 * time.Second
statusErrorInterval = 5 * time.Second
)
func addPeerToProber(p probing.Prober, id string, us []string) {
hus := make([]string, len(us))
for i := range us {
hus[i] = us[i] + ProbingPrefix
}
p.AddHTTP(id, proberInterval, hus)
s, err := p.Status(id)
if err != nil {
plog.Errorf("failed to add peer %s into prober", id)
} else {
go monitorProbingStatus(s, id)
}
}
func monitorProbingStatus(s probing.Status, id string) {
// set the first interval short to log error early.
interval := statusErrorInterval
for {
select {
case <-time.After(interval):
if !s.Health() {
plog.Warningf("health check for peer %s could not connect: %v", id, s.Err())
interval = statusErrorInterval
} else {
interval = statusMonitoringInterval
}
if s.ClockDiff() > time.Second {
plog.Warningf("the clock difference against peer %s is too high [%v > %v]", id, s.ClockDiff(), time.Second)
}
rtts.WithLabelValues(id).Observe(s.SRTT().Seconds())
case <-s.StopNotify():
return
}
}
}

69
vendor/github.com/coreos/etcd/rafthttp/remote.go generated vendored Normal file
View file

@ -0,0 +1,69 @@
// Copyright 2015 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 rafthttp
import (
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
)
type remote struct {
id types.ID
status *peerStatus
pipeline *pipeline
}
func startRemote(tr *Transport, urls types.URLs, id types.ID) *remote {
picker := newURLPicker(urls)
status := newPeerStatus(id)
pipeline := &pipeline{
peerID: id,
tr: tr,
picker: picker,
status: status,
raft: tr.Raft,
errorc: tr.ErrorC,
}
pipeline.start()
return &remote{
id: id,
status: status,
pipeline: pipeline,
}
}
func (g *remote) send(m raftpb.Message) {
select {
case g.pipeline.msgc <- m:
default:
if g.status.isActive() {
plog.MergeWarningf("dropped internal raft message to %s since sending buffer is full (bad/overloaded network)", g.id)
}
plog.Debugf("dropped %s to %s since sending buffer is full", m.Type, g.id)
}
}
func (g *remote) stop() {
g.pipeline.stop()
}
func (g *remote) Pause() {
g.stop()
}
func (g *remote) Resume() {
g.pipeline.start()
}

View file

@ -0,0 +1,155 @@
// Copyright 2015 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 rafthttp
import (
"bytes"
"io"
"io/ioutil"
"net/http"
"time"
"github.com/coreos/etcd/pkg/httputil"
pioutil "github.com/coreos/etcd/pkg/ioutil"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft"
"github.com/coreos/etcd/snap"
)
var (
// timeout for reading snapshot response body
snapResponseReadTimeout = 5 * time.Second
)
type snapshotSender struct {
from, to types.ID
cid types.ID
tr *Transport
picker *urlPicker
status *peerStatus
r Raft
errorc chan error
stopc chan struct{}
}
func newSnapshotSender(tr *Transport, picker *urlPicker, to types.ID, status *peerStatus) *snapshotSender {
return &snapshotSender{
from: tr.ID,
to: to,
cid: tr.ClusterID,
tr: tr,
picker: picker,
status: status,
r: tr.Raft,
errorc: tr.ErrorC,
stopc: make(chan struct{}),
}
}
func (s *snapshotSender) stop() { close(s.stopc) }
func (s *snapshotSender) send(merged snap.Message) {
m := merged.Message
body := createSnapBody(merged)
defer body.Close()
u := s.picker.pick()
req := createPostRequest(u, RaftSnapshotPrefix, body, "application/octet-stream", s.tr.URLs, s.from, s.cid)
plog.Infof("start to send database snapshot [index: %d, to %s]...", m.Snapshot.Metadata.Index, types.ID(m.To))
err := s.post(req)
defer merged.CloseWithError(err)
if err != nil {
plog.Warningf("database snapshot [index: %d, to: %s] failed to be sent out (%v)", m.Snapshot.Metadata.Index, types.ID(m.To), err)
// errMemberRemoved is a critical error since a removed member should
// always be stopped. So we use reportCriticalError to report it to errorc.
if err == errMemberRemoved {
reportCriticalError(err, s.errorc)
}
s.picker.unreachable(u)
s.status.deactivate(failureType{source: sendSnap, action: "post"}, err.Error())
s.r.ReportUnreachable(m.To)
// report SnapshotFailure to raft state machine. After raft state
// machine knows about it, it would pause a while and retry sending
// new snapshot message.
s.r.ReportSnapshot(m.To, raft.SnapshotFailure)
sentFailures.WithLabelValues(types.ID(m.To).String()).Inc()
return
}
s.status.activate()
s.r.ReportSnapshot(m.To, raft.SnapshotFinish)
plog.Infof("database snapshot [index: %d, to: %s] sent out successfully", m.Snapshot.Metadata.Index, types.ID(m.To))
sentBytes.WithLabelValues(types.ID(m.To).String()).Add(float64(merged.TotalSize))
}
// post posts the given request.
// It returns nil when request is sent out and processed successfully.
func (s *snapshotSender) post(req *http.Request) (err error) {
cancel := httputil.RequestCanceler(req)
type responseAndError struct {
resp *http.Response
body []byte
err error
}
result := make(chan responseAndError, 1)
go func() {
resp, err := s.tr.pipelineRt.RoundTrip(req)
if err != nil {
result <- responseAndError{resp, nil, err}
return
}
// close the response body when timeouts.
// prevents from reading the body forever when the other side dies right after
// successfully receives the request body.
time.AfterFunc(snapResponseReadTimeout, func() { httputil.GracefulClose(resp) })
body, err := ioutil.ReadAll(resp.Body)
result <- responseAndError{resp, body, err}
}()
select {
case <-s.stopc:
cancel()
return errStopped
case r := <-result:
if r.err != nil {
return r.err
}
return checkPostResponse(r.resp, r.body, req, s.to)
}
}
func createSnapBody(merged snap.Message) io.ReadCloser {
buf := new(bytes.Buffer)
enc := &messageEncoder{w: buf}
// encode raft message
if err := enc.encode(&merged.Message); err != nil {
plog.Panicf("encode message error (%v)", err)
}
return &pioutil.ReaderAndCloser{
Reader: io.MultiReader(buf, merged.ReadCloser),
Closer: merged.ReadCloser,
}
}

145
vendor/github.com/coreos/etcd/rafthttp/snapshot_test.go generated vendored Normal file
View file

@ -0,0 +1,145 @@
// 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 rafthttp
import (
"fmt"
"io"
"io/ioutil"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
"time"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/snap"
)
type strReaderCloser struct{ *strings.Reader }
func (s strReaderCloser) Close() error { return nil }
func TestSnapshotSend(t *testing.T) {
tests := []struct {
m raftpb.Message
rc io.ReadCloser
size int64
wsent bool
wfiles int
}{
// sent and receive with no errors
{
m: raftpb.Message{Type: raftpb.MsgSnap, To: 1},
rc: strReaderCloser{strings.NewReader("hello")},
size: 5,
wsent: true,
wfiles: 1,
},
// error when reading snapshot for send
{
m: raftpb.Message{Type: raftpb.MsgSnap, To: 1},
rc: &errReadCloser{fmt.Errorf("snapshot error")},
size: 1,
wsent: false,
wfiles: 0,
},
// sends less than the given snapshot length
{
m: raftpb.Message{Type: raftpb.MsgSnap, To: 1},
rc: strReaderCloser{strings.NewReader("hello")},
size: 10000,
wsent: false,
wfiles: 0,
},
// sends less than actual snapshot length
{
m: raftpb.Message{Type: raftpb.MsgSnap, To: 1},
rc: strReaderCloser{strings.NewReader("hello")},
size: 1,
wsent: false,
wfiles: 0,
},
}
for i, tt := range tests {
sent, files := testSnapshotSend(t, snap.NewMessage(tt.m, tt.rc, tt.size))
if tt.wsent != sent {
t.Errorf("#%d: snapshot expected %v, got %v", i, tt.wsent, sent)
}
if tt.wfiles != len(files) {
t.Fatalf("#%d: expected %d files, got %d files", i, tt.wfiles, len(files))
}
}
}
func testSnapshotSend(t *testing.T, sm *snap.Message) (bool, []os.FileInfo) {
d, err := ioutil.TempDir(os.TempDir(), "snapdir")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(d)
r := &fakeRaft{}
tr := &Transport{pipelineRt: &http.Transport{}, ClusterID: types.ID(1), Raft: r}
ch := make(chan struct{}, 1)
h := &syncHandler{newSnapshotHandler(tr, r, snap.New(d), types.ID(1)), ch}
srv := httptest.NewServer(h)
defer srv.Close()
picker := mustNewURLPicker(t, []string{srv.URL})
snapsend := newSnapshotSender(tr, picker, types.ID(1), newPeerStatus(types.ID(1)))
defer snapsend.stop()
snapsend.send(*sm)
sent := false
select {
case <-time.After(time.Second):
t.Fatalf("timed out sending snapshot")
case sent = <-sm.CloseNotify():
}
// wait for handler to finish accepting snapshot
<-ch
files, rerr := ioutil.ReadDir(d)
if rerr != nil {
t.Fatal(rerr)
}
return sent, files
}
type errReadCloser struct{ err error }
func (s *errReadCloser) Read(p []byte) (int, error) { return 0, s.err }
func (s *errReadCloser) Close() error { return s.err }
type syncHandler struct {
h http.Handler
ch chan<- struct{}
}
func (sh *syncHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
sh.h.ServeHTTP(w, r)
sh.ch <- struct{}{}
}

526
vendor/github.com/coreos/etcd/rafthttp/stream.go generated vendored Normal file
View file

@ -0,0 +1,526 @@
// Copyright 2015 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 rafthttp
import (
"fmt"
"io"
"io/ioutil"
"net"
"net/http"
"path"
"strings"
"sync"
"time"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/httputil"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/version"
"github.com/coreos/go-semver/semver"
)
const (
streamTypeMessage streamType = "message"
streamTypeMsgAppV2 streamType = "msgappv2"
streamBufSize = 4096
)
var (
errUnsupportedStreamType = fmt.Errorf("unsupported stream type")
// the key is in string format "major.minor.patch"
supportedStream = map[string][]streamType{
"2.0.0": {},
"2.1.0": {streamTypeMsgAppV2, streamTypeMessage},
"2.2.0": {streamTypeMsgAppV2, streamTypeMessage},
"2.3.0": {streamTypeMsgAppV2, streamTypeMessage},
"3.0.0": {streamTypeMsgAppV2, streamTypeMessage},
"3.1.0": {streamTypeMsgAppV2, streamTypeMessage},
}
)
type streamType string
func (t streamType) endpoint() string {
switch t {
case streamTypeMsgAppV2:
return path.Join(RaftStreamPrefix, "msgapp")
case streamTypeMessage:
return path.Join(RaftStreamPrefix, "message")
default:
plog.Panicf("unhandled stream type %v", t)
return ""
}
}
func (t streamType) String() string {
switch t {
case streamTypeMsgAppV2:
return "stream MsgApp v2"
case streamTypeMessage:
return "stream Message"
default:
return "unknown stream"
}
}
var (
// linkHeartbeatMessage is a special message used as heartbeat message in
// link layer. It never conflicts with messages from raft because raft
// doesn't send out messages without From and To fields.
linkHeartbeatMessage = raftpb.Message{Type: raftpb.MsgHeartbeat}
)
func isLinkHeartbeatMessage(m *raftpb.Message) bool {
return m.Type == raftpb.MsgHeartbeat && m.From == 0 && m.To == 0
}
type outgoingConn struct {
t streamType
io.Writer
http.Flusher
io.Closer
}
// streamWriter writes messages to the attached outgoingConn.
type streamWriter struct {
peerID types.ID
status *peerStatus
fs *stats.FollowerStats
r Raft
mu sync.Mutex // guard field working and closer
closer io.Closer
working bool
msgc chan raftpb.Message
connc chan *outgoingConn
stopc chan struct{}
done chan struct{}
}
// startStreamWriter creates a streamWrite and starts a long running go-routine that accepts
// messages and writes to the attached outgoing connection.
func startStreamWriter(id types.ID, status *peerStatus, fs *stats.FollowerStats, r Raft) *streamWriter {
w := &streamWriter{
peerID: id,
status: status,
fs: fs,
r: r,
msgc: make(chan raftpb.Message, streamBufSize),
connc: make(chan *outgoingConn),
stopc: make(chan struct{}),
done: make(chan struct{}),
}
go w.run()
return w
}
func (cw *streamWriter) run() {
var (
msgc chan raftpb.Message
heartbeatc <-chan time.Time
t streamType
enc encoder
flusher http.Flusher
batched int
)
tickc := time.Tick(ConnReadTimeout / 3)
unflushed := 0
plog.Infof("started streaming with peer %s (writer)", cw.peerID)
for {
select {
case <-heartbeatc:
err := enc.encode(&linkHeartbeatMessage)
unflushed += linkHeartbeatMessage.Size()
if err == nil {
flusher.Flush()
batched = 0
sentBytes.WithLabelValues(cw.peerID.String()).Add(float64(unflushed))
unflushed = 0
continue
}
cw.status.deactivate(failureType{source: t.String(), action: "heartbeat"}, err.Error())
sentFailures.WithLabelValues(cw.peerID.String()).Inc()
cw.close()
plog.Warningf("lost the TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
heartbeatc, msgc = nil, nil
case m := <-msgc:
err := enc.encode(&m)
if err == nil {
unflushed += m.Size()
if len(msgc) == 0 || batched > streamBufSize/2 {
flusher.Flush()
sentBytes.WithLabelValues(cw.peerID.String()).Add(float64(unflushed))
unflushed = 0
batched = 0
} else {
batched++
}
continue
}
cw.status.deactivate(failureType{source: t.String(), action: "write"}, err.Error())
cw.close()
plog.Warningf("lost the TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
heartbeatc, msgc = nil, nil
cw.r.ReportUnreachable(m.To)
sentFailures.WithLabelValues(cw.peerID.String()).Inc()
case conn := <-cw.connc:
cw.mu.Lock()
closed := cw.closeUnlocked()
t = conn.t
switch conn.t {
case streamTypeMsgAppV2:
enc = newMsgAppV2Encoder(conn.Writer, cw.fs)
case streamTypeMessage:
enc = &messageEncoder{w: conn.Writer}
default:
plog.Panicf("unhandled stream type %s", conn.t)
}
flusher = conn.Flusher
unflushed = 0
cw.status.activate()
cw.closer = conn.Closer
cw.working = true
cw.mu.Unlock()
if closed {
plog.Warningf("closed an existing TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
}
plog.Infof("established a TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
heartbeatc, msgc = tickc, cw.msgc
case <-cw.stopc:
if cw.close() {
plog.Infof("closed the TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
}
plog.Infof("stopped streaming with peer %s (writer)", cw.peerID)
close(cw.done)
return
}
}
}
func (cw *streamWriter) writec() (chan<- raftpb.Message, bool) {
cw.mu.Lock()
defer cw.mu.Unlock()
return cw.msgc, cw.working
}
func (cw *streamWriter) close() bool {
cw.mu.Lock()
defer cw.mu.Unlock()
return cw.closeUnlocked()
}
func (cw *streamWriter) closeUnlocked() bool {
if !cw.working {
return false
}
cw.closer.Close()
if len(cw.msgc) > 0 {
cw.r.ReportUnreachable(uint64(cw.peerID))
}
cw.msgc = make(chan raftpb.Message, streamBufSize)
cw.working = false
return true
}
func (cw *streamWriter) attach(conn *outgoingConn) bool {
select {
case cw.connc <- conn:
return true
case <-cw.done:
return false
}
}
func (cw *streamWriter) stop() {
close(cw.stopc)
<-cw.done
}
// streamReader is a long-running go-routine that dials to the remote stream
// endpoint and reads messages from the response body returned.
type streamReader struct {
peerID types.ID
typ streamType
tr *Transport
picker *urlPicker
status *peerStatus
recvc chan<- raftpb.Message
propc chan<- raftpb.Message
errorc chan<- error
mu sync.Mutex
paused bool
cancel func()
closer io.Closer
stopc chan struct{}
done chan struct{}
}
func (r *streamReader) start() {
r.stopc = make(chan struct{})
r.done = make(chan struct{})
if r.errorc == nil {
r.errorc = r.tr.ErrorC
}
go r.run()
}
func (cr *streamReader) run() {
t := cr.typ
plog.Infof("started streaming with peer %s (%s reader)", cr.peerID, t)
for {
rc, err := cr.dial(t)
if err != nil {
if err != errUnsupportedStreamType {
cr.status.deactivate(failureType{source: t.String(), action: "dial"}, err.Error())
}
} else {
cr.status.activate()
plog.Infof("established a TCP streaming connection with peer %s (%s reader)", cr.peerID, cr.typ)
err := cr.decodeLoop(rc, t)
plog.Warningf("lost the TCP streaming connection with peer %s (%s reader)", cr.peerID, cr.typ)
switch {
// all data is read out
case err == io.EOF:
// connection is closed by the remote
case isClosedConnectionError(err):
default:
cr.status.deactivate(failureType{source: t.String(), action: "read"}, err.Error())
}
}
select {
// Wait 100ms to create a new stream, so it doesn't bring too much
// overhead when retry.
case <-time.After(100 * time.Millisecond):
case <-cr.stopc:
plog.Infof("stopped streaming with peer %s (%s reader)", cr.peerID, t)
close(cr.done)
return
}
}
}
func (cr *streamReader) decodeLoop(rc io.ReadCloser, t streamType) error {
var dec decoder
cr.mu.Lock()
switch t {
case streamTypeMsgAppV2:
dec = newMsgAppV2Decoder(rc, cr.tr.ID, cr.peerID)
case streamTypeMessage:
dec = &messageDecoder{r: rc}
default:
plog.Panicf("unhandled stream type %s", t)
}
select {
case <-cr.stopc:
cr.mu.Unlock()
if err := rc.Close(); err != nil {
return err
}
return io.EOF
default:
cr.closer = rc
}
cr.mu.Unlock()
for {
m, err := dec.decode()
if err != nil {
cr.mu.Lock()
cr.close()
cr.mu.Unlock()
return err
}
receivedBytes.WithLabelValues(types.ID(m.From).String()).Add(float64(m.Size()))
cr.mu.Lock()
paused := cr.paused
cr.mu.Unlock()
if paused {
continue
}
if isLinkHeartbeatMessage(&m) {
// raft is not interested in link layer
// heartbeat message, so we should ignore
// it.
continue
}
recvc := cr.recvc
if m.Type == raftpb.MsgProp {
recvc = cr.propc
}
select {
case recvc <- m:
default:
if cr.status.isActive() {
plog.MergeWarningf("dropped internal raft message from %s since receiving buffer is full (overloaded network)", types.ID(m.From))
}
plog.Debugf("dropped %s from %s since receiving buffer is full", m.Type, types.ID(m.From))
recvFailures.WithLabelValues(types.ID(m.From).String()).Inc()
}
}
}
func (cr *streamReader) stop() {
close(cr.stopc)
cr.mu.Lock()
if cr.cancel != nil {
cr.cancel()
}
cr.close()
cr.mu.Unlock()
<-cr.done
}
func (cr *streamReader) dial(t streamType) (io.ReadCloser, error) {
u := cr.picker.pick()
uu := u
uu.Path = path.Join(t.endpoint(), cr.tr.ID.String())
req, err := http.NewRequest("GET", uu.String(), nil)
if err != nil {
cr.picker.unreachable(u)
return nil, fmt.Errorf("failed to make http request to %v (%v)", u, err)
}
req.Header.Set("X-Server-From", cr.tr.ID.String())
req.Header.Set("X-Server-Version", version.Version)
req.Header.Set("X-Min-Cluster-Version", version.MinClusterVersion)
req.Header.Set("X-Etcd-Cluster-ID", cr.tr.ClusterID.String())
req.Header.Set("X-Raft-To", cr.peerID.String())
setPeerURLsHeader(req, cr.tr.URLs)
cr.mu.Lock()
select {
case <-cr.stopc:
cr.mu.Unlock()
return nil, fmt.Errorf("stream reader is stopped")
default:
}
cr.cancel = httputil.RequestCanceler(req)
cr.mu.Unlock()
resp, err := cr.tr.streamRt.RoundTrip(req)
if err != nil {
cr.picker.unreachable(u)
return nil, err
}
rv := serverVersion(resp.Header)
lv := semver.Must(semver.NewVersion(version.Version))
if compareMajorMinorVersion(rv, lv) == -1 && !checkStreamSupport(rv, t) {
httputil.GracefulClose(resp)
cr.picker.unreachable(u)
return nil, errUnsupportedStreamType
}
switch resp.StatusCode {
case http.StatusGone:
httputil.GracefulClose(resp)
cr.picker.unreachable(u)
reportCriticalError(errMemberRemoved, cr.errorc)
return nil, errMemberRemoved
case http.StatusOK:
return resp.Body, nil
case http.StatusNotFound:
httputil.GracefulClose(resp)
cr.picker.unreachable(u)
return nil, fmt.Errorf("peer %s failed to find local node %s", cr.peerID, cr.tr.ID)
case http.StatusPreconditionFailed:
b, err := ioutil.ReadAll(resp.Body)
if err != nil {
cr.picker.unreachable(u)
return nil, err
}
httputil.GracefulClose(resp)
cr.picker.unreachable(u)
switch strings.TrimSuffix(string(b), "\n") {
case errIncompatibleVersion.Error():
plog.Errorf("request sent was ignored by peer %s (server version incompatible)", cr.peerID)
return nil, errIncompatibleVersion
case errClusterIDMismatch.Error():
plog.Errorf("request sent was ignored (cluster ID mismatch: peer[%s]=%s, local=%s)",
cr.peerID, resp.Header.Get("X-Etcd-Cluster-ID"), cr.tr.ClusterID)
return nil, errClusterIDMismatch
default:
return nil, fmt.Errorf("unhandled error %q when precondition failed", string(b))
}
default:
httputil.GracefulClose(resp)
cr.picker.unreachable(u)
return nil, fmt.Errorf("unhandled http status %d", resp.StatusCode)
}
}
func (cr *streamReader) close() {
if cr.closer != nil {
cr.closer.Close()
}
cr.closer = nil
}
func (cr *streamReader) pause() {
cr.mu.Lock()
defer cr.mu.Unlock()
cr.paused = true
}
func (cr *streamReader) resume() {
cr.mu.Lock()
defer cr.mu.Unlock()
cr.paused = false
}
func isClosedConnectionError(err error) bool {
operr, ok := err.(*net.OpError)
return ok && operr.Err.Error() == "use of closed network connection"
}
// checkStreamSupport checks whether the stream type is supported in the
// given version.
func checkStreamSupport(v *semver.Version, t streamType) bool {
nv := &semver.Version{Major: v.Major, Minor: v.Minor}
for _, s := range supportedStream[nv.String()] {
if s == t {
return true
}
}
return false
}

439
vendor/github.com/coreos/etcd/rafthttp/stream_test.go generated vendored Normal file
View file

@ -0,0 +1,439 @@
// Copyright 2015 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 rafthttp
import (
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"reflect"
"sync"
"testing"
"time"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/testutil"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/version"
"github.com/coreos/go-semver/semver"
)
// TestStreamWriterAttachOutgoingConn tests that outgoingConn can be attached
// to streamWriter. After that, streamWriter can use it to send messages
// continuously, and closes it when stopped.
func TestStreamWriterAttachOutgoingConn(t *testing.T) {
sw := startStreamWriter(types.ID(1), newPeerStatus(types.ID(1)), &stats.FollowerStats{}, &fakeRaft{})
// the expected initial state of streamWriter is not working
if _, ok := sw.writec(); ok {
t.Errorf("initial working status = %v, want false", ok)
}
// repeat tests to ensure streamWriter can use last attached connection
var wfc *fakeWriteFlushCloser
for i := 0; i < 3; i++ {
prevwfc := wfc
wfc = newFakeWriteFlushCloser(nil)
sw.attach(&outgoingConn{t: streamTypeMessage, Writer: wfc, Flusher: wfc, Closer: wfc})
// previous attached connection should be closed
if prevwfc != nil {
select {
case <-prevwfc.closed:
case <-time.After(time.Second):
t.Errorf("#%d: close of previous connection timed out", i)
}
}
// if prevwfc != nil, the new msgc is ready since prevwfc has closed
// if prevwfc == nil, the first connection may be pending, but the first
// msgc is already available since it's set on calling startStreamwriter
msgc, _ := sw.writec()
msgc <- raftpb.Message{}
select {
case <-wfc.writec:
case <-time.After(time.Second):
t.Errorf("#%d: failed to write to the underlying connection", i)
}
// write chan is still available
if _, ok := sw.writec(); !ok {
t.Errorf("#%d: working status = %v, want true", i, ok)
}
}
sw.stop()
// write chan is unavailable since the writer is stopped.
if _, ok := sw.writec(); ok {
t.Errorf("working status after stop = %v, want false", ok)
}
if !wfc.Closed() {
t.Errorf("failed to close the underlying connection")
}
}
// TestStreamWriterAttachBadOutgoingConn tests that streamWriter with bad
// outgoingConn will close the outgoingConn and fall back to non-working status.
func TestStreamWriterAttachBadOutgoingConn(t *testing.T) {
sw := startStreamWriter(types.ID(1), newPeerStatus(types.ID(1)), &stats.FollowerStats{}, &fakeRaft{})
defer sw.stop()
wfc := newFakeWriteFlushCloser(errors.New("blah"))
sw.attach(&outgoingConn{t: streamTypeMessage, Writer: wfc, Flusher: wfc, Closer: wfc})
sw.msgc <- raftpb.Message{}
select {
case <-wfc.closed:
case <-time.After(time.Second):
t.Errorf("failed to close the underlying connection in time")
}
// no longer working
if _, ok := sw.writec(); ok {
t.Errorf("working = %v, want false", ok)
}
}
func TestStreamReaderDialRequest(t *testing.T) {
for i, tt := range []streamType{streamTypeMessage, streamTypeMsgAppV2} {
tr := &roundTripperRecorder{rec: &testutil.RecorderBuffered{}}
sr := &streamReader{
peerID: types.ID(2),
tr: &Transport{streamRt: tr, ClusterID: types.ID(1), ID: types.ID(1)},
picker: mustNewURLPicker(t, []string{"http://localhost:2380"}),
}
sr.dial(tt)
act, err := tr.rec.Wait(1)
if err != nil {
t.Fatal(err)
}
req := act[0].Params[0].(*http.Request)
wurl := fmt.Sprintf("http://localhost:2380" + tt.endpoint() + "/1")
if req.URL.String() != wurl {
t.Errorf("#%d: url = %s, want %s", i, req.URL.String(), wurl)
}
if w := "GET"; req.Method != w {
t.Errorf("#%d: method = %s, want %s", i, req.Method, w)
}
if g := req.Header.Get("X-Etcd-Cluster-ID"); g != "1" {
t.Errorf("#%d: header X-Etcd-Cluster-ID = %s, want 1", i, g)
}
if g := req.Header.Get("X-Raft-To"); g != "2" {
t.Errorf("#%d: header X-Raft-To = %s, want 2", i, g)
}
}
}
// TestStreamReaderDialResult tests the result of the dial func call meets the
// HTTP response received.
func TestStreamReaderDialResult(t *testing.T) {
tests := []struct {
code int
err error
wok bool
whalt bool
}{
{0, errors.New("blah"), false, false},
{http.StatusOK, nil, true, false},
{http.StatusMethodNotAllowed, nil, false, false},
{http.StatusNotFound, nil, false, false},
{http.StatusPreconditionFailed, nil, false, false},
{http.StatusGone, nil, false, true},
}
for i, tt := range tests {
h := http.Header{}
h.Add("X-Server-Version", version.Version)
tr := &respRoundTripper{
code: tt.code,
header: h,
err: tt.err,
}
sr := &streamReader{
peerID: types.ID(2),
tr: &Transport{streamRt: tr, ClusterID: types.ID(1)},
picker: mustNewURLPicker(t, []string{"http://localhost:2380"}),
errorc: make(chan error, 1),
}
_, err := sr.dial(streamTypeMessage)
if ok := err == nil; ok != tt.wok {
t.Errorf("#%d: ok = %v, want %v", i, ok, tt.wok)
}
if halt := len(sr.errorc) > 0; halt != tt.whalt {
t.Errorf("#%d: halt = %v, want %v", i, halt, tt.whalt)
}
}
}
// TestStreamReaderStopOnDial tests a stream reader closes the connection on stop.
func TestStreamReaderStopOnDial(t *testing.T) {
defer testutil.AfterTest(t)
h := http.Header{}
h.Add("X-Server-Version", version.Version)
tr := &respWaitRoundTripper{rrt: &respRoundTripper{code: http.StatusOK, header: h}}
sr := &streamReader{
peerID: types.ID(2),
tr: &Transport{streamRt: tr, ClusterID: types.ID(1)},
picker: mustNewURLPicker(t, []string{"http://localhost:2380"}),
errorc: make(chan error, 1),
typ: streamTypeMessage,
status: newPeerStatus(types.ID(2)),
}
tr.onResp = func() {
// stop() waits for the run() goroutine to exit, but that exit
// needs a response from RoundTrip() first; use goroutine
go sr.stop()
// wait so that stop() is blocked on run() exiting
time.Sleep(10 * time.Millisecond)
// sr.run() completes dialing then begins decoding while stopped
}
sr.start()
select {
case <-sr.done:
case <-time.After(time.Second):
t.Fatal("streamReader did not stop in time")
}
}
type respWaitRoundTripper struct {
rrt *respRoundTripper
onResp func()
}
func (t *respWaitRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
resp, err := t.rrt.RoundTrip(req)
resp.Body = newWaitReadCloser()
t.onResp()
return resp, err
}
type waitReadCloser struct{ closec chan struct{} }
func newWaitReadCloser() *waitReadCloser { return &waitReadCloser{make(chan struct{})} }
func (wrc *waitReadCloser) Read(p []byte) (int, error) {
<-wrc.closec
return 0, io.EOF
}
func (wrc *waitReadCloser) Close() error {
close(wrc.closec)
return nil
}
// TestStreamReaderDialDetectUnsupport tests that dial func could find
// out that the stream type is not supported by the remote.
func TestStreamReaderDialDetectUnsupport(t *testing.T) {
for i, typ := range []streamType{streamTypeMsgAppV2, streamTypeMessage} {
// the response from etcd 2.0
tr := &respRoundTripper{
code: http.StatusNotFound,
header: http.Header{},
}
sr := &streamReader{
peerID: types.ID(2),
tr: &Transport{streamRt: tr, ClusterID: types.ID(1)},
picker: mustNewURLPicker(t, []string{"http://localhost:2380"}),
}
_, err := sr.dial(typ)
if err != errUnsupportedStreamType {
t.Errorf("#%d: error = %v, want %v", i, err, errUnsupportedStreamType)
}
}
}
// TestStream tests that streamReader and streamWriter can build stream to
// send messages between each other.
func TestStream(t *testing.T) {
recvc := make(chan raftpb.Message, streamBufSize)
propc := make(chan raftpb.Message, streamBufSize)
msgapp := raftpb.Message{
Type: raftpb.MsgApp,
From: 2,
To: 1,
Term: 1,
LogTerm: 1,
Index: 3,
Entries: []raftpb.Entry{{Term: 1, Index: 4}},
}
tests := []struct {
t streamType
m raftpb.Message
wc chan raftpb.Message
}{
{
streamTypeMessage,
raftpb.Message{Type: raftpb.MsgProp, To: 2},
propc,
},
{
streamTypeMessage,
msgapp,
recvc,
},
{
streamTypeMsgAppV2,
msgapp,
recvc,
},
}
for i, tt := range tests {
h := &fakeStreamHandler{t: tt.t}
srv := httptest.NewServer(h)
defer srv.Close()
sw := startStreamWriter(types.ID(1), newPeerStatus(types.ID(1)), &stats.FollowerStats{}, &fakeRaft{})
defer sw.stop()
h.sw = sw
picker := mustNewURLPicker(t, []string{srv.URL})
tr := &Transport{streamRt: &http.Transport{}, ClusterID: types.ID(1)}
sr := &streamReader{
peerID: types.ID(2),
typ: tt.t,
tr: tr,
picker: picker,
status: newPeerStatus(types.ID(2)),
recvc: recvc,
propc: propc,
}
sr.start()
// wait for stream to work
var writec chan<- raftpb.Message
for {
var ok bool
if writec, ok = sw.writec(); ok {
break
}
time.Sleep(time.Millisecond)
}
writec <- tt.m
var m raftpb.Message
select {
case m = <-tt.wc:
case <-time.After(time.Second):
t.Fatalf("#%d: failed to receive message from the channel", i)
}
if !reflect.DeepEqual(m, tt.m) {
t.Fatalf("#%d: message = %+v, want %+v", i, m, tt.m)
}
sr.stop()
}
}
func TestCheckStreamSupport(t *testing.T) {
tests := []struct {
v *semver.Version
t streamType
w bool
}{
// support
{
semver.Must(semver.NewVersion("2.1.0")),
streamTypeMsgAppV2,
true,
},
// ignore patch
{
semver.Must(semver.NewVersion("2.1.9")),
streamTypeMsgAppV2,
true,
},
// ignore prerelease
{
semver.Must(semver.NewVersion("2.1.0-alpha")),
streamTypeMsgAppV2,
true,
},
}
for i, tt := range tests {
if g := checkStreamSupport(tt.v, tt.t); g != tt.w {
t.Errorf("#%d: check = %v, want %v", i, g, tt.w)
}
}
}
type fakeWriteFlushCloser struct {
mu sync.Mutex
err error
written int
closed chan struct{}
writec chan struct{}
}
func newFakeWriteFlushCloser(err error) *fakeWriteFlushCloser {
return &fakeWriteFlushCloser{
err: err,
closed: make(chan struct{}),
writec: make(chan struct{}, 1),
}
}
func (wfc *fakeWriteFlushCloser) Write(p []byte) (n int, err error) {
wfc.mu.Lock()
defer wfc.mu.Unlock()
select {
case wfc.writec <- struct{}{}:
default:
}
wfc.written += len(p)
return len(p), wfc.err
}
func (wfc *fakeWriteFlushCloser) Flush() {}
func (wfc *fakeWriteFlushCloser) Close() error {
close(wfc.closed)
return wfc.err
}
func (wfc *fakeWriteFlushCloser) Written() int {
wfc.mu.Lock()
defer wfc.mu.Unlock()
return wfc.written
}
func (wfc *fakeWriteFlushCloser) Closed() bool {
select {
case <-wfc.closed:
return true
default:
return false
}
}
type fakeStreamHandler struct {
t streamType
sw *streamWriter
}
func (h *fakeStreamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Add("X-Server-Version", version.Version)
w.(http.Flusher).Flush()
c := newCloseNotifier()
h.sw.attach(&outgoingConn{
t: h.t,
Writer: w,
Flusher: w.(http.Flusher),
Closer: c,
})
<-c.closeNotify()
}

402
vendor/github.com/coreos/etcd/rafthttp/transport.go generated vendored Normal file
View file

@ -0,0 +1,402 @@
// Copyright 2015 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 rafthttp
import (
"net/http"
"sync"
"time"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/logutil"
"github.com/coreos/etcd/pkg/transport"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/snap"
"github.com/coreos/pkg/capnslog"
"github.com/xiang90/probing"
"golang.org/x/net/context"
)
var plog = logutil.NewMergeLogger(capnslog.NewPackageLogger("github.com/coreos/etcd", "rafthttp"))
type Raft interface {
Process(ctx context.Context, m raftpb.Message) error
IsIDRemoved(id uint64) bool
ReportUnreachable(id uint64)
ReportSnapshot(id uint64, status raft.SnapshotStatus)
}
type Transporter interface {
// Start starts the given Transporter.
// Start MUST be called before calling other functions in the interface.
Start() error
// Handler returns the HTTP handler of the transporter.
// A transporter HTTP handler handles the HTTP requests
// from remote peers.
// The handler MUST be used to handle RaftPrefix(/raft)
// endpoint.
Handler() http.Handler
// Send sends out the given messages to the remote peers.
// Each message has a To field, which is an id that maps
// to an existing peer in the transport.
// If the id cannot be found in the transport, the message
// will be ignored.
Send(m []raftpb.Message)
// SendSnapshot sends out the given snapshot message to a remote peer.
// The behavior of SendSnapshot is similar to Send.
SendSnapshot(m snap.Message)
// AddRemote adds a remote with given peer urls into the transport.
// A remote helps newly joined member to catch up the progress of cluster,
// and will not be used after that.
// It is the caller's responsibility to ensure the urls are all valid,
// or it panics.
AddRemote(id types.ID, urls []string)
// AddPeer adds a peer with given peer urls into the transport.
// It is the caller's responsibility to ensure the urls are all valid,
// or it panics.
// Peer urls are used to connect to the remote peer.
AddPeer(id types.ID, urls []string)
// RemovePeer removes the peer with given id.
RemovePeer(id types.ID)
// RemoveAllPeers removes all the existing peers in the transport.
RemoveAllPeers()
// UpdatePeer updates the peer urls of the peer with the given id.
// It is the caller's responsibility to ensure the urls are all valid,
// or it panics.
UpdatePeer(id types.ID, urls []string)
// ActiveSince returns the time that the connection with the peer
// of the given id becomes active.
// If the connection is active since peer was added, it returns the adding time.
// If the connection is currently inactive, it returns zero time.
ActiveSince(id types.ID) time.Time
// Stop closes the connections and stops the transporter.
Stop()
}
// Transport implements Transporter interface. It provides the functionality
// to send raft messages to peers, and receive raft messages from peers.
// User should call Handler method to get a handler to serve requests
// received from peerURLs.
// User needs to call Start before calling other functions, and call
// Stop when the Transport is no longer used.
type Transport struct {
DialTimeout time.Duration // maximum duration before timing out dial of the request
TLSInfo transport.TLSInfo // TLS information used when creating connection
ID types.ID // local member ID
URLs types.URLs // local peer URLs
ClusterID types.ID // raft cluster ID for request validation
Raft Raft // raft state machine, to which the Transport forwards received messages and reports status
Snapshotter *snap.Snapshotter
ServerStats *stats.ServerStats // used to record general transportation statistics
// used to record transportation statistics with followers when
// performing as leader in raft protocol
LeaderStats *stats.LeaderStats
// ErrorC is used to report detected critical errors, e.g.,
// the member has been permanently removed from the cluster
// When an error is received from ErrorC, user should stop raft state
// machine and thus stop the Transport.
ErrorC chan error
streamRt http.RoundTripper // roundTripper used by streams
pipelineRt http.RoundTripper // roundTripper used by pipelines
mu sync.RWMutex // protect the remote and peer map
remotes map[types.ID]*remote // remotes map that helps newly joined member to catch up
peers map[types.ID]Peer // peers map
prober probing.Prober
}
func (t *Transport) Start() error {
var err error
t.streamRt, err = newStreamRoundTripper(t.TLSInfo, t.DialTimeout)
if err != nil {
return err
}
t.pipelineRt, err = NewRoundTripper(t.TLSInfo, t.DialTimeout)
if err != nil {
return err
}
t.remotes = make(map[types.ID]*remote)
t.peers = make(map[types.ID]Peer)
t.prober = probing.NewProber(t.pipelineRt)
return nil
}
func (t *Transport) Handler() http.Handler {
pipelineHandler := newPipelineHandler(t, t.Raft, t.ClusterID)
streamHandler := newStreamHandler(t, t, t.Raft, t.ID, t.ClusterID)
snapHandler := newSnapshotHandler(t, t.Raft, t.Snapshotter, t.ClusterID)
mux := http.NewServeMux()
mux.Handle(RaftPrefix, pipelineHandler)
mux.Handle(RaftStreamPrefix+"/", streamHandler)
mux.Handle(RaftSnapshotPrefix, snapHandler)
mux.Handle(ProbingPrefix, probing.NewHandler())
return mux
}
func (t *Transport) Get(id types.ID) Peer {
t.mu.RLock()
defer t.mu.RUnlock()
return t.peers[id]
}
func (t *Transport) Send(msgs []raftpb.Message) {
for _, m := range msgs {
if m.To == 0 {
// ignore intentionally dropped message
continue
}
to := types.ID(m.To)
t.mu.RLock()
p, pok := t.peers[to]
g, rok := t.remotes[to]
t.mu.RUnlock()
if pok {
if m.Type == raftpb.MsgApp {
t.ServerStats.SendAppendReq(m.Size())
}
p.send(m)
continue
}
if rok {
g.send(m)
continue
}
plog.Debugf("ignored message %s (sent to unknown peer %s)", m.Type, to)
}
}
func (t *Transport) Stop() {
t.mu.Lock()
defer t.mu.Unlock()
for _, r := range t.remotes {
r.stop()
}
for _, p := range t.peers {
p.stop()
}
t.prober.RemoveAll()
if tr, ok := t.streamRt.(*http.Transport); ok {
tr.CloseIdleConnections()
}
if tr, ok := t.pipelineRt.(*http.Transport); ok {
tr.CloseIdleConnections()
}
t.peers = nil
t.remotes = nil
}
// CutPeer drops messages to the specified peer.
func (t *Transport) CutPeer(id types.ID) {
t.mu.RLock()
p, pok := t.peers[id]
g, gok := t.remotes[id]
t.mu.RUnlock()
if pok {
p.(Pausable).Pause()
}
if gok {
g.Pause()
}
}
// MendPeer recovers the message dropping behavior of the given peer.
func (t *Transport) MendPeer(id types.ID) {
t.mu.RLock()
p, pok := t.peers[id]
g, gok := t.remotes[id]
t.mu.RUnlock()
if pok {
p.(Pausable).Resume()
}
if gok {
g.Resume()
}
}
func (t *Transport) AddRemote(id types.ID, us []string) {
t.mu.Lock()
defer t.mu.Unlock()
if t.remotes == nil {
// there's no clean way to shutdown the golang http server
// (see: https://github.com/golang/go/issues/4674) before
// stopping the transport; ignore any new connections.
return
}
if _, ok := t.peers[id]; ok {
return
}
if _, ok := t.remotes[id]; ok {
return
}
urls, err := types.NewURLs(us)
if err != nil {
plog.Panicf("newURLs %+v should never fail: %+v", us, err)
}
t.remotes[id] = startRemote(t, urls, id)
}
func (t *Transport) AddPeer(id types.ID, us []string) {
t.mu.Lock()
defer t.mu.Unlock()
if t.peers == nil {
panic("transport stopped")
}
if _, ok := t.peers[id]; ok {
return
}
urls, err := types.NewURLs(us)
if err != nil {
plog.Panicf("newURLs %+v should never fail: %+v", us, err)
}
fs := t.LeaderStats.Follower(id.String())
t.peers[id] = startPeer(t, urls, id, fs)
addPeerToProber(t.prober, id.String(), us)
plog.Infof("added peer %s", id)
}
func (t *Transport) RemovePeer(id types.ID) {
t.mu.Lock()
defer t.mu.Unlock()
t.removePeer(id)
}
func (t *Transport) RemoveAllPeers() {
t.mu.Lock()
defer t.mu.Unlock()
for id := range t.peers {
t.removePeer(id)
}
}
// the caller of this function must have the peers mutex.
func (t *Transport) removePeer(id types.ID) {
if peer, ok := t.peers[id]; ok {
peer.stop()
} else {
plog.Panicf("unexpected removal of unknown peer '%d'", id)
}
delete(t.peers, id)
delete(t.LeaderStats.Followers, id.String())
t.prober.Remove(id.String())
plog.Infof("removed peer %s", id)
}
func (t *Transport) UpdatePeer(id types.ID, us []string) {
t.mu.Lock()
defer t.mu.Unlock()
// TODO: return error or just panic?
if _, ok := t.peers[id]; !ok {
return
}
urls, err := types.NewURLs(us)
if err != nil {
plog.Panicf("newURLs %+v should never fail: %+v", us, err)
}
t.peers[id].update(urls)
t.prober.Remove(id.String())
addPeerToProber(t.prober, id.String(), us)
plog.Infof("updated peer %s", id)
}
func (t *Transport) ActiveSince(id types.ID) time.Time {
t.mu.Lock()
defer t.mu.Unlock()
if p, ok := t.peers[id]; ok {
return p.activeSince()
}
return time.Time{}
}
func (t *Transport) SendSnapshot(m snap.Message) {
t.mu.Lock()
defer t.mu.Unlock()
p := t.peers[types.ID(m.To)]
if p == nil {
m.CloseWithError(errMemberNotFound)
return
}
p.sendSnap(m)
}
// Pausable is a testing interface for pausing transport traffic.
type Pausable interface {
Pause()
Resume()
}
func (t *Transport) Pause() {
for _, p := range t.peers {
p.(Pausable).Pause()
}
}
func (t *Transport) Resume() {
for _, p := range t.peers {
p.(Pausable).Resume()
}
}
type nopTransporter struct{}
func NewNopTransporter() Transporter {
return &nopTransporter{}
}
func (s *nopTransporter) Start() error { return nil }
func (s *nopTransporter) Handler() http.Handler { return nil }
func (s *nopTransporter) Send(m []raftpb.Message) {}
func (s *nopTransporter) SendSnapshot(m snap.Message) {}
func (s *nopTransporter) AddRemote(id types.ID, us []string) {}
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) ActiveSince(id types.ID) time.Time { return time.Time{} }
func (s *nopTransporter) Stop() {}
func (s *nopTransporter) Pause() {}
func (s *nopTransporter) Resume() {}
type snapTransporter struct {
nopTransporter
snapDoneC chan snap.Message
snapDir string
}
func NewSnapTransporter(snapDir string) (Transporter, <-chan snap.Message) {
ch := make(chan snap.Message, 1)
tr := &snapTransporter{snapDoneC: ch, snapDir: snapDir}
return tr, ch
}
func (s *snapTransporter) SendSnapshot(m snap.Message) {
ss := snap.New(s.snapDir)
ss.SaveDBFrom(m.ReadCloser, m.Snapshot.Metadata.Index+1)
m.CloseWithError(nil)
s.snapDoneC <- m
}

View file

@ -0,0 +1,114 @@
// Copyright 2015 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 rafthttp
import (
"net/http/httptest"
"sync"
"testing"
"time"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft"
"github.com/coreos/etcd/raft/raftpb"
"golang.org/x/net/context"
)
func BenchmarkSendingMsgApp(b *testing.B) {
// member 1
tr := &Transport{
ID: types.ID(1),
ClusterID: types.ID(1),
Raft: &fakeRaft{},
ServerStats: newServerStats(),
LeaderStats: stats.NewLeaderStats("1"),
}
tr.Start()
srv := httptest.NewServer(tr.Handler())
defer srv.Close()
// member 2
r := &countRaft{}
tr2 := &Transport{
ID: types.ID(2),
ClusterID: types.ID(1),
Raft: r,
ServerStats: newServerStats(),
LeaderStats: stats.NewLeaderStats("2"),
}
tr2.Start()
srv2 := httptest.NewServer(tr2.Handler())
defer srv2.Close()
tr.AddPeer(types.ID(2), []string{srv2.URL})
defer tr.Stop()
tr2.AddPeer(types.ID(1), []string{srv.URL})
defer tr2.Stop()
if !waitStreamWorking(tr.Get(types.ID(2)).(*peer)) {
b.Fatalf("stream from 1 to 2 is not in work as expected")
}
b.ReportAllocs()
b.SetBytes(64)
b.ResetTimer()
data := make([]byte, 64)
for i := 0; i < b.N; i++ {
tr.Send([]raftpb.Message{
{
Type: raftpb.MsgApp,
From: 1,
To: 2,
Index: uint64(i),
Entries: []raftpb.Entry{
{
Index: uint64(i + 1),
Data: data,
},
},
},
})
}
// wait until all messages are received by the target raft
for r.count() != b.N {
time.Sleep(time.Millisecond)
}
b.StopTimer()
}
type countRaft struct {
mu sync.Mutex
cnt int
}
func (r *countRaft) Process(ctx context.Context, m raftpb.Message) error {
r.mu.Lock()
defer r.mu.Unlock()
r.cnt++
return nil
}
func (r *countRaft) IsIDRemoved(id uint64) bool { return false }
func (r *countRaft) ReportUnreachable(id uint64) {}
func (r *countRaft) ReportSnapshot(id uint64, status raft.SnapshotStatus) {}
func (r *countRaft) count() int {
r.mu.Lock()
defer r.mu.Unlock()
return r.cnt
}

View file

@ -0,0 +1,185 @@
// Copyright 2015 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 rafthttp
import (
"net/http"
"reflect"
"testing"
"time"
"github.com/coreos/etcd/etcdserver/stats"
"github.com/coreos/etcd/pkg/testutil"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
"github.com/xiang90/probing"
)
// TestTransportSend tests that transport can send messages using correct
// underlying peer, and drop local or unknown-target messages.
func TestTransportSend(t *testing.T) {
ss := &stats.ServerStats{}
ss.Initialize()
peer1 := newFakePeer()
peer2 := newFakePeer()
tr := &Transport{
ServerStats: ss,
peers: map[types.ID]Peer{types.ID(1): peer1, types.ID(2): peer2},
}
wmsgsIgnored := []raftpb.Message{
// bad local message
{Type: raftpb.MsgBeat},
// bad remote message
{Type: raftpb.MsgProp, To: 3},
}
wmsgsTo1 := []raftpb.Message{
// good message
{Type: raftpb.MsgProp, To: 1},
{Type: raftpb.MsgApp, To: 1},
}
wmsgsTo2 := []raftpb.Message{
// good message
{Type: raftpb.MsgProp, To: 2},
{Type: raftpb.MsgApp, To: 2},
}
tr.Send(wmsgsIgnored)
tr.Send(wmsgsTo1)
tr.Send(wmsgsTo2)
if !reflect.DeepEqual(peer1.msgs, wmsgsTo1) {
t.Errorf("msgs to peer 1 = %+v, want %+v", peer1.msgs, wmsgsTo1)
}
if !reflect.DeepEqual(peer2.msgs, wmsgsTo2) {
t.Errorf("msgs to peer 2 = %+v, want %+v", peer2.msgs, wmsgsTo2)
}
}
func TestTransportCutMend(t *testing.T) {
ss := &stats.ServerStats{}
ss.Initialize()
peer1 := newFakePeer()
peer2 := newFakePeer()
tr := &Transport{
ServerStats: ss,
peers: map[types.ID]Peer{types.ID(1): peer1, types.ID(2): peer2},
}
tr.CutPeer(types.ID(1))
wmsgsTo := []raftpb.Message{
// good message
{Type: raftpb.MsgProp, To: 1},
{Type: raftpb.MsgApp, To: 1},
}
tr.Send(wmsgsTo)
if len(peer1.msgs) > 0 {
t.Fatalf("msgs expected to be ignored, got %+v", peer1.msgs)
}
tr.MendPeer(types.ID(1))
tr.Send(wmsgsTo)
if !reflect.DeepEqual(peer1.msgs, wmsgsTo) {
t.Errorf("msgs to peer 1 = %+v, want %+v", peer1.msgs, wmsgsTo)
}
}
func TestTransportAdd(t *testing.T) {
ls := stats.NewLeaderStats("")
tr := &Transport{
LeaderStats: ls,
streamRt: &roundTripperRecorder{},
peers: make(map[types.ID]Peer),
prober: probing.NewProber(nil),
}
tr.AddPeer(1, []string{"http://localhost:2380"})
if _, ok := ls.Followers["1"]; !ok {
t.Errorf("FollowerStats[1] is nil, want exists")
}
s, ok := tr.peers[types.ID(1)]
if !ok {
tr.Stop()
t.Fatalf("senders[1] is nil, want exists")
}
// duplicate AddPeer is ignored
tr.AddPeer(1, []string{"http://localhost:2380"})
ns := tr.peers[types.ID(1)]
if s != ns {
t.Errorf("sender = %v, want %v", ns, s)
}
tr.Stop()
}
func TestTransportRemove(t *testing.T) {
tr := &Transport{
LeaderStats: stats.NewLeaderStats(""),
streamRt: &roundTripperRecorder{},
peers: make(map[types.ID]Peer),
prober: probing.NewProber(nil),
}
tr.AddPeer(1, []string{"http://localhost:2380"})
tr.RemovePeer(types.ID(1))
defer tr.Stop()
if _, ok := tr.peers[types.ID(1)]; ok {
t.Fatalf("senders[1] exists, want removed")
}
}
func TestTransportUpdate(t *testing.T) {
peer := newFakePeer()
tr := &Transport{
peers: map[types.ID]Peer{types.ID(1): peer},
prober: probing.NewProber(nil),
}
u := "http://localhost:2380"
tr.UpdatePeer(types.ID(1), []string{u})
wurls := types.URLs(testutil.MustNewURLs(t, []string{"http://localhost:2380"}))
if !reflect.DeepEqual(peer.peerURLs, wurls) {
t.Errorf("urls = %+v, want %+v", peer.peerURLs, wurls)
}
}
func TestTransportErrorc(t *testing.T) {
errorc := make(chan error, 1)
tr := &Transport{
Raft: &fakeRaft{},
LeaderStats: stats.NewLeaderStats(""),
ErrorC: errorc,
streamRt: newRespRoundTripper(http.StatusForbidden, nil),
pipelineRt: newRespRoundTripper(http.StatusForbidden, nil),
peers: make(map[types.ID]Peer),
prober: probing.NewProber(nil),
}
tr.AddPeer(1, []string{"http://localhost:2380"})
defer tr.Stop()
select {
case <-errorc:
t.Fatalf("received unexpected from errorc")
case <-time.After(10 * time.Millisecond):
}
tr.peers[1].send(raftpb.Message{})
select {
case <-errorc:
case <-time.After(1 * time.Second):
t.Fatalf("cannot receive error from errorc")
}
}

57
vendor/github.com/coreos/etcd/rafthttp/urlpick.go generated vendored Normal file
View file

@ -0,0 +1,57 @@
// Copyright 2015 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 rafthttp
import (
"net/url"
"sync"
"github.com/coreos/etcd/pkg/types"
)
type urlPicker struct {
mu sync.Mutex // guards urls and picked
urls types.URLs
picked int
}
func newURLPicker(urls types.URLs) *urlPicker {
return &urlPicker{
urls: urls,
}
}
func (p *urlPicker) update(urls types.URLs) {
p.mu.Lock()
defer p.mu.Unlock()
p.urls = urls
p.picked = 0
}
func (p *urlPicker) pick() url.URL {
p.mu.Lock()
defer p.mu.Unlock()
return p.urls[p.picked]
}
// unreachable notices the picker that the given url is unreachable,
// and it should use other possible urls.
func (p *urlPicker) unreachable(u url.URL) {
p.mu.Lock()
defer p.mu.Unlock()
if u == p.urls[p.picked] {
p.picked = (p.picked + 1) % len(p.urls)
}
}

73
vendor/github.com/coreos/etcd/rafthttp/urlpick_test.go generated vendored Normal file
View file

@ -0,0 +1,73 @@
// Copyright 2015 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 rafthttp
import (
"net/url"
"testing"
"github.com/coreos/etcd/pkg/testutil"
)
// TestURLPickerPickTwice tests that pick returns a possible url,
// and always returns the same one.
func TestURLPickerPickTwice(t *testing.T) {
picker := mustNewURLPicker(t, []string{"http://127.0.0.1:2380", "http://127.0.0.1:7001"})
u := picker.pick()
urlmap := map[url.URL]bool{
{Scheme: "http", Host: "127.0.0.1:2380"}: true,
{Scheme: "http", Host: "127.0.0.1:7001"}: true,
}
if !urlmap[u] {
t.Errorf("url picked = %+v, want a possible url in %+v", u, urlmap)
}
// pick out the same url when calling pick again
uu := picker.pick()
if u != uu {
t.Errorf("url picked = %+v, want %+v", uu, u)
}
}
func TestURLPickerUpdate(t *testing.T) {
picker := mustNewURLPicker(t, []string{"http://127.0.0.1:2380", "http://127.0.0.1:7001"})
picker.update(testutil.MustNewURLs(t, []string{"http://localhost:2380", "http://localhost:7001"}))
u := picker.pick()
urlmap := map[url.URL]bool{
{Scheme: "http", Host: "localhost:2380"}: true,
{Scheme: "http", Host: "localhost:7001"}: true,
}
if !urlmap[u] {
t.Errorf("url picked = %+v, want a possible url in %+v", u, urlmap)
}
}
func TestURLPickerUnreachable(t *testing.T) {
picker := mustNewURLPicker(t, []string{"http://127.0.0.1:2380", "http://127.0.0.1:7001"})
u := picker.pick()
picker.unreachable(u)
uu := picker.pick()
if u == uu {
t.Errorf("url picked = %+v, want other possible urls", uu)
}
}
func mustNewURLPicker(t *testing.T, us []string) *urlPicker {
urls := testutil.MustNewURLs(t, us)
return newURLPicker(urls)
}

205
vendor/github.com/coreos/etcd/rafthttp/util.go generated vendored Normal file
View file

@ -0,0 +1,205 @@
// Copyright 2015 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 rafthttp
import (
"crypto/tls"
"encoding/binary"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
"time"
"github.com/coreos/etcd/pkg/transport"
"github.com/coreos/etcd/pkg/types"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/version"
"github.com/coreos/go-semver/semver"
)
var (
errMemberRemoved = fmt.Errorf("the member has been permanently removed from the cluster")
errMemberNotFound = fmt.Errorf("member not found")
)
// NewListener returns a listener for raft message transfer between peers.
// It uses timeout listener to identify broken streams promptly.
func NewListener(u url.URL, tlscfg *tls.Config) (net.Listener, error) {
return transport.NewTimeoutListener(u.Host, u.Scheme, tlscfg, ConnReadTimeout, ConnWriteTimeout)
}
// NewRoundTripper returns a roundTripper used to send requests
// to rafthttp listener of remote peers.
func NewRoundTripper(tlsInfo transport.TLSInfo, dialTimeout time.Duration) (http.RoundTripper, error) {
// It uses timeout transport to pair with remote timeout listeners.
// It sets no read/write timeout, because message in requests may
// take long time to write out before reading out the response.
return transport.NewTimeoutTransport(tlsInfo, dialTimeout, 0, 0)
}
// newStreamRoundTripper returns a roundTripper used to send stream requests
// to rafthttp listener of remote peers.
// Read/write timeout is set for stream roundTripper to promptly
// find out broken status, which minimizes the number of messages
// sent on broken connection.
func newStreamRoundTripper(tlsInfo transport.TLSInfo, dialTimeout time.Duration) (http.RoundTripper, error) {
return transport.NewTimeoutTransport(tlsInfo, dialTimeout, ConnReadTimeout, ConnWriteTimeout)
}
func writeEntryTo(w io.Writer, ent *raftpb.Entry) error {
size := ent.Size()
if err := binary.Write(w, binary.BigEndian, uint64(size)); err != nil {
return err
}
b, err := ent.Marshal()
if err != nil {
return err
}
_, err = w.Write(b)
return err
}
func readEntryFrom(r io.Reader, ent *raftpb.Entry) error {
var l uint64
if err := binary.Read(r, binary.BigEndian, &l); err != nil {
return err
}
buf := make([]byte, int(l))
if _, err := io.ReadFull(r, buf); err != nil {
return err
}
return ent.Unmarshal(buf)
}
// createPostRequest creates a HTTP POST request that sends raft message.
func createPostRequest(u url.URL, path string, body io.Reader, ct string, urls types.URLs, from, cid types.ID) *http.Request {
uu := u
uu.Path = path
req, err := http.NewRequest("POST", uu.String(), body)
if err != nil {
plog.Panicf("unexpected new request error (%v)", err)
}
req.Header.Set("Content-Type", ct)
req.Header.Set("X-Server-From", from.String())
req.Header.Set("X-Server-Version", version.Version)
req.Header.Set("X-Min-Cluster-Version", version.MinClusterVersion)
req.Header.Set("X-Etcd-Cluster-ID", cid.String())
setPeerURLsHeader(req, urls)
return req
}
// checkPostResponse checks the response of the HTTP POST request that sends
// raft message.
func checkPostResponse(resp *http.Response, body []byte, req *http.Request, to types.ID) error {
switch resp.StatusCode {
case http.StatusPreconditionFailed:
switch strings.TrimSuffix(string(body), "\n") {
case errIncompatibleVersion.Error():
plog.Errorf("request sent was ignored by peer %s (server version incompatible)", to)
return errIncompatibleVersion
case errClusterIDMismatch.Error():
plog.Errorf("request sent was ignored (cluster ID mismatch: remote[%s]=%s, local=%s)",
to, resp.Header.Get("X-Etcd-Cluster-ID"), req.Header.Get("X-Etcd-Cluster-ID"))
return errClusterIDMismatch
default:
return fmt.Errorf("unhandled error %q when precondition failed", string(body))
}
case http.StatusForbidden:
return errMemberRemoved
case http.StatusNoContent:
return nil
default:
return fmt.Errorf("unexpected http status %s while posting to %q", http.StatusText(resp.StatusCode), req.URL.String())
}
}
// reportCriticalError reports the given error through sending it into
// the given error channel.
// If the error channel is filled up when sending error, it drops the error
// because the fact that error has happened is reported, which is
// good enough.
func reportCriticalError(err error, errc chan<- error) {
select {
case errc <- err:
default:
}
}
// compareMajorMinorVersion returns an integer comparing two versions based on
// their major and minor version. The result will be 0 if a==b, -1 if a < b,
// and 1 if a > b.
func compareMajorMinorVersion(a, b *semver.Version) int {
na := &semver.Version{Major: a.Major, Minor: a.Minor}
nb := &semver.Version{Major: b.Major, Minor: b.Minor}
switch {
case na.LessThan(*nb):
return -1
case nb.LessThan(*na):
return 1
default:
return 0
}
}
// serverVersion returns the server version from the given header.
func serverVersion(h http.Header) *semver.Version {
verStr := h.Get("X-Server-Version")
// backward compatibility with etcd 2.0
if verStr == "" {
verStr = "2.0.0"
}
return semver.Must(semver.NewVersion(verStr))
}
// serverVersion returns the min cluster version from the given header.
func minClusterVersion(h http.Header) *semver.Version {
verStr := h.Get("X-Min-Cluster-Version")
// backward compatibility with etcd 2.0
if verStr == "" {
verStr = "2.0.0"
}
return semver.Must(semver.NewVersion(verStr))
}
// checkVersionCompability checks whether the given version is compatible
// with the local version.
func checkVersionCompability(name string, server, minCluster *semver.Version) error {
localServer := semver.Must(semver.NewVersion(version.Version))
localMinCluster := semver.Must(semver.NewVersion(version.MinClusterVersion))
if compareMajorMinorVersion(server, localMinCluster) == -1 {
return fmt.Errorf("remote version is too low: remote[%s]=%s, local=%s", name, server, localServer)
}
if compareMajorMinorVersion(minCluster, localServer) == 1 {
return fmt.Errorf("local version is too low: remote[%s]=%s, local=%s", name, server, localServer)
}
return nil
}
// setPeerURLsHeader reports local urls for peer discovery
func setPeerURLsHeader(req *http.Request, urls types.URLs) {
if urls == nil {
// often not set in unit tests
return
}
peerURLs := make([]string, urls.Len())
for i := range urls {
peerURLs[i] = urls[i].String()
}
req.Header.Set("X-PeerURLs", strings.Join(peerURLs, ","))
}

193
vendor/github.com/coreos/etcd/rafthttp/util_test.go generated vendored Normal file
View file

@ -0,0 +1,193 @@
// Copyright 2015 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 rafthttp
import (
"bytes"
"net/http"
"reflect"
"testing"
"github.com/coreos/etcd/raft/raftpb"
"github.com/coreos/etcd/version"
"github.com/coreos/go-semver/semver"
)
func TestEntry(t *testing.T) {
tests := []raftpb.Entry{
{},
{Term: 1, Index: 1},
{Term: 1, Index: 1, Data: []byte("some data")},
}
for i, tt := range tests {
b := &bytes.Buffer{}
if err := writeEntryTo(b, &tt); err != nil {
t.Errorf("#%d: unexpected write ents error: %v", i, err)
continue
}
var ent raftpb.Entry
if err := readEntryFrom(b, &ent); err != nil {
t.Errorf("#%d: unexpected read ents error: %v", i, err)
continue
}
if !reflect.DeepEqual(ent, tt) {
t.Errorf("#%d: ent = %+v, want %+v", i, ent, tt)
}
}
}
func TestCompareMajorMinorVersion(t *testing.T) {
tests := []struct {
va, vb *semver.Version
w int
}{
// equal to
{
semver.Must(semver.NewVersion("2.1.0")),
semver.Must(semver.NewVersion("2.1.0")),
0,
},
// smaller than
{
semver.Must(semver.NewVersion("2.0.0")),
semver.Must(semver.NewVersion("2.1.0")),
-1,
},
// bigger than
{
semver.Must(semver.NewVersion("2.2.0")),
semver.Must(semver.NewVersion("2.1.0")),
1,
},
// ignore patch
{
semver.Must(semver.NewVersion("2.1.1")),
semver.Must(semver.NewVersion("2.1.0")),
0,
},
// ignore prerelease
{
semver.Must(semver.NewVersion("2.1.0-alpha.0")),
semver.Must(semver.NewVersion("2.1.0")),
0,
},
}
for i, tt := range tests {
if g := compareMajorMinorVersion(tt.va, tt.vb); g != tt.w {
t.Errorf("#%d: compare = %d, want %d", i, g, tt.w)
}
}
}
func TestServerVersion(t *testing.T) {
tests := []struct {
h http.Header
wv *semver.Version
}{
// backward compatibility with etcd 2.0
{
http.Header{},
semver.Must(semver.NewVersion("2.0.0")),
},
{
http.Header{"X-Server-Version": []string{"2.1.0"}},
semver.Must(semver.NewVersion("2.1.0")),
},
{
http.Header{"X-Server-Version": []string{"2.1.0-alpha.0+git"}},
semver.Must(semver.NewVersion("2.1.0-alpha.0+git")),
},
}
for i, tt := range tests {
v := serverVersion(tt.h)
if v.String() != tt.wv.String() {
t.Errorf("#%d: version = %s, want %s", i, v, tt.wv)
}
}
}
func TestMinClusterVersion(t *testing.T) {
tests := []struct {
h http.Header
wv *semver.Version
}{
// backward compatibility with etcd 2.0
{
http.Header{},
semver.Must(semver.NewVersion("2.0.0")),
},
{
http.Header{"X-Min-Cluster-Version": []string{"2.1.0"}},
semver.Must(semver.NewVersion("2.1.0")),
},
{
http.Header{"X-Min-Cluster-Version": []string{"2.1.0-alpha.0+git"}},
semver.Must(semver.NewVersion("2.1.0-alpha.0+git")),
},
}
for i, tt := range tests {
v := minClusterVersion(tt.h)
if v.String() != tt.wv.String() {
t.Errorf("#%d: version = %s, want %s", i, v, tt.wv)
}
}
}
func TestCheckVersionCompatibility(t *testing.T) {
ls := semver.Must(semver.NewVersion(version.Version))
lmc := semver.Must(semver.NewVersion(version.MinClusterVersion))
tests := []struct {
server *semver.Version
minCluster *semver.Version
wok bool
}{
// the same version as local
{
ls,
lmc,
true,
},
// one version lower
{
lmc,
&semver.Version{},
true,
},
// one version higher
{
&semver.Version{Major: ls.Major + 1},
ls,
true,
},
// too low version
{
&semver.Version{Major: lmc.Major - 1},
&semver.Version{},
false,
},
// too high version
{
&semver.Version{Major: ls.Major + 1, Minor: 1},
&semver.Version{Major: ls.Major + 1},
false,
},
}
for i, tt := range tests {
err := checkVersionCompability("", tt.server, tt.minCluster)
if ok := err == nil; ok != tt.wok {
t.Errorf("#%d: ok = %v, want %v", i, ok, tt.wok)
}
}
}