func sendJoin(node *Node, opts *joinOpts) ([]string, error) {
	joiner, err := newJoinSender(node, opts)
	if err != nil {
		return nil, err
	}
	return joiner.JoinCluster()
}
SendJoin creates a new JoinSender and attempts to join the cluster defined by
the nodes bootstrap hosts
package swim
import (
	"errors"
	"fmt"
	"math/rand"
	"sync"
	"time"
	log "github.com/uber-common/bark"
	"github.com/uber/ringpop-go/logging"
	"github.com/uber/ringpop-go/shared"
	"github.com/uber/ringpop-go/util"
	"github.com/uber/tchannel-go/json"
)
const (
	defaultMaxJoinDuration   = 120 * time.Second
	defaultJoinTimeout       = time.Second
	defaultJoinSize          = 3
	defaultParallelismFactor = 2
)
var errJoinTimeout = errors.New("join timed out")
type joinRequest struct {
	App         string            `json:"app"`
	Source      string            `json:"source"`
	Incarnation int64             `json:"incarnationNumber"`
	Timeout     time.Duration     `json:"timeout"`
	Labels      map[string]string `json:"labels,omitempty"`
}
type joinOpts struct {
	timeout           time.Duration
	size              int
	maxJoinDuration   time.Duration
	parallelismFactor int
	delayer joinDelayer
}
type joinSender struct {
	node    *Node
	timeout time.Duration
	bootstrapHostsMap map[string][]string
	parallelismFactor int
	maxJoinDuration time.Duration
	potentialNodes    []string
	preferredNodes    []string
	nonPreferredNodes []string
	size int
	roundPotentialNodes    []string
	roundPreferredNodes    []string
	roundNonPreferredNodes []string
	numTries int
	delayer joinDelayer
	logger log.Logger
}
func newJoinSender(node *Node, opts *joinOpts) (*joinSender, error) {
	if opts == nil {
		opts = &joinOpts{}
	}
	if node.discoverProvider == nil {
		return nil, errors.New("no discover provider")
	}
	bootstrapHosts, err := node.discoverProvider.Hosts()
	if err != nil {
		return nil, err
	}
	if !util.StringInSlice(bootstrapHosts, node.Address()) {
		bootstrapHosts = append(bootstrapHosts, node.Address())
	}
	js := &joinSender{
		node:   node,
		logger: logging.Logger("join").WithField("local", node.Address()),
	}
	js.parseHosts(bootstrapHosts)
	js.potentialNodes = js.CollectPotentialNodes(nil)
	js.timeout = util.SelectDuration(opts.timeout, defaultJoinTimeout)
	js.maxJoinDuration = util.SelectDuration(opts.maxJoinDuration, defaultMaxJoinDuration)
	js.parallelismFactor = util.SelectInt(opts.parallelismFactor, defaultParallelismFactor)
	js.size = util.SelectInt(opts.size, defaultJoinSize)
	js.size = util.Min(js.size, len(js.potentialNodes))
	js.delayer = opts.delayer
	if js.delayer == nil {
		js.delayer, err = newExponentialDelayer(js.node.address, nil)
		if err != nil {
			return nil, err
		}
	}
	return js, nil
}
func (j *joinSender) parseHosts(hostports []string) {
	j.bootstrapHostsMap = util.HostPortsByHost(hostports)
	err := util.CheckLocalMissing(j.node.address, j.bootstrapHostsMap[util.CaptureHost(j.node.address)])
	if err != nil {
		j.logger.Warn(err.Error())
	}
	mismatched, err := util.CheckHostnameIPMismatch(j.node.address, j.bootstrapHostsMap)
	if err != nil {
		j.logger.WithField("mismatched", mismatched).Warn(err.Error())
	}
}
func (j *joinSender) CollectPotentialNodes(nodesJoined []string) []string {
	if nodesJoined == nil {
		nodesJoined = make([]string, 0)
	}
	var potentialNodes []string
	for _, hostports := range j.bootstrapHostsMap {
		for _, hostport := range hostports {
			if j.node.address != hostport && !util.StringInSlice(nodesJoined, hostport) {
				potentialNodes = append(potentialNodes, hostport)
			}
		}
	}
	return potentialNodes
}
func (j *joinSender) CollectPreferredNodes() []string {
	var preferredNodes []string
	for host, hostports := range j.bootstrapHostsMap {
		if host != util.CaptureHost(j.node.address) {
			preferredNodes = append(preferredNodes, hostports...)
		}
	}
	return preferredNodes
}
func (j *joinSender) CollectNonPreferredNodes() []string {
	if len(j.preferredNodes) == 0 {
		return j.potentialNodes
	}
	var nonPreferredNodes []string
	for _, host := range j.bootstrapHostsMap[util.CaptureHost(j.node.address)] {
		if host != j.node.address {
			nonPreferredNodes = append(nonPreferredNodes, host)
		}
	}
	return nonPreferredNodes
}
func (j *joinSender) Init(nodesJoined []string) {
	rand.Seed(time.Now().UnixNano())
	j.potentialNodes = j.CollectPotentialNodes(nodesJoined)
	j.preferredNodes = j.CollectPreferredNodes()
	j.nonPreferredNodes = j.CollectNonPreferredNodes()
	j.roundPotentialNodes = append(j.roundPotentialNodes, j.potentialNodes...)
	j.roundPreferredNodes = append(j.roundPreferredNodes, j.preferredNodes...)
	j.roundNonPreferredNodes = append(j.roundNonPreferredNodes, j.nonPreferredNodes...)
}
func (j *joinSender) SelectGroup(nodesJoined []string) []string {
	var group []string
	if len(j.roundPreferredNodes) == 0 && len(j.roundNonPreferredNodes) == 0 {
		j.Init(nodesJoined)
	}
	numNodesLeft := j.size - len(nodesJoined)
	cont := func() bool {
		if len(group) == numNodesLeft*j.parallelismFactor {
			return false
		}
		nodesAvailable := len(j.roundPreferredNodes) + len(j.roundNonPreferredNodes)
		if nodesAvailable == 0 {
			return false
		}
		return true
	}
	for cont() {
		if len(j.roundPreferredNodes) > 0 {
			group = append(group, util.TakeNode(&j.roundPreferredNodes, -1))
		} else if len(j.roundNonPreferredNodes) > 0 {
			group = append(group, util.TakeNode(&j.roundNonPreferredNodes, -1))
		}
	}
	return group
}
func (j *joinSender) JoinCluster() ([]string, error) {
	var nodesJoined []string
	var numGroups = 0
	var numJoined = 0
	var numFailed = 0
	var startTime = time.Now()
	if util.SingleNodeCluster(j.node.address, j.bootstrapHostsMap) {
		j.logger.Info("got single node cluster to join")
		return nodesJoined, nil
	}
	for {
		if j.node.Destroyed() {
			j.node.EmitEvent(JoinFailedEvent{
				Reason: Destroyed,
				Error:  nil,
			})
			return nil, errors.New("node destroyed while attempting to join cluster")
		}
		successes, failures := j.JoinGroup(nodesJoined)
		nodesJoined = append(nodesJoined, successes...)
		numJoined += len(successes)
		numFailed += len(failures)
		numGroups++
		if numJoined >= j.size {
			j.logger.WithFields(log.Fields{
				"joinSize":  j.size,
				"joinTime":  time.Now().Sub(startTime),
				"numJoined": numJoined,
				"numFailed": numFailed,
				"numGroups": numGroups,
			}).Debug("join complete")
			break
		}
		joinDuration := time.Now().Sub(startTime)
		if joinDuration > j.maxJoinDuration {
			j.logger.WithFields(log.Fields{
				"joinDuration":    joinDuration,
				"maxJoinDuration": j.maxJoinDuration,
				"numJoined":       numJoined,
				"numFailed":       numFailed,
				"startTime":       startTime,
			}).Warn("max join duration exceeded")
			err := fmt.Errorf("join duration of %v exceeded max %v",
				joinDuration, j.maxJoinDuration)
			j.node.EmitEvent(JoinFailedEvent{
				Reason: Error,
				Error:  err,
			})
			return nodesJoined, err
		}
		j.logger.WithFields(log.Fields{
			"joinSize":  j.size,
			"numJoined": numJoined,
			"numFailed": numFailed,
			"startTime": startTime,
		}).Debug("join not yet complete")
		j.delayer.delay()
	}
	j.node.EmitEvent(JoinCompleteEvent{
		Duration:  time.Now().Sub(startTime),
		NumJoined: numJoined,
		Joined:    nodesJoined,
	})
	return nodesJoined, nil
}
func (j *joinSender) JoinGroup(nodesJoined []string) ([]string, []string) {
	group := j.SelectGroup(nodesJoined)
	var responses struct {
		successes []string
		failures  []string
		sync.Mutex
	}
	var numNodesLeft = j.size - len(nodesJoined)
	var startTime = time.Now()
	var wg sync.WaitGroup
	j.numTries++
	j.node.EmitEvent(JoinTriesUpdateEvent{j.numTries})
	for _, target := range group {
		wg.Add(1)
		go func(target string) {
			defer wg.Done()
			res, err := sendJoinRequest(j.node, target, j.timeout)
			if err != nil {
				msg := "attempt to join node failed"
				if err == errJoinTimeout {
					msg = "attempt to join node timed out"
				}
				j.logger.WithFields(log.Fields{
					"remote":  target,
					"timeout": j.timeout,
				}).Debug(msg)
				responses.Lock()
				responses.failures = append(responses.failures, target)
				responses.Unlock()
				return
			}
			responses.Lock()
			responses.successes = append(responses.successes, target)
			responses.Unlock()
			start := time.Now()
			j.node.memberlist.AddJoinList(res.Membership)
			j.node.EmitEvent(AddJoinListEvent{
				Duration: time.Now().Sub(start),
			})
		}(target)
	}
	wg.Wait()
	j.logger.WithFields(log.Fields{
		"groupSize":    len(group),
		"joinSize":     j.size,
		"joinTime":     time.Now().Sub(startTime),
		"numNodesLeft": numNodesLeft,
		"numFailures":  len(responses.failures),
		"failures":     responses.failures,
		"numSuccesses": len(responses.successes),
		"successes":    responses.successes,
	}).Debug("join group complete")
	return responses.successes, responses.failures
}
func sendJoinRequest(node *Node, target string, timeout time.Duration) (*joinResponse, error) {
	ctx, cancel := shared.NewTChannelContext(timeout)
	defer cancel()
	peer := node.channel.Peers().GetOrAdd(target)
	req := joinRequest{
		App:         node.app,
		Source:      node.address,
		Incarnation: node.Incarnation(),
		Timeout:     timeout,
		Labels:      node.Labels().AsMap(),
	}
	res := &joinResponse{}
	errC := make(chan error, 1)
	go func() {
		errC <- json.CallPeer(ctx, peer, node.service, "/protocol/join", req, res)
	}()
	var err error
	select {
	case err = <-errC:
	case <-ctx.Done():
		err = errJoinTimeout
	}
	if err != nil {
		logging.Logger("join").WithFields(log.Fields{
			"local": node.Address(),
			"error": err,
		}).Debug("could not complete join")
		return nil, err
	}
	return res, err
}