监控与告警场景下数据治理技术选型：替代方案与决策框架

MonitorGovernanceDemo/
├── config/
│   └── settings.yaml          # 主配置文件
├── src/
│   ├── __init__.py
│   ├── data_generator.py      # 模拟监控指标数据生成
│   ├── datastore/
│   │   ├── __init__.py
│   │   ├── base.py            # 抽象存储接口
│   │   ├── influxdb_client.py # InfluxDB实现
│   │   └── timescaledb_client.py # TimescaleDB实现
│   ├── rule_engine/
│   │   ├── __init__.py
│   │   ├── engine.py          # 规则引擎核心
│   │   └── models.py          # 规则数据模型
│   └── app.py                 # 主应用入口，模拟数据管道
├── requirements.txt           # Python依赖
├── init_database.py           # (可选)TimescaleDB表初始化脚本
└── run.py                     # 项目启动脚本

# 数据存储配置 - 通过切换`active_store`来选型
datastore:
  active_store: "timescaledb" # 可选: "influxdb", "timescaledb", "mock"
  stores:
    influxdb:
      url: "http://localhost:8086"
      token: "your_influxdb_token_here" # InfluxDB 2.x使用token
      org: "your_org"
      bucket: "monitoring_bucket"
      timeout: 10000
    timescaledb:
      host: "localhost"
      port: 5432
      database: "monitoring_db"
      user: "postgres"
      password: "password"
      sslmode: "disable"

# 监控指标定义
metrics:

  - name: "cpu_usage"
    tags: ["host", "region"]
    fields: ["usage_percent"]
    interval_sec: 60 # 模拟数据上报间隔

  - name: "memory_usage"
    tags: ["host", "application"]
    fields: ["used_gb", "total_gb"]

# 告警规则定义
rules:

  - id: "high_cpu_rule"
    metric: "cpu_usage"
    condition: "usage_percent > 80"
    duration: "5m" # 持续5分钟超过阈值才告警
    severity: "warning"
    notify_channels: ["console", "email"]

from abc import ABC, abstractmethod
from typing import Dict, List, Any, Optional
import pandas as pd

class DataStore(ABC):
    """抽象数据存储接口，定义监控数据治理的核心操作。"""

    @abstractmethod
    def write_metric(self, metric_name: str, tags: Dict[str, str], fields: Dict[str, float], timestamp: Optional[int] = None):
        """写入一个监控数据点。"""
        pass

    @abstractmethod
    def query_metric(self, metric_name: str, start_time: int, end_time: int, filters: Dict[str, str] = None) -> pd.DataFrame:
        """查询指定时间范围内的监控数据。"""
        pass

    @abstractmethod
    def aggregate_metric(self, metric_name: str, start_time: int, end_time: int, aggregation: str, window: str, group_by_tags: List[str] = None) -> pd.DataFrame:
        """
        对监控数据进行聚合查询（如5分钟平均）。
        aggregation: avg, max, min, count
        window: 时间窗口，如 '5m', '1h'
        """
        pass

    @abstractmethod
    def health_check(self) -> bool:
        """检查存储后端是否健康。"""
        pass

import psycopg2
from psycopg2.extras import RealDictCursor
import pandas as pd
from .base import DataStore
import time
from typing import Dict, List, Any, Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class TimescaleDBClient(DataStore):
    """TimescaleDB (基于PostgreSQL) 存储实现。"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.conn = None
        self._connect()

    def _connect(self):
        try:
            self.conn = psycopg2.connect(
                host=self.config['host'],
                port=self.config['port'],
                database=self.config['database'],
                user=self.config['user'],
                password=self.config['password']
            )
            logger.info("Connected to TimescaleDB successfully.")
        except Exception as e:
            logger.error(f"Failed to connect to TimescaleDB: {e}")
            raise

    def write_metric(self, metric_name: str, tags: Dict[str, str], fields: Dict[str, float], timestamp: Optional[int] = None):
        """写入数据点。假设表名即metric_name，表结构已提前创建。"""
        if timestamp is None:
            timestamp = int(time.time() * 1000)  # 毫秒时间戳

        tag_keys = ', '.join(tags.keys())
        tag_placeholders = ', '.join([f"%s"] * len(tags))
        field_keys = ', '.join(fields.keys())
        field_placeholders = ', '.join([f"%s"] * len(fields))

        # 构建动态SQL - 生产环境需防注入，这里为演示简化
        # 实际应在项目初始化时确保表存在
        sql = f"""
        INSERT INTO {metric_name} (time, {tag_keys}, {field_keys})
        VALUES (to_timestamp(%s/1000.0), {tag_placeholders}, {field_placeholders});
        """
        values = [timestamp] + list(tags.values()) + list(fields.values())

        with self.conn.cursor() as cur:
            cur.execute(sql, values)
            self.conn.commit()

    def query_metric(self, metric_name: str, start_time: int, end_time: int, filters: Dict[str, str] = None) -> pd.DataFrame:
        where_clauses = ["time >= to_timestamp(%s/1000.0)", "time <= to_timestamp(%s/1000.0)"]
        values = [start_time, end_time]

        if filters:
            for key, value in filters.items():
                where_clauses.append(f"{key} = %s")
                values.append(value)

        where_sql = " AND ".join(where_clauses)
        sql = f"SELECT * FROM {metric_name} WHERE {where_sql} ORDER BY time ASC;"

        with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(sql, values)
            rows = cur.fetchall()
            return pd.DataFrame(rows)

    def aggregate_metric(self, metric_name: str, start_time: int, end_time: int, aggregation: str, window: str, group_by_tags: List[str] = None) -> pd.DataFrame:
        # 简化实现：演示TimescaleDB的time_bucket聚合函数
        group_by_clause = ""
        if group_by_tags:
            group_by_clause = ", " + ", ".join(group_by_tags)

        # 假设我们只聚合第一个数值型字段（实际需根据配置）
        field = 'usage_percent' if metric_name == 'cpu_usage' else 'used_gb'
        sql = f"""
        SELECT
            time_bucket('%s', time) AS bucket,
            {aggregation}({field}) AS value
            {group_by_clause}
        FROM {metric_name}
        WHERE time >= to_timestamp(%s/1000.0) AND time <= to_timestamp(%s/1000.0)
        GROUP BY bucket {group_by_clause}
        ORDER BY bucket;
        """
        # 窗口转换，如 '5m' -> '5 minutes'
        window_pg = window[:-1] + ' minutes' if window.endswith('m') else window
        values = [window_pg, start_time, end_time]

        with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(sql, values)
            rows = cur.fetchall()
            return pd.DataFrame(rows)

    def health_check(self) -> bool:
        try:
            with self.conn.cursor() as cur:
                cur.execute("SELECT 1;")
                return True
        except Exception as e:
            logger.error(f"TimescaleDB health check failed: {e}")
            return False

from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
import pandas as pd
from .base import DataStore
import time
from typing import Dict, List, Any, Optional
import logging

logger = logging.getLogger(__name__)

class InfluxDBClientWrapper(DataStore):
    """InfluxDB 2.x 存储实现。"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.client = InfluxDBClient(
            url=config['url'],
            token=config['token'],
            org=config['org'],
            timeout=config.get('timeout', 10000)
        )
        self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
        self.query_api = self.client.query_api()
        self.bucket = config['bucket']

    def write_metric(self, metric_name: str, tags: Dict[str, str], fields: Dict[str, float], timestamp: Optional[int] = None):
        point = Point(metric_name)
        for tag_key, tag_value in tags.items():
            point.tag(tag_key, tag_value)
        for field_key, field_value in fields.items():
            point.field(field_key, field_value)
        if timestamp:
            point.time(timestamp, WritePrecision.MS)
        else:
            point.time(int(time.time() * 1e9), WritePrecision.NS)  # InfluxDB默认纳秒
        self.write_api.write(bucket=self.bucket, record=point)

    def query_metric(self, metric_name: str, start_time: int, end_time: int, filters: Dict[str, str] = None) -> pd.DataFrame:
        # 构建Flux查询语句
        filter_str = ""
        if filters:
            filter_str = " and " + " and ".join([f'r["{k}"] == "{v}"' for k, v in filters.items()])

        flux_query = f'''
        from(bucket:"{self.bucket}")
          |> range(start: {start_time}, stop: {end_time})
          |> filter(fn: (r) => r["_measurement"] == "{metric_name}"{filter_str})
          |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
          |> sort(columns: ["_time"])
        '''
        try:
            result = self.query_api.query_data_frame(flux_query)
            # 处理返回的DataFrame，可能是一个或多个
            if isinstance(result, list):
                df = pd.concat(result, ignore_index=True)
            else:
                df = result
            # 重命名列，去除下划线前缀
            df.rename(columns={"_time": "time", "_measurement": "measurement"}, inplace=True, errors='ignore')
            return df
        except Exception as e:
            logger.error(f"InfluxDB query failed: {e}, query: {flux_query}")
            return pd.DataFrame()

    def aggregate_metric(self, metric_name: str, start_time: int, end_time: int, aggregation: str, window: str, group_by_tags: List[str] = None) -> pd.DataFrame:
        group_by_clause = ""
        if group_by_tags:
            # Flux语法：group_by参数是一个列表
            group_by_fields = ["_time"] + group_by_tags
            group_by_clause = f'|> group(columns: {group_by_fields})'
        else:
            group_by_clause = ''

        # 简化：只查询第一个字段。实际中需要更复杂的逻辑确定字段。
        flux_query = f'''
        from(bucket:"{self.bucket}")
          |> range(start: {start_time}, stop: {end_time})
          |> filter(fn: (r) => r["_measurement"] == "{metric_name}")
          |> filter(fn: (r) => r["_field"] == "usage_percent" or r["_field"] == "used_gb")
          |> aggregateWindow(every: {window}, fn: {aggregation})
          {group_by_clause}
          |> yield(name: "aggregated")
        '''
        try:
            result = self.query_api.query_data_frame(flux_query)
            if isinstance(result, list):
                df = pd.concat(result, ignore_index=True)
            else:
                df = result
            return df
        except Exception as e:
            logger.error(f"InfluxDB aggregation query failed: {e}")
            return pd.DataFrame()

    def health_check(self) -> bool:
        try:
            ready = self.client.ping()
            return ready
        except Exception as e:
            logger.error(f"InfluxDB health check failed: {e}")
            return False

from pydantic import BaseModel, validator
from typing import Dict, List, Any
import re

class AlertRule(BaseModel):
    """告警规则数据模型。"""
    id: str
    metric: str
    condition: str  # 简单表达式，如 "usage_percent > 80"
    duration: str   # 持续时间，如 "5m"
    severity: str   # warning, error, critical
    notify_channels: List[str]

    @validator('condition')
    def validate_condition(cls, v):
        # 简单的安全性/格式校验
        if not re.match(r'^[a-zA-Z0-9_\.\s><=!]+$', v):
            raise ValueError('Condition contains invalid characters')
        return v

import pandas as pd
import time
from typing import Dict, List, Any
from .models import AlertRule
import logging
import re

logger = logging.getLogger(__name__)

class RuleEngine:
    """简单的阈值规则评估引擎。"""

    def __init__(self, datastore):
        self.datastore = datastore
        self.rules: Dict[str, AlertRule] = {}
        self.alert_history = []  # 内存中记录告警历史，生产环境应持久化

    def load_rules(self, rules_config: List[Dict[str, Any]]):
        """从配置加载规则。"""
        self.rules.clear()
        for rule_config in rules_config:
            rule = AlertRule(**rule_config)
            self.rules[rule.id] = rule
        logger.info(f"Loaded {len(self.rules)} alert rules.")

    def evaluate_rule(self, rule_id: str) -> Dict[str, Any]:
        """评估单个规则。"""
        if rule_id not in self.rules:
            return {"triggered": False, "message": f"Rule {rule_id} not found."}

        rule = self.rules[rule_id]
        now = int(time.time() * 1000)
        # 将持续时间（如'5m'）转换为毫秒
        duration_ms = self._parse_duration(rule.duration)
        start_time = now - duration_ms

        # 查询近期数据
        try:
            df = self.datastore.query_metric(rule.metric, start_time, now)
            if df.empty:
                return {"triggered": False, "message": "No data in evaluation window."}

            # 解析条件表达式并评估（简化版：假设条件针对DataFrame的某一列）
            # 例如： "usage_percent > 80"
            field, op, value_str = re.split(r'\s+', rule.condition.strip())
            value = float(value_str)
            triggered_series = None

            if op == '>':
                triggered_series = df[field] > value
            elif op == '>=':
                triggered_series = df[field] >= value
            elif op == '<':
                triggered_series = df[field] < value
            elif op == '<=':
                triggered_series = df[field] <= value
            elif op == '==':
                triggered_series = df[field] == value
            else:
                return {"triggered": False, "message": f"Unsupported operator: {op}"}

            # 检查是否在整个持续时间内都满足条件（简化：检查最后N个点是否都触发）
            # 实际生产环境需要更复杂的窗口和状态保持逻辑。
            if triggered_series.any():
                latest_data = df.iloc[-1]
                alert_message = f"Alert {rule.id} triggered! {rule.condition}. Current value: {latest_data[field]}. Metric: {rule.metric}"
                alert_record = {
                    "rule_id": rule.id,
                    "timestamp": now,
                    "message": alert_message,
                    "severity": rule.severity,
                    "data_point": latest_data.to_dict()
                }
                self.alert_history.append(alert_record)
                logger.warning(alert_message)
                return {"triggered": True, "alert": alert_record}
            else:
                return {"triggered": False, "message": "Condition not met."}

        except Exception as e:
            logger.error(f"Error evaluating rule {rule_id}: {e}")
            return {"triggered": False, "message": f"Evaluation error: {e}"}

    def evaluate_all(self) -> List[Dict[str, Any]]:
        """评估所有规则。"""
        results = []
        for rule_id in self.rules:
            result = self.evaluate_rule(rule_id)
            results.append(result)
        return results

    @staticmethod
    def _parse_duration(duration_str: str) -> int:
        """将'5m', '1h'等字符串转换为毫秒数。"""
        unit = duration_str[-1]
        num = int(duration_str[:-1])
        if unit == 's':
            return num * 1000
        elif unit == 'm':
            return num * 60 * 1000
        elif unit == 'h':
            return num * 60 * 60 * 1000
        else:
            raise ValueError(f"Unsupported duration unit: {unit}")

import random
import time
from typing import Dict, List, Any
import logging

logger = logging.getLogger(__name__)

class MetricDataGenerator:
    """生成模拟的监控指标数据。"""

    def __init__(self, metrics_config: List[Dict[str, Any]]):
        self.metrics_config = metrics_config
        self.hosts = ["web-server-01", "web-server-02", "db-server-01"]
        self.regions = ["us-east-1", "eu-west-1", "ap-southeast-2"]
        self.applications = ["frontend", "backend", "database"]

    def generate_data_point(self, metric_name: str) -> Dict[str, Any]:
        """为指定指标生成一个数据点。"""
        metric_config = next((m for m in self.metrics_config if m['name'] == metric_name), None)
        if not metric_config:
            raise ValueError(f"Metric {metric_name} not configured.")

        tags = {}
        for tag in metric_config['tags']:
            if tag == 'host':
                tags[tag] = random.choice(self.hosts)
            elif tag == 'region':
                tags[tag] = random.choice(self.regions)
            elif tag == 'application':
                tags[tag] = random.choice(self.applications)
            else:
                tags[tag] = f"tag-{random.randint(1,3)}"

        fields = {}
        for field in metric_config['fields']:
            if metric_name == "cpu_usage" and field == "usage_percent":
                # 模拟CPU使用率，偶尔产生高负载
                fields[field] = random.uniform(10, 95) if random.random() > 0.1 else random.uniform(80, 99)
            elif metric_name == "memory_usage":
                if field == "total_gb":
                    fields[field] = 16.0
                elif field == "used_gb":
                    fields[field] = random.uniform(4.0, 15.5)
            else:
                fields[field] = random.uniform(0, 100)

        return {"metric_name": metric_name, "tags": tags, "fields": fields, "timestamp": int(time.time() * 1000)}

    def generate_batch(self, points_per_metric: int = 5) -> List[Dict[str, Any]]:
        """为所有指标生成一批数据点。"""
        batch = []
        for metric_config in self.metrics_config:
            for _ in range(points_per_metric):
                # 模拟数据点的时间轻微偏移
                time.sleep(0.001)
                batch.append(self.generate_data_point(metric_config['name']))
        logger.info(f"Generated {len(batch)} data points.")
        return batch

import yaml
import time
import logging
from pathlib import Path
from typing import Dict, Any

from .datastore.influxdb_client import InfluxDBClientWrapper
from .datastore.timescaledb_client import TimescaleDBClient
from .datastore.base import DataStore
from .data_generator import MetricDataGenerator
from .rule_engine.engine import RuleEngine

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class MonitoringGovernanceDemo:
    """主应用类，编排整个数据治理演示流程。"""

    def __init__(self, config_path: str):
        self.config = self._load_config(config_path)
        self.datastore: DataStore = self._init_datastore()
        self.data_generator = MetricDataGenerator(self.config['metrics'])
        self.rule_engine = RuleEngine(self.datastore)
        self.rule_engine.load_rules(self.config['rules'])

    @staticmethod
    def _load_config(config_path: str) -> Dict[str, Any]:
        with open(config_path, 'r') as f:
            return yaml.safe_load(f)

    def _init_datastore(self) -> DataStore:
        active_store = self.config['datastore']['active_store']
        store_config = self.config['datastore']['stores'].get(active_store)

        if not store_config:
            raise ValueError(f"Configuration for active store '{active_store}' not found.")

        logger.info(f"Initializing '{active_store}' as the active data store.")

        if active_store == "influxdb":
            return InfluxDBClientWrapper(store_config)
        elif active_store == "timescaledb":
            return TimescaleDBClient(store_config)
        elif active_store == "mock":
            # 可以在这里实现一个模拟存储，用于测试
            from .datastore.mock_client import MockDataStore
            return MockDataStore(store_config)
        else:
            raise ValueError(f"Unsupported data store type: {active_store}")

    def run_pipeline(self, cycles: int = 3):
        """运行模拟的数据管道：生成数据 -> 写入 -> 查询 -> 评估告警。"""
        logger.info("Starting monitoring data governance pipeline...")
        # 1. 健康检查
        if not self.datastore.health_check():
            logger.error("Data store health check failed. Exiting.")
            return

        for cycle in range(1, cycles + 1):
            logger.info(f"--- Pipeline Cycle {cycle} ---")
            # 2. 生成并写入数据
            batch = self.data_generator.generate_batch(points_per_metric=2)
            for point in batch:
                self.datastore.write_metric(
                    point['metric_name'],
                    point['tags'],
                    point['fields'],
                    point['timestamp']
                )
            logger.info(f"Cycle {cycle}: Data written.")

            # 3. 等待一小段时间，模拟实时场景
            time.sleep(2)

            # 4. 执行一个简单的聚合查询（演示）
            now = int(time.time() * 1000)
            five_min_ago = now - (5 * 60 * 1000)
            try:
                agg_result = self.datastore.aggregate_metric(
                    "cpu_usage", five_min_ago, now, "avg", "2m", ["host"]
                )
                if not agg_result.empty:
                    logger.info(f"Cycle {cycle}: Sample aggregation result:\n{agg_result.head()}")
                else:
                    logger.warning("Aggregation query returned no data.")
            except Exception as e:
                logger.error(f"Aggregation query failed: {e}")

            # 5. 评估所有告警规则
            alert_results = self.rule_engine.evaluate_all()
            triggered_alerts = [r for r in alert_results if r.get('triggered')]
            if triggered_alerts:
                logger.warning(f"Cycle {cycle}: {len(triggered_alerts)} alert(s) triggered.")
            else:
                logger.info(f"Cycle {cycle}: No alerts triggered.")

            time.sleep(1) # 循环间隔

        logger.info("Pipeline finished.")
        # 打印告警历史
        if self.rule_engine.alert_history:
            logger.info("--- Alert History ---")
            for alert in self.rule_engine.alert_history[-5:]: # 显示最后5条
                logger.info(f"[{alert['severity'].upper()}] {alert['message']}")

#!/usr/bin/env python3
import sys
from pathlib import Path

# 添加src目录到Python路径
src_path = Path(__file__).parent / 'src'
sys.path.insert(0, str(src_path))

from src.app import MonitoringGovernanceDemo

if __name__ == "__main__":
    # 默认配置文件路径
    config_file = Path(__file__).parent / "config" / "settings.yaml"
    if not config_file.exists():
        print(f"Error: Config file not found at {config_file}")
        print("Please ensure 'config/settings.yaml' exists and is properly configured.")
        sys.exit(1)

    # 创建并运行演示应用
    demo_app = MonitoringGovernanceDemo(str(config_file))
    try:
        demo_app.run_pipeline(cycles=5)  # 运行5个周期
    except KeyboardInterrupt:
        print("\nPipeline interrupted by user.")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")
        import traceback
        traceback.print_exc()

# 此脚本用于初始化TimescaleDB中的表结构。
import psycopg2
import sys
from pathlib import Path
import yaml

config_path = Path(__file__).parent / "config" / "settings.yaml"
with open(config_path, 'r') as f:
    config = yaml.safe_load(f)

ts_config = config['datastore']['stores']['timescaledb']

try:
    conn = psycopg2.connect(
        host=ts_config['host'],
        port=ts_config['port'],
        database=ts_config['database'],
        user=ts_config['user'],
        password=ts_config['password']
    )
    cur = conn.cursor()

    # 创建CPU使用率表（超表）
    cur.execute("""
    CREATE TABLE IF NOT EXISTS cpu_usage (
        time TIMESTAMPTZ NOT NULL,
        host TEXT,
        region TEXT,
        usage_percent DOUBLE PRECISION
    );
    SELECT create_hypertable('cpu_usage', 'time', if_not_exists => TRUE);
    CREATE INDEX IF NOT EXISTS idx_cpu_host ON cpu_usage (host, time DESC);
    """)

    # 创建内存使用率表
    cur.execute("""
    CREATE TABLE IF NOT EXISTS memory_usage (
        time TIMESTAMPTZ NOT NULL,
        host TEXT,
        application TEXT,
        used_gb DOUBLE PRECISION,
        total_gb DOUBLE PRECISION
    );
    SELECT create_hypertable('memory_usage', 'time', if_not_exists => TRUE);
    """)

    conn.commit()
    print("TimescaleDB tables initialized successfully.")
    cur.close()
    conn.close()
except Exception as e:
    print(f"Failed to initialize database: {e}")
    sys.exit(1)

pip install -r requirements.txt

# 核心依赖
pandas>=1.3.0
pyyaml>=5.4.0
pydantic>=1.8.0
# 时序数据库客户端
influxdb-client>=1.30.0
psycopg2-binary>=2.9.0
# 可选：用于mock模式或更丰富的功能
# numpy>=1.21.0

python run.py

2023-10-27 10:00:00,000 - src.app - INFO - Initializing 'timescaledb' as the active data store.
2023-10-27 10:00:00,100 - src.datastore.timescaledb_client - INFO - Connected to TimescaleDB successfully.
2023-10-27 10:00:00,101 - src.app - INFO - Starting monitoring data governance pipeline...
2023-10-27 10:00:00,102 - src.data_generator - INFO - Generated 4 data points.
2023-10-27 10:00:00,205 - src.app - INFO - Cycle 1: Data written.
2023-10-27 10:00:02,207 - src.app - INFO - Cycle 1: Sample aggregation result:
                      bucket  value       host
0 2023-10-27 09:55:00+00:00   42.5  web-server-01
...
2023-10-27 10:00:02,300 - src.rule_engine.engine - WARNING - Alert high_cpu_rule triggered! usage_percent > 80. Current value: 92.3. Metric: cpu_usage
2023-10-27 10:00:02,301 - src.app - WARNING - Cycle 1: 1 alert(s) triggered.

import sys
sys.path.insert(0, 'src')
import pytest
from src.rule_engine.engine import RuleEngine
from src.rule_engine.models import AlertRule
from unittest.mock import Mock

def test_rule_condition_parsing():
    """测试规则条件解析逻辑。"""
    mock_store = Mock()
    engine = RuleEngine(mock_store)
    duration_ms = engine._parse_duration("5m")
    assert duration_ms == 5 * 60 * 1000
    with pytest.raises(ValueError):
        engine._parse_duration("5x")