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

优化消息发送相关代码/删除监控相关代码

Browse Source
This commit is contained in:
GoEdgeLab
2023-10-12 20:11:21 +08:00
parent 6b758383b9
commit 444db8ca4c
18 changed files with 178 additions and 740 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
internal/const/const.go
View File

@@ -7,6 +7,8 @@ const (
ProcessName = "edge-api" ProcessName = "edge-api"
ProductNameZH = "Edge" ProductNameZH = "Edge"
GlobalProductName = "GoEdge"
Role = "api" Role = "api"
EncryptKey = "8f983f4d69b83aaa0d74b21a212f6967" EncryptKey = "8f983f4d69b83aaa0d74b21a212f6967"

26
internal/db/models/message_media_dao.go
View File

@@ -34,7 +34,7 @@ func init() {
}) })
} }
// 启用条目 // EnableMessageMedia 启用条目
func (this *MessageMediaDAO) EnableMessageMedia(tx *dbs.Tx, id int64) error { func (this *MessageMediaDAO) EnableMessageMedia(tx *dbs.Tx, id int64) error {
_, err := this.Query(tx). _, err := this.Query(tx).
Pk(id). Pk(id).
@@ -43,7 +43,7 @@ func (this *MessageMediaDAO) EnableMessageMedia(tx *dbs.Tx, id int64) error {
return err return err
} }
// 禁用条目 // DisableMessageMedia 禁用条目
func (this *MessageMediaDAO) DisableMessageMedia(tx *dbs.Tx, id int64) error { func (this *MessageMediaDAO) DisableMessageMedia(tx *dbs.Tx, id int64) error {
_, err := this.Query(tx). _, err := this.Query(tx).
Pk(id). Pk(id).
@@ -52,7 +52,7 @@ func (this *MessageMediaDAO) DisableMessageMedia(tx *dbs.Tx, id int64) error {
return err return err
} }
// 查找启用中的条目 // FindEnabledMessageMedia 查找启用中的条目
func (this *MessageMediaDAO) FindEnabledMessageMedia(tx *dbs.Tx, id int64) (*MessageMedia, error) { func (this *MessageMediaDAO) FindEnabledMessageMedia(tx *dbs.Tx, id int64) (*MessageMedia, error) {
result, err := this.Query(tx). result, err := this.Query(tx).
Pk(id). Pk(id).
@@ -83,19 +83,19 @@ func (this *MessageMediaDAO) FindAllEnabledMessageMedias(tx *dbs.Tx) (result []*
return return
} }
// 设置当前所有可用的媒介 // UpdateMessageMedias 设置当前所有可用的媒介
func (this *MessageMediaDAO) UpdateMessageMedias(tx *dbs.Tx, mediaMaps []maps.Map) error { func (this *MessageMediaDAO) UpdateMessageMedias(tx *dbs.Tx, mediaMaps []maps.Map) error {
// 新的媒介信息 // 新的媒介信息
mediaTypes := []string{} var mediaTypes = []string{}
for index, m := range mediaMaps { for index, m := range mediaMaps {
order := len(mediaMaps) - index var order = len(mediaMaps) - index
mediaType := m.GetString("type") var mediaType = m.GetString("code")
mediaTypes = append(mediaTypes, mediaType) mediaTypes = append(mediaTypes, mediaType)
name := m.GetString("name") var name = m.GetString("name")
description := m.GetString("description") var description = m.GetString("description")
userDescription := m.GetString("userDescription") var userDescription = m.GetString("user")
isOn := m.GetBool("isOn") var isOn = m.GetBool("isOn")
mediaId, err := this.Query(tx). mediaId, err := this.Query(tx).
ResultPk(). ResultPk().
@@ -128,7 +128,7 @@ func (this *MessageMediaDAO) UpdateMessageMedias(tx *dbs.Tx, mediaMaps []maps.Ma
return err return err
} }
for _, one := range ones { for _, one := range ones {
mediaType := one.(*MessageMedia).Type var mediaType = one.(*MessageMedia).Type
if !lists.ContainsString(mediaTypes, mediaType) { if !lists.ContainsString(mediaTypes, mediaType) {
err := this.Query(tx). err := this.Query(tx).
Pk(one.(*MessageMedia).Id). Pk(one.(*MessageMedia).Id).
@@ -142,7 +142,7 @@ func (this *MessageMediaDAO) UpdateMessageMedias(tx *dbs.Tx, mediaMaps []maps.Ma
return nil return nil
} }
// 根据类型查找媒介 // FindEnabledMediaWithType 根据类型查找媒介
func (this *MessageMediaDAO) FindEnabledMediaWithType(tx *dbs.Tx, mediaType string) (*MessageMedia, error) { func (this *MessageMediaDAO) FindEnabledMediaWithType(tx *dbs.Tx, mediaType string) (*MessageMedia, error) {
one, err := this.Query(tx). one, err := this.Query(tx).
Attr("type", mediaType). Attr("type", mediaType).

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

@@ -210,7 +210,7 @@ func (this *MessageTaskDAO) CreateMessageTasks(tx *dbs.Tx, role nodeconfigs.Node
if err != nil { if err != nil {
return err return err
} }
allRecipientIds := []int64{} var allRecipientIds = []int64{}
for _, receiver := range receivers { for _, receiver := range receivers {
if receiver.RecipientId > 0 { if receiver.RecipientId > 0 {
allRecipientIds = append(allRecipientIds, int64(receiver.RecipientId)) allRecipientIds = append(allRecipientIds, int64(receiver.RecipientId))
@@ -223,7 +223,7 @@ func (this *MessageTaskDAO) CreateMessageTasks(tx *dbs.Tx, role nodeconfigs.Node
} }
} }
sentMap := map[int64]bool{} // recipientId => bool 用来检查是否已经发送，防止重复发送给某个接收人 var sentMap = map[int64]bool{} // recipientId => bool 用来检查是否已经发送，防止重复发送给某个接收人
for _, recipientId := range allRecipientIds { for _, recipientId := range allRecipientIds {
_, ok := sentMap[recipientId] _, ok := sentMap[recipientId]
if ok { if ok {

215
internal/db/models/monitor_node_dao.go
View File

@@ -1,215 +0,0 @@
package models
import (
"github.com/TeaOSLab/EdgeAPI/internal/errors"
"github.com/TeaOSLab/EdgeAPI/internal/utils"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
_ "github.com/go-sql-driver/mysql"
"github.com/iwind/TeaGo/Tea"
"github.com/iwind/TeaGo/dbs"
"github.com/iwind/TeaGo/rands"
"github.com/iwind/TeaGo/types"
)
const (
MonitorNodeStateEnabled = 1 // 已启用
MonitorNodeStateDisabled = 0 // 已禁用
)
type MonitorNodeDAO dbs.DAO
func NewMonitorNodeDAO() *MonitorNodeDAO {
return dbs.NewDAO(&MonitorNodeDAO{
DAOObject: dbs.DAOObject{
DB: Tea.Env,
Table: "edgeMonitorNodes",
Model: new(MonitorNode),
PkName: "id",
},
}).(*MonitorNodeDAO)
}
var SharedMonitorNodeDAO *MonitorNodeDAO
func init() {
dbs.OnReady(func() {
SharedMonitorNodeDAO = NewMonitorNodeDAO()
})
}
// EnableMonitorNode 启用条目
func (this *MonitorNodeDAO) EnableMonitorNode(tx *dbs.Tx, id int64) error {
_, err := this.Query(tx).
Pk(id).
Set("state", MonitorNodeStateEnabled).
Update()
return err
}
// DisableMonitorNode 禁用条目
func (this *MonitorNodeDAO) DisableMonitorNode(tx *dbs.Tx, nodeId int64) error {
_, err := this.Query(tx).
Pk(nodeId).
Set("state", MonitorNodeStateDisabled).
Update()
if err != nil {
return err
}
// 删除运行日志
return SharedNodeLogDAO.DeleteNodeLogs(tx, nodeconfigs.NodeRoleMonitor, nodeId)
}
// FindEnabledMonitorNode 查找启用中的条目
func (this *MonitorNodeDAO) FindEnabledMonitorNode(tx *dbs.Tx, id int64) (*MonitorNode, error) {
result, err := this.Query(tx).
Pk(id).
Attr("state", MonitorNodeStateEnabled).
Find()
if result == nil {
return nil, err
}
return result.(*MonitorNode), err
}
// FindMonitorNodeName 根据主键查找名称
func (this *MonitorNodeDAO) FindMonitorNodeName(tx *dbs.Tx, id int64) (string, error) {
return this.Query(tx).
Pk(id).
Result("name").
FindStringCol("")
}
// FindAllEnabledMonitorNodes 列出所有可用监控节点
func (this *MonitorNodeDAO) FindAllEnabledMonitorNodes(tx *dbs.Tx) (result []*MonitorNode, err error) {
_, err = this.Query(tx).
State(MonitorNodeStateEnabled).
Desc("order").
AscPk().
Slice(&result).
FindAll()
return
}
// CountAllEnabledMonitorNodes 计算监控节点数量
func (this *MonitorNodeDAO) CountAllEnabledMonitorNodes(tx *dbs.Tx) (int64, error) {
return this.Query(tx).
State(MonitorNodeStateEnabled).
Count()
}
// ListEnabledMonitorNodes 列出单页的监控节点
func (this *MonitorNodeDAO) ListEnabledMonitorNodes(tx *dbs.Tx, offset int64, size int64) (result []*MonitorNode, err error) {
_, err = this.Query(tx).
State(MonitorNodeStateEnabled).
Offset(offset).
Limit(size).
Desc("order").
DescPk().
Slice(&result).
FindAll()
return
}
// CreateMonitorNode 创建监控节点
func (this *MonitorNodeDAO) CreateMonitorNode(tx *dbs.Tx, name string, description string, isOn bool) (nodeId int64, err error) {
uniqueId, err := this.GenUniqueId(tx)
if err != nil {
return 0, err
}
secret := rands.String(32)
err = NewApiTokenDAO().CreateAPIToken(tx, uniqueId, secret, nodeconfigs.NodeRoleMonitor)
if err != nil {
return
}
var op = NewMonitorNodeOperator()
op.IsOn = isOn
op.UniqueId = uniqueId
op.Secret = secret
op.Name = name
op.Description = description
op.State = NodeStateEnabled
err = this.Save(tx, op)
if err != nil {
return
}
return types.Int64(op.Id), nil
}
// UpdateMonitorNode 修改监控节点
func (this *MonitorNodeDAO) UpdateMonitorNode(tx *dbs.Tx, nodeId int64, name string, description string, isOn bool) error {
if nodeId <= 0 {
return errors.New("invalid nodeId")
}
var op = NewMonitorNodeOperator()
op.Id = nodeId
op.Name = name
op.Description = description
op.IsOn = isOn
err := this.Save(tx, op)
return err
}
// FindEnabledMonitorNodeWithUniqueId 根据唯一ID获取节点信息
func (this *MonitorNodeDAO) FindEnabledMonitorNodeWithUniqueId(tx *dbs.Tx, uniqueId string) (*MonitorNode, error) {
result, err := this.Query(tx).
Attr("uniqueId", uniqueId).
Attr("state", MonitorNodeStateEnabled).
Find()
if result == nil {
return nil, err
}
return result.(*MonitorNode), err
}
// FindEnabledMonitorNodeIdWithUniqueId 根据唯一ID获取节点ID
func (this *MonitorNodeDAO) FindEnabledMonitorNodeIdWithUniqueId(tx *dbs.Tx, uniqueId string) (int64, error) {
return this.Query(tx).
Attr("uniqueId", uniqueId).
Attr("state", MonitorNodeStateEnabled).
ResultPk().
FindInt64Col(0)
}
// GenUniqueId 生成唯一ID
func (this *MonitorNodeDAO) GenUniqueId(tx *dbs.Tx) (string, error) {
for {
uniqueId := rands.HexString(32)
ok, err := this.Query(tx).
Attr("uniqueId", uniqueId).
Exist()
if err != nil {
return "", err
}
if ok {
continue
}
return uniqueId, nil
}
}
// UpdateNodeStatus 更改节点状态
func (this *MonitorNodeDAO) UpdateNodeStatus(tx *dbs.Tx, nodeId int64, statusJSON []byte) error {
if statusJSON == nil {
return nil
}
_, err := this.Query(tx).
Pk(nodeId).
Set("status", string(statusJSON)).
Update()
return err
}
// CountAllLowerVersionNodes 计算所有节点中低于某个版本的节点数量
func (this *MonitorNodeDAO) CountAllLowerVersionNodes(tx *dbs.Tx, version string) (int64, error) {
return this.Query(tx).
State(MonitorNodeStateEnabled).
Attr("isOn", true).
Where("status IS NOT NULL").
Where("(JSON_EXTRACT(status, '$.buildVersionCode') IS NULL OR JSON_EXTRACT(status, '$.buildVersionCode')<:version)").
Param("version", utils.VersionToLong(version)).
Count()
}

6
internal/db/models/monitor_node_dao_test.go
View File

@@ -1,6 +0,0 @@
package models
import (
_ "github.com/go-sql-driver/mysql"
_ "github.com/iwind/TeaGo/bootstrap"
)

38
internal/db/models/monitor_node_model.go
View File

@@ -1,38 +0,0 @@
package models
import "github.com/iwind/TeaGo/dbs"
// MonitorNode 监控节点
type MonitorNode struct {
Id uint32 `field:"id"` // ID
IsOn bool `field:"isOn"` // 是否启用
UniqueId string `field:"uniqueId"` // 唯一ID
Secret string `field:"secret"` // 密钥
Name string `field:"name"` // 名称
Description string `field:"description"` // 描述
Order uint32 `field:"order"` // 排序
State uint8 `field:"state"` // 状态
CreatedAt uint64 `field:"createdAt"` // 创建时间
AdminId uint32 `field:"adminId"` // 管理员ID
Weight uint32 `field:"weight"` // 权重
Status dbs.JSON `field:"status"` // 运行状态
}
type MonitorNodeOperator struct {
Id interface{} // ID
IsOn interface{} // 是否启用
UniqueId interface{} // 唯一ID
Secret interface{} // 密钥
Name interface{} // 名称
Description interface{} // 描述
Order interface{} // 排序
State interface{} // 状态
CreatedAt interface{} // 创建时间
AdminId interface{} // 管理员ID
Weight interface{} // 权重
Status interface{} // 运行状态
}
func NewMonitorNodeOperator() *MonitorNodeOperator {
return &MonitorNodeOperator{}
}

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

@@ -1 +0,0 @@
package models

5
internal/nodes/api_node_services.go
View File

@@ -454,11 +454,6 @@ func (this *APINode) registerServices(server *grpc.Server) {
pb.RegisterNodeClusterFirewallActionServiceServer(server, instance) pb.RegisterNodeClusterFirewallActionServiceServer(server, instance)
this.rest(instance) this.rest(instance)
} }
{
var instance = this.serviceInstance(&services.MonitorNodeService{}).(*services.MonitorNodeService)
pb.RegisterMonitorNodeServiceServer(server, instance)
this.rest(instance)
}
{ {
var instance = this.serviceInstance(&services.AuthorityNodeService{}).(*services.AuthorityNodeService) var instance = this.serviceInstance(&services.AuthorityNodeService{}).(*services.AuthorityNodeService)
pb.RegisterAuthorityNodeServiceServer(server, instance) pb.RegisterAuthorityNodeServiceServer(server, instance)

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

@@ -78,7 +78,7 @@ func (this *APINodeService) DeleteAPINode(ctx context.Context, req *pb.DeleteAPI
// FindAllEnabledAPINodes 列出所有可用API节点 // FindAllEnabledAPINodes 列出所有可用API节点
func (this *APINodeService) FindAllEnabledAPINodes(ctx context.Context, req *pb.FindAllEnabledAPINodesRequest) (*pb.FindAllEnabledAPINodesResponse, error) { func (this *APINodeService) FindAllEnabledAPINodes(ctx context.Context, req *pb.FindAllEnabledAPINodesRequest) (*pb.FindAllEnabledAPINodesResponse, error) {
_, _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin, rpcutils.UserTypeUser, rpcutils.UserTypeNode, rpcutils.UserTypeMonitor, rpcutils.UserTypeDNS, rpcutils.UserTypeAuthority) _, _, _, err := rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAdmin, rpcutils.UserTypeUser, rpcutils.UserTypeNode, rpcutils.UserTypeDNS, rpcutils.UserTypeAuthority)
if err != nil { if err != nil {
return nil, err return nil, err
} }

10
internal/rpc/services/service_base.go
View File

@@ -95,12 +95,6 @@ func (this *BaseService) ValidateUserNode(ctx context.Context, canRest bool) (us
return return
} }
// ValidateMonitorNode 校验监控节点
func (this *BaseService) ValidateMonitorNode(ctx context.Context) (nodeId int64, err error) {
_, _, nodeId, err = rpcutils.ValidateRequest(ctx, rpcutils.UserTypeMonitor)
return
}
// ValidateAuthorityNode 校验认证节点 // ValidateAuthorityNode 校验认证节点
func (this *BaseService) ValidateAuthorityNode(ctx context.Context) (nodeId int64, err error) { func (this *BaseService) ValidateAuthorityNode(ctx context.Context) (nodeId int64, err error) {
_, _, nodeId, err = rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAuthority) _, _, nodeId, err = rpcutils.ValidateRequest(ctx, rpcutils.UserTypeAuthority)
@@ -111,7 +105,7 @@ func (this *BaseService) ValidateAuthorityNode(ctx context.Context) (nodeId int6
func (this *BaseService) ValidateNodeId(ctx context.Context, roles ...rpcutils.UserType) (role rpcutils.UserType, nodeIntId int64, err error) { func (this *BaseService) ValidateNodeId(ctx context.Context, roles ...rpcutils.UserType) (role rpcutils.UserType, nodeIntId int64, err error) {
// 默认包含大部分节点 // 默认包含大部分节点
if len(roles) == 0 { if len(roles) == 0 {
roles = []rpcutils.UserType{rpcutils.UserTypeNode, rpcutils.UserTypeCluster, rpcutils.UserTypeAdmin, rpcutils.UserTypeUser, rpcutils.UserTypeDNS, rpcutils.UserTypeReport, rpcutils.UserTypeMonitor, rpcutils.UserTypeLog, rpcutils.UserTypeAPI} roles = []rpcutils.UserType{rpcutils.UserTypeNode, rpcutils.UserTypeCluster, rpcutils.UserTypeAdmin, rpcutils.UserTypeUser, rpcutils.UserTypeDNS, rpcutils.UserTypeReport, rpcutils.UserTypeLog, rpcutils.UserTypeAPI}
} }
if ctx == nil { if ctx == nil {
@@ -195,8 +189,6 @@ func (this *BaseService) ValidateNodeId(ctx context.Context, roles ...rpcutils.U
nodeIntId, err = models.SharedUserNodeDAO.FindEnabledUserNodeIdWithUniqueId(nil, nodeId) nodeIntId, err = models.SharedUserNodeDAO.FindEnabledUserNodeIdWithUniqueId(nil, nodeId)
case rpcutils.UserTypeAdmin: case rpcutils.UserTypeAdmin:
nodeIntId = 0 nodeIntId = 0
case rpcutils.UserTypeMonitor:
nodeIntId, err = models.SharedMonitorNodeDAO.FindEnabledMonitorNodeIdWithUniqueId(nil, nodeId)
case rpcutils.UserTypeDNS: case rpcutils.UserTypeDNS:
nodeIntId, err = models.SharedNSNodeDAO.FindEnabledNodeIdWithUniqueId(nil, nodeId) nodeIntId, err = models.SharedNSNodeDAO.FindEnabledNodeIdWithUniqueId(nil, nodeId)
case rpcutils.UserTypeReport: case rpcutils.UserTypeReport:

136
internal/rpc/services/service_message_task.go
View File

@@ -6,7 +6,6 @@ import (
"github.com/TeaOSLab/EdgeAPI/internal/db/models" "github.com/TeaOSLab/EdgeAPI/internal/db/models"
"github.com/TeaOSLab/EdgeAPI/internal/utils" "github.com/TeaOSLab/EdgeAPI/internal/utils"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb" "github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/iwind/TeaGo/maps"
"github.com/iwind/TeaGo/types" "github.com/iwind/TeaGo/types"
) )
@@ -23,146 +22,13 @@ func (this *MessageTaskService) CreateMessageTask(ctx context.Context, req *pb.C
} }
var tx = this.NullTx() var tx = this.NullTx()
taskId, err := models.SharedMessageTaskDAO.CreateMessageTask(tx, req.RecipientId, req.InstanceId, req.User, req.Subject, req.Body, req.IsPrimary) taskId, err := models.SharedMessageTaskDAO.CreateMessageTask(tx, req.MessageRecipientId, req.MessageMediaInstanceId, req.User, req.Subject, req.Body, req.IsPrimary)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return &pb.CreateMessageTaskResponse{MessageTaskId: taskId}, nil return &pb.CreateMessageTaskResponse{MessageTaskId: taskId}, nil
} }
// FindSendingMessageTasks 查找要发送的任务
func (this *MessageTaskService) FindSendingMessageTasks(ctx context.Context, req *pb.FindSendingMessageTasksRequest) (*pb.FindSendingMessageTasksResponse, error) {
_, err := this.ValidateMonitorNode(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
var cacheMap = utils.NewCacheMap()
tasks, err := models.SharedMessageTaskDAO.FindSendingMessageTasks(tx, req.Size)
if err != nil {
return nil, err
}
pbTasks := []*pb.MessageTask{}
for _, task := range tasks {
var pbRecipient *pb.MessageRecipient
if task.RecipientId > 0 {
recipient, err := models.SharedMessageRecipientDAO.FindEnabledMessageRecipient(tx, int64(task.RecipientId), cacheMap)
if err != nil {
return nil, err
}
if recipient == nil || !recipient.IsOn {
// 如果发送人已经删除或者禁用，则删除此消息
err = models.SharedMessageTaskDAO.DisableMessageTask(tx, int64(task.Id))
if err != nil {
return nil, err
}
continue
}
// 媒介
instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(recipient.InstanceId), cacheMap)
if err != nil {
return nil, err
}
if instance == nil || !instance.IsOn {
// 如果媒介实例已经删除或者禁用，则删除此消息
err = models.SharedMessageTaskDAO.DisableMessageTask(tx, int64(task.Id))
if err != nil {
return nil, err
}
continue
}
pbRecipient = &pb.MessageRecipient{
Id: int64(recipient.Id),
User: recipient.User,
MessageMediaInstance: &pb.MessageMediaInstance{
Id: int64(instance.Id),
MessageMedia: &pb.MessageMedia{
Type: instance.MediaType,
},
ParamsJSON: instance.Params,
RateJSON: instance.Rate,
},
}
} else { // 没有指定既定的接收人
// 媒介
instance, err := models.SharedMessageMediaInstanceDAO.FindEnabledMessageMediaInstance(tx, int64(task.InstanceId), cacheMap)
if err != nil {
return nil, err
}
if instance == nil || !instance.IsOn {
// 如果媒介实例已经删除或者禁用，则删除此消息
err = models.SharedMessageTaskDAO.DisableMessageTask(tx, int64(task.Id))
if err != nil {
return nil, err
}
continue
}
pbRecipient = &pb.MessageRecipient{
Id: 0,
MessageMediaInstance: &pb.MessageMediaInstance{
Id: int64(instance.Id),
MessageMedia: &pb.MessageMedia{
Type: instance.MediaType,
},
ParamsJSON: instance.Params,
RateJSON: instance.Rate,
},
}
}
pbTasks = append(pbTasks, &pb.MessageTask{
Id: int64(task.Id),
MessageRecipient: pbRecipient,
User: task.User,
Subject: task.Subject,
Body: task.Body,
CreatedAt: int64(task.CreatedAt),
Status: types.Int32(task.Status),
SentAt: int64(task.SentAt),
})
}
return &pb.FindSendingMessageTasksResponse{MessageTasks: pbTasks}, nil
}
// UpdateMessageTaskStatus 修改任务状态
func (this *MessageTaskService) UpdateMessageTaskStatus(ctx context.Context, req *pb.UpdateMessageTaskStatusRequest) (*pb.RPCSuccess, error) {
_, err := this.ValidateMonitorNode(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
resultJSON := []byte{}
if req.Result != nil {
resultJSON, err = json.Marshal(maps.Map{
"isOk": req.Result.IsOk,
"error": req.Result.Error,
"response": req.Result.Response,
})
if err != nil {
return nil, err
}
}
err = models.SharedMessageTaskDAO.UpdateMessageTaskStatus(tx, req.MessageTaskId, int(req.Status), resultJSON)
if err != nil {
return nil, err
}
// 创建发送记录
if (int(req.Status) == models.MessageTaskStatusSuccess || int(req.Status) == models.MessageTaskStatusFailed) && req.Result != nil {
err = models.SharedMessageTaskLogDAO.CreateLog(tx, req.MessageTaskId, req.Result.IsOk, req.Result.Error, req.Result.Response)
if err != nil {
return nil, err
}
}
return this.Success()
}
// DeleteMessageTask 删除消息任务 // DeleteMessageTask 删除消息任务
func (this *MessageTaskService) DeleteMessageTask(ctx context.Context, req *pb.DeleteMessageTaskRequest) (*pb.RPCSuccess, error) { func (this *MessageTaskService) DeleteMessageTask(ctx context.Context, req *pb.DeleteMessageTaskRequest) (*pb.RPCSuccess, error) {
_, err := this.ValidateAdmin(ctx) _, err := this.ValidateAdmin(ctx)

14
internal/rpc/services/service_message_task_ext.go Normal file
View File

@@ -0,0 +1,14 @@
// Copyright 2023 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
//go:build !plus
package services
import (
"context"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
)
// SendMessageTask 发送某个消息任务
func (this *MessageTaskService) SendMessageTask(ctx context.Context, req *pb.SendMessageTaskRequest) (*pb.SendMessageTaskResponse, error) {
return nil, this.NotImplementedYet()
}

233
internal/rpc/services/service_monitor_node.go
View File

@@ -1,233 +0,0 @@
package services
import (
"context"
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"github.com/TeaOSLab/EdgeAPI/internal/errors"
rpcutils "github.com/TeaOSLab/EdgeAPI/internal/rpc/utils"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"google.golang.org/grpc/metadata"
)
type MonitorNodeService struct {
BaseService
}
// CreateMonitorNode 创建监控节点
func (this *MonitorNodeService) CreateMonitorNode(ctx context.Context, req *pb.CreateMonitorNodeRequest) (*pb.CreateMonitorNodeResponse, error) {
_, err := this.ValidateAdmin(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
nodeId, err := models.SharedMonitorNodeDAO.CreateMonitorNode(tx, req.Name, req.Description, req.IsOn)
if err != nil {
return nil, err
}
return &pb.CreateMonitorNodeResponse{MonitorNodeId: nodeId}, nil
}
// UpdateMonitorNode 修改监控节点
func (this *MonitorNodeService) UpdateMonitorNode(ctx context.Context, req *pb.UpdateMonitorNodeRequest) (*pb.RPCSuccess, error) {
_, err := this.ValidateAdmin(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
err = models.SharedMonitorNodeDAO.UpdateMonitorNode(tx, req.MonitorNodeId, req.Name, req.Description, req.IsOn)
if err != nil {
return nil, err
}
return this.Success()
}
// DeleteMonitorNode 删除监控节点
func (this *MonitorNodeService) DeleteMonitorNode(ctx context.Context, req *pb.DeleteMonitorNodeRequest) (*pb.RPCSuccess, error) {
_, err := this.ValidateAdmin(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
err = models.SharedMonitorNodeDAO.DisableMonitorNode(tx, req.MonitorNodeId)
if err != nil {
return nil, err
}
return this.Success()
}
// FindAllEnabledMonitorNodes 列出所有可用监控节点
func (this *MonitorNodeService) FindAllEnabledMonitorNodes(ctx context.Context, req *pb.FindAllEnabledMonitorNodesRequest) (*pb.FindAllEnabledMonitorNodesResponse, error) {
_, err := this.ValidateAdmin(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
nodes, err := models.SharedMonitorNodeDAO.FindAllEnabledMonitorNodes(tx)
if err != nil {
return nil, err
}
result := []*pb.MonitorNode{}
for _, node := range nodes {
result = append(result, &pb.MonitorNode{
Id: int64(node.Id),
IsOn: node.IsOn,
UniqueId: node.UniqueId,
Secret: node.Secret,
Name: node.Name,
Description: node.Description,
})
}
return &pb.FindAllEnabledMonitorNodesResponse{MonitorNodes: result}, nil
}
// CountAllEnabledMonitorNodes 计算监控节点数量
func (this *MonitorNodeService) CountAllEnabledMonitorNodes(ctx context.Context, req *pb.CountAllEnabledMonitorNodesRequest) (*pb.RPCCountResponse, error) {
_, err := this.ValidateAdmin(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
count, err := models.SharedMonitorNodeDAO.CountAllEnabledMonitorNodes(tx)
if err != nil {
return nil, err
}
return this.SuccessCount(count)
}
// ListEnabledMonitorNodes 列出单页的监控节点
func (this *MonitorNodeService) ListEnabledMonitorNodes(ctx context.Context, req *pb.ListEnabledMonitorNodesRequest) (*pb.ListEnabledMonitorNodesResponse, error) {
_, err := this.ValidateAdmin(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
nodes, err := models.SharedMonitorNodeDAO.ListEnabledMonitorNodes(tx, req.Offset, req.Size)
if err != nil {
return nil, err
}
result := []*pb.MonitorNode{}
for _, node := range nodes {
result = append(result, &pb.MonitorNode{
Id: int64(node.Id),
IsOn: node.IsOn,
UniqueId: node.UniqueId,
Secret: node.Secret,
Name: node.Name,
Description: node.Description,
StatusJSON: node.Status,
})
}
return &pb.ListEnabledMonitorNodesResponse{MonitorNodes: result}, nil
}
// FindEnabledMonitorNode 根据ID查找节点
func (this *MonitorNodeService) FindEnabledMonitorNode(ctx context.Context, req *pb.FindEnabledMonitorNodeRequest) (*pb.FindEnabledMonitorNodeResponse, error) {
_, err := this.ValidateAdmin(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
node, err := models.SharedMonitorNodeDAO.FindEnabledMonitorNode(tx, req.MonitorNodeId)
if err != nil {
return nil, err
}
if node == nil {
return &pb.FindEnabledMonitorNodeResponse{MonitorNode: nil}, nil
}
result := &pb.MonitorNode{
Id: int64(node.Id),
IsOn: node.IsOn,
UniqueId: node.UniqueId,
Secret: node.Secret,
Name: node.Name,
Description: node.Description,
}
return &pb.FindEnabledMonitorNodeResponse{MonitorNode: result}, nil
}
// FindCurrentMonitorNode 获取当前监控节点的版本
func (this *MonitorNodeService) FindCurrentMonitorNode(ctx context.Context, req *pb.FindCurrentMonitorNodeRequest) (*pb.FindCurrentMonitorNodeResponse, error) {
_, err := this.ValidateMonitorNode(ctx)
if err != nil {
return nil, err
}
var tx = this.NullTx()
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return nil, errors.New("context: need 'nodeId'")
}
nodeIds := md.Get("nodeid")
if len(nodeIds) == 0 {
return nil, errors.New("invalid 'nodeId'")
}
nodeId := nodeIds[0]
node, err := models.SharedMonitorNodeDAO.FindEnabledMonitorNodeWithUniqueId(tx, nodeId)
if err != nil {
return nil, err
}
if node == nil {
return &pb.FindCurrentMonitorNodeResponse{MonitorNode: nil}, nil
}
result := &pb.MonitorNode{
Id: int64(node.Id),
IsOn: node.IsOn,
UniqueId: node.UniqueId,
Secret: node.Secret,
Name: node.Name,
Description: node.Description,
}
return &pb.FindCurrentMonitorNodeResponse{MonitorNode: result}, nil
}
// UpdateMonitorNodeStatus 更新节点状态
func (this *MonitorNodeService) UpdateMonitorNodeStatus(ctx context.Context, req *pb.UpdateMonitorNodeStatusRequest) (*pb.RPCSuccess, error) {
// 校验节点
_, nodeId, err := this.ValidateNodeId(ctx, rpcutils.UserTypeMonitor)
if err != nil {
return nil, err
}
if req.MonitorNodeId > 0 {
nodeId = req.MonitorNodeId
}
if nodeId <= 0 {
return nil, errors.New("'nodeId' should be greater than 0")
}
var tx = this.NullTx()
err = models.SharedMonitorNodeDAO.UpdateNodeStatus(tx, nodeId, req.StatusJSON)
if err != nil {
return nil, err
}
return this.Success()
}

12
internal/rpc/services/service_sys_locker.go
View File

@@ -14,12 +14,6 @@ type SysLockerService struct {
// SysLockerLock 获得锁 // SysLockerLock 获得锁
func (this *SysLockerService) SysLockerLock(ctx context.Context, req *pb.SysLockerLockRequest) (*pb.SysLockerLockResponse, error) { func (this *SysLockerService) SysLockerLock(ctx context.Context, req *pb.SysLockerLockRequest) (*pb.SysLockerLockResponse, error) {
_, userId, err := this.ValidateAdminAndUser(ctx, false) _, userId, err := this.ValidateAdminAndUser(ctx, false)
if err != nil {
_, err = this.ValidateMonitorNode(ctx)
if err != nil {
return nil, err
}
}
key := req.Key key := req.Key
if userId > 0 { if userId > 0 {
@@ -44,12 +38,6 @@ func (this *SysLockerService) SysLockerLock(ctx context.Context, req *pb.SysLock
// SysLockerUnlock 释放锁 // SysLockerUnlock 释放锁
func (this *SysLockerService) SysLockerUnlock(ctx context.Context, req *pb.SysLockerUnlockRequest) (*pb.RPCSuccess, error) { func (this *SysLockerService) SysLockerUnlock(ctx context.Context, req *pb.SysLockerUnlockRequest) (*pb.RPCSuccess, error) {
_, userId, err := this.ValidateAdminAndUser(ctx, false) _, userId, err := this.ValidateAdminAndUser(ctx, false)
if err != nil {
_, err = this.ValidateMonitorNode(ctx)
if err != nil {
return nil, err
}
}
key := req.Key key := req.Key
if userId > 0 { if userId > 0 {

1
internal/rpc/utils/utils.go
View File

@@ -14,7 +14,6 @@ const (
UserTypeProvider = "provider" UserTypeProvider = "provider"
UserTypeNode = "node" UserTypeNode = "node"
UserTypeCluster = "cluster" UserTypeCluster = "cluster"
UserTypeMonitor = "monitor"
UserTypeStat = "stat" UserTypeStat = "stat"
UserTypeDNS = "dns" UserTypeDNS = "dns"
UserTypeLog = "log" UserTypeLog = "log"

206
internal/setup/sql.json
View File

@@ -104469,7 +104469,144 @@
"definition": "UNIQUE KEY `type` (`type`) USING BTREE" "definition": "UNIQUE KEY `type` (`type`) USING BTREE"
} }
], ],
"records": [] "records": [
{
"id": 1,
"values": {
"description": "通过邮件发送通知",
"id": "1",
"isOn": "1",
"name": "邮件",
"order": "8",
"state": "1",
"type": "email",
"userDescription": "接收人邮箱地址"
},
"uniqueFields": [
"type"
],
"exceptFields": null
},
{
"id": 2,
"values": {
"description": "通过HTTP请求发送通知",
"id": "2",
"isOn": "1",
"name": "WebHook",
"order": "7",
"state": "1",
"type": "webHook",
"userDescription": "通过${MessageUser}参数传递到URL上"
},
"uniqueFields": [
"type"
],
"exceptFields": null
},
{
"id": 3,
"values": {
"description": "通过运行脚本发送通知",
"id": "3",
"isOn": "1",
"name": "脚本",
"order": "6",
"state": "1",
"type": "script",
"userDescription": "可以在脚本中使用${MessageUser}来获取这个标识"
},
"uniqueFields": [
"type"
],
"exceptFields": null
},
{
"id": 4,
"values": {
"description": "通过钉钉群机器人发送通知消息",
"id": "4",
"isOn": "1",
"name": "钉钉群机器人",
"order": "5",
"state": "1",
"type": "dingTalk",
"userDescription": "要At（@）的群成员的手机号，多个手机号用英文逗号隔开，也可以为空"
},
"uniqueFields": [
"type"
],
"exceptFields": null
},
{
"id": 5,
"values": {
"description": "通过企业微信应用发送通知消息",
"id": "5",
"isOn": "1",
"name": "企业微信应用",
"order": "4",
"state": "1",
"type": "qyWeixin",
"userDescription": "接收消息的成员的用户账号，多个成员用竖线（|）分隔，如果所有成员使用@all。留空表示所有成员。"
},
"uniqueFields": [
"type"
],
"exceptFields": null
},
{
"id": 6,
"values": {
"description": "通过微信群机器人发送通知消息",
"id": "6",
"isOn": "1",
"name": "企业微信群机器人",
"order": "3",
"state": "1",
"type": "qyWeixinRobot",
"userDescription": "要At（@）的群成员的手机号，多个手机号用英文逗号隔开，也可以为空"
},
"uniqueFields": [
"type"
],
"exceptFields": null
},
{
"id": 7,
"values": {
"description": "通过\u003ca href=\"https://www.aliyun.com/product/sms?spm=5176.11533447.1097531.2.12055cfa6UnIix\" target=\"_blank\"\u003e阿里云短信服务\u003c/a\u003e发送短信",
"id": "7",
"isOn": "1",
"name": "阿里云短信",
"order": "2",
"state": "1",
"type": "aliyunSms",
"userDescription": "接收消息的手机号"
},
"uniqueFields": [
"type"
],
"exceptFields": null
},
{
"id": 8,
"values": {
"description": "通过机器人向群或者某个用户发送消息，需要确保所在网络能够访问Telegram API服务",
"id": "8",
"isOn": "1",
"name": "Telegram机器人",
"order": "1",
"state": "1",
"type": "telegram",
"userDescription": "群或用户的Chat ID，通常是一个数字，可以通过和 @get_id_bot 建立对话并发送任意消息获得"
},
"uniqueFields": [
"type"
],
"exceptFields": null
}
]
}, },
{ {
"name": "edgeMessageReceivers", "name": "edgeMessageReceivers",
@@ -108529,73 +108666,6 @@
], ],
"records": [] "records": []
}, },
{
"name": "edgeMonitorNodes",
"engine": "InnoDB",
"charset": "utf8mb4_general_ci",
"definition": "CREATE TABLE `edgeMonitorNodes` (\n `id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT 'ID',\n `isOn` tinyint(1) unsigned DEFAULT '1' COMMENT '是否启用',\n `uniqueId` varchar(32) DEFAULT NULL COMMENT '唯一ID',\n `secret` varchar(32) DEFAULT NULL COMMENT '密钥',\n `name` varchar(255) DEFAULT NULL COMMENT '名称',\n `description` varchar(1024) DEFAULT NULL COMMENT '描述',\n `order` int(11) unsigned DEFAULT '0' COMMENT '排序',\n `state` tinyint(1) unsigned DEFAULT '1' COMMENT '状态',\n `createdAt` bigint(11) unsigned DEFAULT '0' COMMENT '创建时间',\n `adminId` int(11) unsigned DEFAULT '0' COMMENT '管理员ID',\n `weight` int(11) unsigned DEFAULT '0' COMMENT '权重',\n `status` json DEFAULT NULL COMMENT '运行状态',\n PRIMARY KEY (`id`),\n UNIQUE KEY `uniqueId` (`uniqueId`) USING BTREE\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='监控节点'",
"fields": [
{
"name": "id",
"definition": "int(11) unsigned auto_increment COMMENT 'ID'"
},
{
"name": "isOn",
"definition": "tinyint(1) unsigned DEFAULT '1' COMMENT '是否启用'"
},
{
"name": "uniqueId",
"definition": "varchar(32) COMMENT '唯一ID'"
},
{
"name": "secret",
"definition": "varchar(32) COMMENT '密钥'"
},
{
"name": "name",
"definition": "varchar(255) COMMENT '名称'"
},
{
"name": "description",
"definition": "varchar(1024) COMMENT '描述'"
},
{
"name": "order",
"definition": "int(11) unsigned DEFAULT '0' COMMENT '排序'"
},
{
"name": "state",
"definition": "tinyint(1) unsigned DEFAULT '1' COMMENT '状态'"
},
{
"name": "createdAt",
"definition": "bigint(11) unsigned DEFAULT '0' COMMENT '创建时间'"
},
{
"name": "adminId",
"definition": "int(11) unsigned DEFAULT '0' COMMENT '管理员ID'"
},
{
"name": "weight",
"definition": "int(11) unsigned DEFAULT '0' COMMENT '权重'"
},
{
"name": "status",
"definition": "json COMMENT '运行状态'"
}
],
"indexes": [
{
"name": "PRIMARY",
"definition": "UNIQUE KEY `PRIMARY` (`id`) USING BTREE"
},
{
"name": "uniqueId",
"definition": "UNIQUE KEY `uniqueId` (`uniqueId`) USING BTREE"
}
],
"records": []
},
{ {
"name": "edgeNSAccessLogs", "name": "edgeNSAccessLogs",
"engine": "InnoDB", "engine": "InnoDB",

5
internal/setup/sql_dump.go
View File

@@ -60,6 +60,11 @@ var recordsTables = []*SQLRecordsTable{
UniqueFields: []string{"agentId", "ip"}, UniqueFields: []string{"agentId", "ip"},
IgnoreId: true, IgnoreId: true,
}, },
{
TableName: "edgeMessageMedias",
UniqueFields: []string{"type"},
IgnoreId: true,
},
} }
type sqlItem struct { type sqlItem struct {

2
internal/tasks/node_monitor_task.go
View File

@@ -144,7 +144,7 @@ func (this *NodeMonitorTask) MonitorCluster(cluster *models.NodeCluster) error {
this.notifiedMap[nodeId] = time.Now().Unix() this.notifiedMap[nodeId] = time.Now().Unix()
var subject = "节点\"" + node.Name + "\"已处于离线状态" var subject = "节点\"" + node.Name + "\"已处于离线状态"
var msg = "集群'" + cluster.Name + "'节点\"" + node.Name + "\"已处于离线状态，请检查节点是否异常" var msg = "集群 \"" + cluster.Name + "\" 节点 \"" + node.Name + "\" 已处于离线状态，请检查节点是否异常"
err = models.SharedMessageDAO.CreateNodeMessage(nil, nodeconfigs.NodeRoleNode, clusterId, int64(node.Id), models.MessageTypeNodeInactive, models.LevelError, subject, msg, nil, false) 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