Skip to content

Commit 3d56a44

Browse files
yuanchaoayuanchaoa
committed
fix: incorrect endpoint
1 parent 75d0497 commit 3d56a44

21 files changed

Lines changed: 58 additions & 106 deletions

‎agent/src/common/l7_protocol_info.rs‎

Lines changed: 26 additions & 83 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ use std::cell::RefMut;
1818

1919
use super::flow::PacketDirection;
2020
use enum_dispatch::enum_dispatch;
21-
use log::{debug, error, warn};
21+
use log::{debug, error};
2222
use serde::Serialize;
2323

2424
use crate::{
@@ -224,15 +224,16 @@ where
224224

225225
/*
226226
calculate rrt
227-
if have previous log cache:
227+
if session
228+
return stats
229+
if request
230+
update rrt cache
231+
return stats
232+
if response
228233
if previous is req and current is resp and current time > previous time
229234
rrt = current time - previous time
230-
if previous is resp and current is req and current time < previous time, likely ebfp disorder
231-
rrt = previous time - current time
232-
233-
otherwise can not calculate rrt, cache current log rrt
234-
235-
if have no previous log cache, cache the current log rrt
235+
remove rrt cache
236+
return stats
236237
*/
237238
fn perf_stats(&self, param: &ParseParam) -> Option<L7PerfStats> {
238239
if param.time == 0 {
@@ -284,14 +285,16 @@ where
284285
//
285286
// If the first log is a response, its perf stats will not be counted here.
286287
// We need to know whether its corresponding request is on blacklist before accounting.
287-
let ret = if cur_info.msg_type == LogMessageType::Request && !cur_info.on_blacklist {
288-
timeout_counter.in_cache[index] += 1;
289-
Some(L7PerfStats::from(&cur_info))
288+
return if !cur_info.on_blacklist {
289+
let stats = L7PerfStats::from(&cur_info);
290+
if cur_info.msg_type == LogMessageType::Request {
291+
timeout_counter.in_cache[index] += 1;
292+
rtt_cache.put(key, cur_info);
293+
}
294+
Some(stats)
290295
} else {
291296
None
292297
};
293-
rtt_cache.put(key, cur_info);
294-
return ret;
295298
};
296299

297300
let mut keep_prev = false;
@@ -343,85 +346,25 @@ where
343346
rtt_cache.pop(&key);
344347
}
345348

346-
result
347-
} else if prev_info.is_response_of(&cur_info) {
348-
// cur_info is request, prev_info is response
349-
// request not accounted before
350-
let result = if !cur_info.on_blacklist {
351-
let mut perf_stats = L7PerfStats::from(&cur_info);
352-
353-
if !prev_info.on_blacklist {
354-
let rrt = prev_info.time - cur_info.time;
355-
if rrt > param.rrt_timeout as u64 {
356-
warn!("l7 log info disorder with long time rrt {}", rrt);
357-
match prev_info.multi_merge_info.as_ref() {
358-
Some(info) if info.merged => (),
359-
_ => timeout_counter.timeout[index] += 1,
360-
}
361-
}
362-
363-
perf_stats.sequential_merge(&L7PerfStats::from(&*prev_info));
364-
perf_stats.update_rrt(rrt);
365-
}
366-
367-
Some(perf_stats)
368-
} else {
369-
None
370-
};
371-
372-
if !keep_prev {
373-
rtt_cache.pop(&key);
374-
}
375-
376349
result
377350
} else if !self.need_merge() {
378351
debug!(
379352
"can not calculate rrt, flow_id: {}, previous log type: {:?}, previous time: {}, current log type: {:?}, current time: {}",
380353
param.flow_id, prev_info.msg_type, prev_info.time, cur_info.msg_type, cur_info.time,
381354
);
382355

383-
if prev_info.time > cur_info.time {
384-
if !cur_info.on_blacklist && cur_info.msg_type == LogMessageType::Request {
385-
timeout_counter.timeout[index] += 1;
386-
}
387-
if !prev_info.on_blacklist && prev_info.msg_type == LogMessageType::Request {
388-
timeout_counter.in_cache[index] += 1;
389-
}
390-
if !cur_info.on_blacklist {
391-
Some(L7PerfStats::from(&cur_info))
392-
} else {
393-
None
356+
if !keep_prev {
357+
rtt_cache.pop(&key);
358+
}
359+
360+
if !cur_info.on_blacklist {
361+
let stats = L7PerfStats::from(&cur_info);
362+
if cur_info.msg_type == LogMessageType::Request {
363+
rtt_cache.put(key, cur_info);
394364
}
365+
Some(stats)
395366
} else {
396-
if !prev_info.on_blacklist && prev_info.msg_type == LogMessageType::Request {
397-
timeout_counter.timeout[index] += 1;
398-
}
399-
if !cur_info.on_blacklist && cur_info.msg_type == LogMessageType::Request {
400-
timeout_counter.in_cache[index] += 1;
401-
}
402-
let cur_is_req = cur_info.msg_type == LogMessageType::Request;
403-
let cur_on_blacklist = cur_info.on_blacklist;
404-
let prev_info = rtt_cache.put(key, cur_info).unwrap();
405-
// Requests are counted (req=1) eagerly when they first enter the cache,
406-
// so re-emitting a displaced Request here would double-count it.
407-
// Responses were cached with None on arrival and must be counted here.
408-
let mut result =
409-
if !prev_info.on_blacklist && prev_info.msg_type == LogMessageType::Response {
410-
L7PerfStats::from(&prev_info)
411-
} else {
412-
L7PerfStats::default()
413-
};
414-
// A new Request entering the cache via this path (replacing a previous entry)
415-
// was never counted by the first-entry path, so emit req=1 now so that
416-
// "request accounted before" holds when its response arrives via is_request_of.
417-
if !cur_on_blacklist && cur_is_req {
418-
result.inc_req();
419-
}
420-
if result == L7PerfStats::default() {
421-
None
422-
} else {
423-
Some(result)
424-
}
367+
None
425368
}
426369
} else {
427370
if !prev_info.on_blacklist

‎agent/src/common/l7_protocol_log.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -353,13 +353,13 @@ impl LogCache {
353353
pub fn is_request_of(&self, other: &Self) -> bool {
354354
self.msg_type == LogMessageType::Request
355355
&& other.msg_type == LogMessageType::Response
356-
&& self.time < other.time
356+
&& self.time <= other.time
357357
}
358358

359359
pub fn is_response_of(&self, other: &Self) -> bool {
360360
self.msg_type == LogMessageType::Response
361361
&& other.msg_type == LogMessageType::Request
362-
&& self.time > other.time
362+
&& self.time >= other.time
363363
}
364364
}
365365

‎agent/src/common/meta_packet.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1288,6 +1288,7 @@ impl<'a> MetaPacket<'a> {
12881288
header_type: self.header_type,
12891289
l2_l3_opt_size: self.l2_l3_opt_size,
12901290
l4_opt_size: self.l4_opt_size,
1291+
flow_id: self.flow_id,
12911292
..Default::default()
12921293
})
12931294
}

‎agent/src/flow_generator/protocol_logs/fastcgi.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -563,7 +563,7 @@ impl L7ProtocolParserInterface for FastCGILog {
563563

564564
if param.parse_perf {
565565
let mut perf_stat = L7PerfStats::default();
566-
if info.msg_type == LogMessageType::Response {
566+
if info.msg_type == LogMessageType::Response && info.endpoint.is_none() {
567567
if let Some(endpoint) = info.load_endpoint_from_cache(param, false) {
568568
info.endpoint = Some(endpoint.to_string());
569569
}

‎agent/src/flow_generator/protocol_logs/http.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1326,7 +1326,7 @@ impl HttpLog {
13261326

13271327
if param.parse_perf {
13281328
let mut perf_stat = L7PerfStats::default();
1329-
if info.msg_type == LogMessageType::Response {
1329+
if info.msg_type == LogMessageType::Response && info.endpoint.is_none() {
13301330
if let Some(endpoint) = info.load_endpoint_from_cache(param, info.is_reversed) {
13311331
info.endpoint = Some(endpoint.to_string());
13321332
}

‎agent/src/flow_generator/protocol_logs/mq/amqp.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1125,7 +1125,7 @@ impl L7ProtocolParserInterface for AmqpLog {
11251125
}
11261126
if param.parse_perf {
11271127
let mut perf_stat = L7PerfStats::default();
1128-
if info.msg_type == LogMessageType::Response {
1128+
if info.msg_type == LogMessageType::Response && info.endpoint.is_none() {
11291129
if let Some(endpoint) = info.load_endpoint_from_cache(param, false) {
11301130
info.endpoint = Some(endpoint.to_string());
11311131
}

‎agent/src/flow_generator/protocol_logs/mq/kafka.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -903,7 +903,7 @@ impl L7ProtocolParserInterface for KafkaLog {
903903

904904
if param.parse_perf {
905905
let mut perf_stat = L7PerfStats::default();
906-
if info.msg_type == LogMessageType::Response {
906+
if info.msg_type == LogMessageType::Response && info.endpoint.is_none() {
907907
if let Some(endpoint) = info.load_endpoint_from_cache(param, false) {
908908
info.endpoint = Some(endpoint.to_string());
909909
}

‎agent/src/flow_generator/protocol_logs/mq/mqtt.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -316,7 +316,7 @@ impl L7ProtocolParserInterface for MqttLog {
316316

317317
if param.parse_perf {
318318
let mut perf_stat = L7PerfStats::default();
319-
if info.msg_type == LogMessageType::Response {
319+
if info.msg_type == LogMessageType::Response && info.endpoint.is_none() {
320320
if let Some(endpoint) = info.load_endpoint_from_cache(param, false) {
321321
info.endpoint = Some(endpoint.to_string());
322322
}

‎agent/src/flow_generator/protocol_logs/mq/nats.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -959,7 +959,7 @@ impl L7ProtocolParserInterface for NatsLog {
959959

960960
if param.parse_perf {
961961
let mut perf_stat = L7PerfStats::default();
962-
if info.msg_type == LogMessageType::Response {
962+
if info.msg_type == LogMessageType::Response && info.endpoint.is_none() {
963963
if let Some(endpoint) =
964964
info.load_endpoint_from_cache(param, info.is_reversed)
965965
{

‎agent/src/flow_generator/protocol_logs/mq/openwire.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1920,7 +1920,7 @@ impl L7ProtocolParserInterface for OpenWireLog {
19201920

19211921
if param.parse_perf {
19221922
let mut perf_stat = L7PerfStats::default();
1923-
if info.msg_type == LogMessageType::Response {
1923+
if info.msg_type == LogMessageType::Response && info.topic.is_none() {
19241924
if let Some(endpoint) = info.load_endpoint_from_cache(param, false) {
19251925
info.topic = Some(endpoint.to_string());
19261926
}

0 commit comments

Comments
 (0)