增加SSH登录建议端口接口

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"
+		var sock = gosock.NewTmpSock("edge-node")
-		conn, err := net.Dial("unix", path)
+		if sock.IsListening() {
-		if err == nil {
+			// 从systemd中停止
-			_ = conn.Close()
+			systemctl, _ := exec.LookPath("systemctl")
-			stderr("test node status: edge node is running now, can not install again")
+			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 {

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)

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)
 	}

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
+}

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)
+}

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)
 	}

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)

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
+}