feat: agent - eBPF Add DPDK packet capture (#8415)

* feat: agent - eBPF Add DPDK packet capture

* eBPF provides a timestamp in nanoseconds, used for sorting

* Adjusting the creation of probes

* Add configuration information

* Explanation of calling DPDK interface

* Adjust the process event structure

* Update template.yml

agent/Cargo.toml

@@ -154,8 +154,8 @@ tonic-build = "0.8.0"
 walkdir = "2"
 
 [features]
-enterprise = ["extended_profile", "enterprise-utils"]
-extended_profile = []
+enterprise = ["extended_observability", "enterprise-utils"]
+extended_observability = []
 dylib_pcap = []
 
 [[bench]]

agent/src/ebpf/kernel/include/common.h

@@ -89,7 +89,8 @@ enum traffic_protocol {
 	PROTO_DNS = 120,
 	PROTO_TLS = 121,
 	PROTO_CUSTOM = 127,
-	PROTO_NUM = 130
+	PROTO_DPDK_PKT = 199,
+	PROTO_NUM = 200
 };
 
 enum process_data_extra_source {
@@ -100,6 +101,7 @@ enum process_data_extra_source {
 	DATA_SOURCE_IO_EVENT,
 	DATA_SOURCE_GO_HTTP2_DATAFRAME_UPROBE,
 	DATA_SOURCE_CLOSE,
+	DATA_SOURCE_DPDK,
 };
 
 struct protocol_message_t {

agent/src/ebpf/kernel/socket_trace.bpf.c

@@ -440,7 +440,6 @@ static __inline struct member_fields_offset *retrieve_ready_kern_offset(void)
 
 #include "uprobe_base.bpf.c"
 #include "include/protocol_inference.h"
-#define EVENT_BURST_NUM            16
 #define CONN_PERSIST_TIME_MAX_NS   100000000000ULL
 
 static __inline struct trace_key_t get_trace_key(__u64 timeout,
@@ -2873,7 +2872,7 @@ static __inline int __output_data_common(void *ctx,
 	__u64 curr_time = bpf_ktime_get_ns();
 	__u64 diff = curr_time - tracer_ctx->last_period_timestamp;
 	if (diff > PERIODIC_PUSH_DELAY_THRESHOLD_NS ||
-	    v_buff->events_num >= EVENT_BURST_NUM ||
+	    v_buff->events_num >= MAX_EVENTS_BURST ||
 	    ((sizeof(v_buff->data) - v_buff->len) < sizeof(*v))) {
 		finalize_data_output(ctx, tracer_ctx, curr_time, diff, v_buff);
 	}
@@ -2972,7 +2971,7 @@ static __inline int output_data_common(void *ctx)
 	__u64 curr_time = bpf_ktime_get_ns();
 	__u64 diff = curr_time - tracer_ctx->last_period_timestamp;
 	if (diff > PERIODIC_PUSH_DELAY_THRESHOLD_NS ||
-	    v_buff->events_num >= EVENT_BURST_NUM ||
+	    v_buff->events_num >= MAX_EVENTS_BURST ||
 	    ((sizeof(v_buff->data) - v_buff->len) < sizeof(*v))) {
 		finalize_data_output(ctx, tracer_ctx, curr_time, diff, v_buff);
 	}

agent/src/ebpf/mod.rs

@@ -148,6 +148,16 @@ pub const DATA_SOURCE_IO_EVENT: u8 = 4;
 pub const DATA_SOURCE_GO_HTTP2_DATAFRAME_UPROBE: u8 = 5;
 #[allow(dead_code)]
 pub const DATA_SOURCE_CLOSE: u8 = 6;
+cfg_if::cfg_if! {
+    if #[cfg(feature = "extended_observability")] {
+        #[allow(dead_code)]
+        pub const DATA_SOURCE_DPDK: u8 = 7;
+        #[allow(dead_code)]
+        pub const DPDK_HOOK_TYPE_RECV: u8 = 0;
+        #[allow(dead_code)]
+        pub const DPDK_HOOK_TYPE_XMIT: u8 = 1;
+    }
+}
 
 // Message types
 // Currently, except for source=EBPF_TYPE_GO_HTTP2_UPROBE,
@@ -185,15 +195,15 @@ pub const PROFILER_TYPE_UNKNOWN: u8 = 0;
 #[allow(dead_code)]
 pub const PROFILER_TYPE_ONCPU: u8 = 1;
 cfg_if::cfg_if! {
-    if #[cfg(feature = "extended_profile")] {
+    if #[cfg(feature = "extended_observability")] {
         #[allow(dead_code)]
         pub const PROFILER_TYPE_OFFCPU: u8 = 2;
         #[allow(dead_code)]
         pub const PROFILER_TYPE_MEMORY: u8 = 3;
     }
 }
 
-#[cfg(feature = "extended_profile")]
+#[cfg(feature = "extended_observability")]
 pub const PROFILER_CTX_MEMORY_IDX: usize = 2;
 pub const PROFILER_CTX_NUM: usize = 3;
 
@@ -271,7 +281,7 @@ pub struct SK_BPF_DATA {
     pub syscall_trace_id_call: u64,
 
     /* data info */
-    pub timestamp: u64, // cap_data获取的时间戳（从1970.1.1开始到数据捕获时的时间间隔，精度为微妙）
+    pub timestamp: u64, // cap_data获取的时间戳（从1970.1.1开始到数据捕获时的时间间隔，精度为纳秒）
     pub direction: u8,  // 数据的收发方向，值是 SOCK_DIR_SND/SOCK_DIR_RCV
 
     /*
@@ -705,7 +715,7 @@ extern "C" {
     pub fn set_dwarf_shard_map_size(size: c_int) -> c_void;
 
     cfg_if::cfg_if! {
-        if #[cfg(feature = "extended_profile")] {
+        if #[cfg(feature = "extended_observability")] {
             pub fn enable_offcpu_profiler() -> c_int;
 
             pub fn disable_offcpu_profiler() -> c_int;
@@ -719,6 +729,60 @@ extern "C" {
             pub fn enable_memory_profiler() -> c_int;
 
             pub fn disable_memory_profiler() -> c_int;
+
+            /**
+             * @brief **set_dpdk_trace_enabled()** DPDK tracing feature enable switch.
+             *
+             * Note: The call must be executed before `running_socket_tracer()` because
+             * `set_dpdk_trace_enabled()` may need to adjust the eBPF maps before loading
+             * the eBPF program. The eBPF program loading process is implemented within
+             * `running_socket_tracer()`.
+             *
+             * @param enabled Used to control whether to enable this feature.
+             *   'true': enbaled; 'false': disabled
+             * @return 0 on success, non-zero on error
+             */
+            pub fn set_dpdk_trace_enabled(enabled: bool) -> c_int;
+
+            /**
+             * @brief **set_dpdk_cmd_name()** Set the command line name of the DPDK application.
+             *
+             * Note: The call must be executed before `dpdk_trace_start()`
+             *
+             * @param name Command name. For example, in the command line '/usr/bin/mydpdk',
+             *   the name selected is the part after the last '/', i.e., 'mydpdk'.
+             *
+             * @return 0 on success, non-zero on error
+             */
+             pub fn set_dpdk_cmd_name(name: *const c_char) -> c_int;
+
+             /**
+              * @brief **set_dpdk_hooks()** Set all DPDK hook points tracked by eBPF.
+              *
+              * Note: The call must be executed before `dpdk_trace_start()`
+              *
+              * @param fucs The list of tracked interfaces,
+              *   for example: i40e_recv_pkts,i40e_xmit_pkts,ixgbe_recv_pkts,ixgbe_xmit_pkts
+              * @param type Is DPDK_HOOK_TYPE_RECV or DPDK_HOOK_TYPE_XMIT, Indicates whether
+              *   it is receiving or transmitting packets.
+              *
+              * @return 0 on success, non-zero on error
+              */
+             pub fn set_dpdk_hooks(func_type: c_int, funcs: *const c_char) -> c_int;
+
+             /**
+              * @brief **dpdk_trace_start()** Start the DPDK tracing module.
+              *
+              * @return 0 on success, non-zero on error
+              */
+             pub fn dpdk_trace_start() -> c_int;
+
+             /**
+              * @brief **dpdk_trace_stop()** Stop the DPDK tracing module.
+              *
+              * @return 0 on success, non-zero on error
+              */
+              pub fn dpdk_trace_stop() -> c_int;
         }
     }
 }

agent/src/ebpf/samples/rust/socket-tracer/src/main.rs

@@ -17,9 +17,9 @@
 use chrono::prelude::DateTime;
 use chrono::FixedOffset;
 use chrono::Utc;
-use std::env;
 use socket_tracer::ebpf::*;
 use std::convert::TryInto;
+use std::env;
 use std::ffi::CString;
 use std::fmt::Write;
 use std::net::IpAddr;
@@ -597,6 +597,9 @@ fn main() {
             CString::new("443".as_bytes()).unwrap().as_c_str().as_ptr(),
         );
 
+	// dpdk enable
+	// set_dpdk_trace_enabled(true);
+
         if running_socket_tracer(
             socket_trace_callback, /* Callback interface rust -> C */
             1, /* Number of worker threads, indicating how many user-space threads participate in data processing */
@@ -617,6 +620,22 @@ fn main() {
         let result = set_feature_pids(feature, pids.as_ptr(), num);
         println!("Result {}", result);
 
+        // Test for dpdk
+        //set_dpdk_cmd_name(
+        //    CString::new("l2fwd".as_bytes()).unwrap().as_c_str().as_ptr
+        //);
+        //// i40e_recv_pkts, virtio_recv_pkts
+        //// i40e_xmit_pkts, virtio_xmit_pkts, virtio_recv_mergeable_pkts
+        //set_dpdk_hooks(
+        //    DPDK_HOOK_TYPE_RECV as c_int,
+        //    CString::new("rte_eth_rx_burst,virtio_recv_mergeable_pkts".
+        //);
+        //set_dpdk_hooks(
+        //    DPDK_HOOK_TYPE_XMIT as c_int,
+        //    CString::new("virtio_xmit_pkts".as_bytes()).unwrap().as_c_s
+        //);
+        //dpdk_trace_start();
+
         // test data limit max
         set_data_limit_max(10000);
 

agent/src/ebpf/user/common.c

@@ -1202,7 +1202,7 @@ int exec_command(const char *cmd, const char *args,
 		/* Read and print the output */
 		char buffer[1024];
 		int write_bytes =
-		    snprintf(ret_buf, ret_buf_size, "\n%s\n", cmd_buf);
+		    snprintf(ret_buf, ret_buf_size, "[ %s ]", cmd_buf);
 		while (fgets(buffer, sizeof(buffer), fp) != NULL) {
 			write_bytes +=
 			    snprintf(ret_buf + write_bytes,

agent/src/ebpf/user/config.h

@@ -18,7 +18,7 @@
 #define DF_EBPF_CONFIG_H
 
 #define EV_NAME_SIZE			1024
-
+#define MAX_EVENTS_BURST           	32 // The number of events in batch processing
 #define BOOT_TIME_UPDATE_PERIOD		60	// 系统启动时间更新周期, 单位：秒
 
 // eBPF Map Name

agent/src/ebpf/user/ctrl_tracer.c

@@ -140,6 +140,8 @@ static void datadump_help(void)
 	fprintf(stderr, "    106: PROTO_ZMTP\n");
 	fprintf(stderr, "    120: PROTO_DNS\n");
 	fprintf(stderr, "    121: PROTO_TLS\n");
+	fprintf(stderr, "    127: PROTO_CUSTOM\n");
+	fprintf(stderr, "    199: PROTO_DPDK_PKT\n");
 	fprintf(stderr, "PID:\n");
 	fprintf(stderr, "    0:   all process/thread\n");
 	fprintf(stderr, "COMM:\n");

agent/src/ebpf/user/extended/extended.c

@@ -25,6 +25,7 @@
 #include "../vec.h"
 #include "../tracer.h"
 #include "../socket.h"
+#include "../proc.h"
 
 int __attribute__ ((weak)) extended_reader_create(struct bpf_tracer *tracer)
 {
@@ -36,26 +37,51 @@ int __attribute__ ((weak)) extended_maps_set(struct bpf_tracer *tracer)
 	return 0;
 }
 
-void __attribute__ ((weak)) extended_prog_jump_tables(struct bpf_tracer *tracer) {
+void __attribute__ ((weak)) extended_prog_jump_tables(struct bpf_tracer *tracer)
+{
 }
 
-int __attribute__ ((weak)) collect_extended_uprobe_syms_from_procfs(struct tracer_probes_conf *conf)
+int __attribute__ ((weak)) collect_extended_uprobe_syms_from_procfs(struct
+								    tracer_probes_conf
+								    *conf)
 {
 	return 0;
 }
 
-void __attribute__ ((weak)) extended_process_exec(int pid) {}
+void __attribute__ ((weak)) extended_process_exec(int pid)
+{
+}
 
-void __attribute__ ((weak)) extended_events_handle(void) {}
+void __attribute__ ((weak)) extended_events_handle(void)
+{
+}
 
-void __attribute__ ((weak)) extended_process_exit(int pid) {}
+void __attribute__ ((weak)) extended_process_exit(int pid)
+{
+}
 
-void __attribute__ ((weak)) extended_match_pid_handle(int feat, int pid, enum match_pids_act act) {}
+void __attribute__ ((weak)) extended_match_pid_handle(int feat, int pid,
+						      enum match_pids_act act)
+{
+}
 
 bool __attribute__ ((weak)) extended_require_dwarf(int pid, const char *path)
 {
 	return false;
 }
 
-void __attribute__ ((weak)) extended_map_preprocess(struct ebpf_map *map) {}
-void __attribute__ ((weak)) extended_print_cp_tracer_status(void) {}
+void __attribute__ ((weak)) extended_map_preprocess(struct ebpf_map *map)
+{
+}
+
+void __attribute__ ((weak)) extended_print_cp_tracer_status(void)
+{
+}
+
+int __attribute__ ((weak)) print_extra_pkt_info(bool datadump_enable,
+						const char *pkt_data, int len,
+						char *buf, int buf_len,
+						u8 direction)
+{
+	return 0;
+}

agent/src/ebpf/user/extended/extended.h

@@ -83,4 +83,18 @@ void extended_map_preprocess(struct ebpf_map *map);
  *        status output.
  */
 void extended_print_cp_tracer_status(void);
+
+/**
+ * @brief **print_extra_pkt_info()** Outputs detailed information of a packet
+ *
+ * @param datadump_enable Enables datadump; when enabled, information will be
+ * @param pkt_data Packet data.
+ * @param len Length of the packet.
+ * @param buf Buffer for outputting packet debug information.
+ * @param buf_len Length of the buffer for outputting packet debug informatio
+ * @param direction Data direction.
+ * @return Length of the output information.
+ */
+int print_extra_pkt_info(bool datadump_enable, const char *pkt_data, int len,
+			 char *buf, int buf_len, u8 direction);
 #endif /* DF_EXTENDED_H */