feat: support icmp close type

Author: yuanchaoa

agent/src/collector/quadruple_generator.rs

@@ -986,6 +986,7 @@ impl QuadrupleGenerator {
                     | CloseType::TcpFin
                     | CloseType::Unknown
                     | CloseType::TcpFinClientRst
+                    | CloseType::IcmpUnknown
                     | CloseType::Max => (),
                 }
             }

agent/src/common/flow.rs

@@ -77,7 +77,8 @@ pub enum CloseType {
     ClientEstablishReset = 18,  // 18: 建连-客户端其他重置
     ServerEstablishReset = 19,  // 19: 建连-服务端其他重置
     TcpFinClientRst = 20,       // 20: 正常结束-客户端重置
-    Max = 21,
+    IcmpUnknown = 21,           // 21: TODO
+    Max = 22,
 }
 
 impl CloseType {
@@ -87,6 +88,7 @@ impl CloseType {
             || self == CloseType::ClientHalfClose
             || self == CloseType::ClientSourcePortReuse
             || self == CloseType::ClientEstablishReset
+            || self == CloseType::IcmpUnknown
     }
 
     pub fn is_server_error(self) -> bool {
@@ -1070,7 +1072,21 @@ impl Flow {
             FlowState::Exception => CloseType::Unknown,
             FlowState::Opening1 => CloseType::ClientSynRepeat,
             FlowState::Opening2 => CloseType::ServerSynAckRepeat,
-            FlowState::Established => CloseType::Timeout,
+            FlowState::Established => {
+                if self.flow_key.proto == IpProtocol::ICMPV4
+                    || self.flow_key.proto == IpProtocol::ICMPV6
+                {
+                    if self.flow_metrics_peers[0].total_packet_count
+                        != self.flow_metrics_peers[1].total_packet_count
+                    {
+                        CloseType::IcmpUnknown
+                    } else {
+                        CloseType::Timeout
+                    }
+                } else {
+                    CloseType::Timeout
+                }
+            }
             FlowState::ClosingTx1 => CloseType::ServerHalfClose,
             FlowState::ClosingRx1 => CloseType::ClientHalfClose,
             FlowState::ClosingTx2 | FlowState::ClosingRx2 | FlowState::Closed => CloseType::TcpFin,

agent/src/flow_generator/flow_config.rs

@@ -26,6 +26,7 @@ pub const TIMEOUT_OTHERS: Timestamp = Timestamp::from_secs(5);
 pub const TIMEOUT_ESTABLISHED: Timestamp = Timestamp::from_secs(300);
 pub const TIMEOUT_CLOSING: Timestamp = Timestamp::from_secs(35);
 pub const TIMEOUT_OPENING_RST: Timestamp = Timestamp::from_secs(1);
+pub const TIMEOUT_ICMP: Timestamp = Timestamp::from_secs(5);
 
 pub struct TcpTimeout {
     pub established: Timestamp,
@@ -55,6 +56,7 @@ pub struct FlowTimeout {
     pub closed_fin: Timestamp,
     pub single_direction: Timestamp,
     pub opening_rst: Timestamp,
+    pub icmp_timeout: Timestamp,
 
     pub min: Timestamp,
     pub max: Timestamp, // time window
@@ -71,6 +73,8 @@ impl From<TcpTimeout> for FlowTimeout {
             closed_fin: Timestamp::from_secs(2),
             single_direction: t.others,
             opening_rst: t.opening_rst,
+            icmp_timeout: TIMEOUT_ICMP,
+
             min: Timestamp::from_secs(0),
             max: Timestamp::from_secs(0),
         };

agent/src/flow_generator/flow_map.rs

@@ -915,7 +915,11 @@ impl FlowMap {
     ) -> bool {
         self.update_flow(config, node, meta_packet);
         let peers = &node.tagged_flow.flow.flow_metrics_peers;
-        if peers[FLOW_METRICS_PEER_SRC].packet_count > 0
+        if node.tagged_flow.flow.flow_key.proto == IpProtocol::ICMPV4
+            || node.tagged_flow.flow.flow_key.proto == IpProtocol::ICMPV6
+        {
+            node.timeout = config.flow.flow_timeout.icmp_timeout;
+        } else if peers[FLOW_METRICS_PEER_SRC].packet_count > 0
             && peers[FLOW_METRICS_PEER_DST].packet_count > 0
         {
             node.timeout = config.flow.flow_timeout.established_rst;

server/libs/datatype/flow.go

@@ -69,13 +69,14 @@ const (
 	CloseTypeClientEstablishReset  // 18: 建连-客户端其他重置
 	CloseTypeServerEstablishReset  // 19: 建连-服务端其他重置
 	CloseTypeTCPFinClientRst       // 20: 正常结束-客户端重置
+	CloseTypeIcmpUnknown           // 21: TODO
 	MaxCloseType
 )
 
 func (t CloseType) IsClientError() bool {
 	return t == CloseTypeClientSYNRepeat || t == CloseTypeTCPClientRst ||
 		t == CloseTypeClientHalfClose || t == CloseTypeClientSourcePortReuse ||
-		t == CloseTypeClientEstablishReset
+		t == CloseTypeClientEstablishReset || t == CloseTypeIcmpUnknown
 }
 
 func (t CloseType) IsServerError() bool {