add runtime v2 join master

pkg/runtime/init.go

@@ -48,7 +48,7 @@ func (k *KubeadmRuntime) loadMetadata() error {
 }
 
 func (k *KubeadmRuntime) ConfigKubeadmOnMaster0() error {
-	if err := k.LoadFromClusterfile(k.RuntimeConfig.Clusterfile); err != nil {
+	if err := k.LoadFromClusterfile(k.Config.Clusterfile); err != nil {
 		return fmt.Errorf("failed to load kubeadm config from clusterfile: %v", err)
 	}
 	// TODO handle the kubeadm config, like kubeproxy config
@@ -105,9 +105,7 @@ func (k *KubeadmRuntime) GenerateCert() error {
 	if err != nil {
 		return fmt.Errorf("generate certs failed %v", err)
 	}
-	k.sendNewCertAndKey([]string{k.getMaster0IP()})
-
-	return nil
+	return k.sendNewCertAndKey([]string{k.getMaster0IP()})
 }
 
 func (k *KubeadmRuntime) CreateKubeConfig() error {
@@ -127,7 +125,9 @@ func (k *KubeadmRuntime) CreateKubeConfig() error {
 }
 
 func (k *KubeadmRuntime) CopyStaticFiles(nodes []string) error {
-	var flag bool
+	errCh := make(chan error, len(nodes))
+	defer close(errCh)
+
 	for _, file := range MasterStaticFiles {
 		staticFilePath := filepath.Join(k.getStaticFileDir(), file.Name)
 		cmdLinkStatic := fmt.Sprintf(RemoteCmdCopyStatic, file.DestinationDir, staticFilePath, filepath.Join(file.DestinationDir, file.Name))
@@ -138,23 +138,19 @@ func (k *KubeadmRuntime) CopyStaticFiles(nodes []string) error {
 				defer wg.Done()
 				ssh, err := k.getHostSSHClient(host)
 				if err != nil {
-					logger.Error("new ssh client failed %v", err)
-					flag = true
+					errCh <- fmt.Errorf("new ssh client failed %v", err)
 					return
 				}
 				err = ssh.CmdAsync(host, cmdLinkStatic)
 				if err != nil {
-					logger.Error("[%s] link static file failed, error:%s", host, err.Error())
-					flag = true
+					errCh <- fmt.Errorf("[%s] link static file failed, error:%s", host, err.Error())
 				}
 			}(host)
-			if flag {
-				return fmt.Errorf("link static files failed %s %s", host, cmdLinkStatic)
-			}
 		}
 		wg.Wait()
 	}
-	return nil
+
+	return ReadChanError(errCh)
 }
 
 //decode output to join token  hash and key
@@ -198,7 +194,9 @@ func (k *KubeadmRuntime) InitMaster0() error {
 		return fmt.Errorf("failed to get master0 ssh client, %v", err)
 	}
 
-	k.SendJoinMasterKubeConfigs([]string{k.getMaster0IP()}, AdminConf, ControllerConf, SchedulerConf, KubeletConf)
+	if err := k.SendJoinMasterKubeConfigs([]string{k.getMaster0IP()}, AdminConf, ControllerConf, SchedulerConf, KubeletConf); err != nil {
+		return err
+	}
 	cmdAddEtcHost := fmt.Sprintf(RemoteAddEtcHosts, getAPIServerHost(k.getMaster0IP(), k.getAPIServerDomain()))
 	cmdAddRegistryHosts := fmt.Sprintf(RemoteAddEtcHosts, getRegistryHost(k.getRootfs(), k.getMaster0IP()))
 	err = ssh.CmdAsync(k.getMaster0IP(), cmdAddEtcHost, cmdAddRegistryHosts)

pkg/runtime/kubeadm_runtime.go

@@ -32,7 +32,7 @@ import (
 	"github.com/alibaba/sealer/utils/ssh"
 )
 
-type RuntimeConfig struct {
+type Config struct {
 	Vlog         int
 	VIP          string
 	RegistryPort string
@@ -48,7 +48,7 @@ type RuntimeConfig struct {
 func newKubeadmRuntime(cluster *v2.Cluster, clusterfile string) (Interface, error) {
 	k := &KubeadmRuntime{
 		Cluster: cluster,
-		RuntimeConfig: &RuntimeConfig{
+		Config: &Config{
 			Clusterfile:     clusterfile,
 			APIServerDomain: DefaultAPIserverDomain,
 		},
@@ -78,11 +78,11 @@ func (k *KubeadmRuntime) checkList() error {
 }
 
 func (k *KubeadmRuntime) checkIPList() error {
-	if len(k.Spec.Hosts) < 1 {
-		return fmt.Errorf("master hosts should not < 1, hosts len is %s", len(k.Spec.Hosts))
+	if len(k.Spec.Hosts) == 1 {
+		return fmt.Errorf("master hosts cannot be empty")
 	}
-	if len(k.Spec.Hosts[0].IPS) < 1 {
-		return fmt.Errorf("master hosts ip should not < 1, hosts ip len is %s", len(k.Spec.Hosts[0].IPS))
+	if len(k.Spec.Hosts[0].IPS) == 1 {
+		return fmt.Errorf("master hosts ip cannot be empty")
 	}
 	return nil
 }
@@ -103,15 +103,15 @@ func (k *KubeadmRuntime) getHostSSHClient(hostIP string) (ssh.Interface, error)
 }
 
 func (k *KubeadmRuntime) getRootfs() string {
-	return common.DefaultTheClusterRootfsDir(k.Cluster.Name)
+	return common.DefaultTheClusterRootfsDir(k.getClusterName())
 }
 
 func (k *KubeadmRuntime) getBasePath() string {
 	return path.Join(common.DefaultClusterRootfsDir, k.Cluster.Name)
 }
 
 func (k *KubeadmRuntime) getMaster0IP() string {
-	// aready check ip list when new the runtime
+	// already check ip list when new the runtime
 	return k.Cluster.Spec.Hosts[0].IPS[0]
 }
 
@@ -144,7 +144,7 @@ func (k *KubeadmRuntime) getDNSDomain() string {
 }
 
 func (k *KubeadmRuntime) getAPIServerDomain() string {
-	return k.RuntimeConfig.APIServerDomain
+	return k.Config.APIServerDomain
 }
 
 func (k *KubeadmRuntime) getKubeVersion() string {
@@ -174,7 +174,9 @@ func (k *KubeadmRuntime) getHostsIPByRole(role string) (nodes []string) {
 }
 
 func (k *KubeadmRuntime) WaitSSHReady(tryTimes int, hosts ...string) error {
-	var err error
+	errCh := make(chan error, len(hosts))
+	defer close(errCh)
+
 	var wg sync.WaitGroup
 	for _, h := range hosts {
 		wg.Add(1)
@@ -192,9 +194,10 @@ func (k *KubeadmRuntime) WaitSSHReady(tryTimes int, hosts ...string) error {
 				}
 				time.Sleep(time.Duration(i) * time.Second)
 			}
-			err = fmt.Errorf("wait for [%s] ssh ready timeout:  %v, ensure that the IP address or password is correct", host, err)
+			err := fmt.Errorf("wait for [%s] ssh ready timeout, ensure that the IP address or password is correct", host)
+			errCh <- err
 		}(h)
 	}
 	wg.Wait()
-	return err
+	return ReadChanError(errCh)
 }