OMC/be.ems
1
0
Fork 0
Code Issues Releases 3 Activity

feat: 调度任务-告警检查CPU/License/导出数据

Browse Source
This commit is contained in:
TsMask
2025-05-09 18:46:25 +08:00
parent 24e684856e
commit 87795e1a7e
15 changed files with 2991 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1442
src/modules/crontask/processor/backup_export_cdr/backup_export_cdr.go Normal file
View File

File diff suppressed because it is too large Load Diff

367
src/modules/crontask/processor/backup_export_log/backup_export_log.go Normal file
View File

@@ -0,0 +1,367 @@
package backup_export_log
import (
"encoding/json"
"fmt"
"path/filepath"
"runtime"
"strconv"
"time"
"be.ems/src/framework/cron"
"be.ems/src/framework/i18n"
"be.ems/src/framework/logger"
"be.ems/src/framework/utils/date"
"be.ems/src/framework/utils/file"
"be.ems/src/framework/utils/parse"
neDataService "be.ems/src/modules/network_data/service"
systemModel "be.ems/src/modules/system/model"
systemService "be.ems/src/modules/system/service"
)
var NewProcessor = &BackupExportLogProcessor{
count: 0,
backupService: neDataService.NewBackup,
sysLogLoginService: systemService.NewSysLogLoginImpl,
sysOperateService: systemService.NewSysLogOperateImpl,
}
// BackupExportLog 队列任务处理
type BackupExportLogProcessor struct {
count int // 执行次数
backupService *neDataService.Backup // 备份相关服务
sysLogLoginService *systemService.SysLogLoginImpl // 系统登录日志服务
sysOperateService *systemService.SysLogOperateImpl // 系统操作日志服务
}
func (s *BackupExportLogProcessor) Execute(data any) (any, error) {
s.count++ // 执行次数加一
options := data.(cron.JobData)
sysJob := options.SysJob
logger.Infof("重复:%v 任务ID:%s 执行次数:%d", options.Repeat, sysJob.JobID, s.count)
// 返回结果，用于记录执行结果
result := map[string]any{
"count": s.count,
}
var params struct {
DataType []string `json:"dataType"` // 类型支持 operate/login
FileType string `json:"fileType"` // 文件类型 csv/xlsx
Hour int `json:"hour"` // 数据时间从任务执行时间前的小时数
}
if err := json.Unmarshal([]byte(sysJob.TargetParams), &params); err != nil {
return nil, err
}
if !(params.FileType == "csv" || params.FileType == "xlsx") {
return nil, fmt.Errorf("file type error, only support csv,xlsx")
}
for _, v := range params.DataType {
switch v {
case "operate":
result[v] = s.exportOperate(params.Hour, params.FileType)
case "login":
result[v] = s.exportLogin(params.Hour, params.FileType)
}
}
// 返回结果，用于记录执行结果
return result, nil
}
// exportOperate 导出系统操作日志数据
func (s BackupExportLogProcessor) exportOperate(hour int, fileType string) string {
// 前 hour 小时
now := time.Now()
end := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, now.Location())
start := end.Add(-time.Duration(hour) * time.Hour)
language := "en"
query := map[string]any{
"beginTime": fmt.Sprint(start.UnixMilli()),
"endTime": fmt.Sprint(end.UnixMilli()),
"pageNum": "1",
"pageSize": "30000",
}
data := s.sysOperateService.SelectSysLogOperatePage(query, "")
// rows, total
if data == nil || parse.Number(data["total"]) == 0 {
return "no data"
}
rows := data["rows"].([]systemModel.SysLogOperate)
// 闭包函数处理多语言
converI18n := func(language string, arr *[]systemModel.SysLogOperate) {
for i := range *arr {
(*arr)[i].Title = i18n.TKey(language, (*arr)[i].Title)
(*arr)[i].OperLocation = i18n.TKey(language, (*arr)[i].OperLocation)
}
}
converI18n(language, &rows)
// 导出文件名称
fileName := fmt.Sprintf("sys_log_operate_export_%d_%s.%s", len(rows), date.ParseDateToStr(end, date.YYYYMMDDHHMMSS), fileType)
filePath := filepath.Join(s.backupService.BACKUP_DIR, "/log/sys_log_operate", fileName)
if runtime.GOOS == "windows" {
filePath = fmt.Sprintf("C:%s", filePath)
}
// 业务类型
businessTypeFunc := func(v string) string {
businessType := ""
switch v {
case "0":
// 业务操作类型-其它
businessType = i18n.TKey(language, "dictData.operType.other")
case "1":
// 业务操作类型-新增
businessType = i18n.TKey(language, "dictData.operType.add")
case "2":
// 业务操作类型-修改
businessType = i18n.TKey(language, "dictData.operType.edit")
case "3":
// 业务操作类型-删除
businessType = i18n.TKey(language, "dictData.operType.delete")
case "4":
// 业务操作类型-授权
businessType = i18n.TKey(language, "dictData.operType.auth")
case "5":
// 业务操作类型-导出
businessType = i18n.TKey(language, "dictData.operType.export")
case "6":
// 业务操作类型-导入
businessType = i18n.TKey(language, "dictData.operType.import")
case "7":
// 业务操作类型-强退
businessType = i18n.TKey(language, "dictData.operType.forced quit")
case "8":
// 业务操作类型-清空数据
businessType = i18n.TKey(language, "dictData.operType.clear")
}
return businessType
}
if fileType == "csv" {
// 转换数据
data := [][]string{
{
i18n.TKey(language, "log.operate.export.id"),
i18n.TKey(language, "log.operate.export.title"),
i18n.TKey(language, "log.operate.export.businessType"),
i18n.TKey(language, "log.operate.export.operName"),
i18n.TKey(language, "log.operate.export.method"),
i18n.TKey(language, "log.operate.export.ip"),
i18n.TKey(language, "log.operate.export.status"),
i18n.TKey(language, "log.operate.export.operTime"),
i18n.TKey(language, "log.operate.export.costTime"),
"Tenant Name", // for multi-tenancy
},
}
for _, row := range rows {
// 业务类型
businessType := businessTypeFunc(row.BusinessType)
// 状态
statusValue := i18n.TKey(language, "dictData.fail")
if row.Status == "1" {
statusValue = i18n.TKey(language, "dictData.success")
}
data = append(data, []string{
fmt.Sprint(row.OperID),
row.Title,
businessType,
row.OperName,
row.RequestMethod,
row.OperIP,
statusValue,
date.ParseDateToStr(row.OperTime, date.YYYY_MM_DDTHH_MM_SSZ),
fmt.Sprint(row.CostTime),
row.TenantName,
})
}
// 输出到文件
if err := file.WriterFileCSV(data, filePath); err != nil {
logger.Errorf("export operate log err => %v", err.Error())
return "export err"
}
}
if fileType == "xlsx" {
// 第一行表头标题
headerCells := map[string]string{
"A1": i18n.TKey(language, "log.operate.export.id"),
"B1": i18n.TKey(language, "log.operate.export.title"),
"C1": i18n.TKey(language, "log.operate.export.businessType"),
"D1": i18n.TKey(language, "log.operate.export.operName"),
"E1": i18n.TKey(language, "log.operate.export.method"),
"F1": i18n.TKey(language, "log.operate.export.ip"),
"G1": i18n.TKey(language, "log.operate.export.status"),
"H1": i18n.TKey(language, "log.operate.export.operTime"),
"I1": i18n.TKey(language, "log.operate.export.costTime"),
"J1": "Tenant Name", // for multi-tenancy
}
// 从第二行开始的数据
dataCells := make([]map[string]any, 0)
for i, row := range rows {
idx := strconv.Itoa(i + 2)
// 业务类型
businessType := businessTypeFunc(row.BusinessType)
// 状态
statusValue := i18n.TKey(language, "dictData.fail")
if row.Status == "1" {
statusValue = i18n.TKey(language, "dictData.success")
}
dataCells = append(dataCells, map[string]any{
"A" + idx: row.OperID,
"B" + idx: row.Title,
"C" + idx: businessType,
"D" + idx: row.OperName,
"E" + idx: row.RequestMethod,
"F" + idx: row.OperIP,
"G" + idx: statusValue,
"H" + idx: date.ParseDateToStr(row.OperTime, date.YYYY_MM_DDTHH_MM_SSZ),
"I" + idx: row.CostTime,
"J" + idx: row.TenantName,
})
}
// 导出数据表格
if err := file.WriterFileExecl(headerCells, dataCells, filePath, ""); err != nil {
logger.Errorf("export operate log err => %v", err.Error())
return "export err"
}
}
// 上传到FTP服务器
if err := s.backupService.FTPPushFile(filePath, "log"); err != nil {
return "ok, ftp err:" + err.Error()
}
return "ok"
}
// exportLogin 导出系统登录日志数据
func (s BackupExportLogProcessor) exportLogin(hour int, fileType string) string {
// 前 hour 小时
now := time.Now()
end := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, now.Location())
start := end.Add(-time.Duration(hour) * time.Hour)
language := "en"
query := map[string]any{
"beginTime": fmt.Sprint(start.UnixMilli()),
"endTime": fmt.Sprint(end.UnixMilli()),
"pageNum": "1",
"pageSize": "30000",
}
data := s.sysLogLoginService.SelectSysLogLoginPage(query, "")
// rows, total
if data == nil || parse.Number(data["total"]) == 0 {
return "no data"
}
rows := data["rows"].([]systemModel.SysLogLogin)
// 闭包函数处理多语言
converI18n := func(language string, arr *[]systemModel.SysLogLogin) {
for i := range *arr {
(*arr)[i].LoginLocation = i18n.TKey(language, (*arr)[i].LoginLocation)
(*arr)[i].OS = i18n.TKey(language, (*arr)[i].OS)
(*arr)[i].Browser = i18n.TKey(language, (*arr)[i].Browser)
(*arr)[i].Msg = i18n.TKey(language, (*arr)[i].Msg)
}
}
converI18n(language, &rows)
// 导出文件名称
fileName := fmt.Sprintf("sys_log_login_export_%d_%s.%s", len(rows), date.ParseDateToStr(end, date.YYYYMMDDHHMMSS), fileType)
filePath := filepath.Join(s.backupService.BACKUP_DIR, "/log/sys_log_login", fileName)
if runtime.GOOS == "windows" {
filePath = fmt.Sprintf("C:%s", filePath)
}
if fileType == "csv" {
// 转换数据
data := [][]string{
{
i18n.TKey(language, "log.login.export.id"),
i18n.TKey(language, "log.login.export.userName"),
i18n.TKey(language, "log.login.export.ip"),
i18n.TKey(language, "log.login.export.location"),
i18n.TKey(language, "log.login.export.os"),
i18n.TKey(language, "log.login.export.browser"),
i18n.TKey(language, "log.login.export.status"),
i18n.TKey(language, "log.login.export.time"),
i18n.TKey(language, "log.login.export.msg"),
},
}
for _, row := range rows {
// 状态
statusValue := i18n.TKey(language, "dictData.fail")
if row.Status == "1" {
statusValue = i18n.TKey(language, "dictData.success")
}
data = append(data, []string{
fmt.Sprint(row.LoginID),
row.UserName,
row.IPAddr,
row.LoginLocation,
row.OS,
row.Browser,
statusValue,
date.ParseDateToStr(row.LoginTime, date.YYYY_MM_DDTHH_MM_SSZ),
row.Msg,
})
}
// 输出到文件
if err := file.WriterFileCSV(data, filePath); err != nil {
logger.Errorf("export login log err => %v", err.Error())
return "export err"
}
}
if fileType == "xlsx" {
// 第一行表头标题
headerCells := map[string]string{
"A1": i18n.TKey(language, "log.login.export.id"),
"B1": i18n.TKey(language, "log.login.export.userName"),
"C1": i18n.TKey(language, "log.login.export.ip"),
"D1": i18n.TKey(language, "log.login.export.location"),
"E1": i18n.TKey(language, "log.login.export.os"),
"F1": i18n.TKey(language, "log.login.export.browser"),
"G1": i18n.TKey(language, "log.login.export.status"),
"H1": i18n.TKey(language, "log.login.export.time"),
"I1": i18n.TKey(language, "log.login.export.msg"),
}
// 从第二行开始的数据
dataCells := make([]map[string]any, 0)
for i, row := range rows {
idx := strconv.Itoa(i + 2)
// 状态
statusValue := i18n.TKey(language, "dictData.fail")
if row.Status == "1" {
statusValue = i18n.TKey(language, "dictData.success")
}
dataCells = append(dataCells, map[string]any{
"A" + idx: row.LoginID,
"B" + idx: row.UserName,
"C" + idx: row.IPAddr,
"D" + idx: row.LoginLocation,
"E" + idx: row.OS,
"F" + idx: row.Browser,
"G" + idx: statusValue,
"H" + idx: date.ParseDateToStr(row.LoginTime, date.YYYY_MM_DDTHH_MM_SSZ),
"I" + idx: row.Msg,
})
}
// 导出数据表格
if err := file.WriterFileExecl(headerCells, dataCells, filePath, ""); err != nil {
logger.Errorf("export login log err => %v", err.Error())
return "export err"
}
}
// 上传到FTP服务器
if err := s.backupService.FTPPushFile(filePath, "log"); err != nil {
return "ok, ftp err:" + err.Error()
}
return "ok"
}

273
src/modules/crontask/processor/backup_export_udm/backup_export_udm.go Normal file
View File

@@ -0,0 +1,273 @@
package backup_export_udm
import (
"encoding/json"
"fmt"
"path/filepath"
"runtime"
"time"
"be.ems/src/framework/cron"
"be.ems/src/framework/logger"
"be.ems/src/framework/utils/file"
neDataModel "be.ems/src/modules/network_data/model"
neDataService "be.ems/src/modules/network_data/service"
neModel "be.ems/src/modules/network_element/model"
neService "be.ems/src/modules/network_element/service"
)
var NewProcessor = &BackupExportUDMProcessor{
count: 0,
neInfoService: neService.NewNeInfo,
backupService: neDataService.NewBackup,
udmAuthService: neDataService.NewUDMAuthUser,
udmSubService: neDataService.NewUDMSubUser,
udmVOIPService: neDataService.NewUDMVOIPUser,
udmVolteIMSService: neDataService.NewUDMVolteIMSUser,
}
// BackupExportUDM 队列任务处理
type BackupExportUDMProcessor struct {
count int // 执行次数
neInfoService *neService.NeInfo // 网元信息服务
backupService *neDataService.Backup // 备份相关服务
udmAuthService *neDataService.UDMAuthUser // UDM鉴权信息服务
udmSubService *neDataService.UDMSubUser // UDM签约信息服务
udmVOIPService *neDataService.UDMVOIPUser // UDMVOIP信息服务
udmVolteIMSService *neDataService.UDMVolteIMSUser // UDMVolteIMS信息服务
}
func (s *BackupExportUDMProcessor) Execute(data any) (any, error) {
s.count++ // 执行次数加一
options := data.(cron.JobData)
sysJob := options.SysJob
logger.Infof("重复:%v 任务ID:%s 执行次数:%d", options.Repeat, sysJob.JobID, s.count)
// 返回结果，用于记录执行结果
result := map[string]any{
"count": s.count,
}
var params struct {
DataType []string `json:"dataType"` // 类型支持 auth/sub/voip/volte
FileType string `json:"fileType"` // 文件类型 csv/txt
}
if err := json.Unmarshal([]byte(sysJob.TargetParams), &params); err != nil {
return nil, err
}
if !(params.FileType == "csv" || params.FileType == "txt") {
return nil, fmt.Errorf("file type error, only support csv,txt")
}
neList := s.neInfoService.SelectList(neModel.NeInfo{NeType: "UDM"}, false, false)
for _, neInfo := range neList {
for _, v := range params.DataType {
switch v {
case "auth":
result[fmt.Sprintf("%s-%s", neInfo.NeId, v)] = s.exportAuth(neInfo.NeId, params.FileType)
case "sub":
result[fmt.Sprintf("%s-%s", neInfo.NeId, v)] = s.exportSub(neInfo.NeId, params.FileType)
case "voip":
result[fmt.Sprintf("%s-%s", neInfo.NeId, v)] = s.exportVOIP(neInfo.NeId, params.FileType)
case "volte":
result[fmt.Sprintf("%s-%s", neInfo.NeId, v)] = s.exportVolte(neInfo.NeId, params.FileType)
}
}
}
// 返回结果，用于记录执行结果
return result, nil
}
// exportAuth 导出鉴权用户数据
func (s BackupExportUDMProcessor) exportAuth(neId, fileType string) string {
rows := s.udmAuthService.SelectList(neDataModel.UDMAuthUser{NeId: neId})
if len(rows) <= 0 {
return "no data"
}
// 文件名
fileName := fmt.Sprintf("auth_%s_export_%s.%s", neId, time.Now().Format("20060102150405"), fileType)
filePath := filepath.Join("/usr/local/omc/backup/udm_data/auth", fileName)
if runtime.GOOS == "windows" {
filePath = fmt.Sprintf("C:%s", filePath)
}
if fileType == "csv" {
// 转换数据
data := [][]string{}
data = append(data, []string{"imsi", "ki", "algo", "amf", "opc"})
for _, v := range rows {
opc := v.Opc
if opc == "-" {
opc = ""
}
data = append(data, []string{v.IMSI, v.Ki, v.AlgoIndex, v.Amf, opc})
}
// 输出到文件
if err := file.WriterFileCSV(data, filePath); err != nil {
return err.Error()
}
}
if fileType == "txt" {
// 转换数据
data := [][]string{}
for _, v := range rows {
opc := v.Opc
if opc == "-" {
opc = ""
}
data = append(data, []string{v.IMSI, v.Ki, v.AlgoIndex, v.Amf, opc})
}
// 输出到文件
if err := file.WriterFileTXT(data, ",", filePath); err != nil {
return err.Error()
}
}
// 上传到FTP服务器
if err := s.backupService.FTPPushFile(filePath, "udm_data"); err != nil {
return "ok, ftp err:" + err.Error()
}
return "ok"
}
// exportSub 导出签约用户数据
func (s BackupExportUDMProcessor) exportSub(neId, fileType string) string {
rows := s.udmSubService.SelectList(neDataModel.UDMSubUser{NeId: neId})
if len(rows) <= 0 {
return "no data"
}
// 文件名
fileName := fmt.Sprintf("sub_%s_export_%s.%s", neId, time.Now().Format("20060102150405"), fileType)
filePath := filepath.Join("/usr/local/omc/backup/udm_data/sub", fileName)
if runtime.GOOS == "windows" {
filePath = fmt.Sprintf("C:%s", filePath)
}
if fileType == "csv" {
// 转换数据
data := [][]string{}
data = append(data, []string{"IMSI", "MSISDN", "UeAmbrTpl", "NssaiTpl", "AreaForbiddenTpl", "ServiceAreaRestrictionTpl", "RatRestrictions", "CnTypeRestrictions", "SmfSel", "SmData", "EPSDat"})
for _, v := range rows {
epsDat := fmt.Sprintf("%s,%s,%s,%s,%s,%s,%s,%s", v.EpsFlag, v.EpsOdb, v.HplmnOdb, v.Ard, v.Epstpl, v.ContextId, v.ApnContext, v.StaticIp)
data = append(data, []string{v.IMSI, v.MSISDN, v.UeAmbrTpl, v.NssaiTpl, v.AreaForbiddenTpl, v.ServiceAreaRestrictionTpl, v.RatRestrictions, v.CnTypeRestrictions, v.SmfSel, v.SmData, epsDat})
}
// 输出到文件
if err := file.WriterFileCSV(data, filePath); err != nil {
return err.Error()
}
}
if fileType == "txt" {
// 转换数据
data := [][]string{}
for _, v := range rows {
epsDat := fmt.Sprintf("%s,%s,%s,%s,%s,%s,%s,%s", v.EpsFlag, v.EpsOdb, v.HplmnOdb, v.Ard, v.Epstpl, v.ContextId, v.ApnContext, v.StaticIp)
data = append(data, []string{v.IMSI, v.MSISDN, v.UeAmbrTpl, v.NssaiTpl, v.AreaForbiddenTpl, v.ServiceAreaRestrictionTpl, v.RatRestrictions, v.CnTypeRestrictions, v.SmfSel, v.SmData, epsDat})
}
// 输出到文件
if err := file.WriterFileTXT(data, ",", filePath); err != nil {
return err.Error()
}
}
// 上传到FTP服务器
if err := s.backupService.FTPPushFile(filePath, "udm_data"); err != nil {
return "ok, ftp err:" + err.Error()
}
return "ok"
}
// exportVOIP 导出VOIP用户数据
func (s BackupExportUDMProcessor) exportVOIP(neId, fileType string) string {
rows := s.udmVOIPService.Find(neDataModel.UDMVOIPUser{NeId: neId})
if len(rows) <= 0 {
return "no data"
}
// 文件名
fileName := fmt.Sprintf("voip_%s_export_%s.%s", neId, time.Now().Format("20060102150405"), fileType)
filePath := filepath.Join("/usr/local/omc/backup/udm_data/voip", fileName)
if runtime.GOOS == "windows" {
filePath = fmt.Sprintf("C:%s", filePath)
}
if fileType == "csv" {
// 转换数据
data := [][]string{}
data = append(data, []string{"username", "password"})
for _, v := range rows {
data = append(data, []string{v.UserName, v.Password})
}
// 输出到文件
if err := file.WriterFileCSV(data, filePath); err != nil {
return err.Error()
}
}
if fileType == "txt" {
// 转换数据
data := [][]string{}
for _, v := range rows {
data = append(data, []string{v.UserName, v.Password})
}
// 输出到文件
if err := file.WriterFileTXT(data, ",", filePath); err != nil {
return err.Error()
}
}
// 上传到FTP服务器
if err := s.backupService.FTPPushFile(filePath, "udm_data"); err != nil {
return "ok, ftp err:" + err.Error()
}
return "ok"
}
// exportVolte 导出Volte用户数据
func (s BackupExportUDMProcessor) exportVolte(neId, fileType string) string {
rows := s.udmVolteIMSService.Find(neDataModel.UDMVolteIMSUser{NeId: neId})
if len(rows) <= 0 {
return "no data"
}
// 文件名
fileName := fmt.Sprintf("volte_%s_export_%s.%s", neId, time.Now().Format("20060102150405"), fileType)
filePath := filepath.Join("/usr/local/omc/backup/udm_data/volte", fileName)
if runtime.GOOS == "windows" {
filePath = fmt.Sprintf("C:%s", filePath)
}
if fileType == "csv" {
// 转换数据
data := [][]string{}
data = append(data, []string{"IMSI", "MSISDN", "TAG", "VNI"})
for _, v := range rows {
data = append(data, []string{v.IMSI, v.MSISDN, v.Tag, v.VNI})
}
// 输出到文件
if err := file.WriterFileCSV(data, filePath); err != nil {
return err.Error()
}
}
if fileType == "txt" {
// 转换数据
data := [][]string{}
for _, v := range rows {
data = append(data, []string{v.IMSI, v.MSISDN, v.Tag, v.VNI})
}
// 输出到文件
if err := file.WriterFileTXT(data, ",", filePath); err != nil {
return err.Error()
}
}
// 上传到FTP服务器
if err := s.backupService.FTPPushFile(filePath, "udm_data"); err != nil {
return "ok, ftp err:" + err.Error()
}
return "ok"
}

113
src/modules/crontask/processor/backup_remove_file/backup_remove_file.go Normal file
View File

@@ -0,0 +1,113 @@
package backup_remove_file
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"runtime"
"sort"
"strings"
"time"
"be.ems/src/framework/cron"
"be.ems/src/framework/logger"
)
type FileInfo struct {
Path string
Info os.FileInfo
}
var NewProcessor = &BackupRemoveFileProcessor{
count: 0,
}
// BackupRemoveFileProcessor 删除备份文件
type BackupRemoveFileProcessor struct {
count int // 执行次数
}
func (s *BackupRemoveFileProcessor) Execute(data any) (any, error) {
s.count++ // 执行次数加一
options := data.(cron.JobData)
sysJob := options.SysJob
logger.Infof("重复:%v 任务ID:%s 执行次数:%d", options.Repeat, sysJob.JobID, s.count)
// 返回结果，用于记录执行结果
result := map[string]any{
"count": s.count,
}
// 读取参数值
var params []struct {
BackupPath string `json:"backupPath"` // 备份路径 /usr/local/omc/backup/{backupPath}
StoreDays int `json:"storeDays"` // 保留天数
StoreNum int `json:"storeNum"` // 保留数量，默认保留7
}
err := json.Unmarshal([]byte(sysJob.TargetParams), &params)
if err != nil {
return nil, fmt.Errorf("json params err: %v", err)
}
for _, item := range params {
result[item.BackupPath] = ""
if item.StoreDays < 0 {
result[item.BackupPath] = "params storeDays less than 0"
continue
}
if item.StoreNum <= 0 {
item.StoreNum = 7
}
// 构建完整备份路径
filePath := filepath.Join("/usr/local/omc/backup", item.BackupPath)
if runtime.GOOS == "windows" {
filePath = fmt.Sprintf("C:%s", filePath)
}
// 获取目录下所有备份文件
files, err := s.files(filePath)
if err != nil {
result[item.BackupPath] = "read files err"
continue
}
// 按修改时间排序（从旧到新）
sort.Slice(files, func(i, j int) bool {
return files[i].Info.ModTime().Before(files[j].Info.ModTime())
})
// 如果文件数量少于保留数量，则不删除
if len(files) <= item.StoreNum {
result[item.BackupPath] = fmt.Sprintf("less StoreNum: %d, file number %d", item.StoreNum, len(files))
continue
}
// 计算截止日期
cutoff := time.Now().AddDate(0, 0, -item.StoreDays)
// 删除超过保留天数的文件
deletedErr := []string{}
for _, file := range files {
if file.Info.ModTime().Before(cutoff) {
if err := os.Remove(file.Path); err != nil {
deletedErr = append(deletedErr, file.Info.Name()) // 记录删除失败的文件名称
}
}
}
result[item.BackupPath] = strings.Join(deletedErr, ", ")
}
// 返回结果，用于记录执行结果
return result, nil
}
func (s *BackupRemoveFileProcessor) files(dir string) ([]FileInfo, error) {
var files []FileInfo
err := filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
files = append(files, FileInfo{Path: path, Info: info})
}
return nil
})
return files, err
}

172
src/modules/crontask/processor/ne_alarm_state_check/ne_alarm_state_check.go Normal file
View File

@@ -0,0 +1,172 @@
package ne_alarm_state_check
import (
"encoding/json"
"fmt"
"time"
"be.ems/src/framework/constants"
"be.ems/src/framework/cron"
"be.ems/src/framework/logger"
"be.ems/src/framework/utils/parse"
neDataModel "be.ems/src/modules/network_data/model"
neDataService "be.ems/src/modules/network_data/service"
neModel "be.ems/src/modules/network_element/model"
neService "be.ems/src/modules/network_element/service"
wsService "be.ems/src/modules/ws/service"
)
var NewProcessor = &NeAlarmStateCheckProcessor{
neConfigBackupService: neService.NewNeConfigBackup,
neInfoService: neService.NewNeInfo,
alarmService: neDataService.NewAlarm,
wsSendService: wsService.NewWSSend,
count: 0,
}
// NeAlarmStateCheckProcessor 网元告警状态检查
type NeAlarmStateCheckProcessor struct {
neConfigBackupService *neService.NeConfigBackup // 网元配置文件备份记录服务
neInfoService *neService.NeInfo // 网元信息服务
alarmService *neDataService.Alarm // 告警信息服务
wsSendService *wsService.WSSend // ws发送服务
count int // 执行次数
}
// alarmParams 告警参数
type alarmParams struct {
AlarmTitle string `json:"alarmTitle"` // NE State Check Alarm
AlarmType string `json:"alarmType"` // EquipmentAlarm=2
OrigSeverity string `json:"origSeverity"` // Major=2
SpecificProblem string `json:"specificProblem"` // alarm cause: the system state of target NE has not been received
SpecificProblemID string `json:"specificProblemId"` // AC10000
AddInfo string `json:"addInfo"`
// === 非参数字段 ===
AlarmId string // 告警ID
}
func (s *NeAlarmStateCheckProcessor) Execute(data any) (any, error) {
s.count++ // 执行次数加一
options := data.(cron.JobData)
sysJob := options.SysJob
logger.Infof("重复:%v 任务ID:%s 执行次数:%d", options.Repeat, sysJob.JobID, s.count)
// 返回结果，用于记录执行结果
result := map[string]any{
"count": s.count,
}
// 读取参数值
var params alarmParams
err := json.Unmarshal([]byte(sysJob.TargetParams), &params)
if err != nil {
return nil, fmt.Errorf("json params err: %v", err)
}
neList := s.neInfoService.SelectList(neModel.NeInfo{}, true, false)
for _, neInfo := range neList {
if neInfo.CreateTime == 0 {
continue
}
neTypeAndId := fmt.Sprintf("%s_%s", neInfo.NeType, neInfo.NeId)
// 网元在线状态
isOnline := parse.Boolean(neInfo.ServerState["online"])
// 告警ID
params.AlarmId = fmt.Sprintf("%d%d", constants.ALARM_STATE_CHECK, neInfo.CreateTime)
// 检查网元告警ID是否唯一
alarmIdArr := s.alarmService.Find(neDataModel.Alarm{
NeType: neInfo.NeType,
NeId: neInfo.NeId,
AlarmId: params.AlarmId,
})
// 告警状态
alarmStatus := ""
if len(alarmIdArr) > 0 {
alarmStatus = fmt.Sprint(alarmIdArr[0].AlarmStatus)
}
// 在线且状态为活动告警
if isOnline && alarmStatus == "1" {
// 进行清除
clearAlarm, err := s.alarmClear(neInfo, alarmIdArr[0])
if err != nil {
result[neTypeAndId] = err.Error()
continue
}
groupID := fmt.Sprintf("%s_%s_%s", wsService.GROUP_ALARM, neInfo.NeType, neInfo.NeId)
s.wsSendService.ByGroupID(groupID, clearAlarm)
result[neTypeAndId] = "alarm clear"
}
// 不在线
if !isOnline && alarmStatus == "" {
// 进行新增
newAlarm, err := s.alarmNew(neInfo, params)
if err != nil {
result[neTypeAndId] = err.Error()
continue
}
groupID := fmt.Sprintf("%s_%s_%s", wsService.GROUP_ALARM, neInfo.NeType, neInfo.NeId)
s.wsSendService.ByGroupID(groupID, newAlarm)
result[neTypeAndId] = "alarm new"
}
}
// 返回结果，用于记录执行结果
return result, nil
}
// alarmClear 清除告警
func (s NeAlarmStateCheckProcessor) alarmClear(neInfo neModel.NeInfo, v neDataModel.Alarm) (neDataModel.Alarm, error) {
// 变更告警ID为告警清除ID
v.AlarmId = fmt.Sprintf("%s%d", v.AlarmCode, v.EventTime.UnixMilli())
v.AlarmStatus = "0"
// 告警清除
v.ClearType = "1"
v.ClearTime = time.Unix(neInfo.UpdateTime, 0)
v.ClearUser = "system"
rows := s.alarmService.Update(v)
if rows > 0 {
return v, nil
}
return neDataModel.Alarm{}, fmt.Errorf("clear alarm fail")
}
// alarmNew 新增告警
func (s NeAlarmStateCheckProcessor) alarmNew(neInfo neModel.NeInfo, v alarmParams) (neDataModel.Alarm, error) {
// seq 告警序号
lastSeq := s.alarmService.FindAlarmSeqLast(neInfo.NeType, neInfo.NeId)
lastTime := neInfo.UpdateTime // 网元最后更新时间
if lastTime < neInfo.CreateTime {
lastTime = time.Now().UnixMilli()
}
alarm := neDataModel.Alarm{
NeType: neInfo.NeType,
NeId: neInfo.NeId,
NeName: neInfo.NeName,
Province: neInfo.Province,
PvFlag: neInfo.PvFlag,
AlarmSeq: fmt.Sprint(lastSeq + 1),
AlarmId: v.AlarmId,
AlarmTitle: v.AlarmTitle,
AlarmCode: fmt.Sprint(constants.ALARM_STATE_CHECK),
EventTime: time.Unix(lastTime, 0),
AlarmType: v.AlarmType,
OrigSeverity: v.OrigSeverity,
PerceivedSeverity: v.OrigSeverity,
ObjectUid: neInfo.RmUID,
ObjectName: "NE State",
ObjectType: "state",
LocationInfo: "NE State: Heartbeat",
AlarmStatus: "1", // 活动告警
SpecificProblem: v.SpecificProblem,
SpecificProblemId: v.SpecificProblemID,
AddInfo: v.AddInfo,
}
insertId := s.alarmService.InsertAndForword(alarm)
if insertId != "" {
alarm.ID = insertId
return alarm, nil
}
return neDataModel.Alarm{}, fmt.Errorf("new alarm fail")
}

288
src/modules/crontask/processor/ne_alarm_state_check_cmd/ne_alarm_state_check_cmd.go Normal file
View File

@@ -0,0 +1,288 @@
package ne_alarm_state_check_cmd
import (
"encoding/json"
"fmt"
"sort"
"strconv"
"strings"
"time"
"be.ems/src/framework/constants"
"be.ems/src/framework/cron"
"be.ems/src/framework/logger"
"be.ems/src/framework/utils/parse"
neDataModel "be.ems/src/modules/network_data/model"
neDataService "be.ems/src/modules/network_data/service"
neModel "be.ems/src/modules/network_element/model"
neService "be.ems/src/modules/network_element/service"
wsService "be.ems/src/modules/ws/service"
)
var NewProcessor = &NeAlarmStateCheckCMDProcessor{
neConfigBackupService: neService.NewNeConfigBackup,
neInfoService: neService.NewNeInfo,
neStateService: neDataService.NewNEState,
alarmService: neDataService.NewAlarm,
wsSendService: wsService.NewWSSend,
count: 0,
}
// NeAlarmStateCheckCMDProcessor 网元告警内存/CPU/磁盘检查
type NeAlarmStateCheckCMDProcessor struct {
neConfigBackupService *neService.NeConfigBackup // 网元配置文件备份记录服务
neInfoService *neService.NeInfo // 网元信息服务
neStateService *neDataService.NEState // 网元状态信息服务
alarmService *neDataService.Alarm // 告警信息服务
wsSendService *wsService.WSSend // ws发送服务
count int // 执行次数
}
// alarmParams 告警参数
type alarmParams struct {
AlarmTitle string `json:"alarmTitle"` // NE State Check Alarm CPU/Menory/Disk
AlarmType string `json:"alarmType"` // EquipmentAlarm=2
OrigSeverity string `json:"origSeverity"` // Major=2
SpecificProblem string `json:"specificProblem"` // Alarm Cause: CPU/Menory/Disk status received from target NE reaches the threshold
SpecificProblemID string `json:"specificProblemId"` // AC10100
AddInfo string `json:"addInfo"` // 告警补充信息
CPUUseGt int64 `json:"cpuUseGt"` // CPU使用率大于， 范围0~100%
MemUseGt int64 `json:"memUseGt"` // 内存使用率大于， 范围0~100%
DiskUseGt int64 `json:"diskUseGt"` // 磁盘使用率大于， 范围0~100%
// === 非参数字段 ===
AlarmId string // 告警ID
}
func (s *NeAlarmStateCheckCMDProcessor) Execute(data any) (any, error) {
s.count++ // 执行次数加一
options := data.(cron.JobData)
sysJob := options.SysJob
logger.Infof("重复:%v 任务ID:%s 执行次数:%d", options.Repeat, sysJob.JobID, s.count)
// 返回结果，用于记录执行结果
result := map[string]any{
"count": s.count,
}
// 读取参数值
var params alarmParams
err := json.Unmarshal([]byte(sysJob.TargetParams), &params)
if err != nil {
return nil, fmt.Errorf("json params err: %v", err)
}
// 检查使用率
if params.CPUUseGt > 100 || params.CPUUseGt < 0 {
return nil, fmt.Errorf("cpuUseGt must be between 0 and 100")
}
if params.MemUseGt > 100 || params.MemUseGt < 0 {
return nil, fmt.Errorf("memUseGt must be between 0 and 100")
}
if params.DiskUseGt > 100 || params.DiskUseGt < 0 {
return nil, fmt.Errorf("diskUseGt must be between 0 and 100")
}
neList := s.neInfoService.SelectList(neModel.NeInfo{}, true, false)
for _, neInfo := range neList {
if neInfo.CreateTime == 0 {
continue
}
// 网元在线状态
isOnline := parse.Boolean(neInfo.ServerState["online"])
if !isOnline {
continue
}
// 检查状态
err := s.serverState(neInfo.ServerState, params.CPUUseGt, params.MemUseGt, params.DiskUseGt)
if err == nil {
continue
}
neTypeAndId := fmt.Sprintf("%s_%s", neInfo.NeType, neInfo.NeId)
// 告警ID
params.AlarmId = fmt.Sprintf("%d%d", constants.ALARM_CMD_CHECK, neInfo.CreateTime)
// 检查网元告警ID是否唯一
alarmIdArr := s.alarmService.Find(neDataModel.Alarm{
NeType: neInfo.NeType,
NeId: neInfo.NeId,
AlarmId: params.AlarmId,
})
// 告警状态, 存在的需要手动清除
alarmStatus := ""
if len(alarmIdArr) > 0 {
alarmStatus = fmt.Sprint(alarmIdArr[0].AlarmStatus)
}
// 活动告警进行清除
if alarmStatus == "1" {
clearAlarm, err := s.alarmClear(neInfo, alarmIdArr[0])
if err != nil {
result[neTypeAndId] = err.Error()
continue
}
groupID := fmt.Sprintf("%s_%s_%s", wsService.GROUP_ALARM, neInfo.NeType, neInfo.NeId)
s.wsSendService.ByGroupID(groupID, clearAlarm)
result[neTypeAndId] = "alarm clear"
alarmStatus = "" // 标记为未记录再次发起新告警
}
// 未记录
if alarmStatus == "" {
addInfo := params.AddInfo
if params.AddInfo != "" {
params.AddInfo = params.AddInfo + ", " + err.Error()
} else {
params.AddInfo = err.Error()
}
// 进行新增
newAlarm, err := s.alarmNew(neInfo, params)
params.AddInfo = addInfo // 恢复附加信息
if err != nil {
result[neTypeAndId] = err.Error()
continue
}
groupID := fmt.Sprintf("%s_%s_%s", wsService.GROUP_ALARM, neInfo.NeType, neInfo.NeId)
s.wsSendService.ByGroupID(groupID, newAlarm)
result[neTypeAndId] = "alarm new"
}
}
// 返回结果，用于记录执行结果
return result, nil
}
// serverState 网元状态
func (s NeAlarmStateCheckCMDProcessor) serverState(state map[string]any, cpuUseGt, memUseGt, diskUseGt int64) error {
// 网元CPU使用率
var nfCpuUsage float64 = 0
// CPU使用率
var sysCpuUsage float64 = 0
if state["cpu"] != nil {
cpu := state["cpu"].(map[string]any)
v := parse.Number(cpu["sysCpuUsage"])
sysCpuUsage = float64(v) / 100
nfv := parse.Number(cpu["nfCpuUsage"])
nfCpuUsage = float64(nfv) / 100
}
// 网元内存使用KB
var nfMemUsed int64 = 0
// 内存使用率
var sysMemUsage float64 = 0
if state["mem"] != nil {
mem := state["mem"].(map[string]any)
v := parse.Number(mem["sysMemUsage"])
sysMemUsage = float64(v) / 100
nfMemUsed = parse.Number(mem["nfUsedMem"])
}
// 磁盘使用率
var sysDiskUsage float64 = 0
if state["disk"] != nil {
mem := state["disk"].(map[string]any)
disks := mem["partitionInfo"].([]any)
sort.Slice(disks, func(i, j int) bool {
iUsed := parse.Number(disks[i].(map[string]any)["used"])
jUsed := parse.Number(disks[j].(map[string]any)["used"])
return iUsed > jUsed
})
disk := disks[0].(map[string]any)
total := parse.Number(disk["total"])
used := parse.Number(disk["used"])
sysDiskUsage = (float64(used) / float64(total)) * 100
sysDiskUsage, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", sysDiskUsage), 64)
}
// 插入网元状态记录
neState := neDataModel.NEState{
NeType: fmt.Sprint(state["neType"]),
NeId: fmt.Sprint(state["neId"]),
Version: fmt.Sprint(state["version"]),
Capability: parse.Number(state["capability"]),
SerialNum: fmt.Sprint(state["sn"]),
ExpiryDate: fmt.Sprint(state["expire"]),
SysCpuUsage: sysCpuUsage,
SysMemUsage: sysMemUsage,
SysDiskUsage: sysDiskUsage,
NfCpuUsage: nfCpuUsage,
NfMemUsed: nfMemUsed,
CreateTime: parse.Number(state["refreshTime"]),
}
s.neStateService.Insert(neState)
// 删除网元状态记录7天前
s.neStateService.DeleteByTime(time.Now().UnixMilli() - 7*24*60*60*1000)
// 推送ws消息
groupID := fmt.Sprintf("%s_%s_%s", wsService.GROUP_NE_STATE, neState.NeType, neState.NeId)
s.wsSendService.ByGroupID(groupID, neState)
// 检查CPU/Menory/Disk使用率
warnMsg := []string{}
if int64(sysCpuUsage) >= cpuUseGt {
warnMsg = append(warnMsg, fmt.Sprintf("cpu usage %.2f%%", sysCpuUsage))
}
if int64(sysMemUsage) >= memUseGt {
warnMsg = append(warnMsg, fmt.Sprintf("memory usage %.2f%%", sysMemUsage))
}
if int64(sysDiskUsage) >= diskUseGt {
warnMsg = append(warnMsg, fmt.Sprintf("disk usage %.2f%%", sysDiskUsage))
}
if len(warnMsg) > 0 {
return fmt.Errorf("greater than %s", strings.Join(warnMsg, ", "))
}
return nil
}
// alarmClear 清除告警
func (s NeAlarmStateCheckCMDProcessor) alarmClear(neInfo neModel.NeInfo, v neDataModel.Alarm) (neDataModel.Alarm, error) {
// 变更告警ID为告警清除ID
v.AlarmId = fmt.Sprintf("%s%d", v.AlarmCode, v.EventTime.UnixMilli())
v.AlarmStatus = "0"
// 告警清除
v.ClearType = "1"
v.ClearTime = time.Unix(neInfo.UpdateTime, 0)
v.ClearUser = "system"
rows := s.alarmService.Update(v)
if rows > 0 {
return v, nil
}
return neDataModel.Alarm{}, fmt.Errorf("clear alarm fail")
}
// alarmNew 新增告警
func (s NeAlarmStateCheckCMDProcessor) alarmNew(neInfo neModel.NeInfo, v alarmParams) (neDataModel.Alarm, error) {
// seq 告警序号
lastSeq := s.alarmService.FindAlarmSeqLast(neInfo.NeType, neInfo.NeId)
lastTime := neInfo.UpdateTime // 网元最后更新时间
if lastTime < neInfo.CreateTime {
lastTime = time.Now().UnixMilli()
}
alarm := neDataModel.Alarm{
NeType: neInfo.NeType,
NeId: neInfo.NeId,
NeName: neInfo.NeName,
Province: neInfo.Province,
PvFlag: neInfo.PvFlag,
AlarmSeq: fmt.Sprint(lastSeq + 1),
AlarmId: v.AlarmId,
AlarmTitle: v.AlarmTitle,
AlarmCode: fmt.Sprint(constants.ALARM_CMD_CHECK),
EventTime: time.Unix(lastTime, 0),
AlarmType: v.AlarmType,
OrigSeverity: v.OrigSeverity,
PerceivedSeverity: v.OrigSeverity,
ObjectUid: neInfo.RmUID,
ObjectName: "NE CPU/Menory/Disk",
ObjectType: "cmd",
LocationInfo: "NE CPU/Menory/Disk: Heartbeat",
AlarmStatus: "1", // 活动告警
SpecificProblem: v.SpecificProblem,
SpecificProblemId: v.SpecificProblemID,
AddInfo: v.AddInfo,
}
insertId := s.alarmService.InsertAndForword(alarm)
if insertId != "" {
alarm.ID = insertId
return alarm, nil
}
return neDataModel.Alarm{}, fmt.Errorf("new alarm fail")
}

213
src/modules/crontask/processor/ne_alarm_state_check_license/ne_alarm_state_check_license.go Normal file
View File

@@ -0,0 +1,213 @@
package ne_alarm_state_check_license
import (
"encoding/json"
"fmt"
"time"
"be.ems/src/framework/constants"
"be.ems/src/framework/cron"
"be.ems/src/framework/logger"
"be.ems/src/framework/utils/parse"
neDataModel "be.ems/src/modules/network_data/model"
neDataService "be.ems/src/modules/network_data/service"
neModel "be.ems/src/modules/network_element/model"
neService "be.ems/src/modules/network_element/service"
wsService "be.ems/src/modules/ws/service"
)
var NewProcessor = &NeAlarmStateCheckLicenseProcessor{
neConfigBackupService: neService.NewNeConfigBackup,
neInfoService: neService.NewNeInfo,
alarmService: neDataService.NewAlarm,
wsSendService: wsService.NewWSSend,
count: 0,
}
// NeAlarmStateCheckLicenseProcessor 网元告警License到期检查
type NeAlarmStateCheckLicenseProcessor struct {
neConfigBackupService *neService.NeConfigBackup // 网元配置文件备份记录服务
neInfoService *neService.NeInfo // 网元信息服务
alarmService *neDataService.Alarm // 告警信息服务
wsSendService *wsService.WSSend // ws发送服务
count int // 执行次数
}
// alarmParams 告警参数
type alarmParams struct {
AlarmTitle string `json:"alarmTitle"` // NE State Check Alarm License
AlarmType string `json:"alarmType"` // EquipmentAlarm=2
OrigSeverity string `json:"origSeverity"` // Major=2
SpecificProblem string `json:"specificProblem"` // Alarm Cause: License received from target NE is about to expire
SpecificProblemID string `json:"specificProblemId"` // AC10200
AddInfo string `json:"addInfo"` // 告警补充信息
DayLt int64 `json:"dayLt"` // 天数小于，默认30天
// === 非参数字段 ===
AlarmId string // 告警ID
}
func (s *NeAlarmStateCheckLicenseProcessor) Execute(data any) (any, error) {
s.count++ // 执行次数加一
options := data.(cron.JobData)
sysJob := options.SysJob
logger.Infof("重复:%v 任务ID:%s 执行次数:%d", options.Repeat, sysJob.JobID, s.count)
// 返回结果，用于记录执行结果
result := map[string]any{
"count": s.count,
}
// 读取参数值
var params alarmParams
err := json.Unmarshal([]byte(sysJob.TargetParams), &params)
if err != nil {
return nil, fmt.Errorf("json params err: %v", err)
}
// 检查参数值
if params.DayLt == 0 {
params.DayLt = 30
}
neList := s.neInfoService.SelectList(neModel.NeInfo{}, true, false)
for _, neInfo := range neList {
if neInfo.CreateTime == 0 {
continue
}
// 网元在线状态
isOnline := parse.Boolean(neInfo.ServerState["online"])
if !isOnline {
continue
}
// 检查状态
err := s.serverState(neInfo.ServerState, params.DayLt)
if err == nil {
continue
}
if params.AddInfo != "" {
params.AddInfo = params.AddInfo + ", " + err.Error()
} else {
params.AddInfo = err.Error()
}
neTypeAndId := fmt.Sprintf("%s_%s", neInfo.NeType, neInfo.NeId)
// 告警ID
params.AlarmId = fmt.Sprintf("%d%d", constants.ALARM_LICENSE_CHECK, neInfo.CreateTime)
// 检查网元告警ID是否唯一
alarmIdArr := s.alarmService.Find(neDataModel.Alarm{
NeType: neInfo.NeType,
NeId: neInfo.NeId,
AlarmId: params.AlarmId,
})
// 告警状态, 存在的需要手动清除
alarmStatus := ""
if len(alarmIdArr) > 0 {
alarmStatus = fmt.Sprint(alarmIdArr[0].AlarmStatus)
}
// 活动告警进行清除
if alarmStatus == "1" {
clearAlarm, err := s.alarmClear(neInfo, alarmIdArr[0])
if err != nil {
result[neTypeAndId] = err.Error()
continue
}
groupID := fmt.Sprintf("%s_%s_%s", wsService.GROUP_ALARM, neInfo.NeType, neInfo.NeId)
s.wsSendService.ByGroupID(groupID, clearAlarm)
result[neTypeAndId] = "alarm clear"
alarmStatus = "" // 标记为未记录再次发起新告警
}
// 未记录
if alarmStatus == "" {
// 进行新增
newAlarm, err := s.alarmNew(neInfo, params)
if err != nil {
result[neTypeAndId] = err.Error()
continue
}
groupID := fmt.Sprintf("%s_%s_%s", wsService.GROUP_ALARM, neInfo.NeType, neInfo.NeId)
s.wsSendService.ByGroupID(groupID, newAlarm)
result[neTypeAndId] = "alarm new"
}
}
// 返回结果，用于记录执行结果
return result, nil
}
// serverState 网元状态
func (s NeAlarmStateCheckLicenseProcessor) serverState(state map[string]any, dayLt int64) error {
expire := fmt.Sprint(state["expire"])
if expire == "" || expire == "<nil>" || expire == "2099-12-31" {
return nil
}
// 解析过期时间
expireTime, err := time.Parse("2006-01-02", expire)
if err != nil {
return fmt.Errorf("parse expire time error: %v", err)
}
// 计算距离天数
daysLeft := int64(time.Since(expireTime).Hours() / 24)
if daysLeft < dayLt {
return fmt.Errorf("license will expire in %d days", daysLeft)
}
return nil
}
// alarmClear 清除告警
func (s NeAlarmStateCheckLicenseProcessor) alarmClear(neInfo neModel.NeInfo, v neDataModel.Alarm) (neDataModel.Alarm, error) {
// 变更告警ID为告警清除ID
v.AlarmId = fmt.Sprintf("%s%d", v.AlarmCode, v.EventTime.UnixMilli())
v.AlarmStatus = "0"
// 告警清除
v.ClearType = "1"
v.ClearTime = time.Unix(neInfo.UpdateTime, 0)
v.ClearUser = "system"
rows := s.alarmService.Update(v)
if rows > 0 {
return v, nil
}
return neDataModel.Alarm{}, fmt.Errorf("clear alarm fail")
}
// alarmNew 新增告警
func (s NeAlarmStateCheckLicenseProcessor) alarmNew(neInfo neModel.NeInfo, v alarmParams) (neDataModel.Alarm, error) {
// seq 告警序号
lastSeq := s.alarmService.FindAlarmSeqLast(neInfo.NeType, neInfo.NeId)
lastTime := neInfo.UpdateTime // 网元最后更新时间
if lastTime < neInfo.CreateTime {
lastTime = time.Now().UnixMilli()
}
alarm := neDataModel.Alarm{
NeType: neInfo.NeType,
NeId: neInfo.NeId,
NeName: neInfo.NeName,
Province: neInfo.Province,
PvFlag: neInfo.PvFlag,
AlarmSeq: fmt.Sprint(lastSeq + 1),
AlarmId: v.AlarmId,
AlarmTitle: v.AlarmTitle,
AlarmCode: fmt.Sprint(constants.ALARM_LICENSE_CHECK),
EventTime: time.Unix(lastTime, 0),
AlarmType: v.AlarmType,
OrigSeverity: v.OrigSeverity,
PerceivedSeverity: v.OrigSeverity,
ObjectUid: neInfo.RmUID,
ObjectName: "NE License",
ObjectType: "license",
LocationInfo: "NE License: Heartbeat",
AlarmStatus: "1", // 活动告警
SpecificProblem: v.SpecificProblem,
SpecificProblemId: v.SpecificProblemID,
AddInfo: v.AddInfo,
}
insertId := s.alarmService.InsertAndForword(alarm)
if insertId != "" {
alarm.ID = insertId
return alarm, nil
}
return neDataModel.Alarm{}, fmt.Errorf("new alarm fail")
}

22
src/modules/crontask/processor/processor.go
View File

@@ -3,13 +3,19 @@ package processor
import (
"be.ems/src/framework/cron"
"be.ems/src/modules/crontask/processor/backupEtcFromNE"
processorBackupExportCDR "be.ems/src/modules/crontask/processor/backup_export_cdr"
processorBackupExportLog "be.ems/src/modules/crontask/processor/backup_export_log"
processorBackupRemoveFile "be.ems/src/modules/crontask/processor/backup_remove_file"
"be.ems/src/modules/crontask/processor/delExpiredNeBackup"
"be.ems/src/modules/crontask/processor/deleteExpiredRecord"
"be.ems/src/modules/crontask/processor/exportUEData"
"be.ems/src/modules/crontask/processor/exportTable"
"be.ems/src/modules/crontask/processor/exportUEData"
"be.ems/src/modules/crontask/processor/genNeStateAlarm"
"be.ems/src/modules/crontask/processor/getStateFromNE"
processorMonitorSysResource "be.ems/src/modules/crontask/processor/monitor_sys_resource"
processorNeAlarmStateCheck "be.ems/src/modules/crontask/processor/ne_alarm_state_check"
processorNeAlarmStateCheckCMD "be.ems/src/modules/crontask/processor/ne_alarm_state_check_cmd"
processorNeAlarmStateCheckLicense "be.ems/src/modules/crontask/processor/ne_alarm_state_check_license"
processorNeConfigBackup "be.ems/src/modules/crontask/processor/ne_config_backup"
processorNeDataUDM "be.ems/src/modules/crontask/processor/ne_data_udm"
"be.ems/src/modules/crontask/processor/removeFile"
@@ -32,4 +38,18 @@ func InitCronQueue() {
cron.CreateQueue("exportTable", exportTable.NewProcessor)
cron.CreateQueue("removeFile", removeFile.NewProcessor)
cron.CreateQueue("exportUEData", exportUEData.NewProcessor)
// 网元告警-状态检查
cron.CreateQueue("ne_alarm_state_check", processorNeAlarmStateCheck.NewProcessor)
// 网元告警-内存/CPU/磁盘检查
cron.CreateQueue("ne_alarm_state_check_cmd", processorNeAlarmStateCheckCMD.NewProcessor)
// 网元告警-License到期检查
cron.CreateQueue("ne_alarm_state_check_license", processorNeAlarmStateCheckLicense.NewProcessor)
// 备份-删除备份目录下文件
cron.CreateQueue("backup_remove_file", processorBackupRemoveFile.NewProcessor)
// 备份-导出CDR数据
cron.CreateQueue("backup_export_cdr", processorBackupExportCDR.NewProcessor)
// 备份-导出Log数据
cron.CreateQueue("backup_export_log", processorBackupExportLog.NewProcessor)
}

2
src/modules/network_data/model/cdr_event_sgwc.go
View File

@@ -13,6 +13,8 @@ type CDREventSGWC struct {
CreatedAt time.Time `json:"createdAt" gorm:"column:created_at;default:CURRENT_TIMESTAMP"`
// ====== 非数据库字段属性 ======
TenantID string `json:"tenantID" gorm:"-"`
TenantName string `json:"tenantName" gorm:"-"`
}
// TableName 表名称