ref: 重构网元跟踪任务功能

2025-04-18 16:12:27 +08:00
parent 243312e5fa
commit 43965222b1
9 changed files with 585 additions and 621 deletions
--- a/src/modules/trace/controller/trace_data.go
+++ b/src/modules/trace/controller/trace_data.go
@@ -31,6 +31,24 @@ func (s *TraceDataController) List(c *gin.Context) {
 	c.JSON(200, resp.OkData(map[string]any{"rows": rows, "total": total}))
 }

+// 跟踪任务数据信息
+//
+// GET /:id
+func (s *TraceDataController) Info(c *gin.Context) {
+	id := parse.Number(c.Param("id"))
+	if id <= 0 {
+		c.JSON(400, resp.CodeMsg(40010, "bind err: id is empty"))
+		return
+	}
+
+	data := s.traceDataService.FindById(id)
+	if data.ID == id {
+		c.JSON(200, resp.OkData(data))
+		return
+	}
+	c.JSON(200, resp.Err(nil))
+}
+
 // 跟踪任务数据删除
 //
 // DELETE /:id
--- a/src/modules/trace/controller/trace_task.go
+++ b/src/modules/trace/controller/trace_task.go
@@ -80,39 +80,6 @@ func (s *TraceTaskController) Add(c *gin.Context) {
 	c.JSON(200, resp.Ok(nil))
 }

-// 跟踪任务修改
-//
-// PUT /
-func (s *TraceTaskController) Edit(c *gin.Context) {
-	language := reqctx.AcceptLanguage(c)
-	var body model.TraceTask
-	err := c.ShouldBindBodyWithJSON(&body)
-	if err != nil {
-		errMsgs := fmt.Sprintf("bind err: %s", resp.FormatBindError(err))
-		c.JSON(422, resp.CodeMsg(40422, errMsgs))
-		return
-	}
-	if body.ID == 0 {
-		c.JSON(400, resp.CodeMsg(40010, "bind err: id is empty"))
-		return
-	}
-
-	// 检查是否存在
-	taskInfo := s.traceTaskService.FindById(body.ID)
-	if taskInfo.ID != body.ID {
-		// 没有可访问任务信息数据！
-		c.JSON(200, resp.ErrMsg(i18n.TKey(language, "task.noData")))
-		return
-	}
-
-	body.UpdateBy = reqctx.LoginUserToUserName(c)
-	if err = s.traceTaskService.Update(body); err != nil {
-		c.JSON(200, resp.ErrMsg(i18n.TKey(language, err.Error())))
-		return
-	}
-	c.JSON(200, resp.Ok(nil))
-}
-
 // 跟踪任务删除
 //
 // DELETE /:id
--- a/src/modules/trace/model/trace_data.go
+++ b/src/modules/trace/model/trace_data.go
@@ -3,18 +3,19 @@ package model
 // TraceData 跟踪_数据 trace_data
 type TraceData struct {
 	ID        int64  `json:"id"  gorm:"column:id;primaryKey;autoIncrement"`
-	TaskId    int64  `json:"taskId"  gorm:"column:task_id"` // 任务ID
-	IMSI      string `json:"imsi"  gorm:"column:imsi"`
-	MSISDN    string `json:"msisdn"  gorm:"column:msisdn"`    // 可能存在
-	SrcAddr   string `json:"srcAddr"  gorm:"column:src_addr"` // 源地址带端口
-	DstAddr   string `json:"dstAddr"  gorm:"column:dst_addr"` // 目标地址带端口
-	IfType    int64  `json:"ifType"  gorm:"column:if_type"`   // 接口类型，未分类
-	MsgType   int64  `json:"msgType"  gorm:"column:msg_type"`
-	MsgDirect int64  `json:"msgDirect"  gorm:"column:msg_direct"`
-	Length    int64  `json:"length"  gorm:"column:length"`       // 去除头后的原始数据byte长度
-	Timestamp int64  `json:"timestamp"  gorm:"column:timestamp"` // 毫秒
-	RawMsg    string `json:"rawMsg"  gorm:"column:raw_msg"`      // 去除头后的原始数据byteBase64
-	DecMsg    string `json:"decMsg"  gorm:"column:dec_msg"`      // TCP内容消息
+	TraceId   int64  `json:"traceId"  gorm:"column:trace_id"`     // 跟踪任务ID
+	IMSI      string `json:"imsi"  gorm:"column:imsi"`            // UE类型 IMSI
+	MSISDN    string `json:"msisdn"  gorm:"column:msisdn"`        // UE类型 可能存在
+	SrcAddr   string `json:"srcAddr"  gorm:"column:src_addr"`     // 源地址
+	DstAddr   string `json:"dstAddr"  gorm:"column:dst_addr"`     // 目标地址
+	IfType    string `json:"ifType"  gorm:"column:if_type"`       // 接口类型，未分类
+	MsgType   int64  `json:"msgType"  gorm:"column:msg_type"`     // 消息类型，0-req, 1-rsp
+	MsgDirect int64  `json:"msgDirect"  gorm:"column:msg_direct"` // 消息方向，0-recv,1-send
+	MsgNe     string `json:"msgNe"  gorm:"column:msg_ne"`         // 消息网元
+	MsgEvent  string `json:"msgEvent"  gorm:"column:msg_event"`   // 消息事件，e.g: CreateContextReq
+	Length    int64  `json:"length"  gorm:"column:length"`        // 原始数据byte长度
+	Timestamp int64  `json:"timestamp"  gorm:"column:timestamp"`  // 纳秒时间戳
+	RawMsg    string `json:"rawMsg"  gorm:"column:raw_msg"`       // 原始数据byteBase64
 }

 // TableName 表名称
--- a/src/modules/trace/model/trace_task.go
+++ b/src/modules/trace/model/trace_task.go
@@ -2,27 +2,21 @@ package model

 // TraceTask 跟踪_任务
 type TraceTask struct {
-	ID         int64  `json:"id"  gorm:"column:id;primaryKey;autoIncrement"` // 跟踪任务ID
-	TraceId    string `json:"traceId"  gorm:"column:trace_id"`               // 任务编号
-	TraceType  string `json:"traceType"  gorm:"column:trace_type"`           // 1-Interface,2-Device,3-User
-	StartTime  int64  `json:"startTime"  gorm:"column:start_time"`           // 开始时间 毫秒
-	EndTime    int64  `json:"endTime"  gorm:"column:end_time"`               // 结束时间 毫秒
-	Interfaces string `json:"interfaces"  gorm:"column:interfaces"`          // 接口跟踪必须 例如 N8,N10
-	IMSI       string `json:"imsi"  gorm:"column:imsi"`                      // 用户跟踪必须
-	MSISDN     string `json:"msisdn"  gorm:"column:msisdn"`                  // 用户跟踪可选
-	UeIp       string `json:"ueIp"  gorm:"column:ue_ip"`                     // 设备跟踪必须 IP
-	SrcIp      string `json:"srcIp"  gorm:"column:src_ip"`                   // 源地址IP
-	DstIp      string `json:"dstIp"  gorm:"column:dst_ip"`                   // 目标地址IP
-	SignalPort int64  `json:"signalPort"  gorm:"column:signal_port"`         // 地址IP端口
-	CreateBy   string `json:"createBy"  gorm:"column:create_by"`             // 创建者
-	CreateTime int64  `json:"createTime"  gorm:"column:create_time"`         // 创建时间
-	UpdateBy   string `json:"updateBy"  gorm:"column:update_by"`             // 更新者
-	UpdateTime int64  `json:"updateTime"  gorm:"column:update_time"`         // 更新时间
-	Remark     string `json:"remark"  gorm:"column:remark"`                  // 备注
-	NeType     string `json:"neType"  gorm:"column:ne_type"`                 // 网元类型
-	NeId       string `json:"neId"  gorm:"column:ne_id"`                     // 网元ID
-	NotifyUrl  string `json:"notifyUrl"  gorm:"column:notify_url"`           // 信息数据通知回调地址UDP 例如udp:192.168.5.58:29500
-	FetchMsg   string `json:"fetchMsg"  gorm:"column:fetch_msg"`             // 任务下发请求响应消息
+	ID         int64  `json:"id"  gorm:"column:id;primaryKey;autoIncrement"`          // 跟踪任务ID
+	TraceId    string `json:"traceId"  gorm:"column:trace_id"`                        // 任务编号
+	TraceType  string `json:"traceType"  gorm:"column:trace_type" binding:"required"` // 1-Interface,2-Device,3-UE
+	StartTime  int64  `json:"startTime"  gorm:"column:start_time"`                    // 开始时间 毫秒
+	EndTime    int64  `json:"endTime"  gorm:"column:end_time" binding:"required"`     // 结束时间 毫秒
+	Interfaces string `json:"interfaces"  gorm:"column:interfaces"`                   // 接口跟踪必须 例如 N8,N10
+	IMSI       string `json:"imsi"  gorm:"column:imsi"`                               // 用户跟踪必须
+	MSISDN     string `json:"msisdn"  gorm:"column:msisdn"`                           // 用户跟踪可选
+	SrcIp      string `json:"srcIp"  gorm:"column:src_ip"`                            // 源地址IP
+	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"`                  // 创建时间
+	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
 }

 // TableName 表名称
--- a/src/modules/trace/model/trace_task_udp_msg.go
+++ b/src/modules/trace/model/trace_task_udp_msg.go
@@ -0,0 +1,42 @@
+package model
+
+type TraceDefType = int64
+
+const (
+	MSG_TYPE_REQ    TraceDefType = 0
+	MSG_TYPE_RSP    TraceDefType = 1
+	MSG_DIRECT_RECV TraceDefType = 0
+	MSG_DIRECT_SEND TraceDefType = 1
+	// NF_TYPE_AMF                  = "AMF"
+	// NF_TYPE_SMF                  = "SMF"
+	// NF_TYPE_AUSF                 = "AUSF"
+	// NF_TYPE_UDM                  = "UDM"
+	// NF_TYPE_PCF                  = "PCF"
+)
+
+// TraceMsg 跟踪_消息
+type TraceMsg struct {
+	TraceId   int
+	MsgType   TraceDefType //0-req, 1-rsp
+	MsgDirect TraceDefType //0-recv,1-send
+	NfName    string
+	MsgEvent  string //e.g: CreateContextReq
+	IMSI      string
+	IfType    string //interface type: N2,N8...
+	SrcIpStr  string
+	DstIpStr  string
+	Timestamp int64
+}
+
+// TraceReq 跟踪_请求
+type TraceReq struct {
+	TraceId    int64    `json:"traceId"`              // Unique identifier for trace operations:delete...
+	TraceType  string   `json:"traceType"`            // Type of trace: Interface;UE; Device
+	StartTime  string   `json:"startTime"`            // Start time of the trace: e.g., "2025-04-12 11:24:59"
+	EndTime    string   `json:"endTime"`              // End time of the trace: e.g., "2025-04-12 14:24:59"
+	Interfaces []string `json:"interfaces,omitempty"` // List of interfaces: e.g., ["N8", "N10", "N1/N2"]
+	IMSI       string   `json:"imsi,omitempty"`       // International Mobile Subscriber Identity
+	SrcIp      string   `json:"srcIp,omitempty"`      // Source IP address (used for device-level filtering)
+	DstIp      string   `json:"dstIp,omitempty"`      // Destination IP address (used for device-level filtering)
+	NotifyUrl  string   `json:"notifyUrl,omitempty"`  // Notification URL: e.g., "udp:192.168.5.58:33033"
+}
--- a/src/modules/trace/repository/trace_data.go
+++ b/src/modules/trace/repository/trace_data.go
@@ -2,6 +2,7 @@ package repository

 import (
 	"fmt"
+	"strings"

 	"be.ems/src/framework/database/db"
 	"be.ems/src/framework/logger"
@@ -18,11 +19,8 @@ type TraceData struct{}
 func (r TraceData) SelectByPage(query map[string]string) ([]model.TraceData, int64) {
 	tx := db.DB("").Model(&model.TraceData{})
 	// 查询条件拼接
-	if v, ok := query["imsi"]; ok && v != "" {
-		tx = tx.Where("imsi like ?", fmt.Sprintf("%s%%", v))
-	}
-	if v, ok := query["msisdn"]; ok && v != "" {
-		tx = tx.Where("msisdn like ?", fmt.Sprintf("%s%%", v))
+	if v, ok := query["traceId"]; ok && v != "" {
+		tx = tx.Where("trace_id = ?", v)
 	}
 	if v, ok := query["startTime"]; ok && v != "" {
 		if len(v) == 10 {
@@ -53,7 +51,33 @@ func (r TraceData) SelectByPage(query map[string]string) ([]model.TraceData, int
 	// 查询数据分页
 	pageNum, pageSize := db.PageNumSize(query["pageNum"], query["pageSize"])
 	tx = tx.Limit(pageSize).Offset(pageSize * pageNum)
-	err := tx.Find(&rows).Error
+	// 排序
+	sortByStr, sortOk := query["sortBy"]
+	sortOrderStr, orderOk := query["sortOrder"]
+	if sortOk && sortByStr != "" && orderOk && sortOrderStr != "" {
+		sortByArr := strings.Split(sortByStr, ",")
+		sortOrderArr := strings.Split(sortOrderStr, ",")
+		for i := range sortByArr {
+			sortBy := sortByArr[i]
+			sortOrder := sortOrderArr[i]
+			// 排序字段
+			sort := "id"
+			if sortBy == "timestamp" {
+				sort = "timestamp"
+			}
+			// 排序方式
+			order := "ASC"
+			if strings.HasPrefix(sortOrder, "asc") {
+				order = "ASC"
+			} else if strings.HasPrefix(sortOrder, "desc") {
+				order = "DESC"
+			}
+			tx = tx.Order(fmt.Sprintf("%s %s", sort, order))
+		}
+	} else {
+		tx = tx.Order("id desc")
+	}
+	err := tx.Omit("raw_msg").Find(&rows).Error
 	if err != nil {
 		logger.Errorf("query find err => %v", err.Error())
 		return rows, total
@@ -100,3 +124,16 @@ func (r TraceData) DeleteByIds(ids []int64) int64 {
 	}
 	return tx.RowsAffected
 }
+
+// DeleteByTraceId 删除跟踪任务数据 返回受影响行数
+func (r TraceData) DeleteByTraceId(traceId string) int64 {
+	if traceId == "" || traceId == "<nil>" {
+		return 0
+	}
+	tx := db.DB("").Where("trace_id = ?", traceId)
+	if err := tx.Delete(&model.TraceData{}).Error; err != nil {
+		logger.Errorf("delete err => %v", err.Error())
+		return 0
+	}
+	return tx.RowsAffected
+}
--- a/src/modules/trace/repository/trace_task.go
+++ b/src/modules/trace/repository/trace_task.go
@@ -2,6 +2,7 @@ package repository

 import (
 	"fmt"
+	"strings"
 	"time"

 	"be.ems/src/framework/database/db"
@@ -19,8 +20,8 @@ type TraceTask struct{}
 func (r TraceTask) SelectByPage(query map[string]string) ([]model.TraceTask, int64) {
 	tx := db.DB("").Model(&model.TraceTask{})
 	// 查询条件拼接
-	if v, ok := query["neType"]; ok && v != "" {
-		tx = tx.Where("ne_type = ?", v)
+	if v, ok := query["traceType"]; ok && v != "" {
+		tx = tx.Where("trace_type = ?", v)
 	}
 	if v, ok := query["imsi"]; ok && v != "" {
 		tx = tx.Where("imsi like ?", fmt.Sprintf("%s%%", v))
@@ -29,10 +30,20 @@ func (r TraceTask) SelectByPage(query map[string]string) ([]model.TraceTask, int
 		tx = tx.Where("msisdn like ?", fmt.Sprintf("%s%%", v))
 	}
 	if v, ok := query["startTime"]; ok && v != "" {
-		tx = tx.Where("start_time >= ?", v)
+		if len(v) == 10 {
+			v = fmt.Sprintf("%s000", v)
+			tx = tx.Where("start_time >= ?", v)
+		} else if len(v) == 13 {
+			tx = tx.Where("start_time >= ?", v)
+		}
 	}
 	if v, ok := query["endTime"]; ok && v != "" {
-		tx = tx.Where("end_time >= ?", v)
+		if len(v) == 10 {
+			v = fmt.Sprintf("%s999", v)
+			tx = tx.Where("end_time <= ?", v)
+		} else if len(v) == 13 {
+			tx = tx.Where("end_time <= ?", v)
+		}
 	}

 	// 查询结果
@@ -47,18 +58,33 @@ func (r TraceTask) SelectByPage(query map[string]string) ([]model.TraceTask, int
 	// 查询数据分页
 	pageNum, pageSize := db.PageNumSize(query["pageNum"], query["pageSize"])
 	tx = tx.Limit(pageSize).Offset(pageSize * pageNum)
-
 	// 排序
-	if v, ok := query["sortField"]; ok && v != "" {
-		sortSql := v
-		if v, ok := query["sortOrder"]; ok && v != "" {
-			if v == "desc" {
-				sortSql += " desc "
-			} else {
-				sortSql += " asc "
+	sortByStr, sortOk := query["sortBy"]
+	sortOrderStr, orderOk := query["sortOrder"]
+	if sortOk && sortByStr != "" && orderOk && sortOrderStr != "" {
+		sortByArr := strings.Split(sortByStr, ",")
+		sortOrderArr := strings.Split(sortOrderStr, ",")
+		for i := range sortByArr {
+			sortBy := sortByArr[i]
+			sortOrder := sortOrderArr[i]
+			// 排序字段
+			sort := "id"
+			if sortBy == "operaBy" {
+				sort = "opera_by"
+			} else if sortBy == "endTime" {
+				sort = "end_time"
 			}
+			// 排序方式
+			order := "ASC"
+			if strings.HasPrefix(sortOrder, "asc") {
+				order = "ASC"
+			} else if strings.HasPrefix(sortOrder, "desc") {
+				order = "DESC"
+			}
+			tx = tx.Order(fmt.Sprintf("%s %s", sort, order))
 		}
-		tx.Order(sortSql)
+	} else {
+		tx = tx.Order("id desc")
 	}

 	err := tx.Find(&rows).Error
@@ -90,8 +116,6 @@ func (r TraceTask) SelectByIds(ids []int64) []model.TraceTask {
 func (r TraceTask) Insert(param model.TraceTask) int64 {
 	if param.CreateBy != "" {
 		ms := time.Now().UnixMilli()
-		param.UpdateBy = param.CreateBy
-		param.UpdateTime = ms
 		param.CreateTime = ms
 	}
 	// 执行插入
@@ -102,26 +126,6 @@ func (r TraceTask) Insert(param model.TraceTask) int64 {
 	return param.ID
 }

-// Update 修改信息 返回受影响行数
-func (r TraceTask) Update(param model.TraceTask) int64 {
-	if param.ID <= 0 {
-		return 0
-	}
-	if param.UpdateBy != "" {
-		param.UpdateTime = time.Now().UnixMilli()
-	}
-	tx := db.DB("").Model(&model.TraceTask{})
-	// 构建查询条件
-	tx = tx.Where("id = ?", param.ID)
-	tx = tx.Omit("id", "create_by", "create_time")
-	// 执行更新
-	if err := tx.Updates(param).Error; err != nil {
-		logger.Errorf("update err => %v", err.Error())
-		return 0
-	}
-	return tx.RowsAffected
-}
-
 // DeleteByIds 批量删除信息 返回受影响行数
 func (r TraceTask) DeleteByIds(ids []int64) int64 {
 	if len(ids) <= 0 {
--- a/src/modules/trace/service/trace_task.go
+++ b/src/modules/trace/service/trace_task.go
@@ -2,7 +2,6 @@ package service

 import (
 	"encoding/base64"
-	"encoding/json"
 	"fmt"
 	"net"
 	"strings"
@@ -13,6 +12,7 @@ import (
 	"be.ems/src/framework/utils/date"
 	"be.ems/src/framework/utils/parse"
 	neFetchlink "be.ems/src/modules/network_element/fetch_link"
+	neModel "be.ems/src/modules/network_element/model"
 	neService "be.ems/src/modules/network_element/service"
 	"be.ems/src/modules/trace/model"
 	"be.ems/src/modules/trace/repository"
@@ -37,20 +37,11 @@ type TraceTask struct {
 }

 // CreateUDP 创建UDP数据通道
-func (r *TraceTask) CreateUDP() error {
+func (r TraceTask) CreateUDP() error {
 	// 跟踪配置是否开启
-	if v := config.Get("trace.enabled"); v != nil {
-		if !v.(bool) {
-			return nil
-		}
-	}
-	host := "127.0.0.1"
-	if v := config.Get("trace.host"); v != nil {
-		host = v.(string)
-	}
-	var port int64 = 33033
-	if v := config.Get("trace.port"); v != nil {
-		port = parse.Number(v)
+	host, port, err := r.traceNotify()
+	if err != nil {
+		return err
 	}

 	// 初始化UDP服务
@@ -67,122 +58,116 @@ func (r *TraceTask) CreateUDP() error {
 		}

 		// 读取数据
-		buf := make([]byte, 2048)
+		buf := make([]byte, 10*1048)
 		n, addr, err := conn.ReadFromUDPAddrPort(buf)
 		if err != nil {
 			logger.Errorf("UDP Resolve ReadFromUDPAddrPort Error: %s", err.Error())
 			return
 		}

-		logger.Infof("socket UDP: %s", string(buf[:n]))
-		// logger.Infof("socket UDP Base64: %s", base64.StdEncoding.EncodeToString(buf[:n]))
-		mData, err := UDPDataHandler(buf, n)
-		if err != nil {
-			logger.Errorf("UDP Resolve UDPDataHandler Error: %s", err.Error())
+		// logger.Infof("socket UDP: %s", string(buf[:n]))
+		logger.Infof("socket UDP Base64 Encode: %s", base64.StdEncoding.EncodeToString(buf[:n]))
+
+		// 解析数据
+		if err := r.pasreUDPData(buf[:n]); err != nil {
+			logger.Errorf("UDP Resolve UDPData Error: %s", err.Error())
 			return
 		}
-		taskId := parse.Number(mData["taskId"])
-
-		// 插入数据库做记录
-		r.traceDataRepository.Insert(model.TraceData{
-			TaskId:    taskId,
-			IMSI:      mData["imsi"].(string),
-			SrcAddr:   mData["srcAddr"].(string),
-			DstAddr:   mData["dstAddr"].(string),
-			IfType:    parse.Number(mData["ifType"]),
-			MsgType:   parse.Number(mData["msgType"]),
-			MsgDirect: parse.Number(mData["msgDirect"]),
-			Length:    parse.Number(mData["dataLen"]),
-			RawMsg:    mData["dataInfo"].(string),
-			Timestamp: parse.Number(mData["timestamp"]),
-			DecMsg:    mData["decMsg"].(string),
-		})
-
-		// 推送文件
-		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)
-		}

 		// 发送响应
 		if _, err := conn.WriteToUDPAddrPort([]byte("udp>"), addr); err != nil {
 			logger.Errorf("UDP Resolve WriteToUDPAddrPort Error: %s", err.Error())
 		}
+		buf = nil
 	})

-	// ============ 测试接收网元UDP发过来的数据
-	// 初始化TCP服务 后续调整TODO
-	tcpService := socket.SocketTCP{Addr: host, Port: port + 1}
-	if _, err := tcpService.New(); err != nil {
-		return err
-	}
-	// 接收处理TCP数据
-	go tcpService.Resolve(func(conn *net.Conn, err error) {
-		if err != nil {
-			logger.Errorf("TCP Resolve %s", err.Error())
-			return
+	// ============ 本地测试接收网元UDP发过来的数据 后续调整TODO
+	if config.Env() == "local" {
+		// 初始化TCP服务
+		tcpService := socket.SocketTCP{Addr: host, Port: port + 1}
+		if _, err := tcpService.New(); err != nil {
+			return err
 		}
+		// 接收处理TCP数据
+		go tcpService.Resolve(func(conn *net.Conn, err error) {
+			if err != nil {
+				logger.Errorf("TCP Resolve %s", err.Error())
+				return
+			}

-		c := (*conn)
-		// 读取数据
-		buf := make([]byte, 2048)
-		n, err := c.Read(buf)
-		if err != nil {
-			logger.Errorf("TCP Resolve Read Error: %s", err.Error())
-			return
-		}
+			c := (*conn)
+			// 读取数据
+			buf := make([]byte, 2048)
+			n, err := c.Read(buf)
+			if err != nil {
+				logger.Errorf("TCP Resolve Read Error: %s", err.Error())
+				return
+			}

-		logger.Infof("socket TCP: %s", string(buf[:n]))
-		deData, _ := base64.StdEncoding.DecodeString(string(buf[:n]))
-		logger.Infof("socket TCP Base64: %s", deData)
-		mData, err := UDPDataHandler(deData, len(deData))
-		if err != nil {
-			logger.Errorf("TCP Resolve UDPDataHandler Error: %s", err.Error())
-			return
-		}
-		taskId := parse.Number(mData["taskId"])
+			// logger.Infof("socket TCP: %s", string(buf[:n]))
+			deData, _ := base64.StdEncoding.DecodeString(string(buf[:n]))
+			// logger.Infof("socket TCP Base64 Decode: %s", deData)

-		// 插入数据库做记录
-		r.traceDataRepository.Insert(model.TraceData{
-			TaskId:    taskId,
-			IMSI:      mData["imsi"].(string),
-			SrcAddr:   mData["srcAddr"].(string),
-			DstAddr:   mData["dstAddr"].(string),
-			IfType:    parse.Number(mData["ifType"]),
-			MsgType:   parse.Number(mData["msgType"]),
-			MsgDirect: parse.Number(mData["msgDirect"]),
-			Length:    parse.Number(mData["dataLen"]),
-			RawMsg:    mData["dataInfo"].(string),
-			Timestamp: parse.Number(mData["timestamp"]),
-			DecMsg:    mData["decMsg"].(string),
+			if err := r.pasreUDPData(deData); err != nil {
+				logger.Errorf("TCP Resolve UDPData Error: %s", err.Error())
+				return
+			}
+
+			// 发送响应
+			if _, err = c.Write([]byte("tcp>")); err != nil {
+				logger.Errorf("TCP Resolve Write Error: %s", err.Error())
+			}
+			buf = nil
 		})
+	}

-		// 推送文件
-		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)
-		}
-
-		// 发送响应
-		if _, err = c.Write([]byte("tcp>")); err != nil {
-			logger.Errorf("TCP Resolve Write Error: %s", err.Error())
-		}
-	})
 	return nil
 }

+// pasreUDPData 解析数据
+func (r TraceTask) pasreUDPData(buf []byte) error {
+	data, err := traceHandler(buf)
+	if err != nil {
+		logger.Errorf("UDP Resolve UDPDataHandler Error: %s", err.Error())
+		return err
+	}
+	taskId := parse.Number(data.NfTraceMsg.TraceId)
+
+	// 插入数据库做记录
+	item := model.TraceData{
+		TraceId:   taskId,
+		IMSI:      data.NfTraceMsg.IMSI,
+		SrcAddr:   data.NfTraceMsg.SrcIpStr,
+		DstAddr:   data.NfTraceMsg.DstIpStr,
+		IfType:    data.NfTraceMsg.IfType,
+		MsgType:   data.NfTraceMsg.MsgType,
+		MsgDirect: data.NfTraceMsg.MsgDirect,
+		MsgNe:     data.NfTraceMsg.NfName,
+		MsgEvent:  data.NfTraceMsg.MsgEvent,
+		Length:    int64(data.TracePayloadLen),
+		RawMsg:    base64.StdEncoding.EncodeToString(data.TracePayload),
+		Timestamp: data.NfTraceMsg.Timestamp,
+	}
+	item.ID = r.traceDataRepository.Insert(item)
+
+	// 推送到ws订阅组
+	item.RawMsg = "" // 不推送原始数据
+	wsService.NewWSSend.ByGroupID(fmt.Sprintf("%s_%d", wsService.GROUP_TRACE_NE, taskId), item)
+	return err
+}
+
 // CloseUDP 关闭UDP数据通道
-func (r *TraceTask) CloseUDP() {
+func (r TraceTask) CloseUDP() {
 	r.udpService.Close()
 }

 // FindByPage 根据条件分页查询
-func (r *TraceTask) FindByPage(query map[string]string) ([]model.TraceTask, int64) {
+func (r TraceTask) FindByPage(query map[string]string) ([]model.TraceTask, int64) {
 	return r.traceTaskRepository.SelectByPage(query)
 }

 // FindById 通过ID查询
-func (r *TraceTask) FindById(id int64) model.TraceTask {
+func (r TraceTask) FindById(id int64) model.TraceTask {
 	tasks := r.traceTaskRepository.SelectByIds([]int64{id})
 	if len(tasks) > 0 {
 		return tasks[0]
@@ -191,12 +176,52 @@ func (r *TraceTask) FindById(id int64) model.TraceTask {
 }

 // Insert 新增信息
-func (r *TraceTask) Insert(task model.TraceTask) error {
+func (r TraceTask) Insert(task model.TraceTask) error {
 	// 跟踪配置是否开启
-	if v := config.Get("trace.enabled"); v != nil {
-		if !v.(bool) {
-			return fmt.Errorf("tracking is not enabled")
+	host, port, err := r.traceNotify()
+	if err != nil {
+		return err
+	}
+	task.NotifyUrl = fmt.Sprintf("udp:%s:%d", host, port)
+	// 网元列表
+	neList := strings.Split(task.NeList, ",")
+	if len(neList) <= 0 {
+		return fmt.Errorf("ne list is empty")
+	}
+	// 生成任务ID
+	traceId := r.traceTaskRepository.LastID() + 1 // 生成任务ID < 65535
+	task.TraceId = fmt.Sprint(traceId)
+
+	// 发送任务给网元
+	for _, neTypeID := range neList {
+		neTypeIDArr := strings.Split(neTypeID, "_")
+		if len(neTypeIDArr) != 2 {
+			logger.Warnf("ne type id is error")
+			continue
 		}
+		neInfo := neService.NewNeInfo.FindByNeTypeAndNeID(neTypeIDArr[0], neTypeIDArr[1])
+		if neInfo.NeType != neTypeIDArr[0] || neInfo.IP == "" {
+			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())
+		}
+	}
+
+	// 插入数据库
+	insertId := r.traceTaskRepository.Insert(task)
+	if insertId <= 0 {
+		return fmt.Errorf("insert task error")
+	}
+	return nil
+}
+
+// traceNotify 网元通知地址
+func (r TraceTask) traceNotify() (string, int64, error) {
+	if v := config.Get("trace.enabled"); !parse.Boolean(v) {
+		return "", 0, fmt.Errorf("trace is not enabled")
 	}
 	host := "127.0.0.1"
 	if v := config.Get("trace.host"); v != nil {
@@ -206,120 +231,43 @@ func (r *TraceTask) Insert(task model.TraceTask) error {
 	if v := config.Get("trace.port"); v != nil {
 		port = parse.Number(v)
 	}
-	task.NotifyUrl = fmt.Sprintf("udp:%s:%d", host, port)
+	return host, port, nil
+}

-	// 查询网元获取IP
-	neInfo := neService.NewNeInfo.FindByNeTypeAndNeID(task.NeType, task.NeId)
-	if neInfo.NeId != task.NeId || neInfo.IP == "" {
-		return fmt.Errorf("app.common.noNEInfo")
-	}
-	traceId := r.traceTaskRepository.LastID() + 1 // 生成任务ID < 65535
-	task.TraceId = fmt.Sprint(traceId)
-
-	// 发送任务给网元
-	data := map[string]any{
-		"neType":    neInfo.NeType,
-		"neId":      neInfo.NeId,
-		"notifyUrl": task.NotifyUrl,
-		"id":        traceId,
-		"startTime": date.ParseDateToStr(task.StartTime, date.YYYY_MM_DD_HH_MM_SS),
-		"endTime":   date.ParseDateToStr(task.EndTime, date.YYYY_MM_DD_HH_MM_SS),
+// createTaskToNe 网元创建任务
+func (r TraceTask) createTaskToNe(neInfo neModel.NeInfo, task model.TraceTask) error {
+	data := model.TraceReq{
+		TraceId:   parse.Number(task.TraceId),
+		NotifyUrl: task.NotifyUrl,
+		StartTime: date.ParseDateToStr(task.StartTime, date.YYYY_MM_DD_HH_MM_SS),
+		EndTime:   date.ParseDateToStr(task.EndTime, date.YYYY_MM_DD_HH_MM_SS),
 	}
 	switch task.TraceType {
 	case "1": // Interface
-		data["traceType"] = "Interface"
-		data["interfaces"] = strings.Split(task.Interfaces, ",")
+		data.TraceType = "Interface"
+		data.Interfaces = strings.Split(task.Interfaces, ",")
 	case "2": // Device
-		data["traceType"] = "Device"
-		data["ueIp"] = task.UeIp
-		data["srcIp"] = task.SrcIp
-		data["dstIp"] = task.DstIp
-		data["signalPort"] = task.SignalPort
-		task.UeIp = neInfo.IP
+		data.TraceType = "Device"
+		data.SrcIp = task.SrcIp
+		data.DstIp = task.DstIp
 	case "3": // UE
-		data["traceType"] = "UE"
-		data["imsi"] = task.IMSI
-		data["msisdn"] = task.MSISDN
+		data.TraceType = "UE"
+		data.IMSI = fmt.Sprintf("imsi-%s", task.IMSI)
 	default:
-		return fmt.Errorf("trace type is not disabled")
+		return fmt.Errorf("trace type is not support")
 	}
 	msg, err := neFetchlink.NeTraceAdd(neInfo, data)
 	if err != nil {
 		return err
 	}
-	s, _ := json.Marshal(msg)
-	task.FetchMsg = string(s)
-
-	// 插入数据库
-	r.traceTaskRepository.Insert(task)
-	return nil
-}
-
-// Update 修改信息
-func (r *TraceTask) Update(task model.TraceTask) error {
-	// 跟踪配置是否开启
-	if v := config.Get("trace.enabled"); v != nil {
-		if !v.(bool) {
-			return fmt.Errorf("tracking is not enabled")
-		}
+	if v, ok := msg["cause"]; ok {
+		return fmt.Errorf("trace task add failed, %v", v)
 	}
-	host := "127.0.0.1"
-	if v := config.Get("trace.host"); v != nil {
-		host = v.(string)
-	}
-	var port int64 = 33033
-	if v := config.Get("trace.port"); v != nil {
-		port = parse.Number(v)
-	}
-	task.NotifyUrl = fmt.Sprintf("udp:%s:%d", host, port)
-
-	// 查询网元获取IP
-	neInfo := neService.NewNeInfo.FindByNeTypeAndNeID(task.NeType, task.NeId)
-	if neInfo.NeId != task.NeId || neInfo.IP == "" {
-		return fmt.Errorf("app.common.noNEInfo")
-	}
-
-	// 查询网元任务信息
-	if msg, err := neFetchlink.NeTraceInfo(neInfo, task.TraceId); err == nil {
-		s, _ := json.Marshal(msg)
-		task.FetchMsg = string(s)
-		// 修改任务信息
-		data := map[string]any{
-			"neType":    neInfo.NeType,
-			"neId":      neInfo.NeId,
-			"notifyUrl": task.NotifyUrl,
-			"id":        parse.Number(task.TraceId),
-			"startTime": date.ParseDateToStr(task.StartTime, date.YYYY_MM_DD_HH_MM_SS),
-			"endTime":   date.ParseDateToStr(task.EndTime, date.YYYY_MM_DD_HH_MM_SS),
-		}
-		switch task.TraceType {
-		case "1": // Interface
-			data["traceType"] = "Interface"
-			data["interfaces"] = strings.Split(task.Interfaces, ",")
-		case "2": // Device
-			task.UeIp = neInfo.IP
-			data["traceType"] = "Device"
-			data["ueIp"] = task.UeIp
-			data["srcIp"] = task.SrcIp
-			data["dstIp"] = task.DstIp
-			data["signalPort"] = task.SignalPort
-		case "3": // UE
-			data["traceType"] = "UE"
-			data["imsi"] = task.IMSI
-			data["msisdn"] = task.MSISDN
-		default:
-			return fmt.Errorf("trace type is not disabled")
-		}
-		neFetchlink.NeTraceEdit(neInfo, data)
-	}
-
-	// 更新数据库
-	r.traceTaskRepository.Update(task)
 	return nil
 }

 // DeleteByIds 批量删除信息
-func (r *TraceTask) DeleteByIds(ids []int64) (int64, error) {
+func (r TraceTask) DeleteByIds(ids []int64) (int64, error) {
 	// 检查是否存在
 	rows := r.traceTaskRepository.SelectByIds(ids)
 	if len(rows) <= 0 {
@@ -327,14 +275,30 @@ func (r *TraceTask) DeleteByIds(ids []int64) (int64, error) {
 	}

 	if len(rows) == len(ids) {
-		// 停止任务
+		// 删除数据同时给网元发送停止任务
 		for _, v := range rows {
-			neInfo := neService.NewNeInfo.FindByNeTypeAndNeID(v.NeType, v.NeId)
-			if neInfo.NeId != v.NeId || neInfo.IP == "" {
+			// 删除数据
+			r.traceDataRepository.DeleteByTraceId(v.TraceId)
+
+			// 网元列表
+			neList := strings.Split(v.NeList, ",")
+			if len(neList) <= 0 {
 				continue
 			}
-			neFetchlink.NeTraceDelete(neInfo, v.TraceId)
+			// 停止任务
+			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, v.TraceId)
+			}
 		}
+
 		num := r.traceTaskRepository.DeleteByIds(ids)
 		return num, nil
 	}
--- a/src/modules/trace/service/trace_task_udp_data.go
+++ b/src/modules/trace/service/trace_task_udp_data.go
@@ -1,334 +1,271 @@
 package service

 import (
-	"encoding/base64"
+	"bytes"
 	"encoding/binary"
 	"fmt"
+	"net"
 	"os"
 	"path/filepath"
 	"runtime"
-	"strings"
 	"time"

-	"golang.org/x/net/http/httpguts"
-	"golang.org/x/net/http2/hpack"
+	"be.ems/src/modules/trace/model"
 )

-const (
-	GTPU_V1_VERSION = 1 << 5
-	GTPU_VER_MASK   = 7 << 5
-	GTPU_PT_GTP     = 1 << 4
-	GTPU_HEADER_LEN = 12
-	GTPU_E_S_PB_BIT = 7
-	GTPU_E_BI       = 1 << 2
-)
-
-const (
-	GTPU_HEADER_VERSION_INDEX  = 0
-	GTPU_HEADER_MSG_TYPE_INDEX = 1
-	GTPU_HEADER_LENGTH_INDEX   = 2
-	GTPU_HEADER_TEID_INDEX     = 4
-)
-
-type ExtHeader struct {
-	TaskId    uint32
-	IMSI      string
-	IfType    int
-	MsgType   int
-	MsgDirect int // 0-recv,1-send
-	TimeStamp int64
-	SrcIP     string
-	DstIP     string
-	SrcPort   uint16
-	DstPort   uint16
-	Proto     int // Protocol
-	PPI       int // only for SCTP
-	DataLen   uint16
-	DataInfo  []byte
+// TraceMsgToOamTraceData 结构体定义
+type TraceMsgToOamTraceData struct {
+	NfTraceMsg      *model.TraceMsg
+	TimestampStr    string
+	TracePayloadLen uint16
+	TracePayload    []byte
 }

-// parseUDPData 解析UDP数据
-func parseUDPData(rvMsg []byte, rvLen int) (ExtHeader, error) {
-	var extHdr ExtHeader
-	// var tr dborm.TraceData
-	var off int
-	msg := rvMsg
-
-	verFlags := msg[GTPU_HEADER_VERSION_INDEX]
-
-	gtpuHdrLen := GTPU_HEADER_LEN
-
-	localTeid := binary.BigEndian.Uint32(msg[GTPU_HEADER_TEID_INDEX:])
-
-	extHdr.TaskId = localTeid
-
-	if (verFlags & GTPU_E_S_PB_BIT) != 0 {
-		if (verFlags & GTPU_E_BI) != 0 {
-			extTypeIndex := GTPU_HEADER_LEN - 1
-
-			extType := msg[extTypeIndex]
-
-			if extType == 0xFE {
-				extHdr.IMSI = string(msg[extTypeIndex+2 : extTypeIndex+17])
-				extHdr.IfType = int(msg[extTypeIndex+17])
-				extHdr.MsgType = int(msg[extTypeIndex+18])
-				extHdr.MsgDirect = int(msg[extTypeIndex+19])
-
-				extHdr.TimeStamp = time.Now().UTC().UnixMilli()
-				// extHdr.TimeStamp = int64(binary.BigEndian.Uint64(msg[extTypeIndex+19:]))
-				// fmt.Printf("ext info %v %s %d %d %d \n", msg[(extTypeIndex+2):(extTypeIndex+20)], extHdr.IMSI, extHdr.IfType, extHdr.MsgType, extHdr.MsgDirect)
-				// set offset of IP Packet
-				off = 40 + 4
-				//src ip: msg+40+12
-				extHdr.SrcIP = fmt.Sprintf("%d.%d.%d.%d", msg[off+12], msg[off+13], msg[off+14], msg[off+15])
-				//dst ip: msg+40+12+4
-				extHdr.DstIP = fmt.Sprintf("%d.%d.%d.%d", msg[off+16], msg[off+17], msg[off+18], msg[off+19])
-				extHdr.SrcPort = uint16(binary.BigEndian.Uint16(msg[off+20:]))
-				extHdr.DstPort = uint16(binary.BigEndian.Uint16(msg[off+22:]))
-				// fmt.Printf("info %s:%d %s:%d \n", extHdr.SrcIP, extHdr.SrcPort, extHdr.DstIP, extHdr.DstPort)
-				// ip header start msg+40
-				extHdr.DataLen = uint16(rvLen - off)
-				extHdr.DataInfo = make([]byte, int(rvLen-off))
-				copy(extHdr.DataInfo, []byte(msg[off:]))
-
-				// 132 SCTP
-				// 6 TCP
-				// 17 UDP
-				extHdr.Proto = int(msg[off+9])
-				if extHdr.Proto == 132 {
-					extHdr.PPI = int(msg[off+47])
-					extHdr.DataLen = uint16(binary.BigEndian.Uint16(msg[(off+34):]) - 16)
-					// fmt.Printf("dat len %d %d \n", extHdr.DataLen, extHdr.PPI)
-				}
-			}
-
-			for extType != 0 && extTypeIndex < rvLen {
-				extLen := msg[extTypeIndex+1] << 2
-				if extLen == 0 {
-					return extHdr, fmt.Errorf("error, extLen is zero")
-				}
-
-				gtpuHdrLen += int(extLen)
-				extTypeIndex += int(extLen)
-				extType = msg[extTypeIndex]
-			}
-		}
-	} else {
-		gtpuHdrLen -= 4
-	}
-	return extHdr, nil
-}
-
-// UDPDataHandler UDP数据处理
-func UDPDataHandler(data []byte, n int) (map[string]any, error) {
-	extHdr, err := parseUDPData(data, n)
+// traceHandler 处理跟踪数据
+func traceHandler(data []byte) (*TraceMsgToOamTraceData, error) {
+	decodeData, err := decodeTraceData(data)
 	if err != nil {
-		return nil, err
+		return decodeData, err
 	}
-	if extHdr.TaskId == 0 || extHdr.DataLen < 1 {
-		return nil, fmt.Errorf("data error")
+	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))
 	}

-	m := map[string]any{
-		"taskId":    extHdr.TaskId,
-		"imsi":      extHdr.IMSI,
-		"ifType":    extHdr.IfType,
-		"srcAddr":   fmt.Sprintf("%s:%d", extHdr.SrcIP, extHdr.SrcPort),
-		"dstAddr":   fmt.Sprintf("%s:%d", extHdr.DstIP, extHdr.DstPort),
-		"msgType":   extHdr.MsgType,
-		"msgDirect": extHdr.MsgDirect,
-		"timestamp": extHdr.TimeStamp,
-		"dataLen":   extHdr.DataLen,
-		// "dataInfo":  extHdr.DataInfo,
-		"decMsg": "",
-	}
-	// Base64 编码
-	m["dataInfo"] = base64.StdEncoding.EncodeToString(extHdr.DataInfo)
-
-	if extHdr.Proto == 6 { // TCP
-		// 取响应数据
-		iplen := uint16(binary.BigEndian.Uint16(extHdr.DataInfo[2:]))
-		tcplen := uint16(iplen - 32 - 20)
-		hdrlen := uint16(binary.BigEndian.Uint16(extHdr.DataInfo[20+32+1:]))
-		offset := uint16(52)
-		// fmt.Printf("HTTP %d %d %d \n", iplen, tcplen, hdrlen)
-		if tcplen > (hdrlen + 9) { // has data
-			doffset := uint16(offset + hdrlen + 9)
-			datlen := uint16(binary.BigEndian.Uint16(extHdr.DataInfo[doffset+1:]))
-			// fmt.Printf("HTTP datlen %d \n", datlen)
-			m["decMsg"], _ = httpDataMsg(extHdr.DataInfo[offset+9:offset+9+hdrlen], extHdr.DataInfo[doffset+9:doffset+datlen+9])
-		} else {
-			m["decMsg"], _ = httpDataMsg(extHdr.DataInfo[offset+9:hdrlen], nil)
-		}
-	}
-
-	// pcap文件
-	m["pcapFile"] = writePcap(extHdr)
-	return m, nil
-}
-
-// =========== TCP协议Body ===========
-
-// httpDataMsg Http数据信息处理
-func httpDataMsg(header []byte, data []byte) (string, error) {
-	var remainSize = uint32(16 << 20)
-	var sawRegular bool
-	var invalid bool // pseudo header field errors
-	var Fields []hpack.HeaderField
-
-	invalid = false
-	hdec := hpack.NewDecoder(4096, nil)
-	hdec.SetEmitEnabled(true)
-	hdec.SetMaxStringLength(int(16 << 20))
-	hdec.SetEmitFunc(func(hf hpack.HeaderField) {
-		if !httpguts.ValidHeaderFieldValue(hf.Value) {
-			// Don't include the value in the error, because it may be sensitive.
-			invalid = true
-		}
-		isPseudo := strings.HasPrefix(hf.Name, ":")
-		if isPseudo {
-			if sawRegular {
-				invalid = true
-			}
-		} else {
-			sawRegular = true
-			if !validWireHeaderFieldName(hf.Name) {
-				invalid = true
-			}
-		}
-
-		if invalid {
-			hdec.SetEmitEnabled(false)
-			return
-		}
-
-		size := hf.Size()
-		if size > remainSize {
-			hdec.SetEmitEnabled(false)
-			//mh.Truncated = true
-			return
-		}
-		remainSize -= size
-
-		Fields = append(Fields, hf)
-	})
-
-	// defer hdec.SetEmitFunc(func(hf hpack.HeaderField) {})
-
-	frag := header
-	if _, err := hdec.Write(frag); err != nil {
-		return "", err
-	}
-
-	if err := hdec.Close(); err != nil {
-		return "", err
-	}
-
-	// hdec.SetEmitFunc(func(hf hpack.HeaderField) {})
-
-	var headers []byte
-	var line string
-	for i := range Fields {
-		line = fmt.Sprintf("\"%s\":\"%s\",", Fields[i].Name, Fields[i].Value)
-		headers = append(headers, []byte(line)...)
-	}
-
-	if len(data) > 0 {
-		return fmt.Sprintf("{ %s \"content\":%s }", string(headers), string(data)), nil
-	} else {
-		return fmt.Sprintf("{ %s }", string(headers)), nil
-	}
-}
-
-// validWireHeaderFieldName 校验报文头字段名称
-func validWireHeaderFieldName(v string) bool {
-	if len(v) == 0 {
-		return false
-	}
-	for _, r := range v {
-		if !httpguts.IsTokenRune(r) {
-			return false
-		}
-		if 'A' <= r && r <= 'Z' {
-			return false
-		}
-	}
-	return true
-}
-
-// =========== writePcap 写Pcap文件 ===========
-
-const magicMicroseconds = 0xA1B2C3D4
-const versionMajor = 2
-const versionMinor = 4
-
-func writeEmptyPcap(filename string, timeStamp int64, length int, data []byte) error {
-	var err error
-	var file *os.File
-	if err := os.MkdirAll(filepath.Dir(filename), 0775); err != nil {
-		return err
-	}
-	if _, err = os.Stat(filename); os.IsNotExist(err) {
-		file, err = os.Create(filename)
-		// File Header
-		var fileHeaderBuf [24]byte
-		binary.LittleEndian.PutUint32(fileHeaderBuf[0:4], magicMicroseconds)
-		binary.LittleEndian.PutUint16(fileHeaderBuf[4:6], versionMajor)
-		binary.LittleEndian.PutUint16(fileHeaderBuf[6:8], versionMinor)
-		// bytes 8:12 stay 0 (timezone = UTC)
-		// bytes 12:16 stay 0 (sigfigs is always set to zero, according to
-		//   http://wiki.wireshark.org/Development/LibpcapFileFormat
-		binary.LittleEndian.PutUint32(fileHeaderBuf[16:20], 0x00040000)
-		binary.LittleEndian.PutUint32(fileHeaderBuf[20:24], 0x00000071)
-		if _, err := file.Write(fileHeaderBuf[:]); err != nil {
-			return err
-		}
-	} else {
-		file, err = os.OpenFile(filename, os.O_WRONLY|os.O_APPEND, 0666)
-	}
-	if err != nil {
-		return err
-	}
-	defer file.Close()
-
-	// Packet Header
-	var packetHeaderBuf [24]byte
-	t := time.UnixMilli(timeStamp)
-	if t.IsZero() {
-		t = time.Now()
-	}
-	secs := t.Unix()
-	usecs := t.Nanosecond() / 1000
-	binary.LittleEndian.PutUint32(packetHeaderBuf[0:4], uint32(secs))
-	binary.LittleEndian.PutUint32(packetHeaderBuf[4:8], uint32(usecs))
-	binary.LittleEndian.PutUint32(packetHeaderBuf[8:12], uint32(length+16))
-	binary.LittleEndian.PutUint32(packetHeaderBuf[12:16], uint32(length+16))
-	if _, err := file.Write(packetHeaderBuf[:]); err != nil {
-		return err
-	}
-
-	// 数据包内容的定义
-	cooked := [...]byte{0x00, 0x00, 0x03, 0x04, 0x00, 0x06, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x08, 0x00}
-	if _, err := file.Write(cooked[:]); err != nil {
-		return err
-	}
-
-	// Packet Data
-	if _, err := file.Write(data); err != nil {
-		return err
-	}
-	return nil
-}
-
-// writePcap 写Pcap文件并返回文件路径
-func writePcap(extHdr ExtHeader) string {
-	filePath := fmt.Sprintf("/tmp/omc/trace/task_%d.pcap", extHdr.TaskId)
+	// 输出到文件
+	filePath := fmt.Sprintf("/tmp/omc/trace/task_%d.pcap", decodeData.NfTraceMsg.TraceId)
 	if runtime.GOOS == "windows" {
 		filePath = fmt.Sprintf("C:%s", filePath)
 	}
-	err := writeEmptyPcap(filePath, extHdr.TimeStamp, int(extHdr.DataLen), extHdr.DataInfo)
-	if err != nil {
-		return ""
-	}
-	return filePath
+	err = writePCAP(filePath, decodeData.NfTraceMsg.Timestamp, decodeData.TracePayload)
+
+	return decodeData, err
+}
+
+// decodeTraceData 解析跟踪数据
+func decodeTraceData(data []byte) (*TraceMsgToOamTraceData, error) {
+	buf := bytes.NewBuffer(data)
+	oamData := new(TraceMsgToOamTraceData)
+	msg := new(model.TraceMsg)
+
+	// 1. Parse Flag (1 byte)
+	var flag uint8
+	if err := binary.Read(buf, binary.BigEndian, &flag); err != nil {
+		return nil, err
+	}
+
+	// Parse message type (bits 1-2)
+	switch flag & 0x3 {
+	case 0x1:
+		msg.MsgType = model.MSG_TYPE_REQ
+	case 0x2:
+		msg.MsgType = model.MSG_TYPE_RSP
+	}
+
+	// Parse message direction (bits 3-4)
+	switch flag & 0xc {
+	case 0x4:
+		msg.MsgDirect = model.MSG_DIRECT_SEND
+	case 0x8:
+		msg.MsgDirect = model.MSG_DIRECT_RECV
+	}
+
+	// 2. Parse TraceId (4 bytes)
+	var traceId uint32
+	if err := binary.Read(buf, binary.BigEndian, &traceId); err != nil {
+		return nil, err
+	}
+	msg.TraceId = int(traceId)
+
+	// 3. Parse IMSI (15 bytes fixed length)
+	imsi := make([]byte, 15)
+	if _, err := buf.Read(imsi); err != nil {
+		return nil, err
+	}
+	msg.IMSI = string(imsi)
+
+	// 4. Parse IfType (1 byte)
+	ifTypeByte, err := buf.ReadByte()
+	if err != nil {
+		return nil, err
+	}
+	msg.IfType = byte2Iftype(ifTypeByte)
+
+	// 5. Parse Timestamp (8 bytes)
+	if err := binary.Read(buf, binary.BigEndian, &msg.Timestamp); err != nil {
+		return nil, err
+	}
+
+	// 6. Parse SrcIp (4 bytes IPv4)
+	srcIp := make([]byte, 4)
+	if flag&0x20 != 0 {
+		srcIp = make([]byte, 16)
+	}
+	if _, err := buf.Read(srcIp); err != nil {
+		return nil, err
+	}
+	msg.SrcIpStr = net.IP(srcIp).String()
+	// 7. Parse DstIp (4 bytes IPv4)
+	dstIp := make([]byte, 4)
+	if flag&0x20 != 0 {
+		dstIp = make([]byte, 16)
+	}
+	if _, err := buf.Read(dstIp); err != nil {
+		return nil, err
+	}
+	msg.DstIpStr = net.IP(dstIp).String()
+
+	// 8. Parse NfName (variable length)
+	nfNameLen, err := buf.ReadByte()
+	if err != nil {
+		return nil, err
+	}
+	nfName := make([]byte, nfNameLen)
+	if _, err := buf.Read(nfName); err != nil {
+		return nil, err
+	}
+	msg.NfName = string(nfName)
+
+	// 9. Parse MsgEvent (variable length)
+	msgEventLen, err := buf.ReadByte()
+	if err != nil {
+		return nil, err
+	}
+	msgEvent := make([]byte, msgEventLen)
+	if _, err := buf.Read(msgEvent); err != nil {
+		return nil, err
+	}
+	msg.MsgEvent = string(msgEvent)
+
+	// 10. Parse ExtenBuf (variable length)
+	extenBufLen, err := buf.ReadByte()
+	if err != nil {
+		return nil, err
+	}
+	if extenBufLen > 0 {
+		extenBuf := make([]byte, extenBufLen)
+		if _, err := buf.Read(extenBuf); err != nil {
+			return nil, err
+		}
+		// TODO: Parse extended fields according to actual protocol
+	}
+	// 11. len of payload(2 byte)
+	var len uint16
+	if err := binary.Read(buf, binary.BigEndian, &len); err != nil {
+		return nil, err
+	}
+
+	oamData.NfTraceMsg = msg
+	oamData.TimestampStr = time.Unix(0, msg.Timestamp).Format("2006-01-02 15:04:05.999999999")
+	oamData.TracePayloadLen = len
+	oamData.TracePayload = buf.Bytes()
+	return oamData, nil
+}
+
+// byte2Iftype 将字节转换为接口类型字符串
+// 参考：3GPP TS 32.422 V17.0.0 (2022-06) 5G; 3GPP TS 29.244 V17.0.0 (2022-06) 5G
+func byte2Iftype(val uint8) string {
+	ret := ""
+	switch val {
+	case 1:
+		ret = "N1"
+	case 2:
+		ret = "N2"
+	case 3:
+		ret = "N1/N2"
+	case 4:
+		ret = "N4"
+	case 7:
+		ret = "N7"
+	case 8:
+		ret = "N8"
+	case 10:
+		ret = "N10"
+	case 11:
+		ret = "N11"
+	case 12:
+		ret = "N12"
+	case 13:
+		ret = "N13"
+	case 15:
+		ret = "N15"
+	}
+	return ret
+}
+
+// writePCAP 写入PCAP文件
+func writePCAP(filename string, timeStamp int64, data []byte) error {
+	var err error
+	var file *os.File
+
+	// 1. 创建目录（含权限控制）
+	if err := os.MkdirAll(filepath.Dir(filename), 0775); err != nil {
+		return err
+	}
+
+	// 2. 智能打开文件（原子操作避免竞态）
+	if _, err = os.Stat(filename); os.IsNotExist(err) {
+		file, err = os.Create(filename)
+		if err != nil {
+			return err
+		}
+		// 写入PCAP全局头（严格兼容Wireshark）
+		fileHeader := [24]byte{
+			0xD4, 0xC3, 0xB2, 0xA1, // magic_number (微秒级)
+			0x02, 0x00, 0x04, 0x00, // version_major(2) + version_minor(4)
+			0x00, 0x00, 0x00, 0x00, // thiszone (UTC)
+			0x00, 0x00, 0x00, 0x00, // sigfigs (固定0)
+			0x00, 0x00, 0x04, 0x00, // snaplen (1024)
+			0x71, 0x00, 0x00, 0x00, // network (LINKTYPE_LINUX_SLL)
+		}
+		if _, err := file.Write(fileHeader[:]); err != nil {
+			return err
+		}
+	} else {
+		// 追加模式打开（避免截断已有内容）
+		file, err = os.OpenFile(filename, os.O_WRONLY|os.O_APPEND|os.O_SYNC, 0666)
+		if err != nil {
+			return err
+		}
+	}
+	defer file.Close()
+
+	// 3. 构造Linux cooked头（RFC 3580规范）
+	linuxSLLHeaderLen := 16 // Linux cooked模式头长度
+	cookedHeader := make([]byte, linuxSLLHeaderLen)
+	binary.BigEndian.PutUint16(cookedHeader[0:2], 0x0000)   // 数据包类型（主机→网络）
+	binary.BigEndian.PutUint16(cookedHeader[2:4], 0x0304)   // 地址类型（ARPHRD_ETHER）
+	binary.BigEndian.PutUint16(cookedHeader[4:6], 0x0008)   // 协议类型（ETH_P_IP）
+	binary.BigEndian.PutUint16(cookedHeader[14:16], 0x0800) // 数据包类型（PACKET_HOST）
+
+	// 4. 合并链路层头与数据（兼容IPv4选项）
+	fullData := append(cookedHeader, data...)
+	pktLen := len(fullData)
+
+	// 5. 生成精确时间戳（处理闰秒和时区）
+	t := time.Unix(0, timeStamp).UTC()
+	if t.IsZero() {
+		t = time.Now().UTC()
+	}
+	secs := t.Unix()
+	usecs := t.Nanosecond() / 1000 // 微秒级时间戳
+
+	// 6. 构造PCAP报文头（16字节）
+	var packetHeader [16]byte
+	binary.LittleEndian.PutUint32(packetHeader[0:4], uint32(secs))     // 时间戳秒
+	binary.LittleEndian.PutUint32(packetHeader[4:8], uint32(usecs))    // 时间戳微秒
+	binary.LittleEndian.PutUint32(packetHeader[8:12], uint32(pktLen))  // 捕获长度
+	binary.LittleEndian.PutUint32(packetHeader[12:16], uint32(pktLen)) // 原始长度
+
+	// 7. 原子写入操作（避免部分写入）
+	buf := new(bytes.Buffer)
+	buf.Write(packetHeader[:])
+	buf.Write(fullData)
+	if _, err := file.Write(buf.Bytes()); err != nil {
+		return err
+	}
+
+	// 8. 强制刷盘并验证写入
+	return file.Sync()
 }