feat: support reverse l7 metrics

Author: yuanchaoa

agent/src/collector/collector.rs

@@ -520,6 +520,7 @@ impl Stash {
                 0,
                 0,
                 L7Protocol::Unknown,
+                false,
                 self.context.agent_mode,
             );
             self.fill_edge_l4_stats(tagger, acc_flow.flow_meter);
@@ -554,6 +555,7 @@ impl Stash {
                     0,
                     0,
                     L7Protocol::Unknown,
+                    false,
                     self.context.agent_mode,
                 );
                 self.fill_single_l4_stats(tagger, flow_meter);
@@ -569,6 +571,7 @@ impl Stash {
                 0,
                 0,
                 L7Protocol::Unknown,
+                false,
                 self.context.agent_mode,
             );
             // edge_stats: If the direction of a certain end is known, the statistical data
@@ -712,6 +715,7 @@ impl Stash {
                 meter.biz_type,
                 meter.time_span,
                 meter.l7_protocol,
+                meter.is_reversed,
                 self.context.agent_mode,
             );
             tagger.code |= Code::L7_PROTOCOL;
@@ -742,6 +746,7 @@ impl Stash {
                     meter.biz_type,
                     meter.time_span,
                     meter.l7_protocol,
+                    meter.is_reversed,
                     self.context.agent_mode,
                 );
                 tagger.code |= Code::L7_PROTOCOL;
@@ -758,6 +763,7 @@ impl Stash {
                 meter.biz_type,
                 meter.time_span,
                 meter.l7_protocol,
+                meter.is_reversed,
                 self.context.agent_mode,
             );
             tagger.code |= Code::L7_PROTOCOL;
@@ -879,13 +885,14 @@ fn get_single_tagger(
     global_thread_id: u8,
     flow: &MiniFlow,
     ep: usize,
-    direction: Direction,
+    mut direction: Direction,
     is_active_host: bool,
     config: &CollectorConfig,
     endpoint: Option<String>,
     biz_type: u8,
     time_span: u32,
     l7_protocol: L7Protocol,
+    is_reversed: bool,
     agent_mode: RunningMode,
 ) -> Tagger {
     let flow_key = &flow.flow_key;
@@ -937,17 +944,35 @@ fn get_single_tagger(
         l3_epc_id: get_l3_epc_id(side.l3_epc_id, flow.signal_source),
         gpid: side.gpid,
         protocol: flow_key.proto,
-        direction,
-        tap_side: TapSide::from(direction),
+        direction: if is_reversed {
+            direction.reverse()
+        } else {
+            direction
+        },
+        tap_side: if is_reversed {
+            TapSide::from(direction).reverse()
+        } else {
+            TapSide::from(direction)
+        },
         tap_port: flow_key.tap_port,
         tap_type: flow_key.tap_type,
         // If the resource is located on the client, the service port is ignored
-        server_port: if ep == FLOW_METRICS_PEER_SRC
-            || ignore_server_port(flow, config.inactive_server_port_aggregation)
-        {
+        server_port: if ignore_server_port(flow, config.inactive_server_port_aggregation) {
             0
         } else {
-            flow.peers[1].nat_real_port
+            if ep == FLOW_METRICS_PEER_SRC {
+                if is_reversed {
+                    flow.peers[0].nat_real_port
+                } else {
+                    0
+                }
+            } else {
+                if is_reversed {
+                    0
+                } else {
+                    flow.peers[1].nat_real_port
+                }
+            }
         },
         is_ipv6,
         code: {
@@ -988,6 +1013,7 @@ fn get_edge_tagger(
     biz_type: u8,
     time_span: u32,
     l7_protocol: L7Protocol,
+    is_reversed: bool,
     agent_mode: RunningMode,
 ) -> Tagger {
     let flow_key = &flow.flow_key;
@@ -1040,7 +1066,7 @@ fn get_edge_tagger(
         (src_mac, dst_mac)
     };
 
-    Tagger {
+    let mut tagger = Tagger {
         global_thread_id,
         agent_id: config.agent_id,
         mac: src_mac,
@@ -1087,7 +1113,18 @@ fn get_edge_tagger(
         biz_type,
         time_span,
         ..Default::default()
+    };
+
+    if is_reversed {
+        let server_port = if ignore_server_port(flow, config.inactive_server_port_aggregation) {
+            0
+        } else {
+            src_ep.nat_real_port
+        };
+        tagger.reverse(server_port);
     }
+
+    tagger
 }
 
 fn get_l3_epc_id(l3_epc_id: i32, signal_source: SignalSource) -> i16 {

agent/src/collector/l7_quadruple_generator.rs

@@ -70,6 +70,7 @@ struct AppMeterWithL7Protocol {
     l7_protocol: L7Protocol,
     biz_type: u8,
     time_in_second: Duration,
+    is_reversed: bool,
 }
 
 struct QuadrupleStash {
@@ -278,6 +279,7 @@ impl SubQuadGen {
                 m.endpoint == l7_stats.endpoint
                     && m.biz_type == l7_stats.biz_type
                     && m.time_span == time_span
+                    && m.is_reversed == l7_stats.is_reversed
             }) {
                 // flow L7Protocol of different client ports on the same server port may be inconsistent.
                 // unknown l7_protocol needs to be judged by the close_type and duration of the flow,
@@ -298,6 +300,7 @@ impl SubQuadGen {
                     biz_type: l7_stats.biz_type,
                     time_span,
                     time_in_second,
+                    is_reversed: l7_stats.is_reversed,
                 };
                 meters.push(meter);
             }
@@ -323,6 +326,7 @@ impl SubQuadGen {
                         is_active_host1,
                         time_in_second: meter.time_in_second.into(),
                         biz_type: meter.biz_type,
+                        is_reversed: meter.is_reversed,
                         time_span,
                     });
 
@@ -359,6 +363,7 @@ impl SubQuadGen {
                     time_in_second: l7_stats.time_in_second.into(),
                     biz_type: l7_stats.biz_type,
                     time_span,
+                    is_reversed: l7_stats.is_reversed,
                 });
                 if close_type != CloseType::Unknown && close_type != CloseType::ForcedReport {
                     Self::push_closed_app_meter(
@@ -378,6 +383,7 @@ impl SubQuadGen {
                     biz_type: l7_stats.biz_type,
                     time_span,
                     time_in_second,
+                    is_reversed: l7_stats.is_reversed,
                 };
                 let _ = stash.l7_stats.insert(l7_stats.flow_id, vec![meter]);
             }

agent/src/collector/types.rs

@@ -162,6 +162,7 @@ pub struct AppMeterWithFlow {
     pub time_in_second: Timestamp,
     // request-reponse time span
     pub time_span: u32,
+    pub is_reversed: bool,
 }
 
 #[derive(Clone)]

agent/src/common/flow.rs

@@ -551,6 +551,7 @@ pub struct L7Stats {
     // request-reponse time span
     pub time_span: u32,
     pub biz_type: u8,
+    pub is_reversed: bool,
 }
 
 #[derive(Serialize, Debug, Default, Clone, PartialEq, Eq)]

agent/src/common/l7_protocol_info.rs

@@ -195,8 +195,12 @@ where
 
     // load endpoint from cache for *responses*
     // call this before calling `perf_stats` because the latter may remove cached entry
-    fn load_endpoint_from_cache<'a>(&mut self, param: &'a ParseParam) -> Option<String> {
-        let key = LogCacheKey::new(param, self.session_id());
+    fn load_endpoint_from_cache<'a>(
+        &mut self,
+        param: &'a ParseParam,
+        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) {
             Some(cached) if cached.endpoint.is_some() => {
                 let log = LogCache {
@@ -258,9 +262,10 @@ where
                 (&mut perf_cache.rrt_cache, &mut perf_cache.timeout_cache)
             },
         );
-        let key = LogCacheKey::new(param, self.session_id());
+        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);
+        let index = if self.is_reversed() { 1 } else { 0 };
 
         let Some(prev_info) = prev_info else {
             // If the first log is a request and on blacklist, we still need to put it in cache to handle the response,
@@ -269,7 +274,7 @@ where
             // If the first log is a response, it's perf stats will not be counted here.
             // We need to know whether its corresponding request is on blacklist before accounting.
             let ret = if cur_info.msg_type == LogMessageType::Request && !cur_info.on_blacklist {
-                timeout_counter.in_cache += 1;
+                timeout_counter.in_cache[index] += 1;
                 Some(L7PerfStats::from(&cur_info))
             } else {
                 None
@@ -287,7 +292,8 @@ where
             if prev_info.msg_type != cur_info.msg_type && !merge_info.merged {
                 merge_info.merged = true;
                 if !(prev_info.on_blacklist || cur_info.on_blacklist) {
-                    timeout_counter.in_cache = timeout_counter.in_cache.saturating_sub(1);
+                    timeout_counter.in_cache[index] =
+                        timeout_counter.in_cache[index].saturating_sub(1);
                 }
             }
 
@@ -297,7 +303,7 @@ where
             keep_prev = !(merge_info.req_end && merge_info.resp_end);
         } else {
             if prev_info.msg_type == LogMessageType::Request && !prev_info.on_blacklist {
-                timeout_counter.in_cache = timeout_counter.in_cache.saturating_sub(1);
+                timeout_counter.in_cache[index] = timeout_counter.in_cache[index].saturating_sub(1);
             }
         }
 
@@ -311,7 +317,7 @@ where
                 if rrt > param.rrt_timeout as u64 {
                     match prev_info.multi_merge_info.as_ref() {
                         Some(info) if info.merged => (),
-                        _ => timeout_counter.timeout += 1,
+                        _ => timeout_counter.timeout[index] += 1,
                     }
                 } else {
                     perf_stats.update_rrt(rrt);
@@ -339,7 +345,7 @@ where
                         warn!("l7 log info disorder with long time rrt {}", rrt);
                         match prev_info.multi_merge_info.as_ref() {
                             Some(info) if info.merged => (),
-                            _ => timeout_counter.timeout += 1,
+                            _ => timeout_counter.timeout[index] += 1,
                         }
                     }
 
@@ -365,10 +371,10 @@ where
 
             if prev_info.time > cur_info.time {
                 if !cur_info.on_blacklist && cur_info.msg_type == LogMessageType::Request {
-                    timeout_counter.timeout += 1;
+                    timeout_counter.timeout[index] += 1;
                 }
                 if !prev_info.on_blacklist && prev_info.msg_type == LogMessageType::Request {
-                    timeout_counter.in_cache += 1;
+                    timeout_counter.in_cache[index] += 1;
                 }
                 if !cur_info.on_blacklist {
                     Some(L7PerfStats::from(&cur_info))
@@ -377,10 +383,10 @@ where
                 }
             } else {
                 if !prev_info.on_blacklist && prev_info.msg_type == LogMessageType::Request {
-                    timeout_counter.timeout += 1;
+                    timeout_counter.timeout[index] += 1;
                 }
                 if !cur_info.on_blacklist && cur_info.msg_type == LogMessageType::Request {
-                    timeout_counter.in_cache += 1;
+                    timeout_counter.in_cache[index] += 1;
                 }
                 let prev_info = rtt_cache.put(key, cur_info).unwrap();
                 if !prev_info.on_blacklist {
@@ -394,7 +400,7 @@ where
                 && prev_info.msg_type != cur_info.msg_type
                 && !prev_info.multi_merge_info.as_ref().unwrap().merged
             {
-                timeout_counter.timeout += 1;
+                timeout_counter.timeout[index] += 1;
             }
             if !keep_prev {
                 rtt_cache.pop(&key);