diff --git a/cmd/installer-helper/main.go b/cmd/installer-helper/main.go index 919567a5..f9b606d5 100644 --- a/cmd/installer-helper/main.go +++ b/cmd/installer-helper/main.go @@ -3,8 +3,9 @@ package main import ( "flag" "github.com/TeaOSLab/EdgeAPI/internal/utils" - "net" + "github.com/iwind/gosock/pkg/gosock" "os" + "os/exec" ) func main() { @@ -24,11 +25,21 @@ func main() { stderr("need '-cmd=COMMAND' argument") } else if cmd == "test" { // 检查是否正在运行 - path := os.TempDir() + "/edge-node.sock" - conn, err := net.Dial("unix", path) - if err == nil { - _ = conn.Close() - stderr("test node status: edge node is running now, can not install again") + var sock = gosock.NewTmpSock("edge-node") + if sock.IsListening() { + // 从systemd中停止 + systemctl, _ := exec.LookPath("systemctl") + if len(systemctl) > 0 { + systemctlCmd := exec.Command(systemctl, "stop", "edge-node") + _ = systemctlCmd.Run() + } + + // 从进程中停止 + if sock.IsListening() { + _, _ = sock.Send(&gosock.Command{ + Code: "stop", + }) + } } } else if cmd == "unzip" { // 解压 if len(zipPath) == 0 { diff --git a/internal/db/models/http_access_log_dao_test.go b/internal/db/models/http_access_log_dao_test.go index 4c989284..c75a6b09 100644 --- a/internal/db/models/http_access_log_dao_test.go +++ b/internal/db/models/http_access_log_dao_test.go @@ -41,7 +41,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs(t *testing.T) { t.Fatal(err) } - accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "", 10, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0, "") + accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "", 10, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0, "", "", "") if err != nil { t.Fatal(err) } @@ -68,7 +68,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs_Page(t *testing.T) { times := 0 // 防止循环次数太多 for { before := time.Now() - accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0, "") + accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd"), 0, false, false, 0, 0, 0, false, 0, "", "", "") cost := time.Since(before).Seconds() if err != nil { t.Fatal(err) @@ -99,7 +99,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs_Reverse(t *testing.T) { } before := time.Now() - accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "16023261176446590001000000000000003500000004", 2, timeutil.Format("Ymd"), 0, true, false, 0, 0, 0, false, 0, "") + accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, "16023261176446590001000000000000003500000004", 2, timeutil.Format("Ymd"), 0, true, false, 0, 0, 0, false, 0, "", "", "") cost := time.Since(before).Seconds() if err != nil { t.Fatal(err) @@ -124,7 +124,7 @@ func TestHTTPAccessLogDAO_ListAccessLogs_Page_NotExists(t *testing.T) { times := 0 // 防止循环次数太多 for { before := time.Now() - accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd", time.Now().AddDate(0, 0, 1)), 0, false, false, 0, 0, 0, false, 0, "") + accessLogs, requestId, hasMore, err := SharedHTTPAccessLogDAO.ListAccessLogs(tx, lastRequestId, 2, timeutil.Format("Ymd", time.Now().AddDate(0, 0, 1)), 0, false, false, 0, 0, 0, false, 0, "", "", "") cost := time.Since(before).Seconds() if err != nil { t.Fatal(err) diff --git a/internal/db/models/message_dao_test.go b/internal/db/models/message_dao_test.go index fee71525..ead42881 100644 --- a/internal/db/models/message_dao_test.go +++ b/internal/db/models/message_dao_test.go @@ -1,6 +1,7 @@ package models import ( + "github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs" _ "github.com/go-sql-driver/mysql" "github.com/iwind/TeaGo/dbs" "testing" @@ -11,7 +12,7 @@ func TestMessageDAO_CreateClusterMessage(t *testing.T) { var tx *dbs.Tx dao := NewMessageDAO() - err := dao.CreateClusterMessage(tx, 1, "test", "error", "123", "123", []byte("456")) + err := dao.CreateClusterMessage(tx, nodeconfigs.NodeRoleNode, 1, "test", "error", "123", "123", []byte("456")) if err != nil { t.Fatal(err) } diff --git a/internal/db/models/node_login_dao.go b/internal/db/models/node_login_dao.go index 78c4d0cc..5b5e0dab 100644 --- a/internal/db/models/node_login_dao.go +++ b/internal/db/models/node_login_dao.go @@ -136,3 +136,22 @@ func (this *NodeLoginDAO) DisableNodeLogins(tx *dbs.Tx, role nodeconfigs.NodeRol Update() return err } + +func (this *NodeLoginDAO) FindFrequentPorts(tx *dbs.Tx) ([]int32, error) { + ones, _, err := this.Query(tx). + Attr("state", NodeLoginStateEnabled). + Result("JSON_EXTRACT(params, '$.port') as `port`", "COUNT(*) AS c"). + Having("port>0"). + Desc("c"). + Limit(10). + Group("port"). + FindOnes() + if err != nil { + return nil, err + } + var ports = []int32{} + for _, one := range ones { + ports = append(ports, one.GetInt32("port")) + } + return ports, nil +} diff --git a/internal/db/models/node_login_dao_test.go b/internal/db/models/node_login_dao_test.go index 97c24b56..87b3e043 100644 --- a/internal/db/models/node_login_dao_test.go +++ b/internal/db/models/node_login_dao_test.go @@ -2,4 +2,16 @@ package models import ( _ "github.com/go-sql-driver/mysql" + "github.com/iwind/TeaGo/dbs" + "testing" ) + +func TestNodeLoginDAO_FindFrequentPorts(t *testing.T) { + dbs.NotifyReady() + + ports, err := SharedNodeLoginDAO.FindFrequentPorts(nil) + if err != nil { + t.Fatal(err) + } + t.Log(ports) +} diff --git a/internal/db/models/node_task_dao_test.go b/internal/db/models/node_task_dao_test.go index 51603a52..c4df7db2 100644 --- a/internal/db/models/node_task_dao_test.go +++ b/internal/db/models/node_task_dao_test.go @@ -1,6 +1,7 @@ package models import ( + "github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs" _ "github.com/go-sql-driver/mysql" "github.com/iwind/TeaGo/dbs" "testing" @@ -10,7 +11,7 @@ func TestNodeTaskDAO_CreateNodeTask(t *testing.T) { dbs.NotifyReady() var tx *dbs.Tx - err := SharedNodeTaskDAO.CreateNodeTask(tx, 1, 2, NodeTaskTypeConfigChanged) + err := SharedNodeTaskDAO.CreateNodeTask(tx, nodeconfigs.NodeRoleNode, 1, 2, NodeTaskTypeConfigChanged) if err != nil { t.Fatal(err) } @@ -21,7 +22,7 @@ func TestNodeTaskDAO_CreateClusterTask(t *testing.T) { dbs.NotifyReady() var tx *dbs.Tx - err := SharedNodeTaskDAO.CreateClusterTask(tx, 1, NodeTaskTypeConfigChanged) + err := SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, 1, NodeTaskTypeConfigChanged) if err != nil { t.Fatal(err) } @@ -32,7 +33,7 @@ func TestNodeTaskDAO_ExtractClusterTask(t *testing.T) { dbs.NotifyReady() var tx *dbs.Tx - err := SharedNodeTaskDAO.ExtractClusterTask(tx, 1, NodeTaskTypeConfigChanged) + err := SharedNodeTaskDAO.ExtractNodeClusterTask(tx, 1, NodeTaskTypeConfigChanged) if err != nil { t.Fatal(err) } diff --git a/internal/nodes/api_node_services.go b/internal/nodes/api_node_services.go index 533e88f5..9caf698a 100644 --- a/internal/nodes/api_node_services.go +++ b/internal/nodes/api_node_services.go @@ -158,6 +158,11 @@ func (this *APINode) registerServices(server *grpc.Server) { pb.RegisterNodeLogServiceServer(server, instance) this.rest(instance) } + { + instance := this.serviceInstance(&services.NodeLoginService{}).(*services.NodeLoginService) + pb.RegisterNodeLoginServiceServer(server, instance) + this.rest(instance) + } { instance := this.serviceInstance(&services.HTTPAccessLogService{}).(*services.HTTPAccessLogService) pb.RegisterHTTPAccessLogServiceServer(server, instance) diff --git a/internal/rpc/services/service_node_login.go b/internal/rpc/services/service_node_login.go new file mode 100644 index 00000000..cdf652f5 --- /dev/null +++ b/internal/rpc/services/service_node_login.go @@ -0,0 +1,68 @@ +// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved. + +package services + +import ( + "context" + "github.com/TeaOSLab/EdgeAPI/internal/db/models" + "github.com/TeaOSLab/EdgeCommon/pkg/configutils" + "github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb" + "github.com/iwind/TeaGo/types" + "net" + "sync" + "time" +) + +// NodeLoginService 节点登录相关 +type NodeLoginService struct { + BaseService +} + +// FindNodeLoginSuggestPorts 读取建议的端口 +func (this *NodeLoginService) FindNodeLoginSuggestPorts(ctx context.Context, req *pb.FindNodeLoginSuggestPortsRequest) (*pb.FindNodeLoginSuggestPortsResponse, error) { + _, err := this.ValidateAdmin(ctx, 0) + if err != nil { + return nil, err + } + + var tx = this.NullTx() + ports, err := models.SharedNodeLoginDAO.FindFrequentPorts(tx) + if err != nil { + return nil, err + } + + var availablePorts = []int32{} + + // 测试端口连通性 + if len(ports) > 0 && len(req.Host) > 0 { + var host = configutils.QuoteIP(req.Host) + + wg := sync.WaitGroup{} + wg.Add(len(ports)) + + var locker sync.Mutex + + for _, port := range ports { + go func(port int32) { + defer wg.Done() + + conn, err := net.DialTimeout("tcp", host+":"+types.String(port), 2*time.Second) + if err != nil { + return + } + _ = conn.Close() + + locker.Lock() + availablePorts = append(availablePorts, port) + locker.Unlock() + }(port) + } + wg.Wait() + + } + + return &pb.FindNodeLoginSuggestPortsResponse{ + Ports: ports, + AvailablePorts: availablePorts, + }, nil +}