feat: eBPF Add the relevant XDP CPU balancer related code

.gitignore

@@ -32,13 +32,15 @@ target/
 /agent/src/ebpf/kernel/*.ll
 /agent/src/ebpf/kernel/*.objdump
 /agent/src/ebpf/kernel/*.elf
+/agent/src/ebpf/kernel/build/
 /agent/src/ebpf/user/socket_trace_bpf_5_2.c
 /agent/src/ebpf/user/socket_trace_bpf_core.c
 /agent/src/ebpf/user/socket_trace_bpf_common.c
 /agent/src/ebpf/user/socket_trace_bpf_5_2_plus.c
 /agent/src/ebpf/user/socket_trace_bpf_kylin.c
 /agent/src/ebpf/user/perf_profiler_bpf_common.c
 /agent/src/ebpf/user/af_packet_fanout_bpf_common.c
+/agent/src/ebpf/user/extended/*_bpf_common.c
 /agent/src/ebpf/user/deepflow_ebpfctl_bin.c
 /agent/src/ebpf/user/profile/java_agent_so_gnu.c
 /agent/src/ebpf/user/profile/java_agent_so_musl.c

agent/src/config/config.rs

@@ -23,8 +23,15 @@ use std::net::{IpAddr, ToSocketAddrs};
 use std::path::Path;
 use std::time::Duration;
 
+#[cfg(feature = "extended_observability")]
+use std::ffi::CString;
+
 #[cfg(any(target_os = "linux", target_os = "android"))]
 use envmnt::{ExpandOptions, ExpansionType};
+#[cfg(feature = "extended_observability")]
+use libc::c_int;
+#[cfg(feature = "extended_observability")]
+use log::warn;
 use log::{debug, error, info};
 use md5::{Digest, Md5};
 #[cfg(any(target_os = "linux", target_os = "android"))]
@@ -39,6 +46,9 @@ use tokio::runtime::Runtime;
 
 use crate::common::l7_protocol_log::{L7ProtocolBitmap, L7ProtocolParser};
 use crate::dispatcher::recv_engine::DEFAULT_BLOCK_SIZE;
+#[cfg(any(target_os = "linux", target_os = "android"))]
+#[cfg(feature = "extended_observability")]
+use crate::ebpf;
 use crate::flow_generator::{DnsLog, MemcachedLog};
 #[cfg(any(target_os = "linux", target_os = "android"))]
 use crate::platform::{OsAppTag, ProcessData};
@@ -1235,6 +1245,97 @@ impl Default for EbpfTunning {
     }
 }
 
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct NicOptimizeConfig {
+    pub interface: String,
+    pub rx_ring_size: u64,
+    pub rss_channel_count: u64,
+    pub irq_cpu_list: String,
+    pub xdp_cpu_redirect: bool,
+    pub xdp_queue_size: u64,
+    pub xdp_cpu_redirect_list: String,
+}
+
+const XDP_QUEUE_SIZE_MIN: u64 = 512;
+const XDP_QUEUE_SIZE_MAX: u64 = 8192;
+
+impl Default for NicOptimizeConfig {
+    fn default() -> Self {
+        Self {
+            interface: "".to_string(),
+            rx_ring_size: 0,
+            rss_channel_count: 0,
+            irq_cpu_list: "".to_string(),
+            xdp_cpu_redirect: false,
+            xdp_queue_size: 2048,
+            xdp_cpu_redirect_list: "".to_string(),
+        }
+    }
+}
+
+impl NicOptimizeConfig {
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    #[cfg(feature = "extended_observability")]
+    pub fn apply(&self) {
+        let to_cstring = |field: &str, value: &str| {
+            CString::new(value)
+                .map_err(|_| {
+                    warn!(
+                        "Skip NIC optimization for interface {:?}: {} contains NUL byte",
+                        self.interface, field
+                    );
+                })
+                .ok()
+        };
+
+        let Some(nic_name) = to_cstring("interface", self.interface.as_str()) else {
+            return;
+        };
+        let Some(irq_cpu) = to_cstring("irq_cpu_list", self.irq_cpu_list.as_str()) else {
+            return;
+        };
+        let Some(xdp_cpu) =
+            to_cstring("xdp_cpu_redirect_list", self.xdp_cpu_redirect_list.as_str())
+        else {
+            return;
+        };
+
+        let ret = unsafe {
+            ebpf::nic_optimize_config(
+                nic_name.as_ptr(),
+                self.rx_ring_size as c_int,
+                self.rss_channel_count as c_int,
+                irq_cpu.as_ptr(),
+                self.xdp_cpu_redirect,
+                self.xdp_queue_size as c_int,
+                xdp_cpu.as_ptr(),
+            )
+        };
+
+        if ret != 0 {
+            warn!(
+                "Failed to configure NIC optimization for interface '{}' \
+                (ret={}, rx_ring_size={}, rss_channel_count={}, \
+                xdp_cpu_redirect={}, xdp_queue_size={})",
+                self.interface,
+                ret,
+                self.rx_ring_size,
+                self.rss_channel_count,
+                self.xdp_cpu_redirect,
+                self.xdp_queue_size,
+            );
+        }
+    }
+}
+
+#[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct EbpfNetwork {
+    pub nic_opt_enabled: bool,
+    pub nic_optimize: Vec<NicOptimizeConfig>,
+}
+
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
 #[serde(default)]
 pub struct EbpfSocketPreprocess {
@@ -1285,6 +1386,7 @@ pub struct Ebpf {
     pub file: EbpfFile,
     pub profile: EbpfProfile,
     pub tunning: EbpfTunning,
+    pub network: EbpfNetwork,
     #[serde(skip)]
     pub java_symbol_file_refresh_defer_interval: i32,
 }
@@ -1297,6 +1399,7 @@ impl Default for Ebpf {
             file: EbpfFile::default(),
             profile: EbpfProfile::default(),
             tunning: EbpfTunning::default(),
+            network: EbpfNetwork::default(),
             java_symbol_file_refresh_defer_interval: 60,
         }
     }
@@ -3219,6 +3322,15 @@ impl UserConfig {
             )));
         }
 
+        for nic in &self.inputs.ebpf.network.nic_optimize {
+            if !(XDP_QUEUE_SIZE_MIN..=XDP_QUEUE_SIZE_MAX).contains(&nic.xdp_queue_size) {
+                return Err(ConfigError::RuntimeConfigInvalid(format!(
+                    "xdp_queue_size {} for interface {} not in [{}, {}]",
+                    nic.xdp_queue_size, nic.interface, XDP_QUEUE_SIZE_MIN, XDP_QUEUE_SIZE_MAX,
+                )));
+            }
+        }
+
         Ok(())
     }
 

agent/src/config/handler.rs

@@ -3553,6 +3553,26 @@ impl ConfigHandler {
             restart_agent = !first_run;
         }
 
+        // The hot-update configuration items in ebpf.network are handled within the set_ebpf() callback.
+        let network = &mut ebpf.network;
+        let new_network = &mut new_ebpf.network;
+
+        if network.nic_opt_enabled != new_network.nic_opt_enabled {
+            info!(
+                "Update inputs.ebpf.network.nic_opt_enabled from {:?} to {:?}.",
+                network.nic_opt_enabled, new_network.nic_opt_enabled
+            );
+            network.nic_opt_enabled = new_network.nic_opt_enabled;
+        }
+
+        if network.nic_optimize != new_network.nic_optimize {
+            info!(
+                "Update inputs.ebpf.network.nic_optimize from {:?} to {:?}.",
+                network.nic_optimize, new_network.nic_optimize
+            );
+            network.nic_optimize = new_network.nic_optimize.clone();
+        }
+
         let kprobe = &mut ebpf.socket.kprobe;
         let new_kprobe = &mut new_ebpf.socket.kprobe;
         if kprobe.disabled != new_kprobe.disabled {

agent/src/ebpf/Makefile

@@ -158,7 +158,7 @@ PERF_PROFILER_ELFS := user/perf_profiler_bpf_common.c \
 	user/perf_profiler_bpf_5_2_plus.c \
 	user/perf_profiler_bpf_5_15_plus.c
 
-ELFFILES := $(SOCKET_TRACE_ELFS) $(PERF_PROFILER_ELFS) $(AF_PACKET_FANOUT_ELFS) $(TCP_OPTION_TRACING_ELFS)
+ELFFILES := $(SOCKET_TRACE_ELFS) $(PERF_PROFILER_ELFS) $(EXTEND_ELFS)
 
 tools/bintobuffer:
 	$(call msg,TOOLS,tools/bintobuffer)

agent/src/ebpf/kernel/Makefile

@@ -60,7 +60,7 @@ FINAL_TARGET = -emit-llvm -D__TARGET_ARCH_$(ARCH) -o ${@:.elf=.ll} -c $^ && $(LL
 
 all: $(TAEGET_KERN_ELF)
 
-EBPF_CLAGS ?= -I. -Ivmlinux -Iinclude -I../../../crates/trace-utils/src \
+EBPF_CLAGS ?= -I. -Ivmlinux -Iinclude -I../user/extended/bpf/include -I../../../crates/trace-utils/src \
         -D__BPF_TRACING__ -D GROUP_LEADER_OFFSET_OVERRIDE=0 \
         -DSTART_BOOTTIME_OFFSET_OVERRIDE=0      \
         -DSTART_BOOTTIME_VARNAME=real_start_time

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

@@ -30,7 +30,6 @@
 #include <errno.h>
 #include <stddef.h>
 #include <bcc/compat/linux/bpf.h>
-#include "utils.h"
 
 struct task_struct;
 
@@ -150,6 +149,13 @@ static int
     __attribute__ ((__unused__)) (*bpf_get_stack) (void *ctx, void *buf, __u32 size,
 						     int flags) = (void *)67;
 
+
+// Linux 4.14: Added support for BPF_MAP_TYPE_CPUMAP, allowing packets to be redirected to specific CPUs.
+static long
+    __attribute__ ((__unused__)) (*bpf_redirect_map)(void *map, __u32 key, __u64 flags) =
+							(void *)51;
+
+
 /* llvm builtin functions that eBPF C program may use to 
  * emit BPF_LD_ABS and BPF_LD_IND instructions 
  */
@@ -334,6 +340,7 @@ _Pragma("GCC error \"PT_GO_REGS_PARM\"");
 
 #define TP_SYSCALL_PROG(F) SEC("tracepoint/syscalls/sys_"__stringify(F)) int df_T_##F
 #define TP_SCHED_PROG(F) SEC("tracepoint/sched/sched_"__stringify(F)) int df_T_##F
+#define TP_XDP_PROG(F) SEC("tracepoint/xdp/xdp_"__stringify(F)) int df_T_##F
 #define PROGTP(F) SEC("prog/tp/"__stringify(F)) int df_TP_##F
 #define PROGKP(F) SEC("prog/kp/"__stringify(F)) int df_KP_##F
 #define PROGPE(F) SEC("prog/pe/"__stringify(F)) int df_PE_##F

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

@@ -19,6 +19,7 @@
  * SPDX-License-Identifier: GPL-2.0
  */
 
+#include <arpa/inet.h>
 #include <linux/bpf_perf_event.h>
 #include "config.h"
 #include "bpf_base.h"

agent/src/ebpf/mod.rs

@@ -941,6 +941,57 @@ extern "C" {
                 */
                 pub fn set_tcp_option_tracing_enabled(enabled: bool) -> c_int;
                 pub fn set_tcp_option_tracing_sample_window(bytes: c_uint) -> c_int;
+
+                /**
+                 * @brief Enable or disable NIC optimization.
+                 *
+                 * This function sets the NIC optimization feature on or off. When enabled,
+                 * the system may adjust RX ring sizes, RSS channels, and CPU affinities
+                 * to improve packet processing performance.
+                 *
+                 * @param enabled Set to 1 to enable, or 0 to disable. On the Rust side,
+                 *                this can be converted from a bool.
+                 *
+                 * @return 0 on success, non-zero on failure.
+                 *
+                 * @note Corresponding C function: `int set_nic_optimization(bool enabled)`
+                 */
+                pub fn set_nic_optimization(enabled: bool) -> c_int;
+
+                /**
+                 * @brief Configure optimization parameters for a single NIC.
+                 *
+                 * This function applies NIC-specific optimization settings, including
+                 * RX ring size, RSS channel count, IRQ CPU binding, and XDP CPU redirection.
+                 * It allows fine-grained performance tuning for each network interface.
+                 *
+                 * @param nic_name The name of the network interface (C string).
+                 * @param rx_ring_size The size of the RX ring buffer.
+                 * @param rss_channel_count Number of RSS channels to use.
+                 * @param irq_cpu_list Comma-separated list of CPU IDs for IRQ binding.
+                 * @param xdp_cpu_redirect Set to 1 to enable XDP CPU redirect, 0 to disable.
+                 * @param xdp_queue_size The size of the XDP queue.
+                 * @param xdp_cpu_list Comma-separated list of CPU IDs for XDP processing.
+                 *
+                 * @return 0 on success, non-zero on failure.
+                 *
+                 * @note Corresponding C function:
+                 * ```c
+                 * int nic_optimize_config(const char* nic_name, int rx_ring_size,
+                 *                         int rss_channel_count, const char* irq_cpu_list,
+                 *                         bool xdp_cpu_redirect, int xdp_queue_size,
+                 *                         const char* xdp_cpu_list);
+                 * ```
+                 */
+                pub fn nic_optimize_config(
+                    nic_name: *const c_char,
+                    rx_ring_size: c_int,
+                    rss_channel_count: c_int,
+                    irq_cpu_list: *const c_char,
+                    xdp_cpu_redirect: bool,
+                    xdp_queue_size: c_int,
+                    xdp_cpu_list: *const c_char,
+                ) -> c_int;
         }
     }
 }

agent/src/ebpf/samples/rust/socket-tracer/Cargo.toml

@@ -21,3 +21,5 @@ lazy_static = "1.4.0"
 log = "0.4"
 env_logger = "0.9"
 trace-utils = { path = "../../../../../crates/trace-utils" }
+# Enable this option when building the Enterprise edition.
+# trace-utils-interp = { path = "../../../../../crates/trace-utils-interp" }

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

@@ -28,6 +28,21 @@ use std::thread;
 use std::time::{Duration, UNIX_EPOCH};
 use log::info;
 
+// Reference trace-utils-interp when building the Enterprise edition.
+// The purpose is to ensure that Rust links against libtrace_utils_interp-xxxx.rlib
+// during the linking stage.
+//
+// interpreter_tracer.c calls is_php_process(),
+// and this function is defined in the trace-utils-interp crate.
+//
+// However, the Rust code in socket_tracer does not directly reference
+// trace-utils-interp. As a result, Cargo considers this dependency unused
+// and excludes it from the linking stage.
+//
+// Therefore, we explicitly reference trace-utils-interp here to force
+// Cargo to include it in the final link.
+//use trace_utils_interp as _;
+
 extern "C" {
     fn print_uprobe_http2_info(data: *mut c_char, len: c_uint);
     fn print_uprobe_grpc_dataframe(data: *mut c_char, len: c_uint);
@@ -746,6 +761,31 @@ fn main() {
             std::thread::sleep(Duration::from_secs(1));
         }
 
+        // ------ Nic Optimization ----
+        //if set_nic_optimization(true) != 0 {
+        //    println!("set_nic_optimization() error.");
+        //    ::std::process::exit(1);
+        //}
+
+        //let nic_name = CString::new("p2p2").unwrap();
+        //let irq_cpu_list = CString::new("").unwrap();
+        //let xdp_cpu_list = CString::new("").unwrap();
+
+        //if nic_optimize_config(
+        //    nic_name.as_c_str().as_ptr(),
+        //    0, // rx_ring_size
+        //    0, // rss_channel_count
+        //    irq_cpu_list.as_c_str().as_ptr(),
+        //    true, // xdp_cpu_redirect
+        //    0,    // xdp_queue_size
+        //    xdp_cpu_list.as_c_str().as_ptr(),
+        //) != 0
+        //{
+        //    println!("nic_optimize_config() error.");
+        //    ::std::process::exit(1);
+        //}
+        // ------ Nic Optimization end ----
+
         //thread::sleep(Duration::from_secs(60));
         //stop_continuous_profiler();
         //print!(

agent/src/ebpf/user/config.h

@@ -180,6 +180,7 @@ enum cfg_feature_idx {
 	FEATURE_PROFILE_PYTHON,
 	FEATURE_PROFILE_PHP,
 	FEATURE_PROFILE_V8,
+	FEATURE_CPU_BALANCER,
 	FEATURE_MAX,
 };
 
@@ -194,6 +195,7 @@ enum cfg_feature_idx {
 #define FEATURE_FLAG_PROFILE_PYTHON		(1 << FEATURE_PROFILE_PYTHON)
 #define FEATURE_FLAG_PROFILE_PHP		(1 << FEATURE_PROFILE_PHP)
 #define FEATURE_FLAG_PROFILE_V8			(1 << FEATURE_PROFILE_V8)
+#define FEATURE_FLAG_CPU_BALANCER		(1 << FEATURE_CPU_BALANCER)
 
 #define FEATURE_FLAG_PROFILE				(FEATURE_FLAG_PROFILE_ONCPU | FEATURE_FLAG_PROFILE_OFFCPU | FEATURE_FLAG_PROFILE_MEMORY)
 

agent/src/ebpf/user/load.c

@@ -630,6 +630,8 @@ static enum bpf_prog_type get_prog_type(struct sec_desc *desc)
 		prog_type = BPF_PROG_TYPE_SOCKET_FILTER;
 	} else if (!strncmp(desc->name, "sockops", 7)) {
 		prog_type = BPF_PROG_TYPE_SOCK_OPS;
+	} else if (!memcmp(desc->name, "xdp", 3)) {
+		prog_type = BPF_PROG_TYPE_XDP;
 	} else {
 		prog_type = BPF_PROG_TYPE_UNSPEC;
 	}