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

优化节点列表

Browse Source
This commit is contained in:
GoEdgeLab
2022-04-07 18:31:38 +08:00
parent 63d070f7e0
commit cc1e806cee
7 changed files with 329 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
internal/db/models/node_cluster_dao.go
View File

@@ -97,6 +97,7 @@ func (this *NodeClusterDAO) FindAllEnableClusters(tx *dbs.Tx) (result []*NodeClu
_, err = this.Query(tx).
State(NodeClusterStateEnabled).
Slice(&result).
Desc("isPinned").
Desc("order").
DescPk().
FindAll()

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

@@ -385,6 +385,10 @@ func (this *NodeDAO) ListEnabledNodesMatch(tx *dbs.Tx,
query.Asc("IF(JSON_EXTRACT(status, '$.updatedAt')>UNIX_TIMESTAMP()-120, IFNULL(JSON_EXTRACT(status, '$.trafficOutBytes'), 0), 0)")
case "trafficOutDesc":
query.Desc("IF(JSON_EXTRACT(status, '$.updatedAt')>UNIX_TIMESTAMP()-120, IFNULL(JSON_EXTRACT(status, '$.trafficOutBytes'), 0), 0)")
case "loadAsc":
query.Asc("IF(JSON_EXTRACT(status, '$.updatedAt')>UNIX_TIMESTAMP()-120, IFNULL(JSON_EXTRACT(status, '$.load1m'), 0), 0)")
case "loadDesc":
query.Desc("IF(JSON_EXTRACT(status, '$.updatedAt')>UNIX_TIMESTAMP()-120, IFNULL(JSON_EXTRACT(status, '$.load1m'), 0), 0)")
default:
query.Desc("level")
}

99
internal/db/models/node_value_dao.go
View File

@@ -1,6 +1,7 @@
package models
import (
"encoding/json"
"github.com/TeaOSLab/EdgeAPI/internal/errors"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
_ "github.com/go-sql-driver/mysql"
@@ -9,6 +10,7 @@ import (
"github.com/iwind/TeaGo/maps"
"github.com/iwind/TeaGo/types"
timeutil "github.com/iwind/TeaGo/utils/time"
"strings"
"time"
)
@@ -35,9 +37,9 @@ func init() {
// CreateValue 创建值
func (this *NodeValueDAO) CreateValue(tx *dbs.Tx, clusterId int64, role nodeconfigs.NodeRole, nodeId int64, item string, valueJSON []byte, createdAt int64) error {
day := timeutil.FormatTime("Ymd", createdAt)
hour := timeutil.FormatTime("YmdH", createdAt)
minute := timeutil.FormatTime("YmdHi", createdAt)
var day = timeutil.FormatTime("Ymd", createdAt)
var hour = timeutil.FormatTime("YmdH", createdAt)
var minute = timeutil.FormatTime("YmdHi", createdAt)
return this.Query(tx).
InsertOrUpdateQuickly(maps.Map{
@@ -171,6 +173,34 @@ func (this *NodeValueDAO) ListValuesForNSNodes(tx *dbs.Tx, item string, key stri
return
}
// SumAllNodeValues 计算所有节点的某项参数值
func (this *NodeValueDAO) SumAllNodeValues(tx *dbs.Tx, role string, item nodeconfigs.NodeValueItem, param string, duration int32, durationUnit nodeconfigs.NodeValueDurationUnit) (total float64, avg float64, max float64, err error) {
if duration <= 0 {
return 0, 0, 0, nil
}
var query = this.Query(tx).
Result("SUM(JSON_EXTRACT(value, '$."+param+"')) AS sumValue", "AVG(JSON_EXTRACT(value, '$."+param+"')) AS avgValue", "MAX(JSON_EXTRACT(value, '$."+param+"')) AS maxValueResult"). // maxValue 是个MySQL Keyword，这里使用maxValueResult代替
Attr("role", role).
Attr("item", item)
switch durationUnit {
case nodeconfigs.NodeValueDurationUnitMinute:
fromMinute := timeutil.FormatTime("YmdHi", time.Now().Unix()-int64(duration*60))
query.Attr("minute", fromMinute)
default:
fromMinute := timeutil.FormatTime("YmdHi", time.Now().Unix()-int64(duration*60))
query.Attr("minute", fromMinute)
}
m, _, err := query.FindOne()
if err != nil {
return 0, 0, 0, err
}
return m.GetFloat64("sumValue"), m.GetFloat64("avgValue"), m.GetFloat64("maxValueResult"), nil
}
// SumNodeValues 计算节点的某项参数值
func (this *NodeValueDAO) SumNodeValues(tx *dbs.Tx, role string, nodeId int64, item string, param string, method nodeconfigs.NodeValueSumMethod, duration int32, durationUnit nodeconfigs.NodeValueDurationUnit) (float64, error) {
if duration <= 0 {
@@ -261,11 +291,13 @@ func (this *NodeValueDAO) SumNodeClusterValues(tx *dbs.Tx, role string, clusterI
// FindLatestNodeValue 获取最近一条数据
func (this *NodeValueDAO) FindLatestNodeValue(tx *dbs.Tx, role string, nodeId int64, item string) (*NodeValue, error) {
fromMinute := timeutil.FormatTime("YmdHi", time.Now().Unix()-int64(60))
one, err := this.Query(tx).
Attr("role", role).
Attr("nodeId", nodeId).
Attr("item", item).
DescPk().
Attr("minute", fromMinute).
Find()
if err != nil {
return nil, err
@@ -275,3 +307,62 @@ func (this *NodeValueDAO) FindLatestNodeValue(tx *dbs.Tx, role string, nodeId in
}
return one.(*NodeValue), nil
}
// ComposeNodeStatus 组合节点状态值
func (this *NodeValueDAO) ComposeNodeStatus(tx *dbs.Tx, role string, nodeId int64, statusConfig *nodeconfigs.NodeStatus) error {
var items = []string{
nodeconfigs.NodeValueItemCPU,
nodeconfigs.NodeValueItemMemory,
nodeconfigs.NodeValueItemLoad,
nodeconfigs.NodeValueItemTrafficOut,
nodeconfigs.NodeValueItemTrafficIn,
}
ones, err := this.Query(tx).
Result("item", "value").
Attr("role", role).
Attr("nodeId", nodeId).
Attr("minute", timeutil.FormatTime("YmdHi", time.Now().Unix()-60)).
Where("item IN ('" + strings.Join(items, "', '") + "')").
FindAll()
if err != nil {
return err
}
for _, one := range ones {
var oneValue = one.(*NodeValue)
var valueMap = oneValue.DecodeMapValue()
switch oneValue.Item {
case nodeconfigs.NodeValueItemCPU:
statusConfig.CPUUsage = valueMap.GetFloat64("usage")
case nodeconfigs.NodeValueItemMemory:
statusConfig.MemoryUsage = valueMap.GetFloat64("usage")
case nodeconfigs.NodeValueItemLoad:
statusConfig.Load1m = valueMap.GetFloat64("load1m")
statusConfig.Load5m = valueMap.GetFloat64("load5m")
statusConfig.Load15m = valueMap.GetFloat64("load15m")
case nodeconfigs.NodeValueItemTrafficOut:
statusConfig.TrafficOutBytes = valueMap.GetUint64("total")
case nodeconfigs.NodeValueItemTrafficIn:
statusConfig.TrafficInBytes = valueMap.GetUint64("total")
}
}
return nil
}
// ComposeNodeStatusJSON 组合节点状态值，并转换为JSON数据
func (this *NodeValueDAO) ComposeNodeStatusJSON(tx *dbs.Tx, role string, nodeId int64, statusJSON []byte) ([]byte, error) {
var statusConfig = &nodeconfigs.NodeStatus{}
if len(statusJSON) > 0 {
err := json.Unmarshal(statusJSON, statusConfig)
if err != nil {
return nil, err
}
}
err := this.ComposeNodeStatus(tx, role, nodeId, statusConfig)
if err != nil {
return nil, err
}
return json.Marshal(statusConfig)
}

27
internal/db/models/node_value_dao_test.go
View File

@@ -6,6 +6,7 @@ import (
_ "github.com/go-sql-driver/mysql"
_ "github.com/iwind/TeaGo/bootstrap"
"github.com/iwind/TeaGo/dbs"
"github.com/iwind/TeaGo/logs"
"github.com/iwind/TeaGo/maps"
"github.com/iwind/TeaGo/rands"
"github.com/iwind/TeaGo/types"
@@ -52,3 +53,29 @@ func TestNodeValueDAO_CreateManyValues(t *testing.T) {
}
t.Log("finished")
}
func TestNodeValueDAO_SumAllNodeValues(t *testing.T) {
var dao = models.NewNodeValueDAO()
sum, avg, max, err := dao.SumAllNodeValues(nil, nodeconfigs.NodeRoleNode, nodeconfigs.NodeValueItemCPU, "usage", 1, nodeconfigs.NodeValueDurationUnitMinute)
if err != nil {
t.Fatal(err)
}
t.Log("sum:", sum, "avg:", avg, "max:", max)
}
func TestNodeValueDAO_ComposeNodeStatus(t *testing.T) {
var dao = models.NewNodeValueDAO()
one, err := dao.Query(nil).DescPk().Find()
if err != nil {
t.Fatal(err)
}
if one != nil {
var config = &nodeconfigs.NodeStatus{}
err = dao.ComposeNodeStatus(nil, one.(*models.NodeValue).Role, int64(one.(*models.NodeValue).NodeId), config)
if err != nil {
t.Fatal(err)
}
logs.PrintAsJSON(config, t)
}
}

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

@@ -19,6 +19,7 @@ import (
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs/shared"
"github.com/andybalholm/brotli"
"github.com/iwind/TeaGo/dbs"
"github.com/iwind/TeaGo/lists"
"github.com/iwind/TeaGo/logs"
"github.com/iwind/TeaGo/types"
stringutil "github.com/iwind/TeaGo/utils/string"
@@ -181,19 +182,21 @@ func (this *NodeService) ListEnabledNodesMatch(ctx context.Context, req *pb.List
tx := this.NullTx()
clusterDNS, err := models.SharedNodeClusterDAO.FindClusterDNSInfo(tx, req.NodeClusterId, nil)
if err != nil {
return nil, err
}
var dnsDomainId = int64(0)
var domainRoutes = []*dnstypes.Route{}
dnsDomainId := int64(0)
domainRoutes := []*dnstypes.Route{}
if clusterDNS != nil {
dnsDomainId = int64(clusterDNS.DnsDomainId)
if clusterDNS.DnsDomainId > 0 {
domainRoutes, err = dns.SharedDNSDomainDAO.FindDomainRoutes(tx, dnsDomainId)
if err != nil {
return nil, err
if req.NodeClusterId > 0 {
clusterDNS, err := models.SharedNodeClusterDAO.FindClusterDNSInfo(tx, req.NodeClusterId, nil)
if err != nil {
return nil, err
}
if clusterDNS != nil {
dnsDomainId = int64(clusterDNS.DnsDomainId)
if clusterDNS.DnsDomainId > 0 {
domainRoutes, err = dns.SharedDNSDomainDAO.FindDomainRoutes(tx, dnsDomainId)
if err != nil {
return nil, err
}
}
}
}
@@ -216,13 +219,18 @@ func (this *NodeService) ListEnabledNodesMatch(ctx context.Context, req *pb.List
order = "trafficOutAsc"
} else if req.TrafficOutDesc {
order = "trafficOutDesc"
} else if req.LoadAsc {
order = "loadAsc"
} else if req.LoadDesc {
order = "loadDesc"
}
nodes, err := models.SharedNodeDAO.ListEnabledNodesMatch(tx, req.NodeClusterId, configutils.ToBoolState(req.InstallState), configutils.ToBoolState(req.ActiveState), req.Keyword, req.NodeGroupId, req.NodeRegionId, req.Level, true, order, req.Offset, req.Size)
if err != nil {
return nil, err
}
result := []*pb.Node{}
var result = []*pb.Node{}
var cacheMap = utils.NewCacheMap()
for _, node := range nodes {
// 主集群信息
clusterName, err := models.SharedNodeClusterDAO.FindNodeClusterName(tx, int64(node.ClusterId))
@@ -277,19 +285,54 @@ func (this *NodeService) ListEnabledNodesMatch(ctx context.Context, req *pb.List
}
// DNS线路
routeCodes, err := node.DNSRouteCodesForDomainId(dnsDomainId)
if err != nil {
return nil, err
}
pbRoutes := []*pb.DNSRoute{}
for _, routeCode := range routeCodes {
for _, route := range domainRoutes {
if route.Code == routeCode {
pbRoutes = append(pbRoutes, &pb.DNSRoute{
Name: route.Name,
Code: route.Code,
})
break
var pbRoutes = []*pb.DNSRoute{}
if dnsDomainId > 0 {
routeCodes, err := node.DNSRouteCodesForDomainId(dnsDomainId)
if err != nil {
return nil, err
}
for _, routeCode := range routeCodes {
for _, route := range domainRoutes {
if route.Code == routeCode {
pbRoutes = append(pbRoutes, &pb.DNSRoute{
Name: route.Name,
Code: route.Code,
})
break
}
}
}
} else if req.NodeClusterId == 0 {
var clusterDomainIds = []int64{}
for _, clusterId := range node.AllClusterIds() {
clusterDNSInfo, err := models.SharedNodeClusterDAO.FindClusterDNSInfo(tx, clusterId, cacheMap)
if err != nil {
return nil, err
}
if clusterDNSInfo != nil && clusterDNSInfo.DnsDomainId > 0 {
clusterDomainIds = append(clusterDomainIds, int64(clusterDNSInfo.DnsDomainId))
}
}
for domainId, routeCodes := range node.DNSRouteCodes() {
if domainId == 0 {
continue
}
if !lists.ContainsInt64(clusterDomainIds, domainId) {
continue
}
for _, routeCode := range routeCodes {
routeName, err := dns.SharedDNSDomainDAO.FindDomainRouteName(tx, domainId, routeCode)
if err != nil {
return nil, err
}
if len(routeName) > 0 {
pbRoutes = append(pbRoutes, &pb.DNSRoute{
Name: routeName,
Code: routeCode,
})
}
}
}
}
@@ -310,12 +353,18 @@ func (this *NodeService) ListEnabledNodesMatch(ctx context.Context, req *pb.List
}
}
// 状态
statusJSON, err := models.SharedNodeValueDAO.ComposeNodeStatusJSON(tx, nodeconfigs.NodeRoleNode, int64(node.Id), node.Status)
if err != nil {
return nil, err
}
result = append(result, &pb.Node{
Id: int64(node.Id),
Name: node.Name,
Version: int64(node.Version),
IsInstalled: node.IsInstalled,
StatusJSON: node.Status,
StatusJSON: statusJSON,
NodeCluster: &pb.NodeCluster{
Id: int64(node.ClusterId),
Name: clusterName,
@@ -455,6 +504,7 @@ func (this *NodeService) FindEnabledNode(ctx context.Context, req *pb.FindEnable
if err != nil {
return nil, err
}
var clusterIds = []int64{int64(node.ClusterId)}
// 从集群信息
var secondaryPBClusters []*pb.NodeCluster
@@ -471,6 +521,7 @@ func (this *NodeService) FindEnabledNode(ctx context.Context, req *pb.FindEnable
IsOn: cluster.IsOn,
Name: cluster.Name,
})
clusterIds = append(clusterIds, int64(cluster.Id))
}
// 认证信息
@@ -556,10 +607,49 @@ func (this *NodeService) FindEnabledNode(ctx context.Context, req *pb.FindEnable
}
}
// 线路
var pbRoutes = []*pb.DNSRoute{}
var clusterDomainIds = []int64{}
for _, clusterId := range node.AllClusterIds() {
clusterDNSInfo, err := models.SharedNodeClusterDAO.FindClusterDNSInfo(tx, clusterId, nil)
if err != nil {
return nil, err
}
if clusterDNSInfo != nil && clusterDNSInfo.DnsDomainId > 0 {
clusterDomainIds = append(clusterDomainIds, int64(clusterDNSInfo.DnsDomainId))
}
}
for domainId, routeCodes := range node.DNSRouteCodes() {
if domainId == 0 {
continue
}
if !lists.ContainsInt64(clusterDomainIds, domainId) {
continue
}
for _, routeCode := range routeCodes {
routeName, err := dns.SharedDNSDomainDAO.FindDomainRouteName(tx, domainId, routeCode)
if err != nil {
return nil, err
}
if len(routeName) > 0 {
pbRoutes = append(pbRoutes, &pb.DNSRoute{
Name: routeName,
Code: routeCode,
})
}
}
}
// 监控状态
statusJSON, err := models.SharedNodeValueDAO.ComposeNodeStatusJSON(tx, nodeconfigs.NodeRoleNode, int64(node.Id), node.Status)
if err != nil {
return nil, err
}
return &pb.FindEnabledNodeResponse{Node: &pb.Node{
Id: int64(node.Id),
Name: node.Name,
StatusJSON: node.Status,
StatusJSON: statusJSON,
UniqueId: node.UniqueId,
Version: int64(node.Version),
LatestVersion: int64(node.LatestVersion),
@@ -582,6 +672,7 @@ func (this *NodeService) FindEnabledNode(ctx context.Context, req *pb.FindEnable
MaxCacheMemoryCapacity: pbMaxCacheMemoryCapacity,
CacheDiskDir: node.CacheDiskDir,
Level: int32(node.Level),
DnsRoutes: pbRoutes,
}}, nil
}

81
internal/rpc/services/service_node_value.go
View File

@@ -6,7 +6,9 @@ import (
"context"
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/iwind/TeaGo/types"
)
type NodeValueService struct {
@@ -60,7 +62,7 @@ func (this *NodeValueService) ListNodeValues(ctx context.Context, req *pb.ListNo
if err != nil {
return nil, err
}
pbValues := []*pb.NodeValue{}
var pbValues = []*pb.NodeValue{}
for _, value := range values {
pbValues = append(pbValues, &pb.NodeValue{
ValueJSON: value.Value,
@@ -70,3 +72,80 @@ func (this *NodeValueService) ListNodeValues(ctx context.Context, req *pb.ListNo
return &pb.ListNodeValuesResponse{NodeValues: pbValues}, nil
}
// SumAllNodeValueStats 读取所有节点的最新数据
func (this *NodeValueService) SumAllNodeValueStats(ctx context.Context, req *pb.SumAllNodeValueStatsRequest) (*pb.SumAllNodeValueStatsResponse, error) {
_, err := this.ValidateAdmin(ctx, 0)
if err != nil {
return nil, err
}
var tx = this.NullTx()
var result = &pb.SumAllNodeValueStatsResponse{}
// traffic
{
total, _, _, err := models.SharedNodeValueDAO.SumAllNodeValues(tx, nodeconfigs.NodeRoleNode, nodeconfigs.NodeValueItemTrafficOut, "total", 1, nodeconfigs.NodeValueDurationUnitMinute)
if err != nil {
return nil, err
}
result.TotalTrafficBytesPerSecond = types.Int64(total) / 60
}
// cpu
{
_, avg, max, err := models.SharedNodeValueDAO.SumAllNodeValues(tx, nodeconfigs.NodeRoleNode, nodeconfigs.NodeValueItemCPU, "usage", 1, nodeconfigs.NodeValueDurationUnitMinute)
if err != nil {
return nil, err
}
result.AvgCPUUsage = types.Float32(avg)
result.MaxCPUUsage = types.Float32(max)
}
{
total, _, _, err := models.SharedNodeValueDAO.SumAllNodeValues(tx, nodeconfigs.NodeRoleNode, nodeconfigs.NodeValueItemCPU, "cores", 1, nodeconfigs.NodeValueDurationUnitMinute)
if err != nil {
return nil, err
}
result.TotalCPUCores = types.Int32(total)
}
// memory
{
_, avg, max, err := models.SharedNodeValueDAO.SumAllNodeValues(tx, nodeconfigs.NodeRoleNode, nodeconfigs.NodeValueItemMemory, "usage", 1, nodeconfigs.NodeValueDurationUnitMinute)
if err != nil {
return nil, err
}
result.AvgMemoryUsage = types.Float32(avg)
result.MaxMemoryUsage = types.Float32(max)
}
{
total, _, _, err := models.SharedNodeValueDAO.SumAllNodeValues(tx, nodeconfigs.NodeRoleNode, nodeconfigs.NodeValueItemMemory, "total", 1, nodeconfigs.NodeValueDurationUnitMinute)
if err != nil {
return nil, err
}
result.TotalMemoryBytes = types.Int64(total)
}
// load
{
_, avg, max, err := models.SharedNodeValueDAO.SumAllNodeValues(tx, nodeconfigs.NodeRoleNode, nodeconfigs.NodeValueItemLoad, "load1m", 1, nodeconfigs.NodeValueDurationUnitMinute)
if err != nil {
return nil, err
}
result.AvgLoad1Min = types.Float32(avg)
result.MaxLoad1Min = types.Float32(max)
}
{
_, avg, _, err := models.SharedNodeValueDAO.SumAllNodeValues(tx, nodeconfigs.NodeRoleNode, nodeconfigs.NodeValueItemLoad, "load5m", 1, nodeconfigs.NodeValueDurationUnitMinute)
if err != nil {
return nil, err
}
result.AvgLoad5Min = types.Float32(avg)
}
return result, nil
}

3
internal/rpc/services/service_server_group.go
View File

@@ -433,6 +433,7 @@ func (this *ServerGroupService) FindEnabledServerGroupConfigInfo(ctx context.Con
ServerGroupId: int64(group.Id),
}
// http
if len(group.HttpReverseProxy) > 0 {
var ref = &serverconfigs.ReverseProxyRef{}
err = json.Unmarshal(group.HttpReverseProxy, ref)
@@ -442,6 +443,7 @@ func (this *ServerGroupService) FindEnabledServerGroupConfigInfo(ctx context.Con
result.HasHTTPReverseProxy = ref.IsPrior
}
// tcp
if len(group.TcpReverseProxy) > 0 {
var ref = &serverconfigs.ReverseProxyRef{}
err = json.Unmarshal(group.TcpReverseProxy, ref)
@@ -451,6 +453,7 @@ func (this *ServerGroupService) FindEnabledServerGroupConfigInfo(ctx context.Con
result.HasTCPReverseProxy = ref.IsPrior
}
// udp
if len(group.UdpReverseProxy) > 0 {
var ref = &serverconfigs.ReverseProxyRef{}
err = json.Unmarshal(group.UdpReverseProxy, ref)