feat: Add pcap replay command

And refactored remote exec code

Author: rvql

agent/Cargo.lock

@@ -91,7 +91,9 @@ packet_segmentation_reassembly = { path = "plugins/packet_segmentation_reassembl
 packet_sequence_block = { path = "plugins/packet_sequence_block" }
 page_size = "0.4.2"
 parking_lot = "0.11"
+pcap = "2.4"
 pcap_assembler = { path = "plugins/pcap_assembler" }
+pcap-parser = { version = "0.17", optional = true }
 pcap-sys = "0.1.3"
 pnet = "^0.29"
 prost.workspace = true
@@ -125,7 +127,6 @@ zstd = "0.13.2"
 [target.'cfg(any(target_os = "linux", target_os = "android"))'.dependencies]
 cgroups-rs = "0.2.9"
 nix = "0.23"
-pcap = "0.9.1"
 # As of procfs 0.16.0, Process::fd().iter() still not giving correct results on kernel 2.6.32
 # ref: https://github.com/eminence/procfs/pull/241
 procfs = { git = "https://github.com/deepflowio/procfs/" }
@@ -145,7 +146,6 @@ schemars = "0.8"
 trace-utils = { path = "crates/trace-utils" }
 
 [target.'cfg(target_os = "windows")'.dependencies]
-pcap = "0.10.1"
 winapi = { version = "0.3.9", features = [
     "errhandlingapi",
     "libloaderapi",
@@ -164,8 +164,10 @@ windows = { version = "0.30", features = [
 
 [dev-dependencies]
 criterion = "0.3.5"
+env_logger = "0.11"
 lru = "0.9.0"
 tempfile = "3.2.0"
+tokio-stream = "0.1"
 uluru = "3.0.0"
 
 [build-dependencies]
@@ -177,7 +179,7 @@ walkdir = "2"
 
 [features]
 default = ["libtrace"]
-enterprise = ["dep:enterprise-utils"]
+enterprise = ["dep:enterprise-utils", "dep:pcap-parser"]
 extended_observability = ["libtrace"]
 dylib_pcap = []
 libtrace = []

agent/Cargo.toml

@@ -86,7 +86,7 @@ pub(super) fn bench(c: &mut Criterion) {
             let rrt_cache = Rc::new(RefCell::new(L7PerfCache::new(8)));
             let req_param = ParseParam::new(
                 &packets[0],
-                rrt_cache.clone(),
+                Some(rrt_cache.clone()),
                 Default::default(),
                 #[cfg(any(target_os = "linux", target_os = "android"))]
                 Default::default(),
@@ -95,7 +95,7 @@ pub(super) fn bench(c: &mut Criterion) {
             );
             let resp_param = ParseParam::new(
                 &packets[1],
-                rrt_cache.clone(),
+                Some(rrt_cache.clone()),
                 Default::default(),
                 #[cfg(any(target_os = "linux", target_os = "android"))]
                 Default::default(),

agent/benches/flow_generator/l7_log.rs

@@ -100,7 +100,7 @@ pub mod l7 {
                     _: public::l7_protocol::L7ProtocolEnum,
                     _: u16,
                 ) -> Option<PolicySlice> {
-                    unimplemented!()
+                    None
                 }
                 pub fn counters(
                     &self,
@@ -123,20 +123,17 @@ pub mod l7 {
                     _: super::enums::TrafficDirection,
                     _: enums::Source,
                 ) {
-                    unimplemented!()
                 }
             }
 
             #[derive(Default, Debug)]
             pub struct Store;
             impl Store {
                 pub fn is_empty(&self) -> bool {
-                    unimplemented!()
+                    true
                 }
 
-                pub fn clear(&mut self) {
-                    unimplemented!()
-                }
+                pub fn clear(&mut self) {}
 
                 pub fn into_iter_with<L: public::l7_protocol::L7Log>(
                     self,

agent/crates/enterprise-utils/src/lib.rs

@@ -24,6 +24,10 @@ pub mod remote_exec {
         SerializeError(#[from] serde_json::Error),
         #[error("transparent")]
         SyscallFailed(String),
+        #[error("batch length too small, need at least {0} bytes")]
+        BatchLengthTooSmall(usize),
+        #[error("pcap parse error failed: {0}")]
+        PcapParseFailed(String),
     }
 
     type Result<T> = std::result::Result<T, Error>;

agent/crates/public/src/rpc.rs

@@ -57,7 +57,7 @@ fn main() {
             &packets[0].get_l4_payload().unwrap(),
             &ParseParam::new(
                 &packets[0],
-                log_cache.clone(),
+                Some(log_cache.clone()),
                 Default::default(),
                 #[cfg(any(target_os = "linux", target_os = "android"))]
                 Default::default(),
@@ -69,7 +69,7 @@ fn main() {
             &packets[1].get_l4_payload().unwrap(),
             &ParseParam::new(
                 &packets[1],
-                log_cache.clone(),
+                Some(log_cache.clone()),
                 Default::default(),
                 #[cfg(any(target_os = "linux", target_os = "android"))]
                 Default::default(),

agent/examples/parse_http_v1.rs

@@ -202,7 +202,10 @@ where
         is_reversed: bool,
     ) -> Option<String> {
         let key = LogCacheKey::new(param, self.session_id(), is_reversed);
-        match param.l7_perf_cache.borrow_mut().rrt_cache.get(&key) {
+        let Some(perf_cache) = param.l7_perf_cache.as_ref() else {
+            return None;
+        };
+        match perf_cache.borrow_mut().rrt_cache.get(&key) {
             Some(cached) if cached.endpoint.is_some() => {
                 let log = LogCache {
                     time: param.time,
@@ -234,6 +237,9 @@ where
             error!("flow_id: {}, packet time 0", param.flow_id);
             return None;
         }
+        let Some(perf_cache) = param.l7_perf_cache.as_ref() else {
+            return None;
+        };
 
         let cur_info = LogCache {
             time: param.time,
@@ -261,12 +267,10 @@ where
             return Some(stats);
         }
 
-        let (mut rtt_cache, mut timeout_cache) = RefMut::map_split(
-            param.l7_perf_cache.borrow_mut(),
-            |perf_cache: &mut L7PerfCache| {
+        let (mut rtt_cache, mut timeout_cache) =
+            RefMut::map_split(perf_cache.borrow_mut(), |perf_cache: &mut L7PerfCache| {
                 (&mut perf_cache.rrt_cache, &mut perf_cache.timeout_cache)
-            },
-        );
+            });
         let key = LogCacheKey::new(param, self.session_id(), self.is_reversed());
         let prev_info = rtt_cache.get_mut(&key);
         let timeout_counter = timeout_cache.get_or_insert_mut(param.flow_id);