feat: support icmp close type

yuanchaoa · yuanchaoa · commit 8926b0d4600b · 2024-09-11T17:49:57.000+08:00
diff --git a/agent/src/collector/quadruple_generator.rs b/agent/src/collector/quadruple_generator.rs
@@ -986,6 +986,19 @@ impl QuadrupleGenerator {
                     | CloseType::TcpFin
                     | CloseType::Unknown
                     | CloseType::TcpFinClientRst
+                    | CloseType::IcmpOtherTimeout
+                    | CloseType::IcmpAddressMaskTimeout
+                    | CloseType::IcmpDestinationUnreachable
+                    | CloseType::IcmpEchoTimeout
+                    | CloseType::IcmpInformationTimeout
+                    | CloseType::IcmpNeighborTimeout
+                    | CloseType::IcmpNormal
+                    | CloseType::IcmpPacketTooBig
+                    | CloseType::IcmpParameterProblem
+                    | CloseType::IcmpRouterTimeout
+                    | CloseType::IcmpSourceQuench
+                    | CloseType::IcmpTimeExceeded
+                    | CloseType::IcmpTimestampTimeout
                     | CloseType::Max => (),
                 }
             }
diff --git a/agent/src/common/flow.rs b/agent/src/common/flow.rs
@@ -62,22 +62,35 @@ const COUNTER_FLOW_ID_MASK: u64 = 0x00FFFFFF;
 #[repr(u8)]
 pub enum CloseType {
     Unknown = 0,
-    TcpFin = 1,                 //  1: 正常结束
-    TcpServerRst = 2,           //  2: 传输-服务端重置
-    Timeout = 3,                //  3: 连接超时
-    ForcedReport = 5,           //  5: 周期性上报
-    ClientSynRepeat = 7,        //  7: 建连-客户端SYN结束
-    ServerHalfClose = 8,        //  8: 断连-服务端半关
-    TcpClientRst = 9,           //  9: 传输-客户端重置
-    ServerSynAckRepeat = 10,    // 10: 建连-服务端SYN结束
-    ClientHalfClose = 11,       // 11: 断连-客户端半关
-    ClientSourcePortReuse = 13, // 13: 建连-客户端端口复用
-    ServerReset = 15,           // 15: 建连-服务端直接重置
-    ServerQueueLack = 17,       // 17: 传输-服务端队列溢出
-    ClientEstablishReset = 18,  // 18: 建连-客户端其他重置
-    ServerEstablishReset = 19,  // 19: 建连-服务端其他重置
-    TcpFinClientRst = 20,       // 20: 正常结束-客户端重置
-    Max = 21,
+    TcpFin = 1,                      //  1: 正常结束
+    TcpServerRst = 2,                //  2: 传输-服务端重置
+    Timeout = 3,                     //  3: 连接超时
+    ForcedReport = 5,                //  5: 周期性上报
+    ClientSynRepeat = 7,             //  7: 建连-客户端SYN结束
+    ServerHalfClose = 8,             //  8: 断连-服务端半关
+    TcpClientRst = 9,                //  9: 传输-客户端重置
+    ServerSynAckRepeat = 10,         // 10: 建连-服务端SYN结束
+    ClientHalfClose = 11,            // 11: 断连-客户端半关
+    ClientSourcePortReuse = 13,      // 13: 建连-客户端端口复用
+    ServerReset = 15,                // 15: 建连-服务端直接重置
+    ServerQueueLack = 17,            // 17: 传输-服务端队列溢出
+    ClientEstablishReset = 18,       // 18: 建连-客户端其他重置
+    ServerEstablishReset = 19,       // 19: 建连-服务端其他重置
+    TcpFinClientRst = 20,            // 20: 正常结束-客户端重置
+    IcmpNormal = 21,                 // 21: ICMP-正常结束
+    IcmpAddressMaskTimeout = 22,     // 22: ICMP-地址掩码超时
+    IcmpDestinationUnreachable = 23, // 23: ICMP-目标不可达
+    IcmpEchoTimeout = 24,            // 24: ICMP-PING超时
+    IcmpInformationTimeout = 25,     // 25: ICMP-信息超时
+    IcmpParameterProblem = 26,       // 26: ICMP-参数问题
+    IcmpRouterTimeout = 27,          // 27: ICMP-路由器超时
+    IcmpSourceQuench = 28,           // 28: ICMP-源站抑制
+    IcmpTimeExceeded = 29,           // 29: ICMP-TTL溢出
+    IcmpTimestampTimeout = 30,       // 30: ICMP-时间戳超时
+    IcmpNeighborTimeout = 31,        // 31: ICMP-邻居发现超时
+    IcmpPacketTooBig = 32,           // 32: ICMP-Packet过大
+    IcmpOtherTimeout = 33,           // 33: ICMP-未知超时
+    Max = 34,
 }
 
 impl CloseType {
@@ -97,6 +110,18 @@ impl CloseType {
             || self == CloseType::ServerReset
             || self == CloseType::ServerQueueLack
             || self == CloseType::ServerEstablishReset
+            || self == CloseType::IcmpAddressMaskTimeout
+            || self == CloseType::IcmpEchoTimeout
+            || self == CloseType::IcmpInformationTimeout
+            || self == CloseType::IcmpRouterTimeout
+            || self == CloseType::IcmpTimestampTimeout
+            || self == CloseType::IcmpNeighborTimeout
+            || self == CloseType::IcmpPacketTooBig
+            || self == CloseType::IcmpTimeExceeded
+            || self == CloseType::IcmpSourceQuench
+            || self == CloseType::IcmpDestinationUnreachable
+            || self == CloseType::IcmpParameterProblem
+            || self == CloseType::IcmpOtherTimeout
     }
 }
 
@@ -1070,7 +1095,40 @@ impl Flow {
             FlowState::Exception => CloseType::Unknown,
             FlowState::Opening1 => CloseType::ClientSynRepeat,
             FlowState::Opening2 => CloseType::ServerSynAckRepeat,
-            FlowState::Established => CloseType::Timeout,
+            FlowState::IcmpAddressMaskReply => CloseType::IcmpNormal,
+            FlowState::IcmpAddressMaskRequest => CloseType::IcmpAddressMaskTimeout,
+            FlowState::IcmpEchoReply => CloseType::IcmpNormal,
+            FlowState::IcmpEchoRequest => CloseType::IcmpEchoTimeout,
+            FlowState::IcmpDestinationUnreachable => CloseType::IcmpDestinationUnreachable,
+            FlowState::IcmpInformationReply => CloseType::IcmpNormal,
+            FlowState::IcmpInformationRequest => CloseType::IcmpInformationTimeout,
+            FlowState::IcmpNeighborAdvert => CloseType::IcmpNormal,
+            FlowState::IcmpNeighborSolicit => CloseType::IcmpNeighborTimeout,
+            FlowState::IcmpPacketTooBig => CloseType::IcmpPacketTooBig,
+            FlowState::IcmpParameterProblem => CloseType::IcmpParameterProblem,
+            FlowState::IcmpRedirectMessage => CloseType::IcmpNormal,
+            FlowState::IcmpRouterAdvertisement => CloseType::IcmpNormal,
+            FlowState::IcmpRouterSolicitation => CloseType::IcmpRouterTimeout,
+            FlowState::IcmpSourceQuench => CloseType::IcmpSourceQuench,
+            FlowState::IcmpTimeExceeded => CloseType::IcmpTimeExceeded,
+            FlowState::IcmpTimestamp => CloseType::IcmpTimestampTimeout,
+            FlowState::IcmpTimestampReply => CloseType::IcmpNormal,
+            FlowState::IcmpTraceroute => CloseType::IcmpNormal,
+            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::IcmpOtherTimeout
+                    } else {
+                        CloseType::IcmpNormal
+                    }
+                } else {
+                    CloseType::Timeout
+                }
+            }
             FlowState::ClosingTx1 => CloseType::ServerHalfClose,
             FlowState::ClosingRx1 => CloseType::ClientHalfClose,
             FlowState::ClosingTx2 | FlowState::ClosingRx2 | FlowState::Closed => CloseType::TcpFin,
diff --git a/agent/src/flow_generator/flow_config.rs b/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),
         };
diff --git a/agent/src/flow_generator/flow_map.rs b/agent/src/flow_generator/flow_map.rs
@@ -915,7 +915,12 @@ 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;
+            node.set_icmp_flow_state(meta_packet);
+        } 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;
@@ -1732,9 +1737,15 @@ impl FlowMap {
         meta_packet.flow_id = node.tagged_flow.flow.flow_id;
         meta_packet.second_in_minute =
             (node.tagged_flow.flow.start_time.as_secs() % SECONDS_IN_MINUTE) as u8;
-        node.flow_state = FlowState::Established;
-        // opening timeout
-        node.timeout = config.flow.flow_timeout.opening;
+        if meta_packet.lookup_key.proto == IpProtocol::ICMPV4
+            || meta_packet.lookup_key.proto == IpProtocol::ICMPV6
+        {
+            node.set_icmp_flow_state(meta_packet);
+            node.timeout = config.flow.flow_timeout.icmp_timeout;
+        } else {
+            node.flow_state = FlowState::Established;
+            node.timeout = config.flow.flow_timeout.opening;
+        }
         if let Some(meta_flow_log) = node.meta_flow_log.as_mut() {
             let _ = meta_flow_log.parse_l3(meta_packet);
         }
@@ -3087,6 +3098,7 @@ mod tests {
                 exception: Timestamp::from_secs(5),
                 closed_fin: Timestamp::ZERO,
                 single_direction: Timestamp::from_millis(10),
+                icmp_timeout: Timestamp::from_secs(5),
                 max: Timestamp::from_secs(300),
                 min: Timestamp::ZERO,
             }),
diff --git a/agent/src/flow_generator/flow_node.rs b/agent/src/flow_generator/flow_node.rs
@@ -16,18 +16,23 @@
 
 use std::{net::IpAddr, sync::Arc};
 
+use pnet::packet::{
+    icmp::{IcmpType, IcmpTypes},
+    icmpv6::{Icmpv6Type, Icmpv6Types},
+};
+
 use super::{perf::FlowLog, FlowState, FLOW_METRICS_PEER_DST, FLOW_METRICS_PEER_SRC};
 use crate::common::{
     decapsulate::TunnelType,
     endpoint::EndpointDataPov,
     enums::{EthernetType, TapType, TcpFlags},
     flow::{FlowMetricsPeer, L7PerfStats, PacketDirection, SignalSource, TcpPerfStats},
     lookup_key::LookupKey,
-    meta_packet::MetaPacket,
+    meta_packet::{MetaPacket, ProtocolData},
     tagged_flow::TaggedFlow,
     TapPort, Timestamp,
 };
-use public::{proto::common::TridentType, utils::net::MacAddr};
+use public::{enums::IpProtocol, proto::common::TridentType, utils::net::MacAddr};
 
 use npb_pcap_policy::PolicyData;
 use packet_sequence_block::PacketSequenceBlock;
@@ -145,6 +150,65 @@ pub struct FlowNode {
 }
 
 impl FlowNode {
+    pub(super) fn set_icmpv4_flow_state(&mut self, meta_packet: &MetaPacket) {
+        let ProtocolData::IcmpData(icmp_data) = &meta_packet.protocol_data else {
+            return;
+        };
+        let icmp_type = IcmpType::new(icmp_data.icmp_type);
+        match icmp_type {
+            IcmpTypes::AddressMaskReply => self.flow_state = FlowState::IcmpAddressMaskReply,
+            IcmpTypes::AddressMaskRequest => self.flow_state = FlowState::IcmpAddressMaskRequest,
+            IcmpTypes::DestinationUnreachable => {
+                self.flow_state = FlowState::IcmpDestinationUnreachable
+            }
+            IcmpTypes::EchoReply => self.flow_state = FlowState::IcmpEchoReply,
+            IcmpTypes::EchoRequest => self.flow_state = FlowState::IcmpEchoRequest,
+            IcmpTypes::InformationReply => self.flow_state = FlowState::IcmpInformationReply,
+            IcmpTypes::InformationRequest => self.flow_state = FlowState::IcmpInformationRequest,
+            IcmpTypes::ParameterProblem => self.flow_state = FlowState::IcmpParameterProblem,
+            IcmpTypes::RedirectMessage => self.flow_state = FlowState::IcmpRedirectMessage,
+            IcmpTypes::RouterAdvertisement => self.flow_state = FlowState::IcmpRouterAdvertisement,
+            IcmpTypes::RouterSolicitation => self.flow_state = FlowState::IcmpRouterSolicitation,
+            IcmpTypes::SourceQuench => self.flow_state = FlowState::IcmpSourceQuench,
+            IcmpTypes::TimeExceeded => self.flow_state = FlowState::IcmpTimeExceeded,
+            IcmpTypes::Timestamp => self.flow_state = FlowState::IcmpTimestamp,
+            IcmpTypes::TimestampReply => self.flow_state = FlowState::IcmpTimestampReply,
+            IcmpTypes::Traceroute => self.flow_state = FlowState::IcmpTraceroute,
+            _ => self.flow_state = FlowState::Established,
+        }
+    }
+
+    pub(super) fn set_icmpv6_flow_state(&mut self, meta_packet: &MetaPacket) {
+        let ProtocolData::IcmpData(icmp_data) = &meta_packet.protocol_data else {
+            return;
+        };
+        let icmp_type = Icmpv6Type::new(icmp_data.icmp_type);
+        match icmp_type {
+            Icmpv6Types::DestinationUnreachable => {
+                self.flow_state = FlowState::IcmpDestinationUnreachable
+            }
+            Icmpv6Types::EchoReply => self.flow_state = FlowState::IcmpEchoReply,
+            Icmpv6Types::EchoRequest => self.flow_state = FlowState::IcmpEchoRequest,
+            Icmpv6Types::NeighborAdvert => self.flow_state = FlowState::IcmpNeighborAdvert,
+            Icmpv6Types::NeighborSolicit => self.flow_state = FlowState::IcmpNeighborSolicit,
+            Icmpv6Types::PacketTooBig => self.flow_state = FlowState::IcmpPacketTooBig,
+            Icmpv6Types::ParameterProblem => self.flow_state = FlowState::IcmpParameterProblem,
+            Icmpv6Types::Redirect => self.flow_state = FlowState::IcmpRedirectMessage,
+            Icmpv6Types::RouterAdvert => self.flow_state = FlowState::IcmpRouterAdvertisement,
+            Icmpv6Types::RouterSolicit => self.flow_state = FlowState::IcmpRouterSolicitation,
+            Icmpv6Types::TimeExceeded => self.flow_state = FlowState::IcmpTimeExceeded,
+            _ => self.flow_state = FlowState::Established,
+        }
+    }
+
+    pub(super) fn set_icmp_flow_state(&mut self, meta_packet: &MetaPacket) {
+        match meta_packet.lookup_key.proto {
+            IpProtocol::ICMPV4 => self.set_icmpv4_flow_state(meta_packet),
+            IpProtocol::ICMPV6 => self.set_icmpv6_flow_state(meta_packet),
+            _ => unreachable!(),
+        }
+    }
+
     pub(super) fn reset_flow_stat_info(&mut self) {
         self.policy_in_tick = [false; 2];
         self.packet_in_tick = false;
diff --git a/agent/src/flow_generator/flow_state.rs b/agent/src/flow_generator/flow_state.rs
@@ -44,6 +44,26 @@ pub enum FlowState {
     EstablishReset,
     OpeningRst,
 
+    IcmpAddressMaskReply,
+    IcmpAddressMaskRequest,
+    IcmpDestinationUnreachable,
+    IcmpEchoReply,
+    IcmpEchoRequest,
+    IcmpInformationReply,
+    IcmpInformationRequest,
+    IcmpParameterProblem,
+    IcmpRedirectMessage,
+    IcmpRouterAdvertisement,
+    IcmpRouterSolicitation,
+    IcmpSourceQuench,
+    IcmpTimeExceeded,
+    IcmpTimestamp,
+    IcmpTimestampReply,
+    IcmpTraceroute,
+    IcmpNeighborAdvert,
+    IcmpNeighborSolicit,
+    IcmpPacketTooBig,
+
     Max,
 }
 
diff --git a/server/ingester/flow_log/log_data/l4_flow_log.go b/server/ingester/flow_log/log_data/l4_flow_log.go
@@ -858,11 +858,11 @@ func (k *KnowledgeGraph) FillL4(f *pb.Flow, isIPv6 bool, platformData *grpc.Plat
 
 func getStatus(t datatype.CloseType, p layers.IPProtocol) datatype.LogMessageStatus {
 	if t == datatype.CloseTypeTCPFin || t == datatype.CloseTypeForcedReport || t == datatype.CloseTypeTCPFinClientRst ||
-		(p != layers.IPProtocolTCP && t == datatype.CloseTypeTimeout) {
+		(p != layers.IPProtocolTCP && t == datatype.CloseTypeTimeout) || t == atatype.CloseTypeIcmpNormal {
 		return datatype.STATUS_OK
 	} else if t.IsClientError() {
 		return datatype.STATUS_CLIENT_ERROR
-	} else if p == layers.IPProtocolTCP && t.IsServerError() {
+	} else if (p == layers.IPProtocolTCP || p == layers.IPProtocolICMPv4 || p == layers.IPProtocolICMPv6) && t.IsServerError() {
 		return datatype.STATUS_SERVER_ERROR
 	} else {
 		return datatype.STATUS_NOT_EXIST
diff --git a/server/libs/datatype/flow.go b/server/libs/datatype/flow.go
diff --git a/server/querier/db_descriptions/clickhouse/tag/enum/close_type.ch b/server/querier/db_descriptions/clickhouse/tag/enum/close_type.ch
diff --git a/server/querier/db_descriptions/clickhouse/tag/enum/close_type.en b/server/querier/db_descriptions/clickhouse/tag/enum/close_type.en
