阶段性提交

internal/installers/installer_node.go

@@ -62,7 +62,7 @@ func (this *NodeInstaller) Install(dir string, params interface{}) error {
 			return err
 		}
 
-		data = bytes.ReplaceAll(data, []byte("${endpoint}"), []byte(nodeParams.Endpoint))
+		data = bytes.ReplaceAll(data, []byte("${endpoints}"), []byte(nodeParams.QuoteEndpoints()))
 		data = bytes.ReplaceAll(data, []byte("${nodeId}"), []byte(nodeParams.NodeId))
 		data = bytes.ReplaceAll(data, []byte("${nodeSecret}"), []byte(nodeParams.Secret))
 

internal/installers/installer_node_test.go

@@ -25,9 +25,9 @@ func TestNodeInstaller_Install(t *testing.T) {
 
 	// 安装
 	err = installer.Install("/opt/edge", &NodeParams{
-		Endpoint: "192.168.2.40:8003",
-		NodeId:   "313fdb1b90d0a63c736f307b4d1ca358",
-		Secret:   "Pl3u5kYqBDZddp7raw6QfHiuGPRCWF54",
+		Endpoints: []string{"192.168.2.40:8003"},
+		NodeId:    "313fdb1b90d0a63c736f307b4d1ca358",
+		Secret:    "Pl3u5kYqBDZddp7raw6QfHiuGPRCWF54",
 	})
 	if err != nil {
 		t.Fatal(err)

internal/installers/params_node.go

@@ -1,15 +1,18 @@
 package installers
 
-import "errors"
+import (
+	"errors"
+	"strings"
+)
 
 type NodeParams struct {
-	Endpoint string
-	NodeId   string
-	Secret   string
+	Endpoints []string
+	NodeId    string
+	Secret    string
 }
 
 func (this *NodeParams) Validate() error {
-	if len(this.Endpoint) == 0 {
+	if len(this.Endpoints) == 0 {
 		return errors.New("'endpoint' should not be empty")
 	}
 	if len(this.NodeId) == 0 {
@@ -20,3 +23,10 @@ func (this *NodeParams) Validate() error {
 	}
 	return nil
 }
+
+func (this *NodeParams) QuoteEndpoints() string {
+	if len(this.Endpoints) == 0 {
+		return ""
+	}
+	return "\"" + strings.Join(this.Endpoints, "\", \"") + "\""
+}

internal/installers/queue.go

@@ -0,0 +1,162 @@
+package installers
+
+import (
+	"errors"
+	"fmt"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
+	"github.com/iwind/TeaGo/logs"
+	"strconv"
+	"time"
+)
+
+var sharedQueue = NewQueue()
+
+type Queue struct {
+}
+
+func NewQueue() *Queue {
+	return &Queue{}
+}
+
+func SharedQueue() *Queue {
+	return sharedQueue
+}
+
+// 安装边缘节点流程控制
+func (this *Queue) InstallNodeProcess(nodeId int64) error {
+	installStatus := models.NewNodeInstallStatus()
+	installStatus.IsRunning = true
+	installStatus.UpdatedAt = time.Now().Unix()
+
+	err := models.SharedNodeDAO.UpdateNodeInstallStatus(nodeId, installStatus)
+	if err != nil {
+		return err
+	}
+
+	// 更新时间
+	ticker := utils.NewTicker(3 * time.Second)
+	go func() {
+		for ticker.Wait() {
+			installStatus.UpdatedAt = time.Now().Unix()
+			err := models.SharedNodeDAO.UpdateNodeInstallStatus(nodeId, installStatus)
+			if err != nil {
+				logs.Println("[INSTALL]" + err.Error())
+				continue
+			}
+		}
+	}()
+	defer func() {
+		ticker.Stop()
+	}()
+
+	// 开始安装
+	err = this.InstallNode(nodeId)
+
+	// 安装结束
+	installStatus.IsRunning = false
+	installStatus.IsFinished = true
+	if err != nil {
+		installStatus.Error = err.Error()
+	} else {
+		installStatus.IsOk = true
+	}
+	err = models.SharedNodeDAO.UpdateNodeInstallStatus(nodeId, installStatus)
+	if err != nil {
+		return err
+	}
+
+	// 修改为已安装
+	if installStatus.IsOk {
+		err = models.SharedNodeDAO.UpdateNodeIsInstalled(nodeId, true)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// 安装边缘节点
+func (this *Queue) InstallNode(nodeId int64) error {
+	node, err := models.SharedNodeDAO.FindEnabledNode(nodeId)
+	if err != nil {
+		return err
+	}
+	if node == nil {
+		return errors.New("can not find node, ID：'" + strconv.FormatInt(nodeId, 10) + "'")
+	}
+
+	// 登录信息
+	login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nodeId)
+	if err != nil {
+		return err
+	}
+	if login == nil {
+		return errors.New("can not find node login information")
+	}
+	loginParams, err := login.DecodeSSHParams()
+	if err != nil {
+		return err
+	}
+
+	grant, err := models.SharedNodeGrantDAO.FindEnabledNodeGrant(loginParams.GrantId)
+	if err != nil {
+		return err
+	}
+	if grant == nil {
+		return errors.New("can not find user grant with id '" + strconv.FormatInt(loginParams.GrantId, 10) + "'")
+	}
+
+	// 安装目录
+	installDir := node.InstallDir
+	if len(installDir) == 0 {
+		clusterId := node.ClusterId
+		cluster, err := models.SharedNodeClusterDAO.FindEnabledNodeCluster(int64(clusterId))
+		if err != nil {
+			return err
+		}
+		if cluster == nil {
+			return errors.New("can not find cluster, ID：'" + fmt.Sprintf("%d", clusterId) + "'")
+		}
+		installDir = cluster.InstallDir
+		if len(installDir) == 0 {
+			return errors.New("unable to find installation dir")
+		}
+	}
+
+	// API终端
+	apiNodes, err := models.SharedAPINodeDAO.FindAllEnabledAPINodes()
+	if err != nil {
+		return err
+	}
+	if len(apiNodes) == 0 {
+		return errors.New("no available api nodes")
+	}
+
+	apiEndpoints := []string{}
+	for _, apiNode := range apiNodes {
+		apiEndpoints = append(apiEndpoints, apiNode.Host+":"+strconv.Itoa(int(apiNode.Port)))
+	}
+
+	params := &NodeParams{
+		Endpoints: apiEndpoints,
+		NodeId:    node.UniqueId,
+		Secret:    node.Secret,
+	}
+
+	installer := &NodeInstaller{}
+	err = installer.Login(&Credentials{
+		Host:       loginParams.Host,
+		Port:       loginParams.Port,
+		Username:   grant.Username,
+		Password:   grant.Password,
+		PrivateKey: grant.PrivateKey,
+	})
+	if err != nil {
+		return err
+	}
+
+	err = installer.Install(installDir, params)
+	return err
+}

internal/installers/queue_test.go

@@ -0,0 +1,19 @@
+package installers
+
+import (
+	"testing"
+	"time"
+)
+
+func TestQueue_InstallNode(t *testing.T) {
+	queue := NewQueue()
+	err := queue.InstallNodeProcess(16)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	time.Sleep(1 * time.Second)
+
+	t.Log("OK")
+
+}