Fix adding etcd-only node to existing cluster

Signed-off-by: Brad Davidson <[email protected]>

Author: brandond

pkg/agent/loadbalancer/loadbalancer.go

@@ -19,16 +19,17 @@ type LoadBalancer struct {
 	dialer *net.Dialer
 	proxy  *tcpproxy.Proxy
 
-	configFile            string
-	localAddress          string
-	localServerURL        string
-	originalServerAddress string
-	ServerURL             string
-	ServerAddresses       []string
-	randomServers         []string
-	currentServerAddress  string
-	nextServerIndex       int
-	Listener              net.Listener
+	serviceName          string
+	configFile           string
+	localAddress         string
+	localServerURL       string
+	defaultServerAddress string
+	ServerURL            string
+	ServerAddresses      []string
+	randomServers        []string
+	currentServerAddress string
+	nextServerIndex      int
+	Listener             net.Listener
 }
 
 const RandomPort = 0
@@ -55,26 +56,27 @@ func New(ctx context.Context, dataDir, serviceName, serverURL string, lbServerPo
 	}
 	localAddress := listener.Addr().String()
 
-	originalServerAddress, localServerURL, err := parseURL(serverURL, localAddress)
+	defaultServerAddress, localServerURL, err := parseURL(serverURL, localAddress)
 	if err != nil {
 		return nil, err
 	}
 
 	if serverURL == localServerURL {
-		logrus.Debugf("Initial server URL for load balancer points at local server URL - starting with empty original server address")
-		originalServerAddress = ""
+		logrus.Debugf("Initial server URL for load balancer %s points at local server URL - starting with empty default server address", serviceName)
+		defaultServerAddress = ""
 	}
 
 	lb := &LoadBalancer{
-		dialer:                &net.Dialer{},
-		configFile:            filepath.Join(dataDir, "etc", serviceName+".json"),
-		localAddress:          localAddress,
-		localServerURL:        localServerURL,
-		originalServerAddress: originalServerAddress,
-		ServerURL:             serverURL,
+		serviceName:          serviceName,
+		dialer:               &net.Dialer{},
+		configFile:           filepath.Join(dataDir, "etc", serviceName+".json"),
+		localAddress:         localAddress,
+		localServerURL:       localServerURL,
+		defaultServerAddress: defaultServerAddress,
+		ServerURL:            serverURL,
 	}
 
-	lb.setServers([]string{lb.originalServerAddress})
+	lb.setServers([]string{lb.defaultServerAddress})
 
 	lb.proxy = &tcpproxy.Proxy{
 		ListenFunc: func(string, string) (net.Listener, error) {
@@ -93,22 +95,27 @@ func New(ctx context.Context, dataDir, serviceName, serverURL string, lbServerPo
 	if err := lb.proxy.Start(); err != nil {
 		return nil, err
 	}
-	logrus.Infof("Running load balancer %s -> %v", lb.localAddress, lb.randomServers)
+	logrus.Infof("Running load balancer %s %s -> %v", serviceName, lb.localAddress, lb.randomServers)
 
 	return lb, nil
 }
 
+func (lb *LoadBalancer) SetDefault(serverAddress string) {
+	logrus.Infof("Updating load balancer %s default server address -> %s", lb.serviceName, serverAddress)
+	lb.defaultServerAddress = serverAddress
+}
+
 func (lb *LoadBalancer) Update(serverAddresses []string) {
 	if lb == nil {
 		return
 	}
 	if !lb.setServers(serverAddresses) {
 		return
 	}
-	logrus.Infof("Updating load balancer server addresses -> %v", lb.randomServers)
+	logrus.Infof("Updating load balancer %s server addresses -> %v", lb.serviceName, lb.randomServers)
 
 	if err := lb.writeConfig(); err != nil {
-		logrus.Warnf("Error updating load balancer config: %s", err)
+		logrus.Warnf("Error updating load balancer %s config: %s", lb.serviceName, err)
 	}
 }
 
@@ -128,14 +135,14 @@ func (lb *LoadBalancer) dialContext(ctx context.Context, network, address string
 		if err == nil {
 			return conn, nil
 		}
-		logrus.Debugf("Dial error from load balancer: %s", err)
+		logrus.Debugf("Dial error from load balancer %s: %s", lb.serviceName, err)
 
 		newServer, err := lb.nextServer(targetServer)
 		if err != nil {
 			return nil, err
 		}
 		if targetServer != newServer {
-			logrus.Debugf("Dial server in load balancer failed over to %s", newServer)
+			logrus.Debugf("Dial server in load balancer %s failed over to %s", lb.serviceName, newServer)
 		}
 		if ctx.Err() != nil {
 			return nil, ctx.Err()
@@ -156,7 +163,7 @@ func onDialError(src net.Conn, dstDialErr error) {
 	src.Close()
 }
 
-// ResetLoadBalancer will delete the local state file for the load balacner on disk
+// ResetLoadBalancer will delete the local state file for the load balancer on disk
 func ResetLoadBalancer(dataDir, serviceName string) error {
 	stateFile := filepath.Join(dataDir, "etc", serviceName+".json")
 	if err := os.Remove(stateFile); err != nil {

pkg/agent/loadbalancer/servers.go

@@ -7,7 +7,7 @@ import (
 )
 
 func (lb *LoadBalancer) setServers(serverAddresses []string) bool {
-	serverAddresses, hasOriginalServer := sortServers(serverAddresses, lb.originalServerAddress)
+	serverAddresses, hasOriginalServer := sortServers(serverAddresses, lb.defaultServerAddress)
 	if len(serverAddresses) == 0 {
 		return false
 	}
@@ -25,7 +25,7 @@ func (lb *LoadBalancer) setServers(serverAddresses []string) bool {
 		lb.randomServers[i], lb.randomServers[j] = lb.randomServers[j], lb.randomServers[i]
 	})
 	if !hasOriginalServer {
-		lb.randomServers = append(lb.randomServers, lb.originalServerAddress)
+		lb.randomServers = append(lb.randomServers, lb.defaultServerAddress)
 	}
 	lb.currentServerAddress = lb.randomServers[0]
 	lb.nextServerIndex = 1

pkg/agent/proxy/apiproxy.go

@@ -15,6 +15,7 @@ import (
 type Proxy interface {
 	Update(addresses []string)
 	SetAPIServerPort(ctx context.Context, port int) error
+	SetSupervisorDefault(address string)
 	SupervisorURL() string
 	SupervisorAddresses() []string
 	APIServerURL() string
@@ -135,6 +136,28 @@ func (p *proxy) SetAPIServerPort(ctx context.Context, port int) error {
 	return nil
 }
 
+// SetSupervisorDefault updates the default (fallback) address for the connection to the
+// supervisor. This is most useful on k3s nodes without apiservers, where the local
+// supervisor must be used to bootstrap the agent config, but then switched over to
+// another node running an apiserver once one is available.
+func (p *proxy) SetSupervisorDefault(address string) {
+	host, port, err := sysnet.SplitHostPort(address)
+	if err != nil {
+		logrus.Errorf("Failed to parse address %s, dropping: %v", address, err)
+		return
+	}
+	if p.apiServerEnabled {
+		port = p.supervisorPort
+		address = sysnet.JoinHostPort(host, port)
+	}
+	p.fallbackSupervisorAddress = address
+	if p.supervisorLB == nil {
+		p.supervisorURL = "https://" + address
+	} else {
+		p.supervisorLB.SetDefault(address)
+	}
+}
+
 func (p *proxy) SupervisorURL() string {
 	return p.supervisorURL
 }

pkg/agent/run.go

@@ -3,9 +3,10 @@ package agent
 import (
 	"context"
 	"fmt"
-	"net/url"
+	"net"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
@@ -350,17 +351,8 @@ func setupTunnelAndRunAgent(ctx context.Context, nodeConfig *daemonconfig.Node,
 			}
 			agentRan = true
 		}
-
-		select {
-		case address := <-cfg.APIAddressCh:
-			cfg.ServerURL = address
-			u, err := url.Parse(cfg.ServerURL)
-			if err != nil {
-				logrus.Warn(err)
-			}
-			proxy.Update([]string{fmt.Sprintf("%s:%d", u.Hostname(), nodeConfig.ServerHTTPSPort)})
-		case <-ctx.Done():
-			return ctx.Err()
+		if err := waitForAPIServerAddresses(ctx, nodeConfig, cfg, proxy); err != nil {
+			return err
 		}
 	} else if cfg.ClusterReset && proxy.IsAPIServerLBEnabled() {
 		// If we're doing a cluster-reset on RKE2, the kubelet needs to be started early to clean
@@ -379,3 +371,26 @@ func setupTunnelAndRunAgent(ctx context.Context, nodeConfig *daemonconfig.Node,
 	}
 	return nil
 }
+
+func waitForAPIServerAddresses(ctx context.Context, nodeConfig *daemonconfig.Node, cfg cmds.Agent, proxy proxy.Proxy) error {
+	for {
+		select {
+		case <-time.After(5 * time.Second):
+			logrus.Info("Waiting for apiserver addresses")
+		case addresses := <-cfg.APIAddressCh:
+			for i, a := range addresses {
+				host, _, err := net.SplitHostPort(a)
+				if err == nil {
+					addresses[i] = net.JoinHostPort(host, strconv.Itoa(nodeConfig.ServerHTTPSPort))
+					if i == 0 {
+						proxy.SetSupervisorDefault(addresses[i])
+					}
+				}
+			}
+			proxy.Update(addresses)
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}

pkg/cli/cmds/agent.go

@@ -14,7 +14,7 @@ type Agent struct {
 	TokenFile                string
 	ClusterSecret            string
 	ServerURL                string
-	APIAddressCh             chan string
+	APIAddressCh             chan []string
 	DisableLoadBalancer      bool
 	DisableServiceLB         bool
 	ETCDAgent                bool

pkg/cli/server/server.go

@@ -394,13 +394,6 @@ func run(app *cli.Context, cfg *cmds.Server, leaderControllers server.CustomCont
 		serverConfig.ControlConfig.DisableScheduler = true
 		serverConfig.ControlConfig.DisableCCM = true
 
-		// only close the agentReady channel in case of k3s restoration, because k3s does not start
-		// the agent until server returns successfully, unlike rke2's agent which starts in parallel
-		// with the server
-		if serverConfig.ControlConfig.SupervisorPort == serverConfig.ControlConfig.HTTPSPort {
-			close(agentReady)
-		}
-
 		dataDir, err := datadir.LocalHome(cfg.DataDir, false)
 		if err != nil {
 			return err
@@ -484,10 +477,12 @@ func run(app *cli.Context, cfg *cmds.Server, leaderControllers server.CustomCont
 	}
 
 	if serverConfig.ControlConfig.DisableAPIServer {
+		if cfg.ServerURL != "" {
+			agentConfig.ServerURL = cfg.ServerURL
+		}
 		// initialize the apiAddress Channel for receiving the api address from etcd
-		agentConfig.APIAddressCh = make(chan string, 1)
-		setAPIAddressChannel(ctx, &serverConfig, &agentConfig)
-		defer close(agentConfig.APIAddressCh)
+		agentConfig.APIAddressCh = make(chan []string)
+		go getAPIAddressFromEtcd(ctx, serverConfig, agentConfig)
 	}
 	return agent.Run(ctx, agentConfig)
 }
@@ -533,29 +528,19 @@ func getArgValueFromList(searchArg string, argList []string) string {
 	return value
 }
 
-// setAPIAddressChannel will try to get the api address key from etcd and when it succeed it will
-// set the APIAddressCh channel with its value, the function works for both k3s and rke2 in case
-// of k3s we block returning back to the agent.Run until we get the api address, however in rke2
-// the code will not block operation and will run the operation in a goroutine
-func setAPIAddressChannel(ctx context.Context, serverConfig *server.Config, agentConfig *cmds.Agent) {
-	// start a goroutine to check for the server ip if set from etcd in case of rke2
-	if serverConfig.ControlConfig.HTTPSPort != serverConfig.ControlConfig.SupervisorPort {
-		go getAPIAddressFromEtcd(ctx, serverConfig, agentConfig)
-		return
-	}
-	getAPIAddressFromEtcd(ctx, serverConfig, agentConfig)
-}
-
-func getAPIAddressFromEtcd(ctx context.Context, serverConfig *server.Config, agentConfig *cmds.Agent) {
-	t := time.NewTicker(5 * time.Second)
-	defer t.Stop()
-	for range t.C {
-		serverAddress, err := etcd.GetAPIServerURLFromETCD(ctx, &serverConfig.ControlConfig)
-		if err == nil {
-			agentConfig.ServerURL = "https://" + serverAddress
-			agentConfig.APIAddressCh <- agentConfig.ServerURL
+func getAPIAddressFromEtcd(ctx context.Context, serverConfig server.Config, agentConfig cmds.Agent) {
+	defer close(agentConfig.APIAddressCh)
+	for {
+		toCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		defer cancel()
+		serverAddresses, err := etcd.GetAPIServerURLsFromETCD(toCtx, &serverConfig.ControlConfig)
+		if err == nil && len(serverAddresses) > 0 {
+			agentConfig.APIAddressCh <- serverAddresses
 			break
 		}
-		logrus.Warn(err)
+		if !errors.Is(err, etcd.ErrAddressNotSet) {
+			logrus.Warnf("Failed to get apiserver address from etcd: %v", err)
+		}
+		<-toCtx.Done()
 	}
 }

pkg/cluster/bootstrap.go

@@ -356,6 +356,7 @@ func (c *Cluster) ReconcileBootstrapData(ctx context.Context, buf io.ReadSeeker,
 			buf.Seek(0, 0)
 		}
 
+		logrus.Debugf("One or more certificate directories do not exist; writing data to disk from datastore")
 		return bootstrap.WriteToDiskFromStorage(files, crb)
 	}
 

pkg/daemons/control/server.go

@@ -55,10 +55,12 @@ func Server(ctx context.Context, cfg *config.Control) error {
 		if err := apiServer(ctx, cfg); err != nil {
 			return err
 		}
+	}
 
-		if err := waitForAPIServerInBackground(ctx, cfg.Runtime); err != nil {
-			return err
-		}
+	// Wait for an apiserver to become available before starting additional controllers,
+	// even if we're not running an apiserver locally.
+	if err := waitForAPIServerInBackground(ctx, cfg.Runtime); err != nil {
+		return err
 	}
 
 	if !cfg.DisableScheduler {