ebpf/ebpftracer/tracer/socket.go

package tracer

import (
	"fmt"
	"net"
	"os"
	"runtime"
	"strings"
	"syscall"
	"time"

	"github.com/cilium/ebpf"
	"github.com/cilium/ebpf/btf"
	"github.com/coroot/coroot-node-agent/flags"
	"github.com/coroot/coroot-node-agent/utils"
	. "github.com/coroot/coroot-node-agent/utils/modelse"
	klog "github.com/sirupsen/logrus"
)

func init() {
	enable_ebpf_protocol(PROTO_HTTP1)
	enable_ebpf_protocol(PROTO_HTTP2)
	enable_ebpf_protocol(PROTO_TLS_HTTP1)
	enable_ebpf_protocol(PROTO_TLS_HTTP2)
	enable_ebpf_protocol(PROTO_DUBBO)
	enable_ebpf_protocol(PROTO_SOFARPC)
	enable_ebpf_protocol(PROTO_MYSQL)
	enable_ebpf_protocol(PROTO_POSTGRESQL)
	enable_ebpf_protocol(PROTO_REDIS)
	enable_ebpf_protocol(PROTO_KAFKA)
	enable_ebpf_protocol(PROTO_MQTT)
	enable_ebpf_protocol(PROTO_DNS)
}

func MapInit(collectionSpec *ebpf.CollectionSpec, opts *ebpf.CollectionOptions) {
	set_offset_map(collectionSpec, opts)
	set_conf_map_default(collectionSpec, opts)

	//offsetData := make([]any, runtime.NumCPU())
	//for i := range offsetData {
	//	offsetData[i] = testStruct{
	//		test_id: 99999,
	//	}
	//}
	//if bpf_table_set_value(collectionSpec, opts, "test_heap", offsetData) != ETR_OK {}
	//insert_output_prog_to_map(collectionSpec, opts)
}

func MapInsert(collection *ebpf.Collection) {
	insert_output_prog_to_map(collection)
	insert_adapt_kern_uid_to_map(collection)

	// Update go offsets to eBPF "proc_info_map"
	//update_proc_info_to_map(collection)

	// Update protocol filter array
	update_protocol_filter_array(collection)

	// 从flags配置获取CodeType配置，默认支持Python和Node
	update_skmsg_header_allowed_to_map(collection)
}

func insert_output_prog_to_map(collection *ebpf.Collection) {
	// jmp for tracepoints
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_TP_NAME, PROG_DATA_SUBMIT_NAME_FOR_TP, PROG_DATA_SUBMIT_TP_IDX)
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_TP_NAME, PROG_OUTPUT_DATA_NAME_FOR_TP, PROG_OUTPUT_DATA_TP_IDX)
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_TP_NAME, PROG_IO_EVENT_NAME_FOR_TP, PROG_IO_EVENT_TP_IDX)

	// jmp for kprobe/uprobe
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_KP_NAME, PROG_DATA_SUBMIT_NAME_FOR_KP, PROG_DATA_SUBMIT_KP_IDX)
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_KP_NAME, PROG_OUTPUT_DATA_NAME_FOR_KP, PROG_OUTPUT_DATA_KP_IDX)

	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_UP_NAME, PROG_SAVE_SC_DATA_FOR_UP, PROG_DATA_SAVE_UP_IDX)
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_UP_NAME, PROG_JAVA_UPDATE_HEADER_FOR_UP, PROG_DATA_JAVA_UPDATE_HEADER_UP_IDX)
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_UP_NAME, PROG_JAVA_FIND_HOST_FOR_UP, PROG_DATA_JAVA_FIND_HOST_UP_IDX)
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_UP_NAME, PROG_JAVA_BUILD_HEADER_FOR_UP, PROG_DATA_JAVA_BUILD_HEADER_UP_IDX)
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_UP_NAME, PROG_GO_UPDATE_HEADER_FOR_UP, PROG_DATA_GO_UPDATE_HEADER_UP_IDX)
	//add for l7.c -> is_http_request(payload) -> bpf_tail_call
	__insert_output_prog_to_map(collection, MAP_PROGS_JMP_TP_NAME, PROG_L7_HTTP_TRACE_ID_FOR_TP, PROG_DATA_L7_HTTP_TRACE_ID_TP_IDX)
}

func __insert_output_prog_to_map(collection *ebpf.Collection, mapName string, progName string, key uint32) {
	// find in programs
	prog, ok := collection.Programs[progName]
	//fmt.Println(prog, ok)
	if ok {
		progFd := prog.FD()
		code, err := bpf_table_set_value(collection, mapName, key, uint32(progFd))
		if err != nil {
			klog.Error(err, code)
		}
	}
}

func update_protocol_filter_array(collection *ebpf.Collection) {
	for i := 0; i < PROTO_NUM; i++ {
		code, err := bpf_table_set_value(collection, MAP_PROTO_FILTER_NAME, uint32(i), EbpfConfigProtocolFilter[i])
		if err != nil || code != ETR_OK {
			klog.Error(err, code)
		}
	}
}

//func update_allow_port_bitmap(collection *ebpf.Collection) {
//	for i := 0; i < PROTO_NUM; i++ {
//		if bpf_table_set_value(collection, MAP_ALLOW_PORT_BITMAP_NAME, 0, &allow_port_bitmap) != ETR_OK {
//			fmt.Println("no")
//		} else {
//			fmt.Println("ok")
//		}
//	}
//}

func insert_adapt_kern_uid_to_map(collection *ebpf.Collection) {
	pid := os.Getpid()
	tid := syscall.Gettid()
	adaptKernUID := uint64(pid)<<32 | uint64(tid)
	code, err := bpf_table_set_value(collection, MAP_ADAPT_KERN_UID_NAME, uint32(0), adaptKernUID)
	if err != nil || code != ETR_OK {
		klog.Error(err, code)
	}
}

func enable_ebpf_protocol(protocol int) {
	if protocol < PROTO_NUM {
		EbpfConfigProtocolFilter[protocol] = 1
	}
}

func Offset() {
	listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", offsetInferServerAddr, offsetInferServerPort))
	if err != nil {
		fmt.Errorf("failed to create server listener: %v", err)
		return
	}
	if err := kernelOffsetInferServer(listener); err != nil {
		fmt.Printf("Error in kernel_offset_infer_server: %v\n", err)
	}
	if err := kernelOffsetInferClient(); err != nil {
		fmt.Printf("Error in kernel_offset_infer_client: %v\n", err)
	}
	defer listener.Close()
}

func set_offset_map(collectionSpec *ebpf.CollectionSpec, opts *ebpf.CollectionOptions) {
	// 解析BTF数据
	if update_offset_map_from_btf_vmlinux(collectionSpec, opts) != ETR_OK {
		klog.Infof("[eBPF Kernel Adapt] Set offsets map from btf_vmlinux, not support.")
		if update_offset_map_default(collectionSpec, opts) != ETR_OK {
			klog.Infof("[eBPF Kernel Adapt] Set offsets error, failed to update default offse.t")
		} else {
			klog.Infof("[eBPF Kernel Adapt] Set offsets map use default.")
		}
	} else {
		klog.Infof("[eBPF Kernel Adapt] Set offsets map from btf_vmlinux, success.")
	}
}

// __u64 socket_id;       // 会话标识
// __u64 coroutine_trace_id;  // 同一协程的数据转发关联
// __u64 thread_trace_id; // 同一进程/线程的数据转发关联，用于多事务流转场景
// __u64 data_limit_max;  // Maximum number of data transfers
// __u64 go_tracing_timeout;
// __u64 io_event_collect_mode;
// __u64 io_event_minimal_duration;
func set_conf_map_default(collectionSpec *ebpf.CollectionSpec, opts *ebpf.CollectionOptions) {
	//charHostID := utils.GetHostIDBPFString()
	//_, charAppID := utils.GetAppID()

	uidBase := uint64(time.Now().UnixNano()/int64(time.Millisecond)) & 0xffffffffffffff
	numCPU := runtime.NumCPU()
	nCPU, _ := utils.GetCPUCount()
	tConf := make([]any, numCPU)
	for i := range tConf {
		socketID := uint64(i)<<56 | uint64(uidBase)
		tracerConf := EbpfTraceConf{
			SocketID:               socketID,
			CoroutineTraceID:       socketID,
			ThreadTraceID:          socketID,
			DataLimitMax:           4096,
			GoTracingTimeout:       120,
			IOEventCollectMode:     1,
			IOEventMinimalDuartion: 1000000,
			//HostID:                 utils.GetHostIDBPFString(),
			//APPID:                  charAppID,
			TotalCpus: uint64(nCPU),
		}
		tConf[i] = tracerConf
	}
	if bpf_table_pre_set_value(collectionSpec, opts, MAP_TRACE_CONF_NAME, tConf) != ETR_OK {
		klog.Infof("[eBPF Kernel Adapt] Set config map from btf_vmlinux, not support.")
	} else {
		klog.Infof("[eBPF Kernel Adapt] Set config map from btf_vmlinux, success.")
	}
}

// 解析BTF数据
func update_offset_map_from_btf_vmlinux(collectionSpec *ebpf.CollectionSpec, opts *ebpf.CollectionOptions) int {
	btfSpec, err := btf.LoadKernelSpec()
	if err != nil || btfSpec == nil {
		klog.WithError(err).Warning("[eBPF Kernel Adapt] Failed to get btf.LoadKernelSpec")
		return ETR_NOTSUPP
	}

	copied_seq_offs := kernel_struct_field_offset(btfSpec, "tcp_sock", "copied_seq")
	write_seq_offs := kernel_struct_field_offset(btfSpec, "tcp_sock", "write_seq")
	files_offs := kernel_struct_field_offset(btfSpec, "task_struct", "files")
	sk_flags_offs := kernel_struct_field_offset(btfSpec, "sock", "__sk_flags_offset")

	if sk_flags_offs == ETR_NOTEXIST {
		sk_flags_offs = kernel_struct_field_offset(btfSpec, "sock", "sk_pacing_shift")
		if sk_flags_offs > 0 {
			sk_flags_offs -= 1
		}
	}

	struct_files_struct_fdt_offset := kernel_struct_field_offset(btfSpec, "files_struct", "fdt")
	struct_files_private_data_offset := kernel_struct_field_offset(btfSpec, "file", "private_data")
	struct_file_f_inode_offset := kernel_struct_field_offset(btfSpec, "file", "f_inode")
	struct_inode_i_mode_offset := kernel_struct_field_offset(btfSpec, "inode", "i_mode")
	struct_file_dentry_offset_1 := kernel_struct_field_offset(btfSpec, "file", "f_path")
	struct_file_dentry_offset_2 := kernel_struct_field_offset(btfSpec, "path", "dentry")
	if struct_file_dentry_offset_1 < 0 ||
		struct_file_dentry_offset_2 < 0 {
		return ETR_NOTSUPP
	}

	struct_file_dentry_offset := struct_file_dentry_offset_1 + struct_file_dentry_offset_2
	struct_dentry_name_offset_1 := kernel_struct_field_offset(btfSpec, "dentry", "d_name")
	struct_dentry_name_offset_2 := kernel_struct_field_offset(btfSpec, "qstr", "name")
	if struct_dentry_name_offset_1 < 0 ||
		struct_dentry_name_offset_2 < 0 {
		return ETR_NOTSUPP
	}

	struct_dentry_name_offset := struct_dentry_name_offset_1 + struct_dentry_name_offset_2
	struct_sock_family_offset := kernel_struct_field_offset(btfSpec, "sock_common", "skc_family")
	struct_sock_saddr_offset := kernel_struct_field_offset(btfSpec, "sock_common", "skc_rcv_saddr")
	struct_sock_daddr_offset := kernel_struct_field_offset(btfSpec, "sock_common", "skc_daddr")
	struct_sock_ip6saddr_offset := kernel_struct_field_offset(btfSpec, "sock_common", "skc_v6_rcv_saddr")
	struct_sock_ip6daddr_offset := kernel_struct_field_offset(btfSpec, "sock_common", "skc_v6_daddr")
	struct_sock_dport_offset := kernel_struct_field_offset(btfSpec, "sock_common", "skc_dport")
	struct_sock_sport_offset := kernel_struct_field_offset(btfSpec, "sock_common", "skc_num")
	struct_sock_skc_state_offset := kernel_struct_field_offset(btfSpec, "sock_common", "skc_state")
	struct_sock_common_ipv6only_offset := kernel_struct_field_offset(btfSpec, "sock_common", "skc_flags")

	klog.Infof("Offsets from BTF vmlinux:")
	klog.Infof("    copied_seq_offs: 0x%x", copied_seq_offs)
	klog.Infof("    write_seq_offs: 0x%x", write_seq_offs)
	klog.Infof("    files_offs: 0x%x", files_offs)
	klog.Infof("    sk_flags_offs: 0x%x", sk_flags_offs)
	klog.Infof("    struct_files_struct_fdt_offset: 0x%x", struct_files_struct_fdt_offset)
	klog.Infof("    struct_files_private_data_offset: 0x%x", struct_files_private_data_offset)
	klog.Infof("    struct_file_f_inode_offset: 0x%x", struct_file_f_inode_offset)
	klog.Infof("    struct_inode_i_mode_offset: 0x%x", struct_inode_i_mode_offset)
	klog.Infof("    struct_file_dentry_offset: 0x%x", struct_file_dentry_offset)
	klog.Infof("    struct_dentry_name_offset: 0x%x", struct_dentry_name_offset)
	klog.Infof("    struct_sock_family_offset: 0x%x", struct_sock_family_offset)
	klog.Infof("    struct_sock_saddr_offset: 0x%x", struct_sock_saddr_offset)
	klog.Infof("    struct_sock_daddr_offset: 0x%x", struct_sock_daddr_offset)
	klog.Infof("    struct_sock_ip6saddr_offset: 0x%x", struct_sock_ip6saddr_offset)
	klog.Infof("    struct_sock_ip6daddr_offset: 0x%x", struct_sock_ip6daddr_offset)
	klog.Infof("    struct_sock_dport_offset: 0x%x", struct_sock_dport_offset)
	klog.Infof("    struct_sock_sport_offset: 0x%x", struct_sock_sport_offset)
	klog.Infof("    struct_sock_skc_state_offset: 0x%x", struct_sock_skc_state_offset)
	klog.Infof("    struct_sock_common_ipv6only_offset: 0x%x", struct_sock_common_ipv6only_offset)

	if copied_seq_offs < 0 || write_seq_offs < 0 || files_offs < 0 ||
		sk_flags_offs < 0 || struct_files_struct_fdt_offset < 0 ||
		struct_files_private_data_offset < 0 ||
		struct_file_f_inode_offset < 0 || struct_inode_i_mode_offset < 0 ||
		struct_inode_i_mode_offset < 0 || struct_file_dentry_offset < 0 ||
		struct_dentry_name_offset < 0 || struct_sock_family_offset < 0 ||
		struct_sock_saddr_offset < 0 || struct_sock_daddr_offset < 0 ||
		struct_sock_ip6saddr_offset < 0 ||
		struct_sock_ip6daddr_offset < 0 || struct_sock_dport_offset < 0 ||
		struct_sock_sport_offset < 0 || struct_sock_skc_state_offset < 0 ||
		struct_sock_common_ipv6only_offset < 0 {
		return ETR_NOTSUPP
	}

	offset := BpfOffsetParam{}
	offset.Ready = 1
	offset.TaskFilesOffset = uint32(files_offs)
	offset.SockFlagsOffset = uint32(sk_flags_offs)
	offset.TcpSockCopiedSeqOffset = uint32(copied_seq_offs)
	offset.TcpSockWriteSeqOffset = uint32(write_seq_offs)
	offset.StructFilesStructFdtOffset = uint32(struct_files_struct_fdt_offset)
	offset.StructFilesPrivateDataOffset = uint32(struct_files_private_data_offset)
	offset.StructFileFInodeOffset = uint32(struct_file_f_inode_offset)
	offset.StructInodeIModeOffset = uint32(struct_inode_i_mode_offset)
	offset.StructFileDentryOffset = uint32(struct_file_dentry_offset)
	offset.StructDentryNameOffset = uint32(struct_dentry_name_offset)
	offset.StructSockFamilyOffset = uint32(struct_sock_family_offset)
	offset.StructSockSaddrOffset = uint32(struct_sock_saddr_offset)
	offset.StructSockDaddrOffset = uint32(struct_sock_daddr_offset)
	offset.StructSockIp6saddrOffset = uint32(struct_sock_ip6saddr_offset)
	offset.StructSockIp6daddrOffset = uint32(struct_sock_ip6daddr_offset)
	offset.StructSockDportOffset = uint32(struct_sock_dport_offset)
	offset.StructSockSportOffset = uint32(struct_sock_sport_offset)
	offset.StructSockSkcStateOffset = uint32(struct_sock_skc_state_offset)
	offset.StructSockCommonIpv6onlyOffset = uint32(struct_sock_common_ipv6only_offset)

	if update_offsets_table(collectionSpec, opts, offset) != ETR_OK {
		return ETR_UPDATE_MAP_FAILD
	}
	return ETR_OK
}

func update_offset_map_default(collectionSpec *ebpf.CollectionSpec, opts *ebpf.CollectionOptions) int {
	offset := BpfOffsetParam{}

	offset.StructFilesStructFdtOffset = 0x20
	offset.StructFilesPrivateDataOffset = 0xc8
	offset.StructFileFInodeOffset = 0x20
	offset.StructInodeIModeOffset = 0x00
	offset.StructFileDentryOffset = 0x18
	offset.StructDentryNameOffset = 0x28
	offset.StructSockFamilyOffset = 0x10
	offset.StructSockSaddrOffset = 0x4
	offset.StructSockDaddrOffset = 0x0
	offset.StructSockIp6saddrOffset = 0x48
	offset.StructSockIp6daddrOffset = 0x38
	offset.StructSockDportOffset = 0xc
	offset.StructSockSportOffset = 0xe
	offset.StructSockSkcStateOffset = 0x12
	offset.StructSockCommonIpv6onlyOffset = 0x13

	if update_offsets_table(collectionSpec, opts, offset) != ETR_OK {
		return ETR_UPDATE_MAP_FAILD
	}
	return ETR_OK
}

func update_offsets_table(collectionSpec *ebpf.CollectionSpec, opts *ebpf.CollectionOptions, offset BpfOffsetParam) int {
	numCPU := runtime.NumCPU()
	offsetData := make([]any, numCPU)
	for i := range offsetData {
		offsetData[i] = offset
	}
	if bpf_table_pre_set_value(collectionSpec, opts, MAP_MEMBERS_OFFSET_NAME, offsetData) != ETR_OK {
		return ETR_UPDATE_MAP_FAILD
	}
	return ETR_OK
}

func SetConstants(collectionSpec *ebpf.CollectionSpec) {
	// nCPU, err := utils.GetCPUCount()
	// consts := map[string]interface{}{
	// TODO go Process
	// "buckets_ptr_pos":  int64(16),
	// "ctx_ptr_pos":      int64(232),
	// "headers_ptr_pos":  int64(56),
	// "request_host_pos": int64(128),
	// "is_registers_abi": true,
	// "method_ptr_pos":   int64(0),
	// "path_ptr_pos":     int64(56),
	// "status_code_pos":  int64(120),
	// "url_ptr_pos":      int64(16),

	// TODO 全局 ***
	// "total_cpus": int64(nCPU),
	//"apm_app_id":  int64(0),
	//"apm_host_id": int64(0),
	// }
	// err = collectionSpec.RewriteConstants(consts)
	// if err != nil {
	// 	fmt.Println("err", err, consts)
	// }
}

// parseCodeTypesFromFlags 从flags配置解析CodeType数组
// 配置格式: "python,nodejs" 或 "python,node" 等
// 支持的语言标识: python, node, nodejs, java, go, php, dotnet, netcore, c, lua, ruby
func parseCodeTypesFromFlags() []CodeType {
	var codeTypes []CodeType
	configValue := *flags.L4HeaderCodeTypes
	if configValue == "" {
		return codeTypes
	}
	parts := strings.Split(configValue, ",")
	for _, part := range parts {
		part = strings.TrimSpace(strings.ToLower(part))
		switch part {
		case "python":
			codeTypes = append(codeTypes, CodeTypePython)
		case "node", "nodejs":
			codeTypes = append(codeTypes, CodeTypeNode)
		case "java":
			codeTypes = append(codeTypes, CodeTypeJava)
		case "go":
			codeTypes = append(codeTypes, CodeTypeGo)
		case "php":
			codeTypes = append(codeTypes, CodeTypePHP)
		case "dotnet":
			codeTypes = append(codeTypes, CodeTypeDotNet)
		case "netcore":
			codeTypes = append(codeTypes, CodeTypeNetCore)
		case "c":
			codeTypes = append(codeTypes, CodeTypeC)
		case "lua":
			codeTypes = append(codeTypes, CodeTypeLua)
		case "ruby":
			codeTypes = append(codeTypes, CodeTypeRuby)
		case "nginx":
			codeTypes = append(codeTypes, CodeTypeNginx)
		default:
			klog.Warnf("Unknown code type in L4_HEADER_CODE_TYPES: %s", part)
		}
	}

	return codeTypes
}

func update_skmsg_header_allowed_to_map(collection *ebpf.Collection) {
	codeTypes := parseCodeTypesFromFlags()
	klog.Infof("[kernel] set l4 header allowed_to_map %s", codeTypes)
	for _, codeType := range codeTypes {
		_, err := bpf_table_set_any_value(collection, MAP_L4_HEADER_CODE_TYPES_NAME, uint16(codeType), uint8(1))
		if err != nil {
			klog.Errorf("[kernel] update_skmsg_header_allowed_to_map err: %v]")
		}
	}
}