Wait for worker node to be ready when joining (#2019)

* Wait for worker node to be ready when joining

Author: sgalsaleh

cmd/installer/cli/install.go

@@ -976,7 +976,7 @@ func waitForNode(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("get hostname: %w", err)
 	}
-	if err := kubeutils.WaitForControllerNode(ctx, kcli, hostname); err != nil {
+	if err := kubeutils.WaitForNode(ctx, kcli, hostname, false); err != nil {
 		return fmt.Errorf("wait for node: %w", err)
 	}
 	return nil

cmd/installer/cli/join.go

@@ -182,16 +182,26 @@ func runJoin(ctx context.Context, name string, flags JoinCmdFlags, jcmd *kotsadm
 		return err
 	}
 
-	if isWorker {
-		logrus.Debugf("worker node join finished")
-		return nil
-	}
-
 	kcli, err := kubeutils.KubeClient()
 	if err != nil {
 		return fmt.Errorf("unable to get kube client: %w", err)
 	}
 
+	hostname, err := os.Hostname()
+	if err != nil {
+		return fmt.Errorf("unable to get hostname: %w", err)
+	}
+
+	logrus.Debugf("waiting for node to join cluster")
+	if err := waitForNodeToJoin(ctx, kcli, hostname, isWorker); err != nil {
+		return fmt.Errorf("unable to wait for node: %w", err)
+	}
+
+	if isWorker {
+		logrus.Debugf("worker node join finished")
+		return nil
+	}
+
 	airgapChartsPath := ""
 	if flags.isAirgap {
 		airgapChartsPath = runtimeconfig.EmbeddedClusterChartsSubDir()
@@ -207,15 +217,6 @@ func runJoin(ctx context.Context, name string, flags JoinCmdFlags, jcmd *kotsadm
 	}
 	defer hcli.Close()
 
-	hostname, err := os.Hostname()
-	if err != nil {
-		return fmt.Errorf("unable to get hostname: %w", err)
-	}
-
-	if err := waitForNodeToJoin(ctx, kcli, hostname); err != nil {
-		return fmt.Errorf("unable to wait for node: %w", err)
-	}
-
 	if flags.enableHighAvailability {
 		if err := maybeEnableHA(ctx, kcli, hcli, flags.isAirgap, cidrCfg.ServiceCIDR, jcmd.InstallationSpec.Proxy, jcmd.InstallationSpec.Config); err != nil {
 			return fmt.Errorf("unable to enable high availability: %w", err)
@@ -246,7 +247,12 @@ func runJoinVerifyAndPrompt(name string, flags JoinCmdFlags, jcmd *kotsadm.JoinC
 	}
 
 	runtimeconfig.Set(jcmd.InstallationSpec.RuntimeConfig)
-	os.Setenv("KUBECONFIG", runtimeconfig.PathToKubeConfig())
+	isWorker := !strings.Contains(jcmd.K0sJoinCommand, "controller")
+	if isWorker {
+		os.Setenv("KUBECONFIG", runtimeconfig.PathToKubeletConfig())
+	} else {
+		os.Setenv("KUBECONFIG", runtimeconfig.PathToKubeConfig())
+	}
 	os.Setenv("TMPDIR", runtimeconfig.EmbeddedClusterTmpSubDir())
 
 	if err := runtimeconfig.WriteToDisk(); err != nil {
@@ -476,11 +482,11 @@ func runK0sInstallCommand(networkInterface string, fullcmd string) error {
 	return nil
 }
 
-func waitForNodeToJoin(ctx context.Context, kcli client.Client, hostname string) error {
+func waitForNodeToJoin(ctx context.Context, kcli client.Client, hostname string, isWorker bool) error {
 	loading := spinner.Start()
 	defer loading.Close()
 	loading.Infof("Waiting for node to join the cluster")
-	if err := kubeutils.WaitForControllerNode(ctx, kcli, hostname); err != nil {
+	if err := kubeutils.WaitForNode(ctx, kcli, hostname, isWorker); err != nil {
 		return fmt.Errorf("unable to wait for node: %w", err)
 	}
 	loading.Infof("Node has joined the cluster!")

pkg/dryrun/kubeutils.go

@@ -44,7 +44,7 @@ func (k *KubeUtils) WaitForNodes(ctx context.Context, cli client.Client) error {
 	return nil
 }
 
-func (k *KubeUtils) WaitForControllerNode(ctx context.Context, kcli client.Client, name string) error {
+func (k *KubeUtils) WaitForNode(ctx context.Context, kcli client.Client, name string, isWorker bool) error {
 	return nil
 }
 

pkg/kubeutils/interface.go

@@ -28,7 +28,7 @@ type KubeUtilsInterface interface {
 	WaitForPodComplete(ctx context.Context, cli client.Client, ns, name string, opts *WaitOptions) error
 	WaitForInstallation(ctx context.Context, cli client.Client, writer *spinner.MessageWriter) error
 	WaitForNodes(ctx context.Context, cli client.Client) error
-	WaitForControllerNode(ctx context.Context, kcli client.Client, name string) error
+	WaitForNode(ctx context.Context, kcli client.Client, name string, isWorker bool) error
 	IsNamespaceReady(ctx context.Context, cli client.Client, ns string) (bool, error)
 	IsDeploymentReady(ctx context.Context, cli client.Client, ns, name string) (bool, error)
 	IsStatefulSetReady(ctx context.Context, cli client.Client, ns, name string) (bool, error)
@@ -86,8 +86,8 @@ func WaitForNodes(ctx context.Context, cli client.Client) error {
 	return kb.WaitForNodes(ctx, cli)
 }
 
-func WaitForControllerNode(ctx context.Context, kcli client.Client, name string) error {
-	return kb.WaitForControllerNode(ctx, kcli, name)
+func WaitForNode(ctx context.Context, kcli client.Client, name string, isWorker bool) error {
+	return kb.WaitForNode(ctx, kcli, name, isWorker)
 }
 
 func IsNamespaceReady(ctx context.Context, cli client.Client, ns string) (bool, error) {

pkg/kubeutils/kubeutils.go

@@ -187,8 +187,8 @@ func (k *KubeUtils) WaitForNodes(ctx context.Context, cli client.Client) error {
 	return nil
 }
 
-// WaitForControllerNode waits for a specific controller node to be registered with the cluster.
-func (k *KubeUtils) WaitForControllerNode(ctx context.Context, kcli client.Client, name string) error {
+// WaitForNode waits for a specific controller node to be registered with the cluster.
+func (k *KubeUtils) WaitForNode(ctx context.Context, kcli client.Client, name string, isWorker bool) error {
 	backoff := wait.Backoff{Steps: 60, Duration: 5 * time.Second, Factor: 1.0, Jitter: 0.1}
 	var lasterr error
 	if err := wait.ExponentialBackoffWithContext(
@@ -198,9 +198,11 @@ func (k *KubeUtils) WaitForControllerNode(ctx context.Context, kcli client.Clien
 				lasterr = fmt.Errorf("unable to get node: %v", err)
 				return false, nil
 			}
-			if _, ok := node.Labels["node-role.kubernetes.io/control-plane"]; !ok {
-				lasterr = fmt.Errorf("control plane label not found")
-				return false, nil
+			if !isWorker {
+				if _, ok := node.Labels["node-role.kubernetes.io/control-plane"]; !ok {
+					lasterr = fmt.Errorf("control plane label not found")
+					return false, nil
+				}
 			}
 			for _, condition := range node.Status.Conditions {
 				if condition.Type == corev1.NodeReady && condition.Status == corev1.ConditionTrue {

pkg/runtimeconfig/runtimeconfig.go

@@ -121,6 +121,11 @@ func PathToKubeConfig() string {
 	return filepath.Join(EmbeddedClusterK0sSubDir(), "pki/admin.conf")
 }
 
+// PathToKubeletConfig returns the path to the kubelet config file.
+func PathToKubeletConfig() string {
+	return filepath.Join(EmbeddedClusterK0sSubDir(), "kubelet.conf")
+}
+
 // EmbeddedClusterSupportSubDir returns the path to the directory where embedded-cluster
 // support files are stored. Things that are useful when providing end user support in
 // a running cluster should be stored into this directory.

tests/dryrun/join_test.go

@@ -176,3 +176,33 @@ func TestJoinRunPreflights(t *testing.T) {
 	dryrunJoin(t, "run-preflights", "10.0.0.1", "some-token")
 	t.Logf("%s: test complete", time.Now().Format(time.RFC3339))
 }
+
+func TestJoinWorkerNode(t *testing.T) {
+	drFile := filepath.Join(t.TempDir(), "ec-dryrun.yaml")
+	client := &dryrun.Client{
+		Kotsadm: dryrun.NewKotsadm(),
+	}
+	clusterID := uuid.New()
+	jcmd := &kotsadm.JoinCommandResponse{
+		K0sJoinCommand:         "/usr/local/bin/k0s install worker --no-taints --labels kots.io/embedded-cluster-role=total-1,kots.io/embedded-cluster-role-0=worker-test,worker-label=worker-label-value",
+		K0sToken:               "some-k0s-token",
+		EmbeddedClusterVersion: "v0.0.0",
+		ClusterID:              clusterID,
+		InstallationSpec: ecv1beta1.InstallationSpec{
+			ClusterID: clusterID.String(),
+			Config: &ecv1beta1.ConfigSpec{
+				UnsupportedOverrides: ecv1beta1.UnsupportedOverrides{},
+			},
+		},
+	}
+	client.Kotsadm.SetGetJoinTokenResponse("10.0.0.1", "some-token", jcmd, nil)
+	dryrun.Init(drFile, client)
+	dr := dryrunJoin(t, "10.0.0.1", "some-token")
+
+	// --- validate os env --- //
+	assertEnv(t, dr.OSEnv, map[string]string{
+		"KUBECONFIG": "/var/lib/embedded-cluster/k0s/kubelet.conf", // uses kubelet config
+	})
+
+	t.Logf("%s: test complete", time.Now().Format(time.RFC3339))
+}