实现节点看板（仅对企业版开放）

2025-12-27 15:46:35 +08:00 · 2021-07-06 20:06:34 +08:00
parent f95b5193ba
commit 7fecfe68bd
9 changed files with 532 additions and 12 deletions
--- a/internal/db/models/metric_stat_dao.go
+++ b/internal/db/models/metric_stat_dao.go
@@ -101,7 +101,7 @@ func (this *MetricStatDAO) ListItemStats(tx *dbs.Tx, itemId int64, version int32
 	return
 }

-// FindItemStatsWithClusterIdAndLastTime 取得最近一次计时前 N 个数据
+// FindItemStatsWithClusterIdAndLastTime 取得集群最近一次计时前 N 个数据
 // 适合每条数据中包含不同的Key的场景
 func (this *MetricStatDAO) FindItemStatsWithClusterIdAndLastTime(tx *dbs.Tx, clusterId int64, itemId int64, version int32, size int64) (result []*MetricStat, err error) {
 	// 最近一次时间
@@ -135,6 +135,40 @@ func (this *MetricStatDAO) FindItemStatsWithClusterIdAndLastTime(tx *dbs.Tx, clu
 	return
 }

+// FindItemStatsWithNodeIdAndLastTime 取得节点最近一次计时前 N 个数据
+// 适合每条数据中包含不同的Key的场景
+func (this *MetricStatDAO) FindItemStatsWithNodeIdAndLastTime(tx *dbs.Tx, nodeId int64, itemId int64, version int32, size int64) (result []*MetricStat, err error) {
+	// 最近一次时间
+	statOne, err := this.Query(tx).
+		Attr("itemId", itemId).
+		Attr("version", version).
+		DescPk().
+		Find()
+	if err != nil {
+		return nil, err
+	}
+	if statOne == nil {
+		return nil, nil
+	}
+	var lastStat = statOne.(*MetricStat)
+	var lastTime = lastStat.Time
+
+	_, err = this.Query(tx).
+		Attr("nodeId", nodeId).
+		Attr("itemId", itemId).
+		Attr("version", version).
+		Attr("time", lastTime).
+		// TODO 增加更多聚合算法，比如 AVG、MEDIAN、MIN、MAX 等
+		// TODO 这里的 MIN(`keys`) 在MySQL8中可以换成FIRST_VALUE
+		Result("MIN(time) AS time", "SUM(value) AS value", "keys").
+		Desc("value").
+		Group("keys").
+		Limit(size).
+		Slice(&result).
+		FindAll()
+	return
+}
+
 // FindLatestItemStatsWithClusterId 取得集群最近 N 个时间的数据
 // 适合同个Key在不同时间段的变化场景
 func (this *MetricStatDAO) FindLatestItemStatsWithClusterId(tx *dbs.Tx, clusterId int64, itemId int64, version int32, size int64) (result []*MetricStat, err error) {
@@ -156,3 +190,25 @@ func (this *MetricStatDAO) FindLatestItemStatsWithClusterId(tx *dbs.Tx, clusterI
 	lists.Reverse(result)
 	return
 }
+
+// FindLatestItemStatsWithNodeId 取得节点最近 N 个时间的数据
+// 适合同个Key在不同时间段的变化场景
+func (this *MetricStatDAO) FindLatestItemStatsWithNodeId(tx *dbs.Tx, nodeId int64, itemId int64, version int32, size int64) (result []*MetricStat, err error) {
+	_, err = this.Query(tx).
+		Attr("nodeId", nodeId).
+		Attr("itemId", itemId).
+		Attr("version", version).
+		// TODO 增加更多聚合算法，比如 AVG、MEDIAN、MIN、MAX 等
+		// TODO 这里的 MIN(`keys`) 在MySQL8中可以换成FIRST_VALUE
+		Result("time", "SUM(value) AS value", "MIN(`keys`) AS `keys`").
+		Desc("time").
+		Group("time").
+		Limit(size).
+		Slice(&result).
+		FindAll()
+	if err != nil {
+		return nil, err
+	}
+	lists.Reverse(result)
+	return
+}
--- a/internal/db/models/metric_sum_stat_dao.go
+++ b/internal/db/models/metric_sum_stat_dao.go
@@ -46,8 +46,8 @@ func (this *MetricSumStatDAO) UpdateSum(tx *dbs.Tx, clusterId int64, nodeId int6
 		})
 }

-// FindNodeSum 查找节点上的统计数据
-func (this *MetricSumStatDAO) FindNodeSum(tx *dbs.Tx, nodeId int64, serverId int64, time string, itemId int64, version int32) (count int64, total float32, err error) {
+// FindServerSum 查找某个服务的统计数据
+func (this *MetricSumStatDAO) FindServerSum(tx *dbs.Tx, nodeId int64, serverId int64, time string, itemId int64, version int32) (count int64, total float32, err error) {
 	one, err := this.Query(tx).
 		Attr("nodeId", nodeId).
 		Attr("serverId", serverId).
@@ -81,3 +81,21 @@ func (this *MetricSumStatDAO) FindClusterSum(tx *dbs.Tx, clusterId int64, time s
 	}
 	return int64(one.(*MetricSumStat).Count), float32(one.(*MetricSumStat).Total), nil
 }
+
+// FindNodeSum 查找节点上的统计数据
+func (this *MetricSumStatDAO) FindNodeSum(tx *dbs.Tx, nodeId int64, time string, itemId int64, version int32) (count int64, total float32, err error) {
+	one, err := this.Query(tx).
+		Attr("nodeId", nodeId).
+		Attr("time", time).
+		Attr("itemId", itemId).
+		Attr("version", version).
+		Result("SUM(count) AS `count`, SUM(total) AS total").
+		Find()
+	if err != nil {
+		return 0, 0, err
+	}
+	if one == nil {
+		return
+	}
+	return int64(one.(*MetricSumStat).Count), float32(one.(*MetricSumStat).Total), nil
+}
--- a/internal/db/models/node_value_dao.go
+++ b/internal/db/models/node_value_dao.go
@@ -92,7 +92,7 @@ func (this *NodeValueDAO) ListValues(tx *dbs.Tx, role string, nodeId int64, item
 }

 // ListValuesWithClusterId 列出集群最近的的平均数据
-func (this *NodeValueDAO) ListValuesWithClusterId(tx *dbs.Tx, clusterId int64, role string, item string, key string, timeRange nodeconfigs.NodeValueRange) (result []*NodeValue, err error) {
+func (this *NodeValueDAO) ListValuesWithClusterId(tx *dbs.Tx, role string, clusterId int64, item string, key string, timeRange nodeconfigs.NodeValueRange) (result []*NodeValue, err error) {
 	query := this.Query(tx).
 		Attr("role", role).
 		Attr("clusterId", clusterId).
@@ -144,3 +144,20 @@ func (this *NodeValueDAO) SumValues(tx *dbs.Tx, role string, nodeId int64, item
 	}
 	return query.FindFloat64Col(0)
 }
+
+// FindLatestNodeValue 获取最近一条数据
+func (this *NodeValueDAO) FindLatestNodeValue(tx *dbs.Tx, role string, nodeId int64, item string) (*NodeValue, error) {
+	one, err := this.Query(tx).
+		Attr("role", role).
+		Attr("nodeId", nodeId).
+		Attr("item", item).
+		DescPk().
+		Find()
+	if err != nil {
+		return nil, err
+	}
+	if one == nil {
+		return nil, nil
+	}
+	return one.(*NodeValue), nil
+}
--- a/internal/db/models/node_value_model_ext.go
+++ b/internal/db/models/node_value_model_ext.go
@@ -1 +1,19 @@
 package models
+
+import (
+	"encoding/json"
+	"github.com/iwind/TeaGo/maps"
+)
+
+func (this *NodeValue) DecodeMapValue() maps.Map {
+	if len(this.Value) == 0 {
+		return maps.Map{}
+	}
+	var m = maps.Map{}
+	err := json.Unmarshal([]byte(this.Value), &m)
+	if err != nil {
+		// 忽略错误
+		return m
+	}
+	return m
+}
--- a/internal/db/models/stats/node_traffic_daily_stat_dao.go
+++ b/internal/db/models/stats/node_traffic_daily_stat_dao.go
@@ -3,6 +3,7 @@ package stats
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
 	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeAPI/internal/utils"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
@@ -78,6 +79,37 @@ func (this *NodeTrafficDailyStatDAO) IncreaseDailyStat(tx *dbs.Tx, clusterId int
 	return nil
 }

+
+// FindDailyStats 获取日期之间统计
+func (this *NodeTrafficDailyStatDAO) FindDailyStats(tx *dbs.Tx, role string, nodeId int64, dayFrom string, dayTo string) (result []*NodeTrafficDailyStat, err error) {
+	ones, err := this.Query(tx).
+		Attr("nodeId", nodeId).
+		Attr("role", role).
+		Between("day", dayFrom, dayTo).
+		FindAll()
+	if err != nil {
+		return nil, err
+	}
+	dayMap := map[string]*NodeTrafficDailyStat{} // day => Stat
+	for _, one := range ones {
+		stat := one.(*NodeTrafficDailyStat)
+		dayMap[stat.Day] = stat
+	}
+	days, err := utils.RangeDays(dayFrom, dayTo)
+	if err != nil {
+		return nil, err
+	}
+	for _, day := range days {
+		stat, ok := dayMap[day]
+		if ok {
+			result = append(result, stat)
+		} else {
+			result = append(result, &NodeTrafficDailyStat{Day: day})
+		}
+	}
+	return result, nil
+}
+
 // Clean 清理历史数据
 func (this *NodeTrafficDailyStatDAO) Clean(tx *dbs.Tx, days int) error {
 	var day = timeutil.Format("Ymd", time.Now().AddDate(0, 0, -days))
--- a/internal/db/models/stats/node_traffic_hourly_stat_dao.go
+++ b/internal/db/models/stats/node_traffic_hourly_stat_dao.go
@@ -110,10 +110,11 @@ func (this *NodeTrafficHourlyStatDAO) FindHourlyStatsWithClusterId(tx *dbs.Tx, c
 	return result, nil
 }

-// FindTopNodeStatsWithClusterId 取得一定时间内的节点排行数据
-func (this *NodeTrafficHourlyStatDAO) FindTopNodeStatsWithClusterId(tx *dbs.Tx, clusterId int64, hourFrom string, hourTo string) (result []*NodeTrafficHourlyStat, err error) {
+// FindTopNodeStatsWithClusterId 取得集群一定时间内的节点排行数据
+func (this *NodeTrafficHourlyStatDAO) FindTopNodeStatsWithClusterId(tx *dbs.Tx, role string, clusterId int64, hourFrom string, hourTo string) (result []*NodeTrafficHourlyStat, err error) {
 	// TODO 节点如果已经被删除，则忽略
 	_, err = this.Query(tx).
+		Attr("role", role).
 		Attr("clusterId", clusterId).
 		Between("hour", hourFrom, hourTo).
 		Result("nodeId, SUM(bytes) AS bytes, SUM(cachedBytes) AS cachedBytes, SUM(countRequests) AS countRequests, SUM(countCachedRequests) AS countCachedRequests").
@@ -124,6 +125,38 @@ func (this *NodeTrafficHourlyStatDAO) FindTopNodeStatsWithClusterId(tx *dbs.Tx,
 	return
 }

+// FindHourlyStatsWithNodeId 获取节点小时之间统计
+func (this *NodeTrafficHourlyStatDAO) FindHourlyStatsWithNodeId(tx *dbs.Tx, role string, nodeId int64, hourFrom string, hourTo string) (result []*NodeTrafficHourlyStat, err error) {
+	ones, err := this.Query(tx).
+		Attr("role", role).
+		Attr("nodeId", nodeId).
+		Between("hour", hourFrom, hourTo).
+		Result("hour, SUM(bytes) AS bytes, SUM(cachedBytes) AS cachedBytes, SUM(countRequests) AS countRequests, SUM(countCachedRequests) AS countCachedRequests").
+		Group("hour").
+		FindAll()
+	if err != nil {
+		return nil, err
+	}
+	hourMap := map[string]*NodeTrafficHourlyStat{} // hour => Stat
+	for _, one := range ones {
+		stat := one.(*NodeTrafficHourlyStat)
+		hourMap[stat.Hour] = stat
+	}
+	hours, err := utils.RangeHours(hourFrom, hourTo)
+	if err != nil {
+		return nil, err
+	}
+	for _, hour := range hours {
+		stat, ok := hourMap[hour]
+		if ok {
+			result = append(result, stat)
+		} else {
+			result = append(result, &NodeTrafficHourlyStat{Hour: hour})
+		}
+	}
+	return result, nil
+}
+
 // Clean 清理历史数据
 func (this *NodeTrafficHourlyStatDAO) Clean(tx *dbs.Tx, days int) error {
 	var hour = timeutil.Format("Ymd00", time.Now().AddDate(0, 0, -days))
--- a/internal/db/models/stats/server_domain_hourly_stat_dao.go
+++ b/internal/db/models/stats/server_domain_hourly_stat_dao.go
@@ -79,7 +79,7 @@ func (this *ServerDomainHourlyStatDAO) IncreaseHourlyStat(tx *dbs.Tx, clusterId
 	return nil
 }

-// FindTopDomainStatsWithClusterId 取得一定时间内的节点排行数据
+// FindTopDomainStatsWithClusterId 取得集群上的一定时间内的域名排行数据
 func (this *ServerDomainHourlyStatDAO) FindTopDomainStatsWithClusterId(tx *dbs.Tx, clusterId int64, hourFrom string, hourTo string) (result []*ServerDomainHourlyStat, err error) {
 	// TODO 节点如果已经被删除，则忽略
 	_, err = this.Query(tx).
@@ -93,6 +93,20 @@ func (this *ServerDomainHourlyStatDAO) FindTopDomainStatsWithClusterId(tx *dbs.T
 	return
 }

+// FindTopDomainStatsWithNodeId 取得节点上的一定时间内的域名排行数据
+func (this *ServerDomainHourlyStatDAO) FindTopDomainStatsWithNodeId(tx *dbs.Tx, nodeId int64, hourFrom string, hourTo string) (result []*ServerDomainHourlyStat, err error) {
+	// TODO 节点如果已经被删除，则忽略
+	_, err = this.Query(tx).
+		Attr("nodeId", nodeId).
+		Between("hour", hourFrom, hourTo).
+		Result("domain, MIN(serverId) AS serverId, SUM(bytes) AS bytes, SUM(cachedBytes) AS cachedBytes, SUM(countRequests) AS countRequests, SUM(countCachedRequests) AS countCachedRequests").
+		Group("domain").
+		Desc("countRequests").
+		Slice(&result).
+		FindAll()
+	return
+}
+
 // Clean 清理历史数据
 func (this *ServerDomainHourlyStatDAO) Clean(tx *dbs.Tx, days int) error {
 	var hour = timeutil.Format("Ymd00", time.Now().AddDate(0, 0, -days))