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

Merge branch 'lichang' into lite

Browse Source
This commit is contained in:
TsMask
2025-05-13 09:51:03 +08:00
parent 641612ab59 1ccbebce22
commit a0cffccbfa
16 changed files with 198 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/framework/database/db/expand.go
View File

@@ -93,8 +93,16 @@ func processSQLFile(db *gorm.DB, filePath string) {
if strings.HasSuffix(line, ";") {
// 执行 SQL 语句
if err := db.Exec(sqlBuilder.String()).Error; err != nil {
log.Fatalln(err.Error())
return
errorStr := err.Error()
if strings.Contains(strings.ToLower(errorStr), "duplicate column") {
// 重复字段错误忽略
// log.Printf("Exec SQL: %s\n", line)
// log.Println(err.Error())
} else {
// 其他错误终止程序
log.Fatalln(errorStr)
return
}
}
sqlBuilder.Reset()

19
src/modules/crontask/processor/ne_alarm_state_check_cmd/ne_alarm_state_check_cmd.go
View File

@@ -3,6 +3,7 @@ package ne_alarm_state_check_cmd
import (
"encoding/json"
"fmt"
"runtime"
"sort"
"strconv"
"strings"
@@ -26,6 +27,8 @@ var NewProcessor = &NeAlarmStateCheckCMDProcessor{
alarmService: neDataService.NewAlarm,
wsSendService: wsService.NewWSSend,
count: 0,
triggerMax: 4,
triggerCount: 0,
}
// NeAlarmStateCheckCMDProcessor 网元告警内存/CPU/磁盘检查
@@ -36,7 +39,8 @@ type NeAlarmStateCheckCMDProcessor struct {
alarmService *neDataService.Alarm // 告警信息服务
wsSendService *wsService.WSSend // ws发送服务
count int // 执行次数
triggerMax int // 阈值连续触发次数大于才会产生告警
triggerCount int // 阈值连续触发次数
}
// alarmParams 告警参数
@@ -47,7 +51,7 @@ type alarmParams struct {
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%
CPUUseGt int64 `json:"cpuUseGt"` // CPU使用率大于， 范围0~100*CPU核心数
MemUseGt int64 `json:"memUseGt"` // 内存使用率大于， 范围0~100%
DiskUseGt int64 `json:"diskUseGt"` // 磁盘使用率大于， 范围0~100%
@@ -72,8 +76,9 @@ func (s *NeAlarmStateCheckCMDProcessor) Execute(data any) (any, error) {
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")
numCPU := runtime.NumCPU()
if params.CPUUseGt > int64(numCPU*100) || params.CPUUseGt < 0 {
return nil, fmt.Errorf("cpuUseGt must be between 0 and 100 * NumCPU")
}
if params.MemUseGt > 100 || params.MemUseGt < 0 {
return nil, fmt.Errorf("memUseGt must be between 0 and 100")
@@ -137,6 +142,7 @@ func (s *NeAlarmStateCheckCMDProcessor) Execute(data any) (any, error) {
// 进行新增
newAlarm, err := s.alarmNew(neInfo, params)
params.AddInfo = addInfo // 恢复附加信息
s.triggerCount = 0 // 重置连续触发次数
if err != nil {
result[neTypeAndId] = err.Error()
continue
@@ -227,7 +233,10 @@ func (s NeAlarmStateCheckCMDProcessor) serverState(state map[string]any, cpuUseG
warnMsg = append(warnMsg, fmt.Sprintf("disk usage %.2f%%", sysDiskUsage))
}
if len(warnMsg) > 0 {
return fmt.Errorf("greater than %s", strings.Join(warnMsg, ", "))
s.triggerCount++
if s.triggerCount > s.triggerMax {
return fmt.Errorf("greater than %s", strings.Join(warnMsg, ", "))
}
}
return nil
}

9
src/modules/crontask/processor/ne_alarm_state_check_license/ne_alarm_state_check_license.go
View File

@@ -120,8 +120,15 @@ func (s *NeAlarmStateCheckLicenseProcessor) Execute(data any) (any, error) {
}
// 未记录
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
@@ -150,7 +157,7 @@ func (s NeAlarmStateCheckLicenseProcessor) serverState(state map[string]any, day
}
// 计算距离天数
daysLeft := int64(time.Since(expireTime).Hours() / 24)
daysLeft := int64(time.Until(expireTime).Hours() / 24)
if daysLeft < dayLt {
return fmt.Errorf("license will expire in %d days", daysLeft)
}

13
src/modules/network_element/service/ne_version.go
View File

@@ -651,7 +651,18 @@ func (r NeVersion) operateRun(sshClient *ssh.ConnSSH, preinput map[string]string
}
// 命令终止符后继续执行命令
suffix := strings.HasSuffix(outputStr, "~]# ") || strings.LastIndex(outputStr, "~# ") != -1 || strings.HasSuffix(outputStr, "~$ ")
// "~]# ":麒麟, "~]$ ":欧拉, "~# ":NXP, "~$ ":Ubuntu
suffixStr := []string{"~]# ", "~]$ ", "~# ", "~$ "}
suffix := false
for _, v := range suffixStr {
if strings.HasSuffix(outputStr, v) {
suffix = true
break
}
}
if !suffix {
suffix = strings.LastIndex(outputStr, "~# ") != -1
}
if len(cmdStrArr) > 0 && suffix {
if firstRead {
firstRead = false

1
src/modules/trace/model/trace_task.go
View File

@@ -14,6 +14,7 @@ type TraceTask struct {
DstIp string `json:"dstIp" gorm:"column:dst_ip"` // 目标地址IP
CreateBy string `json:"createBy" gorm:"column:create_by"` // 创建者
CreateTime int64 `json:"createTime" gorm:"column:create_time"` // 创建时间
Title string `json:"title" gorm:"column:title" binding:"required"` // 任务标题
Remark string `json:"remark" gorm:"column:remark"` // 备注
NeList string `json:"neList" gorm:"column:ne_list" binding:"required"` // 网元列表 neType_neId 例如 UDM_001,AMF_001
NotifyUrl string `json:"notifyUrl" gorm:"column:notify_url"` // 信息数据通知回调地址UDP 例如udp:192.168.5.58:33033

17
src/modules/trace/repository/trace_task.go
View File

@@ -29,6 +29,9 @@ func (r TraceTask) SelectByPage(query map[string]string) ([]model.TraceTask, int
if v, ok := query["msisdn"]; ok && v != "" {
tx = tx.Where("msisdn like ?", fmt.Sprintf("%s%%", v))
}
if v, ok := query["title"]; ok && v != "" {
tx = tx.Where("title like ?", fmt.Sprintf("%s%%", v))
}
if v, ok := query["startTime"]; ok && v != "" {
if len(v) == 10 {
v = fmt.Sprintf("%s000", v)
@@ -112,6 +115,20 @@ func (r TraceTask) SelectByIds(ids []int64) []model.TraceTask {
return rows
}
// SelectByUnstopped 查询未停止的任务补发
func (r TraceTask) SelectByUnstopped() []model.TraceTask {
rows := []model.TraceTask{}
tx := db.DB("").Model(&model.TraceTask{})
// 构建查询条件
tx = tx.Where("end_time > ?", time.Now().UnixMilli())
// 查询数据
if err := tx.Find(&rows).Error; err != nil {
logger.Errorf("query find err => %v", err.Error())
return rows
}
return rows
}
// Insert 新增信息 返回新增数据ID
func (r TraceTask) Insert(param model.TraceTask) int64 {
if param.CreateBy != "" {

54
src/modules/trace/service/trace_task.go
View File

@@ -177,6 +177,19 @@ func (r TraceTask) FindById(id int64) model.TraceTask {
// Insert 新增信息
func (r TraceTask) Insert(task model.TraceTask) error {
if err := r.createTaskToNe(&task, false); err != nil {
return err
}
// 插入数据库
insertId := r.traceTaskRepository.Insert(task)
if insertId <= 0 {
return fmt.Errorf("insert task error")
}
return nil
}
// CreateTaskToNe 创建任务到网元
func (r TraceTask) createTaskToNe(task *model.TraceTask, ignoreErr bool) error {
// 跟踪配置是否开启
host, port, err := r.traceNotify()
if err != nil {
@@ -193,6 +206,7 @@ func (r TraceTask) Insert(task model.TraceTask) error {
task.TraceId = fmt.Sprint(traceId)
// 发送任务给网元
errNe := []string{}
for _, neTypeID := range neList {
neTypeIDArr := strings.Split(neTypeID, "_")
if len(neTypeIDArr) != 2 {
@@ -204,18 +218,28 @@ func (r TraceTask) Insert(task model.TraceTask) error {
logger.Warnf("ne type id is not exist")
continue
}
if err := r.createTaskToNe(neInfo, task); err != nil {
logger.Errorf("task to %s error: %s", neTypeID, err.Error())
return fmt.Errorf("task to %s error: %s", neTypeID, err.Error())
if err := r.traceNeTask(neInfo, *task); err != nil {
logger.Errorf("ne type id is %s to %s error: %s", task.TraceId, neTypeID, err.Error())
errNe = append(errNe, neTypeID)
continue
}
}
// 插入数据库
insertId := r.traceTaskRepository.Insert(task)
if insertId <= 0 {
return fmt.Errorf("insert task error")
if len(errNe) <= 0 && !ignoreErr {
return nil
}
return nil
// 移除任务
for _, neTypeID := range neList {
neTypeIDArr := strings.Split(neTypeID, "_")
if len(neTypeIDArr) != 2 {
continue
}
neInfo := neService.NewNeInfo.FindByNeTypeAndNeID(neTypeIDArr[0], neTypeIDArr[1])
if neInfo.NeType != neTypeIDArr[0] || neInfo.IP == "" {
continue
}
neFetchlink.NeTraceDelete(neInfo, task.TraceId)
}
return fmt.Errorf("task to ne error: %s", strings.Join(errNe, ","))
}
// traceNotify 网元通知地址
@@ -234,8 +258,8 @@ func (r TraceTask) traceNotify() (string, int64, error) {
return host, port, nil
}
// createTaskToNe 网元创建任务
func (r TraceTask) createTaskToNe(neInfo neModel.NeInfo, task model.TraceTask) error {
// traceNeTask 网元创建任务
func (r TraceTask) traceNeTask(neInfo neModel.NeInfo, task model.TraceTask) error {
data := model.TraceReq{
TraceId: parse.Number(task.TraceId),
NotifyUrl: task.NotifyUrl,
@@ -305,3 +329,11 @@ func (r TraceTask) DeleteByIds(ids []int64) (int64, error) {
// 删除信息失败！
return 0, fmt.Errorf("delete fail")
}
// RunUnstopped 启动跟踪未停止的任务
func (r TraceTask) RunUnstopped() {
tasks := r.traceTaskRepository.SelectByUnstopped()
for _, task := range tasks {
r.createTaskToNe(&task, true)
}
}

3
src/modules/trace/service/trace_task_udp_data.go
View File

@@ -27,7 +27,7 @@ func traceHandler(data []byte) (*TraceMsgToOamTraceData, error) {
if err != nil {
return decodeData, err
}
fmt.Printf("TraceHandler get oamData: %s,%+v\n, payload=len(%d,%d)", decodeData.TimestampStr, decodeData.NfTraceMsg, decodeData.TracePayloadLen, len(decodeData.TracePayload))
// fmt.Printf("TraceHandler get oamData: %s,%+v\n, payload=len(%d,%d)", decodeData.TimestampStr, decodeData.NfTraceMsg, decodeData.TracePayloadLen, len(decodeData.TracePayload))
// Return parsed message and payload
if len(decodeData.TracePayload) != int(decodeData.TracePayloadLen) {
return decodeData, fmt.Errorf("trace payload is bad, len=%d, shall be:%d", len(decodeData.TracePayload), int(decodeData.TracePayloadLen))
@@ -106,6 +106,7 @@ func decodeTraceData(data []byte) (*TraceMsgToOamTraceData, error) {
return nil, err
}
msg.SrcIpStr = net.IP(srcIp).String()
// 7. Parse DstIp (4 bytes IPv4)
dstIp := make([]byte, 4)
if flag&0x20 != 0 {