fix: reduce kafka app misidentification

deepflowio · Sep 24, 2024 · fde1bc5 · fde1bc5
1 parent 0f30122
commit fde1bc5
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/agent/src/flow_generator/protocol_logs/mq/kafka.rs b/agent/src/flow_generator/protocol_logs/mq/kafka.rs
@@ -440,6 +440,7 @@ impl KafkaLog {
     const MSG_LEN_SIZE: usize = 4;
     const MAX_TRACE_ID: usize = 255;
     const MAX_SESSION_PER_FLOW: usize = 32;
+    const MAX_VERSION: u16 = 12;
 
     fn decode_varint(buf: &[u8]) -> (usize, usize) {
         let mut shift = 0;
@@ -1496,6 +1497,9 @@ impl KafkaLog {
         info.msg_type = LogMessageType::Request;
         info.api_key = read_u16_be(&payload[4..]);
         info.api_version = read_u16_be(&payload[6..]);
+        if info.api_version > Self::MAX_VERSION {
+            return Err(Error::KafkaLogParseFailed);
+        }
         info.correlation_id = read_u32_be(&payload[8..]);
         info.client_id = String::from_utf8_lossy(&payload[14..14 + client_id_len]).into_owned();
         if !info.client_id.is_ascii() {

diff --git a/agent/src/rpc/synchronizer.rs b/agent/src/rpc/synchronizer.rs
@@ -1256,7 +1256,7 @@ impl Synchronizer {
                         return;
                     }
                     ntp_diff.store(0, Ordering::Relaxed);
-                    time::sleep(sync_interval).await;
+                    time::sleep(Duration::from_secs(1)).await;
                     continue;
                 }