feat(agent): add configurable socket syscall hook list (#11662) (#11702) (#11705)

* feat(agent): add configurable socket syscall hook list (#11662)

* feat(agent): add configurable socket syscall hook list

Add inputs.ebpf.socket.tunning.hooked_socket_syscalls so users can
select which supported socket syscalls install eBPF hooks, while
preserving the existing backend selection logic for mixed, pure-kprobe,
and kfunc modes.

* Fix hooked syscall config handling

* Modify the format of HOOKED_SOCKET_SYSCALLS
Conflicts:
	server/agent_config/README-CH.md
	server/agent_config/README.md

* Add tests for syscall hook configuration

Author: yinjiping

agent/src/config/config.rs

@@ -57,6 +57,35 @@ use enterprise_utils::l7::custom_policy::config::{CustomFieldPolicy, CustomProto
 pub const K8S_CA_CRT_PATH: &str = "/run/secrets/kubernetes.io/serviceaccount/ca.crt";
 const MINUTE: Duration = Duration::from_secs(60);
 const DEFAULT_STANDALONE_CONFIG: &str = "/etc/deepflow-agent-standalone.yaml";
+#[rustfmt::skip]
+const HOOKED_SOCKET_SYSCALLS: [&str; 10] = [
+    "read",
+    "readv",
+    "recvfrom",
+    "recvmsg",
+    "recvmmsg",
+    "sendmsg",
+    "sendmmsg",
+    "sendto",
+    "write",
+    "writev",
+];
+
+fn normalize_hooked_socket_syscalls<T: AsRef<str>>(syscalls: &[T]) -> Vec<String> {
+    HOOKED_SOCKET_SYSCALLS
+        .iter()
+        .filter(|supported| {
+            syscalls
+                .iter()
+                .any(|syscall| syscall.as_ref() == **supported)
+        })
+        .map(|syscall| syscall.to_string())
+        .collect()
+}
+
+fn default_hooked_socket_syscalls() -> Vec<String> {
+    normalize_hooked_socket_syscalls(&HOOKED_SOCKET_SYSCALLS)
+}
 
 #[derive(Debug, Error)]
 pub enum ConfigError {
@@ -980,13 +1009,30 @@ pub struct EbpfSocketKprobe {
     pub whitelist: EbpfSocketKprobePorts,
 }
 
-#[derive(Clone, Default, Debug, Deserialize, PartialEq, Eq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
 #[serde(default)]
 pub struct EbpfSocketTunning {
     pub max_capture_rate: u64,
     pub syscall_trace_id_disabled: bool,
     pub map_prealloc_disabled: bool,
     pub fentry_enabled: bool,
+    #[serde(
+        default = "default_hooked_socket_syscalls",
+        deserialize_with = "deser_hooked_socket_syscalls"
+    )]
+    pub hooked_socket_syscalls: Vec<String>,
+}
+
+impl Default for EbpfSocketTunning {
+    fn default() -> Self {
+        Self {
+            max_capture_rate: 0,
+            syscall_trace_id_disabled: false,
+            map_prealloc_disabled: false,
+            fentry_enabled: false,
+            hooked_socket_syscalls: default_hooked_socket_syscalls(),
+        }
+    }
 }
 
 #[derive(Clone, Default, Debug, Deserialize, PartialEq, Eq)]
@@ -3612,6 +3658,22 @@ where
     Ok(v)
 }
 
+fn deser_hooked_socket_syscalls<'de, D>(deserializer: D) -> Result<Vec<String>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let raw = Vec::<String>::deserialize(deserializer)?;
+    for syscall in &raw {
+        if !HOOKED_SOCKET_SYSCALLS.contains(&syscall.as_str()) {
+            return Err(de::Error::invalid_value(
+                Unexpected::Str(syscall),
+                &"one of read|readv|recvfrom|recvmsg|recvmmsg|sendmsg|sendmmsg|sendto|write|writev",
+            ));
+        }
+    }
+    Ok(normalize_hooked_socket_syscalls(&raw))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -3730,4 +3792,39 @@ processors:
         assert_eq!(apps[1].protocol, L7Protocol::Grpc);
         assert_eq!(apps[1].timeout, Duration::from_secs(130));
     }
+
+    #[test]
+    fn parse_hooked_socket_syscalls_normalizes_order_and_duplicates() {
+        let yaml = r#"
+inputs:
+  ebpf:
+    socket:
+      tunning:
+        hooked_socket_syscalls: [write, read, write, sendto]
+"#;
+        let cfg: UserConfig = serde_yaml::from_str(yaml).unwrap();
+
+        assert_eq!(
+            cfg.inputs.ebpf.socket.tunning.hooked_socket_syscalls,
+            vec!["read", "sendto", "write"]
+        );
+    }
+
+    #[test]
+    fn parse_hooked_socket_syscalls_rejects_invalid_values() {
+        let yaml = r#"
+inputs:
+  ebpf:
+    socket:
+      tunning:
+        hooked_socket_syscalls: [read, invalid_syscall]
+"#;
+        let result: Result<UserConfig, _> = serde_yaml::from_str(yaml);
+
+        assert!(result.is_err());
+        let error = result.unwrap_err().to_string();
+        assert!(error.contains("invalid value: string \"invalid_syscall\""));
+        assert!(error
+            .contains("read|readv|recvfrom|recvmsg|recvmmsg|sendmsg|sendmmsg|sendto|write|writev"));
+    }
 }

agent/src/config/handler.rs

@@ -3610,6 +3610,11 @@ impl ConfigHandler {
                     new_tunning.fentry_enabled,
                     "inputs.ebpf.socket.tunning.fentry_enabled"
                 ),
+                (
+                    tunning.hooked_socket_syscalls,
+                    new_tunning.hooked_socket_syscalls,
+                    "inputs.ebpf.socket.tunning.hooked_socket_syscalls"
+                ),
                 (
                     tunning.map_prealloc_disabled,
                     new_tunning.map_prealloc_disabled,

agent/src/ebpf/mod.rs

@@ -806,6 +806,7 @@ extern "C" {
     pub fn enable_unix_socket_feature();
     pub fn disable_fentry();
     pub fn enable_fentry();
+    pub fn set_hooked_socket_syscalls(bitmap: c_ulonglong);
     pub fn set_virtual_file_collect(enabled: bool) -> c_int;
     cfg_if::cfg_if! {
         if #[cfg(feature = "extended_observability")] {

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

@@ -17,6 +17,7 @@
 use chrono::prelude::DateTime;
 use chrono::FixedOffset;
 use chrono::Utc;
+use log::info;
 use socket_tracer::ebpf::*;
 use std::convert::TryInto;
 use std::env;
@@ -26,7 +27,6 @@ use std::net::IpAddr;
 use std::sync::Mutex;
 use std::thread;
 use std::time::{Duration, UNIX_EPOCH};
-use log::info;
 
 extern "C" {
     fn print_uprobe_http2_info(data: *mut c_char, len: c_uint);
@@ -42,6 +42,88 @@ lazy_static::lazy_static! {
     static ref COUNTER: Mutex<u32> = Mutex::new(0);
 }
 
+#[derive(Clone, Copy, Default, PartialEq, Eq, Debug)]
+struct HookedSocketSyscallBitmap(c_ulonglong);
+
+impl HookedSocketSyscallBitmap {
+    fn set_enabled(&mut self, bit: c_ulonglong) {
+        self.0 |= bit;
+    }
+}
+
+impl<T: AsRef<str>> From<&[T]> for HookedSocketSyscallBitmap {
+    fn from(vs: &[T]) -> Self {
+        let mut bitmap = HookedSocketSyscallBitmap(0);
+        for v in vs.iter() {
+            match v.as_ref() {
+                "read" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_READ),
+                "readv" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_READV),
+                "recvfrom" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_RECVFROM),
+                "recvmsg" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_RECVMSG),
+                "recvmmsg" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_RECVMMSG),
+                "sendmsg" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_SENDMSG),
+                "sendmmsg" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_SENDMMSG),
+                "sendto" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_SENDTO),
+                "write" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_WRITE),
+                "writev" => bitmap.set_enabled(HOOKED_SOCKET_SYSCALL_WRITEV),
+                _ => {}
+            }
+        }
+        bitmap
+    }
+}
+
+const HOOKED_SOCKET_SYSCALL_READ: c_ulonglong = 1 << 0;
+const HOOKED_SOCKET_SYSCALL_READV: c_ulonglong = 1 << 1;
+const HOOKED_SOCKET_SYSCALL_RECVFROM: c_ulonglong = 1 << 2;
+const HOOKED_SOCKET_SYSCALL_RECVMSG: c_ulonglong = 1 << 3;
+const HOOKED_SOCKET_SYSCALL_RECVMMSG: c_ulonglong = 1 << 4;
+const HOOKED_SOCKET_SYSCALL_SENDMSG: c_ulonglong = 1 << 5;
+const HOOKED_SOCKET_SYSCALL_SENDMMSG: c_ulonglong = 1 << 6;
+const HOOKED_SOCKET_SYSCALL_SENDTO: c_ulonglong = 1 << 7;
+const HOOKED_SOCKET_SYSCALL_WRITE: c_ulonglong = 1 << 8;
+const HOOKED_SOCKET_SYSCALL_WRITEV: c_ulonglong = 1 << 9;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn hooked_socket_syscall_bitmap_sets_expected_bits() {
+        let syscalls = ["read", "recvfrom", "sendto", "write"];
+        let bitmap = HookedSocketSyscallBitmap::from(syscalls.as_slice());
+
+        assert_eq!(
+            bitmap,
+            HookedSocketSyscallBitmap(
+                HOOKED_SOCKET_SYSCALL_READ
+                    | HOOKED_SOCKET_SYSCALL_RECVFROM
+                    | HOOKED_SOCKET_SYSCALL_SENDTO
+                    | HOOKED_SOCKET_SYSCALL_WRITE,
+            ),
+        );
+    }
+
+    #[test]
+    fn set_hooked_socket_syscalls_accepts_bitmap_from_syscall_names() {
+        let syscalls = ["write", "read", "write", "sendto"];
+        let bitmap = HookedSocketSyscallBitmap::from(syscalls.as_slice());
+
+        assert_eq!(
+            bitmap,
+            HookedSocketSyscallBitmap(
+                HOOKED_SOCKET_SYSCALL_READ
+                    | HOOKED_SOCKET_SYSCALL_SENDTO
+                    | HOOKED_SOCKET_SYSCALL_WRITE,
+            ),
+        );
+
+        unsafe {
+            set_hooked_socket_syscalls(bitmap.0);
+        }
+    }
+}
+
 #[allow(dead_code)]
 fn increment_counter(num: u32, counter_type: u32) {
     if counter_type == 0 {
@@ -201,7 +283,11 @@ extern "C" fn debug_callback(_data: *mut c_char, len: c_int) {
     }
 }
 
-extern "C" fn socket_trace_callback(_: *mut c_void, queue_id: c_int, sd: *mut SK_BPF_DATA) -> c_int {
+extern "C" fn socket_trace_callback(
+    _: *mut c_void,
+    queue_id: c_int,
+    sd: *mut SK_BPF_DATA,
+) -> c_int {
     unsafe {
         let mut proto_tag = String::from("");
         if sk_proto_safe(sd) == SOCK_DATA_OTHER {
@@ -411,7 +497,10 @@ fn main() {
             if e.kind() == std::io::ErrorKind::NotFound {
                 println!("numad process not found, skipping CPU affinity protection (normal)");
             } else {
-                println!("Failed to protect CPU affinity due to unexpected error: {}", e);
+                println!(
+                    "Failed to protect CPU affinity due to unexpected error: {}",
+                    e
+                );
             }
         }
     }
@@ -640,10 +729,20 @@ fn main() {
                 .as_c_str()
                 .as_ptr(),
         );
-	// dpdk enable
-	// set_dpdk_trace_enabled(true);
-	// disable_kprobe_feature();
-	// set_virtual_file_collect(true);
+        //let hooked_socket_syscalls = [
+        //    "read", "readv", "recvfrom", "recvmsg", "recvmmsg", "sendmsg", "sendmmsg", "sendto",
+        //    "write", "writev",
+        //];
+        let hooked_socket_syscalls = [
+            "readv", "recvfrom", "recvmsg", "recvmmsg", "sendmsg", "sendmmsg", "sendto", "writev",
+        ];
+        set_hooked_socket_syscalls(
+            HookedSocketSyscallBitmap::from(hooked_socket_syscalls.as_slice()).0,
+        );
+        // dpdk enable
+        // set_dpdk_trace_enabled(true);
+        // disable_kprobe_feature();
+        // set_virtual_file_collect(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 */