Feature #TASK_QT-31498 【Q4】eBPF-Go Memcache 适配

containers/container_apm.go

@@ -244,12 +244,18 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 	stats := c.l7Stats.get(r.Protocol, conn.Dest, conn.ActualDest)
 	//trace := tracing.NewTrace(string(c.id), conn.ActualDest)
 	switch r.Protocol {
+	/**
+	 * HTTP
+	 */
 	case l7.ProtocolHTTP:
 		if c.AppInfo.AppName != "" {
 			klog.Debugf("[%s] ->>>>> curl -> %s payload:[%s]", c.AppInfo.AppName, conn.ActualDest, r.Payload)
 		}
 
 		stats.observe(r.Status.Http(), "", r.Duration)
+	/**
+	 * HTTP2
+	 */
 	case l7.ProtocolHTTP2:
 		if conn.http2Parser == nil {
 			conn.http2Parser = l7.NewHttp2Parser()
@@ -259,6 +265,9 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 			stats.observe(req.Status.Http(), "", req.Duration)
 			//trace.Http2Request(req.Method, req.Path, req.Scheme, req.Status, req.Duration)
 		}
+	/**
+	 * PostgreSQL
+	 */
 	case l7.ProtocolPostgres:
 		if r.Method != l7.MethodStatementClose {
 			stats.observe(r.Status.String(), "", r.Duration)
@@ -275,7 +284,7 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 			query := conn.postgresParser.Parse(r.Payload)
 			//trace.MysqlQuery(query, r.Status.Error(), r.Duration)
 			if c.AppInfo.AppName != "" {
-				klog.Debugf("[%s] ->>>>> Postgres -> %s payload:[%s]", c.AppInfo.AppName, conn.ActualDest, query)
+				klog.Debugf("[%s] ->>>>> %s -> %s payload:[%s]", c.AppInfo.AppName, r.Protocol.String(), conn.ActualDest, query)
 			}
 			//apmTrace, ok := c.getTrace(r.TraceId)
 			apmTrace, err := c.getOrInitTrace(r.TraceId)
@@ -285,10 +294,13 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 			if err == nil {
 				//apmTrace.MysqlTraceQuery(query, r.Status.Error(), r.Duration, conn.ActualDest)
 				//apmTrace.PostGreSqlTraceQueryEvent(query, r, conn.ActualDest)
-				apmTrace.SQLTraceQueryEvent(l7.ProtocolPostgres, semconv.DBSystemPostgreSQL, query, r, conn.ActualDest)
+				apmTrace.SQLTraceQueryEvent(r.Protocol, semconv.DBSystemPostgreSQL, query, r, conn.ActualDest)
 				c.SendEvent(apmTrace, r.TraceId)
 			}
 		}
+	/**
+	 * Mysql
+	 */
 	case l7.ProtocolMysql:
 		if r.Method != l7.MethodStatementClose {
 			stats.observe(r.Status.String(), "", r.Duration)
@@ -300,7 +312,7 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 			query := conn.mysqlParser.Parse(r.Payload, r.StatementId)
 			//trace.MysqlQuery(query, r.Status.Error(), r.Duration)
 			if c.AppInfo.AppName != "" {
-				klog.Debugf("[%s] ->>>>> Mysql -> %s payload:[%s]", c.AppInfo.AppName, conn.ActualDest, query)
+				klog.Debugf("[%s] ->>>>> %s -> %s payload:[%s]", c.AppInfo.AppName, r.Protocol.String(), conn.ActualDest, query)
 			}
 			//apmTrace, ok := c.getTrace(r.TraceId)
 			apmTrace, err := c.getOrInitTrace(r.TraceId)
@@ -320,7 +332,9 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 				c.SendEvent(apmTrace, r.TraceId)
 			}
 		}
-
+	/**
+	 * DM (达梦数据库)
+	 */
 	case l7.ProtocolDM:
 		//统计dm的query次数
 		stats.observe(r.Status.String(), "", r.Duration)
@@ -331,24 +345,42 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 			}
 			query := conn.dmParser.Parse(r.Payload, r.StatementId)
 			if c.AppInfo.AppName != "" {
-				klog.Debugf("[%s] ->>>>> Mysql -> %s DMSQL:[%s]", c.AppInfo.AppName, conn.ActualDest, query)
+				klog.Debugf("[%s] ->>>>> %s -> %s payload:[%s]", c.AppInfo.AppName, r.Protocol.String(), conn.ActualDest, query)
 			}
 
 			apmTrace, err := c.getOrInitTrace(r.TraceId)
 			if err == nil {
 				//apmTrace.DmTraceQueryEvent(query, r, conn.ActualDest)
-				apmTrace.SQLTraceQueryEvent(l7.ProtocolDM, semconv.DBSystemDaMengDB, query, r, conn.ActualDest)
+				apmTrace.SQLTraceQueryEvent(r.Protocol, semconv.DBSystemDaMengDB, query, r, conn.ActualDest)
 				c.SendEvent(apmTrace, r.TraceId)
 			}
 		}
+	/**
+	 * Memcached
+	 */
 	case l7.ProtocolMemcached:
 		stats.observe(r.Status.String(), "", r.Duration)
 		if c.l7Attach && c.valuableTrace(r.TraceId) {
-
+			cmd, items := l7.ParseMemcached(r.Payload)
+			if c.AppInfo.AppName != "" {
+				klog.Debugf("[%s] ->>>>> %s -> %s payload:[%s]", c.AppInfo.AppName, r.Protocol.String(), conn.ActualDest, cmd+" "+strings.Join(items, " "))
+			}
+			apmTrace, err := c.getOrInitTrace(r.TraceId)
+			if err == nil {
+				statement := cmd
+				if len(items) == 1 {
+					statement += " " + items[0]
+				} else if len(items) > 1 {
+					joined := fmt.Sprintf("[%s]", strings.Join(items, " "))
+					statement += " " + joined
+				}
+				apmTrace.NoSQLTraceQueryEvent(r.Protocol, semconv.DBSystemMemcached, cmd, statement, r, conn.ActualDest)
+				c.SendEvent(apmTrace, r.TraceId)
+			}
 		}
-
-		//cmd, items := l7.ParseMemcached(r.Payload)
-		//trace.MemcachedQuery(cmd, items, r.Status.Error(), r.Duration)
+	/**
+	 * Redis
+	 */
 	case l7.ProtocolRedis:
 		stats.observe(r.Status.String(), "", r.Duration)
 		if c.l7Attach && c.valuableTrace(r.TraceId) {
@@ -357,17 +389,23 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 			//fmt.Println("args", args)
 			//apmTrace, ok := c.getTrace(r.TraceId)
 			if c.AppInfo.AppName != "" {
-				klog.Debugf("[%s] ->>>>> Redis -> %s DMSQL:[%s]", c.AppInfo.AppName, conn.ActualDest, cmd)
+				klog.Debugf("[%s] ->>>>> %s -> %s payload:[%s]", c.AppInfo.AppName, r.Protocol.String(), conn.ActualDest, cmd)
 			}
 
 			apmTrace, err := c.getOrInitTrace(r.TraceId)
 			if err == nil {
-				//apmTrace.RedisTraceQuery(cmd, args, r.Status.Error(), r.Duration)
-				apmTrace.RedisTraceQueryEvent(cmd, args, r, conn.ActualDest)
+				statement := cmd
+				if args != "" {
+					statement += " " + args
+				}
+				apmTrace.NoSQLTraceQueryEvent(r.Protocol, semconv.DBSystemRedis, cmd, statement, r, conn.ActualDest)
 				c.SendEvent(apmTrace, r.TraceId)
 			}
 		}
 		//trace.RedisQuery(cmd, args, r.Status.Error(), r.Duration)
+	/**
+	 * gRPC
+	 */
 	case l7.ProtocolGrpc:
 		klog.Debugln("enter the l7.ProtocolGrpc")
 		stats.observe(r.Status.String(), "", r.Duration)
@@ -378,6 +416,9 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 				c.SendEvent(apmTrace, r.TraceId)
 			}
 		}
+	/**
+	 * MongoDB
+	 */
 	case l7.ProtocolMongo:
 		stats.observe(r.Status.String(), "", r.Duration)
 		if c.l7Attach && c.valuableTrace(r.TraceId) {
@@ -388,19 +429,28 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 
 			apmTrace, err := c.getOrInitTrace(r.TraceId)
 			if err == nil {
-				//apmTrace.RedisTraceQuery(cmd, args, r.Status.Error(), r.Duration)
-				apmTrace.MongoTraceQueryEvent(query, r, conn.ActualDest)
+				// MongoDB query 格式通常是 JSON，如 {"insert":"users"} 或 {"find":"users","filter":{...}}
+				apmTrace.NoSQLTraceQueryEvent(r.Protocol, semconv.DBSystemMongoDB, "", query, r, conn.ActualDest)
 				c.SendEvent(apmTrace, r.TraceId)
 			}
 		}
+	/**
+	 * Kafka / Cassandra
+	 */
 	case l7.ProtocolKafka, l7.ProtocolCassandra:
 		stats.observe(r.Status.String(), "", r.Duration)
 		if c.l7Attach && c.valuableTrace(r.TraceId) {
 		}
+	/**
+	 * RabbitMQ / NATS
+	 */
 	case l7.ProtocolRabbitmq, l7.ProtocolNats:
 		stats.observe(r.Status.String(), r.Method.String(), 0)
 		if c.l7Attach && c.valuableTrace(r.TraceId) {
 		}
+	/**
+	 * Dubbo2
+	 */
 	case l7.ProtocolDubbo2:
 		stats.observe(r.Status.String(), "", r.Duration)
 		if c.l7Attach && c.valuableTrace(r.TraceId) {

ebpftracer/ebpf/l7/l7.c

@@ -598,6 +598,7 @@ int trace_enter_write(void *ctx, __u64 fd, __u16 is_tls, char *buf, __u64 size,
         cw_bpf_debug("[Enter][Redis]:TGID:%d|type:%s|FD:%d\n",k.pid,"type",k.fd);
         req->protocol = PROTOCOL_REDIS;
     } else if (is_memcached_query(payload, size)) {
+	    cw_bpf_debug("[Enter][MEMCACHE]:TGID:%d|type:%s|FD:%d\n",k.pid,"type",k.fd);
         req->protocol = PROTOCOL_MEMCACHED;
     } else if (is_mysql_query(payload, size, &req->request_type)) {
         cw_bpf_debug("[Enter][Mysql]:thread_id:%d\n",tid);
@@ -995,9 +996,14 @@ int trace_exit_read_common(void *ctx, __u64 id, __u32 pid, __u16 is_tls, long in
         e->component_dport = conn->dport;
         __builtin_memcpy(&e->component_saddr, &conn->saddr, sizeof(e->component_saddr));
         __builtin_memcpy(&e->component_daddr, &conn->daddr, sizeof(e->component_daddr));
-		response = is_redis_response(payload, ret, &e->status);
+		response = is_redis_response(payload, ret, &e->status, e->error_message);
 	} else if (e->protocol == PROTOCOL_MEMCACHED) {
-		response = is_memcached_response(payload, ret, &e->status);
+		cw_bpf_debug("[Response][MEMCACHE]:thread_id:%d\n", tid);
+		e->component_sport = conn->sport;
+		e->component_dport = conn->dport;
+		__builtin_memcpy(&e->component_saddr, &conn->saddr, sizeof(e->component_saddr));
+		__builtin_memcpy(&e->component_daddr, &conn->daddr, sizeof(e->component_daddr));
+		response = is_memcached_response(payload, ret, &e->status, e->error_message);
 	} else if (e->protocol == PROTOCOL_MYSQL) {
 		cw_bpf_debug("[Response][Mysql]:thread_id:%d\n", tid);
 //		__u64 trace_id = get_apm_trace_id(pid, tid);

ebpftracer/ebpf/l7/memcached.c

@@ -1,7 +1,6 @@
 // https://github.com/memcached/memcached/blob/master/doc/protocol.txt
 static __always_inline
 int is_memcached_query(char *buf, __u64 buf_size) {
-	return 0;
     if (buf_size < 9) {
         return 0;
     }
@@ -50,8 +49,7 @@ int is_memcached_query(char *buf, __u64 buf_size) {
 }
 
 static __always_inline
-int is_memcached_response(char *buf, __u64 buf_size, __u32 *status) {
-	return 0;
+int is_memcached_response(char *buf, __u64 buf_size, __u32 *status, unsigned char *error_message) {
     char r[3];
     bpf_read(buf, r);
     char end[2];
@@ -80,6 +78,13 @@ int is_memcached_response(char *buf, __u64 buf_size, __u32 *status) {
         *status = STATUS_OK;
         return 1;
     }
+    if (r[0] == 'E' && r[1] == 'N' && r[2] == 'D') { //END (key not found for get/gets)
+        *status = STATUS_FAILED;
+        if (error_message) {
+            bpf_probe_read(error_message, 27, "key not found for get/gets");
+        }
+        return 1;
+    }
     if (r[0] == 'E' && r[1] == 'X' && r[2] == 'I') { //EXISTS
         *status = STATUS_OK;
         return 1;
@@ -88,12 +93,34 @@ int is_memcached_response(char *buf, __u64 buf_size, __u32 *status) {
         *status = STATUS_FAILED;
         return 1;
     }
-    if (r[0] == 'C' && r[1] == 'L' && r[2] == 'I') { //CLIENT_ERROR
+    if (r[0] == 'C' && r[1] == 'L' && r[2] == 'I') { //CLIENT_ERROR <error>\r\n
         *status = STATUS_FAILED;
-        return 1;
-    }
-    if (r[0] == 'S' && r[1] == 'E' && r[2] == 'R') { //SERVER_ERROR
+        // 提取错误消息：跳过 "CLIENT_ERROR " (14个字符)，到 \r\n 之前
+        if (buf_size > 14 && error_message) {
+            __u64 error_msg_len = buf_size - 16; // 总长度 - "CLIENT_ERROR " (14) - "\r\n" (2)
+            if (error_msg_len > ERROR_MSG_PAYLOAD_SIZE - 1) {
+                error_msg_len = ERROR_MSG_PAYLOAD_SIZE - 1;
+            }
+            if (error_msg_len > 0) {
+                bpf_probe_read(error_message, error_msg_len, buf + 14);
+                error_message[error_msg_len] = '\0';
+            }
+        }
+        return 1;
+    }
+    if (r[0] == 'S' && r[1] == 'E' && r[2] == 'R') { //SERVER_ERROR <error>\r\n
         *status = STATUS_FAILED;
+        // 提取错误消息：跳过 "SERVER_ERROR " (14个字符)，到 \r\n 之前
+        if (buf_size > 14 && error_message) {
+            __u64 error_msg_len = buf_size - 16; // 总长度 - "SERVER_ERROR " (14) - "\r\n" (2)
+            if (error_msg_len > ERROR_MSG_PAYLOAD_SIZE - 1) {
+                error_msg_len = ERROR_MSG_PAYLOAD_SIZE - 1;
+            }
+            if (error_msg_len > 0) {
+                bpf_probe_read(error_message, error_msg_len, buf + 14);
+                error_message[error_msg_len] = '\0';
+            }
+        }
         return 1;
     }
     if (r[0] >= '0' && r[0] <= '9') { // incr/decr response: <value>\r\n

ebpftracer/ebpf/l7/mysql.c

@@ -76,7 +76,7 @@ int is_mysql_response(char *buf, __u64 buf_size, __u8 request_type, __u32 *state
 			error_message[error_message_len] = '\0'; // 确保字符串结束
 		}
 		// 输出错误信息到内核日志
-		bpf_printk("MySQL Error %d: %s", error_code, error_message);
+		cw_bpf_debug("MySQL Error %d: %s", error_code, error_message);
 		*status = STATUS_FAILED;
 		return 1;
 	}

ebpftracer/ebpf/l7/redis.c

@@ -23,7 +23,7 @@ int is_redis_query(char *buf, __u64 buf_size) {
 }
 
 static __always_inline
-int is_redis_response(char *buf, __u64 buf_size, __u32 *status) {
+int is_redis_response(char *buf, __u64 buf_size, __u32 *status, unsigned char *error_message) {
     char type;
     bpf_read(buf, type);
     char end[2];
@@ -38,6 +38,18 @@ int is_redis_response(char *buf, __u64 buf_size, __u32 *status) {
     }
     if (type == '-') {
         *status = STATUS_FAILED;
+        // 提取错误消息：跳过开头的 '-'，到 \r\n 之前
+        // Redis 错误格式：-ERR <error message>\r\n 或 -WRONGTYPE <error message>\r\n
+        if (buf_size > 1 && error_message) {
+            __u64 error_msg_len = buf_size - 3; // 总长度 - '-' (1) - "\r\n" (2)
+            if (error_msg_len > ERROR_MSG_PAYLOAD_SIZE - 1) {
+                error_msg_len = ERROR_MSG_PAYLOAD_SIZE - 1;
+            }
+            if (error_msg_len > 0) {
+                bpf_probe_read(error_message, error_msg_len, buf + 1);
+                error_message[error_msg_len] = '\0';
+            }
+        }
         return 1;
     }
     return 0;

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

@@ -191,8 +191,8 @@ func tracetransformData(sdl []tracesdk.ReadOnlySpan) map[int][]RootDataT {
 					buildDNSMapEvent(&mNode, event)
 				case l7.ProtocolMysql, l7.ProtocolMariaDB, l7.ProtocolPostgres, l7.ProtocolDM:
 					buildSQLMapEvent(&mNode, event)
-				case l7.ProtocolRedis:
-					buildRedisMapEvent(&mNode, event)
+				case l7.ProtocolRedis, l7.ProtocolMemcached:
+					buildNoSqlMapEvent(&mNode, event)
 				case l7.ProtocolMongo:
 					buildMongoMapEvent(&mNode, event)
 				case l7.ProtocolGrpc:
@@ -1096,13 +1096,10 @@ func buildDMMapEvent(mNode *MapInfoT, event tracesdk.Event) {
 	}
 }
 
-func buildRedisMapEvent(mNode *MapInfoT, event tracesdk.Event) {
-	mNode.ServiceName = REDIS_SERVICE_NAME
+func buildNoSqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
+	mNode.ServiceName = l7.Protocol(event.ProtocolType).ServiceNameString()
 	mNode.ServiceType = NOSQL_SERVICE_TYPE
-	//mNode.MethodName = span(sd).Name + " query"
-	//mNode.MethodName = "redis.Do()"
 	for _, attr := range event.Attributes {
-		//fmt.Println(attr.Key, ":", attr.Value.AsInterface())
 		switch attr.Key {
 		case "net.peer.name":
 			mNode.Ip = attr.Value.AsString()
@@ -1112,14 +1109,16 @@ func buildRedisMapEvent(mNode *MapInfoT, event tracesdk.Event) {
 			query := attr.Value.AsString()
 			mNode.MethodName = query
 			mNode.Ps = []string{query}
-			//words := strings.Fields(query)
-			//if len(words) > 0 {
-			//	mNode.OperType = strings.ToUpper(words[0])
-			//}
 		case "nosql.src_addr":
 			mNode.SrcAddr = attr.Value.AsString()
 		case "nosql.destination_addr":
 			mNode.DestinationAddr = attr.Value.AsString()
+		case "nosql.exception":
+			if attr.Value.AsBool() {
+				mNode.Exception = 1
+			}
+		case "nosql.exception_msg":
+			mNode.ExceptionMsg = attr.Value.AsString()
 		}
 	}
 }

tracing/apm_tracing.go

@@ -3,7 +3,6 @@ package tracing
 import (
 	"context"
 	"fmt"
-	klog "github.com/sirupsen/logrus"
 	"os"
 	"strconv"
 	"strings"
@@ -11,6 +10,8 @@ import (
 	"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"
@@ -129,7 +130,7 @@ func (t *Trace) AllEventReady(traceID uint64) bool {
 	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) {
+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),
@@ -326,11 +327,15 @@ func (t *Trace) GetSpan() trace.Span {
 	return t.span
 }
 
-func (t *Trace) createTraceEvent(name string, eventType int, l7Type int, attrs ...attribute.KeyValue) {
-	t.span.AddEventApm(name, eventType, l7Type, trace.WithAttributes(attrs...))
+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)
 }
@@ -429,7 +434,7 @@ func (t *Trace) SQLTraceQueryEvent(l7Type l7.Protocol, semconvVal attribute.KeyV
 
 	// 拼接 destination 的 IP 和端口为 ip:port 格式
 	destAddr := fmt.Sprintf("%s:%d", destination.IP().String(), destination.Port())
-	
+
 	var attr []attribute.KeyValue
 	attr = append(attr,
 		semconvVal,
@@ -450,7 +455,7 @@ func (t *Trace) SQLTraceQueryEvent(l7Type l7.Protocol, semconvVal attribute.KeyV
 			attribute.String("sql.exception_msg", r.ErrorMsg),
 		)
 	}
-	t.createTraceEvent(l7Type.String(), ebpftracer.EventTypeL7Request.Int(), l7Type.Int(), attr...)
+	t.createTraceEvent(l7Type, ebpftracer.EventTypeL7Request.Int(), attr...)
 }
 
 //func (t *Trace) MysqlTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
@@ -547,31 +552,45 @@ func (t *Trace) SQLTraceQueryEvent(l7Type l7.Protocol, semconvVal attribute.KeyV
 //	)
 //}
 
-func (t *Trace) RedisTraceQueryEvent(cmd, args string, r *l7.RequestData, destination netaddr.IPPort) {
+func (t *Trace) NoSQLTraceQueryEvent(l7Type l7.Protocol, semconvVal attribute.KeyValue, operation, statement string, r *l7.RequestData, destination netaddr.IPPort) {
 	if t == nil {
 		return
 	}
 	t.addEvent()
-	if cmd == "" {
+	if statement == "" {
 		return
 	}
-	statement := cmd
-	if args != "" {
-		statement += " " + args
-	}
+
+	// 拼接 destination 的 IP 和端口为 ip:port 格式
+	destAddr := fmt.Sprintf("%s:%d", destination.IP().String(), destination.Port())
 
 	var attr []attribute.KeyValue
 	attr = append(attr,
-		semconv.DBSystemRedis,
-		semconv.DBOperation(cmd),
+		semconvVal,
 		semconv.DBStatement(statement),
 		semconv.NetPeerName(destination.IP().String()),
 		semconv.NetPeerPort(int(destination.Port())),
 		attribute.String("nosql.src_addr", r.ComponentSAddr.String()),
 		attribute.String("nosql.destination_addr", r.ComponentDAddr.String()),
+		attribute.String("nosql.dbn", destAddr),
 	)
+
+	// 如果 operation 不为空，添加 DBOperation 属性
+	if operation != "" {
+		attr = append(attr, semconv.DBOperation(operation))
+	}
+
 	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
-	t.createTraceEvent(l7.ProtocolRedis.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolRedis.Int(), attr...)
+
+	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) GrpcServerTraceQueryEvent(r *l7.RequestData, appInfo AppInfo) {
@@ -602,7 +621,7 @@ func (t *Trace) GrpcServerTraceQueryEvent(r *l7.RequestData, appInfo AppInfo) {
 	)
 
 	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
-	t.createTraceEvent(l7.ProtocolGrpc.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolGrpc.Int(), attr...)
+	t.createTraceEvent(l7.ProtocolGrpc, ebpftracer.EventTypeL7Request.Int(), attr...)
 }
 
 func (t *Trace) GrpcClientTraceQueryEvent(r *l7.RequestData) {
@@ -669,28 +688,7 @@ func (t *Trace) GrpcClientTraceQueryEvent(r *l7.RequestData) {
 	)
 
 	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
-	t.createTraceEvent(l7.ProtocolGrpc.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolGrpc.Int(), attr...)
-}
-func (t *Trace) MongoTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
-	if t == nil {
-		return
-	}
-	t.addEvent()
-	if query == "" {
-		return
-	}
-
-	var attr []attribute.KeyValue
-	attr = append(attr,
-		semconv.DBSystemMongoDB,
-		semconv.DBStatement(query),
-		semconv.NetPeerName(destination.IP().String()),
-		semconv.NetPeerPort(int(destination.Port())),
-		attribute.String("nosql.src_addr", r.ComponentSAddr.String()),
-		attribute.String("nosql.destination_addr", r.ComponentDAddr.String()),
-	)
-	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
-	t.createTraceEvent(l7.ProtocolMongo.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolMongo.Int(), attr...)
+	t.createTraceEvent(l7.ProtocolGrpc, ebpftracer.EventTypeL7Request.Int(), attr...)
 }
 
 func (t *Trace) DNSTraceQueryEvent(r *l7.RequestData, _type string, fqdn string, ttl uint32, ips []netaddr.IP) {
@@ -711,7 +709,7 @@ func (t *Trace) DNSTraceQueryEvent(r *l7.RequestData, _type string, fqdn string,
 		attribute.String("dns.ips", ip),
 	)
 	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
-	t.createTraceEvent(l7.ProtocolDNS.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolDNS.Int(), attr...)
+	t.createTraceEvent(l7.ProtocolDNS, ebpftracer.EventTypeL7Request.Int(), attr...)
 }
 
 func (t *Trace) HttpTraceRequest(method, path, ip string, port uint16, r *l7.RequestData) {
@@ -766,7 +764,7 @@ func (t *Trace) HttpTraceRequestEvent(method, path, ip string, port uint16, r *l
 		attribute.Bool("http.is_tls", r.IsTls),
 	)
 	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
-	t.createTraceEvent(l7.ProtocolHTTP.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolHTTP.Int(), attr...)
+	t.createTraceEvent(l7.ProtocolHTTP, ebpftracer.EventTypeL7Request.Int(), attr...)
 }
 
 func (t *Trace) FuncTraceQuery(funcname string, duration time.Duration, start uint64, end uint64) {
@@ -815,5 +813,5 @@ func (t *Trace) createTraceSpanNoTime2(name string, duration time.Duration, erro
 	//})
 	var attr []attribute.KeyValue
 	t.appendTimestamp(&attr, start, end, int64(end-start))
-	t.createTraceEvent(name, ebpftracer.EventTypeFunEnt.Int(), 0, attr...)
+	t.createFuncTraceEvent(name, ebpftracer.EventTypeFunEnt.Int(), attr...)
 }