ebpf/tracing/apm_tracing.go

package tracing

import (
	"context"
	"fmt"
	"net"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	klog "github.com/sirupsen/logrus"

	"github.com/coroot/coroot-node-agent/ebpftracer"
	"github.com/coroot/coroot-node-agent/ebpftracer/l7"
	"github.com/coroot/coroot-node-agent/utils"
	. "github.com/coroot/coroot-node-agent/utils/modelse"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
	"go.opentelemetry.io/otel/trace"
	"inet.af/netaddr"
)

/**
 * Trace
 */

func (t *Trace) setContext(ctx context.Context) {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.ctx = ctx
}

func (t *Trace) setSpan(span trace.Span) {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.span = span
}

type TimeMap struct {
	Time uint64
	Type int
	Map  *ebpftracer.StackFunEvent
}

//func (t *Trace) buildFun() {
//	mapSlice := []TimeMap{}
//	for i, v := range t.stack {
//		timeStartMap := TimeMap{}
//		if v.StackEvent.Location == 0 {
//			timeStartMap = TimeMap{
//				Time: v.StackEvent.TimeNsStart,
//				Type: 0,
//				Map:  &t.stack[i],
//			}
//		} else {
//			timeStartMap = TimeMap{
//				Time: v.StackEvent.TimeNsEnd,
//				Type: 1,
//				Map:  &t.stack[i],
//			}
//		}
//		mapSlice = append(mapSlice, timeStartMap)
//	}
//	sort.Slice(mapSlice, func(i, j int) bool {
//		return mapSlice[i].Time < mapSlice[j].Time
//	})
//
//	funStack := []TimeMap{}
//
//	currentfunNum := 1
//
//	// for k, v := range mapSlice {
//	// 	fmt.Println("---SliceSliceindex", k, "value", v.Time, v.Type, v.Map.Uprobe.Funcname, v.Map.StackEvent.Nid)
//	// }
//
//	mapSliceLen := len(mapSlice)
//	for k, v := range mapSlice {
//		// fmt.Println("SliceSliceindex", k, "value", v.Time, v.Type, v.Map.Uprobe.Funcname, v.Map.StackEvent.Nid)
//		if v.Type == 0 {
//			// 函数入口
//			funStack = append(funStack, v)
//		} else if v.Type == 1 {
//			// 函数出口
//			len := len(funStack)
//			if len < 1 {
//				fmt.Printf("buildFun ErrorError return before enter: %v\n", v)
//				continue
//			}
//			currnt := funStack[len-1]
//			if currnt.Map.StackEvent.Location != 0 {
//				fmt.Printf("currnt StackEvent ErrorError is not enter: %v\n", v)
//				continue
//			}
//			if k < mapSliceLen-1 && len >= 2 {
//				nextfun := mapSlice[k+1]
//				preCurrnt := funStack[len-2]
//				// // 处理 .netcore 多次 returun
//				// 下一个事件是 return 并且函数名跟当前事件是一样的，且上一个函数不是当前函数
//				if nextfun.Map.StackEvent.Location == 1 && nextfun.Map.Uprobe.Funcname == currnt.Map.Uprobe.Funcname && preCurrnt.Map.Uprobe.Funcname != currnt.Map.Uprobe.Funcname {
//					currentfunNum++
//					continue
//				}
//			}
//			funStack = funStack[:len-1]
//			duration := v.Map.StackEvent.TimeNsEnd - currnt.Map.StackEvent.TimeNsStart
//			t.FuncTraceQuery(currnt.Map.Uprobe.Funcname, time.Duration(duration), currnt.Map.StackEvent.TimeNsStart, v.Map.StackEvent.TimeNsEnd, currentfunNum)
//			currentfunNum = 1
//		}
//	}
//}

func (t *Trace) startReady() {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.startEventReady = true
}

func (t *Trace) endReadyEvent(needCount uint32) {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.endEventReady = true
	t.needEventCount = needCount
}

func (t *Trace) AllEventReady(traceID uint64) bool {
	klog.Debugf("[AllEventReady (current/need)|start|end|traceid](%d/%d)%v|%v|%d", *t.currenEventCount, t.needEventCount, t.startEventReady, t.endEventReady, traceID)
	return t.startEventReady && t.endEventReady && *t.currenEventCount >= t.needEventCount
}

func (t *Trace) GrpcServerTraceStartEvent(sn string, sport uint16, r *l7.RequestData, appInfo AppInfo, container_id string) {
	t.span.SetAttributes(attribute.String("rpc.uri", string(r.Payload)))
	t.commonAttrs = []attribute.KeyValue{
		semconv.NetPeerName(sn),
		semconv.NetPeerPort(int(sport)),
		// buildAppMapFromEvent
		attribute.Int("server.code_type", appInfo.CodeType.Int()),
		attribute.String("server.app_name", appInfo.AppName),
		attribute.String("server.service_name", "GRPC"),
		attribute.String("rpc.service_type", "RPC"),
		attribute.String("rpc.oper_type", "PROVIDER"),
		attribute.Int64("server.app_id", appInfo.AppIdHash.IntVal),
		attribute.Int64("server.agent_id", appInfo.AgentId),
		attribute.Int64("server.instance_id", appInfo.InstanceIdHash.IntVal),
		attribute.String("server.container_id", container_id),
	}
	t.span.SetAttributes(t.commonAttrs...)
	t.startReady()
}

func (t *Trace) MQConsumerTraceStartEvent(sn string, sport uint16, r *l7.RequestData, appInfo AppInfo, container_id string) {
	t.span.SetAttributes(attribute.String("rpc.uri", string(r.Payload)))
	r.Protocol = l7.ProtocolKafka
	t.commonAttrs = []attribute.KeyValue{
		semconv.NetPeerName(sn),
		semconv.NetPeerPort(int(sport)),
		// buildAppMapFromEvent
		attribute.Int("server.code_type", appInfo.CodeType.Int()),
		attribute.String("server.app_name", appInfo.AppName),
		attribute.String("server.service_name", r.Protocol.ServiceNameString()),
		attribute.String("service_type", "MQ"),
		attribute.String("oper_type", "CONSUMER"),
		attribute.String("server.service_type", "TASK"),
		attribute.Int64("server.app_id", appInfo.AppIdHash.IntVal),
		attribute.Int64("server.agent_id", appInfo.AgentId),
		attribute.Int64("server.instance_id", appInfo.InstanceIdHash.IntVal),
		attribute.String("server.container_id", container_id),
		attribute.String("http.uri", "/"+r.MQTopic),
		attribute.Int("server.pid", int(r.Pid)),
		attribute.String("mq.ip", sn),
		attribute.Int("mq.port", int(sport)),
	}
	t.span.SetAttributes(t.commonAttrs...)
	t.startReady()
}

func (t *Trace) TraceStartEvent(method, path, sn, ua string, sport uint16, r *l7.RequestData, addr netaddr.IPPort, pid uint32, appInfo AppInfo, container_id string) {
	t.span.SetAttributes(semconv.HTTPURL(fmt.Sprintf("http://%s:%d%s", sn, sport, path)),
		semconv.HTTPMethod(method),
		attribute.String("http.uri", path))
	if r.Status > 399 {
		t.span.SetStatus(codes.Error, "")
	}
	//host := fmt.Sprintf("%s:%s", sn, sport)
	t.destination = addr
	t.commonAttrs = []attribute.KeyValue{
		semconv.NetPeerName(sn),
		semconv.NetPeerPort(int(sport)),
		// buildAppMapFromEvent
		attribute.Int("server.code_type", appInfo.CodeType.Int()),
		attribute.String("server.app_name", appInfo.AppName),
		attribute.String("server.service_name", appInfo.ServiceType),
		attribute.Int64("server.app_id", appInfo.AppIdHash.IntVal),
		attribute.Int64("server.agent_id", appInfo.AgentId),
		attribute.Int64("server.instance_id", appInfo.InstanceIdHash.IntVal),
		attribute.Int("server.pid", int(pid)),
		attribute.String("server.container_id", container_id),
		attribute.String("server.user_agent", ua),
	}

	if r.SysvcFrom != "" {
		t.commonAttrs = append(t.commonAttrs, attribute.String("server.sysvc_from", r.SysvcFrom))
	}

	t.span.SetAttributes(t.commonAttrs...)
	t.startReady()
}

// set context span
func (t *Trace) CreateRootSpan(traceId uint64) {
	traceIdStr := strconv.Itoa(int(traceId))
	ctx, span := tracer(t.containerId).Start(context.Background(), traceIdStr, trace.WithSpanKind(trace.SpanKindClient))
	t.setContext(ctx)
	t.setSpan(span)
}

// Deprecated: use TraceStartEvent instead.
func (t *Trace) TraceStart(method, path string, status l7.Status, duration time.Duration) {
	if t == nil || method == "" {
		return
	}
	t.createParentSpan("APPLICATION", duration, status >= 400,
		semconv.HTTPURL(fmt.Sprintf("http://%s%s", t.destination.String(), path)),
		semconv.HTTPMethod(method),
		//semconv.HTTPStatusCode(int(status)),
		attribute.String("http.uri", path),
	)
}

// Deprecated: use TraceEndEvent instead.
//func (t *Trace) TraceEnd(r *l7.RequestData) {
//	if t == nil {
//		return
//	}
//	t.span.SetAttributes(
//		semconv.HTTPStatusCode(int(r.Status)),
//		attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
//	)
//
//	CalledId, err := strconv.ParseInt(r.ParentSpanContext.CalledId, 10, 64)
//	if err == nil && CalledId != 0 {
//		t.span.SetAttributes(attribute.Int64("server.called_id", CalledId))
//	}
//
//	InstanceIdFrom, err := strconv.ParseInt(r.ParentSpanContext.InstanceIdFrom, 10, 64)
//	if err == nil && InstanceIdFrom != 0 {
//		t.span.SetAttributes(attribute.Int64("server.instance_id_from", InstanceIdFrom))
//	}
//
//	AppIdFrom, err := strconv.ParseInt(r.ParentSpanContext.AppIdFrom, 10, 64)
//	if err == nil && AppIdFrom != 0 {
//		t.span.SetAttributes(attribute.Int64("server.app_id_from", AppIdFrom))
//	}
//	if r.ParentSpanContext.SpanIdFrom != "0000000000000000" {
//		t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
//	}
//	// klog.Debugf("r.ParentSpanContext.TypeFrom tpp %s", r.ParentSpanContext.TypeFrom)
//	if r.ParentSpanContext.TypeFrom == "10" {
//		t.span.SetAttributes(attribute.String("server.type_from", "SDK"))
//		t.span.SetAttributes(attribute.String("server.app_id_from", "-1"))
//		t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.TraceIdFrom))
//		t.span.SetAttributes(attribute.String("server.instance_id_from", "-1"))
//		t.span.SetAttributes(attribute.String("server.call_id", "-1"))
//	}
//	// for _, v := range t.stack {
//	// 	fmt.Printf("TraceEndTraceEndTraceEnd%s\n", v)
//	// }
//	//t.buildFun()
//	t.span.End(trace.WithTimestamp(time.Now()))
//}

// 新增结束事件
func (t *Trace) TraceEndEvent(r *l7.RequestData) {
	if t == nil {
		return
	}
	var attr []attribute.KeyValue
	attr = append(attr,
		semconv.HTTPStatusCode(int(r.Status)),
		attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
	)
	//t.span.SetAttributes(
	//	semconv.HTTPStatusCode(int(r.Status)),
	//	attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
	//)

	calledId, err := strconv.ParseInt(r.ParentSpanContext.CalledId, 10, 64)
	if err == nil && calledId != 0 {
		attr = append(attr, attribute.Int64("server.called_id", calledId))
		//t.span.SetAttributes(attribute.Int64("server.called_id", CalledId))
	}

	instanceIdFrom, err := strconv.ParseInt(r.ParentSpanContext.InstanceIdFrom, 10, 64)
	if err == nil && instanceIdFrom != 0 {
		attr = append(attr, attribute.Int64("server.instance_id_from", instanceIdFrom))

		//t.span.SetAttributes(attribute.Int64("server.instance_id_from", InstanceIdFrom))
	}

	appIdFrom, err := strconv.ParseInt(r.ParentSpanContext.AppIdFrom, 10, 64)
	if err == nil && appIdFrom != 0 {
		attr = append(attr, attribute.Int64("server.app_id_from", appIdFrom))
		// Add type_from attribute for all cases
		attr = append(attr, attribute.String("server.type_from", r.ParentSpanContext.TypeFrom.String()))
	}
	if r.ParentSpanContext.SpanIdFrom != "0000000000000000" {
		attr = append(attr, attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
		//t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
	}

	// klog.Debugf("r.ParentSpanContext.TypeFrom event tpp %s", r.ParentSpanContext.TypeFrom)
	if r.ParentSpanContext.TypeFrom == "10" {
		attr = append(attr, attribute.String("server.app_id_from", "-1"))
		attr = append(attr, attribute.String("server.span_id_from", r.ParentSpanContext.TraceIdFrom))
		attr = append(attr, attribute.String("server.instance_id_from", "-1"))
		attr = append(attr, attribute.String("server.call_id", "-1"))
	}

	attr = append(attr, attribute.String("server.src_addr", r.DAddr.String()))
	attr = append(attr, attribute.String("server.dst_addr", r.SAddr.String()))
	klog.Debugf("TraceEndEvent SAddr.String %s", r.SAddr.String())
	klog.Debugf("TraceEndEvent DAddr.String %s", r.DAddr.String())

	// 添加 SysvcFrom 到 attributes
	if r.SysvcFrom != "" {
		attr = append(attr, attribute.String("server.sysvc_from", r.SysvcFrom))
	}

	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
	t.span.SetAttributes(attr...)
	t.endReadyEvent(r.EventCount)
}

func (t *Trace) appendTimestamp(attr *[]attribute.KeyValue, s, e uint64, d int64) {
	*attr = append(*attr,
		attribute.Int64("time.start_at", utils.KtimeToTimestamp(s)),
		attribute.Int64("time.end_at", utils.KtimeToTimestamp(e)),
		attribute.Int64("time.duration", d),
	)
}

func (t *Trace) createParentSpan(name string, duration time.Duration, error bool, attrs ...attribute.KeyValue) {
	end := time.Now()
	start := end.Add(-duration)
	ctx, span := tracer(t.containerId).Start(context.Background(), name, trace.WithTimestamp(start), trace.WithSpanKind(trace.SpanKindClient))
	span.SetAttributes(attrs...)
	span.SetAttributes(t.commonAttrs...)
	if error {
		span.SetStatus(codes.Error, "")
	}
	t.setContext(ctx)
	t.setSpan(span)
}

func (t *Trace) SendEvent() {
	t.span.End()
}

func (t *Trace) GetSpan() trace.Span {
	return t.span
}

func (t *Trace) createTraceEvent(protocol l7.Protocol, eventType int, attrs ...attribute.KeyValue) {
	t.span.AddEventApm(protocol.String(), eventType, protocol.Int(), trace.WithAttributes(attrs...))
	//atomic.AddUint32(t.currenEventCount, 1)
}

func (t *Trace) createFuncTraceEvent(name string, eventType int, attrs ...attribute.KeyValue) {
	t.span.AddEventApm(name, eventType, 0, trace.WithAttributes(attrs...))
}

func (t *Trace) addEvent() {
	atomic.AddUint32(t.currenEventCount, 1)
}

func (t *Trace) createTraceSpan(name string, duration time.Duration, error bool, attrs ...attribute.KeyValue) {
	end := time.Now()
	start := end.Add(-duration)
	//fmt.Println("createTraceSpan:", t.ctx)
	_, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(start), trace.WithSpanKind(trace.SpanKindClient))
	span.SetAttributes(t.commonAttrs...)
	span.SetAttributes(attrs...)
	if error {
		span.SetStatus(codes.Error, "")
	}
	span.End(trace.WithTimestamp(end))
}

func (t *Trace) MysqlTraceQuery(query string, error bool, duration time.Duration, destination netaddr.IPPort) {
	if t == nil || query == "" {
		return
	}
	t.createTraceSpan(l7.ProtocolMysql.String(), duration, error,
		semconv.DBSystemMySQL,
		semconv.DBStatement(query),
		semconv.NetPeerName(destination.IP().String()),
		semconv.NetPeerPort(int(destination.Port())),
	)
}

// 内置允许操作（始终放行）
var builtinOps = map[string]struct{}{
	"SELECT": {},
	"INSERT": {},
	"UPDATE": {},
	"DELETE": {},
	"WITH":   {},
}

var (
	allowedOps map[string]struct{}
	initOnce   sync.Once
)

// 解析并缓存环境变量中的前缀
func initAllowedOps() {
	allowedOps = make(map[string]struct{})
	env := os.Getenv("ALLOWED_SQL_PREFIXES")
	for _, s := range strings.Split(env, ",") {
		if trimmed := strings.ToUpper(strings.TrimSpace(s)); trimmed != "" {
			allowedOps[trimmed] = struct{}{}
		}
	}
}

func isCURDOperation(q string) bool {
	if len(q) < 6 {
		return false
	}

	// 提取前缀（考虑最长10个字符）
	prefix := strings.ToUpper(strings.TrimSpace(q))
	if len(prefix) > 10 {
		prefix = prefix[:10]
	}

	// 提取第一个单词作为操作关键字
	spaceIdx := strings.IndexAny(prefix, " \t\n")
	if spaceIdx > 0 {
		prefix = prefix[:spaceIdx]
	}

	// 先查固定允许的操作
	if _, ok := builtinOps[prefix]; ok {
		return true
	} else {
		// 尝试清理所有非字母字符（移除控制字符、BOM、数字、标点等）:
		//`SELECT` 长度 6，字节：[83 69 76 69 67 84]
		//prefix中的值长度 8，字节：[0 1 83 69 76 69 67 84]
		prefixTmp := strings.Map(func(r rune) rune {
			// 只保留大写字母 A-Z
			if r >= 'A' && r <= 'Z' {
				return r
			}
			return -1 // 删除其他所有字符
		}, prefix)
		if _, ok := builtinOps[prefixTmp]; ok {
			return true
		}
	}

	// 加载环境变量控制的操作（只执行一次）
	initOnce.Do(initAllowedOps)

	// 查额外允许操作
	_, ok := allowedOps[prefix]
	return ok
}

func (t *Trace) SQLTraceQueryEvent(l7Type l7.Protocol, semconvVal attribute.KeyValue, query string, r *l7.RequestData, destination netaddr.IPPort) {
	//fmt.Println("query", query)
	if t == nil {
		return
	}
	t.addEvent()

	// 只保留增删改查基本操作
	if !isCURDOperation(query) {
		return
	}

	// 拼接 destination 的 IP 和端口为 ip:port 格式
	destAddr := fmt.Sprintf("%s:%d", destination.IP().String(), destination.Port())

	var attr []attribute.KeyValue
	attr = append(attr,
		semconvVal,
		semconv.DBStatement(query),
		semconv.NetPeerName(destination.IP().String()),
		semconv.NetPeerPort(int(destination.Port())),
		attribute.String("sql.src_addr", r.ComponentSAddr.String()),
		attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
		attribute.String("sql.dbn", destAddr),
	)
	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())

	attr = append(attr,
		attribute.Bool("sql.exception", r.Status.Error()),
	)
	if r.Status.Error() {
		attr = append(attr,
			attribute.String("sql.exception_msg", r.ErrorMsg),
		)
	}
	t.createTraceEvent(l7Type, ebpftracer.EventTypeL7Request.Int(), attr...)
}

//func (t *Trace) MysqlTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
//	fmt.Println("query", query)
//	if t == nil {
//		return
//	}
//
//	var attr []attribute.KeyValue
//	attr = append(attr,
//		semconv.DBSystemMySQL,
//		semconv.DBStatement(query),
//		semconv.NetPeerName(destination.IP().String()),
//		semconv.NetPeerPort(int(destination.Port())),
//		attribute.String("sql.src_addr", r.ComponentSAddr.String()),
//		attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
//	)
//	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
//
//	attr = append(attr,
//		attribute.Bool("sql.exception", r.Status.Error()),
//	)
//	t.createTraceEvent(l7.ProtocolMysql.String(), int(ebpftracer.EventTypeL7Request), int(l7.ProtocolMysql), attr...)
//}

//func (t *Trace) PostGreSqlTraceQuery(query string, error bool, duration time.Duration, destination netaddr.IPPort) {
//	if t == nil || query == "" {
//		return
//	}
//	t.createTraceSpan(l7.ProtocolPostgres.String(), duration, error,
//		semconv.DBSystemPostgreSQL,
//		semconv.DBStatement(query),
//		semconv.NetPeerName(destination.IP().String()),
//		semconv.NetPeerPort(int(destination.Port())),
//	)
//}

//func (t *Trace) PostGreSqlTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
//	fmt.Println("query", query)
//	if t == nil {
//		return
//	}
//
//	var attr []attribute.KeyValue
//	attr = append(attr,
//		semconv.DBSystemPostgreSQL,
//		semconv.DBStatement(query),
//		semconv.NetPeerName(destination.IP().String()),
//		semconv.NetPeerPort(int(destination.Port())),
//		attribute.String("sql.src_addr", r.ComponentSAddr.String()),
//		attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
//	)
//	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
//	attr = append(attr,
//		attribute.Bool("sql.exception", r.Status.Error()),
//	)
//	t.createTraceEvent(l7.ProtocolPostgres.String(), int(ebpftracer.EventTypeL7Request), int(l7.ProtocolPostgres), attr...)
//}

//func (t *Trace) DmTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
//	if t == nil || query == "" {
//		return
//	}
//	l7Type := int(l7.ProtocolDM)
//	var attr []attribute.KeyValue
//	attr = append(attr,
//		semconv.DBSystemDaMengDB,
//		semconv.DBStatement(query),
//		semconv.NetPeerName(destination.IP().String()),
//		semconv.NetPeerPort(int(destination.Port())),
//		attribute.String("sql.src_addr", r.ComponentSAddr.String()),
//		attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
//	)
//	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
//
//	attr = append(attr,
//		attribute.Bool("sql.exception", r.Status.Error()),
//	)
//	t.createTraceEvent(l7.ProtocolHTTP.String(), int(ebpftracer.EventTypeL7Request), l7Type, attr...)
//}

//func (t *Trace) RedisTraceQuery(cmd, args string, error bool, duration time.Duration) {
//	if t == nil || cmd == "" {
//		return
//	}
//	statement := cmd
//	if args != "" {
//		statement += " " + args
//	}
//	t.createTraceSpan(l7.ProtocolRedis.String(), duration, error,
//		semconv.DBSystemRedis,
//		semconv.DBOperation(cmd),
//		semconv.DBStatement(statement),
//	)
//}

func (t *Trace) NoSQLTraceQueryEvent(l7Type l7.Protocol, semconvVal attribute.KeyValue, operation, statement string, r *l7.RequestData, connSrc, connDest netaddr.IPPort) {
	if t == nil {
		return
	}
	t.addEvent()
	if statement == "" {
		return
	}

	// 拼接 destination 的 IP 和端口为 ip:port 格式

	srcAddr := r.ComponentSAddr.String()
	if r.ComponentSAddr.Port() == 0 {
		srcAddr = connSrc.String()
	}

	destinationAddr := r.ComponentDAddr.String()
	if r.ComponentDAddr.Port() == 0 {
		destinationAddr = connDest.String()
	}

	var attr []attribute.KeyValue
	attr = append(attr,
		semconvVal,
		semconv.DBStatement(statement),
		semconv.NetPeerName(connDest.IP().String()),
		semconv.NetPeerPort(int(connDest.Port())),
		attribute.String("nosql.src_addr", srcAddr),
		attribute.String("nosql.destination_addr", destinationAddr),
		attribute.String("nosql.dbn", connDest.String()),
	)

	// 如果 operation 不为空，添加 DBOperation 属性
	if operation != "" {
		attr = append(attr, semconv.DBOperation(operation))
	}

	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())

	attr = append(attr,
		attribute.Bool("nosql.exception", r.Status.Error()),
	)
	if r.Status.Error() {
		attr = append(attr,
			attribute.String("nosql.exception_msg", r.ErrorMsg),
		)
	}
	t.createTraceEvent(l7Type, ebpftracer.EventTypeL7Request.Int(), attr...)
}

func (t *Trace) MQTraceQueryEvent(l7Type l7.Protocol, semconvVal attribute.KeyValue, operation, statement string, r *l7.RequestData, connSrc, connDest netaddr.IPPort) {
	if t == nil {
		return
	}
	t.addEvent()

	if r.MQTopic != "" || r.MQKey != "" {
		statement = fmt.Sprintf("PRODUCE topic:%s key:%s", r.MQTopic, r.MQKey)
	}

	if statement == "" {
		return
	}

	srcAddr := r.ComponentSAddr.String()
	if r.ComponentSAddr.Port() == 0 {
		srcAddr = connSrc.String()
	}

	// 尝试解析目标地址，如果失败则使用 connDest 作为备用
	dAddr, err := net.ResolveTCPAddr("tcp", r.DestAddrString)
	var ip string
	port := 0
	if err != nil {
		klog.Errorln(err)
	} else {
		ip = dAddr.IP.String()
		port = dAddr.Port
	}

	assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
	if err != nil {
		assumedAppID = 0
	}

	var attr []attribute.KeyValue
	attr = append(attr,
		semconvVal,
		semconv.NetPeerName(ip),
		semconv.NetPeerPort(port),
		attribute.String("mq.info", statement),
		attribute.String("mq.src_addr", srcAddr),
		attribute.String("mq.destination_addr", r.DestAddrString),
		attribute.String("mq.topic", r.MQTopic),
		attribute.String("mq.key", r.MQKey),
		attribute.Int64("mq.assumed_app_id", assumedAppID),
		attribute.String("mq.span_id", r.SpanId),
		attribute.String("mq.oper_type", "PRODUCER"),
	)

	// 如果 operation 不为空，添加 DBOperation 属性
	if operation != "" {
		attr = append(attr, semconv.DBOperation(operation))
	}

	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())

	attr = append(attr,
		attribute.Bool("mq.exception", r.Status.Error()),
	)
	if r.Status.Error() {
		attr = append(attr,
			attribute.String("mq.exception_msg", r.ErrorMsg),
		)
	}
	t.createTraceEvent(l7Type, ebpftracer.EventTypeL7Request.Int(), attr...)
}

func (t *Trace) GrpcServerTraceQueryEvent(r *l7.RequestData, appInfo AppInfo) {
	if t == nil {
		return
	}
	t.addEvent()
	// if method == "" {
	// 	return
	// }

	assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
	if err != nil {
		assumedAppID = 0
	}

	var attr []attribute.KeyValue
	attr = append(attr,
		semconv.RPCSystemGRPC,
		attribute.String("rpc.uri", string(r.Payload)),
		attribute.String("rpc.schema", "grpc"),
		attribute.Int64("rpc.assumed_app_id", assumedAppID),
		attribute.String("rpc.span_id", r.SpanId),
		attribute.Int("rpc.port", appInfo.Sport),
		attribute.String("rpc.ip", appInfo.Sn),
		attribute.String("rpc.oper_type", "PROVIDER"),
		attribute.String("rpc.method", "test_method"),
	)

	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
	t.createTraceEvent(l7.ProtocolGrpc, ebpftracer.EventTypeL7Request.Int(), attr...)
}

func (t *Trace) GrpcClientTraceQueryEvent(r *l7.RequestData) {
	if t == nil {
		return
	}
	t.addEvent()
	// if method == "" {
	// 	return
	// }

	assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
	if err != nil {
		assumedAppID = 0
	}

	// 解析 RPCTarget 字符串，格式为 "ip:port"
	klog.Infof("r.RPCTarget is %s", r.DestAddrString)
	var rpcIP, rpcPort string
	if r.DestAddrString != "" {
		// 清理空字节，只保留有效字符
		cleanTarget := strings.TrimRight(r.DestAddrString, "\x00")
		cleanTarget = strings.TrimSpace(cleanTarget)
		klog.Infof("cleanTarget is %s", cleanTarget)

		if colonIndex := strings.LastIndex(cleanTarget, ":"); colonIndex != -1 {
			rpcIP = cleanTarget[:colonIndex]
			rpcPort = cleanTarget[colonIndex+1:]
			klog.Infof("rpcIP is %s, rpcPort is %s", rpcIP, rpcPort)
		} else {
			// 如果没有找到冒号，可能是只有IP或只有端口
			rpcIP = cleanTarget
		}
	}

	// 设置默认值
	if rpcIP == "" {
		rpcIP = "127.0.0.1"
	}
	if rpcPort == "" {
		rpcPort = "80"
	}

	// 转换端口为整数
	portInt, err := strconv.Atoi(rpcPort)
	if err != nil {
		klog.Infof("r.RPCTarget convert port err is %s", err)
		portInt = 80 // 默认端口
	} else {
		klog.Infof("r.RPCTarget convert port success, portInt is %d", portInt)
	}

	var attr []attribute.KeyValue
	attr = append(attr,
		semconv.RPCSystemGRPC,
		attribute.String("rpc.uri", string(r.Payload)),
		attribute.String("rpc.schema", "grpc"),
		attribute.Int64("rpc.assumed_app_id", assumedAppID),
		attribute.String("rpc.span_id", r.SpanId),
		attribute.Int("rpc.port", portInt),
		attribute.String("rpc.ip", rpcIP),
		attribute.String("rpc.oper_type", "CONSUMER"),
		attribute.String("rpc.method", string(r.Payload)),
	)

	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
	t.createTraceEvent(l7.ProtocolGrpc, ebpftracer.EventTypeL7Request.Int(), attr...)
}

func (t *Trace) DNSTraceQueryEvent(r *l7.RequestData, _type string, fqdn string, ttl uint32, ips []netaddr.IP) {
	if t == nil {
		return
	}
	t.addEvent()
	var attr []attribute.KeyValue
	var ipList []string
	for _, ip := range ips {
		ipList = append(ipList, ip.String())
	}
	ip := strings.Join(ipList, ",")
	attr = append(attr,
		attribute.String("dns.type", _type),
		attribute.String("dns.fqdn", fqdn),
		attribute.Int("dns.ttl", int(ttl)),
		attribute.String("dns.ips", ip),
	)
	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
	t.createTraceEvent(l7.ProtocolDNS, ebpftracer.EventTypeL7Request.Int(), attr...)
}

func (t *Trace) HttpTraceRequest(method, path, ip string, port uint16, r *l7.RequestData) {
	if t == nil || method == "" {
		return
	}
	assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
	if err != nil {
		assumedAppID = 0
	}
	status := r.Status
	duration := r.Duration
	t.createTraceSpan(l7.ProtocolHTTP.String(), duration, status >= 400,
		semconv.HTTPURL(fmt.Sprintf("http://%s%s", t.destination.String(), path)),
		semconv.HTTPMethod(method),
		semconv.HTTPStatusCode(int(status)),
		attribute.String("http.uri", path),
		attribute.String("http.ip", ip),
		attribute.Int64("http.assumed_app_id", assumedAppID),
		attribute.String("http.span_id", r.SpanId),
		attribute.Int("http.port", int(port)),
	)
}

// 新增事件处理
func (t *Trace) HttpTraceRequestEvent(method, path, ip string, port uint16, r *l7.RequestData) {
	if t == nil {
		return
	}
	t.addEvent()
	if method == "" {
		return
	}
	assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
	if err != nil {
		assumedAppID = 0
	}
	status := r.Status
	var attr []attribute.KeyValue
	attr = append(attr,
		semconv.HTTPURL(fmt.Sprintf("http://%s:%d%s", ip, port, path)),
		semconv.HTTPMethod(method),
		semconv.HTTPStatusCode(int(status)),
		attribute.Bool("http.status_error", status > 399),
		attribute.String("http.uri", path),
		attribute.String("http.ip", ip),
		attribute.Int64("http.assumed_app_id", assumedAppID),
		attribute.String("http.span_id", r.SpanId),
		attribute.Int("http.port", int(port)),
		attribute.String("http.src_addr", r.ComponentSAddr.String()),
		attribute.String("http.destination_addr", r.ComponentDAddr.String()),
		attribute.Bool("http.is_tls", r.IsTls),
	)
	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
	t.createTraceEvent(l7.ProtocolHTTP, ebpftracer.EventTypeL7Request.Int(), attr...)
}

func (t *Trace) FuncTraceQuery(funcname string, duration time.Duration, start uint64, end uint64) {
	if t == nil {
		return
	}
	t.addEvent()
	if funcname == "" {
		return
	}
	t.createTraceSpanNoTime2(funcname, duration, false, start, end)
	//t.createTraceSpanNoTime2(funcname, duration, false, start, end, attribute.Int("num", num))
}

func (t *Trace) createTraceSpanNoTime(name string, duration time.Duration, error bool, start uint64, end uint64, attrs ...attribute.KeyValue) {
	// end := time.Now()
	// start := end.Add(-duration)
	startTime := time.Unix(0, int64(start))
	endTime := time.Unix(0, int64(end))
	//fmt.Println("createTraceSpan:", t.ctx)
	_, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(startTime), trace.WithSpanKind(trace.SpanKindClient))
	span.SetAttributes(t.commonAttrs...)
	span.SetAttributes(attrs...)
	if error {
		span.SetStatus(codes.Error, "")
	}
	span.End(trace.WithTimestamp(endTime))
}

func (t *Trace) createTraceSpanNoTime2(name string, duration time.Duration, error bool, start uint64, end uint64) {
	// end := time.Now()
	// start := end.Add(-duration)
	//startTime := time.Unix(0, int64(start))
	//endTime := time.Unix(0, int64(end))
	////fmt.Println("createTraceSpan:", t.ctx)
	//_, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(startTime), trace.WithSpanKind(trace.SpanKindClient))
	//span.SetAttributes(t.commonAttrs...)
	//span.SetAttributes(attrs...)
	//if error {
	//	span.SetStatus(codes.Error, "")
	//}
	//span.End(trace.WithTimestamp(endTime))
	//attrs = append([]attribute.KeyValue{
	//	attribute.Int64("startAt", int64(start)),
	//	attribute.Int64("endAt", int64(end)),
	//})
	var attr []attribute.KeyValue
	t.appendTimestamp(&attr, start, end, int64(end-start))
	t.createFuncTraceEvent(name, ebpftracer.EventTypeFunEnt.Int(), attr...)
}