коммит произвёл
GitHub
родитель
65733f4969
Коммит
9dc64173dc
15
vendor/github.com/hashicorp/memberlist/.travis.yml
сгенерированный
поставляемый
15
vendor/github.com/hashicorp/memberlist/.travis.yml
сгенерированный
поставляемый
@@ -1,15 +0,0 @@
|
||||
language: go
|
||||
|
||||
sudo: true
|
||||
|
||||
go:
|
||||
- "1.x"
|
||||
|
||||
branches:
|
||||
only:
|
||||
- master
|
||||
|
||||
install: true
|
||||
|
||||
env:
|
||||
- GO111MODULE=on # Enable Go modules in 1.11
|
||||
29
vendor/github.com/hashicorp/memberlist/Makefile
сгенерированный
поставляемый
29
vendor/github.com/hashicorp/memberlist/Makefile
сгенерированный
поставляемый
@@ -1,6 +1,10 @@
|
||||
DEPS := $(shell go list -f '{{range .Imports}}{{.}} {{end}}' ./...)
|
||||
SHELL := bash
|
||||
|
||||
test: subnet
|
||||
GOFILES ?= $(shell go list ./... | grep -v /vendor/)
|
||||
|
||||
default: test
|
||||
|
||||
test: vet subnet
|
||||
go test ./...
|
||||
|
||||
integ: subnet
|
||||
@@ -10,11 +14,20 @@ subnet:
|
||||
./test/setup_subnet.sh
|
||||
|
||||
cov:
|
||||
gocov test github.com/hashicorp/memberlist | gocov-html > /tmp/coverage.html
|
||||
open /tmp/coverage.html
|
||||
go test ./... -coverprofile=coverage.out
|
||||
go tool cover -html=coverage.out
|
||||
|
||||
deps:
|
||||
go get -t -d -v ./...
|
||||
echo $(DEPS) | xargs -n1 go get -d
|
||||
format:
|
||||
@echo "--> Running go fmt"
|
||||
@go fmt $(GOFILES)
|
||||
|
||||
.PHONY: test cov integ
|
||||
vet:
|
||||
@echo "--> Running go vet"
|
||||
@go vet -tags '$(GOTAGS)' $(GOFILES); if [ $$? -eq 1 ]; then \
|
||||
echo ""; \
|
||||
echo "Vet found suspicious constructs. Please check the reported constructs"; \
|
||||
echo "and fix them if necessary before submitting the code for review."; \
|
||||
exit 1; \
|
||||
fi
|
||||
|
||||
.PHONY: default test integ subnet cov format vet
|
||||
|
||||
4
vendor/github.com/hashicorp/memberlist/README.md
сгенерированный
поставляемый
4
vendor/github.com/hashicorp/memberlist/README.md
сгенерированный
поставляемый
@@ -1,4 +1,4 @@
|
||||
# memberlist [](https://godoc.org/github.com/hashicorp/memberlist) [](https://travis-ci.org/hashicorp/memberlist)
|
||||
# memberlist [](https://godoc.org/github.com/hashicorp/memberlist) [](https://circleci.com/gh/hashicorp/memberlist)
|
||||
|
||||
memberlist is a [Go](http://www.golang.org) library that manages cluster
|
||||
membership and member failure detection using a gossip based protocol.
|
||||
@@ -23,8 +23,6 @@ Please check your installation with:
|
||||
go version
|
||||
```
|
||||
|
||||
Run `make deps` to fetch dependencies before building
|
||||
|
||||
## Usage
|
||||
|
||||
Memberlist is surprisingly simple to use. An example is shown below:
|
||||
|
||||
61
vendor/github.com/hashicorp/memberlist/config.go
сгенерированный
поставляемый
61
vendor/github.com/hashicorp/memberlist/config.go
сгенерированный
поставляемый
@@ -1,10 +1,15 @@
|
||||
package memberlist
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
multierror "github.com/hashicorp/go-multierror"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
@@ -116,6 +121,10 @@ type Config struct {
|
||||
// indirect UDP pings.
|
||||
DisableTcpPings bool
|
||||
|
||||
// DisableTcpPingsForNode is like DisableTcpPings, but lets you control
|
||||
// whether to perform TCP pings on a node-by-node basis.
|
||||
DisableTcpPingsForNode func(nodeName string) bool
|
||||
|
||||
// AwarenessMaxMultiplier will increase the probe interval if the node
|
||||
// becomes aware that it might be degraded and not meeting the soft real
|
||||
// time requirements to reliably probe other nodes.
|
||||
@@ -220,6 +229,39 @@ type Config struct {
|
||||
// reclaimed by one with a different address or port. By default, this is 0,
|
||||
// meaning nodes cannot be reclaimed this way.
|
||||
DeadNodeReclaimTime time.Duration
|
||||
|
||||
// RequireNodeNames controls if the name of a node is required when sending
|
||||
// a message to that node.
|
||||
RequireNodeNames bool
|
||||
// CIDRsAllowed If nil, allow any connection (default), otherwise specify all networks
|
||||
// allowed to connect (you must specify IPv6/IPv4 separately)
|
||||
// Using [] will block all connections.
|
||||
CIDRsAllowed []net.IPNet
|
||||
}
|
||||
|
||||
// ParseCIDRs return a possible empty list of all Network that have been parsed
|
||||
// In case of error, it returns succesfully parsed CIDRs and the last error found
|
||||
func ParseCIDRs(v []string) ([]net.IPNet, error) {
|
||||
nets := make([]net.IPNet, 0)
|
||||
if v == nil {
|
||||
return nets, nil
|
||||
}
|
||||
var errs error
|
||||
hasErrors := false
|
||||
for _, p := range v {
|
||||
_, net, err := net.ParseCIDR(strings.TrimSpace(p))
|
||||
if err != nil {
|
||||
err = fmt.Errorf("invalid cidr: %s", p)
|
||||
errs = multierror.Append(errs, err)
|
||||
hasErrors = true
|
||||
} else {
|
||||
nets = append(nets, *net)
|
||||
}
|
||||
}
|
||||
if !hasErrors {
|
||||
errs = nil
|
||||
}
|
||||
return nets, errs
|
||||
}
|
||||
|
||||
// DefaultLANConfig returns a sane set of configurations for Memberlist.
|
||||
@@ -263,6 +305,7 @@ func DefaultLANConfig() *Config {
|
||||
|
||||
HandoffQueueDepth: 1024,
|
||||
UDPBufferSize: 1400,
|
||||
CIDRsAllowed: nil, // same as allow all
|
||||
}
|
||||
}
|
||||
|
||||
@@ -282,6 +325,24 @@ func DefaultWANConfig() *Config {
|
||||
return conf
|
||||
}
|
||||
|
||||
// IPMustBeChecked return true if IPAllowed must be called
|
||||
func (c *Config) IPMustBeChecked() bool {
|
||||
return len(c.CIDRsAllowed) > 0
|
||||
}
|
||||
|
||||
// IPAllowed return an error if access to memberlist is denied
|
||||
func (c *Config) IPAllowed(ip net.IP) error {
|
||||
if !c.IPMustBeChecked() {
|
||||
return nil
|
||||
}
|
||||
for _, n := range c.CIDRsAllowed {
|
||||
if n.Contains(ip) {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("%s is not allowed", ip)
|
||||
}
|
||||
|
||||
// DefaultLocalConfig works like DefaultConfig, however it returns a configuration
|
||||
// that is optimized for a local loopback environments. The default configuration is
|
||||
// still very conservative and errs on the side of caution.
|
||||
|
||||
8
vendor/github.com/hashicorp/memberlist/go.mod
сгенерированный
поставляемый
8
vendor/github.com/hashicorp/memberlist/go.mod
сгенерированный
поставляемый
@@ -1,5 +1,7 @@
|
||||
module github.com/hashicorp/memberlist
|
||||
|
||||
go 1.12
|
||||
|
||||
require (
|
||||
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da
|
||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||
@@ -8,13 +10,9 @@ require (
|
||||
github.com/hashicorp/go-msgpack v0.5.3
|
||||
github.com/hashicorp/go-multierror v1.0.0
|
||||
github.com/hashicorp/go-sockaddr v1.0.0
|
||||
github.com/miekg/dns v1.0.14
|
||||
github.com/miekg/dns v1.1.26
|
||||
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529
|
||||
github.com/stretchr/testify v1.2.2
|
||||
golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3 // indirect
|
||||
golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519 // indirect
|
||||
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4 // indirect
|
||||
golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5 // indirect
|
||||
)
|
||||
|
||||
30
vendor/github.com/hashicorp/memberlist/go.sum
сгенерированный
поставляемый
30
vendor/github.com/hashicorp/memberlist/go.sum
сгенерированный
поставляемый
@@ -18,8 +18,8 @@ github.com/hashicorp/go-uuid v1.0.0 h1:RS8zrF7PhGwyNPOtxSClXXj9HA8feRnJzgnI1RJCS
|
||||
github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
|
||||
github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo=
|
||||
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
|
||||
github.com/miekg/dns v1.0.14 h1:9jZdLNd/P4+SfEJ0TNyxYpsK8N4GtfylBLqtbYN1sbA=
|
||||
github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
|
||||
github.com/miekg/dns v1.1.26 h1:gPxPSwALAeHJSjarOs00QjVdV9QoBvc1D2ujQUr5BzU=
|
||||
github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso=
|
||||
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c h1:Lgl0gzECD8GnQ5QCWA8o6BtfL6mDH5rQgM4/fX3avOs=
|
||||
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
@@ -28,11 +28,21 @@ github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUt
|
||||
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
|
||||
github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
|
||||
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
|
||||
golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3 h1:KYQXGkl6vs02hK7pK4eIbw0NpNPedieTSTEiJ//bwGs=
|
||||
golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
|
||||
golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519 h1:x6rhz8Y9CjbgQkccRGmELH6K+LJj7tOoh3XWeC1yaQM=
|
||||
golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4 h1:YUO/7uOKsKeq9UokNS62b8FYywz3ker1l1vDZRCRefw=
|
||||
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5 h1:x6r4Jo0KNzOOzYd8lbcRsqjuqEASK6ob3auvWYM4/8U=
|
||||
golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20190923035154-9ee001bba392 h1:ACG4HJsFiNMf47Y4PeRoebLNy/2lXT9EtprMuTFWt1M=
|
||||
golang.org/x/crypto v0.0.0-20190923035154-9ee001bba392/go.mod h1:/lpIB1dKB+9EgE3H3cr1v9wB50oz8l4C4h62xy7jSTY=
|
||||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20190923162816-aa69164e4478 h1:l5EDrHhldLYb3ZRHDUhXF7Om7MvYXnkV9/iQNo1lX6g=
|
||||
golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58 h1:8gQV6CLnAEikrhgkHFbMAEhagSSnXWGV915qUMm9mrU=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190922100055-0a153f010e69/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe h1:6fAMxZRR6sl1Uq8U61gxU+kPTs2tR8uOySCbBP7BN/M=
|
||||
golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20190907020128-2ca718005c18/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
|
||||
8
vendor/github.com/hashicorp/memberlist/logging.go
сгенерированный
поставляемый
8
vendor/github.com/hashicorp/memberlist/logging.go
сгенерированный
поставляемый
@@ -13,6 +13,14 @@ func LogAddress(addr net.Addr) string {
|
||||
return fmt.Sprintf("from=%s", addr.String())
|
||||
}
|
||||
|
||||
func LogStringAddress(addr string) string {
|
||||
if addr == "" {
|
||||
return "from=<unknown address>"
|
||||
}
|
||||
|
||||
return fmt.Sprintf("from=%s", addr)
|
||||
}
|
||||
|
||||
func LogConn(conn net.Conn) string {
|
||||
if conn == nil {
|
||||
return LogAddress(nil)
|
||||
|
||||
125
vendor/github.com/hashicorp/memberlist/memberlist.go
сгенерированный
поставляемый
125
vendor/github.com/hashicorp/memberlist/memberlist.go
сгенерированный
поставляемый
@@ -16,6 +16,7 @@ package memberlist
|
||||
|
||||
import (
|
||||
"container/list"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
@@ -31,12 +32,18 @@ import (
|
||||
"github.com/miekg/dns"
|
||||
)
|
||||
|
||||
var errNodeNamesAreRequired = errors.New("memberlist: node names are required by configuration but one was not provided")
|
||||
|
||||
type Memberlist struct {
|
||||
sequenceNum uint32 // Local sequence number
|
||||
incarnation uint32 // Local incarnation number
|
||||
numNodes uint32 // Number of known nodes (estimate)
|
||||
pushPullReq uint32 // Number of push/pull requests
|
||||
|
||||
advertiseLock sync.RWMutex
|
||||
advertiseAddr net.IP
|
||||
advertisePort uint16
|
||||
|
||||
config *Config
|
||||
shutdown int32 // Used as an atomic boolean value
|
||||
shutdownCh chan struct{}
|
||||
@@ -46,7 +53,7 @@ type Memberlist struct {
|
||||
shutdownLock sync.Mutex // Serializes calls to Shutdown
|
||||
leaveLock sync.Mutex // Serializes calls to Leave
|
||||
|
||||
transport Transport
|
||||
transport NodeAwareTransport
|
||||
|
||||
handoffCh chan struct{}
|
||||
highPriorityMsgQueue *list.List
|
||||
@@ -55,8 +62,8 @@ type Memberlist struct {
|
||||
|
||||
nodeLock sync.RWMutex
|
||||
nodes []*nodeState // Known nodes
|
||||
nodeMap map[string]*nodeState // Maps Addr.String() -> NodeState
|
||||
nodeTimers map[string]*suspicion // Maps Addr.String() -> suspicion timer
|
||||
nodeMap map[string]*nodeState // Maps Node.Name -> NodeState
|
||||
nodeTimers map[string]*suspicion // Maps Node.Name -> suspicion timer
|
||||
awareness *awareness
|
||||
|
||||
tickerLock sync.Mutex
|
||||
@@ -174,11 +181,17 @@ func newMemberlist(conf *Config) (*Memberlist, error) {
|
||||
transport = nt
|
||||
}
|
||||
|
||||
nodeAwareTransport, ok := transport.(NodeAwareTransport)
|
||||
if !ok {
|
||||
logger.Printf("[DEBUG] memberlist: configured Transport is not a NodeAwareTransport and some features may not work as desired")
|
||||
nodeAwareTransport = &shimNodeAwareTransport{transport}
|
||||
}
|
||||
|
||||
m := &Memberlist{
|
||||
config: conf,
|
||||
shutdownCh: make(chan struct{}),
|
||||
leaveBroadcast: make(chan struct{}, 1),
|
||||
transport: transport,
|
||||
transport: nodeAwareTransport,
|
||||
handoffCh: make(chan struct{}, 1),
|
||||
highPriorityMsgQueue: list.New(),
|
||||
lowPriorityMsgQueue: list.New(),
|
||||
@@ -192,6 +205,14 @@ func newMemberlist(conf *Config) (*Memberlist, error) {
|
||||
m.broadcasts.NumNodes = func() int {
|
||||
return m.estNumNodes()
|
||||
}
|
||||
|
||||
// Get the final advertise address from the transport, which may need
|
||||
// to see which address we bound to. We'll refresh this each time we
|
||||
// send out an alive message.
|
||||
if _, _, err := m.refreshAdvertise(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
go m.streamListen()
|
||||
go m.packetListen()
|
||||
go m.packetHandler()
|
||||
@@ -239,7 +260,8 @@ func (m *Memberlist) Join(existing []string) (int, error) {
|
||||
|
||||
for _, addr := range addrs {
|
||||
hp := joinHostPort(addr.ip.String(), addr.port)
|
||||
if err := m.pushPullNode(hp, true); err != nil {
|
||||
a := Address{Addr: hp, Name: addr.nodeName}
|
||||
if err := m.pushPullNode(a, true); err != nil {
|
||||
err = fmt.Errorf("Failed to join %s: %v", addr.ip, err)
|
||||
errs = multierror.Append(errs, err)
|
||||
m.logger.Printf("[DEBUG] memberlist: %v", err)
|
||||
@@ -257,8 +279,9 @@ func (m *Memberlist) Join(existing []string) (int, error) {
|
||||
|
||||
// ipPort holds information about a node we want to try to join.
|
||||
type ipPort struct {
|
||||
ip net.IP
|
||||
port uint16
|
||||
ip net.IP
|
||||
port uint16
|
||||
nodeName string // optional
|
||||
}
|
||||
|
||||
// tcpLookupIP is a helper to initiate a TCP-based DNS lookup for the given host.
|
||||
@@ -267,7 +290,7 @@ type ipPort struct {
|
||||
// Consul's. By doing the TCP lookup directly, we get the best chance for the
|
||||
// largest list of hosts to join. Since joins are relatively rare events, it's ok
|
||||
// to do this rather expensive operation.
|
||||
func (m *Memberlist) tcpLookupIP(host string, defaultPort uint16) ([]ipPort, error) {
|
||||
func (m *Memberlist) tcpLookupIP(host string, defaultPort uint16, nodeName string) ([]ipPort, error) {
|
||||
// Don't attempt any TCP lookups against non-fully qualified domain
|
||||
// names, since those will likely come from the resolv.conf file.
|
||||
if !strings.Contains(host, ".") {
|
||||
@@ -309,9 +332,9 @@ func (m *Memberlist) tcpLookupIP(host string, defaultPort uint16) ([]ipPort, err
|
||||
for _, r := range in.Answer {
|
||||
switch rr := r.(type) {
|
||||
case (*dns.A):
|
||||
ips = append(ips, ipPort{rr.A, defaultPort})
|
||||
ips = append(ips, ipPort{ip: rr.A, port: defaultPort, nodeName: nodeName})
|
||||
case (*dns.AAAA):
|
||||
ips = append(ips, ipPort{rr.AAAA, defaultPort})
|
||||
ips = append(ips, ipPort{ip: rr.AAAA, port: defaultPort, nodeName: nodeName})
|
||||
case (*dns.CNAME):
|
||||
m.logger.Printf("[DEBUG] memberlist: Ignoring CNAME RR in TCP-first answer for '%s'", host)
|
||||
}
|
||||
@@ -325,6 +348,16 @@ func (m *Memberlist) tcpLookupIP(host string, defaultPort uint16) ([]ipPort, err
|
||||
// resolveAddr is used to resolve the address into an address,
|
||||
// port, and error. If no port is given, use the default
|
||||
func (m *Memberlist) resolveAddr(hostStr string) ([]ipPort, error) {
|
||||
// First peel off any leading node name. This is optional.
|
||||
nodeName := ""
|
||||
if slashIdx := strings.Index(hostStr, "/"); slashIdx >= 0 {
|
||||
if slashIdx == 0 {
|
||||
return nil, fmt.Errorf("empty node name provided")
|
||||
}
|
||||
nodeName = hostStr[0:slashIdx]
|
||||
hostStr = hostStr[slashIdx+1:]
|
||||
}
|
||||
|
||||
// This captures the supplied port, or the default one.
|
||||
hostStr = ensurePort(hostStr, m.config.BindPort)
|
||||
host, sport, err := net.SplitHostPort(hostStr)
|
||||
@@ -341,13 +374,15 @@ func (m *Memberlist) resolveAddr(hostStr string) ([]ipPort, error) {
|
||||
// will make sure the host part is in good shape for parsing, even for
|
||||
// IPv6 addresses.
|
||||
if ip := net.ParseIP(host); ip != nil {
|
||||
return []ipPort{ipPort{ip, port}}, nil
|
||||
return []ipPort{
|
||||
ipPort{ip: ip, port: port, nodeName: nodeName},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// First try TCP so we have the best chance for the largest list of
|
||||
// hosts to join. If this fails it's not fatal since this isn't a standard
|
||||
// way to query DNS, and we have a fallback below.
|
||||
ips, err := m.tcpLookupIP(host, port)
|
||||
ips, err := m.tcpLookupIP(host, port, nodeName)
|
||||
if err != nil {
|
||||
m.logger.Printf("[DEBUG] memberlist: TCP-first lookup failed for '%s', falling back to UDP: %s", hostStr, err)
|
||||
}
|
||||
@@ -364,7 +399,7 @@ func (m *Memberlist) resolveAddr(hostStr string) ([]ipPort, error) {
|
||||
}
|
||||
ips = make([]ipPort, 0, len(ans))
|
||||
for _, ip := range ans {
|
||||
ips = append(ips, ipPort{ip, port})
|
||||
ips = append(ips, ipPort{ip: ip, port: port, nodeName: nodeName})
|
||||
}
|
||||
return ips, nil
|
||||
}
|
||||
@@ -375,10 +410,9 @@ func (m *Memberlist) resolveAddr(hostStr string) ([]ipPort, error) {
|
||||
func (m *Memberlist) setAlive() error {
|
||||
// Get the final advertise address from the transport, which may need
|
||||
// to see which address we bound to.
|
||||
addr, port, err := m.transport.FinalAdvertiseAddr(
|
||||
m.config.AdvertiseAddr, m.config.AdvertisePort)
|
||||
addr, port, err := m.refreshAdvertise()
|
||||
if err != nil {
|
||||
return fmt.Errorf("Failed to get final advertise address: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
// Check if this is a public address without encryption
|
||||
@@ -414,9 +448,33 @@ func (m *Memberlist) setAlive() error {
|
||||
Vsn: m.config.BuildVsnArray(),
|
||||
}
|
||||
m.aliveNode(&a, nil, true)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Memberlist) getAdvertise() (net.IP, uint16) {
|
||||
m.advertiseLock.RLock()
|
||||
defer m.advertiseLock.RUnlock()
|
||||
return m.advertiseAddr, m.advertisePort
|
||||
}
|
||||
|
||||
func (m *Memberlist) setAdvertise(addr net.IP, port int) {
|
||||
m.advertiseLock.Lock()
|
||||
defer m.advertiseLock.Unlock()
|
||||
m.advertiseAddr = addr
|
||||
m.advertisePort = uint16(port)
|
||||
}
|
||||
|
||||
func (m *Memberlist) refreshAdvertise() (net.IP, int, error) {
|
||||
addr, port, err := m.transport.FinalAdvertiseAddr(
|
||||
m.config.AdvertiseAddr, m.config.AdvertisePort)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("Failed to get final advertise address: %v", err)
|
||||
}
|
||||
m.setAdvertise(addr, port)
|
||||
return addr, port, nil
|
||||
}
|
||||
|
||||
// LocalNode is used to return the local Node
|
||||
func (m *Memberlist) LocalNode() *Node {
|
||||
m.nodeLock.RLock()
|
||||
@@ -472,24 +530,29 @@ func (m *Memberlist) UpdateNode(timeout time.Duration) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SendTo is deprecated in favor of SendBestEffort, which requires a node to
|
||||
// target.
|
||||
// Deprecated: SendTo is deprecated in favor of SendBestEffort, which requires a node to
|
||||
// target. If you don't have a node then use SendToAddress.
|
||||
func (m *Memberlist) SendTo(to net.Addr, msg []byte) error {
|
||||
a := Address{Addr: to.String(), Name: ""}
|
||||
return m.SendToAddress(a, msg)
|
||||
}
|
||||
|
||||
func (m *Memberlist) SendToAddress(a Address, msg []byte) error {
|
||||
// Encode as a user message
|
||||
buf := make([]byte, 1, len(msg)+1)
|
||||
buf[0] = byte(userMsg)
|
||||
buf = append(buf, msg...)
|
||||
|
||||
// Send the message
|
||||
return m.rawSendMsgPacket(to.String(), nil, buf)
|
||||
return m.rawSendMsgPacket(a, nil, buf)
|
||||
}
|
||||
|
||||
// SendToUDP is deprecated in favor of SendBestEffort.
|
||||
// Deprecated: SendToUDP is deprecated in favor of SendBestEffort.
|
||||
func (m *Memberlist) SendToUDP(to *Node, msg []byte) error {
|
||||
return m.SendBestEffort(to, msg)
|
||||
}
|
||||
|
||||
// SendToTCP is deprecated in favor of SendReliable.
|
||||
// Deprecated: SendToTCP is deprecated in favor of SendReliable.
|
||||
func (m *Memberlist) SendToTCP(to *Node, msg []byte) error {
|
||||
return m.SendReliable(to, msg)
|
||||
}
|
||||
@@ -505,7 +568,8 @@ func (m *Memberlist) SendBestEffort(to *Node, msg []byte) error {
|
||||
buf = append(buf, msg...)
|
||||
|
||||
// Send the message
|
||||
return m.rawSendMsgPacket(to.Address(), to, buf)
|
||||
a := Address{Addr: to.Address(), Name: to.Name}
|
||||
return m.rawSendMsgPacket(a, to, buf)
|
||||
}
|
||||
|
||||
// SendReliable uses the reliable stream-oriented interface of the transport to
|
||||
@@ -513,7 +577,7 @@ func (m *Memberlist) SendBestEffort(to *Node, msg []byte) error {
|
||||
// mechanism). Delivery is guaranteed if no error is returned, and there is no
|
||||
// limit on the size of the message.
|
||||
func (m *Memberlist) SendReliable(to *Node, msg []byte) error {
|
||||
return m.sendUserMsg(to.Address(), msg)
|
||||
return m.sendUserMsg(to.FullAddress(), msg)
|
||||
}
|
||||
|
||||
// Members returns a list of all known live nodes. The node structures
|
||||
@@ -525,7 +589,7 @@ func (m *Memberlist) Members() []*Node {
|
||||
|
||||
nodes := make([]*Node, 0, len(m.nodes))
|
||||
for _, n := range m.nodes {
|
||||
if n.State != stateDead {
|
||||
if !n.DeadOrLeft() {
|
||||
nodes = append(nodes, &n.Node)
|
||||
}
|
||||
}
|
||||
@@ -542,7 +606,7 @@ func (m *Memberlist) NumMembers() (alive int) {
|
||||
defer m.nodeLock.RUnlock()
|
||||
|
||||
for _, n := range m.nodes {
|
||||
if n.State != stateDead {
|
||||
if !n.DeadOrLeft() {
|
||||
alive++
|
||||
}
|
||||
}
|
||||
@@ -579,9 +643,14 @@ func (m *Memberlist) Leave(timeout time.Duration) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// This dead message is special, because Node and From are the
|
||||
// same. This helps other nodes figure out that a node left
|
||||
// intentionally. When Node equals From, other nodes know for
|
||||
// sure this node is gone.
|
||||
d := dead{
|
||||
Incarnation: state.Incarnation,
|
||||
Node: state.Name,
|
||||
From: state.Name,
|
||||
}
|
||||
m.deadNode(&d)
|
||||
|
||||
@@ -607,7 +676,7 @@ func (m *Memberlist) anyAlive() bool {
|
||||
m.nodeLock.RLock()
|
||||
defer m.nodeLock.RUnlock()
|
||||
for _, n := range m.nodes {
|
||||
if n.State != stateDead && n.Name != m.config.Name {
|
||||
if !n.DeadOrLeft() && n.Name != m.config.Name {
|
||||
return true
|
||||
}
|
||||
}
|
||||
@@ -630,7 +699,7 @@ func (m *Memberlist) ProtocolVersion() uint8 {
|
||||
return m.config.ProtocolVersion
|
||||
}
|
||||
|
||||
// Shutdown will stop any background maintanence of network activity
|
||||
// Shutdown will stop any background maintenance of network activity
|
||||
// for this memberlist, causing it to appear "dead". A leave message
|
||||
// will not be broadcasted prior, so the cluster being left will have
|
||||
// to detect this node's shutdown using probing. If you wish to more
|
||||
@@ -667,7 +736,7 @@ func (m *Memberlist) hasLeft() bool {
|
||||
return atomic.LoadInt32(&m.leave) == 1
|
||||
}
|
||||
|
||||
func (m *Memberlist) getNodeState(addr string) nodeStateType {
|
||||
func (m *Memberlist) getNodeState(addr string) NodeStateType {
|
||||
m.nodeLock.RLock()
|
||||
defer m.nodeLock.RUnlock()
|
||||
|
||||
|
||||
100
vendor/github.com/hashicorp/memberlist/mock_transport.go
сгенерированный
поставляемый
100
vendor/github.com/hashicorp/memberlist/mock_transport.go
сгенерированный
поставляемый
@@ -1,7 +1,9 @@
|
||||
package memberlist
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"strconv"
|
||||
"time"
|
||||
@@ -10,26 +12,33 @@ import (
|
||||
// MockNetwork is used as a factory that produces MockTransport instances which
|
||||
// are uniquely addressed and wired up to talk to each other.
|
||||
type MockNetwork struct {
|
||||
transports map[string]*MockTransport
|
||||
port int
|
||||
transportsByAddr map[string]*MockTransport
|
||||
transportsByName map[string]*MockTransport
|
||||
port int
|
||||
}
|
||||
|
||||
// NewTransport returns a new MockTransport with a unique address, wired up to
|
||||
// talk to the other transports in the MockNetwork.
|
||||
func (n *MockNetwork) NewTransport() *MockTransport {
|
||||
func (n *MockNetwork) NewTransport(name string) *MockTransport {
|
||||
n.port += 1
|
||||
addr := fmt.Sprintf("127.0.0.1:%d", n.port)
|
||||
transport := &MockTransport{
|
||||
net: n,
|
||||
addr: &MockAddress{addr},
|
||||
addr: &MockAddress{addr, name},
|
||||
packetCh: make(chan *Packet),
|
||||
streamCh: make(chan net.Conn),
|
||||
}
|
||||
|
||||
if n.transports == nil {
|
||||
n.transports = make(map[string]*MockTransport)
|
||||
if n.transportsByAddr == nil {
|
||||
n.transportsByAddr = make(map[string]*MockTransport)
|
||||
}
|
||||
n.transports[addr] = transport
|
||||
n.transportsByAddr[addr] = transport
|
||||
|
||||
if n.transportsByName == nil {
|
||||
n.transportsByName = make(map[string]*MockTransport)
|
||||
}
|
||||
n.transportsByName[name] = transport
|
||||
|
||||
return transport
|
||||
}
|
||||
|
||||
@@ -37,6 +46,7 @@ func (n *MockNetwork) NewTransport() *MockTransport {
|
||||
// address scheme.
|
||||
type MockAddress struct {
|
||||
addr string
|
||||
name string
|
||||
}
|
||||
|
||||
// See net.Addr.
|
||||
@@ -57,6 +67,8 @@ type MockTransport struct {
|
||||
streamCh chan net.Conn
|
||||
}
|
||||
|
||||
var _ NodeAwareTransport = (*MockTransport)(nil)
|
||||
|
||||
// See Transport.
|
||||
func (t *MockTransport) FinalAdvertiseAddr(string, int) (net.IP, int, error) {
|
||||
host, portStr, err := net.SplitHostPort(t.addr.String())
|
||||
@@ -79,9 +91,15 @@ func (t *MockTransport) FinalAdvertiseAddr(string, int) (net.IP, int, error) {
|
||||
|
||||
// See Transport.
|
||||
func (t *MockTransport) WriteTo(b []byte, addr string) (time.Time, error) {
|
||||
dest, ok := t.net.transports[addr]
|
||||
if !ok {
|
||||
return time.Time{}, fmt.Errorf("No route to %q", addr)
|
||||
a := Address{Addr: addr, Name: ""}
|
||||
return t.WriteToAddress(b, a)
|
||||
}
|
||||
|
||||
// See NodeAwareTransport.
|
||||
func (t *MockTransport) WriteToAddress(b []byte, a Address) (time.Time, error) {
|
||||
dest, err := t.getPeer(a)
|
||||
if err != nil {
|
||||
return time.Time{}, err
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
@@ -98,11 +116,45 @@ func (t *MockTransport) PacketCh() <-chan *Packet {
|
||||
return t.packetCh
|
||||
}
|
||||
|
||||
// See NodeAwareTransport.
|
||||
func (t *MockTransport) IngestPacket(conn net.Conn, addr net.Addr, now time.Time, shouldClose bool) error {
|
||||
if shouldClose {
|
||||
defer conn.Close()
|
||||
}
|
||||
|
||||
// Copy everything from the stream into packet buffer.
|
||||
var buf bytes.Buffer
|
||||
if _, err := io.Copy(&buf, conn); err != nil {
|
||||
return fmt.Errorf("failed to read packet: %v", err)
|
||||
}
|
||||
|
||||
// Check the length - it needs to have at least one byte to be a proper
|
||||
// message. This is checked elsewhere for writes coming in directly from
|
||||
// the UDP socket.
|
||||
if n := buf.Len(); n < 1 {
|
||||
return fmt.Errorf("packet too short (%d bytes) %s", n, LogAddress(addr))
|
||||
}
|
||||
|
||||
// Inject the packet.
|
||||
t.packetCh <- &Packet{
|
||||
Buf: buf.Bytes(),
|
||||
From: addr,
|
||||
Timestamp: now,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// See Transport.
|
||||
func (t *MockTransport) DialTimeout(addr string, timeout time.Duration) (net.Conn, error) {
|
||||
dest, ok := t.net.transports[addr]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("No route to %q", addr)
|
||||
a := Address{Addr: addr, Name: ""}
|
||||
return t.DialAddressTimeout(a, timeout)
|
||||
}
|
||||
|
||||
// See NodeAwareTransport.
|
||||
func (t *MockTransport) DialAddressTimeout(a Address, timeout time.Duration) (net.Conn, error) {
|
||||
dest, err := t.getPeer(a)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
p1, p2 := net.Pipe()
|
||||
@@ -115,7 +167,29 @@ func (t *MockTransport) StreamCh() <-chan net.Conn {
|
||||
return t.streamCh
|
||||
}
|
||||
|
||||
// See NodeAwareTransport.
|
||||
func (t *MockTransport) IngestStream(conn net.Conn) error {
|
||||
t.streamCh <- conn
|
||||
return nil
|
||||
}
|
||||
|
||||
// See Transport.
|
||||
func (t *MockTransport) Shutdown() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *MockTransport) getPeer(a Address) (*MockTransport, error) {
|
||||
var (
|
||||
dest *MockTransport
|
||||
ok bool
|
||||
)
|
||||
if a.Name != "" {
|
||||
dest, ok = t.net.transportsByName[a.Name]
|
||||
} else {
|
||||
dest, ok = t.net.transportsByAddr[a.Addr]
|
||||
}
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("No route to %s", a)
|
||||
}
|
||||
return dest, nil
|
||||
}
|
||||
|
||||
191
vendor/github.com/hashicorp/memberlist/net.go
сгенерированный
поставляемый
191
vendor/github.com/hashicorp/memberlist/net.go
сгенерированный
поставляемый
@@ -84,15 +84,28 @@ type ping struct {
|
||||
// the intended recipient. This is to protect again an agent
|
||||
// restart with a new name.
|
||||
Node string
|
||||
|
||||
SourceAddr []byte `codec:",omitempty"` // Source address, used for a direct reply
|
||||
SourcePort uint16 `codec:",omitempty"` // Source port, used for a direct reply
|
||||
SourceNode string `codec:",omitempty"` // Source name, used for a direct reply
|
||||
}
|
||||
|
||||
// indirect ping sent to an indirect ndoe
|
||||
// indirect ping sent to an indirect node
|
||||
type indirectPingReq struct {
|
||||
SeqNo uint32
|
||||
Target []byte
|
||||
Port uint16
|
||||
Node string
|
||||
Nack bool // true if we'd like a nack back
|
||||
|
||||
// Node is sent so the target can verify they are
|
||||
// the intended recipient. This is to protect against an agent
|
||||
// restart with a new name.
|
||||
Node string
|
||||
|
||||
Nack bool // true if we'd like a nack back
|
||||
|
||||
SourceAddr []byte `codec:",omitempty"` // Source address, used for a direct reply
|
||||
SourcePort uint16 `codec:",omitempty"` // Source port, used for a direct reply
|
||||
SourceNode string `codec:",omitempty"` // Source name, used for a direct reply
|
||||
}
|
||||
|
||||
// ack response is sent for a ping
|
||||
@@ -163,7 +176,7 @@ type pushNodeState struct {
|
||||
Port uint16
|
||||
Meta []byte
|
||||
Incarnation uint32
|
||||
State nodeStateType
|
||||
State NodeStateType
|
||||
Vsn []uint8 // Protocol versions
|
||||
}
|
||||
|
||||
@@ -207,9 +220,9 @@ func (m *Memberlist) streamListen() {
|
||||
|
||||
// handleConn handles a single incoming stream connection from the transport.
|
||||
func (m *Memberlist) handleConn(conn net.Conn) {
|
||||
defer conn.Close()
|
||||
m.logger.Printf("[DEBUG] memberlist: Stream connection %s", LogConn(conn))
|
||||
|
||||
defer conn.Close()
|
||||
metrics.IncrCounter([]string{"memberlist", "tcp", "accept"}, 1)
|
||||
|
||||
conn.SetDeadline(time.Now().Add(m.config.TCPTimeout))
|
||||
@@ -483,7 +496,19 @@ func (m *Memberlist) handlePing(buf []byte, from net.Addr) {
|
||||
if m.config.Ping != nil {
|
||||
ack.Payload = m.config.Ping.AckPayload()
|
||||
}
|
||||
if err := m.encodeAndSendMsg(from.String(), ackRespMsg, &ack); err != nil {
|
||||
|
||||
addr := ""
|
||||
if len(p.SourceAddr) > 0 && p.SourcePort > 0 {
|
||||
addr = joinHostPort(net.IP(p.SourceAddr).String(), p.SourcePort)
|
||||
} else {
|
||||
addr = from.String()
|
||||
}
|
||||
|
||||
a := Address{
|
||||
Addr: addr,
|
||||
Name: p.SourceNode,
|
||||
}
|
||||
if err := m.encodeAndSendMsg(a, ackRespMsg, &ack); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send ack: %s %s", err, LogAddress(from))
|
||||
}
|
||||
}
|
||||
@@ -503,7 +528,25 @@ func (m *Memberlist) handleIndirectPing(buf []byte, from net.Addr) {
|
||||
|
||||
// Send a ping to the correct host.
|
||||
localSeqNo := m.nextSeqNo()
|
||||
ping := ping{SeqNo: localSeqNo, Node: ind.Node}
|
||||
selfAddr, selfPort := m.getAdvertise()
|
||||
ping := ping{
|
||||
SeqNo: localSeqNo,
|
||||
Node: ind.Node,
|
||||
// The outbound message is addressed FROM us.
|
||||
SourceAddr: selfAddr,
|
||||
SourcePort: selfPort,
|
||||
SourceNode: m.config.Name,
|
||||
}
|
||||
|
||||
// Forward the ack back to the requestor. If the request encodes an origin
|
||||
// use that otherwise assume that the other end of the UDP socket is
|
||||
// usable.
|
||||
indAddr := ""
|
||||
if len(ind.SourceAddr) > 0 && ind.SourcePort > 0 {
|
||||
indAddr = joinHostPort(net.IP(ind.SourceAddr).String(), ind.SourcePort)
|
||||
} else {
|
||||
indAddr = from.String()
|
||||
}
|
||||
|
||||
// Setup a response handler to relay the ack
|
||||
cancelCh := make(chan struct{})
|
||||
@@ -511,18 +554,25 @@ func (m *Memberlist) handleIndirectPing(buf []byte, from net.Addr) {
|
||||
// Try to prevent the nack if we've caught it in time.
|
||||
close(cancelCh)
|
||||
|
||||
// Forward the ack back to the requestor.
|
||||
ack := ackResp{ind.SeqNo, nil}
|
||||
if err := m.encodeAndSendMsg(from.String(), ackRespMsg, &ack); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to forward ack: %s %s", err, LogAddress(from))
|
||||
a := Address{
|
||||
Addr: indAddr,
|
||||
Name: ind.SourceNode,
|
||||
}
|
||||
if err := m.encodeAndSendMsg(a, ackRespMsg, &ack); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to forward ack: %s %s", err, LogStringAddress(indAddr))
|
||||
}
|
||||
}
|
||||
m.setAckHandler(localSeqNo, respHandler, m.config.ProbeTimeout)
|
||||
|
||||
// Send the ping.
|
||||
addr := joinHostPort(net.IP(ind.Target).String(), ind.Port)
|
||||
if err := m.encodeAndSendMsg(addr, pingMsg, &ping); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send indirect ping: %s %s", err, LogAddress(from))
|
||||
a := Address{
|
||||
Addr: addr,
|
||||
Name: ind.Node,
|
||||
}
|
||||
if err := m.encodeAndSendMsg(a, pingMsg, &ping); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send indirect ping: %s %s", err, LogStringAddress(indAddr))
|
||||
}
|
||||
|
||||
// Setup a timer to fire off a nack if no ack is seen in time.
|
||||
@@ -533,8 +583,12 @@ func (m *Memberlist) handleIndirectPing(buf []byte, from net.Addr) {
|
||||
return
|
||||
case <-time.After(m.config.ProbeTimeout):
|
||||
nack := nackResp{ind.SeqNo}
|
||||
if err := m.encodeAndSendMsg(from.String(), nackRespMsg, &nack); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send nack: %s %s", err, LogAddress(from))
|
||||
a := Address{
|
||||
Addr: indAddr,
|
||||
Name: ind.SourceNode,
|
||||
}
|
||||
if err := m.encodeAndSendMsg(a, nackRespMsg, &nack); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send nack: %s %s", err, LogStringAddress(indAddr))
|
||||
}
|
||||
}
|
||||
}()
|
||||
@@ -568,12 +622,47 @@ func (m *Memberlist) handleSuspect(buf []byte, from net.Addr) {
|
||||
m.suspectNode(&sus)
|
||||
}
|
||||
|
||||
// ensureCanConnect return the IP from a RemoteAddress
|
||||
// return error if this client must not connect
|
||||
func (m *Memberlist) ensureCanConnect(from net.Addr) error {
|
||||
if !m.config.IPMustBeChecked() {
|
||||
return nil
|
||||
}
|
||||
source := from.String()
|
||||
if source == "pipe" {
|
||||
return nil
|
||||
}
|
||||
host, _, err := net.SplitHostPort(source)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
ip := net.ParseIP(host)
|
||||
if ip == nil {
|
||||
return fmt.Errorf("Cannot parse IP from %s", host)
|
||||
}
|
||||
return m.config.IPAllowed(ip)
|
||||
}
|
||||
|
||||
func (m *Memberlist) handleAlive(buf []byte, from net.Addr) {
|
||||
if err := m.ensureCanConnect(from); err != nil {
|
||||
m.logger.Printf("[DEBUG] memberlist: Blocked alive message: %s %s", err, LogAddress(from))
|
||||
return
|
||||
}
|
||||
var live alive
|
||||
if err := decode(buf, &live); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to decode alive message: %s %s", err, LogAddress(from))
|
||||
return
|
||||
}
|
||||
if m.config.IPMustBeChecked() {
|
||||
innerIP := net.IP(live.Addr)
|
||||
if innerIP != nil {
|
||||
if err := m.config.IPAllowed(innerIP); err != nil {
|
||||
m.logger.Printf("[DEBUG] memberlist: Blocked alive.Addr=%s message from: %s %s", innerIP.String(), err, LogAddress(from))
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// For proto versions < 2, there is no port provided. Mask old
|
||||
// behavior by using the configured port
|
||||
@@ -615,12 +704,12 @@ func (m *Memberlist) handleCompressed(buf []byte, from net.Addr, timestamp time.
|
||||
}
|
||||
|
||||
// encodeAndSendMsg is used to combine the encoding and sending steps
|
||||
func (m *Memberlist) encodeAndSendMsg(addr string, msgType messageType, msg interface{}) error {
|
||||
func (m *Memberlist) encodeAndSendMsg(a Address, msgType messageType, msg interface{}) error {
|
||||
out, err := encode(msgType, msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := m.sendMsg(addr, out.Bytes()); err != nil {
|
||||
if err := m.sendMsg(a, out.Bytes()); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
@@ -628,7 +717,7 @@ func (m *Memberlist) encodeAndSendMsg(addr string, msgType messageType, msg inte
|
||||
|
||||
// sendMsg is used to send a message via packet to another host. It will
|
||||
// opportunistically create a compoundMsg and piggy back other broadcasts.
|
||||
func (m *Memberlist) sendMsg(addr string, msg []byte) error {
|
||||
func (m *Memberlist) sendMsg(a Address, msg []byte) error {
|
||||
// Check if we can piggy back any messages
|
||||
bytesAvail := m.config.UDPBufferSize - len(msg) - compoundHeaderOverhead
|
||||
if m.config.EncryptionEnabled() && m.config.GossipVerifyOutgoing {
|
||||
@@ -638,7 +727,7 @@ func (m *Memberlist) sendMsg(addr string, msg []byte) error {
|
||||
|
||||
// Fast path if nothing to piggypack
|
||||
if len(extra) == 0 {
|
||||
return m.rawSendMsgPacket(addr, nil, msg)
|
||||
return m.rawSendMsgPacket(a, nil, msg)
|
||||
}
|
||||
|
||||
// Join all the messages
|
||||
@@ -650,12 +739,16 @@ func (m *Memberlist) sendMsg(addr string, msg []byte) error {
|
||||
compound := makeCompoundMessage(msgs)
|
||||
|
||||
// Send the message
|
||||
return m.rawSendMsgPacket(addr, nil, compound.Bytes())
|
||||
return m.rawSendMsgPacket(a, nil, compound.Bytes())
|
||||
}
|
||||
|
||||
// rawSendMsgPacket is used to send message via packet to another host without
|
||||
// modification, other than compression or encryption if enabled.
|
||||
func (m *Memberlist) rawSendMsgPacket(addr string, node *Node, msg []byte) error {
|
||||
func (m *Memberlist) rawSendMsgPacket(a Address, node *Node, msg []byte) error {
|
||||
if a.Name == "" && m.config.RequireNodeNames {
|
||||
return errNodeNamesAreRequired
|
||||
}
|
||||
|
||||
// Check if we have compression enabled
|
||||
if m.config.EnableCompression {
|
||||
buf, err := compressPayload(msg)
|
||||
@@ -669,11 +762,12 @@ func (m *Memberlist) rawSendMsgPacket(addr string, node *Node, msg []byte) error
|
||||
}
|
||||
}
|
||||
|
||||
// Try to look up the destination node
|
||||
// Try to look up the destination node. Note this will only work if the
|
||||
// bare ip address is used as the node name, which is not guaranteed.
|
||||
if node == nil {
|
||||
toAddr, _, err := net.SplitHostPort(addr)
|
||||
toAddr, _, err := net.SplitHostPort(a.Addr)
|
||||
if err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to parse address %q: %v", addr, err)
|
||||
m.logger.Printf("[ERR] memberlist: Failed to parse address %q: %v", a.Addr, err)
|
||||
return err
|
||||
}
|
||||
m.nodeLock.RLock()
|
||||
@@ -708,14 +802,14 @@ func (m *Memberlist) rawSendMsgPacket(addr string, node *Node, msg []byte) error
|
||||
}
|
||||
|
||||
metrics.IncrCounter([]string{"memberlist", "udp", "sent"}, float32(len(msg)))
|
||||
_, err := m.transport.WriteTo(msg, addr)
|
||||
_, err := m.transport.WriteToAddress(msg, a)
|
||||
return err
|
||||
}
|
||||
|
||||
// rawSendMsgStream is used to stream a message to another host without
|
||||
// modification, other than applying compression and encryption if enabled.
|
||||
func (m *Memberlist) rawSendMsgStream(conn net.Conn, sendBuf []byte) error {
|
||||
// Check if compresion is enabled
|
||||
// Check if compression is enabled
|
||||
if m.config.EnableCompression {
|
||||
compBuf, err := compressPayload(sendBuf)
|
||||
if err != nil {
|
||||
@@ -748,8 +842,12 @@ func (m *Memberlist) rawSendMsgStream(conn net.Conn, sendBuf []byte) error {
|
||||
}
|
||||
|
||||
// sendUserMsg is used to stream a user message to another host.
|
||||
func (m *Memberlist) sendUserMsg(addr string, sendBuf []byte) error {
|
||||
conn, err := m.transport.DialTimeout(addr, m.config.TCPTimeout)
|
||||
func (m *Memberlist) sendUserMsg(a Address, sendBuf []byte) error {
|
||||
if a.Name == "" && m.config.RequireNodeNames {
|
||||
return errNodeNamesAreRequired
|
||||
}
|
||||
|
||||
conn, err := m.transport.DialAddressTimeout(a, m.config.TCPTimeout)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -774,14 +872,18 @@ func (m *Memberlist) sendUserMsg(addr string, sendBuf []byte) error {
|
||||
|
||||
// sendAndReceiveState is used to initiate a push/pull over a stream with a
|
||||
// remote host.
|
||||
func (m *Memberlist) sendAndReceiveState(addr string, join bool) ([]pushNodeState, []byte, error) {
|
||||
func (m *Memberlist) sendAndReceiveState(a Address, join bool) ([]pushNodeState, []byte, error) {
|
||||
if a.Name == "" && m.config.RequireNodeNames {
|
||||
return nil, nil, errNodeNamesAreRequired
|
||||
}
|
||||
|
||||
// Attempt to connect
|
||||
conn, err := m.transport.DialTimeout(addr, m.config.TCPTimeout)
|
||||
conn, err := m.transport.DialAddressTimeout(a, m.config.TCPTimeout)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
defer conn.Close()
|
||||
m.logger.Printf("[DEBUG] memberlist: Initiating push/pull sync with: %s", conn.RemoteAddr())
|
||||
m.logger.Printf("[DEBUG] memberlist: Initiating push/pull sync with: %s %s", a.Name, conn.RemoteAddr())
|
||||
metrics.IncrCounter([]string{"memberlist", "tcp", "connect"}, 1)
|
||||
|
||||
// Send our state
|
||||
@@ -1046,16 +1148,17 @@ func (m *Memberlist) mergeRemoteState(join bool, remoteNodes []pushNodeState, us
|
||||
nodes := make([]*Node, len(remoteNodes))
|
||||
for idx, n := range remoteNodes {
|
||||
nodes[idx] = &Node{
|
||||
Name: n.Name,
|
||||
Addr: n.Addr,
|
||||
Port: n.Port,
|
||||
Meta: n.Meta,
|
||||
PMin: n.Vsn[0],
|
||||
PMax: n.Vsn[1],
|
||||
PCur: n.Vsn[2],
|
||||
DMin: n.Vsn[3],
|
||||
DMax: n.Vsn[4],
|
||||
DCur: n.Vsn[5],
|
||||
Name: n.Name,
|
||||
Addr: n.Addr,
|
||||
Port: n.Port,
|
||||
Meta: n.Meta,
|
||||
State: n.State,
|
||||
PMin: n.Vsn[0],
|
||||
PMax: n.Vsn[1],
|
||||
PCur: n.Vsn[2],
|
||||
DMin: n.Vsn[3],
|
||||
DMax: n.Vsn[4],
|
||||
DCur: n.Vsn[5],
|
||||
}
|
||||
}
|
||||
if err := m.config.Merge.NotifyMerge(nodes); err != nil {
|
||||
@@ -1108,8 +1211,12 @@ func (m *Memberlist) readUserMsg(bufConn io.Reader, dec *codec.Decoder) error {
|
||||
// a ping, and waits for an ack. All of this is done as a series of blocking
|
||||
// operations, given the deadline. The bool return parameter is true if we
|
||||
// we able to round trip a ping to the other node.
|
||||
func (m *Memberlist) sendPingAndWaitForAck(addr string, ping ping, deadline time.Time) (bool, error) {
|
||||
conn, err := m.transport.DialTimeout(addr, deadline.Sub(time.Now()))
|
||||
func (m *Memberlist) sendPingAndWaitForAck(a Address, ping ping, deadline time.Time) (bool, error) {
|
||||
if a.Name == "" && m.config.RequireNodeNames {
|
||||
return false, errNodeNamesAreRequired
|
||||
}
|
||||
|
||||
conn, err := m.transport.DialAddressTimeout(a, deadline.Sub(time.Now()))
|
||||
if err != nil {
|
||||
// If the node is actually dead we expect this to fail, so we
|
||||
// shouldn't spam the logs with it. After this point, errors
|
||||
|
||||
54
vendor/github.com/hashicorp/memberlist/net_transport.go
сгенерированный
поставляемый
54
vendor/github.com/hashicorp/memberlist/net_transport.go
сгенерированный
поставляемый
@@ -1,7 +1,9 @@
|
||||
package memberlist
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net"
|
||||
"sync"
|
||||
@@ -48,6 +50,8 @@ type NetTransport struct {
|
||||
shutdown int32
|
||||
}
|
||||
|
||||
var _ NodeAwareTransport = (*NetTransport)(nil)
|
||||
|
||||
// NewNetTransport returns a net transport with the given configuration. On
|
||||
// success all the network listeners will be created and listening.
|
||||
func NewNetTransport(config *NetTransportConfig) (*NetTransport, error) {
|
||||
@@ -170,6 +174,14 @@ func (t *NetTransport) FinalAdvertiseAddr(ip string, port int) (net.IP, int, err
|
||||
|
||||
// See Transport.
|
||||
func (t *NetTransport) WriteTo(b []byte, addr string) (time.Time, error) {
|
||||
a := Address{Addr: addr, Name: ""}
|
||||
return t.WriteToAddress(b, a)
|
||||
}
|
||||
|
||||
// See NodeAwareTransport.
|
||||
func (t *NetTransport) WriteToAddress(b []byte, a Address) (time.Time, error) {
|
||||
addr := a.Addr
|
||||
|
||||
udpAddr, err := net.ResolveUDPAddr("udp", addr)
|
||||
if err != nil {
|
||||
return time.Time{}, err
|
||||
@@ -188,8 +200,44 @@ func (t *NetTransport) PacketCh() <-chan *Packet {
|
||||
return t.packetCh
|
||||
}
|
||||
|
||||
// See IngestionAwareTransport.
|
||||
func (t *NetTransport) IngestPacket(conn net.Conn, addr net.Addr, now time.Time, shouldClose bool) error {
|
||||
if shouldClose {
|
||||
defer conn.Close()
|
||||
}
|
||||
|
||||
// Copy everything from the stream into packet buffer.
|
||||
var buf bytes.Buffer
|
||||
if _, err := io.Copy(&buf, conn); err != nil {
|
||||
return fmt.Errorf("failed to read packet: %v", err)
|
||||
}
|
||||
|
||||
// Check the length - it needs to have at least one byte to be a proper
|
||||
// message. This is checked elsewhere for writes coming in directly from
|
||||
// the UDP socket.
|
||||
if n := buf.Len(); n < 1 {
|
||||
return fmt.Errorf("packet too short (%d bytes) %s", n, LogAddress(addr))
|
||||
}
|
||||
|
||||
// Inject the packet.
|
||||
t.packetCh <- &Packet{
|
||||
Buf: buf.Bytes(),
|
||||
From: addr,
|
||||
Timestamp: now,
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// See Transport.
|
||||
func (t *NetTransport) DialTimeout(addr string, timeout time.Duration) (net.Conn, error) {
|
||||
a := Address{Addr: addr, Name: ""}
|
||||
return t.DialAddressTimeout(a, timeout)
|
||||
}
|
||||
|
||||
// See NodeAwareTransport.
|
||||
func (t *NetTransport) DialAddressTimeout(a Address, timeout time.Duration) (net.Conn, error) {
|
||||
addr := a.Addr
|
||||
|
||||
dialer := net.Dialer{Timeout: timeout}
|
||||
return dialer.Dial("tcp", addr)
|
||||
}
|
||||
@@ -199,6 +247,12 @@ func (t *NetTransport) StreamCh() <-chan net.Conn {
|
||||
return t.streamCh
|
||||
}
|
||||
|
||||
// See IngestionAwareTransport.
|
||||
func (t *NetTransport) IngestStream(conn net.Conn) error {
|
||||
t.streamCh <- conn
|
||||
return nil
|
||||
}
|
||||
|
||||
// See Transport.
|
||||
func (t *NetTransport) Shutdown() error {
|
||||
// This will avoid log spam about errors when we shut down.
|
||||
|
||||
5
vendor/github.com/hashicorp/memberlist/security.go
сгенерированный
поставляемый
5
vendor/github.com/hashicorp/memberlist/security.go
сгенерированный
поставляемый
@@ -106,7 +106,10 @@ func encryptPayload(vsn encryptionVersion, key []byte, msg []byte, data []byte,
|
||||
dst.WriteByte(byte(vsn))
|
||||
|
||||
// Add a random nonce
|
||||
io.CopyN(dst, rand.Reader, nonceSize)
|
||||
_, err = io.CopyN(dst, rand.Reader, nonceSize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
afterNonce := dst.Len()
|
||||
|
||||
// Ensure we are correctly padded (only version 0)
|
||||
|
||||
173
vendor/github.com/hashicorp/memberlist/state.go
сгенерированный
поставляемый
173
vendor/github.com/hashicorp/memberlist/state.go
сгенерированный
поставляемый
@@ -13,26 +13,28 @@ import (
|
||||
metrics "github.com/armon/go-metrics"
|
||||
)
|
||||
|
||||
type nodeStateType int
|
||||
type NodeStateType int
|
||||
|
||||
const (
|
||||
stateAlive nodeStateType = iota
|
||||
stateSuspect
|
||||
stateDead
|
||||
StateAlive NodeStateType = iota
|
||||
StateSuspect
|
||||
StateDead
|
||||
StateLeft
|
||||
)
|
||||
|
||||
// Node represents a node in the cluster.
|
||||
type Node struct {
|
||||
Name string
|
||||
Addr net.IP
|
||||
Port uint16
|
||||
Meta []byte // Metadata from the delegate for this node.
|
||||
PMin uint8 // Minimum protocol version this understands
|
||||
PMax uint8 // Maximum protocol version this understands
|
||||
PCur uint8 // Current version node is speaking
|
||||
DMin uint8 // Min protocol version for the delegate to understand
|
||||
DMax uint8 // Max protocol version for the delegate to understand
|
||||
DCur uint8 // Current version delegate is speaking
|
||||
Name string
|
||||
Addr net.IP
|
||||
Port uint16
|
||||
Meta []byte // Metadata from the delegate for this node.
|
||||
State NodeStateType // State of the node.
|
||||
PMin uint8 // Minimum protocol version this understands
|
||||
PMax uint8 // Maximum protocol version this understands
|
||||
PCur uint8 // Current version node is speaking
|
||||
DMin uint8 // Min protocol version for the delegate to understand
|
||||
DMax uint8 // Max protocol version for the delegate to understand
|
||||
DCur uint8 // Current version delegate is speaking
|
||||
}
|
||||
|
||||
// Address returns the host:port form of a node's address, suitable for use
|
||||
@@ -41,6 +43,15 @@ func (n *Node) Address() string {
|
||||
return joinHostPort(n.Addr.String(), n.Port)
|
||||
}
|
||||
|
||||
// FullAddress returns the node name and host:port form of a node's address,
|
||||
// suitable for use with a transport.
|
||||
func (n *Node) FullAddress() Address {
|
||||
return Address{
|
||||
Addr: joinHostPort(n.Addr.String(), n.Port),
|
||||
Name: n.Name,
|
||||
}
|
||||
}
|
||||
|
||||
// String returns the node name
|
||||
func (n *Node) String() string {
|
||||
return n.Name
|
||||
@@ -50,7 +61,7 @@ func (n *Node) String() string {
|
||||
type nodeState struct {
|
||||
Node
|
||||
Incarnation uint32 // Last known incarnation number
|
||||
State nodeStateType // Current state
|
||||
State NodeStateType // Current state
|
||||
StateChange time.Time // Time last state change happened
|
||||
}
|
||||
|
||||
@@ -60,6 +71,16 @@ func (n *nodeState) Address() string {
|
||||
return n.Node.Address()
|
||||
}
|
||||
|
||||
// FullAddress returns the node name and host:port form of a node's address,
|
||||
// suitable for use with a transport.
|
||||
func (n *nodeState) FullAddress() Address {
|
||||
return n.Node.FullAddress()
|
||||
}
|
||||
|
||||
func (n *nodeState) DeadOrLeft() bool {
|
||||
return n.State == StateDead || n.State == StateLeft
|
||||
}
|
||||
|
||||
// ackHandler is used to register handlers for incoming acks and nacks.
|
||||
type ackHandler struct {
|
||||
ackFn func([]byte, time.Time)
|
||||
@@ -218,7 +239,7 @@ START:
|
||||
node = *m.nodes[m.probeIndex]
|
||||
if node.Name == m.config.Name {
|
||||
skip = true
|
||||
} else if node.State == stateDead {
|
||||
} else if node.DeadOrLeft() {
|
||||
skip = true
|
||||
}
|
||||
|
||||
@@ -271,7 +292,14 @@ func (m *Memberlist) probeNode(node *nodeState) {
|
||||
}
|
||||
|
||||
// Prepare a ping message and setup an ack handler.
|
||||
ping := ping{SeqNo: m.nextSeqNo(), Node: node.Name}
|
||||
selfAddr, selfPort := m.getAdvertise()
|
||||
ping := ping{
|
||||
SeqNo: m.nextSeqNo(),
|
||||
Node: node.Name,
|
||||
SourceAddr: selfAddr,
|
||||
SourcePort: selfPort,
|
||||
SourceNode: m.config.Name,
|
||||
}
|
||||
ackCh := make(chan ackMessage, m.config.IndirectChecks+1)
|
||||
nackCh := make(chan struct{}, m.config.IndirectChecks+1)
|
||||
m.setProbeChannels(ping.SeqNo, ackCh, nackCh, probeInterval)
|
||||
@@ -294,8 +322,8 @@ func (m *Memberlist) probeNode(node *nodeState) {
|
||||
defer func() {
|
||||
m.awareness.ApplyDelta(awarenessDelta)
|
||||
}()
|
||||
if node.State == stateAlive {
|
||||
if err := m.encodeAndSendMsg(addr, pingMsg, &ping); err != nil {
|
||||
if node.State == StateAlive {
|
||||
if err := m.encodeAndSendMsg(node.FullAddress(), pingMsg, &ping); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send ping: %s", err)
|
||||
if failedRemote(err) {
|
||||
goto HANDLE_REMOTE_FAILURE
|
||||
@@ -320,7 +348,7 @@ func (m *Memberlist) probeNode(node *nodeState) {
|
||||
}
|
||||
|
||||
compound := makeCompoundMessage(msgs)
|
||||
if err := m.rawSendMsgPacket(addr, &node.Node, compound.Bytes()); err != nil {
|
||||
if err := m.rawSendMsgPacket(node.FullAddress(), &node.Node, compound.Bytes()); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send compound ping and suspect message to %s: %s", addr, err)
|
||||
if failedRemote(err) {
|
||||
goto HANDLE_REMOTE_FAILURE
|
||||
@@ -369,13 +397,22 @@ HANDLE_REMOTE_FAILURE:
|
||||
kNodes := kRandomNodes(m.config.IndirectChecks, m.nodes, func(n *nodeState) bool {
|
||||
return n.Name == m.config.Name ||
|
||||
n.Name == node.Name ||
|
||||
n.State != stateAlive
|
||||
n.State != StateAlive
|
||||
})
|
||||
m.nodeLock.RUnlock()
|
||||
|
||||
// Attempt an indirect ping.
|
||||
expectedNacks := 0
|
||||
ind := indirectPingReq{SeqNo: ping.SeqNo, Target: node.Addr, Port: node.Port, Node: node.Name}
|
||||
selfAddr, selfPort = m.getAdvertise()
|
||||
ind := indirectPingReq{
|
||||
SeqNo: ping.SeqNo,
|
||||
Target: node.Addr,
|
||||
Port: node.Port,
|
||||
Node: node.Name,
|
||||
SourceAddr: selfAddr,
|
||||
SourcePort: selfPort,
|
||||
SourceNode: m.config.Name,
|
||||
}
|
||||
for _, peer := range kNodes {
|
||||
// We only expect nack to be sent from peers who understand
|
||||
// version 4 of the protocol.
|
||||
@@ -383,7 +420,7 @@ HANDLE_REMOTE_FAILURE:
|
||||
expectedNacks++
|
||||
}
|
||||
|
||||
if err := m.encodeAndSendMsg(peer.Address(), indirectPingMsg, &ind); err != nil {
|
||||
if err := m.encodeAndSendMsg(peer.FullAddress(), indirectPingMsg, &ind); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send indirect ping: %s", err)
|
||||
}
|
||||
}
|
||||
@@ -399,10 +436,13 @@ HANDLE_REMOTE_FAILURE:
|
||||
// which protocol version we are speaking. That's why we've included a
|
||||
// config option to turn this off if desired.
|
||||
fallbackCh := make(chan bool, 1)
|
||||
if (!m.config.DisableTcpPings) && (node.PMax >= 3) {
|
||||
|
||||
disableTcpPings := m.config.DisableTcpPings ||
|
||||
(m.config.DisableTcpPingsForNode != nil && m.config.DisableTcpPingsForNode(node.Name))
|
||||
if (!disableTcpPings) && (node.PMax >= 3) {
|
||||
go func() {
|
||||
defer close(fallbackCh)
|
||||
didContact, err := m.sendPingAndWaitForAck(node.Address(), ping, deadline)
|
||||
didContact, err := m.sendPingAndWaitForAck(node.FullAddress(), ping, deadline)
|
||||
if err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed fallback ping: %s", err)
|
||||
} else {
|
||||
@@ -459,12 +499,21 @@ HANDLE_REMOTE_FAILURE:
|
||||
// Ping initiates a ping to the node with the specified name.
|
||||
func (m *Memberlist) Ping(node string, addr net.Addr) (time.Duration, error) {
|
||||
// Prepare a ping message and setup an ack handler.
|
||||
ping := ping{SeqNo: m.nextSeqNo(), Node: node}
|
||||
selfAddr, selfPort := m.getAdvertise()
|
||||
ping := ping{
|
||||
SeqNo: m.nextSeqNo(),
|
||||
Node: node,
|
||||
SourceAddr: selfAddr,
|
||||
SourcePort: selfPort,
|
||||
SourceNode: m.config.Name,
|
||||
}
|
||||
ackCh := make(chan ackMessage, m.config.IndirectChecks+1)
|
||||
m.setProbeChannels(ping.SeqNo, ackCh, nil, m.config.ProbeInterval)
|
||||
|
||||
a := Address{Addr: addr.String(), Name: node}
|
||||
|
||||
// Send a ping to the node.
|
||||
if err := m.encodeAndSendMsg(addr.String(), pingMsg, &ping); err != nil {
|
||||
if err := m.encodeAndSendMsg(a, pingMsg, &ping); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
@@ -525,10 +574,10 @@ func (m *Memberlist) gossip() {
|
||||
}
|
||||
|
||||
switch n.State {
|
||||
case stateAlive, stateSuspect:
|
||||
case StateAlive, StateSuspect:
|
||||
return false
|
||||
|
||||
case stateDead:
|
||||
case StateDead:
|
||||
return time.Since(n.StateChange) > m.config.GossipToTheDeadTime
|
||||
|
||||
default:
|
||||
@@ -553,13 +602,13 @@ func (m *Memberlist) gossip() {
|
||||
addr := node.Address()
|
||||
if len(msgs) == 1 {
|
||||
// Send single message as is
|
||||
if err := m.rawSendMsgPacket(addr, &node.Node, msgs[0]); err != nil {
|
||||
if err := m.rawSendMsgPacket(node.FullAddress(), &node.Node, msgs[0]); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", addr, err)
|
||||
}
|
||||
} else {
|
||||
// Otherwise create and send a compound message
|
||||
compound := makeCompoundMessage(msgs)
|
||||
if err := m.rawSendMsgPacket(addr, &node.Node, compound.Bytes()); err != nil {
|
||||
if err := m.rawSendMsgPacket(node.FullAddress(), &node.Node, compound.Bytes()); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", addr, err)
|
||||
}
|
||||
}
|
||||
@@ -575,7 +624,7 @@ func (m *Memberlist) pushPull() {
|
||||
m.nodeLock.RLock()
|
||||
nodes := kRandomNodes(1, m.nodes, func(n *nodeState) bool {
|
||||
return n.Name == m.config.Name ||
|
||||
n.State != stateAlive
|
||||
n.State != StateAlive
|
||||
})
|
||||
m.nodeLock.RUnlock()
|
||||
|
||||
@@ -586,17 +635,17 @@ func (m *Memberlist) pushPull() {
|
||||
node := nodes[0]
|
||||
|
||||
// Attempt a push pull
|
||||
if err := m.pushPullNode(node.Address(), false); err != nil {
|
||||
if err := m.pushPullNode(node.FullAddress(), false); err != nil {
|
||||
m.logger.Printf("[ERR] memberlist: Push/Pull with %s failed: %s", node.Name, err)
|
||||
}
|
||||
}
|
||||
|
||||
// pushPullNode does a complete state exchange with a specific node.
|
||||
func (m *Memberlist) pushPullNode(addr string, join bool) error {
|
||||
func (m *Memberlist) pushPullNode(a Address, join bool) error {
|
||||
defer metrics.MeasureSince([]string{"memberlist", "pushPullNode"}, time.Now())
|
||||
|
||||
// Attempt to send and receive with the node
|
||||
remote, userState, err := m.sendAndReceiveState(addr, join)
|
||||
remote, userState, err := m.sendAndReceiveState(a, join)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -633,7 +682,7 @@ func (m *Memberlist) verifyProtocol(remote []pushNodeState) error {
|
||||
|
||||
for _, rn := range remote {
|
||||
// If the node isn't alive, then skip it
|
||||
if rn.State != stateAlive {
|
||||
if rn.State != StateAlive {
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -662,7 +711,7 @@ func (m *Memberlist) verifyProtocol(remote []pushNodeState) error {
|
||||
|
||||
for _, n := range m.nodes {
|
||||
// Ignore non-alive nodes
|
||||
if n.State != stateAlive {
|
||||
if n.State != StateAlive {
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -921,6 +970,11 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) {
|
||||
// store this node in our node map.
|
||||
var updatesNode bool
|
||||
if !ok {
|
||||
errCon := m.config.IPAllowed(a.Addr)
|
||||
if errCon != nil {
|
||||
m.logger.Printf("[WARN] memberlist: Rejected node %s (%v): %s", a.Node, net.IP(a.Addr), errCon)
|
||||
return
|
||||
}
|
||||
state = &nodeState{
|
||||
Node: Node{
|
||||
Name: a.Node,
|
||||
@@ -928,7 +982,7 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) {
|
||||
Port: a.Port,
|
||||
Meta: a.Meta,
|
||||
},
|
||||
State: stateDead,
|
||||
State: StateDead,
|
||||
}
|
||||
if len(a.Vsn) > 5 {
|
||||
state.PMin = a.Vsn[0]
|
||||
@@ -958,13 +1012,18 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) {
|
||||
} else {
|
||||
// Check if this address is different than the existing node unless the old node is dead.
|
||||
if !bytes.Equal([]byte(state.Addr), a.Addr) || state.Port != a.Port {
|
||||
errCon := m.config.IPAllowed(a.Addr)
|
||||
if errCon != nil {
|
||||
m.logger.Printf("[WARN] memberlist: Rejected IP update from %v to %v for node %s: %s", a.Node, state.Addr, net.IP(a.Addr), errCon)
|
||||
return
|
||||
}
|
||||
// If DeadNodeReclaimTime is configured, check if enough time has elapsed since the node died.
|
||||
canReclaim := (m.config.DeadNodeReclaimTime > 0 &&
|
||||
time.Since(state.StateChange) > m.config.DeadNodeReclaimTime)
|
||||
|
||||
// Allow the address to be updated if a dead node is being replaced.
|
||||
if state.State == stateDead && canReclaim {
|
||||
m.logger.Printf("[INFO] memberlist: Updating address for failed node %s from %v:%d to %v:%d",
|
||||
if state.State == StateLeft || (state.State == StateDead && canReclaim) {
|
||||
m.logger.Printf("[INFO] memberlist: Updating address for left or failed node %s from %v:%d to %v:%d",
|
||||
state.Name, state.Addr, state.Port, net.IP(a.Addr), a.Port)
|
||||
updatesNode = true
|
||||
} else {
|
||||
@@ -1048,8 +1107,8 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) {
|
||||
state.Meta = a.Meta
|
||||
state.Addr = a.Addr
|
||||
state.Port = a.Port
|
||||
if state.State != stateAlive {
|
||||
state.State = stateAlive
|
||||
if state.State != StateAlive {
|
||||
state.State = StateAlive
|
||||
state.StateChange = time.Now()
|
||||
}
|
||||
}
|
||||
@@ -1059,8 +1118,8 @@ func (m *Memberlist) aliveNode(a *alive, notify chan struct{}, bootstrap bool) {
|
||||
|
||||
// Notify the delegate of any relevant updates
|
||||
if m.config.Events != nil {
|
||||
if oldState == stateDead {
|
||||
// if Dead -> Alive, notify of join
|
||||
if oldState == StateDead || oldState == StateLeft {
|
||||
// if Dead/Left -> Alive, notify of join
|
||||
m.config.Events.NotifyJoin(&state.Node)
|
||||
|
||||
} else if !bytes.Equal(oldMeta, state.Meta) {
|
||||
@@ -1099,7 +1158,7 @@ func (m *Memberlist) suspectNode(s *suspect) {
|
||||
}
|
||||
|
||||
// Ignore non-alive nodes
|
||||
if state.State != stateAlive {
|
||||
if state.State != StateAlive {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -1117,7 +1176,7 @@ func (m *Memberlist) suspectNode(s *suspect) {
|
||||
|
||||
// Update the state
|
||||
state.Incarnation = s.Incarnation
|
||||
state.State = stateSuspect
|
||||
state.State = StateSuspect
|
||||
changeTime := time.Now()
|
||||
state.StateChange = changeTime
|
||||
|
||||
@@ -1141,7 +1200,7 @@ func (m *Memberlist) suspectNode(s *suspect) {
|
||||
fn := func(numConfirmations int) {
|
||||
m.nodeLock.Lock()
|
||||
state, ok := m.nodeMap[s.Node]
|
||||
timeout := ok && state.State == stateSuspect && state.StateChange == changeTime
|
||||
timeout := ok && state.State == StateSuspect && state.StateChange == changeTime
|
||||
m.nodeLock.Unlock()
|
||||
|
||||
if timeout {
|
||||
@@ -1179,7 +1238,7 @@ func (m *Memberlist) deadNode(d *dead) {
|
||||
delete(m.nodeTimers, d.Node)
|
||||
|
||||
// Ignore if node is already dead
|
||||
if state.State == stateDead {
|
||||
if state.DeadOrLeft() {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -1203,7 +1262,14 @@ func (m *Memberlist) deadNode(d *dead) {
|
||||
|
||||
// Update the state
|
||||
state.Incarnation = d.Incarnation
|
||||
state.State = stateDead
|
||||
|
||||
// If the dead message was send by the node itself, mark it is left
|
||||
// instead of dead.
|
||||
if d.Node == d.From {
|
||||
state.State = StateLeft
|
||||
} else {
|
||||
state.State = StateDead
|
||||
}
|
||||
state.StateChange = time.Now()
|
||||
|
||||
// Notify of death
|
||||
@@ -1217,7 +1283,7 @@ func (m *Memberlist) deadNode(d *dead) {
|
||||
func (m *Memberlist) mergeState(remote []pushNodeState) {
|
||||
for _, r := range remote {
|
||||
switch r.State {
|
||||
case stateAlive:
|
||||
case StateAlive:
|
||||
a := alive{
|
||||
Incarnation: r.Incarnation,
|
||||
Node: r.Name,
|
||||
@@ -1228,11 +1294,14 @@ func (m *Memberlist) mergeState(remote []pushNodeState) {
|
||||
}
|
||||
m.aliveNode(&a, nil, false)
|
||||
|
||||
case stateDead:
|
||||
case StateLeft:
|
||||
d := dead{Incarnation: r.Incarnation, Node: r.Name, From: r.Name}
|
||||
m.deadNode(&d)
|
||||
case StateDead:
|
||||
// If the remote node believes a node is dead, we prefer to
|
||||
// suspect that node instead of declaring it dead instantly
|
||||
fallthrough
|
||||
case stateSuspect:
|
||||
case StateSuspect:
|
||||
s := suspect{Incarnation: r.Incarnation, Node: r.Name, From: m.config.Name}
|
||||
m.suspectNode(&s)
|
||||
}
|
||||
|
||||
62
vendor/github.com/hashicorp/memberlist/transport.go
сгенерированный
поставляемый
62
vendor/github.com/hashicorp/memberlist/transport.go
сгенерированный
поставляемый
@@ -1,6 +1,7 @@
|
||||
package memberlist
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"time"
|
||||
)
|
||||
@@ -63,3 +64,64 @@ type Transport interface {
|
||||
// transport a chance to clean up any listeners.
|
||||
Shutdown() error
|
||||
}
|
||||
|
||||
type Address struct {
|
||||
// Addr is a network address as a string, similar to Dial. This usually is
|
||||
// in the form of "host:port". This is required.
|
||||
Addr string
|
||||
|
||||
// Name is the name of the node being addressed. This is optional but
|
||||
// transports may require it.
|
||||
Name string
|
||||
}
|
||||
|
||||
func (a *Address) String() string {
|
||||
if a.Name != "" {
|
||||
return fmt.Sprintf("%s (%s)", a.Name, a.Addr)
|
||||
}
|
||||
return a.Addr
|
||||
}
|
||||
|
||||
type IngestionAwareTransport interface {
|
||||
Transport
|
||||
// IngestPacket pulls a single packet off the conn, and only closes it if shouldClose is true.
|
||||
IngestPacket(conn net.Conn, addr net.Addr, now time.Time, shouldClose bool) error
|
||||
// IngestStream hands off the conn to the transport and doesn't close it.
|
||||
IngestStream(conn net.Conn) error
|
||||
}
|
||||
|
||||
type NodeAwareTransport interface {
|
||||
IngestionAwareTransport
|
||||
WriteToAddress(b []byte, addr Address) (time.Time, error)
|
||||
DialAddressTimeout(addr Address, timeout time.Duration) (net.Conn, error)
|
||||
}
|
||||
|
||||
type shimNodeAwareTransport struct {
|
||||
Transport
|
||||
}
|
||||
|
||||
var _ NodeAwareTransport = (*shimNodeAwareTransport)(nil)
|
||||
|
||||
func (t *shimNodeAwareTransport) IngestPacket(conn net.Conn, addr net.Addr, now time.Time, shouldClose bool) error {
|
||||
iat, ok := t.Transport.(IngestionAwareTransport)
|
||||
if !ok {
|
||||
panic("shimNodeAwareTransport does not support IngestPacket")
|
||||
}
|
||||
return iat.IngestPacket(conn, addr, now, shouldClose)
|
||||
}
|
||||
|
||||
func (t *shimNodeAwareTransport) IngestStream(conn net.Conn) error {
|
||||
iat, ok := t.Transport.(IngestionAwareTransport)
|
||||
if !ok {
|
||||
panic("shimNodeAwareTransport does not support IngestStream")
|
||||
}
|
||||
return iat.IngestStream(conn)
|
||||
}
|
||||
|
||||
func (t *shimNodeAwareTransport) WriteToAddress(b []byte, addr Address) (time.Time, error) {
|
||||
return t.WriteTo(b, addr.Addr)
|
||||
}
|
||||
|
||||
func (t *shimNodeAwareTransport) DialAddressTimeout(addr Address, timeout time.Duration) (net.Conn, error) {
|
||||
return t.DialTimeout(addr.Addr, timeout)
|
||||
}
|
||||
|
||||
2
vendor/github.com/hashicorp/memberlist/util.go
сгенерированный
поставляемый
2
vendor/github.com/hashicorp/memberlist/util.go
сгенерированный
поставляемый
@@ -102,7 +102,7 @@ func moveDeadNodes(nodes []*nodeState, gossipToTheDeadTime time.Duration) int {
|
||||
numDead := 0
|
||||
n := len(nodes)
|
||||
for i := 0; i < n-numDead; i++ {
|
||||
if nodes[i].State != stateDead {
|
||||
if nodes[i].State != StateDead {
|
||||
continue
|
||||
}
|
||||
|
||||
|
||||
Ссылка в новой задаче
Block a user