feat: 跟踪任务添加标题，任务错误移除已发任务

src/framework/database/db/expand.go

@@ -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()

src/modules/trace/model/trace_task.go

@@ -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

src/modules/trace/repository/trace_task.go

@@ -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 != "" {

src/modules/trace/service/trace_task.go

@@ -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)
+	}
+}