Adds a direct TCP fallback ping piplined with the UDP indirect pings.

Author: James Phillips

net.go

@@ -196,17 +196,19 @@ func (m *Memberlist) handleConn(conn *net.TCPConn) {
 	defer conn.Close()
 	metrics.IncrCounter([]string{"memberlist", "tcp", "accept"}, 1)
 
+	conn.SetDeadline(time.Now().Add(m.config.TCPTimeout))
 	msgType, bufConn, dec, err := m.readTCP(conn)
 	if err != nil {
 		m.logger.Printf("[ERR] memberlist: failed to receive: %s", err)
 		return
 	}
 
-	if msgType == userMsg {
+	switch msgType {
+	case userMsg:
 		if err := m.readUserMsg(bufConn, dec); err != nil {
 			m.logger.Printf("[ERR] memberlist: Failed to receive user message: %s", err)
 		}
-	} else if msgType == pushPullMsg {
+	case pushPullMsg:
 		join, remoteNodes, userState, err := m.readRemoteState(bufConn, dec)
 		if err != nil {
 			m.logger.Printf("[ERR] memberlist: Failed to read remote state: %s", err)
@@ -221,7 +223,31 @@ func (m *Memberlist) handleConn(conn *net.TCPConn) {
 		if err := m.mergeRemoteState(join, remoteNodes, userState); err != nil {
 			return
 		}
-	} else {
+	case pingMsg:
+		var p ping
+		if err := dec.Decode(&p); err != nil {
+			m.logger.Printf("[ERR] memberlist: Failed to decode TCP ping: %s", err)
+			return
+		}
+
+		if p.Node != "" && p.Node != m.config.Name {
+			m.logger.Printf("[WARN] memberlist: Got ping for unexpected node %s", p.Node)
+			return
+		}
+
+		ack := ackResp{p.SeqNo, nil}
+		out, err := encode(ackRespMsg, &ack)
+		if err != nil {
+			m.logger.Printf("[ERR] memeberlist: Failed to encode TCP ack: %s", err)
+			return
+		}
+
+		err = m.rawSendMsgTCP(conn, out.Bytes())
+		if err != nil {
+			m.logger.Printf("[ERR] memberlist: Failed to send TCP ack: %s", err)
+			return
+		}
+	default:
 		m.logger.Printf("[ERR] memberlist: Received invalid msgType (%d)", msgType)
 	}
 }
@@ -654,6 +680,7 @@ func (m *Memberlist) sendAndReceiveState(addr []byte, port uint16, join bool) ([
 		return nil, nil, err
 	}
 
+	conn.SetDeadline(time.Now().Add(m.config.TCPTimeout))
 	msgType, bufConn, dec, err := m.readTCP(conn)
 	if err != nil {
 		return nil, nil, err
@@ -789,9 +816,6 @@ func (m *Memberlist) decryptRemoteState(bufConn io.Reader) ([]byte, error) {
 // readTCP is used to read the start of a TCP stream.
 // it decrypts and decompresses the stream if necessary
 func (m *Memberlist) readTCP(conn net.Conn) (messageType, io.Reader, *codec.Decoder, error) {
-	// Setup a deadline
-	conn.SetDeadline(time.Now().Add(m.config.TCPTimeout))
-
 	// Created a buffered reader
 	var bufConn io.Reader = bufio.NewReader(conn)
 
@@ -934,7 +958,7 @@ func (m *Memberlist) mergeRemoteState(join bool, remoteNodes []pushNodeState, us
 	return nil
 }
 
-// readUserMsg is used to decode a UserMsg from a TCP stream
+// readUserMsg is used to decode a userMsg from a TCP stream
 func (m *Memberlist) readUserMsg(bufConn io.Reader, dec *codec.Decoder) error {
 	// Read the user message header
 	var header userMsgHeader
@@ -964,3 +988,37 @@ func (m *Memberlist) readUserMsg(bufConn io.Reader, dec *codec.Decoder) error {
 
 	return nil
 }
+
+// sendPingAndWaitForAck uses the given TCP connection, sends a ping, and waits
+// for an ack. All of this is done in blocking fashion, and if the exchange is
+// good, a nil error will be returned to the caller.
+func (m *Memberlist) sendPingAndWaitForAck(conn net.Conn, ping ping) error {
+	out, err := encode(pingMsg, &ping)
+	if err != nil {
+		return err
+	}
+
+	if err = m.rawSendMsgTCP(conn, out.Bytes()); err != nil {
+		return err
+	}
+
+	msgType, _, dec, err := m.readTCP(conn)
+	if err != nil {
+		return err
+	}
+
+	if msgType != ackRespMsg {
+		return fmt.Errorf("unexpected msgType (%d) from TCP ping", msgType)
+	}
+
+	var ack ackResp
+	if err := dec.Decode(&ack); err != nil {
+		return err
+	}
+
+	if ack.SeqNo != ping.SeqNo {
+		return fmt.Errorf("sequence number from ack (%d) doesn't match ping (%d)", ack.SeqNo, ping.SeqNo)
+	}
+
+	return nil
+}

ping_delegate.go

@@ -4,7 +4,8 @@ import "time"
 
 // PingDelegate is used to notify an observer how long it took for a ping message to
 // complete a round trip.  It can also be used for writing arbitrary byte slices
-// into ack messages.
+// into ack messages. Note that in order to be meaningful for RTT estimates, this
+// delegate does not apply to indirect pings, nor fallback pings sent over TCP.
 type PingDelegate interface {
 	// AckPayload is invoked when an ack is being sent; the returned bytes will be appended to the ack
 	AckPayload() []byte

state.go

@@ -216,6 +216,7 @@ func (m *Memberlist) probeNode(node *nodeState) {
 	// Setup an ack handler
 	ackCh := make(chan ackMessage, m.config.IndirectChecks+1)
 	sent := time.Now()
+	deadline := sent.Add(m.config.ProbeInterval)
 	m.setAckChannel(ping.SeqNo, ackCh, m.config.ProbeInterval)
 
 	// Send the ping message
@@ -257,6 +258,37 @@ func (m *Memberlist) probeNode(node *nodeState) {
 		}
 	}
 
+	// Also make an attempt to contact the node directly over TCP. This
+	// helps prevent confused clients who get isolated from UDP traffic
+	// but can still speak TCP (which also means they can possibly report
+	// misinformation to other nodes via anti-entropy), avoiding flapping in
+	// the cluster.
+	fallbackCh := make(chan bool)
+	if node.PMax >= 3 {
+		destAddr := &net.TCPAddr{IP: node.Addr, Port: int(node.Port)}
+		dialer := net.Dialer{Timeout: deadline.Sub(time.Now())}
+		conn, err := dialer.Dial("tcp", destAddr.String())
+		if err != nil {
+			// This will fail if the node is actually dead, so we
+			// shouldn't spam the logs with it.
+			close(fallbackCh)
+		} else {
+			go func() {
+				defer close(fallbackCh)
+				defer conn.Close()
+
+				conn.SetDeadline(deadline)
+				if err := m.sendPingAndWaitForAck(conn, ping); err != nil {
+					m.logger.Printf("[ERR] memberlist: Failed TCP fallback ping: %s", err)
+				} else {
+					fallbackCh <- true
+				}
+			}()
+		}
+	} else {
+		close(fallbackCh)
+	}
+
 	// Wait for the acks or timeout
 	select {
 	case v := <-ackCh:
@@ -265,6 +297,15 @@ func (m *Memberlist) probeNode(node *nodeState) {
 		}
 	}
 
+	// Finally, poll the fallback channel - anything in there means that the
+	// TCP fallback ping was successful. The timeouts are set such that the
+	// channel will have something or be closed without having to wait any
+	// additional time here.
+	for _ = range fallbackCh {
+		m.logger.Printf("memberlist: Was able to reach %s via TCP but not UDP, network may be misconfigured and not allowing bidirectional UDP", node.Name)
+		return
+	}
+
 	// No acks received from target, suspect
 	m.logger.Printf("[INFO] memberlist: Suspect %s has failed, no acks received", node.Name)
 	s := suspect{Incarnation: node.Incarnation, Node: node.Name, From: m.config.Name}

state_test.go

@@ -105,17 +105,97 @@ func TestMemberList_ProbeNode_Suspect(t *testing.T) {
 	}
 }
 
-func TestMemberList_ProbeNode(t *testing.T) {
+func TestMemberList_ProbeNode_FallbackTCP(t *testing.T) {
 	addr1 := getBindAddr()
 	addr2 := getBindAddr()
+	addr3 := getBindAddr()
+	addr4 := getBindAddr()
 	ip1 := []byte(addr1)
 	ip2 := []byte(addr2)
+	ip3 := []byte(addr3)
+	ip4 := []byte(addr4)
 
 	m1 := HostMemberlist(addr1.String(), t, func(c *Config) {
 		c.ProbeTimeout = time.Millisecond
 		c.ProbeInterval = 10 * time.Millisecond
 	})
 	m2 := HostMemberlist(addr2.String(), t, nil)
+	m3 := HostMemberlist(addr3.String(), t, nil)
+	m4 := HostMemberlist(addr4.String(), t, nil)
+
+	a1 := alive{Node: addr1.String(), Addr: ip1, Port: 7946, Incarnation: 1}
+	m1.aliveNode(&a1, nil, true)
+	a2 := alive{Node: addr2.String(), Addr: ip2, Port: 7946, Incarnation: 1}
+	m1.aliveNode(&a2, nil, false)
+	a3 := alive{Node: addr3.String(), Addr: ip3, Port: 7946, Incarnation: 1}
+	m1.aliveNode(&a3, nil, false)
+
+	// Make sure m4 is configured with the same protocol version as m1 so
+	// the TCP fallback behavior is enabled.
+	a4 := alive{
+		Node: addr4.String(),
+		Addr: ip4,
+		Port: 7946,
+		Incarnation: 1,
+		Vsn: []uint8{
+			ProtocolVersionMin, ProtocolVersionMax,
+			m1.config.ProtocolVersion, m1.config.DelegateProtocolMin,
+			m1.config.DelegateProtocolMax, m1.config.DelegateProtocolVersion,
+		},
+	}
+	m1.aliveNode(&a4, nil, false)
+
+	// Isolate m4 from all inbound UDP traffic to force the TCP fallback
+	// path to get executed.
+	if err := m4.udpListener.Close(); err != nil {
+		t.Fatalf("could not close UDP listener")
+	}
+	n := m1.nodeMap[addr4.String()]
+	m1.probeNode(n)
+	if n.State != stateAlive {
+		t.Fatalf("expect node to be alive")
+	}
+
+	// Confirm the peers attempted an indirect probe.
+	time.Sleep(10 * time.Millisecond)
+	if m2.sequenceNum != 1 {
+		t.Fatalf("bad seqno %v", m2.sequenceNum)
+	}
+	if m3.sequenceNum != 1 {
+		t.Fatalf("bad seqno %v", m3.sequenceNum)
+	}
+
+	// Now shutdown all inbound TCP traffic to make sure the TCP fallback
+	// path properly fails when the node is really unreachable.
+	if err := m4.tcpListener.Close(); err != nil {
+		t.Fatalf("could not close TCP listener")
+	}
+	m1.probeNode(n)
+	if n.State != stateSuspect {
+		t.Fatalf("expect node to be suspect")
+	}
+
+	// Confirm the peers attempted an indirect probe.
+	time.Sleep(10 * time.Millisecond)
+	if m2.sequenceNum != 2 {
+		t.Fatalf("bad seqno %v", m2.sequenceNum)
+	}
+	if m3.sequenceNum != 2 {
+		t.Fatalf("bad seqno %v", m3.sequenceNum)
+	}
+}
+
+func TestMemberList_ProbeNode(t *testing.T) {
+	addr1 := getBindAddr()
+	addr2 := getBindAddr()
+	ip1 := []byte(addr1)
+	ip2 := []byte(addr2)
+
+	m1 := HostMemberlist(addr1.String(), t, func(c *Config) {
+		c.ProbeTimeout = time.Millisecond
+		c.ProbeInterval = 10 * time.Millisecond
+	})
+	_ = HostMemberlist(addr2.String(), t, nil)
 
 	a1 := alive{Node: addr1.String(), Addr: ip1, Port: 7946, Incarnation: 1}
 	m1.aliveNode(&a1, nil, true)
@@ -125,14 +205,14 @@ func TestMemberList_ProbeNode(t *testing.T) {
 	n := m1.nodeMap[addr2.String()]
 	m1.probeNode(n)
 
-	// Should be marked suspect
+	// Should be marked alive
 	if n.State != stateAlive {
 		t.Fatalf("Expect node to be alive")
 	}
 
 	// Should increment seqno
 	if m1.sequenceNum != 1 {
-		t.Fatalf("bad seqno %v", m2.sequenceNum)
+		t.Fatalf("bad seqno %v", m1.sequenceNum)
 	}
 }