Fixed #JiraBug28023 sql增删改查,数据库默认值为[-]

containers/container_apm.go

@@ -24,6 +24,7 @@ import (
 	. "github.com/coroot/coroot-node-agent/utils/modelse"
 	"github.com/pkg/errors"
 	klog "github.com/sirupsen/logrus"
+	semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
 	"inet.af/netaddr"
 )
 
@@ -188,13 +189,13 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 
 			//apmTrace, ok := c.getTrace(r.TraceId)
 			apmTrace, err := c.getOrInitTrace(r.TraceId)
-			fmt.Println(err)
 			//fmt.Println("mysql r.TraceId:", r.TraceId)
 			//fmt.Println("ok:", ok)
 			//fmt.Println("traceMap:", len(c.traceMap))
 			if err == nil {
 				//apmTrace.MysqlTraceQuery(query, r.Status.Error(), r.Duration, conn.ActualDest)
-				apmTrace.PostGreSqlTraceQueryEvent(query, r, conn.ActualDest)
+				//apmTrace.PostGreSqlTraceQueryEvent(query, r, conn.ActualDest)
+				apmTrace.SQLTraceQueryEvent(l7.ProtocolPostgres, semconv.DBSystemPostgreSQL, query, r, conn.ActualDest)
 				c.SendEvent(apmTrace, r.TraceId)
 			}
 		}
@@ -211,13 +212,13 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 
 			//apmTrace, ok := c.getTrace(r.TraceId)
 			apmTrace, err := c.getOrInitTrace(r.TraceId)
-			fmt.Println(err)
 			//fmt.Println("mysql r.TraceId:", r.TraceId)
 			//fmt.Println("ok:", ok)
 			//fmt.Println("traceMap:", len(c.traceMap))
 			if err == nil {
 				//apmTrace.MysqlTraceQuery(query, r.Status.Error(), r.Duration, conn.ActualDest)
-				apmTrace.MysqlTraceQueryEvent(query, r, conn.ActualDest)
+				//apmTrace.MysqlTraceQueryEvent(query, r, conn.ActualDest)
+				apmTrace.SQLTraceQueryEvent(l7.ProtocolMysql, semconv.DBSystemMySQL, query, r, conn.ActualDest)
 				c.SendEvent(apmTrace, r.TraceId)
 			}
 		}
@@ -233,7 +234,8 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 			query := conn.dmParser.Parse(r.Payload, r.StatementId)
 			apmTrace, err := c.getOrInitTrace(r.TraceId)
 			if err == nil {
-				apmTrace.DmTraceQueryEvent(query, r, conn.ActualDest)
+				//apmTrace.DmTraceQueryEvent(query, r, conn.ActualDest)
+				apmTrace.SQLTraceQueryEvent(l7.ProtocolDM, semconv.DBSystemDaMengDB, query, r, conn.ActualDest)
 				c.SendEvent(apmTrace, r.TraceId)
 			}
 		}

ebpftracer/l7/l7.go

@@ -1,9 +1,9 @@
 package l7
 
 import (
+	"inet.af/netaddr"
 	"strconv"
 	"time"
-	"inet.af/netaddr"
 )
 
 type Protocol uint8
@@ -27,6 +27,10 @@ const (
 	ProtocolDM        Protocol = 14
 )
 
+func (p Protocol) Int() int {
+	return int(p)
+}
+
 func (p Protocol) String() string {
 	switch p {
 	case ProtocolTrace:
@@ -63,6 +67,42 @@ func (p Protocol) String() string {
 	return "UNKNOWN:" + strconv.Itoa(int(p))
 }
 
+func (p Protocol) ServiceNameString() string {
+	switch p {
+	case ProtocolTrace:
+		return "TRACE"
+	case ProtocolHTTP:
+		return "HTTP"
+	case ProtocolPostgres:
+		return "POSTGRESQL"
+	case ProtocolRedis:
+		return "REDIS"
+	case ProtocolMemcached:
+		return "Memcached"
+	case ProtocolMysql:
+		return "MYSQL"
+	case ProtocolMongo:
+		return "Mongo"
+	case ProtocolKafka:
+		return "Kafka"
+	case ProtocolCassandra:
+		return "Cassandra"
+	case ProtocolRabbitmq:
+		return "Rabbitmq"
+	case ProtocolNats:
+		return "NATS"
+	case ProtocolHTTP2:
+		return "HTTP2"
+	case ProtocolDubbo2:
+		return "Dubbo2"
+	case ProtocolDNS:
+		return "DNS"
+	case ProtocolDM:
+		return "DM"
+	}
+	return "UNKNOWN:" + strconv.Itoa(int(p))
+}
+
 type Method uint8
 
 const (
@@ -156,8 +196,8 @@ type RequestData struct {
 	SpanId            string
 	StartAt           uint64
 	EndAt             uint64
-    SAddr          	  netaddr.IPPort
-	DAddr          	  netaddr.IPPort
+	SAddr             netaddr.IPPort
+	DAddr             netaddr.IPPort
 	ComponentSAddr    netaddr.IPPort
 	ComponentDAddr    netaddr.IPPort
 	ParentSpanContext struct {

ebpftracer/tracer.go

@@ -399,6 +399,10 @@ func (t *Tracer) UnlinkEbpfProg() {
 	}
 }
 
+func (t EventType) Int() int {
+	return int(t)
+}
+
 func (t EventType) String() string {
 	switch t {
 	case EventTypeProcessStart:
@@ -482,13 +486,13 @@ type l7Event struct {
 	TraceStart          uint32
 	TraceEnd            uint32
 	EventCount          uint32
-	Sport				uint16
-    Dport				uint16
-    SAddr          		[16]byte
-	DAddr          		[16]byte
-	ComponentSport		uint16
-    ComponentDport		uint16
-    ComponentSAddr      [16]byte
+	Sport               uint16
+	Dport               uint16
+	SAddr               [16]byte
+	DAddr               [16]byte
+	ComponentSport      uint16
+	ComponentDport      uint16
+	ComponentSAddr      [16]byte
 	ComponentDAddr      [16]byte
 	AssumedAppId        HashByte
 	SpanId              HashByte
@@ -710,23 +714,23 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy
 
 			payload := reader.Bytes()
 			req := &l7.RequestData{
-				Protocol:     l7.Protocol(v.Protocol),
-				Status:       l7.Status(v.Status),
-				Duration:     time.Duration(v.Duration),
-				Method:       l7.Method(v.Method),
-				StatementId:  v.StatementId,
-				TraceId:      v.TraceId,
-				TraceStart:   v.TraceStart,
-				TraceEnd:     v.TraceEnd,
-				EventCount:   v.EventCount,
-				AssumedAppId: hex.EncodeToString(v.AssumedAppId[:]),
-				SpanId:       hex.EncodeToString(v.SpanId[:]),
-				StartAt:      v.StartAt,
-				EndAt:        v.EndtAt,
-				ComponentSAddr: ipPort(v.ComponentSAddr,v.ComponentSport),
-				ComponentDAddr: ipPort(v.ComponentDAddr,v.ComponentDport),
+				Protocol:       l7.Protocol(v.Protocol),
+				Status:         l7.Status(v.Status),
+				Duration:       time.Duration(v.Duration),
+				Method:         l7.Method(v.Method),
+				StatementId:    v.StatementId,
+				TraceId:        v.TraceId,
+				TraceStart:     v.TraceStart,
+				TraceEnd:       v.TraceEnd,
+				EventCount:     v.EventCount,
+				AssumedAppId:   hex.EncodeToString(v.AssumedAppId[:]),
+				SpanId:         hex.EncodeToString(v.SpanId[:]),
+				StartAt:        v.StartAt,
+				EndAt:          v.EndtAt,
+				ComponentSAddr: ipPort(v.ComponentSAddr, v.ComponentSport),
+				ComponentDAddr: ipPort(v.ComponentDAddr, v.ComponentDport),
 			}
-			if req.Protocol == l7.ProtocolHTTP{
+			if req.Protocol == l7.ProtocolHTTP {
 				klog.Infof("runEventsReader ComponentSAddr.String %s", req.ComponentSAddr.String())
 				klog.Infof("runEventsReader ComponentDAddr.String %s", req.ComponentDAddr.String())
 			}

flags/flags.go

@@ -12,6 +12,7 @@ import (
 	"net/url"
 	"os"
 	"path"
+	"runtime"
 	"strings"
 )
 
@@ -20,6 +21,7 @@ var (
 	ConfigServer        = kingpin.Flag("config-server", "The URL of the endpoint to send traces to").Envar("CONFIG_SERVER").Default("http://10.0.16.250:18080").String()
 	DataServer          = kingpin.Flag("data-server", "The URL of the endpoint to send traces to").Envar("DATA_SERVER").Default("http://10.0.16.250:18080").String()
 	DumpApps            = kingpin.Flag("dump", "Dump app snap").Default("false").Bool()
+	Version             = kingpin.Flag("version", "show app version").Short('v').Bool()
 	LogLevel            = kingpin.Flag("log-level", "Log level").Envar("LOG_LEVEL").Default("info").String()
 	EbpfFilePath        = kingpin.Flag("ebpf-path", "Set ebpf file path").Envar("EBPF_FILE").Default("").String()
 	CommonIni           = kingpin.Flag("common.ini", "Set ebpf file path").Envar("COMMON_INI").Default("/opt/cloudwise/omniagent/conf/common.ini").String()
@@ -67,6 +69,11 @@ var (
 	HostDirPathPrefix = kingpin.Flag("host-dir-path-prefix", "Set the prefix of path about the mount point of the host directory").Envar("HOST_DIR_PATH_PREFIX").Default("").String()
 )
 
+var AgentName = "euspace"
+var AgentVersion = "0.0.1"
+var GitCommit = ""
+var BuildDate = ""
+
 func GetString(fl *string) string {
 	if fl == nil {
 		return ""
@@ -81,6 +88,9 @@ func init() {
 
 	kingpin.HelpFlag.Short('h').Hidden()
 	kingpin.Parse()
+	if *Version {
+		ShowVersion()
+	}
 	if *DumpApps {
 		DumpTableFeatures()
 	}
@@ -203,3 +213,12 @@ func DumpTableFeatures() {
 	fmt.Println(t.Render())
 	os.Exit(0)
 }
+
+func ShowVersion() {
+	fmt.Printf("%-20s %-8s\n", "Name:", AgentName)
+	fmt.Printf("%-20s %-8s\n", "Version:", AgentVersion)
+	fmt.Printf("%-20s %-8s\n", "Commit:", GitCommit)
+	fmt.Printf("%-20s %-8s\n", "Date:", BuildDate)
+	fmt.Printf("%-20s %-8s\n", "go version:", runtime.Version())
+	os.Exit(0)
+}

main.go

@@ -273,7 +273,7 @@ func main() {
 				metric.GetName() != "process_net_tcp_bytes_sent_total" &&
 				metric.GetName() != "process_net_tcp_data_latency_time" &&
 				metric.GetName() != "process_net_tcp_flow_duration_time" &&
-				metric.GetName() != "process_net_tcp_connection_establish_time"{
+				metric.GetName() != "process_net_tcp_connection_establish_time" {
 				continue
 			}
 
@@ -324,9 +324,9 @@ func main() {
 		// 创建请求
 		urlRoute := "/api/v2/ebpf/receive"
 
-		log.Infoln("send url is ", *flags.DataServer + *flags.ServerPrefix  + urlRoute)
+		log.Infoln("send url is ", *flags.DataServer+*flags.ServerPrefix+urlRoute)
 		// req, err := http.NewRequest("POST", "http://10.0.7.115:18080/api/v2/ebpf/receive", bytes.NewBuffer(jsonData))
-		req, err := http.NewRequest("POST", *flags.DataServer + *flags.ServerPrefix  + urlRoute, bytes.NewBuffer(jsonData))
+		req, err := http.NewRequest("POST", *flags.DataServer+*flags.ServerPrefix+urlRoute, bytes.NewBuffer(jsonData))
 		if err != nil {
 			fmt.Println("Error:", err)
 			return
@@ -351,16 +351,16 @@ func main() {
 		}
 		defer response.Body.Close()
 
-		// 读取响应内容  
-		responseData, err := io.ReadAll(response.Body)  
-		if err != nil {  
-			log.Infoln("Error:", err)  
-			return  
-		}  
+		// 读取响应内容
+		responseData, err := io.ReadAll(response.Body)
+		if err != nil {
+			log.Infoln("Error:", err)
+			return
+		}
 
-		// 输出响应状态码和响应正文  
-		log.Infoln("Status Code:", response.StatusCode)  
-		log.Infoln("Response Body:", string(responseData)) 
+		// 输出响应状态码和响应正文
+		log.Infoln("Status Code:", response.StatusCode)
+		log.Infoln("Response Body:", string(responseData))
 	}
 	sendNetDataDone := make(chan struct{})
 	go func() {

pkg/go.opentelemetry.io/otel/exporters/otlp/otlptrace/apm_exporter.go

@@ -177,14 +177,14 @@ func tracetransformData(sdl []tracesdk.ReadOnlySpan) map[int][]RootDataT {
 				case l7.ProtocolHTTP:
 					buildHttpMapFromEvent(&mNode, event)
 				case l7.ProtocolMysql:
-					buildMysqlMapEvent(&mNode, event)
+					buildSQLMapEvent(&mNode, event)
 				case l7.ProtocolRedis:
 					buildRedisMapEvent(&mNode, event)
 				// dm
 				case l7.ProtocolDM:
-					buildDMMapEvent(&mNode, event)
+					buildSQLMapEvent(&mNode, event)
 				case l7.ProtocolPostgres:
-					buildPostGreSqlMapEvent(&mNode, event)
+					buildSQLMapEvent(&mNode, event)
 				}
 			}
 
@@ -857,11 +857,11 @@ func buildHttpMapFromEvent(mNode *MapInfoT, event tracesdk.Event) {
 //	}
 //}
 
-func buildPostGreSqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
-	mNode.Dbn = "unknown"
-	mNode.ServiceName = POSTGRESQL_SERVICE_NAME
+func buildSQLMapEvent(mNode *MapInfoT, event tracesdk.Event) {
+	mNode.Dbn = "-"
+	mNode.ServiceName = l7.Protocol(event.ProtocolType).ServiceNameString()
 	mNode.ServiceType = SQL_SERVICE_TYPE
-	mNode.MethodName = "database/sql.Query()"
+	//mNode.MethodName = "database/sql.Query()"
 	for _, attr := range event.Attributes {
 		//fmt.Println(attr.Key, ":", attr.Value.AsInterface())
 		switch attr.Key {
@@ -871,6 +871,7 @@ func buildPostGreSqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
 			mNode.Port = attr.Value.AsInt64()
 		case "db.statement":
 			query := attr.Value.AsString()
+			mNode.MethodName = query
 			mNode.Ps = []string{query}
 			words := strings.Fields(query)
 			if len(words) > 0 {
@@ -890,9 +891,9 @@ func buildPostGreSqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
 	}
 }
 
-func buildMysqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
-	mNode.Dbn = "unknown"
-	mNode.ServiceName = MYSQL_SERVICE_NAME
+func buildPostGreSqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
+	mNode.Dbn = "-"
+	mNode.ServiceName = POSTGRESQL_SERVICE_NAME
 	mNode.ServiceType = SQL_SERVICE_TYPE
 	//mNode.MethodName = "database/sql.Query()"
 	for _, attr := range event.Attributes {
@@ -904,7 +905,6 @@ func buildMysqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
 			mNode.Port = attr.Value.AsInt64()
 		case "db.statement":
 			query := attr.Value.AsString()
-			mNode.MethodName = query
 			mNode.Ps = []string{query}
 			words := strings.Fields(query)
 			if len(words) > 0 {
@@ -924,11 +924,11 @@ func buildMysqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
 	}
 }
 
-func buildDMMapEvent(mNode *MapInfoT, event tracesdk.Event) {
-	mNode.Dbn = "TEST"
-	mNode.ServiceName = DM_SERVICE_NAME
+func buildMysqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
+	mNode.Dbn = "-"
+	mNode.ServiceName = MYSQL_SERVICE_NAME
 	mNode.ServiceType = SQL_SERVICE_TYPE
-	mNode.MethodName = "database/sql.Query()"
+	//mNode.MethodName = "database/sql.Query()"
 	for _, attr := range event.Attributes {
 		//fmt.Println(attr.Key, ":", attr.Value.AsInterface())
 		switch attr.Key {
@@ -938,6 +938,7 @@ func buildDMMapEvent(mNode *MapInfoT, event tracesdk.Event) {
 			mNode.Port = attr.Value.AsInt64()
 		case "db.statement":
 			query := attr.Value.AsString()
+			mNode.MethodName = query
 			mNode.Ps = []string{query}
 			words := strings.Fields(query)
 			if len(words) > 0 {
@@ -957,12 +958,12 @@ func buildDMMapEvent(mNode *MapInfoT, event tracesdk.Event) {
 	}
 }
 
-func buildRedisMap(mNode *MapInfoT, sd apmTraceSpan) {
-	mNode.ServiceName = REDIS_SERVICE_NAME
-	mNode.ServiceType = NOSQL_SERVICE_TYPE
-	//mNode.MethodName = span(sd).Name + " query"
-	mNode.MethodName = "redis.Do()"
-	for _, attr := range sd.Attributes() {
+func buildDMMapEvent(mNode *MapInfoT, event tracesdk.Event) {
+	mNode.Dbn = "TEST"
+	mNode.ServiceName = DM_SERVICE_NAME
+	mNode.ServiceType = SQL_SERVICE_TYPE
+	mNode.MethodName = "database/sql.Query()"
+	for _, attr := range event.Attributes {
 		//fmt.Println(attr.Key, ":", attr.Value.AsInterface())
 		switch attr.Key {
 		case "net.peer.name":
@@ -976,9 +977,15 @@ func buildRedisMap(mNode *MapInfoT, sd apmTraceSpan) {
 			if len(words) > 0 {
 				mNode.OperType = strings.ToUpper(words[0])
 			}
-		case "nosql.src_addr":
+		case "sql.exception":
+			if attr.Value.AsBool() {
+				mNode.Exception = 1
+			} else {
+				mNode.Exception = 0
+			}
+		case "sql.src_addr":
 			mNode.SrcAddr = attr.Value.AsString()
-		case "nosql.destination_addr":
+		case "sql.destination_addr":
 			mNode.DestinationAddr = attr.Value.AsString()
 		}
 	}

tracing/apm_tracing.go

@@ -14,6 +14,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	"inet.af/netaddr"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"time"
 )
@@ -281,6 +282,10 @@ func (t *Trace) GetSpan() trace.Span {
 
 func (t *Trace) createTraceEvent(name string, eventType int, l7Type int, attrs ...attribute.KeyValue) {
 	t.span.AddEventApm(name, eventType, l7Type, trace.WithAttributes(attrs...))
+	//atomic.AddUint32(t.currenEventCount, 1)
+}
+
+func (t *Trace) addEvent() {
 	atomic.AddUint32(t.currenEventCount, 1)
 }
 
@@ -309,71 +314,29 @@ func (t *Trace) MysqlTraceQuery(query string, error bool, duration time.Duration
 	)
 }
 
-func (t *Trace) MysqlTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
-	fmt.Println("query", query)
-	if t == nil {
-		return
+func isCURDOperation(q string) bool {
+	if len(q) < 6 {
+		return false
 	}
-
-	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...)
+	q = strings.ToUpper(q[:6])
+	return q == "SELECT" || q == "INSERT" || q == "UPDATE" || q == "DELETE"
 }
 
-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) {
+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()
 
-	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 == "" {
+	// 只保留增删改查基本操作
+	if !isCURDOperation(query) {
 		return
 	}
-	l7Type := int(l7.ProtocolDM)
+
 	var attr []attribute.KeyValue
 	attr = append(attr,
-		semconv.DBSystemDaMengDB,
+		semconvVal,
 		semconv.DBStatement(query),
 		semconv.NetPeerName(destination.IP().String()),
 		semconv.NetPeerPort(int(destination.Port())),
@@ -385,26 +348,109 @@ func (t *Trace) DmTraceQueryEvent(query string, r *l7.RequestData, destination n
 	attr = append(attr,
 		attribute.Bool("sql.exception", r.Status.Error()),
 	)
-	t.createTraceEvent(l7.ProtocolHTTP.String(), int(ebpftracer.EventTypeL7Request), l7Type, attr...)
+	t.createTraceEvent(l7Type.String(), ebpftracer.EventTypeL7Request.Int(), l7Type.Int(), 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) 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) RedisTraceQueryEvent(cmd, args string, r *l7.RequestData, destination netaddr.IPPort) {
-	if t == nil || cmd == "" {
+	if t == nil {
+		return
+	}
+	t.addEvent()
+	if cmd == "" {
 		return
 	}
 	statement := cmd
@@ -430,7 +476,7 @@ func (t *Trace) DNSTraceQueryEvent(r *l7.RequestData, _type, fqdn string) {
 	if t == nil {
 		return
 	}
-
+	t.addEvent()
 	var attr []attribute.KeyValue
 	attr = append(attr,
 		attribute.String("dns.type", _type),
@@ -464,7 +510,11 @@ func (t *Trace) HttpTraceRequest(method, path, ip string, port uint16, r *l7.Req
 
 // 新增事件处理
 func (t *Trace) HttpTraceRequestEvent(method, path, ip string, port uint16, r *l7.RequestData) {
-	if t == nil || method == "" {
+	if t == nil {
+		return
+	}
+	t.addEvent()
+	if method == "" {
 		return
 	}
 	assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
@@ -491,7 +541,11 @@ func (t *Trace) HttpTraceRequestEvent(method, path, ip string, port uint16, r *l
 }
 
 func (t *Trace) FuncTraceQuery(funcname string, duration time.Duration, start uint64, end uint64) {
-	if t == nil || funcname == "" {
+	if t == nil {
+		return
+	}
+	t.addEvent()
+	if funcname == "" {
 		return
 	}
 	t.createTraceSpanNoTime2(funcname, duration, false, start, end)