sshd (#850)

* refactor(master): fix assh to ssh

* refactor(master): add sealctl

* refactor(master): add sealctl cert,hostname,ipvs,route,static_pod

* refactor(master): add sealctl password

* refactor(master): add sealctl token

* refactor(master): add sealctl token

* refactor(master): add sealctl kube

* refactor(master): add sealctl kube

* refactor(master): add sealctl version

* refactor(master): add sealctl .github

pkg/utils/ssh/README.md

@@ -0,0 +1,3 @@
+# 执行远程命令
+
+执行远程ssh命令标准库

pkg/utils/ssh/connect.go

@@ -1,4 +1,4 @@
-// Copyright © 2021 sealos.
+// Copyright © 2021 Alibaba Group Holding Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -17,88 +17,94 @@ package ssh
 import (
 	"fmt"
 	"io/ioutil"
+	"net"
 	"os"
+	"path"
+	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
+	"github.com/fanux/sealos/pkg/utils/iputils"
+
 	"github.com/fanux/sealos/pkg/utils/logger"
 
 	"golang.org/x/crypto/ssh"
 )
 
 /**
-这里主要是做连接ssh操作的
+  SSH connection operation
 */
-func (ss *SSH) connect(host string) (*ssh.Client, error) {
-	auth := ss.sshAuthMethod(ss.Password, ss.PkFile, ss.PkPassword)
+func (s *SSH) connect(host string) (*ssh.Client, error) {
+	auth := s.sshAuthMethod(s.Password, s.PkFile, s.PkPassword)
 	config := ssh.Config{
 		Ciphers: []string{"aes128-ctr", "aes192-ctr", "aes256-ctr", "aes128-gcm@openssh.com", "arcfour256", "arcfour128", "aes128-cbc", "3des-cbc", "aes192-cbc", "aes256-cbc"},
 	}
-	DefaultTimeout := time.Duration(1) * time.Minute
-	if ss.Timeout == nil {
-		ss.Timeout = &DefaultTimeout
+	DefaultTimeout := time.Duration(15) * time.Second
+	if s.Timeout == nil {
+		s.Timeout = &DefaultTimeout
 	}
 	clientConfig := &ssh.ClientConfig{
-		User:            ss.User,
-		Auth:            auth,
-		Timeout:         *ss.Timeout,
-		Config:          config,
-		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+		User:    s.User,
+		Auth:    auth,
+		Timeout: *s.Timeout,
+		Config:  config,
+		HostKeyCallback: func(hostname string, remote net.Addr, key ssh.PublicKey) error {
+			return nil
+		},
 	}
-
-	addr := ss.addrReformat(host)
+	ip, port := iputils.GetSSHHostIPAndPort(host)
+	addr := s.addrReformat(ip, port)
 	return ssh.Dial("tcp", addr, clientConfig)
 }
 
-func (ss *SSH) Connect(host string) (*ssh.Session, error) {
-	client, err := ss.connect(host)
+func (s *SSH) Connect(host string) (*ssh.Client, *ssh.Session, error) {
+	client, err := s.connect(host)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	session, err := client.NewSession()
 	if err != nil {
-		return nil, err
+		_ = client.Close()
+		return nil, nil, err
 	}
 
 	modes := ssh.TerminalModes{
-		ssh.ECHO:          0,     // disable echoing
+		ssh.ECHO:          0,     //disable echoing
 		ssh.TTY_OP_ISPEED: 14400, // input speed = 14.4kbaud
 		ssh.TTY_OP_OSPEED: 14400, // output speed = 14.4kbaud
 	}
 
 	if err := session.RequestPty("xterm", 80, 40, modes); err != nil {
-		return nil, err
+		_ = session.Close()
+		_ = client.Close()
+		return nil, nil, err
 	}
 
-	return session, nil
+	return client, session, nil
 }
 
-func (ss *SSH) sshAuthMethod(passwd, pkFile, pkPasswd string) (auth []ssh.AuthMethod) {
-	// pkfile存在， 就进行密钥验证， 如果不存在，则跳过密钥验证。
+func (s *SSH) sshAuthMethod(password, pkFile, pkPasswd string) (auth []ssh.AuthMethod) {
 	if fileExist(pkFile) {
-		am, err := ss.sshPrivateKeyMethod(pkFile, pkPasswd)
-		// 获取到密钥验证就添加， 没获取到就直接跳过。
+		am, err := s.sshPrivateKeyMethod(pkFile, pkPasswd)
 		if err == nil {
 			auth = append(auth, am)
 		}
 	}
-	// 密码不为空， 则添加密码验证。
-	if passwd != "" {
-		auth = append(auth, ss.sshPasswordMethod(passwd))
+	if password != "" {
+		auth = append(auth, s.sshPasswordMethod(password))
 	}
-
 	return auth
 }
 
-func fileExist(path string) bool {
-	_, err := os.Stat(path)
-	return err == nil || os.IsExist(err)
-}
+//Authentication with a private key,private key has password and no password to verify in this
+func (s *SSH) sshPrivateKeyMethod(pkFile, pkPassword string) (am ssh.AuthMethod, err error) {
+	pkData, err := ioutil.ReadFile(filepath.Clean(pkFile))
+	if err != nil {
+		return nil, err
+	}
 
-// 使用 pk认证， pk路径为 "$HOME/.ssh/id_rsa", pk有密码和无密码在这里面验证
-func (ss *SSH) sshPrivateKeyMethod(pkFile, pkPassword string) (am ssh.AuthMethod, err error) {
-	pkData := ss.readFile(pkFile)
 	var pk ssh.Signer
 	if pkPassword == "" {
 		pk, err = ssh.ParsePrivateKey(pkData)
@@ -115,24 +121,48 @@ func (ss *SSH) sshPrivateKeyMethod(pkFile, pkPassword string) (am ssh.AuthMethod
 	return ssh.PublicKeys(pk), nil
 }
 
-func (ss *SSH) sshPasswordMethod(passwd string) ssh.AuthMethod {
-	return ssh.Password(passwd)
+func fileExist(path string) bool {
+	_, err := os.Stat(path)
+	return err == nil || os.IsExist(err)
+}
+func (s *SSH) sshPasswordMethod(password string) ssh.AuthMethod {
+	return ssh.Password(password)
 }
 
-// readFile 从文件读取privateKey， 并返回[]byte 无需返回string。
-// 直接返回[]byte， 避免重复 []byte -> string -> []byte
-func (ss *SSH) readFile(name string) []byte {
-	content, err := ioutil.ReadFile(name)
-	if err != nil {
-		logger.Error("[globals] read %s file err is : %s", name, err)
-		os.Exit(1)
-	}
-	return content
-}
-
-func (ss *SSH) addrReformat(host string) string {
+func (s *SSH) addrReformat(host, port string) string {
 	if !strings.Contains(host, ":") {
-		host = fmt.Sprintf("%s:22", host)
+		host = fmt.Sprintf("%s:%s", host, port)
 	}
 	return host
 }
+
+//RemoteFileExist is
+func (s *SSH) IsFileExist(host, remoteFilePath string) bool {
+	// if remote file is
+	// ls -l | grep aa | wc -l
+	remoteFileName := path.Base(remoteFilePath) // aa
+	remoteFileDirName := path.Dir(remoteFilePath)
+	//it's bug: if file is aa.bak, `ls -l | grep aa | wc -l` is 1 ,should use `ll aa 2>/dev/null |wc -l`
+	//remoteFileCommand := fmt.Sprintf("ls -l %s| grep %s | grep -v grep |wc -l", remoteFileDirName, remoteFileName)
+	remoteFileCommand := fmt.Sprintf("ls -l %s/%s 2>/dev/null |wc -l", remoteFileDirName, remoteFileName)
+
+	data, err := s.CmdToString(host, remoteFileCommand, " ")
+	defer func() {
+		if r := recover(); r != nil {
+			logger.Error("[ssh][%s]remoteFileCommand err:%s", host, err)
+		}
+	}()
+	if err != nil {
+		panic(1)
+	}
+	count, err := strconv.Atoi(strings.TrimSpace(data))
+	defer func() {
+		if r := recover(); r != nil {
+			logger.Error("[ssh][%s]RemoteFileExist:%s", host, err)
+		}
+	}()
+	if err != nil {
+		panic(1)
+	}
+	return count != 0
+}

pkg/utils/ssh/hostname.go

@@ -1,50 +0,0 @@
-/*
-Copyright 2021 cuisongliu@qq.com.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package ssh
-
-import (
-	"strings"
-
-	"github.com/fanux/sealos/pkg/utils/logger"
-)
-
-//HostName 判断当前host的hostname
-func HostName(sshConfig SSH, master, host string) string {
-	hostString := sshConfig.CmdToString(master, "kubectl get nodes | grep -v NAME  | awk '{print $1}'", ",")
-	hostName := sshConfig.CmdToString(host, "hostname", "")
-	logger.Debug("hosts %v", hostString)
-	hosts := strings.Split(hostString, ",")
-	var name string
-	for _, h := range hosts {
-		if strings.TrimSpace(h) == "" {
-			continue
-		} else {
-			hh := strings.ToLower(h)
-			fromH := strings.ToLower(hostName)
-			if hh == fromH {
-				name = h
-				break
-			}
-		}
-	}
-	return name
-}
-
-func RemoteHostName(sshConfig SSH, hostIP string) string {
-	hostName := sshConfig.CmdToString(hostIP, "hostname", "")
-	return strings.ToLower(hostName)
-}

pkg/utils/ssh/scp.go

@@ -1,4 +1,4 @@
-// Copyright © 2021 sealos.
+// Copyright © 2021 Alibaba Group Holding Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,444 +15,244 @@
 package ssh
 
 import (
-	"bytes"
 	"fmt"
 	"io"
 	"io/ioutil"
-	"net"
 	"os"
 	"path"
 	"path/filepath"
 	"strings"
-	"sync"
-	"time"
 
 	"github.com/fanux/sealos/pkg/utils/hash"
 
-	"github.com/fanux/sealos/pkg/utils/http"
+	"github.com/fanux/sealos/pkg/utils/file"
+	"github.com/fanux/sealos/pkg/utils/iputils"
 
 	"github.com/fanux/sealos/pkg/utils/logger"
+	"github.com/fanux/sealos/pkg/utils/progress"
+	"github.com/schollz/progressbar/v3"
 
 	"github.com/pkg/sftp"
 	"golang.org/x/crypto/ssh"
 )
 
-//Copy is
-func (ss *SSH) CopyForMD5(host, localFilePath, remoteFilePath, md5 string) bool {
-	//如果有md5则可以验证
-	//如果没有md5则拿到本地数据后验证
-	if md5 == "" {
-		md5 = hash.FileMD5(localFilePath)
-	}
-	logger.Debug("[ssh]source file md5 value is %s", md5)
-	ss.Copy(host, localFilePath, remoteFilePath)
-	remoteMD5 := ss.Md5Sum(host, remoteFilePath)
-	logger.Debug("[ssh]host: %s , remote md5: %s", host, remoteMD5)
-	remoteMD5 = strings.TrimSpace(remoteMD5)
-	md5 = strings.TrimSpace(md5)
-	if remoteMD5 == md5 {
-		logger.Info("[ssh]md5 validate true")
-		return true
-	}
-	logger.Error("[ssh]md5 validate false")
-	return false
-}
-
-func (ss *SSH) Md5Sum(host, remoteFilePath string) string {
+func (s *SSH) RemoteMd5Sum(host, remoteFilePath string) string {
 	cmd := fmt.Sprintf("md5sum %s | cut -d\" \" -f1", remoteFilePath)
-	remoteMD5 := ss.CmdToString(host, cmd, "")
+	remoteMD5, err := s.CmdToString(host, cmd, "")
+	if err != nil {
+		logger.Error("count remote md5 failed %s %s %v", host, remoteFilePath, err)
+	}
 	return remoteMD5
 }
 
-//Copy is
-func (ss *SSH) Copy(host, localFilePath, remoteFilePath string) {
-	sftpClient, err := ss.sftpConnect(host)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]scpCopy: %s", host, err)
-		}
-	}()
+//CmdToString is in host exec cmd and replace to spilt str
+func (s *SSH) CmdToString(host, cmd, spilt string) (string, error) {
+	data, err := s.Cmd(host, cmd)
+	str := string(data)
 	if err != nil {
-		panic(1)
+		return str, fmt.Errorf("exec remote command failed %s %s %v", host, cmd, err)
 	}
-	defer sftpClient.Close()
-	srcFile, err := os.Open(localFilePath)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]scpCopy: %s", host, err)
-		}
-	}()
-	if err != nil {
-		panic(1)
-	}
-	defer srcFile.Close()
-
-	dstFile, err := sftpClient.Create(remoteFilePath)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]scpCopy: %s", host, err)
-		}
-	}()
-	if err != nil {
-		panic(1)
-	}
-	defer dstFile.Close()
-	buf := make([]byte, 100*oneMBByte) //100mb
-	total := 0
-	unit := ""
-	for {
-		n, _ := srcFile.Read(buf)
-		if n == 0 {
-			break
-		}
-		length, _ := dstFile.Write(buf[0:n])
-		isKb := length/oneMBByte < 1
-		speed := 0
-		if isKb {
-			total += length
-			unit = "KB"
-			speed = length / oneKBByte
-		} else {
-			total += length
-			unit = "MB"
-			speed = length / oneMBByte
-		}
-		totalLength, totalUnit := toSizeFromInt(total)
-		logger.Info("[ssh][%s]transfer total size is: %.2f%s ;speed is %d%s", host, totalLength, totalUnit, speed, unit)
-	}
-}
-
-//Copy is
-func (ss *SSH) CopyConfigFile(host, remoteFilePath string, localFilePathOrBytes interface{}) {
-	var (
-		data io.Reader
-	)
-	sftpClient, err := ss.sftpConnect(host)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]scpCopy: %s", host, err)
-		}
-	}()
-	if err != nil {
-		panic(1)
-	}
-	defer sftpClient.Close()
-
-	switch v := localFilePathOrBytes.(type) {
-	case string:
-		srcFile, err := os.Open(v)
-		defer func() {
-			if r := recover(); r != nil {
-				logger.Error("[ssh][%s]scpCopy: %s", host, err)
-			}
-		}()
-		if err != nil {
-			panic(1)
-		}
-		defer srcFile.Close()
-		data = srcFile
-	case []byte:
-		data = bytes.NewReader(v)
-	default:
-		panic("must use path or []bytes")
-	}
-
-	dstFile, err := sftpClient.Create(remoteFilePath)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]scpCopy: %s", host, err)
-		}
-	}()
-	if err != nil {
-		panic(1)
-	}
-	defer dstFile.Close()
-	buf := make([]byte, 100*oneMBByte) //100mb
-	totalMB := 0
-	for {
-		n, _ := data.Read(buf)
-		if n == 0 {
-			break
-		}
-		length, _ := dstFile.Write(buf[0:n])
-		totalMB += length / oneMBByte
-		logger.Info("[ssh][%s]transfer total size is: %d%s", host, totalMB, "MB")
+	if data != nil {
+		str = strings.ReplaceAll(str, "\r\n", spilt)
+		str = strings.ReplaceAll(str, "\n", spilt)
+		return str, nil
 	}
+	return str, fmt.Errorf("command %s %s return nil", host, cmd)
 }
 
 //SftpConnect  is
-func (ss *SSH) sftpConnect(host string) (*sftp.Client, error) {
-	var (
-		auth         []ssh.AuthMethod
-		addr         string
-		clientConfig *ssh.ClientConfig
-		sshClient    *ssh.Client
-		sftpClient   *sftp.Client
-		err          error
-	)
-	// get auth method
-	auth = ss.sshAuthMethod(ss.Password, ss.PkFile, ss.PkPassword)
-
-	clientConfig = &ssh.ClientConfig{
-		User:    ss.User,
-		Auth:    auth,
-		Timeout: 30 * time.Second,
-		HostKeyCallback: func(hostname string, remote net.Addr, key ssh.PublicKey) error {
-			return nil
-		},
-		Config: ssh.Config{
-			Ciphers: []string{"aes128-ctr", "aes192-ctr", "aes256-ctr", "aes128-gcm@openssh.com", "arcfour256", "arcfour128", "aes128-cbc", "3des-cbc", "aes192-cbc", "aes256-cbc"},
-		},
-	}
-
-	// connet to ssh
-	addr = ss.addrReformat(host)
-
-	if sshClient, err = ssh.Dial("tcp", addr, clientConfig); err != nil {
-		return nil, err
+func (s *SSH) sftpConnect(host string) (*ssh.Client, *sftp.Client, error) {
+	sshClient, err := s.connect(host)
+	if err != nil {
+		return nil, nil, err
 	}
 
 	// create sftp client
-	if sftpClient, err = sftp.NewClient(sshClient); err != nil {
-		return nil, err
-	}
-
-	return sftpClient, nil
+	sftpClient, err := sftp.NewClient(sshClient)
+	return sshClient, sftpClient, err
 }
 
 // CopyRemoteFileToLocal is scp remote file to local
-func (ss *SSH) CopyRemoteFileToLocal(host, localFilePath, remoteFilePath string) {
-	sftpClient, err := ss.sftpConnect(host)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]scpCopy: %s", host, err)
+func (s *SSH) Fetch(host, localFilePath, remoteFilePath string) error {
+	if iputils.IsLocalIP(host, s.LocalAddress) {
+		if remoteFilePath != localFilePath {
+			logger.Debug("local copy files src %s to dst %s", remoteFilePath, localFilePath)
+			return file.RecursionCopy(remoteFilePath, localFilePath)
 		}
-	}()
-	if err != nil {
-		panic(1)
+		return nil
 	}
-	defer sftpClient.Close()
+	sshClient, sftpClient, err := s.sftpConnect(host)
+	if err != nil {
+		return fmt.Errorf("new sftp client failed %v", err)
+	}
+	defer func() {
+		_ = sftpClient.Close()
+		_ = sshClient.Close()
+	}()
 	// open remote source file
 	srcFile, err := sftpClient.Open(remoteFilePath)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]scpCopy: %s", host, err)
-		}
-	}()
 	if err != nil {
-		panic(1)
+		return fmt.Errorf("open remote file failed %v, remote path: %s", err, remoteFilePath)
 	}
 	defer srcFile.Close()
 
+	err = file.MkFileFullPathDir(localFilePath)
+	if err != nil {
+		return err
+	}
 	// open local Destination file
 	dstFile, err := os.Create(localFilePath)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]scpCopy: %s", host, err)
-		}
-	}()
 	if err != nil {
-		panic(1)
+		return fmt.Errorf("create local file failed %v", err)
 	}
 	defer dstFile.Close()
 	// copy to local file
-	_, _ = srcFile.WriteTo(dstFile)
+	_, err = srcFile.WriteTo(dstFile)
+	return err
 }
 
 // CopyLocalToRemote is copy file or dir to remotePath, add md5 validate
-func (ss *SSH) CopyLocalToRemote(host, localPath, remotePath string) {
-	sftpClient, err := ss.sftpConnect(host)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]sshConnect err: %s", host, err)
-		}
-	}()
-	if err != nil {
-		panic(1)
+func (s *SSH) Copy(host, localPath, remotePath string) error {
+	if iputils.IsLocalIP(host, s.LocalAddress) {
+		logger.Debug("local copy files src %s to dst %s", localPath, remotePath)
+		return file.RecursionCopy(localPath, remotePath)
 	}
-	sshClient, err := ss.connect(host)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]scpConnect err: %s", host, err)
-		}
-	}()
+	logger.Debug("remote copy files src %s to dst %s", localPath, remotePath)
+	sshClient, sftpClient, err := s.sftpConnect(host)
 	if err != nil {
-		panic(1)
+		return fmt.Errorf("new sftp client failed %s", err)
 	}
-	defer sftpClient.Close()
-	defer sshClient.Close()
-	s, _ := os.Stat(localPath)
-	if s.IsDir() {
-		ss.copyLocalDirToRemote(host, sshClient, sftpClient, localPath, remotePath)
+	defer func() {
+		_ = sftpClient.Close()
+		_ = sshClient.Close()
+	}()
+
+	f, err := os.Stat(localPath)
+	if err != nil {
+		return fmt.Errorf("get file stat failed %s", err)
+	}
+
+	baseRemoteFilePath := filepath.Dir(remotePath)
+	_, err = sftpClient.ReadDir(baseRemoteFilePath)
+	if err != nil {
+		if err = sftpClient.MkdirAll(baseRemoteFilePath); err != nil {
+			return err
+		}
+	}
+	number := 1
+	if f.IsDir() {
+		number = file.CountDirFiles(localPath)
+	}
+	// no file in dir, do need to send
+	if number == 0 {
+		return nil
+	}
+	var (
+		bar = progress.Simple("copying files to "+host, number)
+	)
+	defer func() {
+		_ = bar.Close()
+	}()
+
+	if f.IsDir() {
+		s.copyLocalDirToRemote(host, sftpClient, localPath, remotePath, bar)
 	} else {
-		baseRemoteFilePath := filepath.Dir(remotePath)
-		mkDstDir := fmt.Sprintf("mkdir -p %s || true", baseRemoteFilePath)
-		_ = ss.CmdAsync(host, mkDstDir)
-		ss.copyLocalFileToRemote(host, sshClient, sftpClient, localPath, remotePath)
+		err = s.copyLocalFileToRemote(host, sftpClient, localPath, remotePath)
+		if err != nil {
+			errMsg := fmt.Sprintf("copy local file to remote failed %v %s %s %s", err, host, localPath, remotePath)
+			logger.Error(errMsg)
+		}
+		_ = bar.Add(1)
 	}
+	return nil
 }
 
-// ssh session is a problem, 复用ssh链接
-func (ss *SSH) copyLocalDirToRemote(host string, sshClient *ssh.Client, sftpClient *sftp.Client, localPath, remotePath string) {
+func (s *SSH) copyLocalDirToRemote(host string, sftpClient *sftp.Client, localPath, remotePath string, epu *progressbar.ProgressBar) {
 	localFiles, err := ioutil.ReadDir(localPath)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("readDir err : %s", err)
-		}
-	}()
 	if err != nil {
-		panic(1)
+		logger.Error("read local path dir failed %s %s", host, localPath)
+		return
+	}
+	if err = sftpClient.MkdirAll(remotePath); err != nil {
+		logger.Error("failed to create remote path %s:%v", remotePath, err)
+		return
 	}
-	_ = sftpClient.Mkdir(remotePath)
 	for _, file := range localFiles {
 		lfp := path.Join(localPath, file.Name())
 		rfp := path.Join(remotePath, file.Name())
 		if file.IsDir() {
-			_ = sftpClient.Mkdir(rfp)
-			ss.copyLocalDirToRemote(host, sshClient, sftpClient, lfp, rfp)
+			if err = sftpClient.MkdirAll(rfp); err != nil {
+				logger.Error("failed to create remote path %s:%v", rfp, err)
+				return
+			}
+			s.copyLocalDirToRemote(host, sftpClient, lfp, rfp, epu)
 		} else {
-			ss.copyLocalFileToRemote(host, sshClient, sftpClient, lfp, rfp)
+			err := s.copyLocalFileToRemote(host, sftpClient, lfp, rfp)
+			if err != nil {
+				errMsg := fmt.Sprintf("copy local file to remote failed %v %s %s %s", err, host, lfp, rfp)
+				logger.Error(errMsg)
+				return
+			}
+			_ = epu.Add(1)
 		}
 	}
 }
 
-// solve the session
-func (ss *SSH) copyLocalFileToRemote(host string, sshClient *ssh.Client, sftpClient *sftp.Client, localPath, remotePath string) {
-	srcFile, err := os.Open(localPath)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("open file [%s] err : %s", localPath, err)
+// check the remote file existence before copying
+// solve the sesion
+func (s *SSH) copyLocalFileToRemote(host string, sftpClient *sftp.Client, localPath, remotePath string) error {
+	var (
+		srcMd5, dstMd5 string
+	)
+	srcMd5 = hash.FileMD5(localPath)
+	if s.IsFileExist(host, remotePath) {
+		dstMd5 = s.RemoteMd5Sum(host, remotePath)
+		if srcMd5 == dstMd5 {
+			logger.Debug("remote dst %s already exists and is the latest version , skip copying process", remotePath)
+			return nil
 		}
-	}()
+	}
+	srcFile, err := os.Open(filepath.Clean(localPath))
 	if err != nil {
-		panic(1)
+		return err
 	}
 	defer srcFile.Close()
 	dstFile, err := sftpClient.Create(remotePath)
 	if err != nil {
-		logger.Error("err:", err)
+		return err
+	}
+	fileStat, err := srcFile.Stat()
+	if err != nil {
+		return fmt.Errorf("get file stat failed %v", err)
+	}
+	// TODO seems not work
+	if err := dstFile.Chmod(fileStat.Mode()); err != nil {
+		return fmt.Errorf("chmod remote file failed %v", err)
 	}
 	defer dstFile.Close()
-	buf := make([]byte, 100*oneMBByte) //100mb
-	total := 0
-	unit := ""
-	for {
-		n, _ := srcFile.Read(buf)
-		if n == 0 {
-			break
-		}
-		length, _ := dstFile.Write(buf[0:n])
-		isKb := length/oneMBByte < 1
-		speed := 0
-		if isKb {
-			total += length
-			unit = "KB"
-			speed = length / oneKBByte
-		} else {
-			total += length
-			unit = "MB"
-			speed = length / oneMBByte
-		}
-		totalLength, totalUnit := toSizeFromInt(total)
-		logger.Debug("[ssh][%s]transfer local [%s] to Dst [%s] total size is: %.2f%s ;speed is %d%s", host, localPath, remotePath, totalLength, totalUnit, speed, unit)
+	_, err = io.Copy(dstFile, srcFile)
+	if err != nil {
+		return err
 	}
-	if !ss.isCopyMd5Success(sshClient, localPath, remotePath) {
-		//	logger.Debug("[ssh][%s] copy local file: %s to remote file: %s validate md5sum success", host, localPath, remotePath)
-		//} else {
-		logger.Error("[ssh][%s] copy local file: %s to remote file: %s validate md5sum failed", host, localPath, remotePath)
+	dstMd5 = s.RemoteMd5Sum(host, remotePath)
+	if srcMd5 != dstMd5 {
+		return fmt.Errorf("[ssh][%s] validate md5sum failed %s != %s", host, srcMd5, dstMd5)
 	}
+	return nil
 }
 
-func (ss *SSH) isCopyMd5Success(sshClient *ssh.Client, localFile, remoteFile string) bool {
-	cmd := fmt.Sprintf("md5sum %s | cut -d\" \" -f1", remoteFile)
-	localMd5 := hash.FileMD5(localFile)
-	sshSession, err := sshClient.NewSession()
+//if remote file not exist return false and nil
+func (s *SSH) RemoteDirExist(host, remoteDirpath string) (bool, error) {
+	sshClient, sftpClient, err := s.sftpConnect(host)
 	if err != nil {
-		return false
+		return false, err
 	}
-	modes := ssh.TerminalModes{
-		ssh.ECHO:          0,     // disable echoing
-		ssh.TTY_OP_ISPEED: 14400, // input speed = 14.4kbaud
-		ssh.TTY_OP_OSPEED: 14400, // output speed = 14.4kbaud
-	}
-
-	if err := sshSession.RequestPty("xterm", 80, 40, modes); err != nil {
-		return false
-	}
-
-	if sshSession.Stdout != nil {
-		sshSession.Stdout = nil
-		sshSession.Stderr = nil
-	}
-	b, err := sshSession.CombinedOutput(cmd)
 	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh]Error exec command failed: %s", err)
-		}
+		_ = sftpClient.Close()
+		_ = sshClient.Close()
 	}()
-	if err != nil {
-		panic(1)
+	if _, err := sftpClient.ReadDir(remoteDirpath); err != nil {
+		return false, err
 	}
-	var remoteMd5 string
-	if b != nil {
-		remoteMd5 = string(b)
-		remoteMd5 = strings.ReplaceAll(remoteMd5, "\r\n", "")
-	}
-	return localMd5 == remoteMd5
-}
-
-func (ss *SSH) ValidateMd5sumLocalWithRemote(host, localFile, remoteFile string) bool {
-	localMd5 := hash.FileMD5(localFile)
-	return localMd5 == ss.Md5Sum(host, remoteFile)
-}
-
-//location : url
-//md5
-//dst: /root
-//hook: cd /root && rm -rf kube && tar zxvf %s  && cd /root/kube/shell && sh init.sh
-func CopyFiles(sshConfig SSH, location string, hosts []string, dst string, before, after *string) string {
-	var md5 string
-	location, md5 = http.DownloadFile(location)
-	pkg := path.Base(location)
-	fullPath := fmt.Sprintf("%s/%s", dst, pkg)
-	mkDstDir := fmt.Sprintf("mkdir -p %s || true", dst)
-	var wm sync.WaitGroup
-	for _, host := range hosts {
-		wm.Add(1)
-		go func(host string) {
-			defer wm.Done()
-			_ = sshConfig.CmdAsync(host, mkDstDir)
-			logger.Debug("[%s]please wait for mkDstDir", host)
-			if before != nil {
-				logger.Debug("[%s]please wait for before hook", host)
-				_ = sshConfig.CmdAsync(host, *before)
-			}
-			if sshConfig.IsFileExist(host, fullPath) {
-				if sshConfig.ValidateMd5sumLocalWithRemote(host, location, fullPath) {
-					logger.Info("[%s]CopyFiles:  %s file is exist and ValidateMd5 success", host, fullPath)
-				} else {
-					rm := fmt.Sprintf("rm -f %s", fullPath)
-					_ = sshConfig.Cmd(host, rm)
-					// del then copy
-					if ok := sshConfig.CopyForMD5(host, location, fullPath, md5); ok {
-						logger.Info("[%s]copy file md5 validate success", host)
-					} else {
-						logger.Error("[%s]copy file md5 validate failed", host)
-					}
-				}
-			} else {
-				if ok := sshConfig.CopyForMD5(host, location, fullPath, md5); ok {
-					logger.Info("[%s]copy file md5 validate success", host)
-				} else {
-					logger.Error("[%s]copy file md5 validate failed", host)
-				}
-			}
-			if after != nil {
-				logger.Debug("[%s]please wait for after hook", host)
-				_ = sshConfig.CmdAsync(host, *after)
-			}
-		}(host)
-	}
-	wm.Wait()
-	return location
+	return true, nil
 }

pkg/utils/ssh/scp_test.go

@@ -1,89 +0,0 @@
-// Copyright © 2021 sealos.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package ssh
-
-import (
-	"testing"
-
-	"github.com/fanux/sealos/pkg/utils/logger"
-)
-
-func TestSSHCopyLocalToRemote(t *testing.T) {
-	type args struct {
-		host       string
-		localPath  string
-		remotePath string
-	}
-	var (
-		host = "192.168.160.243"
-		ssh  = SSH{
-			User:       "root",
-			Password:   "centos",
-			PkFile:     "",
-			PkPassword: "",
-			Timeout:    nil,
-		}
-	)
-	tests := []struct {
-		name   string
-		fields SSH
-		args   args
-	}{
-		{"test for copy file to remote server", ssh, args{
-			host,
-			"/home/louis/temp/01",
-			"/data/temp/01",
-		}},
-		{"test copy for file when local file is not exist", ssh, args{
-			host,
-			// local file  001 is not exist.
-			"/home/louis/temp/001",
-			"/data/temp/01",
-		}},
-		{"test copy dir to remote server", ssh, args{
-			host,
-			"/home/louis/temp",
-			"/data/temp01",
-		}},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			ss := &SSH{
-				User:       tt.fields.User,
-				Password:   tt.fields.Password,
-				PkFile:     tt.fields.PkFile,
-				PkPassword: tt.fields.PkPassword,
-				Timeout:    tt.fields.Timeout,
-			}
-
-			if !fileExist(tt.args.localPath) {
-				logger.Error("local filepath is not exit")
-				return
-			}
-			if ss.IsFileExist(host, tt.args.remotePath) {
-				logger.Error("remote filepath is exit")
-				return
-			}
-			// test copy dir
-			ss.CopyLocalToRemote(tt.args.host, tt.args.localPath, tt.args.remotePath)
-
-			// test the copy result
-			ss.Cmd(tt.args.host, "ls -lh "+tt.args.remotePath)
-
-			// rm remote file
-			ss.Cmd(tt.args.host, "rm -rf "+tt.args.remotePath)
-		})
-	}
-}

pkg/utils/ssh/ssh.go

@@ -1,4 +1,4 @@
-// Copyright © 2021 sealos.
+// Copyright © 2021 Alibaba Group Holding Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,102 +15,142 @@
 package ssh
 
 import (
-	"bufio"
-	"io"
-	"strings"
-
-	"github.com/fanux/sealos/pkg/utils/logger"
+	"fmt"
+	"net"
+	"sync"
+	"time"
 )
 
-//Cmd is in host exec cmd
-func (ss *SSH) Cmd(host string, cmd string) []byte {
-	logger.Info("[ssh][%s] %s", host, cmd)
-	session, err := ss.Connect(host)
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]Error create ssh session failed,%s", host, err)
-		}
-	}()
-	if err != nil {
-		panic(1)
-	}
-	defer session.Close()
-	b, err := session.CombinedOutput(cmd)
-	logger.Debug("[ssh][%s]command result is: %s", host, string(b))
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]Error exec command failed: %s", host, err)
-		}
-	}()
-	if err != nil {
-		panic(1)
-	}
-	return b
+type Interface interface {
+	// Copy is copy local files to remote host
+	// scp -r /tmp root@192.168.0.2:/root/tmp => Copy("192.168.0.2","tmp","/root/tmp")
+	// need check md5sum
+	Copy(host, srcFilePath, dstFilePath string) error
+	// Fetch is copy remote host files to localhost
+	Fetch(host, srcFilePath, dstFilePath string) error
+	// CmdAsync is exec command on remote host, and asynchronous return logs
+	CmdAsync(host string, cmd ...string) error
+	// Cmd is exec command on remote host, and return combined standard output and standard error
+	Cmd(host, cmd string) ([]byte, error)
+	// IsFileExist is check remote file exist or not
+	IsFileExist(host, remoteFilePath string) bool
+	//RemoteDirExist is remote file existence returns true, nil
+	RemoteDirExist(host, remoteDirpath string) (bool, error)
+	//CmdToString is exec command on remote host, and return spilt standard output and standard error
+	CmdToString(host, cmd, spilt string) (string, error)
+	Ping(host string) error
 }
 
-func readPipe(host string, pipe io.Reader, isErr bool) {
-	r := bufio.NewReader(pipe)
-	for {
-		line, _, err := r.ReadLine()
-		if line == nil {
-			return
-		} else if err != nil {
-			logger.Info("[%s] %s", host, line)
-			logger.Error("[ssh] [%s] %s", host, err)
-			return
-		} else {
-			if isErr {
-				logger.Error("[%s] %s", host, line)
-			} else {
-				logger.Info("[%s] %s", host, line)
+type SSH struct {
+	User         string
+	Password     string
+	PkFile       string
+	PkPassword   string
+	Timeout      *time.Duration
+	LocalAddress *[]net.Addr
+}
+
+//func NewSSHByCluster(cluster *v1.Cluster) Interface {
+//	if cluster.Spec.SSH.User == "" {
+//		cluster.Spec.SSH.User = common.ROOT
+//	}
+//	address, err := utils.IsLocalHostAddrs()
+//	if err != nil {
+//		logger.Warn("failed to get local address, %v", err)
+//	}
+//	return &SSH{
+//		User:         cluster.Spec.SSH.User,
+//		Password:     cluster.Spec.SSH.Passwd,
+//		PkFile:       cluster.Spec.SSH.Pk,
+//		PkPassword:   cluster.Spec.SSH.PkPasswd,
+//		LocalAddress: address,
+//	}
+//}
+//
+//func NewSSHClient(ssh *v1.SSH) Interface {
+//	if ssh.User == "" {
+//		ssh.User = common.ROOT
+//	}
+//	address, err := utils.IsLocalHostAddrs()
+//	if err != nil {
+//		logger.Warn("failed to get local address, %v", err)
+//	}
+//	return &SSH{
+//		User:         ssh.User,
+//		Password:     ssh.Passwd,
+//		PkFile:       ssh.Pk,
+//		PkPassword:   ssh.PkPasswd,
+//		LocalAddress: address,
+//	}
+//}
+//
+//func GetHostSSHClient(hostIP string, cluster *v2.Cluster) (Interface, error) {
+//	for _, host := range cluster.Spec.Hosts {
+//		for _, ip := range host.IPS {
+//			if hostIP == ip {
+//				if err := mergo.Merge(&host.SSH, &cluster.Spec.SSH); err != nil {
+//					return nil, err
+//				}
+//
+//				return NewSSHClient(&host.SSH), nil
+//			}
+//		}
+//	}
+//	return nil, fmt.Errorf("get host ssh client failed, host ip %s not in hosts ip list", hostIP)
+//}
+//
+
+type Client struct {
+	SSH  Interface
+	Host string
+}
+//
+//func NewSSHClientWithCluster(cluster *v1.Cluster) (*Client, error) {
+//	var (
+//		ipList []string
+//		host   string
+//	)
+//	sshClient := NewSSHByCluster(cluster)
+//	if cluster.Spec.Provider == common.AliCloud {
+//		host = cluster.GetAnnotationsByKey(common.Eip)
+//		if host == "" {
+//			return nil, fmt.Errorf("get cluster EIP failed")
+//		}
+//		ipList = append(ipList, host)
+//	} else {
+//		host = cluster.Spec.Masters.IPList[0]
+//		ipList = append(ipList, append(cluster.Spec.Masters.IPList, cluster.Spec.Nodes.IPList...)...)
+//	}
+//	err := WaitSSHReady(sshClient, 6, ipList...)
+//	if err != nil {
+//		return nil, err
+//	}
+//	if sshClient == nil {
+//		return nil, fmt.Errorf("cloud build init ssh client failed")
+//	}
+//	return &Client{
+//		SSH:  sshClient,
+//		Host: host,
+//	}, nil
+//}
+
+func WaitSSHReady(ssh Interface, tryTimes int, hosts ...string) error {
+	var err error
+	var wg sync.WaitGroup
+	for _, h := range hosts {
+		wg.Add(1)
+		go func(host string) {
+			defer wg.Done()
+			for i := 0; i < tryTimes; i++ {
+				err = ssh.Ping(host)
+				if err == nil {
+					return
+				}
+				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)
+		}(h)
 	}
-}
-
-func (ss *SSH) CmdAsync(host string, cmd string) error {
-	logger.Debug("[%s] %s", host, cmd)
-	session, err := ss.Connect(host)
-	if err != nil {
-		logger.Error("[ssh][%s]Error create ssh session failed,%s", host, err)
-		return err
-	}
-	defer session.Close()
-	stdout, err := session.StdoutPipe()
-	if err != nil {
-		logger.Error("[ssh][%s]Unable to request StdoutPipe(): %s", host, err)
-		return err
-	}
-	stderr, err := session.StderrPipe()
-	if err != nil {
-		logger.Error("[ssh][%s]Unable to request StderrPipe(): %s", host, err)
-		return err
-	}
-	if err := session.Start(cmd); err != nil {
-		logger.Error("[ssh][%s]Unable to execute command: %s", host, err)
-		return err
-	}
-	doneout := make(chan bool, 1)
-	doneerr := make(chan bool, 1)
-	go func() {
-		readPipe(host, stderr, true)
-		doneerr <- true
-	}()
-	go func() {
-		readPipe(host, stdout, false)
-		doneout <- true
-	}()
-	<-doneerr
-	<-doneout
-	return session.Wait()
-}
-
-//CmdToString is in host exec cmd and replace to spilt str
-func (ss *SSH) CmdToString(host, cmd, spilt string) string {
-	if data := ss.Cmd(host, cmd); data != nil {
-		str := string(data)
-		str = strings.ReplaceAll(str, "\r\n", spilt)
-		return str
-	}
-	return ""
+	wg.Wait()
+	return err
 }

pkg/utils/ssh/sshcmd.go

@@ -0,0 +1,124 @@
+// Copyright © 2021 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ssh
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"strings"
+	"sync"
+
+	strings2 "github.com/fanux/sealos/pkg/utils/strings"
+)
+
+var DebugMode bool
+
+func (s *SSH) Ping(host string) error {
+	client, _, err := s.Connect(host)
+	if err != nil {
+		return fmt.Errorf("[ssh %s]create ssh session failed, %v", host, err)
+	}
+	err = client.Close()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *SSH) CmdAsync(host string, cmds ...string) error {
+	for _, cmd := range cmds {
+		if cmd == "" {
+			continue
+		}
+
+		if err := func(cmd string) error {
+			client, session, err := s.Connect(host)
+			if err != nil {
+				return fmt.Errorf("failed to create ssh session for %s: %v", host, err)
+			}
+			defer client.Close()
+			defer session.Close()
+			stdout, err := session.StdoutPipe()
+			if err != nil {
+				return fmt.Errorf("failed to create stdout pipe for %s: %v", host, err)
+			}
+			stderr, err := session.StderrPipe()
+			if err != nil {
+				return fmt.Errorf("failed to create stderr pipe for %s: %v", host, err)
+			}
+
+			if err := session.Start(cmd); err != nil {
+				return fmt.Errorf("failed to start command %s on %s: %v", cmd, host, err)
+			}
+
+			var combineSlice []string
+			var combineLock sync.Mutex
+			doneout := make(chan error, 1)
+			doneerr := make(chan error, 1)
+			go func() {
+				doneerr <- readPipe(stderr, &combineSlice, &combineLock)
+			}()
+			go func() {
+				doneout <- readPipe(stdout, &combineSlice, &combineLock)
+			}()
+			<-doneerr
+			<-doneout
+
+			err = session.Wait()
+			if err != nil {
+				return strings2.WrapExecResult(host, cmd, []byte(strings.Join(combineSlice, "\n")), err)
+			}
+
+			return nil
+		}(cmd); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *SSH) Cmd(host, cmd string) ([]byte, error) {
+	client, session, err := s.Connect(host)
+	if err != nil {
+		return nil, fmt.Errorf("[ssh][%s] create ssh session failed, %s", host, err)
+	}
+	defer client.Close()
+	defer session.Close()
+	b, err := session.CombinedOutput(cmd)
+	if err != nil {
+		return b, fmt.Errorf("[ssh][%s]run command failed [%s]", host, cmd)
+	}
+
+	return b, nil
+}
+
+func readPipe(pipe io.Reader, combineSlice *[]string, combineLock *sync.Mutex) error {
+	r := bufio.NewReader(pipe)
+	for {
+		line, _, err := r.ReadLine()
+		if err != nil {
+			return err
+		}
+
+		combineLock.Lock()
+		*combineSlice = append(*combineSlice, string(line))
+		if DebugMode {
+			fmt.Println(string(line))
+		}
+		combineLock.Unlock()
+	}
+}

pkg/utils/ssh/sshcmd_test.go

@@ -0,0 +1,217 @@
+// Copyright © 2021 Alibaba Group Holding Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ssh
+
+import (
+	"net"
+	"testing"
+)
+
+func TestSSH_Cmd(t *testing.T) {
+	type args struct {
+		ssh       SSH
+		host, cmd string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    string
+		wantErr bool
+	}{
+		{
+			name: "touch test.txt",
+			args: args{
+				ssh: SSH{
+					"root",
+					"huaijiahui.com",
+					"",
+					"",
+					nil,
+					&[]net.Addr{},
+				},
+				host: "192.168.56.103",
+				cmd:  "bash /opt/touchTxt.sh",
+			},
+			want:    "success touch test.txt\r\n", //命令返回值后缀为/r/n
+			wantErr: false,
+		},
+		{
+			name: "ls /opt/test",
+			args: args{
+				ssh: SSH{
+					"root",
+					"huaijiahui.com",
+					"",
+					"",
+					nil,
+					&[]net.Addr{},
+				},
+				host: "192.168.56.103",
+				cmd:  "ls /opt/test",
+			},
+			want:    "test.txt\r\n",
+			wantErr: false,
+		},
+		{
+			name: "remove test.txt",
+			args: args{
+				ssh: SSH{
+					"root",
+					"huaijiahui.com",
+					"",
+					"",
+					nil,
+					&[]net.Addr{},
+				},
+				host: "192.168.56.103",
+				cmd:  "bash /opt/removeTxt.sh",
+			},
+			want:    "test remove success\r\n",
+			wantErr: false,
+		},
+		{
+			name: "exist 1",
+			args: args{
+				ssh: SSH{
+					"root",
+					"huaijiahui.com",
+					"",
+					"",
+					nil,
+					&[]net.Addr{},
+				},
+				host: "192.168.56.103",
+				cmd:  "bash /opt/exit1.sh",
+			},
+			want:    "",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := tt.args.ssh.Cmd(tt.args.host, tt.args.cmd)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("Cmd err : %v,  wangErr is %v", err, tt.wantErr)
+			}
+
+			if string(got) != tt.want {
+				t.Errorf("got={%s},want={%s}", string(got), tt.want)
+			}
+		})
+	}
+}
+
+func TestSSH_CmdAsync(t *testing.T) {
+	type args struct {
+		ssh       SSH
+		host, cmd string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    string
+		wantErr bool
+	}{
+		{
+			name: "touch test.txt",
+			args: args{
+				ssh: SSH{
+					"root",
+					"huaijiahui.com",
+					"",
+					"",
+					nil,
+					&[]net.Addr{},
+				},
+				host: "192.168.56.103",
+				cmd:  "bash /opt/touchTxt.sh",
+			},
+			wantErr: false,
+		},
+		{
+			name: "ls /opt/test",
+			args: args{
+				ssh: SSH{
+					"root",
+					"huaijiahui.com",
+					"",
+					"",
+					nil,
+					&[]net.Addr{},
+				},
+				host: "192.168.56.103",
+				cmd:  "ls /opt/test",
+			},
+			wantErr: false,
+		},
+		{
+			name: "remove test.txt",
+			args: args{
+				ssh: SSH{
+					"root",
+					"huaijiahui.com",
+					"",
+					"",
+					nil,
+					&[]net.Addr{},
+				},
+				host: "192.168.56.103",
+				cmd:  "bash /opt/removeTxt.sh",
+			},
+			wantErr: false,
+		},
+		{
+			name: "exist 1",
+			args: args{
+				ssh: SSH{
+					"root",
+					"huaijiahui.com",
+					"",
+					"",
+					nil,
+					&[]net.Addr{},
+				},
+				host: "192.168.56.103",
+				cmd:  "bash /opt/exit1.sh",
+			},
+			wantErr: true, //Process exited with status 1
+		},
+		{
+			name: "exist 1",
+			args: args{
+				ssh: SSH{
+					"root",
+					"",
+					"/Users/cuisongliu/.ssh/id_rsa",
+					"",
+					nil,
+					&[]net.Addr{},
+				},
+				host: "192.168.64.15",
+				cmd:  "cd kube/scripts && bash check-node.sh",
+			},
+			wantErr: true, //Process exited with status 1
+		},
+	}
+	DebugMode = true
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := tt.args.ssh.CmdAsync(tt.args.host, tt.args.cmd); (err != nil) != tt.wantErr {
+				t.Errorf("Cmd err : %v,  wangErr is %v", err, tt.wantErr)
+			}
+		})
+	}
+}

pkg/utils/ssh/types.go

@@ -1,27 +0,0 @@
-// Copyright © 2021 sealos.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package ssh
-
-import (
-	"time"
-)
-
-type SSH struct {
-	User       string
-	Password   string
-	PkFile     string
-	PkPassword string
-	Timeout    *time.Duration
-}

pkg/utils/ssh/watch.go

@@ -1,80 +0,0 @@
-// Copyright © 2021 sealos.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package ssh
-
-import (
-	"fmt"
-	"path"
-	"strconv"
-	"strings"
-	"time"
-
-	"github.com/fanux/sealos/pkg/utils/logger"
-)
-
-const oneKBByte = 1024
-const oneMBByte = 1024 * 1024
-
-//WatchFileSize is
-func (ss *SSH) LoggerFileSize(host, filename string, size int) {
-	t := time.NewTicker(3 * time.Second) //every 3s check file
-	defer t.Stop()
-	for {
-		if <-t.C; true {
-			length := ss.CmdToString(host, "ls -l "+filename+" | awk '{print $5}'", "")
-			length = strings.Replace(length, "\n", "", -1)
-			length = strings.Replace(length, "\r", "", -1)
-			lengthByte, _ := strconv.Atoi(length)
-			if lengthByte == size {
-				t.Stop()
-			}
-			lengthFloat := float64(lengthByte)
-			value, _ := strconv.ParseFloat(fmt.Sprintf("%.2f", lengthFloat/oneMBByte), 64)
-			logger.Info("[ssh][%s]transfer total size is: %.2f%s", host, value, "MB")
-		}
-	}
-}
-
-//RemoteFileExist is
-func (ss *SSH) IsFileExist(host, remoteFilePath string) bool {
-	// if remote file is
-	// ls -l | grep aa | wc -l
-	remoteFileName := path.Base(remoteFilePath) // aa
-	remoteFileDirName := path.Dir(remoteFilePath)
-	//it's bug: if file is aa.bak, `ls -l | grep aa | wc -l` is 1 ,should use `ll aa 2>/dev/null |wc -l`
-	//remoteFileCommand := fmt.Sprintf("ls -l %s| grep %s | grep -v grep |wc -l", remoteFileDirName, remoteFileName)
-	remoteFileCommand := fmt.Sprintf("ls -l %s/%s 2>/dev/null |wc -l", remoteFileDirName, remoteFileName)
-
-	data := ss.CmdToString(host, remoteFileCommand, " ")
-	count, err := strconv.Atoi(strings.TrimSpace(data))
-	defer func() {
-		if r := recover(); r != nil {
-			logger.Error("[ssh][%s]RemoteFileExist:%s", host, err)
-		}
-	}()
-	if err != nil {
-		panic(1)
-	}
-	return count != 0
-}
-
-func toSizeFromInt(length int) (float64, string) {
-	value, _ := strconv.ParseFloat(fmt.Sprintf("%.2f", float64(length)/oneMBByte), 64)
-	if isMb := length/oneMBByte > 1; isMb {
-		return value, "MB"
-	}
-	value, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", float64(length)/oneKBByte), 64)
-	return value, "KB"
-}