Move container runtime ready channel into executor

pkg/agent/cri/cri.go

@@ -5,10 +5,37 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
+	k8sutil "k8s.io/kubernetes/pkg/kubelet/util"
 )
 
 const maxMsgSize = 1024 * 1024 * 16
 
+// Connection connects to a CRI socket at the given path.
+func Connection(ctx context.Context, address string) (*grpc.ClientConn, error) {
+	addr, dialer, err := k8sutil.GetAddressAndDialer(socketPrefix + address)
+	if err != nil {
+		return nil, err
+	}
+
+	conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(3*time.Second), grpc.WithContextDialer(dialer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))
+	if err != nil {
+		return nil, err
+	}
+
+	c := runtimeapi.NewRuntimeServiceClient(conn)
+	_, err = c.Version(ctx, &runtimeapi.VersionRequest{
+		Version: "0.1.0",
+	})
+	if err != nil {
+		conn.Close()
+		return nil, err
+	}
+
+	return conn, nil
+}
+
 // WaitForService blocks in a retry loop until the CRI service
 // is functional at the provided socket address. It will return only on success,
 // or when the context is cancelled.

pkg/agent/cri/cri_linux.go

@@ -3,37 +3,4 @@
 
 package cri
 
-import (
-	"context"
-	"time"
-
-	"google.golang.org/grpc"
-	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
-	k8sutil "k8s.io/kubernetes/pkg/kubelet/util"
-)
-
 const socketPrefix = "unix://"
-
-// Connection connects to a CRI socket at the given path.
-func Connection(ctx context.Context, address string) (*grpc.ClientConn, error) {
-	addr, dialer, err := k8sutil.GetAddressAndDialer(socketPrefix + address)
-	if err != nil {
-		return nil, err
-	}
-
-	conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(3*time.Second), grpc.WithContextDialer(dialer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))
-	if err != nil {
-		return nil, err
-	}
-
-	c := runtimeapi.NewRuntimeServiceClient(conn)
-	_, err = c.Version(ctx, &runtimeapi.VersionRequest{
-		Version: "0.1.0",
-	})
-	if err != nil {
-		conn.Close()
-		return nil, err
-	}
-
-	return conn, nil
-}

pkg/agent/cri/cri_windows.go

@@ -3,35 +3,4 @@
 
 package cri
 
-import (
-	"context"
-	"time"
-
-	"google.golang.org/grpc"
-	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
-	"k8s.io/kubernetes/pkg/kubelet/util"
-)
-
-// Connection connects to a CRI socket at the given path.
-func Connection(ctx context.Context, address string) (*grpc.ClientConn, error) {
-	addr, dialer, err := util.GetAddressAndDialer(address)
-	if err != nil {
-		return nil, err
-	}
-
-	conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithTimeout(3*time.Second), grpc.WithContextDialer(dialer), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMsgSize)))
-	if err != nil {
-		return nil, err
-	}
-
-	c := runtimeapi.NewRuntimeServiceClient(conn)
-	_, err = c.Version(ctx, &runtimeapi.VersionRequest{
-		Version: "0.1.0",
-	})
-	if err != nil {
-		conn.Close()
-		return nil, err
-	}
-
-	return conn, nil
-}
+const socketPrefix = "npipe://"

pkg/agent/run.go

@@ -158,13 +158,10 @@ func run(ctx context.Context, cfg cmds.Agent, proxy proxy.Proxy) error {
 		if err := executor.Containerd(ctx, nodeConfig); err != nil {
 			return err
 		}
-	}
-	// the container runtime is ready to host workloads when containerd is up and the airgap
-	// images have finished loading, as that portion of startup may block for an arbitrary
-	// amount of time depending on how long it takes to import whatever the user has placed
-	// in the images directory.
-	if cfg.ContainerRuntimeReady != nil {
-		close(cfg.ContainerRuntimeReady)
+	} else {
+		if err := executor.CRI(ctx, nodeConfig); err != nil {
+			return err
+		}
 	}
 
 	notifySocket := os.Getenv("NOTIFY_SOCKET")
@@ -282,8 +279,9 @@ func RunStandalone(ctx context.Context, cfg cmds.Agent) error {
 		return err
 	}
 
-	if cfg.ContainerRuntimeReady != nil {
-		close(cfg.ContainerRuntimeReady)
+	// this is a no-op just to get the cri ready channel closed
+	if err := executor.CRI(ctx, nodeConfig); err != nil {
+		return err
 	}
 
 	if err := tunnelSetup(ctx, nodeConfig, cfg, proxy); err != nil {
@@ -513,53 +511,34 @@ func updateAddressAnnotations(nodeConfig *daemonconfig.Node, nodeAnnotations map
 	return result, !equality.Semantic.DeepEqual(nodeAnnotations, result)
 }
 
-// setupTunnelAndRunAgent should start the setup tunnel before starting kubelet and kubeproxy
-// there are special case for etcd agents, it will wait until it can find the apiaddress from
-// the address channel and update the proxy with the servers addresses, if in rke2 we need to
-// start the agent before the tunnel is setup to allow kubelet to start first and start the pods
+// setupTunnelAndRunAgent starts the agent tunnel, cert expiry monitoring, and
+// runs the Agent (cri+kubelet). On etcd-only nodes, an extra goroutine is
+// started to retrieve apiserver addresses from the datastore. On other node
+// types, this is done later by the tunnel setup, which starts goroutines to
+// watch apiserver endpoints.
 func setupTunnelAndRunAgent(ctx context.Context, nodeConfig *daemonconfig.Node, cfg cmds.Agent, proxy proxy.Proxy) error {
-	var agentRan bool
-	// IsAPIServerLBEnabled is used as a shortcut for detecting RKE2, where the kubelet needs to
-	// be run earlier in order to manage static pods. This should probably instead query a
-	// flag on the executor or something.
+	// only need to get apiserver addresses from the datastore on an etcd-only node that is not being reset
 	if !cfg.ClusterReset && cfg.ETCDAgent {
-		// ETCDAgent is only set to true on servers that are started with --disable-apiserver.
-		// In this case, we may be running without an apiserver available in the cluster, and need
-		// to wait for one to register and post it's address into APIAddressCh so that we can update
-		// the LB proxy with its address.
-		if proxy.IsAPIServerLBEnabled() {
-			// On RKE2, the agent needs to be started early to run the etcd static pod.
-			if err := agent.Agent(ctx, nodeConfig, proxy); err != nil {
-				return err
-			}
-			agentRan = true
-		}
-		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
-		// up static pods.
-		if err := agent.Agent(ctx, nodeConfig, proxy); err != nil {
-			return err
-		}
-		agentRan = true
+		go waitForAPIServerAddresses(ctx, nodeConfig, cfg, proxy)
 	}
 
 	if err := tunnelSetup(ctx, nodeConfig, cfg, proxy); err != nil {
 		return err
 	}
+
 	if err := certMonitorSetup(ctx, nodeConfig, cfg); err != nil {
 		return err
 	}
 
-	if !agentRan {
-		return agent.Agent(ctx, nodeConfig, proxy)
-	}
-	return nil
+	return agent.Agent(ctx, nodeConfig, proxy)
 }
 
-func waitForAPIServerAddresses(ctx context.Context, nodeConfig *daemonconfig.Node, cfg cmds.Agent, proxy proxy.Proxy) error {
+// waitForAPIServerAddresses syncs apiserver addresses from the datastore. This
+// is also handled by the agent tunnel watch, but on etcd-only nodes we need to
+// read apiserver addresses from APIAddressCh before the agent has a
+// connection to the apiserver. This does not return until addresses or set,
+// or the context is cancelled.
+func waitForAPIServerAddresses(ctx context.Context, nodeConfig *daemonconfig.Node, cfg cmds.Agent, proxy proxy.Proxy) {
 	var localSupervisorDefault bool
 	if addresses := proxy.SupervisorAddresses(); len(addresses) > 0 {
 		host, _, _ := net.SplitHostPort(addresses[0])
@@ -586,9 +565,9 @@ func waitForAPIServerAddresses(ctx context.Context, nodeConfig *daemonconfig.Nod
 				proxy.SetSupervisorDefault(addresses[0])
 			}
 			proxy.Update(addresses)
-			return nil
+			return
 		case <-ctx.Done():
-			return ctx.Err()
+			return
 		}
 	}
 }

pkg/agent/tunnel/tunnel.go

@@ -8,7 +8,6 @@ import (
 	"net"
 	"os"
 	"strconv"
-	"sync"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -49,6 +48,7 @@ var (
 
 type agentTunnel struct {
 	client      kubernetes.Interface
+	tlsConfig   *tls.Config
 	cidrs       cidranger.Ranger
 	ports       map[string]bool
 	mode        string
@@ -69,6 +69,8 @@ func (p *podEntry) Network() net.IPNet {
 	return p.cidr
 }
 
+// Setup sets up the agent tunnel, which is reponsible for connecting websocket tunnels to
+// control-plane nodes, syncing endpoints for the tunnel authorizer, and updating proxy endpoints.
 func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) error {
 	client, err := util.GetClientSet(config.AgentConfig.KubeConfigK3sController)
 	if err != nil {
@@ -87,6 +89,7 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
 
 	tunnel := &agentTunnel{
 		client:      client,
+		tlsConfig:   tlsConfig,
 		cidrs:       cidranger.NewPCTrieRanger(),
 		ports:       map[string]bool{},
 		mode:        config.EgressSelectorMode,
@@ -95,6 +98,14 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
 		startTime:   time.Now().Truncate(time.Second),
 	}
 
+	go tunnel.startWatches(ctx, config, proxy)
+
+	return nil
+}
+
+// startWatches starts watching for changes to endpoints, both for the tunnel authorizer,
+// and to sync supervisor addresses into the proxy. This will block until the context is cancelled.
+func (a *agentTunnel) startWatches(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) {
 	rbacReady := make(chan struct{})
 	go func() {
 		<-executor.APIServerReadyChan()
@@ -113,15 +124,15 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
 	// signifying that this is an agentless server that will not register a node.
 	if config.ContainerRuntimeEndpoint != "/dev/null" {
 		// Allow the kubelet port, as published via our node object.
-		go tunnel.setKubeletPort(ctx, rbacReady)
+		go a.setKubeletPort(ctx, rbacReady)
 
-		switch tunnel.mode {
+		switch a.mode {
 		case daemonconfig.EgressSelectorModeCluster:
 			// In Cluster mode, we allow the cluster CIDRs, and any connections to the node's IPs for pods using host network.
-			tunnel.clusterAuth(config)
+			a.clusterAuth(config)
 		case daemonconfig.EgressSelectorModePod:
 			// In Pod mode, we watch pods assigned to this node, and allow their addresses, as well as ports used by containers with host network.
-			go tunnel.watchPods(ctx, rbacReady, config)
+			go a.watchPods(ctx, rbacReady, config)
 		}
 	}
 
@@ -150,7 +161,7 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
 			}
 			proxy.Update(addresses)
 		} else {
-			if endpoint, err := client.CoreV1().Endpoints(metav1.NamespaceDefault).Get(ctx, "kubernetes", metav1.GetOptions{}); err != nil {
+			if endpoint, err := a.client.CoreV1().Endpoints(metav1.NamespaceDefault).Get(ctx, "kubernetes", metav1.GetOptions{}); err != nil {
 				logrus.Errorf("Failed to get apiserver addresses from kubernetes endpoints: %v", err)
 			} else {
 				addresses := util.GetAddresses(endpoint)
@@ -162,24 +173,7 @@ func Setup(ctx context.Context, config *daemonconfig.Node, proxy proxy.Proxy) er
 		}
 	}
 
-	wg := &sync.WaitGroup{}
-
-	go tunnel.watchEndpoints(ctx, rbacReady, wg, tlsConfig, config, proxy)
-
-	wait := make(chan int, 1)
-	go func() {
-		wg.Wait()
-		wait <- 0
-	}()
-
-	select {
-	case <-ctx.Done():
-		logrus.Error("Tunnel context canceled while waiting for connection")
-		return ctx.Err()
-	case <-wait:
-	}
-
-	return nil
+	a.watchEndpoints(ctx, rbacReady, config, proxy)
 }
 
 // setKubeletPort retrieves the configured kubelet port from our node object
@@ -307,8 +301,8 @@ func (a *agentTunnel) watchPods(ctx context.Context, rbacReady <-chan struct{},
 // WatchEndpoints attempts to create tunnels to all supervisor addresses.  Once the
 // apiserver is up, go into a watch loop, adding and removing tunnels as endpoints come
 // and go from the cluster.
-func (a *agentTunnel) watchEndpoints(ctx context.Context, rbacReady <-chan struct{}, wg *sync.WaitGroup, tlsConfig *tls.Config, node *daemonconfig.Node, proxy proxy.Proxy) {
-	syncProxyAddresses := a.getProxySyncer(ctx, wg, tlsConfig, proxy)
+func (a *agentTunnel) watchEndpoints(ctx context.Context, rbacReady <-chan struct{}, node *daemonconfig.Node, proxy proxy.Proxy) {
+	syncProxyAddresses := a.getProxySyncer(ctx, proxy)
 	refreshFromSupervisor := getAPIServersRequester(node, proxy, syncProxyAddresses)
 
 	<-rbacReady
@@ -395,17 +389,12 @@ type agentConnection struct {
 
 // connect initiates a connection to the remotedialer server. Incoming dial requests from
 // the server will be checked by the authorizer function prior to being fulfilled.
-func (a *agentTunnel) connect(rootCtx context.Context, waitGroup *sync.WaitGroup, address string, tlsConfig *tls.Config) agentConnection {
+func (a *agentTunnel) connect(rootCtx context.Context, address string) agentConnection {
 	var status loadbalancer.HealthCheckResult
 
 	wsURL := fmt.Sprintf("wss://%s/v1-"+version.Program+"/connect", address)
 	ws := &websocket.Dialer{
-		TLSClientConfig: tlsConfig,
-	}
-
-	once := sync.Once{}
-	if waitGroup != nil {
-		waitGroup.Add(1)
+		TLSClientConfig: a.tlsConfig,
 	}
 
 	ctx, cancel := context.WithCancel(rootCtx)
@@ -416,9 +405,6 @@ func (a *agentTunnel) connect(rootCtx context.Context, waitGroup *sync.WaitGroup
 	onConnect := func(_ context.Context, _ *remotedialer.Session) error {
 		status = loadbalancer.HealthCheckResultOK
 		logrus.WithField("url", wsURL).Info("Remotedialer connected to proxy")
-		if waitGroup != nil {
-			once.Do(waitGroup.Done)
-		}
 		return nil
 	}
 
@@ -435,9 +421,6 @@ func (a *agentTunnel) connect(rootCtx context.Context, waitGroup *sync.WaitGroup
 			}
 			// If the context has been cancelled, exit the goroutine instead of retrying
 			if ctx.Err() != nil {
-				if waitGroup != nil {
-					once.Do(waitGroup.Done)
-				}
 				return
 			}
 		}
@@ -476,13 +459,13 @@ type proxySyncer func(addresses []string)
 // getProxySyncer returns a function that can be called to update the list of supervisors.
 // This function is responsible for connecting to or disconnecting websocket tunnels,
 // as well as updating the proxy loadbalancer server list.
-func (a *agentTunnel) getProxySyncer(ctx context.Context, wg *sync.WaitGroup, tlsConfig *tls.Config, proxy proxy.Proxy) proxySyncer {
+func (a *agentTunnel) getProxySyncer(ctx context.Context, proxy proxy.Proxy) proxySyncer {
 	disconnect := map[string]context.CancelFunc{}
 	// Attempt to connect to inital list of addresses, storing their cancellation
 	// function for later when we need to disconnect.
 	for _, address := range proxy.SupervisorAddresses() {
 		if _, ok := disconnect[address]; !ok {
-			conn := a.connect(ctx, wg, address, tlsConfig)
+			conn := a.connect(ctx, address)
 			disconnect[address] = conn.cancel
 			proxy.SetHealthCheck(address, conn.healthCheck)
 		}
@@ -538,7 +521,7 @@ func (a *agentTunnel) getProxySyncer(ctx context.Context, wg *sync.WaitGroup, tl
 			// add new servers
 			for address := range addedAddresses {
 				if _, ok := disconnect[address]; !ok {
-					conn := a.connect(ctx, nil, address, tlsConfig)
+					conn := a.connect(ctx, address)
 					logrus.Infof("Started tunnel to %s", address)
 					disconnect[address] = conn.cancel
 					proxy.SetHealthCheck(address, conn.healthCheck)

pkg/cli/cert/cert.go

@@ -46,7 +46,7 @@ func commandSetup(app *cli.Context, cfg *cmds.Server, sc *server.Config) (string
 		cfg.Token = string(bytes.TrimRight(tokenByte, "\n"))
 	}
 	sc.ControlConfig.Token = cfg.Token
-	sc.ControlConfig.Runtime = config.NewRuntime(nil)
+	sc.ControlConfig.Runtime = config.NewRuntime()
 
 	return dataDir, nil
 }
@@ -287,7 +287,7 @@ func rotateCA(app *cli.Context, cfg *cmds.Server, sync *cmds.CertRotateCA) error
 
 	// Set up dummy server config for reading new bootstrap data from disk.
 	tmpServer := &config.Control{
-		Runtime: config.NewRuntime(nil),
+		Runtime: config.NewRuntime(),
 		DataDir: sync.CACertPath,
 	}
 	deps.CreateRuntimeCertFiles(tmpServer)

pkg/cli/cmds/agent.go

@@ -56,7 +56,6 @@ type Agent struct {
 	Taints                   cli.StringSlice
 	ImageCredProvBinDir      string
 	ImageCredProvConfig      string
-	ContainerRuntimeReady    chan<- struct{}
 	AgentShared
 }
 

pkg/cli/server/server.go

@@ -113,11 +113,9 @@ func run(app *cli.Context, cfg *cmds.Server, leaderControllers server.CustomCont
 		}
 	}
 
-	containerRuntimeReady := make(chan struct{})
-
 	serverConfig := server.Config{}
 	serverConfig.DisableAgent = cfg.DisableAgent
-	serverConfig.ControlConfig.Runtime = config.NewRuntime(containerRuntimeReady)
+	serverConfig.ControlConfig.Runtime = config.NewRuntime()
 	serverConfig.ControlConfig.Token = cfg.Token
 	serverConfig.ControlConfig.AgentToken = cfg.AgentToken
 	serverConfig.ControlConfig.JoinURL = cfg.ServerURL
@@ -525,7 +523,6 @@ func run(app *cli.Context, cfg *cmds.Server, leaderControllers server.CustomCont
 	}
 
 	agentConfig := cmds.AgentConfig
-	agentConfig.ContainerRuntimeReady = containerRuntimeReady
 	agentConfig.Debug = app.GlobalBool("debug")
 	agentConfig.DataDir = filepath.Dir(serverConfig.ControlConfig.DataDir)
 	agentConfig.ServerURL = url

pkg/daemons/config/types.go

@@ -312,7 +312,6 @@ type ControlRuntimeBootstrap struct {
 type ControlRuntime struct {
 	ControlRuntimeBootstrap
 
-	ContainerRuntimeReady                <-chan struct{}
 	ETCDReady                            <-chan struct{}
 	StartupHooksWg                       *sync.WaitGroup
 	ClusterControllerStarts              map[string]leader.Callback
@@ -392,9 +391,8 @@ type CoreFactory interface {
 	Start(ctx context.Context, defaultThreadiness int) error
 }
 
-func NewRuntime(containerRuntimeReady <-chan struct{}) *ControlRuntime {
+func NewRuntime() *ControlRuntime {
 	return &ControlRuntime{
-		ContainerRuntimeReady:                containerRuntimeReady,
 		ClusterControllerStarts:              map[string]leader.Callback{},
 		LeaderElectedClusterControllerStarts: map[string]leader.Callback{},
 	}

pkg/daemons/executor/embed.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/k3s-io/k3s/pkg/agent/containerd"
+	"github.com/k3s-io/k3s/pkg/agent/cri"
 	"github.com/k3s-io/k3s/pkg/agent/cridockerd"
 	"github.com/k3s-io/k3s/pkg/cli/cmds"
 	daemonconfig "github.com/k3s-io/k3s/pkg/daemons/config"
@@ -44,6 +45,7 @@ func init() {
 
 func (e *Embedded) Bootstrap(ctx context.Context, nodeConfig *daemonconfig.Node, cfg cmds.Agent) error {
 	e.apiServerReady = util.APIServerReadyChan(ctx, nodeConfig.AgentConfig.KubeConfigK3sController, util.DefaultAPIServerReadyTimeout)
+	e.criReady = make(chan struct{})
 	e.nodeConfig = nodeConfig
 
 	go func() {
@@ -235,16 +237,34 @@ func (e *Embedded) CurrentETCDOptions() (InitialOptions, error) {
 }
 
 func (e *Embedded) Containerd(ctx context.Context, cfg *daemonconfig.Node) error {
+	defer close(e.criReady)
 	return containerd.Run(ctx, cfg)
 }
 
 func (e *Embedded) Docker(ctx context.Context, cfg *daemonconfig.Node) error {
+	defer close(e.criReady)
 	return cridockerd.Run(ctx, cfg)
 }
 
+func (e *Embedded) CRI(ctx context.Context, cfg *daemonconfig.Node) error {
+	defer close(e.criReady)
+	// agentless sets cri socket path to /dev/null to indicate no CRI is needed
+	if cfg.ContainerRuntimeEndpoint != "/dev/null" {
+		return cri.WaitForService(ctx, cfg.ContainerRuntimeEndpoint, "CRI")
+	}
+	return nil
+}
+
 func (e *Embedded) APIServerReadyChan() <-chan struct{} {
 	if e.apiServerReady == nil {
 		panic("executor not bootstrapped")
 	}
 	return e.apiServerReady
 }
+
+func (e *Embedded) CRIReadyChan() <-chan struct{} {
+	if e.criReady == nil {
+		panic("executor not bootstrapped")
+	}
+	return e.criReady
+}

pkg/daemons/executor/etcd.go

@@ -17,6 +17,7 @@ import (
 // of the embedded execututor is disabled by build flags
 type Embedded struct {
 	apiServerReady <-chan struct{}
+	criReady       chan struct{}
 	nodeConfig     *daemonconfig.Node
 }
 

pkg/daemons/executor/executor.go

@@ -33,7 +33,9 @@ type Executor interface {
 	CloudControllerManager(ctx context.Context, ccmRBACReady <-chan struct{}, args []string) error
 	Containerd(ctx context.Context, node *daemonconfig.Node) error
 	Docker(ctx context.Context, node *daemonconfig.Node) error
+	CRI(ctx context.Context, node *daemonconfig.Node) error
 	APIServerReadyChan() <-chan struct{}
+	CRIReadyChan() <-chan struct{}
 }
 
 type ETCDConfig struct {
@@ -183,6 +185,14 @@ func Docker(ctx context.Context, config *daemonconfig.Node) error {
 	return executor.Docker(ctx, config)
 }
 
+func CRI(ctx context.Context, config *daemonconfig.Node) error {
+	return executor.CRI(ctx, config)
+}
+
 func APIServerReadyChan() <-chan struct{} {
 	return executor.APIServerReadyChan()
 }
+
+func CRIReadyChan() <-chan struct{} {
+	return executor.CRIReadyChan()
+}

pkg/etcd/etcd.go

@@ -343,7 +343,7 @@ func (e *ETCD) IsInitialized() (bool, error) {
 func (e *ETCD) Reset(ctx context.Context, rebootstrap func() error) error {
 	// Wait for etcd to come up as a new single-node cluster, then exit
 	go func() {
-		<-e.config.Runtime.ContainerRuntimeReady
+		<-executor.CRIReadyChan()
 		t := time.NewTicker(5 * time.Second)
 		defer t.Stop()
 		for range t.C {
@@ -497,7 +497,7 @@ func (e *ETCD) Start(ctx context.Context, clientAccessInfo *clientaccess.Info) e
 			select {
 			case <-time.After(30 * time.Second):
 				logrus.Infof("Waiting for container runtime to become ready before joining etcd cluster")
-			case <-e.config.Runtime.ContainerRuntimeReady:
+			case <-executor.CRIReadyChan():
 				if err := wait.PollUntilContextCancel(ctx, time.Second, true, func(ctx context.Context) (bool, error) {
 					if err := e.join(ctx, clientAccessInfo); err != nil {
 						// Retry the join if waiting for another member to be promoted, or waiting for peers to connect after promotion
@@ -726,7 +726,6 @@ func (e *ETCD) handler(next http.Handler) http.Handler {
 }
 
 // infoHandler returns etcd cluster information. This is used by new members when joining the cluster.
-// If we can't retrieve an actual MemberList from etcd, we return a canned response with only the local node listed.
 func (e *ETCD) infoHandler() http.Handler {
 	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
 		if req.Method != http.MethodGet {
@@ -734,6 +733,11 @@ func (e *ETCD) infoHandler() http.Handler {
 			return
 		}
 
+		if e.client == nil {
+			util.SendError(errors.New("failed to get etcd MemberList: etcd not started"), rw, req, http.StatusInternalServerError)
+			return
+		}
+
 		ctx, cancel := context.WithTimeout(req.Context(), 2*time.Second)
 		defer cancel()
 
@@ -1157,7 +1161,7 @@ func (e *ETCD) RemovePeer(ctx context.Context, name, address string, allowSelfRe
 // being promoted to full voting member. The checks only run on the cluster member that is
 // the etcd leader.
 func (e *ETCD) manageLearners(ctx context.Context) {
-	<-e.config.Runtime.ContainerRuntimeReady
+	<-executor.CRIReadyChan()
 	t := time.NewTicker(manageTickerTime)
 	defer t.Stop()
 

pkg/etcd/etcd_linux_test.go

@@ -61,7 +61,7 @@ func generateTestConfig() *config.Control {
 	}
 	return &config.Control{
 		ServerNodeName:        hostname,
-		Runtime:               config.NewRuntime(containerRuntimeReady),
+		Runtime:               config.NewRuntime(),
 		HTTPSPort:             6443,
 		SupervisorPort:        6443,
 		AdvertisePort:         6443,

pkg/server/handlers/cert.go

@@ -56,7 +56,7 @@ func caCertReplace(control *config.Control, buf io.ReadCloser, force bool) error
 	}
 	defer os.RemoveAll(tmpdir)
 
-	runtime := config.NewRuntime(nil)
+	runtime := config.NewRuntime()
 	runtime.EtcdConfig = control.Runtime.EtcdConfig
 	runtime.ServerToken = control.Runtime.ServerToken
 

tests/unit.go

@@ -44,11 +44,7 @@ func CleanupDataDir(cnf *config.Control) {
 // config.ControlRuntime with all the appropriate certificate keys.
 func GenerateRuntime(cnf *config.Control) error {
 	// reuse ready channel from existing runtime if set
-	var readyCh <-chan struct{}
-	if cnf.Runtime != nil {
-		readyCh = cnf.Runtime.ContainerRuntimeReady
-	}
-	cnf.Runtime = config.NewRuntime(readyCh)
+	cnf.Runtime = config.NewRuntime()
 	if err := GenerateDataDir(cnf); err != nil {
 		return err
 	}