ebpf/ebpftracer/tracer.go

package ebpftracer

import (
	"bytes"
	debugelf "debug/elf"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/coroot/coroot-node-agent/utils"
	"github.com/coroot/coroot-node-agent/utils/try"

	"github.com/cilium/ebpf"
	"github.com/cilium/ebpf/link"
	"github.com/cilium/ebpf/perf"
	"github.com/coroot/coroot-node-agent/common"
	"github.com/coroot/coroot-node-agent/ebpftracer/l7"
	"github.com/coroot/coroot-node-agent/ebpftracer/tracer"
	"github.com/coroot/coroot-node-agent/proc"
	. "github.com/coroot/coroot-node-agent/utils/modelse"
	klog "github.com/sirupsen/logrus"
	"golang.org/x/sys/unix"
	"inet.af/netaddr"
)

/*
#define TASK_COMM_LEN	16
#define BURST_DATA_BUF_SIZE 8192	// For brust send buffer

#include <linux/types.h>

struct __tuple_t {
	__u8 daddr[16];
	__u8 rcv_saddr[16];
	__u8 addr_len;
	__u8 l4_protocol;
	__u16 dport;
	__u16 num;
};

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];
} __attribute__((packed));

struct __socket_data_buffer {
	__u32 events_num;
	__u32 len;
	char data[32760];
};
*/
import "C"

type SocketData C.struct___socket_data
type SocketDataBuffer C.struct___socket_data_buffer

const MaxPayloadSize = 1024

type EventType uint32
type EventReason uint32

const (
	EventTypeProcessStart    EventType = 1
	EventTypeProcessExit     EventType = 2
	EventTypeConnectionOpen  EventType = 3
	EventTypeConnectionClose EventType = 4
	EventTypeConnectionError EventType = 5
	EventTypeListenOpen      EventType = 6
	EventTypeListenClose     EventType = 7
	EventTypeFileOpen        EventType = 8
	EventTypeTCPRetransmit   EventType = 9
	EventTypeL7Request       EventType = 10
	EventTypeFunEnt          EventType = 11
	EventTypeFunRet          EventType = 12
	EventTypeAcceptOpen      EventType = 13
	EventTypeAcceptClose     EventType = 14

	EventReasonNone    EventReason = 0
	EventReasonOOMKill EventReason = 1
)

type TrafficStats struct {
	BytesSent     uint64
	BytesReceived uint64
}

type Event struct {
	StackEvent     *StackEvent
	Type           EventType
	Reason         EventReason
	Pid            uint32
	SrcAddr        netaddr.IPPort
	DstAddr        netaddr.IPPort
	Fd             uint64
	Timestamp      uint64
	Duration       time.Duration
	L7Request      *l7.RequestData
	TrafficStats   *TrafficStats
	FirstReadTime  uint64
	FirstWriteTime uint64
	NewReadTime    uint64
}

type perfMapType uint8

const (
	perfMapTypeProcEvents         perfMapType = 1
	perfMapTypeTCPEvents          perfMapType = 2
	perfMapTypeFileEvents         perfMapType = 3
	perfMapTypeL7Events           perfMapType = 4
	perfMapTypeSocketEvents       perfMapType = 5
	perfMapTypeEventQueue         perfMapType = 6
	perfMapTypePythonThreadEvents perfMapType = 7
)

type Tracer struct {
	kernelVersion       string
	disableL7Tracing    bool
	disableE2ETracing   bool
	disableStackTracing bool

	collection     *ebpf.Collection
	collectionSpec *ebpf.CollectionSpec
	readers        map[string]*perf.Reader
	links          []link.Link
	uprobes        map[string]*ebpf.Program
	Symbols        []debugelf.Symbol
	Uprobes        []tracer.Uprobe
	UprobesMap     map[string]tracer.Uprobe
}

func NewTracer(kernelVersion string, disableL7Tracing, disableE2ETracing, disableStackTracing bool) *Tracer {
	klog.Infof(
		"Tracing config | L7=%t, E2E=%t, L7Stack=%t",
		!disableL7Tracing,
		!disableE2ETracing,
		!disableStackTracing,
	)
	return &Tracer{
		kernelVersion:       kernelVersion,
		disableL7Tracing:    disableL7Tracing,
		disableE2ETracing:   disableE2ETracing,
		disableStackTracing: disableStackTracing,

		readers: map[string]*perf.Reader{},
		uprobes: map[string]*ebpf.Program{},
		links:   []link.Link{},
	}
}

func (t *Tracer) Run(events chan<- Event) error {
	if err := t.ebpf(events); err != nil {
		return err
	}
	if err := t.init(events); err != nil {
		return err
	}
	return nil
}

func (t *Tracer) Close() {
	for k, p := range t.uprobes {
		if p != nil {
			err := p.Close()
			klog.WithError(err).Infof("[close] uprobes %s", k)
		}
	}
	for _, l := range t.links {
		if l != nil {
			err := l.Close()
			klog.WithError(err).Infof("[close] links")
		}
	}
	for k, r := range t.readers {
		if r != nil {
			err := r.Close()
			klog.WithError(err).Infof("[close] readers %s", k)
		}
	}
	if t.collection != nil {
		t.collection.Close()
	}
}

func (t *Tracer) init(ch chan<- Event) error {
	pids, err := proc.ListPids()
	if err != nil {
		return fmt.Errorf("failed to list pids: %w", err)
	}
	for _, pid := range pids {
		ch <- Event{Type: EventTypeProcessStart, Pid: pid}
	}

	fds, sockets := readFds(pids)
	for _, fd := range fds {
		ch <- Event{Type: EventTypeFileOpen, Pid: fd.pid, Fd: fd.fd}
	}

	listens := map[uint64]bool{}
	for _, s := range sockets {
		if s.Listen {
			listens[uint64(s.pid)<<32|uint64(s.SAddr.Port())] = true
		}
	}

	ebpfConnectionsMap := t.collection.Maps["active_connections"]
	timestamp := uint64(time.Now().UnixNano())
	for _, s := range sockets {
		typ := EventTypeConnectionOpen
		if s.Listen {
			typ = EventTypeListenOpen
			//} else if listens[uint64(s.pid)<<32|uint64(s.SAddr.Port())] || s.DAddr.Port() > s.SAddr.Port() { // inbound
		} else if listens[uint64(s.pid)<<32|uint64(s.SAddr.Port())] { // 存在误判
			continue
		}
		ch <- Event{
			Type:      typ,
			Pid:       s.pid,
			Timestamp: timestamp,
			Fd:        s.fd,
			SrcAddr:   s.SAddr,
			DstAddr:   s.DAddr,
		}
		if typ == EventTypeConnectionOpen {
			id := ConnectionId{FD: s.fd, PID: s.pid}
			sip := s.SAddr.IP()
			sipbytes := sip.As16()
			dip := s.DAddr.IP()
			dipbytes := dip.As16()
			conn := Connection{Timestamp: timestamp, Saddr: sipbytes, Sport: s.SAddr.Port(), Daddr: dipbytes, Dport: s.DAddr.Port()}
			if err := ebpfConnectionsMap.Update(id, conn, ebpf.UpdateNoExist); err != nil {
				klog.Warningln(err)
			}
		}
	}
	return nil
}

func (t *Tracer) ActiveConnectionsIterator() *ebpf.MapIterator {
	return t.collection.Maps["active_connections"].Iterate()
}

func (t *Tracer) ActiveAcceptsIterator() *ebpf.MapIterator {
	return t.collection.Maps["active_accepts"].Iterate()
}

type perfMap struct {
	name                  string
	perCPUBufferSizePages int
	typ                   perfMapType
}

func (t *Tracer) ebpf(ch chan<- Event) error {
	kv := "v" + common.KernelMajorMinor(t.kernelVersion)
	path, prg, err := EbpfCode(kv)
	klog.Infof("kv is [%s], kernel version: [%s] path: [%s]", kv, t.kernelVersion, path)
	if len(prg) == 0 || err != nil {
		return fmt.Errorf("kv is %s, unsupported kernel version: [%s] path: [%s] err:<%v>", kv, t.kernelVersion, path, err)
	}
	_, debugFsErr := os.Stat("/sys/kernel/debug/tracing")
	_, traceFsErr := os.Stat("/sys/kernel/tracing")

	if debugFsErr != nil && traceFsErr != nil {
		return fmt.Errorf("kernel tracing is not available: debugfs or tracefs must be mounted")
	}

	collectionSpec, err := ebpf.LoadCollectionSpecFromReader(bytes.NewReader(prg))
	if err != nil {
		return fmt.Errorf("failed to load collection spec: %w", err)
	}
	_ = unix.Setrlimit(unix.RLIMIT_MEMLOCK, &unix.Rlimit{Cur: unix.RLIM_INFINITY, Max: unix.RLIM_INFINITY})
	tracer.PidFilter(collectionSpec)
	opts := &ebpf.CollectionOptions{MapReplacements: make(map[string]*ebpf.Map)}
	klog.Infof("[start] Look eBPF .maps")
	for _, spec := range collectionSpec.Maps {
		klog.Infoln(spec.Name)
	}
	klog.Infof("[end] Look eBPF .maps")
	tracer.MapInit(collectionSpec, opts)
	// TODO 多进程
	// tracer.SetConstants(collectionSpec)
	c, err := ebpf.NewCollectionWithOptions(collectionSpec, *opts)
	if err != nil {
		var verr *ebpf.VerifierError
		if errors.As(err, &verr) {
			klog.Errorf("----%+v", verr)
		}
		return fmt.Errorf("failed to load collection: %w", err)
	}
	tracer.Offset()
	t.collectionSpec = collectionSpec
	t.collection = c

	perfMaps := []perfMap{
		{name: "proc_events", typ: perfMapTypeProcEvents, perCPUBufferSizePages: 4},
		{name: "tcp_listen_events", typ: perfMapTypeTCPEvents, perCPUBufferSizePages: 4},
		{name: "tcp_connect_events", typ: perfMapTypeTCPEvents, perCPUBufferSizePages: 8},
		{name: "tcp_accept_events", typ: perfMapTypeTCPEvents, perCPUBufferSizePages: 8},
		{name: "tcp_retransmit_events", typ: perfMapTypeTCPEvents, perCPUBufferSizePages: 4},
		{name: "file_events", typ: perfMapTypeFileEvents, perCPUBufferSizePages: 4},
		{name: "event_queue", typ: perfMapTypeEventQueue, perCPUBufferSizePages: 32},
		{name: "python_thread_events", typ: perfMapTypePythonThreadEvents, perCPUBufferSizePages: 4},
	}
	tracer.MapInsert(c)
	if !t.DisableL7Tracing() {
		perfMaps = append(perfMaps, perfMap{name: "l7_events", typ: perfMapTypeL7Events, perCPUBufferSizePages: 32})
	}
	perfMaps = append(perfMaps, perfMap{name: MAP_PERF_SOCKET_DATA_NAME, typ: perfMapTypeSocketEvents, perCPUBufferSizePages: 64})
	klog.Infof("[start] Look eBPF perf_maps")
	for _, pm := range perfMaps {
		klog.Infoln(pm.name)
		m, ok := t.collection.Maps[pm.name]
		if ok {
			r, err := perf.NewReader(m, pm.perCPUBufferSizePages*os.Getpagesize())
			if err != nil {
				t.Close()
				return fmt.Errorf("failed to create ebpf reader: %w", err)
			}
			t.readers[pm.name] = r
			// event监听
			//go runEventsReader(pm.name, r, ch, pm.typ)
			try.GoParams(runEventsReader, utils.CatchFn, pm.name, r, ch, pm.typ)
		}
	}
	klog.Infof("[end] Look eBPF perf_maps")

	klog.Infof("[start] Look eBPF specPrograms")
	if err = t.LinkEbpfProg(); err != nil {
		return err
	}
	klog.Infof("[end] Look eBPF specPrograms")

	return nil
}

func (t *Tracer) LinkEbpfProg() error {
	klog.Infof("[start] Look eBPF specPrograms")
	var (
		l       link.Link
		err     error
		lastErr error
	)
	for _, programSpec := range t.collectionSpec.Programs {
		program := t.collection.Programs[programSpec.Name]
		klog.Infof("%s:[%s]", programSpec.SectionName, programSpec.Name)
		if t.DisableL7Tracing() {
			switch programSpec.Name {
			case "sys_enter_writev", "sys_enter_write", "sys_enter_sendto", "sys_enter_sendmsg", "sys_enter_sendmmsg":
				continue
			case "sys_enter_read", "sys_enter_readv", "sys_enter_recvfrom", "sys_enter_recvmsg":
				continue
			case "sys_exit_read", "sys_exit_readv", "sys_exit_recvfrom", "sys_exit_recvmsg":
				continue
			}
		}
		switch programSpec.Type {
		case ebpf.TracePoint:
			if strings.Contains(programSpec.SectionName, "prog") {
				continue
			}
			parts := strings.SplitN(programSpec.AttachTo, "/", 2)
			l, err = link.Tracepoint(parts[0], parts[1], program, nil)
		case ebpf.Kprobe:
			if strings.HasPrefix(programSpec.SectionName, "uprobe/") {
				t.uprobes[programSpec.Name] = program
				continue
			}
			if strings.HasPrefix(programSpec.SectionName, "kretprobe/") {
				l, err = link.Kretprobe(programSpec.AttachTo, program, nil)
				if err == nil {
					t.links = append(t.links, l)
				}
				continue
			}
			l, err = link.Kprobe(programSpec.AttachTo, program, nil)
		}
		if err != nil {
			lastErr = err
			t.Close()
			klog.Errorf("LinkEbpfProg failed to program[%s] link program: %s", programSpec.Name, err)
			//return fmt.Errorf("failed to link program: %w", err)
		} else {
			t.links = append(t.links, l)
		}
	}
	klog.Infof("[end] Look eBPF specPrograms")

	if lastErr != nil {
		return fmt.Errorf("failed to link program: %w", lastErr)
	}
	return nil
}

func (t *Tracer) UnlinkEbpfProg() error {
	var (
		lastErr error
		err     error
	)
	/* 此处不应该处理 t.uprobes 中 ebpf-program
	for pName, p := range t.uprobes {
		if err = p.Close(); err != nil {
			lastErr = err
			klog.Errorf("UnlinkEbpfProg close program[%s] uprobe occurs error: %s", pName, err.Error())
		}
	}
	*/

	for _, l := range t.links {
		if err = l.Close(); err != nil {
			lastErr = err
			klog.Errorf("UnlinkEbpfProg close link occurs error: %s", err.Error())
		}
	}
	return lastErr
}

func (t EventType) Int() int {
	return int(t)
}

func (t EventType) String() string {
	switch t {
	case EventTypeProcessStart:
		return "process-start"
	case EventTypeProcessExit:
		return "process-exit"
	case EventTypeConnectionOpen:
		return "connection-open"
	case EventTypeConnectionClose:
		return "connection-close"
	case EventTypeConnectionError:
		return "connection-error"
	case EventTypeListenOpen:
		return "listen-open"
	case EventTypeListenClose:
		return "listen-close"
	case EventTypeFileOpen:
		return "file-open"
	case EventTypeTCPRetransmit:
		return "tcp-retransmit"
	case EventTypeL7Request:
		return "l7-request"
	}
	return "unknown: " + strconv.Itoa(int(t))
}

func (t EventReason) String() string {
	switch t {
	case EventReasonNone:
		return "none"
	case EventReasonOOMKill:
		return "oom-kill"
	}
	return "unknown: " + strconv.Itoa(int(t))
}

type procEvent struct {
	Type   EventType
	Pid    uint32
	Reason uint32
}

type tcpEvent struct {
	Fd             uint64
	Timestamp      uint64
	Duration       uint64
	FirstReadTime  uint64
	FirstWriteTime uint64
	NewReadTime    uint64
	Type           EventType
	Pid            uint32
	BytesSent      uint64
	BytesReceived  uint64
	SPort          uint16
	DPort          uint16
	SAddr          [16]byte
	DAddr          [16]byte
}

type fileEvent struct {
	Type EventType
	Pid  uint32
	Fd   uint64
}

// struct l7_event in l7.c
type l7Event struct {
	Fd                  uint64
	ConnectionTimestamp uint64
	Pid                 uint32
	Status              uint32
	Duration            uint64
	Protocol            uint8
	Method              uint8
	Padding             uint16
	StatementId         uint32
	PayloadSize         uint64
	TraceId             uint64
	StartAt             uint64 // ns
	EndtAt              uint64 // ns
	TraceStart          uint32
	TraceEnd            uint32
	TraceType           uint32
	EventCount          uint32
	Sport               uint16
	Dport               uint16
	SAddr               HashByte16
	DAddr               HashByte16
	ComponentSport      uint16
	ComponentDport      uint16
	IsTls               uint16
	ComponentSAddr      HashByte16
	ComponentDAddr      HashByte16
	AssumedAppId        HashByte
	SpanId              HashByte
	TraceIdFrom         HashByte16
	CalledId            HashByte
	InstanceIdFrom      HashByte
	AppIdFrom           HashByte
	SpanIdFrom          HashByte
	TypeFrom            [1]byte
	SysvcFrom           [76]byte // apmother header value: NN:ParentServiceName:MM:ParentServiceSysTag
	RPCTarget           [64]byte
	ErrorMsg            HashByte128
	MQ                  struct {
		Topic [256]byte // MQ topic (e.g., Kafka topic)
		Key   [256]byte // MQ key (e.g., Kafka message key)
	}
}

type SocketDataBufferddd struct {
	EventsNum uint32
	Len       uint32
	Data      [32760]byte
}

const (
	TASK_COMM_LEN       = 16
	BURST_DATA_BUF_SIZE = 8192
)

type Tuple struct {
	Daddr      [16]uint8
	RcvSaddr   [16]uint8
	AddrLen    uint8
	L4Protocol uint8
	Dport      uint16
	Num        uint16
}

type SocketDatadddd struct {
	Pid            uint32 // 表示线程号 如果'pid == tgid'表示一个进程, 否则是线程
	Tgid           uint32 // 进程号
	CoroutineID    uint64
	Source         uint8
	Comm           [TASK_COMM_LEN]byte
	SocketID       uint64
	Tuple          Tuple
	ExtraData      uint32
	ExtraDataCount uint32
	TcpSeq         uint32
	ThreadTraceID  uint64
	Timestamp      uint64
	Direction      uint8
	MsgType        uint8
	SyscallLen     uint64
	DataSeq        uint64
	DataType       uint16
	DataLen        uint16
	Data           [BURST_DATA_BUF_SIZE]byte
}

type StackEvent struct {
	Type        uint64
	Pid         uint64
	TraceId     uint64
	Goid        uint64
	Ip          uint64
	Bp          uint64
	CallerIp    uint64
	CallerBp    uint64
	TimeNsStart uint64
	TimeNsEnd   uint64
	// Nid         uint64
	// Fpid        uint64
	// Level       uint64
	Location   byte
	ClassName  [100]byte
	MethedName [100]byte
}

type StackFunEvent struct {
	StackEvent StackEvent
	Uprobe     *tracer.Uprobe
}

type pythonThreadEvent struct {
	Type     EventType
	Pid      uint32
	Duration uint64
}

func runEventsReader(name string, r *perf.Reader, ch chan<- Event, typ perfMapType) {
	for {
		rec, err := r.Read()
		if err != nil {
			if errors.Is(err, perf.ErrClosed) {
				break
			}
			continue
		}
		if rec.LostSamples > 0 {
			klog.Errorln(name, "lost samples:", rec.LostSamples)
			continue
		}
		var event Event

		switch typ {
		case perfMapTypeSocketEvents:
			//fmt.Println("perfMapTypeSocketEvents")
			// 假设 rec.RawSample 包含数据，类型为 []byte
			//rawData := rec.RawSample
			//fmt.Println("perfMapTypeSocketEvents2")
			//
			//// 创建一个 SocketDataBuffer 结构体实例
			//var buffer SocketDataBuffer
			//
			//// 创建一个字节缓冲区，并将数据填充到其中
			//reader := bytes.NewReader(rawData)
			//fmt.Println("perfMapTypeSocketEvents3")
			//fmt.Println(len(rawData))
			//// 使用 binary.Read 函数读取数据并解析为 SocketDataBuffer 结构体实例
			//if err := binary.Read(reader, binary.LittleEndian, &buffer); err != nil {
			//	fmt.Println(reader.Len())
			//	fmt.Println("Failed to read data:", err)
			//	continue
			//}
			//fmt.Println("perfMapTypeSocketEvents4")
			//
			//// 打印解析后的数据
			//fmt.Println("EventsNum:", buffer.EventsNum)
			//fmt.Println("Len:", buffer.Len)
			//
			//// 打印 char data 的内容
			//fmt.Printf("Data: %s\n", string(buffer.Data[:buffer.Len])) // 仅打印实际长度的数据

			//socketDataBuffer := rec.RawSample

			// 解析 __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
			//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))

			/*todo */
			//socketData := (*(*[128]byte)(unsafe.Pointer(&eventC.line)))
			//dataPtr := unsafe.Pointer(&buf.data[0])
			//socketData := (*SocketData)(dataPtr)

			//reader2 := bytes.NewBuffer(rec.RawSample)

			// 222222

			//fmt.Println("socketData.Pid:", socketData.pid)
			//fmt.Println("socketData.Tgid:", socketData.tgid)
			//fmt.Println("socketData.CoroutineID:", socketData.coroutine_id)
			//fmt.Println("socketData.Source:", socketData.source)
			//
			//fmt.Printf("socketData.Comm: %s \n", socketData.comm)
			//fmt.Printf("socketData.SocketID :%v \n", socketData.socket_id)
			//fmt.Println("socketData.Tuple:", socketData.Tuple)
			//fmt.Println("socketData.ExtraData:", socketData.ExtraData)
			//fmt.Println("socketData.ExtraDataCount:", socketData.ExtraDataCount)
			//fmt.Println("socketData.TCPSeq:", socketData.TcpSeq)
			//fmt.Println("socketData.ThreadTraceID:", socketData.ThreadTraceID)
			//fmt.Println("socketData.Timestamp:", socketData.Timestamp)
			//fmt.Println("socketData.Direction:", socketData.Direction)
			//fmt.Println("socketData.MsgType:", socketData.MsgType)
			//fmt.Println("socketData.SyscallLen:", socketData.SyscallLen)
			//fmt.Println("socketData.DataSeq:", socketData.DataSeq)

			//socketData := &SocketData{}
			//reader := bytes.NewBuffer(rec.RawSample)
			//if err := binary.Read(reader, binary.LittleEndian, v); err != nil {
			//	klog.Warningln("failed1 to read msg:", err)
			//	continue
			//}

			//
			//var data []byte
			//payload := reader.Bytes()
			//switch {
			//case v.Len == 0:
			//case v.Len > 32760:
			//	data = payload[:32760]
			//default:
			//	data = payload[:v.Len]
			//}
			//////data2 := data[:v.Len]
			////fmt.Println("perfMapTypeSocketEvents")
			//fmt.Println(v.EventsNum)
			//fmt.Println(v.Len)
			//fmt.Println(string(data))
			//
			//var data2 SocketData
			//reader2 := bytes.NewBuffer(data)
			//if err := binary.Read(reader2, binary.LittleEndian, data2); err != nil {
			//	klog.Warningln("failed2 to read msg:", err)
			//	continue
			//}
			//
			//fmt.Println(data2.Pid)
			//fmt.Println(data2.Tgid)

			//fmt.Println(string(v.Data))
			//continue
		case perfMapTypeL7Events:
			v := &l7Event{}
			reader := bytes.NewBuffer(rec.RawSample)
			if err := binary.Read(reader, binary.LittleEndian, v); err != nil {
				klog.Warningln("failed to read msg:", err)
				continue
			}
			//fmt.Println("v.TraceIdFrom")
			//fmt.Println(v.TraceIdFrom)
			//a := hex.EncodeToString(v.TraceIdFrom[:])
			//for _, b := range v.AssumedAppId {
			//	fmt.Printf("v.AssumedAppId- %02\n", b)
			//}
			//fmt.Println(a)

			payload := reader.Bytes()
			req := &l7.RequestData{
				Protocol:       l7.Protocol(v.Protocol),
				Pid:            v.Pid,
				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,
				TraceType:      v.TraceType,
				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),
				DestAddrString: utils.BytesToString(v.RPCTarget[:]),
				ErrorMsg:       utils.BytesToString(v.ErrorMsg[:]),
				IsTls:          v.IsTls > 0,
				MQTopic:        utils.BytesToString(v.MQ.Topic[:]),
				MQKey:          utils.BytesToString(v.MQ.Key[:]),
			}
			if req.Protocol == l7.ProtocolHTTP {
				klog.Debugf("runEventsReader ComponentSAddr.String %s", req.ComponentSAddr.String())
				klog.Debugf("runEventsReader ComponentDAddr.String %s", req.ComponentDAddr.String())
			}
			if v.TraceEnd == TRACE_STATUS {
				req.ParentSpanContext.TraceIdFrom = hex.EncodeToString(v.TraceIdFrom[:])
				req.ParentSpanContext.CalledId = hex.EncodeToString(v.CalledId[:])
				req.ParentSpanContext.InstanceIdFrom = hex.EncodeToString(v.InstanceIdFrom[:])
				req.ParentSpanContext.AppIdFrom = hex.EncodeToString(v.AppIdFrom[:])
				req.ParentSpanContext.SpanIdFrom = hex.EncodeToString(v.SpanIdFrom[:])
				req.ParentSpanContext.TypeFrom = l7.TypeFrom(hex.EncodeToString(v.TypeFrom[:]))
				req.SysvcFrom = utils.BytesToString(v.SysvcFrom[:])
					klog.Debugf("apmother '%s'", req.SysvcFrom)
				// klog.Debugf("req.ParentSpanContext.TraceIdFrom %s", req.ParentSpanContext.TraceIdFrom)
				// klog.Debugf("req.ParentSpanContext.TypeFrom %s", req.ParentSpanContext.TypeFrom)
				req.SAddr = ipPort(v.SAddr, v.Sport)
				req.DAddr = ipPort(v.DAddr, v.Dport)
				// klog.Debugf("runEventsReader SAddr.String %s", req.SAddr.String())
				// klog.Debugf("runEventsReader DAddr.String %s", req.DAddr.String())
			}
			switch {
			case v.PayloadSize == 0:
			case v.PayloadSize > MaxPayloadSize:
				req.Payload = payload[:MaxPayloadSize]
			default:
				req.Payload = payload[:v.PayloadSize]
			}
			//fmt.Println("==========")
			//fmt.Println("req.Payload:", string(req.Payload))
			//fmt.Println("==========")
			event = Event{Type: EventTypeL7Request, Pid: v.Pid, Fd: v.Fd, Timestamp: v.ConnectionTimestamp, L7Request: req}
		case perfMapTypeFileEvents:
			v := &fileEvent{}
			if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil {
				klog.Warningln("failed to read msg:", err)
				continue
			}
			event = Event{Type: v.Type, Pid: v.Pid, Fd: v.Fd}
		case perfMapTypeProcEvents:
			v := &procEvent{}
			if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil {
				klog.Warningln("failed to read msg:", err)
				continue
			}
			event = Event{Type: v.Type, Reason: EventReason(v.Reason), Pid: v.Pid}
		case perfMapTypeTCPEvents:
			v := &tcpEvent{}
			if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil {
				klog.Warningln("failed to read msg:", err)
				continue
			}
			event = Event{
				Type:      v.Type,
				Pid:       v.Pid,
				SrcAddr:   ipPort(v.SAddr, v.SPort),
				DstAddr:   ipPort(v.DAddr, v.DPort),
				Fd:        v.Fd,
				Timestamp: v.Timestamp,
				Duration:  time.Duration(v.Duration),
			}
			if v.Type == EventTypeConnectionClose {
				event.TrafficStats = &TrafficStats{
					BytesSent:     v.BytesSent,
					BytesReceived: v.BytesReceived,
				}
			}
			event.FirstReadTime = v.FirstReadTime
			event.FirstWriteTime = v.FirstWriteTime
			event.NewReadTime = v.NewReadTime
			if v.Type == EventTypeAcceptClose {
				event.TrafficStats = &TrafficStats{
					BytesSent:     v.BytesSent,
					BytesReceived: v.BytesReceived,
				}
			}
		case perfMapTypePythonThreadEvents:
			v := &pythonThreadEvent{}
			if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil {
				klog.Warningln("failed to read msg:", err)
				continue
			}
			event = Event{
				Type:     v.Type,
				Pid:      v.Pid,
				Duration: time.Duration(v.Duration),
			}
		case perfMapTypeEventQueue:
			v := &StackEvent{}
			if err := binary.Read(bytes.NewBuffer(rec.RawSample), binary.LittleEndian, v); err != nil {
				klog.Warningln("failed to read msg:", err)
				continue
			}

			event = Event{
				Type:       EventTypeFunEnt,
				StackEvent: v,
			}
		default:
			continue
		}

		ch <- event
	}
}

func ipPort(ip [16]byte, port uint16) netaddr.IPPort {
	i, _ := netaddr.FromStdIP(ip[:])
	return netaddr.IPPortFrom(i, port)
}

func (t *Tracer) InitKProcInfo(pid uint32, appInfo *AppInfo) error {
	fmt.Println("InitKProcInfo111111111111111111111111111111111")
	var err error
	var info EbpfProcInfo
	if appInfo.EBPFProcInfo == nil {
		info = EbpfProcInfo{
			InstanceId: appInfo.InstanceIdHash.HashtVal,
			AppId:      appInfo.AppIdHash.HashtVal,
			CodeType:   uint16(appInfo.CodeType),
		}
	} else {
		info = *appInfo.EBPFProcInfo
		info.AppId = appInfo.AppIdHash.HashtVal
	}

	// 构建 sysvc 值：{app_name_len}:app_name:{appServiceType_len}:appServiceType:{SysTagLen}[:SysTag]
	// appServiceType 固定为 "APPLICATION"
	appName := appInfo.AppName
	nodeInfo := NODE_INFO.GetNodeInfo()
	sysTag := ""
	if nodeInfo != nil {
		sysTag = nodeInfo.SysTag
	}
	appServiceType := "APPLICATION"

	appNameLen := len(appName)
	appServiceTypeLen := len(appServiceType)
	sysTagLen := len(sysTag)

	// 限制长度，避免超出数组大小
	if appNameLen > 32 {
		appNameLen = 32
		appName = appName[:32]
	}
	if sysTagLen > 32 {
		sysTagLen = 32
		sysTag = sysTag[:32]
	}

	// 格式：{app_name_len}:app_name:{appServiceType_len}:appServiceType[:{SysTagLen}:SysTag]
	// 如果 sysTag 为空，则为 {app_name_len}:app_name:{appServiceType_len}:appServiceType
	// 如果 sysTag 不为空，则为 {app_name_len}:app_name:{appServiceType_len}:appServiceType:{SysTagLen}:SysTag
	// 长度不足2位时前面补0，例如：08:service:12:APPLICATION 或 08:service:12:APPLICATION:03:tag
	var sysvcStr string
	if sysTagLen == 0 {
		sysvcStr = fmt.Sprintf("%02d:%s:%02d:%s", appNameLen, appName, appServiceTypeLen, appServiceType)
	} else {
		sysvcStr = fmt.Sprintf("%02d:%s:%02d:%s:%02d:%s", appNameLen, appName, appServiceTypeLen, appServiceType, sysTagLen, sysTag)
	}

	// 复制到字节数组，确保不超过 76 字节
	sysvcBytes := []byte(sysvcStr)
	if len(sysvcBytes) > 76 {
		sysvcBytes = sysvcBytes[:76]
	}
	copy(info.Sysvc[:], sysvcBytes)
	_, err = tracer.UpdateProcInfoToMap(t.collection, pid, info)
	if err != nil {
		klog.Error("failed to update program info", err)
	}
	appInfo.EBPFProcInfo = &info
	return err
}

func (t *Tracer) DelKProcInfo(pid uint32) error {
	_, err := tracer.DelProcInfoFromMap(t.collection, pid)
	if err != nil {
		klog.WithField("pid", pid).Error("failed to delete proc info", err)
	}
	return err
}

// TODO check language
func (t *Tracer) DisableL7Tracing() bool {
	return t.disableL7Tracing
}

func (t *Tracer) DisableE2ETracing() bool {
	return t.disableE2ETracing
}

func (t *Tracer) DisableStackTracing() bool {
	return t.disableStackTracing
}