mirror of
https://github.com/TeaOSLab/EdgeAPI.git
synced 2025-11-03 15:00:27 +08:00
502 lines
15 KiB
Go
502 lines
15 KiB
Go
package models
|
|
|
|
import (
|
|
"encoding/json"
|
|
"github.com/TeaOSLab/EdgeAPI/internal/goman"
|
|
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
|
|
_ "github.com/go-sql-driver/mysql"
|
|
"github.com/iwind/TeaGo/Tea"
|
|
"github.com/iwind/TeaGo/dbs"
|
|
"github.com/iwind/TeaGo/lists"
|
|
"github.com/iwind/TeaGo/logs"
|
|
"github.com/iwind/TeaGo/maps"
|
|
"github.com/iwind/TeaGo/rands"
|
|
timeutil "github.com/iwind/TeaGo/utils/time"
|
|
"strconv"
|
|
"time"
|
|
)
|
|
|
|
type MetricStatDAO dbs.DAO
|
|
|
|
func init() {
|
|
dbs.OnReadyDone(func() {
|
|
// 清理数据任务
|
|
var ticker = time.NewTicker(time.Duration(rands.Int(24, 48)) * time.Hour)
|
|
goman.New(func() {
|
|
for range ticker.C {
|
|
err := SharedMetricStatDAO.Clean(nil)
|
|
if err != nil {
|
|
logs.Println("SharedMetricStatDAO: clean expired data failed: " + err.Error())
|
|
}
|
|
}
|
|
})
|
|
})
|
|
}
|
|
|
|
func NewMetricStatDAO() *MetricStatDAO {
|
|
return dbs.NewDAO(&MetricStatDAO{
|
|
DAOObject: dbs.DAOObject{
|
|
DB: Tea.Env,
|
|
Table: "edgeMetricStats",
|
|
Model: new(MetricStat),
|
|
PkName: "id",
|
|
},
|
|
}).(*MetricStatDAO)
|
|
}
|
|
|
|
var SharedMetricStatDAO *MetricStatDAO
|
|
|
|
func init() {
|
|
dbs.OnReady(func() {
|
|
SharedMetricStatDAO = NewMetricStatDAO()
|
|
})
|
|
}
|
|
|
|
// CreateStat 创建统计数据
|
|
func (this *MetricStatDAO) CreateStat(tx *dbs.Tx, hash string, clusterId int64, nodeId int64, serverId int64, itemId int64, keys []string, value float64, time string, version int32) error {
|
|
hash += "@" + strconv.FormatInt(nodeId, 10)
|
|
var keysString string
|
|
if len(keys) > 0 {
|
|
keysJSON, err := json.Marshal(keys)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
keysString = string(keysJSON)
|
|
} else {
|
|
keysString = "[]"
|
|
}
|
|
return this.Query(tx).
|
|
Param("value", value).
|
|
InsertOrUpdateQuickly(maps.Map{
|
|
"hash": hash,
|
|
"clusterId": clusterId,
|
|
"nodeId": nodeId,
|
|
"serverId": serverId,
|
|
"itemId": itemId,
|
|
"value": value,
|
|
"time": time,
|
|
"version": version,
|
|
"keys": keysString,
|
|
"createdDay": timeutil.Format("Ymd"),
|
|
}, maps.Map{
|
|
"value": value,
|
|
})
|
|
}
|
|
|
|
// DeleteOldVersionItemStats 删除以前版本的统计数据
|
|
func (this *MetricStatDAO) DeleteOldVersionItemStats(tx *dbs.Tx, itemId int64, version int32) error {
|
|
_, err := this.Query(tx).
|
|
Attr("itemId", itemId).
|
|
Where("version<:version").
|
|
Param("version", version).
|
|
Delete()
|
|
return err
|
|
}
|
|
|
|
// DeleteItemStats 删除某个指标相关的统计数据
|
|
func (this *MetricStatDAO) DeleteItemStats(tx *dbs.Tx, itemId int64) error {
|
|
_, err := this.Query(tx).
|
|
Attr("itemId", itemId).
|
|
Delete()
|
|
return err
|
|
}
|
|
|
|
// DeleteNodeItemStats 删除某个节点的统计数据
|
|
func (this *MetricStatDAO) DeleteNodeItemStats(tx *dbs.Tx, nodeId int64, serverId int64, itemId int64, time string) error {
|
|
_, err := this.Query(tx).
|
|
Attr("nodeId", nodeId).
|
|
Attr("serverId", serverId).
|
|
Attr("itemId", itemId).
|
|
Attr("time", time).
|
|
Delete()
|
|
return err
|
|
}
|
|
|
|
// CountItemStats 计算统计数据数量
|
|
func (this *MetricStatDAO) CountItemStats(tx *dbs.Tx, itemId int64, version int32) (int64, error) {
|
|
return this.Query(tx).
|
|
Attr("itemId", itemId).
|
|
Attr("version", version).
|
|
Count()
|
|
}
|
|
|
|
// ListItemStats 列出单页统计数据
|
|
func (this *MetricStatDAO) ListItemStats(tx *dbs.Tx, itemId int64, version int32, offset int64, size int64) (result []*MetricStat, err error) {
|
|
_, err = this.Query(tx).
|
|
Attr("itemId", itemId).
|
|
Attr("version", version).
|
|
Offset(offset).
|
|
Limit(size).
|
|
Desc("time").
|
|
Desc("serverId").
|
|
Desc("value").
|
|
Slice(&result).
|
|
FindAll()
|
|
return
|
|
}
|
|
|
|
// FindItemStatsAtLastTime 取得所有集群最近一次计时前 N 个数据
|
|
// 适合每条数据中包含不同的Key的场景
|
|
func (this *MetricStatDAO) FindItemStatsAtLastTime(tx *dbs.Tx, itemId int64, ignoreEmptyKeys bool, ignoreKeys []string, 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
|
|
var query = this.Query(tx).
|
|
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)
|
|
if ignoreEmptyKeys {
|
|
query.Where("NOT JSON_CONTAINS(`keys`, '\"\"')")
|
|
}
|
|
if len(ignoreKeys) > 0 {
|
|
ignoreKeysJSON, err := json.Marshal(ignoreKeys)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query.Where("NOT JSON_CONTAINS(:ignoredKeys, JSON_EXTRACT(`keys`, '$[0]'))") // TODO $[0] 需要换成keys中的primary key位置
|
|
query.Param("ignoredKeys", string(ignoreKeysJSON))
|
|
}
|
|
_, err = query.
|
|
FindAll()
|
|
return
|
|
}
|
|
|
|
// FindItemStatsWithClusterIdAndLastTime 取得集群最近一次计时前 N 个数据
|
|
// 适合每条数据中包含不同的Key的场景
|
|
func (this *MetricStatDAO) FindItemStatsWithClusterIdAndLastTime(tx *dbs.Tx, clusterId int64, itemId int64, ignoreEmptyKeys bool, ignoreKeys []string, 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
|
|
|
|
var query = this.Query(tx).
|
|
UseIndex("cluster_item_time").
|
|
Attr("clusterId", clusterId).
|
|
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)
|
|
if ignoreEmptyKeys {
|
|
query.Where("NOT JSON_CONTAINS(`keys`, '\"\"')")
|
|
}
|
|
if len(ignoreKeys) > 0 {
|
|
ignoreKeysJSON, err := json.Marshal(ignoreKeys)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query.Where("NOT JSON_CONTAINS(:ignoredKeys, JSON_EXTRACT(`keys`, '$[0]'))") // TODO $[0] 需要换成keys中的primary key位置
|
|
query.Param("ignoredKeys", string(ignoreKeysJSON))
|
|
}
|
|
|
|
_, err = query.
|
|
FindAll()
|
|
return
|
|
}
|
|
|
|
// FindItemStatsWithNodeIdAndLastTime 取得节点最近一次计时前 N 个数据
|
|
// 适合每条数据中包含不同的Key的场景
|
|
func (this *MetricStatDAO) FindItemStatsWithNodeIdAndLastTime(tx *dbs.Tx, nodeId int64, itemId int64, ignoreEmptyKeys bool, ignoreKeys []string, 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
|
|
var query = this.Query(tx).
|
|
UseIndex("node_item_time").
|
|
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)
|
|
if ignoreEmptyKeys {
|
|
query.Where("NOT JSON_CONTAINS(`keys`, '\"\"')")
|
|
}
|
|
if len(ignoreKeys) > 0 {
|
|
ignoreKeysJSON, err := json.Marshal(ignoreKeys)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query.Where("NOT JSON_CONTAINS(:ignoredKeys, JSON_EXTRACT(`keys`, '$[0]'))") // TODO $[0] 需要换成keys中的primary key位置
|
|
query.Param("ignoredKeys", string(ignoreKeysJSON))
|
|
}
|
|
|
|
_, err = query.
|
|
FindAll()
|
|
return
|
|
}
|
|
|
|
// FindItemStatsWithServerIdAndLastTime 取得节点最近一次计时前 N 个数据
|
|
// 适合每条数据中包含不同的Key的场景
|
|
func (this *MetricStatDAO) FindItemStatsWithServerIdAndLastTime(tx *dbs.Tx, serverId int64, itemId int64, ignoreEmptyKeys bool, ignoreKeys []string, 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
|
|
|
|
var query = this.Query(tx).
|
|
UseIndex("server_item_time").
|
|
Attr("serverId", serverId).
|
|
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)
|
|
if ignoreEmptyKeys {
|
|
query.Where("NOT JSON_CONTAINS(`keys`, '\"\"')")
|
|
}
|
|
if len(ignoreKeys) > 0 {
|
|
ignoreKeysJSON, err := json.Marshal(ignoreKeys)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query.Where("NOT JSON_CONTAINS(:ignoredKeys, JSON_EXTRACT(`keys`, '$[0]'))") // TODO $[0] 需要换成keys中的primary key位置
|
|
query.Param("ignoredKeys", string(ignoreKeysJSON))
|
|
}
|
|
|
|
_, err = query.
|
|
FindAll()
|
|
return
|
|
}
|
|
|
|
// FindLatestItemStats 取得所有集群上最近 N 个时间的数据
|
|
// 适合同个Key在不同时间段的变化场景
|
|
func (this *MetricStatDAO) FindLatestItemStats(tx *dbs.Tx, itemId int64, ignoreEmptyKeys bool, ignoreKeys []string, version int32, size int64) (result []*MetricStat, err error) {
|
|
var query = this.Query(tx).
|
|
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)
|
|
if ignoreEmptyKeys {
|
|
query.Where("NOT JSON_CONTAINS(`keys`, '\"\"')")
|
|
}
|
|
if len(ignoreKeys) > 0 {
|
|
ignoreKeysJSON, err := json.Marshal(ignoreKeys)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query.Where("NOT JSON_CONTAINS(:ignoredKeys, JSON_EXTRACT(`keys`, '$[0]'))") // TODO $[0] 需要换成keys中的primary key位置
|
|
query.Param("ignoredKeys", string(ignoreKeysJSON))
|
|
}
|
|
|
|
_, err = query.
|
|
FindAll()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
lists.Reverse(result)
|
|
return
|
|
}
|
|
|
|
// FindLatestItemStatsWithClusterId 取得集群最近 N 个时间的数据
|
|
// 适合同个Key在不同时间段的变化场景
|
|
func (this *MetricStatDAO) FindLatestItemStatsWithClusterId(tx *dbs.Tx, clusterId int64, itemId int64, ignoreEmptyKeys bool, ignoreKeys []string, version int32, size int64) (result []*MetricStat, err error) {
|
|
var query = this.Query(tx).
|
|
Attr("clusterId", clusterId).
|
|
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)
|
|
if ignoreEmptyKeys {
|
|
query.Where("NOT JSON_CONTAINS(`keys`, '\"\"')")
|
|
}
|
|
if len(ignoreKeys) > 0 {
|
|
ignoreKeysJSON, err := json.Marshal(ignoreKeys)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query.Where("NOT JSON_CONTAINS(:ignoredKeys, JSON_EXTRACT(`keys`, '$[0]'))") // TODO $[0] 需要换成keys中的primary key位置
|
|
query.Param("ignoredKeys", string(ignoreKeysJSON))
|
|
}
|
|
|
|
_, err = query.
|
|
FindAll()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
lists.Reverse(result)
|
|
return
|
|
}
|
|
|
|
// FindLatestItemStatsWithNodeId 取得节点最近 N 个时间的数据
|
|
// 适合同个Key在不同时间段的变化场景
|
|
func (this *MetricStatDAO) FindLatestItemStatsWithNodeId(tx *dbs.Tx, nodeId int64, itemId int64, ignoreEmptyKeys bool, ignoreKeys []string, version int32, size int64) (result []*MetricStat, err error) {
|
|
var query = 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)
|
|
if ignoreEmptyKeys {
|
|
query.Where("NOT JSON_CONTAINS(`keys`, '\"\"')")
|
|
}
|
|
if len(ignoreKeys) > 0 {
|
|
ignoreKeysJSON, err := json.Marshal(ignoreKeys)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query.Where("NOT JSON_CONTAINS(:ignoredKeys, JSON_EXTRACT(`keys`, '$[0]'))") // TODO $[0] 需要换成keys中的primary key位置
|
|
query.Param("ignoredKeys", string(ignoreKeysJSON))
|
|
}
|
|
|
|
_, err = query.
|
|
FindAll()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
lists.Reverse(result)
|
|
return
|
|
}
|
|
|
|
// FindLatestItemStatsWithServerId 取得服务最近 N 个时间的数据
|
|
// 适合同个Key在不同时间段的变化场景
|
|
func (this *MetricStatDAO) FindLatestItemStatsWithServerId(tx *dbs.Tx, serverId int64, itemId int64, ignoreEmptyKeys bool, ignoreKeys []string, version int32, size int64) (result []*MetricStat, err error) {
|
|
var query = this.Query(tx).
|
|
Attr("serverId", serverId).
|
|
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)
|
|
if ignoreEmptyKeys {
|
|
query.Where("NOT JSON_CONTAINS(`keys`, '\"\"')")
|
|
}
|
|
if len(ignoreKeys) > 0 {
|
|
ignoreKeysJSON, err := json.Marshal(ignoreKeys)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
query.Where("NOT JSON_CONTAINS(:ignoredKeys, JSON_EXTRACT(`keys`, '$[0]'))") // TODO $[0] 需要换成keys中的primary key位置
|
|
query.Param("ignoredKeys", string(ignoreKeysJSON))
|
|
}
|
|
|
|
_, err = query.
|
|
FindAll()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
lists.Reverse(result)
|
|
return
|
|
}
|
|
|
|
// Clean 清理数据
|
|
func (this *MetricStatDAO) Clean(tx *dbs.Tx) error {
|
|
for _, category := range serverconfigs.FindAllMetricItemCategoryCodes() {
|
|
var offset int64 = 0
|
|
var size int64 = 100
|
|
for {
|
|
items, err := SharedMetricItemDAO.ListEnabledItems(tx, category, offset, size)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, item := range items {
|
|
var config = &serverconfigs.MetricItemConfig{
|
|
Id: int64(item.Id),
|
|
Period: int(item.Period),
|
|
PeriodUnit: item.PeriodUnit,
|
|
}
|
|
var expiresDay = config.ServerExpiresDay()
|
|
_, err := this.Query(tx).
|
|
Attr("itemId", item.Id).
|
|
Lte("createdDay", expiresDay).
|
|
UseIndex("createdDay").
|
|
Limit(100_000). // 一次性不要删除太多，防止阻塞其他操作
|
|
Delete()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
if len(items) == 0 {
|
|
break
|
|
}
|
|
|
|
offset += size
|
|
}
|
|
}
|
|
return nil
|
|
}
|