TRKJ/EdgeAPI
2
0
Fork 0
mirror of https://github.com/TeaOSLab/EdgeAPI.git synced 2025-11-04 07:50:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

实现DNS节点远程安装

Browse Source
This commit is contained in:
GoEdgeLab
2021-08-11 21:00:29 +08:00
parent 30ba617f94
commit 6460489a91
18 changed files with 859 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
build/build.sh
View File

@@ -50,6 +50,7 @@ function build() {
done
# build edge-dns
if [ "$TAG" = "plus" ]; then
DNS_ROOT=$ROOT"/../../EdgeDNS"
if [ -d $DNS_ROOT ]; then
DNSNodeVersion=$(lookup-version $ROOT"/../../EdgeDNS/internal/const/const.go")
@@ -72,6 +73,7 @@ function build() {
cp $ROOT"/../../EdgeDNS/dist/edge-dns-linux-${arch}-v${DNSNodeVersion}.zip" $ROOT/deploy/edge-dns-linux-${arch}-v${DNSNodeVersion}.zip
done
fi
fi
# build sql
echo "building sql ..."
@@ -94,14 +96,22 @@ function build() {
rm -f $DIST/resources/ipdata/ip2region/global_region.csv
rm -f $DIST/resources/ipdata/ip2region/ip.merge.txt
# building installer
echo "building installer ..."
# building edge installer
echo "building node installer ..."
architects=("amd64" "386" "arm64")
for arch in "${architects[@]}"; do
# TODO support arm, mips ...
env GOOS=linux GOARCH=${arch} go build -tags $TAG --ldflags="-s -w" -o $ROOT/installers/edge-installer-helper-linux-${arch} $ROOT/../cmd/installer-helper/main.go
done
# building edge dns installer
echo "building dns node installer ..."
architects=("amd64" "386" "arm64")
for arch in "${architects[@]}"; do
# TODO support arm, mips ...
env GOOS=linux GOARCH=${arch} go build -tags $TAG --ldflags="-s -w" -o $ROOT/installers/edge-installer-dns-helper-linux-${arch} $ROOT/../cmd/installer-dns-helper/main.go
done
# building api node
env GOOS=$OS GOARCH=$ARCH go build -tags $TAG --ldflags="-s -w" -o $DIST/bin/edge-api $ROOT/../cmd/edge-api/main.go

62
cmd/installer-dns-helper/main.go Normal file
View File

@@ -0,0 +1,62 @@
package main
import (
"flag"
"github.com/TeaOSLab/EdgeAPI/internal/utils"
"net"
"os"
)
func main() {
cmd := ""
flag.StringVar(&cmd, "cmd", "", "command name: [unzip]")
// unzip
zipPath := ""
targetPath := ""
flag.StringVar(&zipPath, "zip", "", "zip path")
flag.StringVar(&targetPath, "target", "", "target dir")
// parse
flag.Parse()
if len(cmd) == 0 {
stderr("need '-cmd=COMMAND' argument")
} else if cmd == "test" {
// 检查是否正在运行
path := os.TempDir() + "/edge-dns.sock"
conn, err := net.Dial("unix", path)
if err == nil {
_ = conn.Close()
stderr("test dns node status: edge dns node is running now, can not install again")
}
} else if cmd == "unzip" { // 解压
if len(zipPath) == 0 {
stderr("ERROR: need '-zip=PATH' argument")
return
}
if len(targetPath) == 0 {
stderr("ERROR: need '-target=TARGET' argument")
return
}
unzip := utils.NewUnzip(zipPath, targetPath)
err := unzip.Run()
if err != nil {
stderr("ERROR: " + err.Error())
return
}
stdout("ok")
} else {
stderr("ERROR: not recognized command '" + cmd + "'")
}
}
func stdout(s string) {
_, _ = os.Stdout.WriteString(s + "\n")
}
func stderr(s string) {
_, _ = os.Stderr.WriteString(s + "\n")
}

4
internal/db/models/node_grant_model.go
View File

@@ -1,6 +1,6 @@
package models
// 节点授权
// NodeGrant 节点授权
type NodeGrant struct {
Id uint32 `field:"id"` // ID
AdminId uint32 `field:"adminId"` // 管理员ID
@@ -12,6 +12,7 @@ type NodeGrant struct {
PrivateKey string `field:"privateKey"` // 密钥
Description string `field:"description"` // 备注
NodeId uint32 `field:"nodeId"` // 专有节点
Role string `field:"role"` // 角色
State uint8 `field:"state"` // 状态
CreatedAt uint64 `field:"createdAt"` // 创建时间
}
@@ -27,6 +28,7 @@ type NodeGrantOperator struct {
PrivateKey interface{} // 密钥
Description interface{} // 备注
NodeId interface{} // 专有节点
Role interface{} // 角色
State interface{} // 状态
CreatedAt interface{} // 创建时间
}

42
internal/db/models/node_login_dao.go
View File

@@ -2,6 +2,7 @@ package models
import (
"errors"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
_ "github.com/go-sql-driver/mysql"
"github.com/iwind/TeaGo/Tea"
"github.com/iwind/TeaGo/dbs"
@@ -36,7 +37,7 @@ func init() {
})
}
// 启用条目
// EnableNodeLogin 启用条目
func (this *NodeLoginDAO) EnableNodeLogin(tx *dbs.Tx, id uint32) (rowsAffected int64, err error) {
return this.Query(tx).
Pk(id).
@@ -44,16 +45,16 @@ func (this *NodeLoginDAO) EnableNodeLogin(tx *dbs.Tx, id uint32) (rowsAffected i
Update()
}
// 禁用条目
func (this *NodeLoginDAO) DisableNodeLogin(tx *dbs.Tx, id uint32) (rowsAffected int64, err error) {
// DisableNodeLogin 禁用条目
func (this *NodeLoginDAO) DisableNodeLogin(tx *dbs.Tx, loginId int64) (rowsAffected int64, err error) {
return this.Query(tx).
Pk(id).
Pk(loginId).
Set("state", NodeLoginStateDisabled).
Update()
}
// 查找启用中的条目
func (this *NodeLoginDAO) FindEnabledNodeLogin(tx *dbs.Tx, id uint32) (*NodeLogin, error) {
// FindEnabledNodeLogin 查找启用中的条目
func (this *NodeLoginDAO) FindEnabledNodeLogin(tx *dbs.Tx, id int64) (*NodeLogin, error) {
result, err := this.Query(tx).
Pk(id).
Attr("state", NodeLoginStateEnabled).
@@ -64,7 +65,7 @@ func (this *NodeLoginDAO) FindEnabledNodeLogin(tx *dbs.Tx, id uint32) (*NodeLogi
return result.(*NodeLogin), err
}
// 根据主键查找名称
// FindNodeLoginName 根据主键查找名称
func (this *NodeLoginDAO) FindNodeLoginName(tx *dbs.Tx, id uint32) (string, error) {
name, err := this.Query(tx).
Pk(id).
@@ -73,9 +74,14 @@ func (this *NodeLoginDAO) FindNodeLoginName(tx *dbs.Tx, id uint32) (string, erro
return name.(string), err
}
// 创建认证
func (this *NodeLoginDAO) CreateNodeLogin(tx *dbs.Tx, nodeId int64, name string, loginType string, paramsJSON []byte) (loginId int64, err error) {
// CreateNodeLogin 创建认证
func (this *NodeLoginDAO) CreateNodeLogin(tx *dbs.Tx, role nodeconfigs.NodeRole, nodeId int64, name string, loginType string, paramsJSON []byte) (loginId int64, err error) {
if len(role) == 0 {
role = nodeconfigs.NodeRoleNode
}
login := NewNodeLoginOperator()
login.Role = role
login.NodeId = nodeId
login.Name = name
login.Type = loginType
@@ -85,7 +91,7 @@ func (this *NodeLoginDAO) CreateNodeLogin(tx *dbs.Tx, nodeId int64, name string,
return types.Int64(login.Id), err
}
// 修改认证
// UpdateNodeLogin 修改认证
func (this *NodeLoginDAO) UpdateNodeLogin(tx *dbs.Tx, loginId int64, name string, loginType string, paramsJSON []byte) error {
if loginId <= 0 {
return errors.New("invalid loginId")
@@ -99,9 +105,13 @@ func (this *NodeLoginDAO) UpdateNodeLogin(tx *dbs.Tx, loginId int64, name string
return err
}
// 查找认证
func (this *NodeLoginDAO) FindEnabledNodeLoginWithNodeId(tx *dbs.Tx, nodeId int64) (*NodeLogin, error) {
// FindEnabledNodeLoginWithNodeId 查找认证
func (this *NodeLoginDAO) FindEnabledNodeLoginWithNodeId(tx *dbs.Tx, role nodeconfigs.NodeRole, nodeId int64) (*NodeLogin, error) {
if len(role) == 0 {
role = nodeconfigs.NodeRoleNode
}
one, err := this.Query(tx).
Attr("role", role).
Attr("nodeId", nodeId).
State(NodeLoginStateEnabled).
Find()
@@ -114,9 +124,13 @@ func (this *NodeLoginDAO) FindEnabledNodeLoginWithNodeId(tx *dbs.Tx, nodeId int6
return one.(*NodeLogin), nil
}
// 禁用某个节点的认证
func (this *NodeLoginDAO) DisableNodeLogins(tx *dbs.Tx, nodeId int64) error {
// DisableNodeLogins 禁用某个节点的认证
func (this *NodeLoginDAO) DisableNodeLogins(tx *dbs.Tx, role nodeconfigs.NodeRole, nodeId int64) error {
if len(role) == 0 {
role = nodeconfigs.NodeRoleNode
}
_, err := this.Query(tx).
Attr("role", role).
Attr("nodeId", nodeId).
Set("state", NodeLoginStateDisabled).
Update()

4
internal/db/models/node_login_model.go
View File

@@ -1,9 +1,10 @@
package models
//
// NodeLogin 节点登录信息
type NodeLogin struct {
Id uint32 `field:"id"` // ID
NodeId uint32 `field:"nodeId"` // 节点ID
Role string `field:"role"` // 角色
Name string `field:"name"` // 名称
Type string `field:"type"` // 类型：ssh,agent
Params string `field:"params"` // 配置参数
@@ -13,6 +14,7 @@ type NodeLogin struct {
type NodeLoginOperator struct {
Id interface{} // ID
NodeId interface{} // 节点ID
Role interface{} // 角色
Name interface{} // 名称
Type interface{} // 类型：ssh,agent
Params interface{} // 配置参数

8
internal/db/models/ns_cluster_dao.go
View File

@@ -161,6 +161,14 @@ func (this *NSClusterDAO) FindClusterAccessLog(tx *dbs.Tx, clusterId int64) ([]b
return []byte(accessLog), err
}
// FindClusterGrantId 查找集群的认证ID
func (this *NSClusterDAO) FindClusterGrantId(tx *dbs.Tx, clusterId int64) (int64, error) {
return this.Query(tx).
Pk(clusterId).
Result("grantId").
FindInt64Col(0)
}
// NotifyUpdate 通知更改
func (this *NSClusterDAO) NotifyUpdate(tx *dbs.Tx, clusterId int64) error {
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleDNS, clusterId, NSNodeTaskTypeConfigChanged)

2
internal/db/models/ns_cluster_model.go
View File

@@ -8,6 +8,7 @@ type NSCluster struct {
InstallDir string `field:"installDir"` // 安装目录
State uint8 `field:"state"` // 状态
AccessLog string `field:"accessLog"` // 访问日志配置
GrantId uint32 `field:"grantId"` // 授权ID
}
type NSClusterOperator struct {
@@ -17,6 +18,7 @@ type NSClusterOperator struct {
InstallDir interface{} // 安装目录
State interface{} // 状态
AccessLog interface{} // 访问日志配置
GrantId interface{} // 授权ID
}
func NewNSClusterOperator() *NSClusterOperator {

21
internal/db/models/ns_node_dao.go
View File

@@ -521,6 +521,27 @@ func (this *NSNodeDAO) FindAllNodeIdsMatch(tx *dbs.Tx, clusterId int64, includeS
return
}
// UpdateNodeInstallStatus 修改节点的安装状态
func (this *NSNodeDAO) UpdateNodeInstallStatus(tx *dbs.Tx, nodeId int64, status *NodeInstallStatus) error {
if status == nil {
_, err := this.Query(tx).
Pk(nodeId).
Set("installStatus", "null").
Update()
return err
}
data, err := json.Marshal(status)
if err != nil {
return err
}
_, err = this.Query(tx).
Pk(nodeId).
Set("installStatus", string(data)).
Update()
return err
}
// NotifyUpdate 通知更新
func (this *NSNodeDAO) NotifyUpdate(tx *dbs.Tx, nodeId int64) error {
// TODO 先什么都不做

2
internal/installers/installer_base.go
View File

@@ -164,7 +164,7 @@ func (this *BaseInstaller) InstallHelper(targetDir string) (env *Env, err error)
archName = "386"
}
exeName := "edge-installer-helper-" + osName + "-" + archName
exeName := "edge-installer-dns-helper-" + osName + "-" + archName
exePath := Tea.Root + "/installers/" + exeName
err = this.client.Copy(exePath, targetDir+"/"+exeName, 0777)

9
internal/installers/installer_node.go
View File

@@ -95,12 +95,11 @@ func (this *NodeInstaller) Install(dir string, params interface{}, installStatus
// 修改配置文件
{
templateFile := dir + "/edge-node/configs/api.template.yaml"
configFile := dir + "/edge-node/configs/api.yaml"
data, err := this.client.ReadFile(templateFile)
if err != nil {
return err
}
var data = []byte(`rpc:
endpoints: [ ${endpoints} ]
nodeId: "${nodeId}"
secret: "${nodeSecret}"`)
data = bytes.ReplaceAll(data, []byte("${endpoints}"), []byte(nodeParams.QuoteEndpoints()))
data = bytes.ReplaceAll(data, []byte("${nodeId}"), []byte(nodeParams.NodeId))

2
internal/installers/installer_node_test.go
View File

@@ -28,7 +28,7 @@ func TestNodeInstaller_Install(t *testing.T) {
// 安装
err = installer.Install("/opt/edge", &NodeParams{
Endpoints: []string{"192.168.2.40:8003"},
Endpoints: []string{"http://192.168.2.40:8003"},
NodeId: "313fdb1b90d0a63c736f307b4d1ca358",
Secret: "Pl3u5kYqBDZddp7raw6QfHiuGPRCWF54",
}, &models.NodeInstallStatus{})

139
internal/installers/installer_ns_node.go Normal file
View File

@@ -0,0 +1,139 @@
package installers
import (
"bytes"
"errors"
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"path/filepath"
"regexp"
)
type NSNodeInstaller struct {
BaseInstaller
}
func (this *NSNodeInstaller) Install(dir string, params interface{}, installStatus *models.NodeInstallStatus) error {
if params == nil {
return errors.New("'params' required for node installation")
}
nodeParams, ok := params.(*NodeParams)
if !ok {
return errors.New("'params' should be *NodeParams")
}
err := nodeParams.Validate()
if err != nil {
return errors.New("params validation: " + err.Error())
}
// 检查目标目录是否存在
_, err = this.client.Stat(dir)
if err != nil {
err = this.client.MkdirAll(dir)
if err != nil {
installStatus.ErrorCode = "CREATE_ROOT_DIRECTORY_FAILED"
return errors.New("create directory '" + dir + "' failed: " + err.Error())
}
}
// 安装助手
env, err := this.InstallHelper(dir)
if err != nil {
installStatus.ErrorCode = "INSTALL_HELPER_FAILED"
return err
}
// 上传安装文件
filePrefix := "edge-dns-" + env.OS + "-" + env.Arch
zipFile, err := this.LookupLatestInstaller(filePrefix)
if err != nil {
return err
}
if len(zipFile) == 0 {
return errors.New("can not find installer file for " + env.OS + "/" + env.Arch)
}
targetZip := dir + "/" + filepath.Base(zipFile)
err = this.client.Copy(zipFile, targetZip, 0777)
if err != nil {
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-dns/bin/edge-dns"
if nodeParams.IsUpgrading {
_, err = this.client.Stat(exePath)
if err == nil {
_, _, _ = this.client.Exec(exePath + " stop")
}
// 删除可执行文件防止冲突
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 + "\"")
if err != nil {
return err
}
if len(stderr) > 0 {
return errors.New("unzip installer failed: " + stderr)
}
// 修改配置文件
{
configFile := dir + "/edge-dns/configs/api.yaml"
var data = []byte(`rpc:
endpoints: [ ${endpoints} ]
nodeId: "${nodeId}"
secret: "${nodeSecret}"`)
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))
_, err = this.client.WriteFile(configFile, data)
if err != nil {
return errors.New("write 'configs/api.yaml': " + err.Error())
}
}
// 测试
_, stderr, err = this.client.Exec(dir + "/edge-dns/bin/edge-dns test")
if err != nil {
installStatus.ErrorCode = "TEST_FAILED"
return errors.New("test edge node failed: " + err.Error())
}
if len(stderr) > 0 {
if regexp.MustCompile(`(?i)rpc`).MatchString(stderr) {
installStatus.ErrorCode = "RPC_TEST_FAILED"
}
return errors.New("test edge dns node failed: " + stderr)
}
// 启动
_, stderr, err = this.client.Exec(dir + "/edge-dns/bin/edge-dns start")
if err != nil {
return errors.New("start edge dns failed: " + err.Error())
}
if len(stderr) > 0 {
return errors.New("start edge dns failed: " + stderr)
}
return nil
}

39
internal/installers/installer_ns_node_test.go Normal file
View File

@@ -0,0 +1,39 @@
package installers
import (
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"testing"
)
func TestDNSNodeInstaller_Install(t *testing.T) {
var installer InstallerInterface = &DNSNodeInstaller{}
err := installer.Login(&Credentials{
Host: "192.168.2.30",
Port: 22,
Username: "root",
Password: "123456",
PrivateKey: "",
Method: "user",
})
if err != nil {
t.Fatal(err)
}
// 关闭连接
defer func() {
err := installer.Close()
if err != nil {
t.Fatal(err)
}
}()
// 安装
err = installer.Install("/opt/edge", &NodeParams{
Endpoints: []string{"http://192.168.2.40:8003"},
NodeId: "b3f0690c793db5daaa666e89bd7b2301",
Secret: "H6nbSzjN3tLYi0ecdtUeDpQdZZPjKL7S",
}, &models.NodeInstallStatus{})
if err != nil {
t.Fatal(err)
}
}

35
internal/installers/queue.go → internal/installers/queue_node.go
View File

@@ -6,25 +6,26 @@ import (
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"github.com/TeaOSLab/EdgeAPI/internal/utils"
"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/iwind/TeaGo/logs"
"time"
)
var sharedQueue = NewQueue()
var sharedNodeQueue = NewNodeQueue()
type Queue struct {
type NodeQueue struct {
}
func NewQueue() *Queue {
return &Queue{}
func NewNodeQueue() *NodeQueue {
return &NodeQueue{}
}
func SharedQueue() *Queue {
return sharedQueue
func SharedNodeQueue() *NodeQueue {
return sharedNodeQueue
}
// 安装边缘节点流程控制
func (this *Queue) InstallNodeProcess(nodeId int64, isUpgrading bool) error {
// InstallNodeProcess 安装边缘节点流程控制
func (this *NodeQueue) InstallNodeProcess(nodeId int64, isUpgrading bool) error {
installStatus := models.NewNodeInstallStatus()
installStatus.IsRunning = true
installStatus.UpdatedAt = time.Now().Unix()
@@ -77,8 +78,8 @@ func (this *Queue) InstallNodeProcess(nodeId int64, isUpgrading bool) error {
return nil
}
// 安装边缘节点
func (this *Queue) InstallNode(nodeId int64, installStatus *models.NodeInstallStatus, isUpgrading bool) error {
// InstallNode 安装边缘节点
func (this *NodeQueue) InstallNode(nodeId int64, installStatus *models.NodeInstallStatus, isUpgrading bool) error {
node, err := models.SharedNodeDAO.FindEnabledNode(nil, nodeId)
if err != nil {
return err
@@ -88,7 +89,7 @@ func (this *Queue) InstallNode(nodeId int64, installStatus *models.NodeInstallSt
}
// 登录信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nil, nodeId)
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nil, nodeconfigs.NodeRoleNode, nodeId)
if err != nil {
return err
}
@@ -198,8 +199,8 @@ func (this *Queue) InstallNode(nodeId int64, installStatus *models.NodeInstallSt
return err
}
// 启动边缘节点
func (this *Queue) StartNode(nodeId int64) error {
// StartNode 启动边缘节点
func (this *NodeQueue) StartNode(nodeId int64) error {
node, err := models.SharedNodeDAO.FindEnabledNode(nil, nodeId)
if err != nil {
return err
@@ -209,7 +210,7 @@ func (this *Queue) StartNode(nodeId int64) error {
}
// 登录信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nil, nodeId)
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nil, nodeconfigs.NodeRoleNode, nodeId)
if err != nil {
return err
}
@@ -303,8 +304,8 @@ func (this *Queue) StartNode(nodeId int64) error {
return nil
}
// 停止节点
func (this *Queue) StopNode(nodeId int64) error {
// StopNode 停止节点
func (this *NodeQueue) StopNode(nodeId int64) error {
node, err := models.SharedNodeDAO.FindEnabledNode(nil, nodeId)
if err != nil {
return err
@@ -314,7 +315,7 @@ func (this *Queue) StopNode(nodeId int64) error {
}
// 登录信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nil, nodeId)
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nil, nodeconfigs.NodeRoleNode, nodeId)
if err != nil {
return err
}

0
internal/installers/queue_test.go → internal/installers/queue_node_test.go
View File

410
internal/installers/queue_ns_node.go Normal file
View File

@@ -0,0 +1,410 @@
package installers
import (
"errors"
"fmt"
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"github.com/TeaOSLab/EdgeAPI/internal/utils"
"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/iwind/TeaGo/logs"
"time"
)
var sharedNSNodeQueue = NewNSNodeQueue()
type NSNodeQueue struct {
}
func NewNSNodeQueue() *NSNodeQueue {
return &NSNodeQueue{}
}
func SharedNSNodeQueue() *NSNodeQueue {
return sharedNSNodeQueue
}
// InstallNodeProcess 安装边缘节点流程控制
func (this *NSNodeQueue) InstallNodeProcess(nodeId int64, isUpgrading bool) error {
installStatus := models.NewNodeInstallStatus()
installStatus.IsRunning = true
installStatus.UpdatedAt = time.Now().Unix()
err := models.SharedNSNodeDAO.UpdateNodeInstallStatus(nil, 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.SharedNSNodeDAO.UpdateNodeInstallStatus(nil, nodeId, installStatus)
if err != nil {
logs.Println("[INSTALL]" + err.Error())
continue
}
}
}()
defer func() {
ticker.Stop()
}()
// 开始安装
err = this.InstallNode(nodeId, installStatus, isUpgrading)
// 安装结束
installStatus.IsRunning = false
installStatus.IsFinished = true
if err != nil {
installStatus.Error = err.Error()
} else {
installStatus.IsOk = true
}
err = models.SharedNSNodeDAO.UpdateNodeInstallStatus(nil, nodeId, installStatus)
if err != nil {
return err
}
// 修改为已安装
if installStatus.IsOk {
err = models.SharedNSNodeDAO.UpdateNodeIsInstalled(nil, nodeId, true)
if err != nil {
return err
}
}
return nil
}
// InstallNode 安装边缘节点
func (this *NSNodeQueue) InstallNode(nodeId int64, installStatus *models.NodeInstallStatus, isUpgrading bool) error {
node, err := models.SharedNSNodeDAO.FindEnabledNSNode(nil, nodeId)
if err != nil {
return err
}
if node == nil {
return errors.New("can not find node, ID：'" + numberutils.FormatInt64(nodeId) + "'")
}
// 登录信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nil, nodeconfigs.NodeRoleDNS, nodeId)
if err != nil {
return err
}
if login == nil {
installStatus.ErrorCode = "EMPTY_LOGIN"
return errors.New("can not find node login information")
}
loginParams, err := login.DecodeSSHParams()
if err != nil {
return err
}
if len(loginParams.Host) == 0 {
installStatus.ErrorCode = "EMPTY_SSH_HOST"
return errors.New("ssh host should not be empty")
}
if loginParams.Port <= 0 {
installStatus.ErrorCode = "EMPTY_SSH_PORT"
return errors.New("ssh port is invalid")
}
if loginParams.GrantId == 0 {
// 从集群中读取
grantId, err := models.SharedNSClusterDAO.FindClusterGrantId(nil, int64(node.ClusterId))
if err != nil {
return err
}
if grantId == 0 {
installStatus.ErrorCode = "EMPTY_GRANT"
return errors.New("can not find node grant")
}
loginParams.GrantId = grantId
}
grant, err := models.SharedNodeGrantDAO.FindEnabledNodeGrant(nil, loginParams.GrantId)
if err != nil {
return err
}
if grant == nil {
installStatus.ErrorCode = "EMPTY_GRANT"
return errors.New("can not find user grant with id '" + numberutils.FormatInt64(loginParams.GrantId) + "'")
}
// 安装目录
installDir := node.InstallDir
if len(installDir) == 0 {
clusterId := node.ClusterId
cluster, err := models.SharedNSClusterDAO.FindEnabledNSCluster(nil, 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 {
// 默认是 $登录用户/edge-dns
installDir = "/" + grant.Username + "/edge-dns"
}
}
// API终端
apiNodes, err := models.SharedAPINodeDAO.FindAllEnabledAndOnAPINodes(nil)
if err != nil {
return err
}
if len(apiNodes) == 0 {
return errors.New("no available api nodes")
}
apiEndpoints := []string{}
for _, apiNode := range apiNodes {
addrConfigs, err := apiNode.DecodeAccessAddrs()
if err != nil {
return errors.New("decode api node access addresses failed: " + err.Error())
}
for _, addrConfig := range addrConfigs {
apiEndpoints = append(apiEndpoints, addrConfig.FullAddresses()...)
}
}
params := &NodeParams{
Endpoints: apiEndpoints,
NodeId: node.UniqueId,
Secret: node.Secret,
IsUpgrading: isUpgrading,
}
installer := &NSNodeInstaller{}
err = installer.Login(&Credentials{
Host: loginParams.Host,
Port: loginParams.Port,
Username: grant.Username,
Password: grant.Password,
PrivateKey: grant.PrivateKey,
Method: grant.Method,
})
if err != nil {
installStatus.ErrorCode = "SSH_LOGIN_FAILED"
return err
}
defer func() {
_ = installer.Close()
}()
err = installer.Install(installDir, params, installStatus)
return err
}
// StartNode 启动边缘节点
func (this *NSNodeQueue) StartNode(nodeId int64) error {
node, err := models.SharedNSNodeDAO.FindEnabledNSNode(nil, nodeId)
if err != nil {
return err
}
if node == nil {
return errors.New("can not find node, ID：'" + numberutils.FormatInt64(nodeId) + "'")
}
// 登录信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nil, nodeconfigs.NodeRoleDNS, 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
}
if len(loginParams.Host) == 0 {
return errors.New("ssh host should not be empty")
}
if loginParams.Port <= 0 {
return errors.New("ssh port is invalid")
}
if loginParams.GrantId == 0 {
// 从集群中读取
grantId, err := models.SharedNSClusterDAO.FindClusterGrantId(nil, int64(node.ClusterId))
if err != nil {
return err
}
if grantId == 0 {
return errors.New("can not find node grant")
}
loginParams.GrantId = grantId
}
grant, err := models.SharedNodeGrantDAO.FindEnabledNodeGrant(nil, loginParams.GrantId)
if err != nil {
return err
}
if grant == nil {
return errors.New("can not find user grant with id '" + numberutils.FormatInt64(loginParams.GrantId) + "'")
}
// 安装目录
installDir := node.InstallDir
if len(installDir) == 0 {
clusterId := node.ClusterId
cluster, err := models.SharedNSClusterDAO.FindEnabledNSCluster(nil, 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 {
// 默认是 $登录用户/edge-dns
installDir = "/" + grant.Username + "/edge-dns"
}
}
installer := &NSNodeInstaller{}
err = installer.Login(&Credentials{
Host: loginParams.Host,
Port: loginParams.Port,
Username: grant.Username,
Password: grant.Password,
PrivateKey: grant.PrivateKey,
Method: grant.Method,
})
if err != nil {
return err
}
defer func() {
_ = installer.Close()
}()
// 检查命令是否存在
exeFile := installDir + "/edge-dns/bin/edge-dns"
_, err = installer.client.Stat(exeFile)
if err != nil {
return errors.New("edge node is not installed correctly, can not find executable file: " + exeFile)
}
// 我们先尝试Systemd启动
_, _, _ = installer.client.Exec("systemctl start edge-dns")
_, stderr, err := installer.client.Exec(exeFile + " start")
if err != nil {
return errors.New("start failed: " + err.Error())
}
if len(stderr) > 0 {
return errors.New("start failed: " + stderr)
}
return nil
}
// StopNode 停止节点
func (this *NSNodeQueue) StopNode(nodeId int64) error {
node, err := models.SharedNSNodeDAO.FindEnabledNSNode(nil, nodeId)
if err != nil {
return err
}
if node == nil {
return errors.New("can not find node, ID：'" + numberutils.FormatInt64(nodeId) + "'")
}
// 登录信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(nil, nodeconfigs.NodeRoleDNS, 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
}
if len(loginParams.Host) == 0 {
return errors.New("ssh host should not be empty")
}
if loginParams.Port <= 0 {
return errors.New("ssh port is invalid")
}
if loginParams.GrantId == 0 {
// 从集群中读取
grantId, err := models.SharedNSClusterDAO.FindClusterGrantId(nil, int64(node.ClusterId))
if err != nil {
return err
}
if grantId == 0 {
return errors.New("can not find node grant")
}
loginParams.GrantId = grantId
}
grant, err := models.SharedNodeGrantDAO.FindEnabledNodeGrant(nil, loginParams.GrantId)
if err != nil {
return err
}
if grant == nil {
return errors.New("can not find user grant with id '" + numberutils.FormatInt64(loginParams.GrantId) + "'")
}
// 安装目录
installDir := node.InstallDir
if len(installDir) == 0 {
clusterId := node.ClusterId
cluster, err := models.SharedNSClusterDAO.FindEnabledNSCluster(nil, 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 {
// 默认是 $登录用户/edge-dns
installDir = "/" + grant.Username + "/edge-dns"
}
}
installer := &NSNodeInstaller{}
err = installer.Login(&Credentials{
Host: loginParams.Host,
Port: loginParams.Port,
Username: grant.Username,
Password: grant.Password,
PrivateKey: grant.PrivateKey,
Method: grant.Method,
})
if err != nil {
return err
}
defer func() {
_ = installer.Close()
}()
// 检查命令是否存在
exeFile := installDir + "/edge-dns/bin/edge-dns"
_, err = installer.client.Stat(exeFile)
if err != nil {
return errors.New("edge node is not installed correctly, can not find executable file: " + exeFile)
}
// 我们先尝试Systemd停止
_, _, _ = installer.client.Exec("systemctl stop edge-dns")
_, stderr, err := installer.client.Exec(exeFile + " stop")
if err != nil {
return errors.New("stop failed: " + err.Error())
}
if len(stderr) > 0 {
return errors.New("stop failed: " + stderr)
}
return nil
}

75
internal/rpc/services/nameservers/service_ns_node.go
View File

@@ -13,6 +13,7 @@ import (
"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/iwind/TeaGo/logs"
stringutil "github.com/iwind/TeaGo/utils/string"
"path/filepath"
)
@@ -169,6 +170,14 @@ func (this *NSNodeService) CreateNSNode(ctx context.Context, req *pb.CreateNSNod
return nil, err
}
// 增加认证相关
if req.NodeLogin != nil {
_, err = models.SharedNodeLoginDAO.CreateNodeLogin(tx, nodeconfigs.NodeRoleDNS, nodeId, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
if err != nil {
return nil, err
}
}
return &pb.CreateNSNodeResponse{
NsNodeId: nodeId,
}, nil
@@ -220,6 +229,21 @@ func (this *NSNodeService) FindEnabledNSNode(ctx context.Context, req *pb.FindEn
return nil, err
}
// 认证信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(tx, nodeconfigs.NodeRoleDNS, req.NsNodeId)
if err != nil {
return nil, err
}
var respLogin *pb.NodeLogin = nil
if login != nil {
respLogin = &pb.NodeLogin{
Id: int64(login.Id),
Name: login.Name,
Type: login.Type,
Params: []byte(login.Params),
}
}
// 安装信息
installStatus, err := node.DecodeInstallStatus()
if err != nil {
@@ -251,6 +275,7 @@ func (this *NSNodeService) FindEnabledNSNode(ctx context.Context, req *pb.FindEn
},
InstallStatus: installStatusResult,
IsOn: node.IsOn == 1,
NodeLogin: respLogin,
}}, nil
}
@@ -268,6 +293,26 @@ func (this *NSNodeService) UpdateNSNode(ctx context.Context, req *pb.UpdateNSNod
return nil, err
}
// 登录信息
if req.NodeLogin == nil {
err = models.SharedNodeLoginDAO.DisableNodeLogins(tx, nodeconfigs.NodeRoleDNS, req.NsNodeId)
if err != nil {
return nil, err
}
} else {
if req.NodeLogin.Id > 0 {
err = models.SharedNodeLoginDAO.UpdateNodeLogin(tx, req.NodeLogin.Id, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
if err != nil {
return nil, err
}
} else {
_, err = models.SharedNodeLoginDAO.CreateNodeLogin(tx, nodeconfigs.NodeRoleDNS, req.NsNodeId, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
if err != nil {
return nil, err
}
}
}
return this.Success()
}
@@ -278,8 +323,12 @@ func (this *NSNodeService) InstallNSNode(ctx context.Context, req *pb.InstallNSN
return nil, err
}
// TODO 需要实现
return nil, errors.New("尚未实现此功能")
go func() {
err = installers.SharedNSNodeQueue().InstallNodeProcess(req.NsNodeId, false)
if err != nil {
logs.Println("[RPC]install dns node:" + err.Error())
}
}()
return &pb.InstallNSNodeResponse{}, nil
}
@@ -446,3 +495,25 @@ func (this *NSNodeService) UpdateNSNodeConnectedAPINodes(ctx context.Context, re
return this.Success()
}
// UpdateNSNodeLogin 修改节点登录信息
func (this *NSNodeService) UpdateNSNodeLogin(ctx context.Context, req *pb.UpdateNSNodeLoginRequest) (*pb.RPCSuccess, error) {
// 校验请求
_, err := this.ValidateAdmin(ctx, 0)
if err != nil {
return nil, err
}
tx := this.NullTx()
if req.NodeLogin.Id <= 0 {
_, err := models.SharedNodeLoginDAO.CreateNodeLogin(tx, nodeconfigs.NodeRoleDNS, req.NsNodeId, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
if err != nil {
return nil, err
}
}
err = models.SharedNodeLoginDAO.UpdateNodeLogin(tx, req.NodeLogin.Id, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
return this.Success()
}

29
internal/rpc/services/service_node.go
View File

@@ -43,7 +43,7 @@ func (this *NodeService) CreateNode(ctx context.Context, req *pb.CreateNodeReque
// 增加认证相关
if req.NodeLogin != nil {
_, err = models.SharedNodeLoginDAO.CreateNodeLogin(tx, nodeId, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
_, err = models.SharedNodeLoginDAO.CreateNodeLogin(tx, nodeconfigs.NodeRoleNode, nodeId, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
if err != nil {
return nil, err
}
@@ -435,8 +435,9 @@ func (this *NodeService) UpdateNode(ctx context.Context, req *pb.UpdateNodeReque
return nil, err
}
// 登录信息
if req.NodeLogin == nil {
err = models.SharedNodeLoginDAO.DisableNodeLogins(tx, req.NodeId)
err = models.SharedNodeLoginDAO.DisableNodeLogins(tx, nodeconfigs.NodeRoleNode, req.NodeId)
if err != nil {
return nil, err
}
@@ -447,7 +448,7 @@ func (this *NodeService) UpdateNode(ctx context.Context, req *pb.UpdateNodeReque
return nil, err
}
} else {
_, err = models.SharedNodeLoginDAO.CreateNodeLogin(tx, req.NodeId, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
_, err = models.SharedNodeLoginDAO.CreateNodeLogin(tx, nodeconfigs.NodeRoleNode, req.NodeId, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
if err != nil {
return nil, err
}
@@ -524,7 +525,7 @@ func (this *NodeService) FindEnabledNode(ctx context.Context, req *pb.FindEnable
}
// 认证信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(tx, req.NodeId)
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(tx, nodeconfigs.NodeRoleNode, req.NodeId)
if err != nil {
return nil, err
}
@@ -621,7 +622,7 @@ func (this *NodeService) FindEnabledNode(ctx context.Context, req *pb.FindEnable
Name: clusterName,
},
SecondaryNodeClusters: secondaryPBClusters,
Login: respLogin,
NodeLogin: respLogin,
InstallStatus: installStatusResult,
MaxCPU: types.Int32(node.MaxCPU),
IsOn: node.IsOn == 1,
@@ -717,7 +718,7 @@ func (this *NodeService) InstallNode(ctx context.Context, req *pb.InstallNodeReq
}
go func() {
err = installers.SharedQueue().InstallNodeProcess(req.NodeId, false)
err = installers.SharedNodeQueue().InstallNodeProcess(req.NodeId, false)
if err != nil {
logs.Println("[RPC]install node:" + err.Error())
}
@@ -757,7 +758,7 @@ func (this *NodeService) UpgradeNode(ctx context.Context, req *pb.UpgradeNodeReq
}
go func() {
err = installers.SharedQueue().InstallNodeProcess(req.NodeId, true)
err = installers.SharedNodeQueue().InstallNodeProcess(req.NodeId, true)
if err != nil {
logs.Println("[RPC]install node:" + err.Error())
}
@@ -774,7 +775,7 @@ func (this *NodeService) StartNode(ctx context.Context, req *pb.StartNodeRequest
return nil, err
}
err = installers.SharedQueue().StartNode(req.NodeId)
err = installers.SharedNodeQueue().StartNode(req.NodeId)
if err != nil {
return &pb.StartNodeResponse{
IsOk: false,
@@ -793,7 +794,7 @@ func (this *NodeService) StopNode(ctx context.Context, req *pb.StopNodeRequest)
return nil, err
}
err = installers.SharedQueue().StopNode(req.NodeId)
err = installers.SharedNodeQueue().StopNode(req.NodeId)
if err != nil {
return &pb.StopNodeResponse{
IsOk: false,
@@ -909,7 +910,7 @@ func (this *NodeService) FindAllNotInstalledNodesWithNodeClusterId(ctx context.C
result := []*pb.Node{}
for _, node := range nodes {
// 认证信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(tx, int64(node.Id))
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(tx, nodeconfigs.NodeRoleNode, int64(node.Id))
if err != nil {
return nil, err
}
@@ -967,7 +968,7 @@ func (this *NodeService) FindAllNotInstalledNodesWithNodeClusterId(ctx context.C
IsInstalled: node.IsInstalled == 1,
StatusJSON: []byte(node.Status),
IsOn: node.IsOn == 1,
Login: pbLogin,
NodeLogin: pbLogin,
IpAddresses: pbAddresses,
InstallStatus: pbInstallStatus,
})
@@ -1018,7 +1019,7 @@ func (this *NodeService) FindAllUpgradeNodesWithNodeClusterId(ctx context.Contex
}
for _, node := range nodes {
// 认证信息
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(tx, int64(node.Id))
login, err := models.SharedNodeLoginDAO.FindEnabledNodeLoginWithNodeId(tx, nodeconfigs.NodeRoleNode, int64(node.Id))
if err != nil {
return nil, err
}
@@ -1086,7 +1087,7 @@ func (this *NodeService) FindAllUpgradeNodesWithNodeClusterId(ctx context.Contex
StatusJSON: []byte(node.Status),
IsOn: node.IsOn == 1,
IpAddresses: pbAddresses,
Login: pbLogin,
NodeLogin: pbLogin,
InstallStatus: pbInstallStatus,
}
@@ -1144,7 +1145,7 @@ func (this *NodeService) UpdateNodeLogin(ctx context.Context, req *pb.UpdateNode
tx := this.NullTx()
if req.NodeLogin.Id <= 0 {
_, err := models.SharedNodeLoginDAO.CreateNodeLogin(tx, req.NodeId, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
_, err := models.SharedNodeLoginDAO.CreateNodeLogin(tx, nodeconfigs.NodeRoleNode, req.NodeId, req.NodeLogin.Name, req.NodeLogin.Type, req.NodeLogin.Params)
if err != nil {
return nil, err
}