Level sets dependency graph to consume etcd 3.1.5

vendor/github.com/coreos/etcd/rafthttp/coder.go

@@ -18,7 +18,7 @@ import "github.com/coreos/etcd/raft/raftpb"
 
 type encoder interface {
 	// encode encodes the given message to an output stream.
-	encode(m raftpb.Message) error
+	encode(m *raftpb.Message) error
 }
 
 type decoder interface {

vendor/github.com/coreos/etcd/rafthttp/http.go

@@ -104,6 +104,7 @@ func (h *pipelineHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	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
 	}
 
@@ -111,6 +112,7 @@ func (h *pipelineHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	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
 	}
 
@@ -123,6 +125,9 @@ func (h *pipelineHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		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
 	}
@@ -183,6 +188,7 @@ func (h *snapshotHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		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
 	}
 

vendor/github.com/coreos/etcd/rafthttp/metrics.go

@@ -16,7 +16,6 @@ package rafthttp
 
 import "github.com/prometheus/client_golang/prometheus"
 
-// TODO: record write/recv failures.
 var (
 	sentBytes = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace: "etcd",
@@ -36,6 +35,24 @@ var (
 		[]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",
@@ -50,5 +67,7 @@ var (
 func init() {
 	prometheus.MustRegister(sentBytes)
 	prometheus.MustRegister(receivedBytes)
+	prometheus.MustRegister(sentFailures)
+	prometheus.MustRegister(recvFailures)
 	prometheus.MustRegister(rtts)
 }

vendor/github.com/coreos/etcd/rafthttp/msg_codec.go

@@ -16,6 +16,7 @@ package rafthttp
 
 import (
 	"encoding/binary"
+	"errors"
 	"io"
 
 	"github.com/coreos/etcd/pkg/pbutil"
@@ -28,11 +29,11 @@ type messageEncoder struct {
 	w io.Writer
 }
 
-func (enc *messageEncoder) encode(m raftpb.Message) error {
+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))
+	_, err := enc.w.Write(pbutil.MustMarshal(m))
 	return err
 }
 
@@ -41,12 +42,20 @@ 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

vendor/github.com/coreos/etcd/rafthttp/msgappv2_codec.go

@@ -82,7 +82,7 @@ func newMsgAppV2Encoder(w io.Writer, fs *stats.FollowerStats) *msgAppV2Encoder {
 	}
 }
 
-func (enc *msgAppV2Encoder) encode(m raftpb.Message) error {
+func (enc *msgAppV2Encoder) encode(m *raftpb.Message) error {
 	start := time.Now()
 	switch {
 	case isLinkHeartbeatMessage(m):
@@ -135,7 +135,7 @@ func (enc *msgAppV2Encoder) encode(m raftpb.Message) error {
 			return err
 		}
 		// write message
-		if _, err := enc.w.Write(pbutil.MustMarshal(&m)); err != nil {
+		if _, err := enc.w.Write(pbutil.MustMarshal(m)); err != nil {
 			return err
 		}
 

vendor/github.com/coreos/etcd/rafthttp/peer.go

@@ -106,7 +106,6 @@ type peer struct {
 	msgAppV2Reader *streamReader
 	msgAppReader   *streamReader
 
-	sendc chan raftpb.Message
 	recvc chan raftpb.Message
 	propc chan raftpb.Message
 
@@ -145,7 +144,6 @@ func startPeer(transport *Transport, urls types.URLs, peerID types.ID, fs *stats
 		writer:         startStreamWriter(peerID, status, fs, r),
 		pipeline:       pipeline,
 		snapSender:     newSnapshotSender(transport, picker, peerID, status),
-		sendc:          make(chan raftpb.Message),
 		recvc:          make(chan raftpb.Message, recvBufSize),
 		propc:          make(chan raftpb.Message, maxPendingProposals),
 		stopc:          make(chan struct{}),

vendor/github.com/coreos/etcd/rafthttp/pipeline.go

@@ -93,6 +93,7 @@ func (p *pipeline) handle() {
 				if isMsgSnap(m) {
 					p.raft.ReportSnapshot(m.To, raft.SnapshotFailure)
 				}
+				sentFailures.WithLabelValues(types.ID(m.To).String()).Inc()
 				continue
 			}
 
@@ -117,7 +118,7 @@ func (p *pipeline) post(data []byte) (err error) {
 	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(p.tr.pipelineRt, req)
+	cancel := httputil.RequestCanceler(req)
 	go func() {
 		select {
 		case <-done:

vendor/github.com/coreos/etcd/rafthttp/probing_status.go

@@ -25,6 +25,7 @@ var (
 	// 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) {
@@ -44,11 +45,16 @@ func addPeerToProber(p probing.Prober, id string, us []string) {
 }
 
 func monitorProbingStatus(s probing.Status, id string) {
+	// set the first interval short to log error early.
+	interval := statusErrorInterval
 	for {
 		select {
-		case <-time.After(statusMonitoringInterval):
+		case <-time.After(interval):
 			if !s.Health() {
-				plog.Warningf("health check for peer %s failed", id)
+				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)

vendor/github.com/coreos/etcd/rafthttp/remote.go

@@ -59,3 +59,11 @@ func (g *remote) send(m raftpb.Message) {
 func (g *remote) stop() {
 	g.pipeline.stop()
 }
+
+func (g *remote) Pause() {
+	g.stop()
+}
+
+func (g *remote) Resume() {
+	g.pipeline.start()
+}

vendor/github.com/coreos/etcd/rafthttp/snapshot_sender.go

@@ -91,6 +91,7 @@ func (s *snapshotSender) send(merged snap.Message) {
 		// 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()
@@ -103,7 +104,7 @@ func (s *snapshotSender) send(merged snap.Message) {
 // 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(s.tr.pipelineRt, req)
+	cancel := httputil.RequestCanceler(req)
 
 	type responseAndError struct {
 		resp *http.Response
@@ -143,7 +144,7 @@ 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 {
+	if err := enc.encode(&merged.Message); err != nil {
 		plog.Panicf("encode message error (%v)", err)
 	}
 

vendor/github.com/coreos/etcd/rafthttp/stream.go

@@ -50,6 +50,7 @@ var (
 		"2.2.0": {streamTypeMsgAppV2, streamTypeMessage},
 		"2.3.0": {streamTypeMsgAppV2, streamTypeMessage},
 		"3.0.0": {streamTypeMsgAppV2, streamTypeMessage},
+		"3.1.0": {streamTypeMsgAppV2, streamTypeMessage},
 	}
 )
 
@@ -85,7 +86,7 @@ var (
 	linkHeartbeatMessage = raftpb.Message{Type: raftpb.MsgHeartbeat}
 )
 
-func isLinkHeartbeatMessage(m raftpb.Message) bool {
+func isLinkHeartbeatMessage(m *raftpb.Message) bool {
 	return m.Type == raftpb.MsgHeartbeat && m.From == 0 && m.To == 0
 }
 
@@ -147,7 +148,7 @@ func (cw *streamWriter) run() {
 	for {
 		select {
 		case <-heartbeatc:
-			err := enc.encode(linkHeartbeatMessage)
+			err := enc.encode(&linkHeartbeatMessage)
 			unflushed += linkHeartbeatMessage.Size()
 			if err == nil {
 				flusher.Flush()
@@ -159,12 +160,13 @@ func (cw *streamWriter) run() {
 
 			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)
+			err := enc.encode(&m)
 			if err == nil {
 				unflushed += m.Size()
 
@@ -185,11 +187,11 @@ func (cw *streamWriter) run() {
 			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:
-			if cw.close() {
-				plog.Warningf("closed an existing TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
-			}
+			cw.mu.Lock()
+			closed := cw.closeUnlocked()
 			t = conn.t
 			switch conn.t {
 			case streamTypeMsgAppV2:
@@ -201,19 +203,22 @@ func (cw *streamWriter) run() {
 			}
 			flusher = conn.Flusher
 			unflushed = 0
-			cw.mu.Lock()
 			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)
 			}
-			close(cw.done)
 			plog.Infof("stopped streaming with peer %s (writer)", cw.peerID)
+			close(cw.done)
 			return
 		}
 	}
@@ -228,6 +233,10 @@ func (cw *streamWriter) writec() (chan<- raftpb.Message, bool) {
 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
 	}
@@ -315,8 +324,8 @@ func (cr *streamReader) run() {
 		// overhead when retry.
 		case <-time.After(100 * time.Millisecond):
 		case <-cr.stopc:
-			close(cr.done)
 			plog.Infof("stopped streaming with peer %s (%s reader)", cr.peerID, t)
+			close(cr.done)
 			return
 		}
 	}
@@ -364,7 +373,7 @@ func (cr *streamReader) decodeLoop(rc io.ReadCloser, t streamType) error {
 			continue
 		}
 
-		if isLinkHeartbeatMessage(m) {
+		if isLinkHeartbeatMessage(&m) {
 			// raft is not interested in link layer
 			// heartbeat message, so we should ignore
 			// it.
@@ -383,6 +392,7 @@ func (cr *streamReader) decodeLoop(rc io.ReadCloser, t streamType) error {
 				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()
 		}
 	}
 }
@@ -423,7 +433,7 @@ func (cr *streamReader) dial(t streamType) (io.ReadCloser, error) {
 		return nil, fmt.Errorf("stream reader is stopped")
 	default:
 	}
-	cr.cancel = httputil.RequestCanceler(cr.tr.streamRt, req)
+	cr.cancel = httputil.RequestCanceler(req)
 	cr.mu.Unlock()
 
 	resp, err := cr.tr.streamRt.RoundTrip(req)
@@ -444,18 +454,14 @@ func (cr *streamReader) dial(t streamType) (io.ReadCloser, error) {
 	case http.StatusGone:
 		httputil.GracefulClose(resp)
 		cr.picker.unreachable(u)
-		err := fmt.Errorf("the member has been permanently removed from the cluster")
-		select {
-		case cr.errorc <- err:
-		default:
-		}
-		return nil, err
+		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 faild to fine local node %s", cr.peerID, cr.tr.ID)
+		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 {

vendor/github.com/coreos/etcd/rafthttp/transport.go

@@ -206,6 +206,36 @@ func (t *Transport) Stop() {
 	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()