Alarm information
Sequence: %d
-NE name: %s
+NE Name: %s
Title: %s
Severity: %s
Event Time: %s
-Alarm Status: %d
+Alarm Status: %s
Automatic sent by OMC, please do not reply!
- `, alarmData.AlarmSeq, alarmData.NeName, alarmData.AlarmTitle, alarmData.OrigSeverity, alarmData.EventTime, alarmData.AlarmStatus) + `, + alarmData.AlarmSeq, + alarmData.NeName, + alarmData.AlarmTitle, + severity, + alarmData.EventTime, + alarmStatus, + ) // message := fmt.Sprintf(` // Alarm information @@ -70,7 +90,7 @@ func AlarmEmailForward(alarmData *Alarm) error { //m.SetHeader("To", "zhangshuzhong@agrandtech.com", "simonzhangsz@outlook.com") // 收件人，可以多个收件人，但必须使用相同的 SMTP 连接 //m.SetHeader("Cc", "******@qq.com") // 抄送，可以多个 //m.SetHeader("Bcc", "******@qq.com") // 暗送，可以多个 - m.SetHeader("Subject", "Alarm from OMC!") // 邮件主题 + m.SetHeader("Subject", subjectTime) // 邮件主题 // text/html 的意思是将文件的 content-type 设置为 text/html 的形式，浏览器在获取到这种文件时会自动调用html的解析器对文件进行相应的处理。 // 可以通过 text/html 处理文本格式进行特殊处理，如换行、缩进、加粗等等 diff --git a/features/pm/kpi_c_report/controller.go b/features/pm/kpi_c_report/controller.go index e546217c..1a3c5d8f 100644 --- a/features/pm/kpi_c_report/controller.go +++ b/features/pm/kpi_c_report/controller.go @@ -103,7 +103,7 @@ func (k *KpiCReport) GetReport2FE(c *gin.Context) { dbg := db.DB("").Model(&KpiCReport{}).Table(tableName) if querys.NeID != "" { - conditions = append(conditions, "rm_uid = (select n.rm_uid from ne_info n where n.ne_type=? and n.ne_id=? and n.status=1)") + conditions = append(conditions, "rm_uid = (select n.rm_uid from ne_info n where n.ne_type=? and n.ne_id=?)") params = append(params, querys.NeType, querys.NeID) } else { c.JSON(http.StatusBadRequest, services.ErrResp("Not found required parameter NE ID")) diff --git a/features/pm/performance.go b/features/pm/performance.go index bd690c0e..213021f4 100644 --- a/features/pm/performance.go +++ b/features/pm/performance.go @@ -199,7 +199,10 @@ func saveKPIData(kpiReport KpiReport, index int64) int64 { // 推送到ws订阅组 wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s%s_%s", wsService.GROUP_KPI, neInfo.NeType, neInfo.NeId), kpiEvent) if neInfo.NeType == "UPF" { - wsService.NewWSSend.ByGroupID(wsService.GROUP_KPI_UPF+neInfo.NeId, kpiEvent) + // 更新UPF总流量 + upValue := parse.Number(kpiEvent["UPF.03"]) + downValue := parse.Number(kpiEvent["UPF.06"]) + neDataService.NewKpiReport.UPFTodayFlowUpdate(neInfo.RmUID, upValue, downValue, false) } } } @@ -274,7 +277,7 @@ func saveKPIDataC(kpiReport KpiReport, index int64) int64 { neInfo := neService.NewNeInfo.FindByRmuid(kpiData.RmUid) if neInfo.RmUID == kpiData.RmUid { // 推送自定义KPI到ws订阅组 - wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s%s_%s", wsService.GROUP_KPI_C, neInfo.NeType, neInfo.NeId), kpiEvent) + wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s_%s_%s", wsService.GROUP_KPI_C, neInfo.NeType, neInfo.NeId), kpiEvent) } } return insertId @@ -412,12 +415,9 @@ func PostKPIReportFromNFOld(w http.ResponseWriter, r *http.Request) { if neInfo.RmUID == kpiData.RmUid { // 推送到ws订阅组 - wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s%s_%s", wsService.GROUP_KPI, neInfo.NeType, neInfo.NeId), kpiEvent) + wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s_%s_%s", wsService.GROUP_KPI, neInfo.NeType, neInfo.NeId), kpiEvent) // 推送自定义KPI到ws订阅组 - wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s%s_%s", wsService.GROUP_KPI_C, neInfo.NeType, neInfo.NeId), kpiCEvent) - if neInfo.NeType == "UPF" { - wsService.NewWSSend.ByGroupID(wsService.GROUP_KPI_UPF+neInfo.NeId, kpiEvent) - } + wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s_%s_%s", wsService.GROUP_KPI_C, neInfo.NeType, neInfo.NeId), kpiCEvent) } services.ResponseStatusOK204NoContent(w) @@ -465,7 +465,7 @@ type Task struct { } type KpiSetJ struct { Code string `json:"Code"` // 统计编码 如：SMFHA01 - KPIs []string `json:"KPIs` // 指标项集合 ["SMF.AttCreatePduSession", "SMF.AttCreatePduSession._Dnn"] + KPIs []string `json:"KPIs"` // 指标项集合 ["SMF.AttCreatePduSession", "SMF.AttCreatePduSession._Dnn"] } type MeasureTask struct { @@ -475,7 +475,7 @@ type MeasureTask struct { KpiSet []KpiSetJ `json:"KPISet" xorm:"kpi_set"` StartTime string `json:"startTime" xorm:"start_time"` EndTime string `json:"endTime" xorm:"end_time"` - Periods []Period `json:"Periods" xorm:"periods` + Periods []Period `json:"Periods" xorm:"periods"` Schedule []ScheduleJ `json:"Schedule" xorm:"schedule"` GranulOption string `json:"granulOption" xorm:"granul_option"` Status string `json:"status" xorm:"status"` @@ -483,7 +483,7 @@ type MeasureTask struct { Comment string `json:"comment" xorm:"comment"` CreateTime string `json:"createTime" xorm:"create_time"` UpdateTime string `json:"updateTime" xorm:"update_time"` - DeleteTime string `json:"deleteTime xorm:"delete_time"` + DeleteTime string `json:"deleteTime" xorm:"delete_time"` Tasks []Task `json:"Tasks"` NotifyUrl string `json:"NotifyUrl"` /* "http://xEngine.xEngine.xEngine.x:xxxx/api/rest/performanceManagement/v1/elementType/smf/objectType/measureReport */ diff --git a/src/framework/database/redis/expand.go b/src/framework/database/redis/expand.go index 8eba226e..85a8076a 100644 --- a/src/framework/database/redis/expand.go +++ b/src/framework/database/redis/expand.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "sync" + "time" "be.ems/src/framework/logger" "github.com/redis/go-redis/v9" @@ -137,3 +138,38 @@ func SetHash(source, key string, value map[string]any) error { } return nil } + +// IncrBy 累加统计数据 +func IncrBy(source, key, field string, value int64) error { + // 数据源 + rdb := RDB(source) + if rdb == nil { + return fmt.Errorf("redis not client") + } + + // 使用HINCRBY命令进行累加统计 + ctx := context.Background() + err := rdb.HIncrBy(ctx, key, field, value).Err() + if err != nil { + logger.Errorf("redis HIncrBy err %v", err) + return err + } + return nil +} + +// Expire 过期时间设置 +func Expire(source, key string, expiration time.Duration) error { + // 数据源 + rdb := RDB(source) + if rdb == nil { + return fmt.Errorf("redis not client") + } + // 过期时间设置 + ctx := context.Background() + err := rdb.Expire(ctx, key, expiration).Err() + if err != nil { + logger.Errorf("redis HIncrBy err %v", err) + return err + } + return nil +} diff --git a/src/modules/network_data/controller/upf.go b/src/modules/network_data/controller/upf.go index 7cd329b2..72e90efa 100644 --- a/src/modules/network_data/controller/upf.go +++ b/src/modules/network_data/controller/upf.go @@ -57,6 +57,6 @@ func (s UPFController) TotalFlow(c *gin.Context) { return } - data := s.kpiReportService.FindUPFTotalFlow(neInfo.RmUID, querys.Day) - c.JSON(200, resp.OkData(data)) + up, down := s.kpiReportService.UPFTodayFlowFind(neInfo.RmUID, querys.Day) + c.JSON(200, resp.OkData(map[string]int64{"up": up, "down": down})) } diff --git a/src/modules/network_data/network_data.go b/src/modules/network_data/network_data.go index 43a3840b..7c419b4b 100644 --- a/src/modules/network_data/network_data.go +++ b/src/modules/network_data/network_data.go @@ -6,6 +6,7 @@ import ( "be.ems/src/framework/middleware/collectlogs" "be.ems/src/framework/middleware/repeat" "be.ems/src/modules/network_data/controller" + "be.ems/src/modules/network_data/service" "github.com/gin-gonic/gin" ) @@ -14,6 +15,9 @@ import ( func Setup(router *gin.Engine) { logger.Infof("开始加载 ====> network_data 模块路由") + // 启动时需要的初始参数 + InitLoad() + neDataGroup := router.Group("/neData") // 性能统计信息 @@ -349,3 +353,9 @@ func Setup(router *gin.Engine) { ) } } + +// InitLoad 初始参数 +func InitLoad() { + // 启动时，加载UPF上下行流量 + go service.NewKpiReport.UPFTodayFlowLoad() +} diff --git a/src/modules/network_data/repository/udm_auth.go b/src/modules/network_data/repository/udm_auth.go index 93694628..e8cdb888 100644 --- a/src/modules/network_data/repository/udm_auth.go +++ b/src/modules/network_data/repository/udm_auth.go @@ -42,7 +42,6 @@ func (r *UDMAuthUser) SelectPage(query map[string]string) (int64, []model.UDMAut // 查询数量 长度为0直接返回 if err := tx.Count(&total).Error; err != nil || total <= 0 { - logger.Errorf("total err => %v", err) return total, rows } diff --git a/src/modules/network_data/repository/udm_sub.go b/src/modules/network_data/repository/udm_sub.go index 8b56a87d..1ec6a71e 100644 --- a/src/modules/network_data/repository/udm_sub.go +++ b/src/modules/network_data/repository/udm_sub.go @@ -45,7 +45,6 @@ func (r *UDMSubUser) SelectPage(query map[string]string) (int64, []model.UDMSubU // 查询数量 长度为0直接返回 if err := tx.Count(&total).Error; err != nil || total <= 0 { - logger.Errorf("total err => %v", err) return total, rows } diff --git a/src/modules/network_data/service/kpi_report.go b/src/modules/network_data/service/kpi_report.go index 2bd3f922..53134adc 100644 --- a/src/modules/network_data/service/kpi_report.go +++ b/src/modules/network_data/service/kpi_report.go @@ -11,6 +11,7 @@ import ( "be.ems/src/framework/utils/parse" "be.ems/src/modules/network_data/model" "be.ems/src/modules/network_data/repository" + neModel "be.ems/src/modules/network_element/model" ) // 实例化数据层 KpiReport 结构体 @@ -140,55 +141,111 @@ func (r KpiReport) FindTitle(neType string) []model.KpiTitle { return r.kpiReportRepository.SelectKPITitle(neType) } -// FindUPFTotalFlow 查询UPF总流量 N3上行 N6下行 -func (r KpiReport) FindUPFTotalFlow(rmUID string, day int) map[string]int64 { - now := time.Now() +// UPFTodayFlowFind 查询UPF总流量 N3上行 N6下行 +// down * 8 / 1000 / 1000 单位M +func (r KpiReport) UPFTodayFlowFind(rmUID string, day int) (int64, int64) { // 获取当前日期 - endTime := now.UnixMilli() - // 将当前日期前几天数 - beginTime := now.AddDate(0, 0, -day).Truncate(24 * time.Hour).UnixMilli() + now := time.Now() + var upTotal, downTotal int64 - data := map[string]int64{ - "up": 0, - "down": 0, - } + // 查询最近7天的数据 + for i := 0; i <= day; i++ { + dateKey := now.AddDate(0, 0, -i).Format("2006-01-02") + key := fmt.Sprintf("%s:UPF_FLOW:%s:%s", constants.CACHE_NE_DATA, rmUID, dateKey) - // 读取缓存数据 小于2分钟重新缓存 - key := fmt.Sprintf("%s:UPF_FLOW:%s_%d", constants.CACHE_NE_DATA, rmUID, day) - if infoStr, err := redis.Get("", key); err == nil && infoStr != "" { - json.Unmarshal([]byte(infoStr), &data) - if expireSecond, _ := redis.GetExpire("", key); expireSecond > 120 { - return data + // 读取缓存数据 + up, err := redis.GetHash("", key, "up") + if err != nil || up == "" { + up = "0" } + down, err := redis.GetHash("", key, "down") + if err != nil || down == "" { + down = "0" + } + + upTotal += parse.Number(up) + downTotal += parse.Number(down) } - // down * 8 / 1000 / 1000 单位M - rows := r.kpiReportRepository.SelectUPF(rmUID, beginTime, endTime) - for _, row := range rows { - // 解析 JSON 字符串为 map - var kpiValues []map[string]any - err := json.Unmarshal([]byte(row.KpiValues), &kpiValues) + return upTotal, downTotal +} + +// UPFTodayFlow UPF流量今日统计 +func (r KpiReport) UPFTodayFlowUpdate(rmUID string, upValue, downValue int64, rest bool) error { + // 按日期存储统计数据 + dateKey := time.Now().Format("2006-01-02") + key := fmt.Sprintf("%s:UPF_FLOW:%s:%s", constants.CACHE_NE_DATA, rmUID, dateKey) + + // 重置数据 + if rest { + err := redis.SetHash("", key, map[string]any{ + "up": upValue, + "down": downValue, + }) if err != nil { - continue + return err } + // 设置key的过期时间为30天，自动清理旧数据 + err = redis.Expire("", key, 30*24*time.Hour) + if err != nil { + return err + } + } - // 遍历 kpiValues 数组 - for _, v := range kpiValues { - if k, ok := v["kpiId"]; ok { - if k == "UPF.03" { - data["up"] = data["up"] + parse.Number(v["value"]) + // 使用HIncrBy实时累加统计值 + if err := redis.IncrBy("", key, "up", upValue); err != nil { + return err + } + if err := redis.IncrBy("", key, "down", downValue); err != nil { + return err + } + return nil +} + +// UPFTodayFlowLoad UPF上下行数据到redis +func (r KpiReport) UPFTodayFlowLoad() { + cacheKeys, _ := redis.GetKeys("", constants.CACHE_NE_INFO+":UPF:*") + if len(cacheKeys) == 0 { + return + } + + // 今日流量 + now := time.Now() + beginTime := now.Truncate(24 * time.Hour).UnixMilli() + endTime := beginTime + 24*60*60*1000 - 1 + + for _, key := range cacheKeys { + var v neModel.NeInfo + jsonStr, _ := redis.Get("", key) + if len(jsonStr) > 7 { + json.Unmarshal([]byte(jsonStr), &v) + } + if v.NeType == "UPF" && v.RmUID != "" { + // 查询历史数据 + rows := r.kpiReportRepository.SelectUPF(v.RmUID, beginTime, endTime) + var upTotal, downTotal int64 + + // 处理历史数据 + for _, row := range rows { + var kpiValues []map[string]any + if err := json.Unmarshal([]byte(row.KpiValues), &kpiValues); err != nil { + continue } - if k == "UPF.06" { - data["down"] = data["down"] + parse.Number(v["value"]) + + for _, v := range kpiValues { + if k, ok := v["kpiId"]; ok { + if k == "UPF.03" { + upTotal += parse.Number(v["value"]) + } + if k == "UPF.06" { + downTotal += parse.Number(v["value"]) + } + } } } + + // 将历史数据添加到Redis + r.UPFTodayFlowUpdate(v.RmUID, upTotal, downTotal, true) } } - - // 保存到缓存 - if infoJSON, err := json.Marshal(data); err == nil { - redis.SetByExpire("", key, string(infoJSON), time.Duration(10)*time.Minute) - } - - return data } diff --git a/src/modules/network_element/service/ne_info.go b/src/modules/network_element/service/ne_info.go index 4cc00721..13437225 100644 --- a/src/modules/network_element/service/ne_info.go +++ b/src/modules/network_element/service/ne_info.go @@ -86,12 +86,17 @@ func (r NeInfo) ClearNeCacheByNeType(neType string) bool { func (r NeInfo) FindByNeType(neType string) []model.NeInfo { neInfo := make([]model.NeInfo, 0) key := fmt.Sprintf("%s:%s:*", constants.CACHE_NE_INFO, strings.ToUpper(neType)) - jsonStr, _ := redis.Get("", key) - if len(jsonStr) > 7 { - err := json.Unmarshal([]byte(jsonStr), &neInfo) - if err != nil { - return neInfo + cacheKeys, _ := redis.GetKeys("", key) + if len(cacheKeys) > 0 { + for _, key := range cacheKeys { + var v model.NeInfo + jsonStr, _ := redis.Get("", key) + if len(jsonStr) > 7 { + json.Unmarshal([]byte(jsonStr), &v) + } + neInfo = append(neInfo, v) } + return neInfo } else { neInfo = r.neInfoRepository.SelectList(model.NeInfo{NeType: neType}) for _, v := range neInfo { diff --git a/src/modules/system/repository/sys_menu.go b/src/modules/system/repository/sys_menu.go index 48972b98..48959ad1 100644 --- a/src/modules/system/repository/sys_menu.go +++ b/src/modules/system/repository/sys_menu.go @@ -68,7 +68,7 @@ func (r SysMenu) SelectByIds(menuIds []int64) []model.SysMenu { // Insert 新增信息 func (r SysMenu) Insert(sysMenu model.SysMenu) int64 { sysMenu.DelFlag = "0" - if sysMenu.MenuId <= 0 { + if sysMenu.MenuId > 0 { return 0 } if sysMenu.Icon == "" { diff --git a/src/modules/trace/packet_task/packet.go b/src/modules/trace/packet_task/packet.go index 5df723b3..420b5a39 100644 --- a/src/modules/trace/packet_task/packet.go +++ b/src/modules/trace/packet_task/packet.go @@ -141,7 +141,7 @@ func capturePacketSource(taskInfo *task) { frameMeta := parsePacketFrame(frameNumber, frameTime, packet) // 推送到ws订阅组 - wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s%s", wsService.GROUP_TRACE_PACKET, taskInfo.TaskNo), frameMeta) + wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s_%s", wsService.GROUP_TRACE_PACKET, taskInfo.TaskNo), frameMeta) } } } diff --git a/src/modules/trace/service/trace_task.go b/src/modules/trace/service/trace_task.go index d5259ab1..03948e2e 100644 --- a/src/modules/trace/service/trace_task.go +++ b/src/modules/trace/service/trace_task.go @@ -101,7 +101,7 @@ func (r *TraceTask) CreateUDP() error { // 推送文件 if v, ok := mData["pcapFile"]; ok && v != "" { logger.Infof("pcapFile: %s", v) - wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s%d", wsService.GROUP_TRACE_NE, taskId), taskId) + wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s_%d", wsService.GROUP_TRACE_NE, taskId), taskId) } // 发送响应 @@ -160,7 +160,7 @@ func (r *TraceTask) CreateUDP() error { // 推送文件 if v, ok := mData["pcapFile"]; ok && v != "" { logger.Infof("pcapFile: %s", v) - wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s%d", wsService.GROUP_TRACE_NE, taskId), taskId) + wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s_%d", wsService.GROUP_TRACE_NE, taskId), taskId) } // 发送响应 diff --git a/src/modules/ws/processor/upf_total_flow.go b/src/modules/ws/processor/upf_total_flow.go index a0397014..7d83ba79 100644 --- a/src/modules/ws/processor/upf_total_flow.go +++ b/src/modules/ws/processor/upf_total_flow.go @@ -34,11 +34,11 @@ func GetUPFTotalFlow(requestID string, data any) ([]byte, error) { return nil, fmt.Errorf("no matching network element information found") } - dataMap := neDataService.NewKpiReport.FindUPFTotalFlow(neInfo.RmUID, querys.Day) + up, down := neDataService.NewKpiReport.UPFTodayFlowFind(neInfo.RmUID, querys.Day) resultByte, err := json.Marshal(resp.Ok(map[string]any{ "requestId": requestID, - "data": dataMap, + "data": map[string]int64{"up": up, "down": down}, })) return resultByte, err } diff --git a/src/modules/ws/service/ws_send.go b/src/modules/ws/service/ws_send.go index fda2979c..3cf88541 100644 --- a/src/modules/ws/service/ws_send.go +++ b/src/modules/ws/service/ws_send.go @@ -13,23 +13,21 @@ const ( // 组号-其他 GROUP_OTHER = "0" // 组号-跟踪任务网元数据变更 2_traceId - GROUP_TRACE_NE = "2_" + GROUP_TRACE_NE = "2" // 组号-信令跟踪Packet 4_taskNo - GROUP_TRACE_PACKET = "4_" + GROUP_TRACE_PACKET = "4" // 组号-指标通用 10_neType_neId - GROUP_KPI = "10_" - // 组号-指标UPF 12_neId - GROUP_KPI_UPF = "12_" + GROUP_KPI = "10" // 组号-自定义KPI指标 20_neType_neId - GROUP_KPI_C = "20_" + GROUP_KPI_C = "20" // 组号-IMS_CDR会话事件 1005_neId - GROUP_IMS_CDR = "1005_" + GROUP_IMS_CDR = "1005" // 组号-SMF_CDR会话事件 1006_neId - GROUP_SMF_CDR = "1006_" + GROUP_SMF_CDR = "1006" // 组号-SMSC_CDR会话事件 1007_neId - GROUP_SMSC_CDR = "1007_" + GROUP_SMSC_CDR = "1007" // 组号-SGWC_CDR会话事件 1008_neId - GROUP_SGWC_CDR = "1008_" + GROUP_SGWC_CDR = "1008" // 组号-AMF_UE会话事件 1010_neId GROUP_AMF_UE = "1010" // 组号-MME_UE会话事件 1011_neId