实现远程升级节点

internal/installers/deploy_manager.go

@@ -0,0 +1,62 @@
+package installers
+
+import (
+	"github.com/iwind/TeaGo/Tea"
+	"github.com/iwind/TeaGo/files"
+	stringutil "github.com/iwind/TeaGo/utils/string"
+	"regexp"
+)
+
+var SharedDeployManager = NewDeployManager()
+
+type DeployFile struct {
+	OS      string
+	Arch    string
+	Version string
+	Path    string
+}
+
+type DeployManager struct {
+	dir string
+}
+
+func NewDeployManager() *DeployManager {
+	return &DeployManager{
+		dir: Tea.Root + "/deploy",
+	}
+}
+
+// 加载所有文件
+func (this *DeployManager) LoadFiles() []*DeployFile {
+	keyMap := map[string]*DeployFile{} // key => File
+
+	reg := regexp.MustCompile(`(\w+)-(\w+)-v([0-9.]+)\.zip`)
+	for _, file := range files.NewFile(this.dir).List() {
+		name := file.Name()
+		if !reg.MatchString(name) {
+			continue
+		}
+		matches := reg.FindStringSubmatch(name)
+		osName := matches[1]
+		arch := matches[2]
+		version := matches[3]
+
+		key := osName + "_" + arch
+		oldFile, ok := keyMap[key]
+		if ok && stringutil.VersionCompare(oldFile.Version, version) > 0 {
+			continue
+		}
+		keyMap[key] = &DeployFile{
+			OS:      osName,
+			Arch:    arch,
+			Version: version,
+			Path:    file.Path(),
+		}
+	}
+
+	result := []*DeployFile{}
+	for _, v := range keyMap {
+		result = append(result, v)
+	}
+	return result
+}

internal/installers/deploy_manager_test.go

@@ -0,0 +1,10 @@
+package installers
+
+import "testing"
+
+func TestDeployManager_LoadFiles(t *testing.T) {
+	files := NewDeployManager().LoadFiles()
+	for _, file := range files {
+		t.Logf("%#v", file)
+	}
+}

internal/installers/installer_node.go

@@ -42,15 +42,6 @@ func (this *NodeInstaller) Install(dir string, params interface{}, installStatus
 		return err
 	}
 
-	// 测试环境
-	_, stderr, err := this.client.Exec(dir + "/" + env.HelperName + " -cmd=test")
-	if err != nil {
-		return errors.New("test failed: " + err.Error())
-	}
-	if len(stderr) > 0 {
-		return errors.New("test failed: " + stderr)
-	}
-
 	// 上传安装文件
 	filePrefix := "edge-node-" + env.OS + "-" + env.Arch
 	zipFile, err := this.LookupLatestInstaller(filePrefix)
@@ -66,8 +57,35 @@ func (this *NodeInstaller) Install(dir string, params interface{}, installStatus
 		return err
 	}
 
+	// 测试运行环境
+	// 升级的节点暂时不列入测试
+	if !nodeParams.IsUpgrading {
+		_, stderr, err := this.client.Exec(dir + "/" + env.HelperName + " -cmd=test")
+		if err != nil {
+			return errors.New("test failed: " + err.Error())
+		}
+		if len(stderr) > 0 {
+			return errors.New("test failed: " + stderr)
+		}
+	}
+
+	// 如果是升级则优雅停止先前的进程
+	exePath := dir + "/edge-node/bin/edge-node"
+	if nodeParams.IsUpgrading {
+		_, err = this.client.Stat(exePath)
+		if err == nil {
+			_, _, _ = this.client.Exec(exePath + " quit")
+		}
+
+		// 删除可执行文件防止冲突
+		err = this.client.Remove(exePath)
+		if err != nil {
+			return errors.New("remove old file failed: " + err.Error())
+		}
+	}
+
 	// 解压
-	_, stderr, err = this.client.Exec(dir + "/" + env.HelperName + " -cmd=unzip -zip=\"" + targetZip + "\" -target=\"" + dir + "\"")
+	_, stderr, err := this.client.Exec(dir + "/" + env.HelperName + " -cmd=unzip -zip=\"" + targetZip + "\" -target=\"" + dir + "\"")
 	if err != nil {
 		return err
 	}

internal/installers/params_node.go

@@ -6,9 +6,10 @@ import (
 )
 
 type NodeParams struct {
-	Endpoints []string
-	NodeId    string
-	Secret    string
+	Endpoints   []string
+	NodeId      string
+	Secret      string
+	IsUpgrading bool // 是否为升级
 }
 
 func (this *NodeParams) Validate() error {

internal/installers/queue.go

@@ -24,7 +24,7 @@ func SharedQueue() *Queue {
 }
 
 // 安装边缘节点流程控制
-func (this *Queue) InstallNodeProcess(nodeId int64) error {
+func (this *Queue) InstallNodeProcess(nodeId int64, isUpgrading bool) error {
 	installStatus := models.NewNodeInstallStatus()
 	installStatus.IsRunning = true
 	installStatus.UpdatedAt = time.Now().Unix()
@@ -51,7 +51,7 @@ func (this *Queue) InstallNodeProcess(nodeId int64) error {
 	}()
 
 	// 开始安装
-	err = this.InstallNode(nodeId, installStatus)
+	err = this.InstallNode(nodeId, installStatus, isUpgrading)
 
 	// 安装结束
 	installStatus.IsRunning = false
@@ -78,7 +78,7 @@ func (this *Queue) InstallNodeProcess(nodeId int64) error {
 }
 
 // 安装边缘节点
-func (this *Queue) InstallNode(nodeId int64, installStatus *models.NodeInstallStatus) error {
+func (this *Queue) InstallNode(nodeId int64, installStatus *models.NodeInstallStatus, isUpgrading bool) error {
 	node, err := models.SharedNodeDAO.FindEnabledNode(nodeId)
 	if err != nil {
 		return err
@@ -171,9 +171,10 @@ func (this *Queue) InstallNode(nodeId int64, installStatus *models.NodeInstallSt
 	}
 
 	params := &NodeParams{
-		Endpoints: apiEndpoints,
-		NodeId:    node.UniqueId,
-		Secret:    node.Secret,
+		Endpoints:   apiEndpoints,
+		NodeId:      node.UniqueId,
+		Secret:      node.Secret,
+		IsUpgrading: isUpgrading,
 	}
 
 	installer := &NodeInstaller{}
@@ -187,6 +188,9 @@ func (this *Queue) InstallNode(nodeId int64, installStatus *models.NodeInstallSt
 	if err != nil {
 		return err
 	}
+	defer func() {
+		_ = installer.Close()
+	}()
 
 	err = installer.Install(installDir, params, installStatus)
 	return err
@@ -271,6 +275,9 @@ func (this *Queue) StartNode(nodeId int64) error {
 	if err != nil {
 		return err
 	}
+	defer func() {
+		_ = installer.Close()
+	}()
 
 	// 检查命令是否存在
 	exeFile := installDir + "/edge-node/bin/edge-node"
@@ -369,6 +376,9 @@ func (this *Queue) StopNode(nodeId int64) error {
 	if err != nil {
 		return err
 	}
+	defer func() {
+		_ = installer.Close()
+	}()
 
 	// 检查命令是否存在
 	exeFile := installDir + "/edge-node/bin/edge-node"

internal/installers/ssh_client.go

@@ -147,3 +147,8 @@ func (this *SSHClient) WriteFile(path string, data []byte) (n int, err error) {
 	n, err = fp.Write(data)
 	return
 }
+
+// 删除文件
+func (this *SSHClient) Remove(path string) error {
+	return this.sftp.Remove(path)
+}