From acc21cc5fe1f9bf4782b3a9fa8b3a5633da4cd7d Mon Sep 17 00:00:00 2001 From: Kyle Havlovitz Date: Mon, 5 Dec 2016 17:27:52 -0500 Subject: [PATCH] Update memberlist vendor deps --- .../github.com/hashicorp/memberlist/config.go | 7 +++ .../hashicorp/memberlist/memberlist.go | 6 +-- vendor/github.com/hashicorp/memberlist/net.go | 51 ++++++++++++++++--- .../github.com/hashicorp/memberlist/state.go | 4 +- .../github.com/hashicorp/memberlist/util.go | 12 ++++- vendor/vendor.json | 6 +-- 6 files changed, 69 insertions(+), 17 deletions(-) diff --git a/vendor/github.com/hashicorp/memberlist/config.go b/vendor/github.com/hashicorp/memberlist/config.go index 85a93f4228d4..27a52ea56121 100644 --- a/vendor/github.com/hashicorp/memberlist/config.go +++ b/vendor/github.com/hashicorp/memberlist/config.go @@ -179,6 +179,11 @@ type Config struct { // behavior for using LogOutput. You cannot specify both LogOutput and Logger // at the same time. Logger *log.Logger + + // Size of Memberlist's internal channel which handles UDP messages. The + // size of this determines the size of the queue which Memberlist will keep + // while UDP messages are handled. + HandoffQueueDepth int } // DefaultLANConfig returns a sane set of configurations for Memberlist. @@ -216,6 +221,8 @@ func DefaultLANConfig() *Config { Keyring: nil, DNSConfigPath: "/etc/resolv.conf", + + HandoffQueueDepth: 1024, } } diff --git a/vendor/github.com/hashicorp/memberlist/memberlist.go b/vendor/github.com/hashicorp/memberlist/memberlist.go index 7d7e563ef0b8..7e696762599d 100644 --- a/vendor/github.com/hashicorp/memberlist/memberlist.go +++ b/vendor/github.com/hashicorp/memberlist/memberlist.go @@ -129,7 +129,7 @@ func newMemberlist(conf *Config) (*Memberlist, error) { leaveBroadcast: make(chan struct{}, 1), udpListener: udpLn, tcpListener: tcpLn, - handoff: make(chan msgHandoff, 1024), + handoff: make(chan msgHandoff, conf.HandoffQueueDepth), nodeMap: make(map[string]*nodeState), nodeTimers: make(map[string]*suspicion), awareness: newAwareness(conf.AwarenessMaxMultiplier), @@ -496,7 +496,7 @@ func (m *Memberlist) SendTo(to net.Addr, msg []byte) error { buf = append(buf, msg...) // Send the message - return m.rawSendMsgUDP(to, buf) + return m.rawSendMsgUDP(to, nil, buf) } // SendToUDP is used to directly send a message to another node, without @@ -513,7 +513,7 @@ func (m *Memberlist) SendToUDP(to *Node, msg []byte) error { // Send the message destAddr := &net.UDPAddr{IP: to.Addr, Port: int(to.Port)} - return m.rawSendMsgUDP(destAddr, buf) + return m.rawSendMsgUDP(destAddr, to, buf) } // SendToTCP is used to directly send a message to another node, without diff --git a/vendor/github.com/hashicorp/memberlist/net.go b/vendor/github.com/hashicorp/memberlist/net.go index 7f66f5d0a49f..378adba78c2d 100644 --- a/vendor/github.com/hashicorp/memberlist/net.go +++ b/vendor/github.com/hashicorp/memberlist/net.go @@ -5,6 +5,7 @@ import ( "bytes" "encoding/binary" "fmt" + "hash/crc32" "io" "net" "time" @@ -32,7 +33,7 @@ const ( // understand version 4 or greater. ProtocolVersion2Compatible = 2 - ProtocolVersionMax = 4 + ProtocolVersionMax = 5 ) // messageType is an integer ID of a type of message that can be received @@ -53,6 +54,7 @@ const ( compressMsg encryptMsg nackRespMsg + hasCrcMsg ) // compressionType is used to specify the compression algorithm @@ -338,8 +340,18 @@ func (m *Memberlist) ingestPacket(buf []byte, from net.Addr, timestamp time.Time buf = plain } - // Handle the command - m.handleCommand(buf, from, timestamp) + // See if there's a checksum included to verify the contents of the message + if len(buf) >= 5 && messageType(buf[0]) == hasCrcMsg { + crc := crc32.ChecksumIEEE(buf[5:]) + expected := binary.BigEndian.Uint32(buf[1:5]) + if crc != expected { + m.logger.Printf("[WARN] memberlist: Got invalid checksum for UDP packet: %x, %x", crc, expected) + return + } + m.handleCommand(buf[5:], from, timestamp) + } else { + m.handleCommand(buf, from, timestamp) + } } func (m *Memberlist) handleCommand(buf []byte, from net.Addr, timestamp time.Time) { @@ -601,7 +613,7 @@ func (m *Memberlist) sendMsg(to net.Addr, msg []byte) error { // Fast path if nothing to piggypack if len(extra) == 0 { - return m.rawSendMsgUDP(to, msg) + return m.rawSendMsgUDP(to, nil, msg) } // Join all the messages @@ -613,11 +625,11 @@ func (m *Memberlist) sendMsg(to net.Addr, msg []byte) error { compound := makeCompoundMessage(msgs) // Send the message - return m.rawSendMsgUDP(to, compound.Bytes()) + return m.rawSendMsgUDP(to, nil, compound.Bytes()) } // rawSendMsgUDP is used to send a UDP message to another host without modification -func (m *Memberlist) rawSendMsgUDP(to net.Addr, msg []byte) error { +func (m *Memberlist) rawSendMsgUDP(addr net.Addr, node *Node, msg []byte) error { // Check if we have compression enabled if m.config.EnableCompression { buf, err := compressPayload(msg) @@ -631,6 +643,31 @@ func (m *Memberlist) rawSendMsgUDP(to net.Addr, msg []byte) error { } } + // Try to look up the destination node + if node == nil { + toAddr, _, err := net.SplitHostPort(addr.String()) + if err != nil { + m.logger.Printf("[ERR] memberlist: Failed to parse address %q: %v", addr.String(), err) + return err + } + m.nodeLock.RLock() + nodeState, ok := m.nodeMap[toAddr] + m.nodeLock.RUnlock() + if ok { + node = &nodeState.Node + } + } + + // Add a CRC to the end of the payload if the recipient understands + // ProtocolVersion >= 5 + if node != nil && node.PMax >= 5 { + crc := crc32.ChecksumIEEE(msg) + header := make([]byte, 5, 5+len(msg)) + header[0] = byte(hasCrcMsg) + binary.BigEndian.PutUint32(header[1:], crc) + msg = append(header, msg...) + } + // Check if we have encryption enabled if m.config.EncryptionEnabled() { // Encrypt the payload @@ -645,7 +682,7 @@ func (m *Memberlist) rawSendMsgUDP(to net.Addr, msg []byte) error { } metrics.IncrCounter([]string{"memberlist", "udp", "sent"}, float32(len(msg))) - _, err := m.udpListener.WriteTo(msg, to) + _, err := m.udpListener.WriteTo(msg, addr) return err } diff --git a/vendor/github.com/hashicorp/memberlist/state.go b/vendor/github.com/hashicorp/memberlist/state.go index 77871299d75d..d5d34e40d783 100644 --- a/vendor/github.com/hashicorp/memberlist/state.go +++ b/vendor/github.com/hashicorp/memberlist/state.go @@ -261,7 +261,7 @@ func (m *Memberlist) probeNode(node *nodeState) { } compound := makeCompoundMessage(msgs) - if err := m.rawSendMsgUDP(destAddr, compound.Bytes()); err != nil { + if err := m.rawSendMsgUDP(destAddr, &node.Node, compound.Bytes()); err != nil { m.logger.Printf("[ERR] memberlist: Failed to send compound ping and suspect message to %s: %s", destAddr, err) return } @@ -484,7 +484,7 @@ func (m *Memberlist) gossip() { // Send the compound message destAddr := &net.UDPAddr{IP: node.Addr, Port: int(node.Port)} - if err := m.rawSendMsgUDP(destAddr, compound.Bytes()); err != nil { + if err := m.rawSendMsgUDP(destAddr, &node.Node, compound.Bytes()); err != nil { m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", destAddr, err) } } diff --git a/vendor/github.com/hashicorp/memberlist/util.go b/vendor/github.com/hashicorp/memberlist/util.go index 7a59e3b3705b..c082ec69044d 100644 --- a/vendor/github.com/hashicorp/memberlist/util.go +++ b/vendor/github.com/hashicorp/memberlist/util.go @@ -327,10 +327,18 @@ func isLoopbackIP(ip_str string) bool { return loopbackBlock.Contains(ip) } -// Given a string of the form "host", "host:port", or "[ipv6::address]:port", +// Given a string of the form "host", "host:port", +// "ipv6::addr" or "[ipv6::address]:port", // return true if the string includes a port. func hasPort(s string) bool { - return strings.LastIndex(s, ":") > strings.LastIndex(s, "]") + last := strings.LastIndex(s, ":") + if last == -1 { + return false + } + if s[0] == '[' { + return s[last-1] == ']' + } + return strings.Index(s, ":") == last } // compressPayload takes an opaque input buffer, compresses it diff --git a/vendor/vendor.json b/vendor/vendor.json index d3a08ca82a41..d4c8b1cec5ed 100644 --- a/vendor/vendor.json +++ b/vendor/vendor.json @@ -528,10 +528,10 @@ "revisionTime": "2015-06-09T07:04:31Z" }, { - "checksumSHA1": "AY1/cRsuWpoJMG0J821TqFo9nDE=", + "checksumSHA1": "F2jm1h5jRic/Q0e3UEk4aqzS7k0=", "path": "github.com/hashicorp/memberlist", - "revision": "0c5ba075f8520c65572f001331a1a43b756e01d7", - "revisionTime": "2016-08-12T18:27:57Z" + "revision": "56f5fd70afa73f13bbe529192aeb3dc7bc4bc960", + "revisionTime": "2016-12-05T22:01:58Z" }, { "checksumSHA1": "qnlqWJYV81ENr61SZk9c65R1mDo=",