Add server discovery

go.mod

@@ -31,7 +31,9 @@ replace (
 )
 
 require (
+	github.com/hashicorp/go-discover v0.0.0-20201029210230-738cb3105cd0 // indirect
 	github.com/pkg/errors v0.9.1
+	github.com/rancher/dynamiclistener v0.3.1-0.20210616080009-9865ae859c7f // indirect
 	github.com/rancher/rancher/pkg/apis v0.0.0-20210616082234-54d4afc27c36
 	github.com/rancher/system-agent v0.0.1-alpha30
 	github.com/rancher/wharfie v0.3.2

pkg/config/types.go

@@ -38,10 +38,11 @@ var (
 
 type Config struct {
 	RuntimeConfig
-	KubernetesVersion string `json:"kubernetesVersion,omitempty"`
-	RancherVersion    string `json:"rancherVersion,omitempty"`
-	Server            string `json:"server,omitempty"`
-	Role              string `json:"role,omitempty"`
+	KubernetesVersion string            `json:"kubernetesVersion,omitempty"`
+	RancherVersion    string            `json:"rancherVersion,omitempty"`
+	Server            string            `json:"server,omitempty"`
+	Discovery         map[string]string `json:"discovery,omitempty"`
+	Role              string            `json:"role,omitempty"`
 
 	RancherValues      map[string]interface{} `json:"rancherValues,omitempty"`
 	PreInstructions    []plan.Instruction     `json:"preInstructions,omitempty"`

pkg/discovery/discovery.go

@@ -2,10 +2,197 @@ package discovery
 
 import (
 	"context"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"sort"
+	"sync"
+	"time"
 
+	"github.com/hashicorp/go-discover"
+	"github.com/rancher/dynamiclistener/server"
 	"github.com/rancher/rancherd/pkg/config"
+	"github.com/rancher/wrangler/pkg/data/convert"
+	"github.com/rancher/wrangler/pkg/randomtoken"
+	"github.com/rancher/wrangler/pkg/slice"
+	"github.com/sirupsen/logrus"
+
+	// Include kubernetes provider
+	_ "github.com/hashicorp/go-discover/provider/k8s"
+)
+
+var (
+	insecureHTTPClient = http.Client{
+		Timeout: 10 * time.Second,
+		Transport: &http.Transport{
+			Proxy:               http.ProxyFromEnvironment,
+			TLSHandshakeTimeout: 5 * time.Second,
+			TLSClientConfig: &tls.Config{
+				InsecureSkipVerify: true,
+			},
+		},
+	}
 )
 
-func FindServer(ctx context.Context, cfg *config.Config) (string, error) {
-	return cfg.Server, nil
+func DiscoverServerAndRole(ctx context.Context, cfg *config.Config) error {
+	if len(cfg.Discovery) == 0 {
+		return nil
+	}
+
+	server, clusterInit, err := discoverServerAndRole(ctx, cfg)
+	if err != nil {
+		return err
+	}
+	if clusterInit {
+		cfg.Role = "cluster-init"
+	} else if server != "" {
+		cfg.Server = server
+	}
+	return nil
+
+}
+func discoverServerAndRole(ctx context.Context, cfg *config.Config) (string, bool, error) {
+	discovery, err := discover.New()
+	if err != nil {
+		return "", false, err
+	}
+
+	port, err := convert.ToNumber(cfg.RancherValues["hostPort"])
+	if err != nil || port == 0 {
+		port = 8443
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	server, err := NewJoinServer(ctx, port)
+	if err != nil {
+		return "", false, err
+	}
+
+	for {
+		server, clusterInit := server.loop(ctx, cfg.Discovery, port, discovery)
+		if clusterInit {
+			return "", true, nil
+		}
+		if server != "" {
+			return server, false, nil
+		}
+		logrus.Info("Waiting to discover server")
+		select {
+		case <-ctx.Done():
+			return "", false, fmt.Errorf("interrupted waiting to discover server: %w", ctx.Err())
+		case <-time.After(5 * time.Second):
+		}
+	}
+}
+
+func (j *joinServer) loop(ctx context.Context, params map[string]string, port int64, discovery *discover.Discover) (string, bool) {
+	addrs, err := discovery.Addrs(discover.Config(params).String(), log.Default())
+	if err != nil {
+		logrus.Errorf("failed to discover peers to: %v", err)
+		return "", false
+	}
+
+	sort.Strings(addrs)
+	j.setPeers(addrs)
+
+	var (
+		allAgree = true
+		firstID  = ""
+	)
+	for i, addr := range addrs {
+		url := fmt.Sprintf("https://%s:%d/cacerts", addr, port)
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+		if err != nil {
+			logrus.Errorf("failed to construct request for %s: %v", url, err)
+			allAgree = false
+			return "", false
+		}
+		resp, err := insecureHTTPClient.Do(req)
+		if err != nil {
+			logrus.Errorf("failed to connect to %s: %v", url, err)
+			allAgree = false
+			continue
+		}
+
+		data, err := ioutil.ReadAll(resp.Body)
+		resp.Body.Close()
+		if err != nil || resp.StatusCode != http.StatusOK {
+			logrus.Errorf("failed to read response from %s: code %d: %v", url, resp.StatusCode, err)
+			allAgree = false
+			continue
+		}
+
+		rancherID := resp.Header.Get("X-Cattle-Rancherd-Id")
+		if rancherID == "" {
+			return fmt.Sprintf("https://%s:%d", addr, port), false
+		}
+		if i == 0 {
+			firstID = rancherID
+		}
+
+		var pingResponse pingResponse
+		if err := json.Unmarshal(data, &pingResponse); err != nil {
+			logrus.Errorf("failed to unmarshal response (%s) from %s: %v", data, url, err)
+			allAgree = false
+			continue
+		}
+
+		if !slice.StringsEqual(addrs, pingResponse.Peers) {
+			logrus.Infof("Peer %s does not agree on peer list, %v != %v", addr, addrs, pingResponse.Peers)
+			allAgree = false
+			continue
+		}
+	}
+
+	if allAgree && len(addrs) > 2 && firstID == j.id {
+		return "", true
+	}
+
+	return "", false
+}
+
+type joinServer struct {
+	lock  sync.Mutex
+	id    string
+	peers []string
+}
+
+type pingResponse struct {
+	Peers []string `json:"peers,omitempty"`
+}
+
+func NewJoinServer(ctx context.Context, port int64) (*joinServer, error) {
+	id, err := randomtoken.Generate()
+	if err != nil {
+		return nil, err
+	}
+
+	j := &joinServer{
+		id: id,
+	}
+
+	return j, server.ListenAndServe(ctx, int(port), 0, j, nil)
+}
+
+func (j *joinServer) setPeers(peers []string) {
+	j.lock.Lock()
+	defer j.lock.Unlock()
+	logrus.Infof("current set of peers: %v", peers)
+	j.peers = peers
+}
+
+func (j *joinServer) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
+	j.lock.Lock()
+	defer j.lock.Unlock()
+
+	rw.Header().Set("X-Cattle-Rancherd-Id", j.id)
+	rw.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(rw).Encode(pingResponse{
+		Peers: j.peers,
+	})
 }

pkg/plan/bootstrap.go

@@ -1,9 +1,11 @@
 package plan
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/rancher/rancherd/pkg/config"
+	"github.com/rancher/rancherd/pkg/discovery"
 	"github.com/rancher/rancherd/pkg/join"
 	"github.com/rancher/rancherd/pkg/probe"
 	"github.com/rancher/rancherd/pkg/rancher"
@@ -62,11 +64,15 @@ func toJoinPlan(cfg *config.Config, dataDir string) (*applyinator.Plan, error) {
 	return (*applyinator.Plan)(&plan), nil
 }
 
-func ToPlan(config *config.Config, dataDir string) (*applyinator.Plan, error) {
-	if config.Role == "cluster-init" {
-		return toInitPlan(config, dataDir)
+func ToPlan(ctx context.Context, config *config.Config, dataDir string) (*applyinator.Plan, error) {
+	newCfg := *config
+	if err := discovery.DiscoverServerAndRole(ctx, &newCfg); err != nil {
+		return nil, err
+	}
+	if newCfg.Role == "cluster-init" {
+		return toInitPlan(&newCfg, dataDir)
 	}
-	return toJoinPlan(config, dataDir)
+	return toJoinPlan(&newCfg, dataDir)
 }
 
 func (p *plan) addInstructions(cfg *config.Config, dataDir string) error {

pkg/rancherd/rancher.go

@@ -57,7 +57,7 @@ func (r *Rancherd) execute(ctx context.Context) error {
 
 	logrus.Infof("Bootstrapping Rancher (%s/%s)", rancherVersion, k8sVersion)
 
-	nodePlan, err := plan.ToPlan(&cfg, r.cfg.DataDir)
+	nodePlan, err := plan.ToPlan(ctx, &cfg, r.cfg.DataDir)
 	if err != nil {
 		return fmt.Errorf("generating plan: %w", err)
 	}