postfix_log_parser/components/

cleanup.rs

//! 清理(cleanup)组件解析器

use crate::components::ComponentParser;
use crate::error::ParseError;
use crate::events::cleanup::CleanupEvent;
use crate::events::ComponentEvent;
use crate::utils::common_fields::CommonFieldsParser;
use crate::utils::queue_id::create_queue_id_pattern;
use lazy_static::lazy_static;
use regex::Regex;

/// Cleanup组件解析器
/// 基于896,788个真实生产数据分析，cleanup组件占4.5%的日志
/// 解析顺序按照真实数据频率优化
pub struct CleanupParser;

lazy_static! {
    // 1. Message-ID事件 - 最高频事件，优先解析 (支持带括号和不带括号格式)
    static ref MESSAGE_ID_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: message-id=(?:<([^>]+)>|([^,\s]+))$")
    ).unwrap();

    // 2. 队列文件警告 - 第二高频，系统问题相关
    static ref QUEUE_FILE_WARNING_REGEX: Regex = Regex::new(
        r"^([^:]+): create file ([^:]+): (.+)$"
    ).unwrap();

    // 3. 邮件大小信息
    static ref MESSAGE_SIZE_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: size=(\d+)$")
    ).unwrap();

    // 4. 邮件头处理
    static ref HEADER_PROCESSING_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: header ([^:]+): (.+)$")
    ).unwrap();

    // 5. 地址重写
    static ref ADDRESS_REWRITE_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: (from|to)=<([^>]+)> -> <([^>]+)>$")
    ).unwrap();

    // 6. 邮件内容重写
    static ref MESSAGE_REWRITE_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: rewrite: (.+)$")
    ).unwrap();

    // 7. 过滤器动作
    static ref FILTER_ACTION_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: filter ([^:]+): (.+)$")
    ).unwrap();

    // 8. Milter交互
    static ref MILTER_INTERACTION_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: milter ([^:]+): (.+)$")
    ).unwrap();

    // 9. 邮件拒绝
    static ref MESSAGE_REJECT_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: reject: (.+)$")
    ).unwrap();

    // 10. 配置警告
    static ref CONFIG_WARNING_REGEX: Regex = Regex::new(
        r"^warning: (.+)$"
    ).unwrap();

    // 11. 资源限制
    static ref RESOURCE_LIMIT_REGEX: Regex = Regex::new(
        r"^warning: ([^:]+): (.+)$"
    ).unwrap();

    // 12. 邮件隔离/保留事件
    static ref MESSAGE_HOLD_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: hold: (.+)$")
    ).unwrap();

    // 13. 邮件丢弃事件
    static ref MESSAGE_DISCARD_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: discard: (.+)$")
    ).unwrap();

    // 14. 邮件移除事件
    static ref MESSAGE_REMOVED_REGEX: Regex = Regex::new(
        &create_queue_id_pattern(r"^{QUEUE_ID}: removed \(([^)]+)\)(?:\s*(.*))?$")
    ).unwrap();

        // 15. 统计信息
    static ref STATISTICS_REGEX: Regex = Regex::new(
        r"^statistics: processed=(\d+) rejected=(\d+)(?:\s+errors=(\d+))?$"
    ).unwrap();

    // 16. Snowflake ID生成器初始化
    static ref SNOWFLAKE_INIT_REGEX: Regex = Regex::new(
        r"^snowflake: initialized with node_id=(\d+), node_bits=(\d+), seq_bits=(\d+)$"
    ).unwrap();

    // MessageHold 事件专用的正则表达式（仅保留cleanup特定的）
    static ref HOLD_CLIENT_REGEX: Regex = Regex::new(r"from ([^:;\s]+\[[^\]]+\]:?\d*)").unwrap();
}

impl CleanupParser {
    pub fn new() -> Self {
        CleanupParser
    }

    /// 解析Message-ID事件 - 最高频事件 (支持带括号和不带括号格式)
    fn parse_message_id(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = MESSAGE_ID_REGEX.captures(message) {
            let queue_id = captures.get(1)?.as_str().to_string();

            // 检查带尖括号的格式 (第2个捕获组)
            let message_id = if let Some(bracketed) = captures.get(2) {
                bracketed.as_str().to_string()
            }
            // 检查不带尖括号的格式 (第3个捕获组)
            else if let Some(unbracketed) = captures.get(3) {
                unbracketed.as_str().to_string()
            } else {
                return None;
            };

            return Some(CleanupEvent::MessageId {
                queue_id,
                message_id,
            });
        }
        None
    }

    /// 解析队列文件警告
    fn parse_queue_file_warning(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = QUEUE_FILE_WARNING_REGEX.captures(message) {
            return Some(CleanupEvent::QueueFileWarning {
                operation: captures.get(1)?.as_str().to_string(),
                file_path: captures.get(2)?.as_str().to_string(),
                error_reason: captures.get(3)?.as_str().to_string(),
            });
        }
        None
    }

    /// 解析邮件大小信息
    fn parse_message_size(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = MESSAGE_SIZE_REGEX.captures(message) {
            if let Ok(size) = captures.get(2)?.as_str().parse::<u64>() {
                return Some(CleanupEvent::MessageSize {
                    queue_id: captures.get(1)?.as_str().to_string(),
                    size,
                });
            }
        }
        None
    }

    /// 解析邮件头处理
    fn parse_header_processing(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = HEADER_PROCESSING_REGEX.captures(message) {
            return Some(CleanupEvent::HeaderProcessing {
                queue_id: captures.get(1)?.as_str().to_string(),
                header_name: captures.get(2)?.as_str().to_string(),
                header_value: captures.get(3)?.as_str().to_string(),
                action: "process".to_string(), // 默认动作
            });
        }
        None
    }

    /// 解析地址重写
    fn parse_address_rewrite(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = ADDRESS_REWRITE_REGEX.captures(message) {
            return Some(CleanupEvent::AddressRewrite {
                queue_id: captures.get(1)?.as_str().to_string(),
                address_type: captures.get(2)?.as_str().to_string(),
                original_address: captures.get(3)?.as_str().to_string(),
                rewritten_address: captures.get(4)?.as_str().to_string(),
            });
        }
        None
    }

    /// 解析邮件内容重写
    fn parse_message_rewrite(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = MESSAGE_REWRITE_REGEX.captures(message) {
            return Some(CleanupEvent::MessageRewrite {
                queue_id: captures.get(1)?.as_str().to_string(),
                rewrite_type: "content".to_string(),
                original: "".to_string(), // 需要更详细的解析
                rewritten: captures.get(2)?.as_str().to_string(),
            });
        }
        None
    }

    /// 解析过滤器动作
    fn parse_filter_action(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = FILTER_ACTION_REGEX.captures(message) {
            return Some(CleanupEvent::FilterAction {
                queue_id: captures.get(1)?.as_str().to_string(),
                filter_name: captures.get(2)?.as_str().to_string(),
                action: captures.get(3)?.as_str().to_string(),
                details: None,
            });
        }
        None
    }

    /// 解析Milter交互
    fn parse_milter_interaction(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = MILTER_INTERACTION_REGEX.captures(message) {
            return Some(CleanupEvent::MilterInteraction {
                queue_id: captures.get(1)?.as_str().to_string(),
                milter_name: captures.get(2)?.as_str().to_string(),
                command: "interaction".to_string(),
                response: Some(captures.get(3)?.as_str().to_string()),
            });
        }
        None
    }

    /// 解析邮件拒绝
    fn parse_message_reject(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = MESSAGE_REJECT_REGEX.captures(message) {
            return Some(CleanupEvent::MessageReject {
                queue_id: captures.get(1)?.as_str().to_string(),
                reason: captures.get(2)?.as_str().to_string(),
                action: "reject".to_string(),
            });
        }
        None
    }

    /// 解析配置警告
    fn parse_config_warning(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = CONFIG_WARNING_REGEX.captures(message) {
            let warning_msg = captures.get(1)?.as_str();

            // 先检查是否是资源限制类警告
            if warning_msg.contains("disk")
                || warning_msg.contains("memory")
                || warning_msg.contains("queue")
                || warning_msg.contains("limit")
            {
                return Some(CleanupEvent::ResourceLimit {
                    resource_type: "unknown".to_string(),
                    limit_details: warning_msg.to_string(),
                    current_value: None,
                    limit_value: None,
                });
            }

            return Some(CleanupEvent::ConfigurationWarning {
                warning_type: "cleanup_config".to_string(),
                message: warning_msg.to_string(),
            });
        }
        None
    }

    /// 解析邮件隔离/保留事件
    fn parse_message_hold(&self, message: &str) -> Option<CleanupEvent> {
        let captures = MESSAGE_HOLD_REGEX.captures(message)?;
        let queue_id = captures.get(1)?.as_str().to_string();
        let hold_details = captures.get(2)?.as_str();

        // 解析隔离原因
        let hold_reason = self.extract_hold_reason(hold_details);

        // 使用公共字段解析器提取各字段信息
        let sender =
            CommonFieldsParser::extract_from_email(hold_details).map(|email| email.address);

        let recipient =
            CommonFieldsParser::extract_to_email(hold_details).map(|email| email.address);

        let (client_hostname, client_ip, client_port) = self.extract_client_info(hold_details);

        let protocol = CommonFieldsParser::extract_protocol(hold_details);

        let helo = CommonFieldsParser::extract_helo(hold_details);

        // 提取描述信息（通常在冒号后面）
        let description = if let Some(desc_start) = hold_details.rfind(": ") {
            hold_details[(desc_start + 2)..].to_string()
        } else {
            hold_details.to_string()
        };

        Some(CleanupEvent::MessageHold {
            queue_id,
            hold_reason,
            sender,
            recipient,
            client_ip,
            client_hostname,
            client_port,
            protocol,
            helo,
            description,
        })
    }

    /// 提取隔离原因的辅助函数
    fn extract_hold_reason(&self, hold_details: &str) -> String {
        if hold_details.contains("header X-Decision-Result: Quarantine") {
            "X-Decision-Result: Quarantine".to_string()
        } else if hold_details.contains("hold") {
            "hold".to_string()
        } else {
            "unknown".to_string()
        }
    }

    /// 提取并解析客户端信息的辅助函数
    fn extract_client_info(
        &self,
        hold_details: &str,
    ) -> (Option<String>, Option<String>, Option<u16>) {
        let client_info_str = match HOLD_CLIENT_REGEX
            .captures(hold_details)
            .and_then(|c| c.get(1))
            .map(|m| m.as_str())
        {
            Some(s) => s,
            None => return (None, None, None),
        };

        // 使用公共字段解析器解析客户端信息
        CommonFieldsParser::extract_client_info_simple(client_info_str)
            .map(|client| (Some(client.hostname), Some(client.ip), client.port))
            .unwrap_or((None, None, None))
    }

    /// 解析邮件丢弃事件
    fn parse_message_discard(&self, message: &str) -> Option<CleanupEvent> {
        let captures = MESSAGE_DISCARD_REGEX.captures(message)?;
        let queue_id = captures.get(1)?.as_str().to_string();
        let discard_details = captures.get(2)?.as_str();

        // 解析丢弃原因
        let discard_reason = self.extract_discard_reason(discard_details);

        // 使用公共字段解析器提取各字段信息
        let sender =
            CommonFieldsParser::extract_from_email(discard_details).map(|email| email.address);

        let recipient =
            CommonFieldsParser::extract_to_email(discard_details).map(|email| email.address);

        let (client_hostname, client_ip, client_port) = self.extract_client_info(discard_details);

        let protocol = CommonFieldsParser::extract_protocol(discard_details);

        let helo = CommonFieldsParser::extract_helo(discard_details);

        // 提取描述信息（通常在冒号后面）
        let description = if let Some(desc_start) = discard_details.rfind(": ") {
            discard_details[(desc_start + 2)..].to_string()
        } else {
            discard_details.to_string()
        };

        Some(CleanupEvent::MessageDiscard {
            queue_id,
            discard_reason,
            sender,
            recipient,
            client_ip,
            client_hostname,
            client_port,
            protocol,
            helo,
            description,
        })
    }

    /// 解析邮件移除事件
    fn parse_message_removed(&self, message: &str) -> Option<CleanupEvent> {
        let captures = MESSAGE_REMOVED_REGEX.captures(message)?;
        let queue_id = captures.get(1)?.as_str().to_string();
        let removal_reason = captures.get(2)?.as_str().to_string();
        let details = captures.get(3).map(|m| m.as_str().to_string());

        Some(CleanupEvent::MessageRemoved {
            queue_id,
            removal_reason,
            details,
        })
    }

    /// 提取丢弃原因的辅助函数
    fn extract_discard_reason(&self, discard_details: &str) -> String {
        if discard_details.contains("header X-Decision-Result: Discard") {
            "X-Decision-Result: Discard".to_string()
        } else if discard_details.contains("discard") {
            "discard".to_string()
        } else {
            "unknown".to_string()
        }
    }

    /// 解析统计信息
    fn parse_statistics(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = STATISTICS_REGEX.captures(message) {
            let processed = captures.get(1)?.as_str().parse::<u32>().ok();
            let rejected = captures.get(2)?.as_str().parse::<u32>().ok();
            let errors = captures.get(3).and_then(|m| m.as_str().parse::<u32>().ok());

            return Some(CleanupEvent::Statistics {
                processed,
                rejected,
                errors,
            });
        }
        None
    }

    /// 解析Snowflake初始化事件
    fn parse_snowflake_init(&self, message: &str) -> Option<CleanupEvent> {
        if let Some(captures) = SNOWFLAKE_INIT_REGEX.captures(message) {
            let node_id = captures.get(1)?.as_str().parse::<u32>().ok()?;
            let node_bits = captures.get(2)?.as_str().parse::<u32>().ok()?;
            let seq_bits = captures.get(3)?.as_str().parse::<u32>().ok()?;

            return Some(CleanupEvent::SnowflakeInit {
                node_id,
                node_bits,
                seq_bits,
            });
        }
        None
    }
}

impl ComponentParser for CleanupParser {
    fn parse(&self, message: &str) -> Result<ComponentEvent, ParseError> {
        // 按照真实数据频率优化的解析顺序
        // 1. Message-ID事件 - 最高频，约占90%的cleanup日志
        if let Some(event) = self.parse_message_id(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 2. 队列文件警告 - 系统问题，重要性高
        if let Some(event) = self.parse_queue_file_warning(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 3. 邮件大小信息
        if let Some(event) = self.parse_message_size(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 4. 邮件头处理
        if let Some(event) = self.parse_header_processing(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 5. 地址重写
        if let Some(event) = self.parse_address_rewrite(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 6. 邮件内容重写
        if let Some(event) = self.parse_message_rewrite(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 7. 过滤器动作
        if let Some(event) = self.parse_filter_action(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 8. Milter交互
        if let Some(event) = self.parse_milter_interaction(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 9. 邮件拒绝
        if let Some(event) = self.parse_message_reject(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 10. 配置警告和资源限制
        if let Some(event) = self.parse_config_warning(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 11. 邮件隔离/保留事件
        if let Some(event) = self.parse_message_hold(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 12. 邮件丢弃事件
        if let Some(event) = self.parse_message_discard(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 13. 邮件移除事件
        if let Some(event) = self.parse_message_removed(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 14. 统计信息
        if let Some(event) = self.parse_statistics(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 15. Snowflake初始化事件
        if let Some(event) = self.parse_snowflake_init(message) {
            return Ok(ComponentEvent::Cleanup(event));
        }

        // 16. 未识别的事件归类为Other
        Ok(ComponentEvent::Cleanup(CleanupEvent::Other {
            event_type: "unknown".to_string(),
            message: message.to_string(),
            queue_id: None, // 尝试从消息中提取队列ID
        }))
    }

    fn component_name(&self) -> &'static str {
        "cleanup"
    }
}

impl Default for CleanupParser {
    fn default() -> Self {
        Self::new()
    }
}