Fixed #TASK_GK-2944 横向串联6.25

Makefile2

@@ -27,4 +27,6 @@ go: go-build run
 
 run:
 	ssh [email protected] 'cd /opt/github/euspace && TRACES_ENDPOINT=http://10.0.12.192:18080/docp/api/v2/data/receive ${FILTER} ./euspace --listen="0.0.0.0:8123"'
+send_run:
+	ssh [email protected] 'cd /opt/github/euspace && SEND=1 TRACES_ENDPOINT=http://10.0.12.192:18080/docp/api/v2/data/receive ${FILTER} ./euspace --listen="0.0.0.0:8123"'
 all: c go run

containers/container_apm.go

@@ -71,20 +71,24 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 			return nil
 		}
 	}
-	//if r.Protocol != l7.ProtocolHTTP {
-	//
-	//}
+	if r.Protocol == l7.ProtocolHTTP {
+		//stats.observe(r.Status.Http(), "", r.Duration)
+		method, path, hostIp, port := l7.ParseHttpHost(r.Payload)
+		//trace.HttpRequest(method, path, r.Status, r.Duration)
+
+		apmTrace, ok := c.getTrace(r.TraceId)
+		if ok {
+			apmTrace.HttpTraceRequest(method, path, hostIp, port, r)
+		}
+		return nil
+	}
 	conn := c.connectionsByPidFd[PidFd{Pid: pid, Fd: fd}]
 	fmt.Println("l7.connectionsByPidFd", conn, pid, fd)
+
 	if conn == nil {
 		return nil
 	}
-
-	fmt.Println("l7.timestamp", timestamp, conn.Timestamp)
-	//if r.Protocol != l7.ProtocolHTTP {
-	//
-	//}
-	if r.Protocol != l7.ProtocolHTTP && timestamp != 0 && conn.Timestamp != timestamp {
+	if timestamp != 0 && conn.Timestamp != timestamp {
 		return nil
 	}
 	stats := c.l7Stats.get(r.Protocol, conn.Dest, conn.ActualDest)
@@ -92,7 +96,7 @@ func (c *Container) onL7RequestApm(pid uint32, fd uint64, timestamp uint64, r *l
 	switch r.Protocol {
 	case l7.ProtocolHTTP:
 		fmt.Println("l7.ProtocolHTTP", r.TraceId)
-		stats.observe(r.Status.Http(), "", r.Duration)
+		//stats.observe(r.Status.Http(), "", r.Duration)
 		method, path, hostIp, port := l7.ParseHttpHost(r.Payload)
 		//trace.HttpRequest(method, path, r.Status, r.Duration)
 

ebpftracer/ebpf/common/common.h

@@ -10,8 +10,8 @@
 #ifdef _DEBUG_MODE
 #define cw_bpf_debug(fmt, ...)                                 \
 ({                                                             \
-    char ____fmt[] = fmt;                                      \
-    bpf_trace_printk(____fmt, sizeof(____fmt), ##__VA_ARGS__); \
+                               \
+  \
 })
 #define debug(fmt, ...)                                        \
 ({                                                             \

ebpftracer/ebpf/ebpf.c

@@ -36,19 +36,6 @@
 })
 
 
-#ifdef _DEBUG_MODE
-#define cw_bpf_debug(fmt, ...)                                 \
-({                                                             \
-    char ____fmt[] = fmt;                                      \
-    bpf_trace_printk(____fmt, sizeof(____fmt), ##__VA_ARGS__); \
-})
-#else
-#define cw_bpf_debug(fmt, ...)                                  \
-({                                                              \
-                                                                \
-})
-#endif
-
 //#define GOROUTINE(x) ((x)->r14)
 
 #include "common/common.h"

ebpftracer/ebpf/include/apm_trace.h

@@ -43,6 +43,10 @@ struct fd_trace_key_t {
 	__u32 fd;
 };
 
+struct thread_trace_key_t {
+	__u64 pid_tgid;
+};
+
 
 //struct apm_span_context {
 //	unsigned char TraceID[APM_TRACE_ID_SIZE];
@@ -71,9 +75,15 @@ struct apm_trace_info_t {
 	//	__u64 thread_trace_id; // 线程追踪ID
 	//	__u64 socket_id; // Records the socket associated when tracing was created (记录创建追踪时关联的socket)
 	__u64 trace_id;
-	struct apm_span_context psc;
+	struct apm_trace_key_t trace_key;
+
+	struct fd_trace_key_t fd_trace_key;
+
+	struct thread_trace_key_t thread_trace_key;
+
+//	struct apm_span_context psc;
 
-	struct apm_span_context sc;
+//	struct apm_span_context sc;
 };
 
 static __always_inline void cw_copy_byte_arrays(unsigned char *src, unsigned char *dst, __u32 size)

ebpftracer/ebpf/l7/apm_trace.c

@@ -30,6 +30,13 @@ struct {
 	__uint(max_entries, 32768);
 } fd_trace_info_heap SEC(".maps");
 
+struct {
+	__uint(type, BPF_MAP_TYPE_LRU_HASH);
+	__uint(key_size, sizeof(struct thread_trace_key_t));
+	__uint(value_size, sizeof(struct apm_trace_info_t));
+	__uint(max_entries, 32768);
+} thread_trace_info_heap SEC(".maps");
+
 static __inline __attribute__((__always_inline__))
 struct apm_trace_key_t get_apm_trace_key(__u64 timeout, bool is_socket_io) {
 	__u64 pid_tgid = bpf_get_current_pid_tgid();
@@ -53,7 +60,7 @@ struct apm_trace_key_t get_apm_trace_key(__u64 timeout, bool is_socket_io) {
 //    debug("apm key.pid:%d",(__u32)pid_tgid);
 //    debug("apm key.goid:%llu",key.goid);
 //    cw_bpf_debug("[Trace ID] thread id:%d",(__u32)pid_tgid);
-	bpf_printk("[Trace ID] GOID:%llu", key.goid);
+//	bpf_printk("[get_apm_trace_key] pid:[%d]|GOID:[%llu]", (__u32) pid_tgid, key.goid);
 	return key;
 }
 
@@ -70,13 +77,31 @@ __u64 get_apm_trace_id(__u32 pid, __u32 tid) {
 }
 
 static __inline __attribute__((__always_inline__))
-struct fd_trace_key_t get_fd_trace_key(__u32 pid, __u32 fd) {
+struct apm_trace_info_t *get_apm_trace_info(__u32 pid, __u32 tid) {
+	struct apm_trace_key_t trace_key = get_apm_trace_key(120 * NS_PER_SEC, true);
+	struct apm_trace_info_t *trace_info = bpf_map_lookup_elem(&trace_info_heap, &trace_key);
+	if (trace_info) {
+		return trace_info;
+	}
+	return 0;
+}
+
+static __inline __attribute__((__always_inline__))
+struct fd_trace_key_t get_fd_trace_key(__u32 pid, __u64 fd) {
 	struct fd_trace_key_t trace_key = {};
 	trace_key.tgid = pid;
 	trace_key.fd = fd;
 	return trace_key;
 }
 
+static __inline __attribute__((__always_inline__))
+struct thread_trace_key_t get_thread_trace_key(__u64 id) {
+	struct thread_trace_key_t trace_key = {};
+	trace_key.pid_tgid = id;
+	return trace_key;
+}
+
+
 static __inline __attribute__((__always_inline__))
 __u64 get_fd_trace_id(__u32 pid, __u32 fd) {
 	struct fd_trace_key_t trace_key = get_fd_trace_key(pid, fd);
@@ -104,7 +129,7 @@ struct apm_span_context *cw_get_parent_tracking_span() {
 	struct apm_trace_key_t trace_key = get_apm_trace_key(120 * NS_PER_SEC, true);
 	struct apm_span_context *apm_sc = {0};
 	struct apm_span_context *span_contexts = bpf_map_lookup_elem(&apm_parent_span_context_map, &trace_key);
-	bpf_printk("-------");
+//	bpf_printk("-------");
 
 	if (span_contexts) {
 		/*for (int i = 0; i < APM_TYPE_FROM_SIZE; i++) {
@@ -133,7 +158,7 @@ struct apm_span_context *cw_get_parent_tracking_span() {
 		}*/
 		apm_sc = span_contexts;
 	}
-	bpf_printk("-------end");
+//	bpf_printk("-------end");
 
 	return apm_sc;
 }
@@ -153,9 +178,14 @@ __u64 clear_current_span_context() {
 }
 
 static __inline __attribute__((__always_inline__))
-__u64 clear_trace(__u32 pid, __u32 tid, __u32 fd) {
+__u64 cw_clear_trace(__u32 tgid, __u32 pid, __u32 fd) {
 	struct apm_trace_key_t trace_key = get_apm_trace_key(120 * NS_PER_SEC, true);
-	struct fd_trace_key_t fd_trace_key = get_fd_trace_key(pid, fd);
+	struct fd_trace_key_t fd_trace_key = get_fd_trace_key(tgid, fd);
+	// 清除trace信息
+	struct apm_trace_info_t *trace_info = bpf_map_lookup_elem(&trace_info_heap, &trace_key);
+	if (trace_info) {
+		bpf_map_delete_elem(&thread_trace_info_heap, &trace_info->thread_trace_key);
+	}
 	// 清除trace信息
 	bpf_map_delete_elem(&trace_info_heap, &trace_key);
 	bpf_map_delete_elem(&fd_trace_info_heap, &fd_trace_key);
@@ -180,7 +210,7 @@ struct apm_span_context *cw_get_current_tracking_span() {
 	struct apm_trace_key_t trace_key = get_apm_trace_key(120 * NS_PER_SEC, true);
 	struct apm_span_context *apm_sc = {0};
 	struct apm_span_context *span_contexts = bpf_map_lookup_elem(&apm_current_span_context_map, &trace_key);
-	bpf_printk("-------");
+//	bpf_printk("-------");
 
 	if (span_contexts) {
 		/*for (int i = 0; i < APM_TYPE_FROM_SIZE; i++) {
@@ -209,7 +239,36 @@ struct apm_span_context *cw_get_current_tracking_span() {
 		}*/
 		apm_sc = span_contexts;
 	}
-	bpf_printk("-------end");
+//	bpf_printk("-------end");
 
 	return apm_sc;
-}
+}
+
+static __inline __attribute__((__always_inline__))
+struct apm_trace_info_t cw_save_trace_info(__u64 id, __u32 pid, __u64 fd) {
+	// apm trace
+//	struct apm_trace_key_t trace_key = {0};
+//	trace_key = get_apm_trace_key(120 * NS_PER_SEC, true);
+
+	// fd trace
+//	struct fd_trace_key_t fd_trace_key = {0};
+//	fd_trace_key = get_fd_trace_key(pid, fd);
+
+	// trace info
+	struct apm_trace_info_t trace_info = {0};
+	__u64 uid_base = bpf_ktime_get_ns();
+	trace_info.trace_id = bpf_get_current_pid_tgid() + uid_base;
+	// apm trace
+	trace_info.trace_key = get_apm_trace_key(120 * NS_PER_SEC, true);
+	// fd trace
+	trace_info.fd_trace_key = get_fd_trace_key(pid, fd);
+	// thread trace
+	trace_info.thread_trace_key = get_thread_trace_key(id);
+
+	//	入口方法缓存
+	bpf_map_update_elem(&trace_info_heap, &trace_info.trace_key, &trace_info, BPF_NOEXIST);
+	bpf_map_update_elem(&fd_trace_info_heap, &trace_info.fd_trace_key, &trace_info, BPF_NOEXIST);
+	bpf_map_update_elem(&thread_trace_info_heap, &trace_info.thread_trace_key, &trace_info, BPF_NOEXIST);
+	return trace_info;
+}
+

ebpftracer/ebpf/l7/l7.c

@@ -286,9 +286,9 @@ int trace_enter_write(void *ctx, __u64 fd, __u16 is_tls, char *buf, __u64 size,
     {
 //        __u64 trace_id = get_apm_trace_id(pid, fd);
         __u64 trace_id = get_fd_trace_id(pid, fd);
-        cw_bpf_debug("[trace end][Response][HTTP] pid:%d,fd:%d,trace_id:%llu", pid, fd, trace_id);
+        cw_bpf_debug("[Trace End in l7][Response][HTTP] pid:%d,fd:%d,trace_id:%llu", tid, fd, trace_id);
         // 清除trace信息
-        clear_trace(pid, tid, fd);
+        cw_clear_trace(pid, tid, fd);
         // 发送事件到用户空间 start
         struct l7_event *e = bpf_map_lookup_elem(&l7_event_heap, &zero);
         if (!e) {
@@ -350,8 +350,10 @@ int trace_enter_write(void *ctx, __u64 fd, __u16 is_tls, char *buf, __u64 size,
     }
 
     if (is_http_request(payload)) {
-        cw_bpf_debug("[Kernel Enter] [HTTP]:TGID:%d|FD:%d",k.pid,k.fd);
-        req->protocol = PROTOCOL_HTTP;
+	    cw_bpf_debug("");
+	    cw_bpf_debug("-----[Kernel Enter][HTTP]:pid:[%d]|CURRENT-GOID:[%llu]|FD:[%d]", tid, get_current_goroutine(), k.fd);
+
+	    req->protocol = PROTOCOL_HTTP;
 //        struct apm_trace_key_t trace_key = {0};
 //        trace_key = get_apm_trace_key(120 * NS_PER_SEC, true);
 //        cw_bpf_debug("[Enter] [HTTP]:goid:%llu",trace_key.goid);
@@ -547,7 +549,6 @@ int trace_exit_read(void *ctx, __u64 id, __u32 pid, __u16 is_tls, long int ret)
             bpf_map_update_elem(&active_l7_requests, &k, req, BPF_NOEXIST);
         }
 
-        cw_bpf_debug("[trace start][Receive][HTTP]:thread_id:%d\n",tid);
 //        cw_bpf_debug("[Receive][HTTP] payload1:%s|type:%s\n",payload,"type");
 //        __u64 goid = 0;
 //        goid = get_rw_goid(120 * NS_PER_SEC, 1);
@@ -556,18 +557,20 @@ int trace_exit_read(void *ctx, __u64 id, __u32 pid, __u16 is_tls, long int ret)
 //        cw_bpf_debug("[Receive][HTTP]:thread_id:%d|goid:%d|FD:%d\n", tid, goid, k.fd);
 
 
-		// apm trace
-	    struct apm_trace_key_t trace_key = {0};
-	    trace_key = get_apm_trace_key(120 * NS_PER_SEC, true);
-
-	    // fd trace
-        struct fd_trace_key_t fd_trace_key = {0};
-	    fd_trace_key = get_fd_trace_key(pid, k.fd);
-
-		// trace info
-	    struct apm_trace_info_t trace_info = {0};
-	    __u64 uid_base = bpf_ktime_get_ns();
-	    trace_info.trace_id = bpf_get_current_pid_tgid() + uid_base;
+//		// apm trace
+//	    struct apm_trace_key_t trace_key = {0};
+//	    trace_key = get_apm_trace_key(120 * NS_PER_SEC, true);
+//
+//	    // fd trace
+//        struct fd_trace_key_t fd_trace_key = {0};
+//	    fd_trace_key = get_fd_trace_key(pid, k.fd);
+//
+//		// trace info
+	    struct apm_trace_info_t trace_info = cw_save_trace_info(id,pid, k.fd);
+//	    __u64 uid_base = bpf_ktime_get_ns();
+//	    trace_info.trace_id = bpf_get_current_pid_tgid() + uid_base;
+	    cw_bpf_debug("\n");
+	    cw_bpf_debug("[Trace Start in l7][Receive][HTTP]pid:[%d]|GOID:[%d]|FD:%d", tid, trace_info.trace_key.goid,k.fd);
 
         e->trace_start = 1;
         e->trace_end = 0;
@@ -576,9 +579,11 @@ int trace_exit_read(void *ctx, __u64 id, __u32 pid, __u16 is_tls, long int ret)
         e->payload_size = ret;
         COPY_PAYLOAD(e->payload, ret, payload);
 
+		// pid_tgid:trace_id
+//		thread_trace_key =
         // 入口方法缓存
-        bpf_map_update_elem(&trace_info_heap, &trace_key, &trace_info, BPF_NOEXIST);
-	    bpf_map_update_elem(&fd_trace_info_heap, &fd_trace_key, &trace_info, BPF_NOEXIST);
+//        bpf_map_update_elem(&trace_info_heap, &trace_key, &trace_info, BPF_NOEXIST);
+//	    bpf_map_update_elem(&fd_trace_info_heap, &fd_trace_key, &trace_info, BPF_NOEXIST);
         bpf_perf_event_output(ctx, &l7_events, BPF_F_CURRENT_CPU, e, sizeof(*e));
         cw_bpf_debug("[Receive][HTTP] to user space");
         return 0;
@@ -638,8 +643,6 @@ int trace_exit_read(void *ctx, __u64 id, __u32 pid, __u16 is_tls, long int ret)
     if (e->protocol == PROTOCOL_HTTP) {
         __u64 trace_id = get_apm_trace_id(pid, tid);
         e->trace_id = trace_id;
-        cw_bpf_debug("[Kernel Response][HTTP] [Trace ID] trace_id:%llu", e->trace_id);
-        cw_bpf_debug("[Kernel Response][HTTP]:TGID:%d|type:%s|FD:%d\n",k.pid,"",k.fd);
 	    struct  apm_span_context * sc = cw_get_current_tracking_span();
 	    if (sc) {
 		    cw_copy_byte_arrays(sc->assumed_app_id, e->assumed_app_id, APM_ASSUMED_APP_ID_SIZE);
@@ -656,6 +659,25 @@ int trace_exit_read(void *ctx, __u64 id, __u32 pid, __u16 is_tls, long int ret)
 //        cw_bpf_debug("[Response][HTTP222] resp-payload:%s",payload);
 
         response = is_http_response(payload, &e->status);
+//	    if (trace_id == 0) {
+//		    __u64 goid = get_current_goroutine();
+//		    cw_bpf_debug("[Kernel End] go_id:%llu",goid);
+//
+//		}
+	    if (e->trace_id == 0){
+		    struct thread_trace_key_t t_key = get_thread_trace_key(id);
+		    struct apm_trace_info_t *trace_info = bpf_map_lookup_elem(&thread_trace_info_heap, &t_key);
+		    if (trace_info) {
+			    if (trace_info->trace_id == 0){
+				    cw_bpf_debug("get trace_id from enter pid:[%d]|thread_trace_key_t:[%llu]", tid,trace_info->trace_id);
+				    e->trace_id=trace_info->trace_id;
+				}
+		    } else {
+				// 从子事件的头获取线程id
+			}
+	    }
+	    cw_bpf_debug("[Kernel End][HTTP]:pid:[%d]|CURRENT-GOID:[%llu]|trace_id:[%llu]---------\n", tid, get_current_goroutine(),e->trace_id);
+
     } else if (e->protocol == PROTOCOL_POSTGRES) {
         response = is_postgres_response(payload, ret, &e->status);
         if (req->request_type == POSTGRES_FRAME_PARSE) {

ebpftracer/ebpf/socket_trace.c

@@ -332,8 +332,8 @@ static __inline struct trace_key_t get_trace_key(__u64 timeout, bool is_socket_i
 	}
 
 //    debug("key.tgid:%llu|%llu",key.tgid,key.goid);
-    debug("key.pid:%d",(__u32)pid_tgid);
-    debug("key.goid:%llu",key.goid);
+//    debug("key.pid:%d",(__u32)pid_tgid);
+//    debug("key.goid:%llu",key.goid);
 
 	return key;
 }
@@ -1948,32 +1948,33 @@ static __inline int output_data_common(void *ctx) {
 	}
 
 	v->data_len = len;
-//    debug("start=======================");
-//    debug("Pid: %u", v->pid);
-//    debug("Tgid: %u", v->tgid);
+
+    debug("start=======================");
+    debug("Pid: %u", v->pid);
+    debug("Tgid: %u", v->tgid);
     debug("CoroutineID: %llu", v->coroutine_id);
-//    debug("Source: %u", v->source);
-//    debug("Comm: %s", v->comm);
-//    debug("SocketID: %llu", v->socket_id);
-//    debug("ExtraData: %u", v->extra_data);
-//    debug("ExtraDataCount: %u", v->extra_data_count);
-//    debug("TCPSeq: %u", v->tcp_seq);
-//    debug("ThreadTraceID: %llu", v->thread_trace_id);
-//    debug("Timestamp: %llu", v->timestamp);
-//    debug("Direction: %u", v->direction);
-//    debug("MsgType: %u", v->msg_type);
-//    debug("SyscallLen: %llu", v->syscall_len);
-//    debug("DataSeq: %llu", v->data_seq);
-//    debug("DataType: %u", v->data_type);
-//    debug("DataLen: %u", v->data_len);
-//    debug("data: %s", v->data);
-////    for (size_t i = 0; i < v->data_len; ++i) {
-////        debug("%02x ", (unsigned char)v->data[i]);
-////        if ((i + 1) % 16 == 0) {
-////            debug("\n");
-////        }
-////    }
-//    debug("=======================end");
+    debug("Source: %u", v->source);
+    debug("Comm: %s", v->comm);
+    debug("SocketID: %llu", v->socket_id);
+    debug("ExtraData: %u", v->extra_data);
+    debug("ExtraDataCount: %u", v->extra_data_count);
+    debug("TCPSeq: %u", v->tcp_seq);
+    debug("ThreadTraceID: %llu", v->thread_trace_id);
+    debug("Timestamp: %llu", v->timestamp);
+    debug("Direction: %u", v->direction);
+    debug("MsgType: %u", v->msg_type);
+    debug("SyscallLen: %llu", v->syscall_len);
+    debug("DataSeq: %llu", v->data_seq);
+    debug("DataType: %u", v->data_type);
+    debug("DataLen: %u", v->data_len);
+    debug("data: %s", v->data);
+//    for (size_t i = 0; i < v->data_len; ++i) {
+//        debug("%02x ", (unsigned char)v->data[i]);
+//        if ((i + 1) % 16 == 0) {
+//            debug("\n");
+//        }
+//    }
+    debug("=======================end");
 
 	v_buff->len += offsetof(typeof(struct __socket_data), data) + v->data_len;
 	v_buff->events_num++;
@@ -2203,23 +2204,23 @@ static __inline void trace_io_event_common(void *ctx,
 
 PROGTP(io_event)(void *ctx)
 {
-//	__u64 id = bpf_get_current_pid_tgid();
-//
-//	struct data_args_t *data_args = NULL;
-//
-//	data_args = active_read_args_map__lookup(&id);
-//	if (data_args) {
-//		trace_io_event_common(ctx, data_args, T_INGRESS, id);
-//		active_read_args_map__delete(&id);
-//		return 0;
-//	}
-//
-//	data_args = active_write_args_map__lookup(&id);
-//	if (data_args) {
-//		trace_io_event_common(ctx, data_args, T_EGRESS, id);
-//		active_write_args_map__delete(&id);
-//		return 0;
-//	}
+	__u64 id = bpf_get_current_pid_tgid();
+
+	struct data_args_t *data_args = NULL;
+
+	data_args = active_read_args_map__lookup(&id);
+	if (data_args) {
+		trace_io_event_common(ctx, data_args, T_INGRESS, id);
+		active_read_args_map__delete(&id);
+		return 0;
+	}
+
+	data_args = active_write_args_map__lookup(&id);
+	if (data_args) {
+		trace_io_event_common(ctx, data_args, T_EGRESS, id);
+		active_write_args_map__delete(&id);
+		return 0;
+	}
 
 	return 0;
 }

ebpftracer/ebpf/utrace/go/net/client.probe.bpf.c

@@ -369,7 +369,7 @@ static __always_inline long cw_inject_header2(void* headers_ptr, struct apm_span
 // func net/http/transport.roundTrip(req *Request) (*Response, error)
 SEC("uprobe/Transport_roundTrip")
 int uprobe_Transport_roundTrip(struct pt_regs *ctx) {
-	bpf_printk("[Uprobe HTTP Client] start");
+	bpf_printk("--------[Uprobe HTTP Client] start");
 	__u64 goroutine_id = GOROUTINE(ctx);
 	bpf_printk("[Uprobe HTTP Client] goroutine_id:%llu",goroutine_id);
 
@@ -516,7 +516,7 @@ int uprobe_Transport_roundTrip(struct pt_regs *ctx) {
 // func net/http/transport.roundTrip(req *Request) (*Response, error)
 SEC("uprobe/Transport_roundTrip")
 int uprobe_Transport_roundTrip_Returns(struct pt_regs *ctx) {
-    bpf_printk("[Uprobe HTTP Client] Return start");
+	bpf_printk("[Uprobe HTTP Client] start Return----------");
 	clear_current_span_context();
     u64 end_time = bpf_ktime_get_ns();
     void *req_ctx_ptr = get_Go_context(ctx, 2, ctx_ptr_pos, false);

ebpftracer/tracer.go

@@ -23,6 +23,7 @@ import (
 	"strconv"
 	"strings"
 	"time"
+	"unsafe"
 )
 
 /*
@@ -30,6 +31,7 @@ import (
 #define BURST_DATA_BUF_SIZE 8192	// For brust send buffer
 
 #include <linux/types.h>
+
 struct __tuple_t {
 	__u8 daddr[16];
 	__u8 rcv_saddr[16];
@@ -40,29 +42,29 @@ struct __tuple_t {
 };
 
 struct __socket_data {
-__u32 pid;
-__u32 tgid;
-__u64 coroutine_id;
-__u8 source;
-__u8 comm[TASK_COMM_LEN];
-
-__u64 socket_id;
-struct __tuple_t tuple;
-__u32 extra_data;
-__u32 extra_data_count;
-
-__u32 tcp_seq;
-__u64 thread_trace_id;
-
-__u64 timestamp;
-__u8  direction: 1;
-__u8  msg_type:  7;
-
-__u64 syscall_len;
-__u64 data_seq;
-__u16 data_type;
-__u16 data_len;
-char data[BURST_DATA_BUF_SIZE];
+	__u32 pid;
+	__u32 tgid;
+	__u64 coroutine_id;
+	__u8 source;
+	__u8 comm[TASK_COMM_LEN];
+
+	__u64 socket_id;
+	struct __tuple_t tuple;
+	__u32 extra_data;
+	__u32 extra_data_count;
+
+	__u32 tcp_seq;
+	__u64 thread_trace_id;
+
+	__u64 timestamp;
+	__u8  direction: 1;
+	__u8  msg_type:  7;
+
+	__u64 syscall_len;
+	__u64 data_seq;
+	__u16 data_type;
+	__u16 data_len;
+	char data[BURST_DATA_BUF_SIZE];
 } __attribute__((packed));
 
 struct __socket_data_buffer {
@@ -473,7 +475,7 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy
 		switch typ {
 		case perfMapTypeSocketEvents:
 			//fmt.Println("perfMapTypeSocketEvents")
-			//// 假设 rec.RawSample 包含数据，类型为 []byte
+			// 假设 rec.RawSample 包含数据，类型为 []byte
 			//rawData := rec.RawSample
 			//fmt.Println("perfMapTypeSocketEvents2")
 			//
@@ -500,9 +502,13 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy
 			//fmt.Printf("Data: %s\n", string(buffer.Data[:buffer.Len])) // 仅打印实际长度的数据
 
 			//socketDataBuffer := rec.RawSample
-			/*todo */
-			//buf := (*SocketDataBuffer)(unsafe.Pointer(&rec.RawSample[0])) //nolint:gosec
-			//socketData := (*SocketData)(unsafe.Pointer(&buf.data[0]))     //nolint:gosec
+
+			// 解析 __socket_data_buffer
+			buf := (*SocketDataBuffer)(unsafe.Pointer(&rec.RawSample[0])) //nolint:gosec
+
+			// 获取 char data[32760];
+			socketData := (*SocketData)(unsafe.Pointer(&buf.data[0])) //nolint:gosec
+
 			/*todo */
 			//socketData := (*(*[128]byte)(unsafe.Pointer(&eventC.line)))
 			//dataPtr := unsafe.Pointer(&buf.data[0])
@@ -530,9 +536,13 @@ func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapTy
 			//fmt.Println("socketData.SyscallLen:", socketData.SyscallLen)
 			//fmt.Println("socketData.DataSeq:", socketData.DataSeq)
 			// todo
-			//			fmt.Printf("socketData.DataType:%d \n", (socketData.data_type))
-			//			fmt.Printf("socketData.DataLen:%d \n", (socketData.data_len))
-			//fmt.Println("socketData.Data:", len(socketData.Data))
+			fmt.Printf("socketData.DataType:%d \n", (socketData.data_type))
+			fmt.Printf("socketData.DataLen:%d \n", (socketData.data_len))
+
+			// 解析C结构体中的data字段
+			dataSlice := C.GoBytes(unsafe.Pointer(&socketData.data[0]), C.int(socketData.data_len))
+			// 打印或处理包含的数据
+			fmt.Printf("socketData.Payload:%v \n", string(dataSlice))
 
 			//socketData := &SocketData{}
 			//reader := bytes.NewBuffer(rec.RawSample)

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

@@ -240,7 +240,7 @@ func buildAndAssemblyMap(sd apmTraceSpan, traceRoot *TraceMapT) MapInfoT {
 func buildAppMap(mNode *MapInfoT, traceRoot *TraceMapT, sd apmTraceSpan) {
 	mNode.ServiceName = GO_SERVICE_NAME
 	mNode.ServiceType = APP_SERVICE_TYPE
-	mNode.MethodName = "Endpoint"
+	mNode.MethodName = "net/http.(*Transport).roundTrip()"
 	mNode.Level = 1
 	mNode.Pid = 0
 	// 构建root节点
@@ -282,7 +282,7 @@ func buildHttpMap(mNode *MapInfoT, sd apmTraceSpan) {
 	mNode.ServiceName = HTTP_SERVICE_NAME
 	mNode.ServiceType = HTTP_SERVICE_TYPE
 	mNode.Schema = "http"
-	mNode.MethodName = "(External Services)"
+	mNode.MethodName = "net/http.serverHandler.ServeHTTP()"
 	var descAddr string
 	for _, attr := range sd.Attributes() {
 		fmt.Println(attr.Key, ":", attr.Value.AsInterface())
@@ -308,7 +308,7 @@ func buildMysqlMap(mNode *MapInfoT, sd apmTraceSpan) {
 	mNode.Dbn = "unknown"
 	mNode.ServiceName = MYSQL_SERVICE_NAME
 	mNode.ServiceType = SQL_SERVICE_TYPE
-	mNode.MethodName = "Mysql query"
+	mNode.MethodName = "database/sql.Query()"
 	for _, attr := range sd.Attributes() {
 		//fmt.Println(attr.Key, ":", attr.Value.AsInterface())
 		switch attr.Key {
@@ -330,7 +330,8 @@ func buildMysqlMap(mNode *MapInfoT, sd apmTraceSpan) {
 func buildRedisMap(mNode *MapInfoT, sd apmTraceSpan) {
 	mNode.ServiceName = REDIS_SERVICE_NAME
 	mNode.ServiceType = NOSQL_SERVICE_TYPE
-	mNode.MethodName = span(sd).Name + " query"
+	//mNode.MethodName = span(sd).Name + " query"
+	mNode.MethodName = "redis.Do()"
 	for _, attr := range sd.Attributes() {
 		//fmt.Println(attr.Key, ":", attr.Value.AsInterface())
 		switch attr.Key {

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

@@ -17,6 +17,8 @@ package otlptrace // import "go.opentelemetry.io/otel/exporters/otlp/otlptrace"
 import (
 	"context"
 	"errors"
+	"fmt"
+	"os"
 	"sync"
 
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/internal/tracetransform"
@@ -44,12 +46,16 @@ func (e *Exporter) ExportSpans(ctx context.Context, ss []tracesdk.ReadOnlySpan)
 	if len(protoSpans) == 0 {
 		return nil
 	}
-	//sendData := tracetransformData(ss)
-	tracetransformData(ss)
-	//err := e.client.UploadApmTraces(ctx, sendData)
-	//if err != nil {
-	//	return fmt.Errorf("traces export: %w", err)
-	//}
+	if os.Getenv("SEND") == "1" {
+		sendData := tracetransformData(ss)
+		err := e.client.UploadApmTraces(ctx, sendData)
+		if err != nil {
+			return fmt.Errorf("traces export: %w", err)
+		}
+	} else {
+		tracetransformData(ss)
+	}
+
 	return nil
 }