阶段性提交

internal/rpc/services/service_node.go

@@ -2,9 +2,13 @@ package services
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
 	"github.com/TeaOSLab/EdgeAPI/internal/rpc/pb"
 	rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
+	"github.com/iwind/TeaGo/logs"
+	"github.com/iwind/TeaGo/maps"
 )
 
 type NodeService struct {
@@ -66,8 +70,9 @@ func (this *NodeService) ListEnabledNodes(ctx context.Context, req *pb.ListEnabl
 		}
 
 		result = append(result, &pb.Node{
-			Id:   int64(node.Id),
-			Name: node.Name,
+			Id:     int64(node.Id),
+			Name:   node.Name,
+			Status: node.Status,
 			Cluster: &pb.NodeCluster{
 				Id:   int64(node.ClusterId),
 				Name: clusterName,
@@ -166,8 +171,9 @@ func (this *NodeService) FindEnabledNode(ctx context.Context, req *pb.FindEnable
 	}
 
 	return &pb.FindEnabledNodeResponse{Node: &pb.Node{
-		Id:   int64(node.Id),
-		Name: node.Name,
+		Id:     int64(node.Id),
+		Name:   node.Name,
+		Status: node.Status,
 		Cluster: &pb.NodeCluster{
 			Id:   int64(node.ClusterId),
 			Name: clusterName,
@@ -175,3 +181,113 @@ func (this *NodeService) FindEnabledNode(ctx context.Context, req *pb.FindEnable
 		Login: respLogin,
 	}}, nil
 }
+
+// 组合节点配置
+func (this *NodeService) ComposeNodeConfig(ctx context.Context, req *pb.ComposeNodeConfigRequest) (*pb.ComposeNodeConfigResponse, error) {
+	// 校验节点
+	_, nodeId, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeNode)
+	if err != nil {
+		return nil, err
+	}
+
+	node, err := models.SharedNodeDAO.FindEnabledNode(nodeId)
+	if err != nil {
+		return nil, err
+	}
+	if node == nil {
+		return nil, errors.New("node validate failed, please check 'nodeId' or 'secret'")
+	}
+
+	nodeMap := maps.Map{
+		"id":      node.UniqueId,
+		"isOn":    node.IsOn == 1,
+		"servers": []maps.Map{},
+		"version": node.Version,
+	}
+
+	// 获取所有的服务
+	servers, err := models.SharedServerDAO.FindAllEnabledServersWithNode(int64(node.Id))
+	if err != nil {
+		return nil, err
+	}
+
+	serverMaps := []maps.Map{}
+	for _, server := range servers {
+		if len(server.Config) == 0 {
+			continue
+		}
+		configMap := maps.Map{}
+		err = json.Unmarshal([]byte(server.Config), &configMap)
+		if err != nil {
+			return nil, err
+		}
+		configMap["id"] = server.UniqueId
+		configMap["version"] = server.Version
+		serverMaps = append(serverMaps, configMap)
+	}
+	nodeMap["servers"] = serverMaps
+
+	data, err := json.Marshal(nodeMap)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.ComposeNodeConfigResponse{ConfigJSON: data}, nil
+}
+
+// 节点stream
+func (this *NodeService) NodeStream(server pb.NodeService_NodeStreamServer) error {
+	// 校验节点
+	_, nodeId, err := rpcutils.ValidateRequest(server.Context(), rpcutils.UserTypeNode)
+	if err != nil {
+		return err
+	}
+	logs.Println("nodeId:", nodeId)
+
+	for {
+		req, err := server.Recv()
+		if err != nil {
+			return err
+		}
+		logs.Println("received:", req)
+	}
+}
+
+// 更新节点状态
+func (this *NodeService) UpdateNodeStatus(ctx context.Context, req *pb.UpdateNodeStatusRequest) (*pb.UpdateNodeStatusResponse, error) {
+	// 校验节点
+	_, nodeId, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeNode)
+	if err != nil {
+		return nil, err
+	}
+
+	if req.NodeId > 0 {
+		nodeId = req.NodeId
+	}
+
+	if nodeId <= 0 {
+		return nil, errors.New("'nodeId' should be greater than 0")
+	}
+
+	err = models.SharedNodeDAO.UpdateNodeStatus(nodeId, req.StatusJSON)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.UpdateNodeStatusResponse{}, nil
+}
+
+// 同步集群中的节点版本
+func (this *NodeService) SyncNodesVersionWithCluster(ctx context.Context, req *pb.SyncNodesVersionWithClusterRequest) (*pb.SyncNodesVersionWithClusterResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	err = models.SharedNodeDAO.SyncNodeVersionsWithCluster(req.ClusterId)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.SyncNodesVersionWithClusterResponse{}, nil
+}

internal/rpc/services/service_node_cluster.go

@@ -36,3 +36,37 @@ func (this *NodeClusterService) FindAllEnabledClusters(ctx context.Context, req
 		Clusters: result,
 	}, nil
 }
+
+// 查找所有变更的集群
+func (this *NodeClusterService) FindAllChangedClusters(ctx context.Context, req *pb.FindAllChangedClustersRequest) (*pb.FindAllChangedClustersResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	clusterIds, err := models.SharedNodeDAO.FindChangedClusterIds()
+	if err != nil {
+		return nil, err
+	}
+	if len(clusterIds) == 0 {
+		return &pb.FindAllChangedClustersResponse{
+			Clusters: []*pb.NodeCluster{},
+		}, nil
+	}
+	result := []*pb.NodeCluster{}
+	for _, clusterId := range clusterIds {
+		cluster, err := models.SharedNodeClusterDAO.FindEnabledNodeCluster(clusterId)
+		if err != nil {
+			return nil, err
+		}
+		if cluster == nil {
+			continue
+		}
+		result = append(result, &pb.NodeCluster{
+			Id:        int64(cluster.Id),
+			Name:      cluster.Name,
+			CreatedAt: int64(cluster.CreatedAt),
+		})
+	}
+	return &pb.FindAllChangedClustersResponse{Clusters: result}, nil
+}

internal/rpc/services/service_server.go

@@ -0,0 +1,186 @@
+package services
+
+import (
+	"context"
+	"errors"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/rpc/pb"
+	rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
+)
+
+type ServerService struct {
+}
+
+// 创建服务
+func (this *ServerService) CreateServer(ctx context.Context, req *pb.CreateServerRequest) (*pb.CreateServerResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+	serverId, err := models.SharedServerDAO.CreateServer(req.AdminId, req.UserId, req.ClusterId, string(req.Config), string(req.IncludeNodesJSON), string(req.ExcludeNodesJSON))
+	if err != nil {
+		return nil, err
+	}
+
+	// 更新节点版本
+	err = models.SharedNodeDAO.UpdateAllNodesLatestVersionMatch(req.ClusterId)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.CreateServerResponse{ServerId: serverId}, nil
+}
+
+// 修改服务
+func (this *ServerService) UpdateServer(ctx context.Context, req *pb.UpdateServerRequest) (*pb.UpdateServerResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	if req.ServerId <= 0 {
+		return nil, errors.New("invalid serverId")
+	}
+
+	// 查询老的节点信息
+	server, err := models.SharedServerDAO.FindEnabledServer(req.ServerId)
+	if err != nil {
+		return nil, err
+	}
+	if server == nil {
+		return nil, errors.New("can not find server")
+	}
+
+	err = models.SharedServerDAO.UpdateServer(req.ServerId, req.ClusterId, string(req.Config), string(req.IncludeNodesJSON), string(req.ExcludeNodesJSON))
+	if err != nil {
+		return nil, err
+	}
+
+	// 更新老的节点版本
+	if req.ClusterId != int64(server.ClusterId) {
+		err = models.SharedNodeDAO.UpdateAllNodesLatestVersionMatch(int64(server.ClusterId))
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// 更新新的节点版本
+	err = models.SharedNodeDAO.UpdateAllNodesLatestVersionMatch(req.ClusterId)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.UpdateServerResponse{}, nil
+}
+
+// 计算服务数量
+func (this *ServerService) CountAllEnabledServers(ctx context.Context, req *pb.CountAllEnabledServersRequest) (*pb.CountAllEnabledServersResponse, error) {
+	_ = req
+	_, _, err := rpcutils.ValidateRequest(ctx)
+	if err != nil {
+		return nil, err
+	}
+	count, err := models.SharedServerDAO.CountAllEnabledServers()
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.CountAllEnabledServersResponse{Count: count}, nil
+}
+
+// 列出单页服务
+func (this *ServerService) ListEnabledServers(ctx context.Context, req *pb.ListEnabledServersRequest) (*pb.ListEnabledServersResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+	servers, err := models.SharedServerDAO.ListEnabledServers(req.Offset, req.Size)
+	if err != nil {
+		return nil, err
+	}
+	result := []*pb.Server{}
+	for _, server := range servers {
+		clusterName, err := models.SharedNodeClusterDAO.FindNodeClusterName(int64(server.ClusterId))
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, &pb.Server{
+			Id:           int64(server.Id),
+			Config:       []byte(server.Config),
+			IncludeNodes: []byte(server.IncludeNodes),
+			ExcludeNodes: []byte(server.ExcludeNodes),
+			CreatedAt:    int64(server.CreatedAt),
+			Cluster: &pb.NodeCluster{
+				Id:   int64(server.ClusterId),
+				Name: clusterName,
+			},
+		})
+	}
+
+	return &pb.ListEnabledServersResponse{Servers: result}, nil
+}
+
+// 禁用某服务
+func (this *ServerService) DisableServer(ctx context.Context, req *pb.DisableServerRequest) (*pb.DisableServerResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	// 查找服务
+	server, err := models.SharedServerDAO.FindEnabledServer(req.ServerId)
+	if err != nil {
+		return nil, err
+	}
+	if server == nil {
+		return nil, errors.New("can not find the server")
+	}
+
+	// 禁用服务
+	err = models.SharedServerDAO.DisableServer(req.ServerId)
+	if err != nil {
+		return nil, err
+	}
+
+	// 更新节点版本
+	err = models.SharedNodeDAO.UpdateAllNodesLatestVersionMatch(int64(server.ClusterId))
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.DisableServerResponse{}, nil
+}
+
+// 查找单个服务
+func (this *ServerService) FindEnabledServer(ctx context.Context, req *pb.FindEnabledServerRequest) (*pb.FindEnabledServerResponse, error) {
+	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
+	if err != nil {
+		return nil, err
+	}
+
+	server, err := models.SharedServerDAO.FindEnabledServer(req.ServerId)
+	if err != nil {
+		return nil, err
+	}
+
+	if server == nil {
+		return &pb.FindEnabledServerResponse{}, nil
+	}
+
+	clusterName, err := models.SharedNodeClusterDAO.FindNodeClusterName(int64(server.ClusterId))
+	if err != nil {
+		return nil, err
+	}
+
+	return &pb.FindEnabledServerResponse{Server: &pb.Server{
+		Id:           int64(server.Id),
+		Config:       []byte(server.Config),
+		IncludeNodes: []byte(server.IncludeNodes),
+		ExcludeNodes: []byte(server.ExcludeNodes),
+		CreatedAt:    int64(server.CreatedAt),
+		Cluster: &pb.NodeCluster{
+			Id:   int64(server.ClusterId),
+			Name: clusterName,
+		},
+	}}, nil
+}

internal/rpc/services/servie_server.go

@@ -1,68 +0,0 @@
-package services
-
-import (
-	"context"
-	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
-	"github.com/TeaOSLab/EdgeAPI/internal/rpc/pb"
-	rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
-)
-
-type ServerService struct {
-}
-
-func (this *ServerService) CreateServer(ctx context.Context, req *pb.CreateServerRequest) (*pb.CreateServerResponse, error) {
-	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
-	if err != nil {
-		return nil, err
-	}
-	serverId, err := models.SharedServerDAO.CreateServer(req.AdminId, req.UserId, req.ClusterId, string(req.Config), string(req.IncludeNodesJSON), string(req.ExcludeNodesJSON))
-	if err != nil {
-		return nil, err
-	}
-	return &pb.CreateServerResponse{ServerId: serverId}, nil
-}
-
-func (this *ServerService) CountAllEnabledServers(ctx context.Context, req *pb.CountAllEnabledServersRequest) (*pb.CountAllEnabledServersResponse, error) {
-	_ = req
-	_, _, err := rpcutils.ValidateRequest(ctx)
-	if err != nil {
-		return nil, err
-	}
-	count, err := models.SharedServerDAO.CountAllEnabledServers()
-	if err != nil {
-		return nil, err
-	}
-
-	return &pb.CountAllEnabledServersResponse{Count: count}, nil
-}
-
-func (this *ServerService) ListEnabledServers(ctx context.Context, req *pb.ListEnabledServersRequest) (*pb.ListEnabledServersResponse, error) {
-	_, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin)
-	if err != nil {
-		return nil, err
-	}
-	servers, err := models.SharedServerDAO.ListEnabledServers(req.Offset, req.Size)
-	if err != nil {
-		return nil, err
-	}
-	result := []*pb.Server{}
-	for _, server := range servers {
-		clusterName, err := models.SharedNodeClusterDAO.FindNodeClusterName(int64(server.ClusterId))
-		if err != nil {
-			return nil, err
-		}
-		result = append(result, &pb.Server{
-			Id:           int64(server.Id),
-			Config:       []byte(server.Config),
-			IncludeNodes: []byte(server.IncludeNodes),
-			ExcludeNodes: []byte(server.ExcludeNodes),
-			CreatedAt:    int64(server.CreatedAt),
-			Cluster: &pb.NodeCluster{
-				Id:   int64(server.ClusterId),
-				Name: clusterName,
-			},
-		})
-	}
-
-	return &pb.ListEnabledServersResponse{Servers: result}, nil
-}