feat: Handle Oracle cross-payload binds

Cache Oracle request info across frames so custom SQL extraction can resolve bind values when SQL and binds arrive separately. Keep the local enterprise-utils stubs aligned with the parser changes.

Author: rvql

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

@@ -311,6 +311,13 @@ pub mod l7 {
                 }
             }
 
+            pub fn find_bind_values<'a>(
+                _: &'a [u8],
+                _: usize,
+            ) -> Option<impl Iterator<Item = ()> + 'a> {
+                Some(std::iter::empty())
+            }
+
             #[derive(Serialize, Clone, Copy, Debug, Default, PartialEq)]
             pub enum TnsPacketType {
                 #[default]

agent/src/flow_generator/protocol_logs/sql/oracle.rs

@@ -44,12 +44,13 @@ use enterprise_utils::l7::{
     custom_policy::{
         custom_field_policy::{
             enums::{Op, Source},
-            Store,
+            PolicySlice, Store,
         },
         enums::TrafficDirection,
     },
     sql::oracle::{
-        Body, CallId, DataFlags, DataId, OracleParseConfig, OracleParser, TnsPacketType,
+        find_bind_values, Body, CallId, DataFlags, DataId, OracleParseConfig, OracleParser,
+        TnsPacketType,
     },
 };
 use public::l7_protocol::{Field, FieldSetter, L7Log, L7LogAttribute, L7Protocol, LogMessageType};
@@ -294,6 +295,66 @@ impl From<&OracleInfo> for LogCache {
 pub struct OracleLog {
     perf_stats: Vec<L7PerfStats>,
     custom_field_store: Store,
+    cached_info: Option<CachedOracleInfo>,
+}
+
+struct CachedOracleInfo {
+    info: OracleInfo,
+    bind_count: usize,
+}
+
+fn sql_max_bind_placeholder(sql: &str) -> usize {
+    let bytes = sql.as_bytes();
+    let mut max_n: usize = 0;
+    let mut i = 0;
+    while i < bytes.len() {
+        if bytes[i] != b':' {
+            i += 1;
+            continue;
+        }
+
+        let mut start = i + 1;
+        if start < bytes.len() && (bytes[start] == b'x' || bytes[start] == b'X') {
+            start += 1;
+        }
+        let mut end = start;
+        while end < bytes.len() && bytes[end].is_ascii_digit() {
+            end += 1;
+        }
+        if end > start {
+            if let Ok(n) = sql[start..end].parse::<usize>() {
+                if n > max_n {
+                    max_n = n;
+                }
+            }
+            i = end;
+            continue;
+        }
+        i += 1;
+    }
+    max_n
+}
+
+impl OracleLog {
+    fn apply_custom_field_operations(
+        &mut self,
+        policies: PolicySlice,
+        info: &mut OracleInfo,
+        frame_payload: &[u8],
+    ) {
+        policies.apply(
+            &mut self.custom_field_store,
+            info,
+            TrafficDirection::REQUEST,
+            Source::Sql(&info.sql, Some(frame_payload)),
+        );
+        for op in self.custom_field_store.drain_with(policies, info) {
+            match &op.op {
+                Op::AddMetric(_, _) | Op::SaveHeader(_) | Op::SavePayload(_) => (),
+                _ => auto_merge_custom_field(op, info),
+            }
+        }
+    }
 }
 
 impl L7ProtocolParserInterface for OracleLog {
@@ -335,6 +396,7 @@ impl L7ProtocolParserInterface for OracleLog {
         let mut info = vec![];
         for frame in frames {
             let frame_payload = frame.payload;
+            let is_request = matches!(&frame.body, Body::Request(_));
             let mut log_info = match frame.body {
                 Body::Request(req) => OracleInfo {
                     msg_type: param.direction.into(),
@@ -367,18 +429,35 @@ impl L7ProtocolParserInterface for OracleLog {
             };
 
             if let Some(policies) = custom_policies {
-                if !log_info.sql.is_empty() {
-                    policies.apply(
-                        &mut self.custom_field_store,
-                        &log_info,
-                        TrafficDirection::REQUEST,
-                        Source::Sql(&log_info.sql, Some(frame_payload)),
-                    );
-                    for op in self.custom_field_store.drain_with(policies, &log_info) {
-                        match &op.op {
-                            Op::AddMetric(_, _) | Op::SaveHeader(_) | Op::SavePayload(_) => (),
-                            _ => auto_merge_custom_field(op, &mut log_info),
-                        }
+                // Oracle 请求里，SQL 文本和 bind value 可能分布在两个 request frame 中。
+                // 先缓存只携带 SQL 的请求，等待下一帧带上 bind value 后再重新执行自定义 SQL 提取。
+                //
+                // In Oracle requests, the SQL text and bind values may be split across two
+                // request frames. Cache the SQL-only request first, then re-run custom SQL
+                // extraction when the next request frame carries the bind values.
+                if !is_request {
+                    self.cached_info = None;
+                } else if !log_info.sql.is_empty() {
+                    let bind_count = sql_max_bind_placeholder(&log_info.sql);
+                    let can_apply_custom_fields =
+                        bind_count == 0 || find_bind_values(frame_payload, bind_count).is_some();
+                    if can_apply_custom_fields {
+                        self.apply_custom_field_operations(policies, &mut log_info, frame_payload);
+                        self.cached_info = None;
+                    } else {
+                        self.cached_info = Some(CachedOracleInfo {
+                            info: log_info,
+                            bind_count,
+                        });
+                        continue;
+                    }
+                } else if let Some(mut cached) = self.cached_info.take() {
+                    if find_bind_values(frame_payload, cached.bind_count).is_some() {
+                        cached.info.merge(&mut log_info);
+                        log_info = cached.info;
+                        self.apply_custom_field_operations(policies, &mut log_info, frame_payload);
+                    } else {
+                        self.cached_info = Some(cached);
                     }
                 }
             }