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

IP阈值增加节点分组和集群相关统计项目

Browse Source
This commit is contained in:
GoEdgeLab
2021-09-14 11:36:22 +08:00
parent 187bb3bc0c
commit c2831398c8
9 changed files with 83 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
internal/db/models/message_dao.go
View File

@@ -118,21 +118,23 @@ func (this *MessageDAO) CreateClusterMessage(tx *dbs.Tx, role string, clusterId
} }
// CreateNodeMessage 创建节点消息 // CreateNodeMessage 创建节点消息
func (this *MessageDAO) CreateNodeMessage(tx *dbs.Tx, role string, clusterId int64, nodeId int64, messageType MessageType, level string, subject string, body string, paramsJSON []byte) error { func (this *MessageDAO) CreateNodeMessage(tx *dbs.Tx, role string, clusterId int64, nodeId int64, messageType MessageType, level string, subject string, body string, paramsJSON []byte, force bool) error {
// 检查N分钟内是否已经发送过 // 检查N分钟内是否已经发送过
hash := this.calHash(role, clusterId, nodeId, subject, body, paramsJSON) hash := this.calHash(role, clusterId, nodeId, subject, body, paramsJSON)
exists, err := this.Query(tx). if !force {
Attr("hash", hash). exists, err := this.Query(tx).
Gt("createdAt", time.Now().Unix()-10*60). // 10分钟 Attr("hash", hash).
Exist() Gt("createdAt", time.Now().Unix()-10*60). // 10分钟
if err != nil { Exist()
return err if err != nil {
} return err
if exists { }
return nil if exists {
return nil
}
} }
_, err = this.createMessage(tx, role, clusterId, nodeId, messageType, level, subject, body, paramsJSON) _, err := this.createMessage(tx, role, clusterId, nodeId, messageType, level, subject, body, paramsJSON)
if err != nil { if err != nil {
return err return err
} }

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

@@ -113,7 +113,7 @@ func (this *NodeDAO) FindEnabledBasicNode(tx *dbs.Tx, nodeId int64) (*Node, erro
one, err := this.Query(tx). one, err := this.Query(tx).
State(NodeStateEnabled). State(NodeStateEnabled).
Pk(nodeId). Pk(nodeId).
Result("id", "name", "clusterId", "isOn", "isUp"). Result("id", "name", "clusterId", "groupId", "isOn", "isUp").
Find() Find()
if one == nil { if one == nil {
return nil, err return nil, err

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

@@ -221,7 +221,7 @@ func (this *NodeThresholdDAO) FireNodeThreshold(tx *dbs.Tx, role string, nodeId
if len(threshold.Param) == 0 || threshold.Duration <= 0 { if len(threshold.Param) == 0 || threshold.Duration <= 0 {
continue continue
} }
paramValue, err := SharedNodeValueDAO.SumValues(tx, role, nodeId, item, threshold.Param, threshold.SumMethod, types.Int32(threshold.Duration), threshold.DurationUnit) paramValue, err := SharedNodeValueDAO.SumNodeValues(tx, role, nodeId, item, threshold.Param, threshold.SumMethod, types.Int32(threshold.Duration), threshold.DurationUnit)
if err != nil { if err != nil {
return err return err
} }
@@ -252,7 +252,7 @@ func (this *NodeThresholdDAO) FireNodeThreshold(tx *dbs.Tx, role string, nodeId
body = strings.Replace(body, "${item.name}", itemName, -1) body = strings.Replace(body, "${item.name}", itemName, -1)
body = strings.Replace(body, "${value}", fmt.Sprintf("%.2f", paramValue), -1) body = strings.Replace(body, "${value}", fmt.Sprintf("%.2f", paramValue), -1)
} }
err = SharedMessageDAO.CreateNodeMessage(tx, role, clusterId, nodeId, MessageTypeThresholdSatisfied, MessageLevelWarning, subject, body, maps.Map{}.AsJSON()) err = SharedMessageDAO.CreateNodeMessage(tx, role, clusterId, nodeId, MessageTypeThresholdSatisfied, MessageLevelWarning, subject, body, maps.Map{}.AsJSON(), true)
if err != nil { if err != nil {
return err return err
} }

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

@@ -173,8 +173,8 @@ func (this *NodeValueDAO) ListValuesForNSNodes(tx *dbs.Tx, item string, key stri
return return
} }
// SumValues 计算某项参数值 // SumNodeValues 计算节点的某项参数值
func (this *NodeValueDAO) SumValues(tx *dbs.Tx, role string, nodeId int64, item string, param string, method nodeconfigs.NodeValueSumMethod, duration int32, durationUnit nodeconfigs.NodeValueDurationUnit) (float64, error) { 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 { if duration <= 0 {
return 0, nil return 0, nil
} }
@@ -202,6 +202,65 @@ func (this *NodeValueDAO) SumValues(tx *dbs.Tx, role string, nodeId int64, item
return query.FindFloat64Col(0) return query.FindFloat64Col(0)
} }
// SumNodeGroupValues 计算节点分组的某项参数值
func (this *NodeValueDAO) SumNodeGroupValues(tx *dbs.Tx, role string, groupId int64, item string, param string, method nodeconfigs.NodeValueSumMethod, duration int32, durationUnit nodeconfigs.NodeValueDurationUnit) (float64, error) {
if duration <= 0 {
return 0, nil
}
query := this.Query(tx).
Attr("role", role).
Where("nodeId IN (SELECT id FROM "+SharedNodeDAO.Table+" WHERE groupId=:groupId AND state=1)").
Param("groupId", groupId).
Attr("item", item)
switch method {
case nodeconfigs.NodeValueSumMethodAvg:
query.Result("AVG(JSON_EXTRACT(value, '$." + param + "'))")
case nodeconfigs.NodeValueSumMethodSum:
query.Result("SUM(JSON_EXTRACT(value, '$." + param + "'))")
default:
query.Result("AVG(JSON_EXTRACT(value, '$." + param + "'))")
}
switch durationUnit {
case nodeconfigs.NodeValueDurationUnitMinute:
fromMinute := timeutil.FormatTime("YmdHi", time.Now().Unix()-int64(duration*60))
query.Gte("minute", fromMinute)
default:
fromMinute := timeutil.FormatTime("YmdHi", time.Now().Unix()-int64(duration*60))
query.Gte("minute", fromMinute)
}
return query.FindFloat64Col(0)
}
// SumNodeClusterValues 计算节点集群的某项参数值
func (this *NodeValueDAO) SumNodeClusterValues(tx *dbs.Tx, role string, clusterId int64, item string, param string, method nodeconfigs.NodeValueSumMethod, duration int32, durationUnit nodeconfigs.NodeValueDurationUnit) (float64, error) {
if duration <= 0 {
return 0, nil
}
query := this.Query(tx).
Attr("role", role).
Attr("clusterId", clusterId).
Attr("item", item)
switch method {
case nodeconfigs.NodeValueSumMethodAvg:
query.Result("AVG(JSON_EXTRACT(value, '$." + param + "'))")
case nodeconfigs.NodeValueSumMethodSum:
query.Result("SUM(JSON_EXTRACT(value, '$." + param + "'))")
default:
query.Result("AVG(JSON_EXTRACT(value, '$." + param + "'))")
}
switch durationUnit {
case nodeconfigs.NodeValueDurationUnitMinute:
fromMinute := timeutil.FormatTime("YmdHi", time.Now().Unix()-int64(duration*60))
query.Gte("minute", fromMinute)
default:
fromMinute := timeutil.FormatTime("YmdHi", time.Now().Unix()-int64(duration*60))
query.Gte("minute", fromMinute)
}
return query.FindFloat64Col(0)
}
// FindLatestNodeValue 获取最近一条数据 // FindLatestNodeValue 获取最近一条数据
func (this *NodeValueDAO) FindLatestNodeValue(tx *dbs.Tx, role string, nodeId int64, item string) (*NodeValue, error) { func (this *NodeValueDAO) FindLatestNodeValue(tx *dbs.Tx, role string, nodeId int64, item string) (*NodeValue, error) {
one, err := this.Query(tx). one, err := this.Query(tx).

2
internal/rpc/services/nameservers/service_ns_node_stream.go
View File

@@ -156,7 +156,7 @@ func (this *NSNodeService) NsNodeStream(server pb.NSNodeService_NsNodeStreamServ
} }
subject := "DNS节点\"" + nodeName + "\"已经恢复在线" subject := "DNS节点\"" + nodeName + "\"已经恢复在线"
msg := "DNS节点\"" + nodeName + "\"已经恢复在线" msg := "DNS节点\"" + nodeName + "\"已经恢复在线"
err = models.SharedMessageDAO.CreateNodeMessage(tx, nodeconfigs.NodeRoleDNS, clusterId, nodeId, models.MessageTypeNSNodeActive, models.MessageLevelSuccess, subject, msg, nil) err = models.SharedMessageDAO.CreateNodeMessage(tx, nodeconfigs.NodeRoleDNS, clusterId, nodeId, models.MessageTypeNSNodeActive, models.MessageLevelSuccess, subject, msg, nil, false)
if err != nil { if err != nil {
return err return err
} }

2
internal/rpc/services/service_node_stream.go
View File

@@ -120,7 +120,7 @@ func (this *NodeService) NodeStream(server pb.NodeService_NodeStreamServer) erro
} }
subject := "节点\"" + nodeName + "\"已经恢复在线" subject := "节点\"" + nodeName + "\"已经恢复在线"
msg := "节点\"" + nodeName + "\"已经恢复在线" msg := "节点\"" + nodeName + "\"已经恢复在线"
err = models.SharedMessageDAO.CreateNodeMessage(tx, nodeconfigs.NodeRoleNode, clusterId, nodeId, models.MessageTypeNodeActive, models.MessageLevelSuccess, subject, msg, nil) err = models.SharedMessageDAO.CreateNodeMessage(tx, nodeconfigs.NodeRoleNode, clusterId, nodeId, models.MessageTypeNodeActive, models.MessageLevelSuccess, subject, msg, nil, false)
if err != nil { if err != nil {
return err return err
} }

4
internal/tasks/health_check_executor.go
View File

@@ -140,10 +140,10 @@ func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
// 通知恢复或下线 // 通知恢复或下线
if result.IsOk { if result.IsOk {
message := "健康检查成功，节点\"" + result.Node.Name + "\"已恢复上线" message := "健康检查成功，节点\"" + result.Node.Name + "\"已恢复上线"
err = models.NewMessageDAO().CreateNodeMessage(nil, nodeconfigs.NodeRoleNode, this.clusterId, int64(result.Node.Id), models.MessageTypeHealthCheckNodeUp, models.MessageLevelSuccess, message, message, nil) err = models.NewMessageDAO().CreateNodeMessage(nil, nodeconfigs.NodeRoleNode, this.clusterId, int64(result.Node.Id), models.MessageTypeHealthCheckNodeUp, models.MessageLevelSuccess, message, message, nil, false)
} else { } else {
message := "健康检查失败，节点\"" + result.Node.Name + "\"已自动下线" message := "健康检查失败，节点\"" + result.Node.Name + "\"已自动下线"
err = models.NewMessageDAO().CreateNodeMessage(nil, nodeconfigs.NodeRoleNode, this.clusterId, int64(result.Node.Id), models.MessageTypeHealthCheckNodeDown, models.MessageLevelError, message, message, nil) err = models.NewMessageDAO().CreateNodeMessage(nil, nodeconfigs.NodeRoleNode, this.clusterId, int64(result.Node.Id), models.MessageTypeHealthCheckNodeDown, models.MessageLevelError, message, message, nil, false)
} }
} }
} }

2
internal/tasks/node_monitor_task.go
View File

@@ -83,7 +83,7 @@ func (this *NodeMonitorTask) monitorCluster(cluster *models.NodeCluster) error {
for _, node := range inactiveNodes { for _, node := range inactiveNodes {
subject := "节点\"" + node.Name + "\"已处于离线状态" subject := "节点\"" + node.Name + "\"已处于离线状态"
msg := "节点\"" + node.Name + "\"已处于离线状态" msg := "节点\"" + node.Name + "\"已处于离线状态"
err = models.SharedMessageDAO.CreateNodeMessage(nil, nodeconfigs.NodeRoleNode, clusterId, int64(node.Id), models.MessageTypeNodeInactive, models.LevelError, subject, msg, nil) err = models.SharedMessageDAO.CreateNodeMessage(nil, nodeconfigs.NodeRoleNode, clusterId, int64(node.Id), models.MessageTypeNodeInactive, models.LevelError, subject, msg, nil, false)
if err != nil { if err != nil {
return err return err
} }

2
internal/tasks/ns_node_monitor_task.go
View File

@@ -83,7 +83,7 @@ func (this *NSNodeMonitorTask) monitorCluster(cluster *models.NSCluster) error {
for _, node := range inactiveNodes { for _, node := range inactiveNodes {
subject := "DNS节点\"" + node.Name + "\"已处于离线状态" subject := "DNS节点\"" + node.Name + "\"已处于离线状态"
msg := "DNS节点\"" + node.Name + "\"已处于离线状态" msg := "DNS节点\"" + node.Name + "\"已处于离线状态"
err = models.SharedMessageDAO.CreateNodeMessage(nil, nodeconfigs.NodeRoleDNS, clusterId, int64(node.Id), models.MessageTypeNSNodeInactive, models.LevelError, subject, msg, nil) err = models.SharedMessageDAO.CreateNodeMessage(nil, nodeconfigs.NodeRoleDNS, clusterId, int64(node.Id), models.MessageTypeNSNodeInactive, models.LevelError, subject, msg, nil, false)
if err != nil { if err != nil {
return err return err
} }