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

所有数据库相关的操作支持事务

Browse Source
This commit is contained in:
刘祥超
2021-01-01 23:31:30 +08:00
parent 0cf398a436
commit f49c26cdab
146 changed files with 2845 additions and 2068 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
internal/tasks/event_looper.go
View File

@@ -34,12 +34,12 @@ func (this *EventLooper) Start() {
func (this *EventLooper) loop() error {
lockerKey := "eventLooper"
isOk, err := models.SharedSysLockerDAO.Lock(lockerKey, 3600)
isOk, err := models.SharedSysLockerDAO.Lock(nil, lockerKey, 3600)
if err != nil {
return err
}
defer func() {
err = models.SharedSysLockerDAO.Unlock(lockerKey)
err = models.SharedSysLockerDAO.Unlock(nil, lockerKey)
if err != nil {
logs.Println("[EVENT_LOOPER]" + err.Error())
}
@@ -48,7 +48,7 @@ func (this *EventLooper) loop() error {
return nil
}
events, err := models.SharedSysEventDAO.FindEvents(100)
events, err := models.SharedSysEventDAO.FindEvents(nil, 100)
if err != nil {
return err
}
@@ -63,7 +63,7 @@ func (this *EventLooper) loop() error {
logs.Println("[EVENT_LOOPER]" + err.Error())
continue
}
err = models.SharedSysEventDAO.DeleteEvent(int64(eventOne.Id))
err = models.SharedSysEventDAO.DeleteEvent(nil, int64(eventOne.Id))
if err != nil {
return err
}

6
internal/tasks/health_check_cluster_task.go
View File

@@ -89,7 +89,7 @@ func (this *HealthCheckClusterTask) loop(seconds int64) error {
// 检查上次运行时间，防止重复运行
settingKey := models.SettingCodeClusterHealthCheck + "Loop" + numberutils.FormatInt64(this.clusterId)
timestamp := time.Now().Unix()
c, err := models.SharedSysSettingDAO.CompareInt64Setting(settingKey, timestamp-seconds)
c, err := models.SharedSysSettingDAO.CompareInt64Setting(nil, settingKey, timestamp-seconds)
if err != nil {
return err
}
@@ -98,7 +98,7 @@ func (this *HealthCheckClusterTask) loop(seconds int64) error {
}
// 记录时间
err = models.SharedSysSettingDAO.UpdateSetting(settingKey, []byte(numberutils.FormatInt64(timestamp)))
err = models.SharedSysSettingDAO.UpdateSetting(nil, settingKey, []byte(numberutils.FormatInt64(timestamp)))
if err != nil {
return err
}
@@ -131,7 +131,7 @@ func (this *HealthCheckClusterTask) loop(seconds int64) error {
return err
}
message := "有" + numberutils.FormatInt(len(failedResults)) + "个节点在健康检查中出现问题"
err = models.NewMessageDAO().CreateClusterMessage(this.clusterId, models.MessageTypeHealthCheckFailed, models.MessageLevelError, message, failedResultsJSON)
err = models.NewMessageDAO().CreateClusterMessage(nil, this.clusterId, models.MessageTypeHealthCheckFailed, models.MessageLevelError, message, failedResultsJSON)
if err != nil {
return err
}

12
internal/tasks/health_check_executor.go
View File

@@ -26,7 +26,7 @@ func NewHealthCheckExecutor(clusterId int64) *HealthCheckExecutor {
}
func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
cluster, err := models.NewNodeClusterDAO().FindEnabledNodeCluster(this.clusterId)
cluster, err := models.NewNodeClusterDAO().FindEnabledNodeCluster(nil, this.clusterId)
if err != nil {
return nil, err
}
@@ -44,7 +44,7 @@ func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
}
results := []*HealthCheckResult{}
nodes, err := models.NewNodeDAO().FindAllEnabledNodesWithClusterId(this.clusterId)
nodes, err := models.NewNodeDAO().FindAllEnabledNodesWithClusterId(nil, this.clusterId)
if err != nil {
return nil, err
}
@@ -56,7 +56,7 @@ func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
Node: node,
}
ipAddr, err := models.NewNodeIPAddressDAO().FindFirstNodeIPAddress(int64(node.Id))
ipAddr, err := models.NewNodeIPAddressDAO().FindFirstNodeIPAddress(nil, int64(node.Id))
if err != nil {
return nil, err
}
@@ -129,7 +129,7 @@ func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
// 修改节点状态
if healthCheckConfig.AutoDown {
isChanged, err := models.SharedNodeDAO.UpdateNodeUp(int64(result.Node.Id), result.IsOk, healthCheckConfig.CountUp, healthCheckConfig.CountDown)
isChanged, err := models.SharedNodeDAO.UpdateNodeUp(nil, int64(result.Node.Id), result.IsOk, healthCheckConfig.CountUp, healthCheckConfig.CountDown)
if err != nil {
logs.Println("[HEALTH_CHECK]" + err.Error())
} else if isChanged {
@@ -142,9 +142,9 @@ func (this *HealthCheckExecutor) Run() ([]*HealthCheckResult, error) {
// 通知恢复或下线
if result.IsOk {
err = models.NewMessageDAO().CreateNodeMessage(this.clusterId, int64(result.Node.Id), models.MessageTypeHealthCheckNodeUp, models.MessageLevelSuccess, "健康检查成功，节点\""+result.Node.Name+"\"已恢复上线", nil)
err = models.NewMessageDAO().CreateNodeMessage(nil, this.clusterId, int64(result.Node.Id), models.MessageTypeHealthCheckNodeUp, models.MessageLevelSuccess, "健康检查成功，节点\""+result.Node.Name+"\"已恢复上线", nil)
} else {
err = models.NewMessageDAO().CreateNodeMessage(this.clusterId, int64(result.Node.Id), models.MessageTypeHealthCheckNodeDown, models.MessageLevelError, "健康检查失败，节点\""+result.Node.Name+"\"已自动下线", nil)
err = models.NewMessageDAO().CreateNodeMessage(nil, this.clusterId, int64(result.Node.Id), models.MessageTypeHealthCheckNodeDown, models.MessageLevelError, "健康检查失败，节点\""+result.Node.Name+"\"已自动下线", nil)
}
}
}

2
internal/tasks/health_check_task.go
View File

@@ -46,7 +46,7 @@ func (this *HealthCheckTask) Run() {
}
func (this *HealthCheckTask) loop() error {
clusters, err := models.NewNodeClusterDAO().FindAllEnableClusters()
clusters, err := models.NewNodeClusterDAO().FindAllEnableClusters(nil)
if err != nil {
return err
}

16
internal/tasks/log_task.go
View File

@@ -44,7 +44,7 @@ func (this *LogTask) loopClean(seconds int64) error {
// 检查上次运行时间，防止重复运行
settingKey := "logTaskCleanLoop"
timestamp := time.Now().Unix()
c, err := models.SharedSysSettingDAO.CompareInt64Setting(settingKey, timestamp-seconds)
c, err := models.SharedSysSettingDAO.CompareInt64Setting(nil, settingKey, timestamp-seconds)
if err != nil {
return err
}
@@ -53,13 +53,13 @@ func (this *LogTask) loopClean(seconds int64) error {
}
// 记录时间
err = models.SharedSysSettingDAO.UpdateSetting(settingKey, []byte(numberutils.FormatInt64(timestamp)))
err = models.SharedSysSettingDAO.UpdateSetting(nil, settingKey, []byte(numberutils.FormatInt64(timestamp)))
if err != nil {
return err
}
configKey := "adminLogConfig"
valueJSON, err := models.SharedSysSettingDAO.ReadSetting(configKey)
valueJSON, err := models.SharedSysSettingDAO.ReadSetting(nil, configKey)
if err != nil {
return err
}
@@ -73,7 +73,7 @@ func (this *LogTask) loopClean(seconds int64) error {
return err
}
if config.Days > 0 {
err = models.SharedLogDAO.DeleteLogsPermanentlyBeforeDays(config.Days)
err = models.SharedLogDAO.DeleteLogsPermanentlyBeforeDays(nil, config.Days)
if err != nil {
return err
}
@@ -95,7 +95,7 @@ func (this *LogTask) loopMonitor(seconds int64) error {
// 检查上次运行时间，防止重复运行
settingKey := "logTaskMonitorLoop"
timestamp := time.Now().Unix()
c, err := models.SharedSysSettingDAO.CompareInt64Setting(settingKey, timestamp-seconds)
c, err := models.SharedSysSettingDAO.CompareInt64Setting(nil, settingKey, timestamp-seconds)
if err != nil {
return err
}
@@ -104,13 +104,13 @@ func (this *LogTask) loopMonitor(seconds int64) error {
}
// 记录时间
err = models.SharedSysSettingDAO.UpdateSetting(settingKey, []byte(numberutils.FormatInt64(timestamp)))
err = models.SharedSysSettingDAO.UpdateSetting(nil, settingKey, []byte(numberutils.FormatInt64(timestamp)))
if err != nil {
return err
}
configKey := "adminLogConfig"
valueJSON, err := models.SharedSysSettingDAO.ReadSetting(configKey)
valueJSON, err := models.SharedSysSettingDAO.ReadSetting(nil, configKey)
if err != nil {
return err
}
@@ -132,7 +132,7 @@ func (this *LogTask) loopMonitor(seconds int64) error {
return err
}
if sumBytes > capacityBytes {
err := models.SharedMessageDAO.CreateMessage(0, 0, models.MessageTypeLogCapacityOverflow, models.MessageLevelError, "日志用量已经超出最大限制，当前的用量为"+this.formatBytes(sumBytes)+"，而设置的最大容量为"+this.formatBytes(capacityBytes)+"。", nil)
err := models.SharedMessageDAO.CreateMessage(nil, 0, 0, models.MessageTypeLogCapacityOverflow, models.MessageLevelError, "日志用量已经超出最大限制，当前的用量为"+this.formatBytes(sumBytes)+"，而设置的最大容量为"+this.formatBytes(capacityBytes)+"。", nil)
if err != nil {
return err
}

2
internal/tasks/message_task.go
View File

@@ -37,5 +37,5 @@ func (this *MessageTask) Run() {
// 单次运行
func (this *MessageTask) loop() error {
dayTime := time.Now().AddDate(0, 0, -30) // TODO 这个30天应该可以在界面上设置
return models.NewMessageDAO().DeleteMessagesBeforeDay(dayTime)
return models.NewMessageDAO().DeleteMessagesBeforeDay(nil, dayTime)
}

2
internal/tasks/node_log_cleaner_task.go
View File

@@ -36,5 +36,5 @@ func (this *NodeLogCleanerTask) Start() {
func (this *NodeLogCleanerTask) loop() error {
// TODO 30天这个数值改成可以设置
return models.SharedNodeLogDAO.DeleteExpiredLogs(30)
return models.SharedNodeLogDAO.DeleteExpiredLogs(nil, 30)
}

12
internal/tasks/node_monitor_task.go
View File

@@ -42,7 +42,7 @@ func (this *NodeMonitorTask) loop() error {
// 检查上次运行时间，防止重复运行
settingKey := models.SettingCodeNodeMonitor + "Loop"
timestamp := time.Now().Unix()
c, err := models.SharedSysSettingDAO.CompareInt64Setting(settingKey, timestamp-int64(this.intervalSeconds))
c, err := models.SharedSysSettingDAO.CompareInt64Setting(nil, settingKey, timestamp-int64(this.intervalSeconds))
if err != nil {
return err
}
@@ -51,12 +51,12 @@ func (this *NodeMonitorTask) loop() error {
}
// 记录时间
err = models.SharedSysSettingDAO.UpdateSetting(settingKey, []byte(numberutils.FormatInt64(timestamp)))
err = models.SharedSysSettingDAO.UpdateSetting(nil, settingKey, []byte(numberutils.FormatInt64(timestamp)))
if err != nil {
return err
}
clusters, err := models.SharedNodeClusterDAO.FindAllEnableClusters()
clusters, err := models.SharedNodeClusterDAO.FindAllEnableClusters(nil)
if err != nil {
return err
}
@@ -74,18 +74,18 @@ func (this *NodeMonitorTask) monitorCluster(cluster *models.NodeCluster) error {
clusterId := int64(cluster.Id)
// 检查离线节点
inactiveNodes, err := models.SharedNodeDAO.FindAllInactiveNodesWithClusterId(clusterId)
inactiveNodes, err := models.SharedNodeDAO.FindAllInactiveNodesWithClusterId(nil, clusterId)
if err != nil {
return err
}
for _, node := range inactiveNodes {
err = models.SharedMessageDAO.CreateNodeMessage(clusterId, int64(node.Id), models.MessageTypeNodeInactive, models.LevelError, "节点已处于离线状态", nil)
err = models.SharedMessageDAO.CreateNodeMessage(nil, clusterId, int64(node.Id), models.MessageTypeNodeInactive, models.LevelError, "节点已处于离线状态", nil)
if err != nil {
return err
}
// 修改在线状态
err = models.SharedNodeDAO.UpdateNodeActive(int64(node.Id), false)
err = models.SharedNodeDAO.UpdateNodeActive(nil, int64(node.Id), false)
if err != nil {
return err
}

36
internal/tasks/ssl_cert_expire_check_executor.go
View File

@@ -42,7 +42,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
// 检查上次运行时间，防止重复运行
settingKey := "sslCertExpiringCheckLoop"
timestamp := time.Now().Unix()
c, err := models.SharedSysSettingDAO.CompareInt64Setting(settingKey, timestamp-seconds)
c, err := models.SharedSysSettingDAO.CompareInt64Setting(nil, settingKey, timestamp-seconds)
if err != nil {
return err
}
@@ -51,7 +51,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
}
// 记录时间
err = models.SharedSysSettingDAO.UpdateSetting(settingKey, []byte(numberutils.FormatInt64(timestamp)))
err = models.SharedSysSettingDAO.UpdateSetting(nil, settingKey, []byte(numberutils.FormatInt64(timestamp)))
if err != nil {
return err
}
@@ -59,7 +59,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
// 查找需要自动更新的证书
// 30, 14 ... 是到期的天数
for _, days := range []int{30, 14, 7} {
certs, err := models.SharedSSLCertDAO.FindAllExpiringCerts(days)
certs, err := models.SharedSSLCertDAO.FindAllExpiringCerts(nil, days)
if err != nil {
return err
}
@@ -69,7 +69,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
// 是否有自动更新任务
if cert.AcmeTaskId > 0 {
task, err := models.SharedACMETaskDAO.FindEnabledACMETask(int64(cert.AcmeTaskId))
task, err := models.SharedACMETaskDAO.FindEnabledACMETask(nil, int64(cert.AcmeTaskId))
if err != nil {
return err
}
@@ -84,7 +84,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
msg += "请及时更新证书。"
}
err = models.SharedMessageDAO.CreateMessage(int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertExpiring, models.MessageLevelWarning, msg, maps.Map{
err = models.SharedMessageDAO.CreateMessage(nil, int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertExpiring, models.MessageLevelWarning, msg, maps.Map{
"certId": cert.Id,
"acmeTaskId": cert.AcmeTaskId,
}.AsJSON())
@@ -93,7 +93,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
}
// 设置最后通知时间
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(int64(cert.Id))
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(nil, int64(cert.Id))
if err != nil {
return err
}
@@ -102,7 +102,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
// 自动续期
for _, days := range []int{3, 2, 1} {
certs, err := models.SharedSSLCertDAO.FindAllExpiringCerts(days)
certs, err := models.SharedSSLCertDAO.FindAllExpiringCerts(nil, days)
if err != nil {
return err
}
@@ -112,36 +112,36 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
// 是否有自动更新任务
if cert.AcmeTaskId > 0 {
task, err := models.SharedACMETaskDAO.FindEnabledACMETask(int64(cert.AcmeTaskId))
task, err := models.SharedACMETaskDAO.FindEnabledACMETask(nil, int64(cert.AcmeTaskId))
if err != nil {
return err
}
if task != nil {
if task.AutoRenew == 1 {
isOk, errMsg, _ := models.SharedACMETaskDAO.RunTask(int64(cert.AcmeTaskId))
isOk, errMsg, _ := models.SharedACMETaskDAO.RunTask(nil, int64(cert.AcmeTaskId))
if isOk {
// 发送成功通知
msg = "系统已成功为你自动更新了证书\"" + cert.Name + "\"（" + cert.DnsNames + "）。"
err = models.SharedMessageDAO.CreateMessage(int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertACMETaskSuccess, models.MessageLevelSuccess, msg, maps.Map{
err = models.SharedMessageDAO.CreateMessage(nil, int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertACMETaskSuccess, models.MessageLevelSuccess, msg, maps.Map{
"certId": cert.Id,
"acmeTaskId": cert.AcmeTaskId,
}.AsJSON())
// 更新通知时间
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(int64(cert.Id))
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(nil, int64(cert.Id))
if err != nil {
return err
}
} else {
// 发送失败通知
msg = "系统在尝试自动更新证书\"" + cert.Name + "\"（" + cert.DnsNames + "）时发生错误：" + errMsg + "。请检查系统设置并修复错误。"
err = models.SharedMessageDAO.CreateMessage(int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertACMETaskFailed, models.MessageLevelError, msg, maps.Map{
err = models.SharedMessageDAO.CreateMessage(nil, int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertACMETaskFailed, models.MessageLevelError, msg, maps.Map{
"certId": cert.Id,
"acmeTaskId": cert.AcmeTaskId,
}.AsJSON())
// 更新通知时间
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(int64(cert.Id))
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(nil, int64(cert.Id))
if err != nil {
return err
}
@@ -158,7 +158,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
msg += "请及时更新证书。"
}
err = models.SharedMessageDAO.CreateMessage(int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertExpiring, models.MessageLevelWarning, msg, maps.Map{
err = models.SharedMessageDAO.CreateMessage(nil, int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertExpiring, models.MessageLevelWarning, msg, maps.Map{
"certId": cert.Id,
"acmeTaskId": cert.AcmeTaskId,
}.AsJSON())
@@ -167,7 +167,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
}
// 设置最后通知时间
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(int64(cert.Id))
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(nil, int64(cert.Id))
if err != nil {
return err
}
@@ -176,7 +176,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
// 当天过期
for _, days := range []int{0} {
certs, err := models.SharedSSLCertDAO.FindAllExpiringCerts(days)
certs, err := models.SharedSSLCertDAO.FindAllExpiringCerts(nil, days)
if err != nil {
return err
}
@@ -184,7 +184,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
// 发送消息
today := timeutil.Format("Y-m-d")
msg := "SSL证书\"" + cert.Name + "\"（" + cert.DnsNames + "）在今天（" + today + "）过期，请及时更新证书，之后将不再重复提醒。"
err = models.SharedMessageDAO.CreateMessage(int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertExpiring, models.MessageLevelWarning, msg, maps.Map{
err = models.SharedMessageDAO.CreateMessage(nil, int64(cert.AdminId), int64(cert.UserId), models.MessageTypeSSLCertExpiring, models.MessageLevelWarning, msg, maps.Map{
"certId": cert.Id,
"acmeTaskId": cert.AcmeTaskId,
}.AsJSON())
@@ -193,7 +193,7 @@ func (this *SSLCertExpireCheckExecutor) loop(seconds int64) error {
}
// 设置最后通知时间
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(int64(cert.Id))
err = models.SharedSSLCertDAO.UpdateCertNotifiedAt(nil, int64(cert.Id))
if err != nil {
return err
}