AIOps监控实践笔记

AIOps Prometheus 监控告警 自愈机制
作者: 醉月思 发布日期: 2024-03-15 分类: 技术实践

别急着给AIOps监控下定义，先看这次卡在哪。

这次折腾，从传统告警到AIOps自愈，大概走了半年。

传统告警的尴尬

告警泛滥的问题

最开始用的是 Prometheus + Alertmanager，配置了一个基础告警规则：

# prometheus/rules/node_alerts.yml
groups:
- name: node_alerts
  rules:
  - alert: HighMemoryUsage
    expr: (node_memory_MemTotal_bytes - node_memory_MemAvailable_bytes) / node_memory_MemTotal_bytes > 0.85
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "High memory usage on {{ $labels.instance }}"
      description: "Memory usage is {{ $value | humanizePercentage }}"

跑了一段时间，发现问题来了：

  • 告警太频繁：内存使用率85%就告警，实际上有些节点就是长期这么高
  • 误报太多：偶发尖峰触发告警，过几分钟就自己恢复了
  • 通知疲劳：研发团队收到太多告警，最后直接忽略了
  • 响应慢：真正出问题时，大家已经对告警麻木了

第一个坑：阈值告警的局限性

尝试调整阈值，从85%调到90%，结果漏掉了真正的内存泄漏问题。

后来才知道，单靠固定阈值根本不够。不同业务的节点，内存模式不一样；同一个节点，不同时间段、不同业务场景，正常的内存使用率也不一样。

智能告警的探索

基于基线的动态阈值

换了个思路，不用固定阈值，而是用历史数据建立基线：

# 基于移动平均的动态阈值
import numpy as np
from prometheus_api_client import PrometheusConnect

def get_dynamic_threshold(metric, days=7):
    """
    获取基于历史数据的动态阈值
    """
    prom = PrometheusConnect(url="http://prometheus:9090", disable_ssl=True)

    # 获取过去7天的数据
    query = f'{metric}[{days}d:5m]'
    result = prom.custom_query(query)

    if not result or not result[0]['values']:
        return None

    values = [float(v[1]) for v in result[0]['values']]

    # 计算移动平均
    window_size = 12  # 1小时
    ma = np.convolve(values, np.ones(window_size)/window_size, mode='valid')

    # 计算2σ区间
    std = np.std(ma[-24:])  # 最近2小时的标准差
    mean = np.mean(ma[-24:])

    return {
        'baseline': mean,
        'upper_threshold': mean + 2 * std,
        'lower_threshold': max(0, mean - 2 * std),
        'current': values[-1]
    }

# 使用示例
threshold = get_dynamic_threshold('node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes')
if threshold and threshold['current'] < threshold['lower_threshold']:
    # 触发告警
    print(f"内存使用异常: 当前 {threshold['current']:.2f}, 阈值 {threshold['lower_threshold']:.2f}")

这样做的效果明显改善，但也有问题：

  • 冷启动问题：新节点没历史数据怎么办？
  • 周期性变化：周一早上的高峰和周末不同
  • 突发场景：促销活动、新功能上线，基线会失真

异常检测：从规则到机器学习

后来尝试了 isolation forest 异常检测：

# anomaly_detection.py
from sklearn.ensemble import IsolationForest
import numpy as np
from prometheus_api_client import PrometheusConnect
import joblib

class MetricsAnomalyDetector:
    def __init__(self, model_path='models/anomaly_detector.pkl'):
        self.model = IsolationForest(
            contamination=0.1,
            n_estimators=100,
            max_samples='auto',
            random_state=42
        )
        self.model_path = model_path
        self.fitted = False

    def train(self, X):
        """
        训练异常检测模型
        X: 历史指标数据，shape (n_samples, n_features)
        """
        self.model.fit(X)
        self.fitted = True
        joblib.dump(self.model, self.model_path)

    def predict(self, X):
        """
        预测是否异常
        返回: 1 (正常) 或 -1 (异常)
        """
        if not self.fitted:
            self.load_model()
        return self.model.predict(X)

    def load_model(self):
        """加载训练好的模型"""
        try:
            self.model = joblib.load(self.model_path)
            self.fitted = True
        except FileNotFoundError:
            print("模型文件不存在，需要先训练")

# 使用示例
detector = MetricsAnomalyDetector()

# 收集多维度指标作为特征
def collect_metrics(instance):
    prom = PrometheusConnect(url="http://prometheus:9090", disable_ssl=True)

    queries = {
        'cpu': f'rate(node_cpu_seconds_total{{instance="{instance}", mode!="idle"}}[5m])',
        'memory': f'(1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)',
        'disk_io': f'rate(node_disk_io_time_seconds_total{{instance="{instance}"}}[5m])',
        'network': f'rate(node_network_receive_bytes_total{{instance="{instance}"}}[5m])'
    }

    features = []
    for name, query in queries.items():
        result = prom.custom_query(query)
        if result:
            features.append(float(result[0]['value'][1]))
        else:
            features.append(0)

    return np.array(features).reshape(1, -1)

# 训练模型（基于正常历史数据）
# 假设我们已经收集了30天的正常数据
# X_normal = collect_training_data()
# detector.train(X_normal)

# 实时检测
# metrics = collect_metrics('192.168.1.100:9100')
# prediction = detector.predict(metrics)
# if prediction[0] == -1:
#     print("检测到异常行为")

这里踩了个坑：训练数据的质量直接影响检测效果。如果训练数据里本身就包含了异常，模型就会学到错误的"正常模式"。

后来加了数据预处理步骤：

from scipy import signal

def preprocess_metrics(raw_data):
    """
    预处理指标数据
    1. 平滑处理
    2. 去除噪点
    3. 归一化
    """
    # 使用Savitzky-Golay滤波器平滑
    window_length = 11
    polyorder = 3
    smoothed = signal.savgol_filter(raw_data, window_length, polyorder)

    # 去除异常值（使用IQR方法）
    Q1 = np.percentile(smoothed, 25)
    Q3 = np.percentile(smoothed, 75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR

    cleaned = np.clip(smoothed, lower_bound, upper_bound)

    # 归一化
    min_val = np.min(cleaned)
    max_val = np.max(cleaned)
    normalized = (cleaned - min_val) / (max_val - min_val)

    return normalized

从告警到自愈

告警分类与优先级

不同问题需要不同处理方式，不能一概而论。做了个简单的分类：

# alertmanager/config/classification.yml
receivers:
- name: 'critical-alerts'
  webhook_configs:
  - url: 'http://auto-healer:8080/critical'
    send_resolved: true

- name: 'warning-alerts'
  webhook_configs:
  - url: 'http://auto-healer:8080/warning'
    send_resolved: true

- name: 'info-alerts'
  webhook_configs:
  - url: 'http://auto-healer:8080/info'

route:
  group_by: ['alertname', 'cluster']
  group_wait: 10s
  group_interval: 5m
  repeat_interval: 12h

  receiver: 'warning-alerts'

  routes:
  - match:
      severity: critical
    receiver: 'critical-alerts'
    continue: false

  - match:
      severity: warning
    receiver: 'warning-alerts'

自愈机制：从脚本到编排

最开始写了个简单的自愈脚本，处理常见问题：

#!/bin/bash
# auto_healer.sh - 简单的自愈脚本

ALERT_TYPE=$1
SERVICE_NAME=$2
INSTANCE=$3

case $ALERT_TYPE in
  "high_memory")
    # 尝试重启服务
    echo "检测到高内存使用，尝试重启 $SERVICE_NAME"
    systemctl restart $SERVICE_NAME
    ;;

  "high_disk")
    # 清理日志文件
    echo "检测到磁盘空间不足，清理日志文件"
    find /var/log/$SERVICE_NAME -name "*.log" -mtime +7 -delete
    ;;

  "high_cpu")
    # 检查是否有僵尸进程
    echo "检测到CPU使用率过高，检查进程状态"
    ps aux | grep $SERVICE_NAME | grep -v grep
    ;;

  "service_down")
    # 启动服务
    echo "检测到服务 $SERVICE_NAME 停止，尝试启动"
    systemctl start $SERVICE_NAME
    ;;

  *)
    echo "未知告警类型: $ALERT_TYPE"
    exit 1
    ;;
esac

这个脚本太简单了，处理不了复杂场景。后来用了 Ansible Playbook：

# auto_healer/playbook/heal_service.yml
---
- name: Auto healing playbook
  hosts: "{{ target_instance }}"
  become: yes
  vars:
    service_name: "{{ service }}"
    alert_type: "{{ alert }}"
    max_retries: 3

  tasks:
  - name: Check service status
    systemd:
      name: "{{ service_name }}"
    register: service_status
    ignore_errors: yes

  - name: Heal based on alert type
    block:
      - name: Restart service (memory issue)
        systemd:
          name: "{{ service_name }}"
          state: restarted
        when: alert_type == "high_memory"

      - name: Clean up log files (disk issue)
        shell: |
          find /var/log/{{ service_name }} -name "*.log" -mtime +7 -delete
          find /var/log/{{ service_name }} -name "*.log.*" -mtime +7 -delete
        when: alert_type == "high_disk"

      - name: Kill zombie processes (CPU issue)
        shell: |
          pkill -9 -f "{{ service_name }}: defunct"
        when: alert_type == "high_cpu"
        ignore_errors: yes

      - name: Wait for service recovery
        systemd:
          name: "{{ service_name }}"
          state: started
        retries: "{{ max_retries }}"
        delay: 5
        register: recovery_result
        until: recovery_result is succeeded

    rescue:
      - name: Log failure
        debug:
          msg: "Auto heal failed for {{ service_name }} on {{ inventory_hostname }}"

      - name: Notify on-call engineer
        uri:
          url: "http://alertmanager:9093/api/v1/alerts"
          method: POST
          body_format: json
          body:
            - labels:
                alertname: "auto_heal_failed"
                service: "{{ service_name }}"
                instance: "{{ inventory_hostname }}"
                severity: critical
              annotations:
                summary: "Auto heal failed for {{ service_name }}"
                description: "Failed to recover from {{ alert_type }}"

智能决策引擎

同一个问题，不同场景下处理方式可能不一样。比如内存高，可能是：

  • 内存泄漏 → 需要重启服务
  • 突发流量 → 需要扩容
  • 配置问题 → 需要调整参数

搞了个简单的决策引擎：

# decision_engine.py
class HealingDecisionEngine:
    def __init__(self):
        self.rules = {
            'memory_leak': {
                'conditions': [
                    lambda metrics: metrics['memory'] > 0.9,
                    lambda metrics: metrics['memory_trend'] > 0.01,  # 持续上升
                    lambda metrics: metrics['service_uptime'] > 3600  # 运行超过1小时
                ],
                'action': 'restart_service'
            },
            'traffic_spike': {
                'conditions': [
                    lambda metrics: metrics['memory'] > 0.85,
                    lambda metrics: metrics['request_rate'] > 1000,
                    lambda metrics: metrics['response_time'] < 0.5  # 响应时间还好
                ],
                'action': 'scale_up'
            },
            'resource_exhaustion': {
                'conditions': [
                    lambda metrics: metrics['memory'] > 0.95,
                    lambda metrics: metrics['disk_io'] > 0.8,
                    lambda metrics: metrics['cpu'] > 0.9
                ],
                'action': 'scale_up_plus'
            }
        }

    def decide(self, metrics):
        """
        基于当前指标决策自愈策略
        """
        for pattern, config in self.rules.items():
            if all(condition(metrics) for condition in config['conditions']):
                return {
                    'pattern': pattern,
                    'action': config['action'],
                    'confidence': self._calculate_confidence(metrics, config['conditions'])
                }

        # 没有匹配到已知模式，返回默认策略
        return {
            'pattern': 'unknown',
            'action': 'manual_intervention',
            'confidence': 0
        }

    def _calculate_confidence(self, metrics, conditions):
        """
        计算决策置信度
        """
        # 简单实现：满足条件的越多，置信度越高
        satisfied = sum(1 for condition in conditions if condition(metrics))
        return satisfied / len(conditions)

# 使用示例
engine = HealingDecisionEngine()

# 收集多维度指标
metrics = {
    'memory': 0.92,
    'memory_trend': 0.02,
    'service_uptime': 7200,
    'request_rate': 800,
    'response_time': 0.3,
    'disk_io': 0.6,
    'cpu': 0.85
}

decision = engine.decide(metrics)
print(f"决策: {decision}")

踩过的坑

坑一：误触发自愈

有一次，自动扩容被触发，但其实是监控指标本身有问题。本来只是Prometheus采集延迟，被当成了业务负载。

解决：加了数据质量检查和熔断机制：

def check_data_quality(metrics):
    """
    检查数据质量
    """
    # 检查数据时效性
    if time.time() - metrics['timestamp'] > 60:
        return False, "数据过旧"

    # 检查数值合理性
    if not (0 <= metrics['memory'] <= 1):
        return False, "内存使用率超出合理范围"

    # 检查变化率
    if abs(metrics['memory_delta']) > 0.5:
        return False, "指标变化异常"

    return True, "数据质量正常"

class CircuitBreaker:
    def __init__(self, failure_threshold=3, timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.last_failure_time = 0
        self.state = 'closed'  # closed, open, half-open

    def allow_request(self):
        if self.state == 'open':
            if time.time() - self.last_failure_time > self.timeout:
                self.state = 'half-open'
                return True
            return False

        return True

    def record_success(self):
        self.failure_count = 0
        self.state = 'closed'

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = 'open'

坑二：自愈失败没回滚

有一次，自动重启服务后，服务没起来，但系统又没检测到失败，导致服务持续不可用。

解决：加了自愈结果验证和回滚机制：

# auto_healer/playbook/heal_with_rollback.yml
---
- name: Heal with rollback
  hosts: "{{ target_instance }}"
  become: yes
  tasks:
  - name: Take snapshot before action
    shell: |
      systemctl status {{ service_name }} > /tmp/pre_heal_snapshot.txt
      ps aux | grep {{ service_name }} >> /tmp/pre_heal_snapshot.txt
    register: snapshot
    when: action_type != "manual"

  - name: Execute healing action
    block:
      - name: Do the healing
        include_tasks: "tasks/{{ action_type }}.yml"

      - name: Verify healing result
        uri:
          url: "http://{{ target_instance }}:{{ health_check_port }}/health"
          method: GET
          status_code: 200
        register: health_check
        retries: 5
        delay: 10
        until: health_check.status == 200

    rescue:
      - name: Rollback on failure
        block:
          - name: Restore previous state
            shell: |
              systemctl stop {{ service_name }}
              # 恢复配置文件
              cp /etc/{{ service_name }}/config.backup /etc/{{ service_name }}/config.yml
              systemctl start {{ service_name }}
            ignore_errors: yes

          - name: Notify team
            debug:
              msg: "Healing failed, rolled back. Manual intervention required."

        always:
          - name: Log incident
            uri:
              url: "http://incident-logger:8080/log"
              method: POST
              body_format: json
              body:
                service: "{{ service_name }}"
                action: "{{ action_type }}"
                status: "failed"
                instance: "{{ target_instance }}"
                timestamp: "{{ ansible_date_time.iso8601 }}"

坑三：自愈变成自动化运维

有些团队把自愈当成了"自动运维"，什么都想自动化，结果系统越来越复杂，没人知道什么情况下会触发什么动作。

解决：明确自愈的边界，只处理以下场景：

# 自愈场景白名单
AUTO_HEAL_SCENARIOS = {
    'service_restart': {
        'max_frequency': '3次/小时',
        'conditions': ['memory > 90%', 'service_unresponsive'],
        'require_approval': False
    },
    'log_cleanup': {
        'max_frequency': '1次/天',
        'conditions': ['disk_usage > 85%'],
        'require_approval': False
    },
    'config_rollback': {
        'max_frequency': '1次/天',
        'conditions': ['recent_config_change', 'error_rate > 5%'],
        'require_approval': True  # 需要审批
    },
    'scale_up': {
        'max_frequency': '5次/天',
        'conditions': ['cpu > 80%', 'request_rate > threshold'],
        'require_approval': False
    }
}

# 需要人工干预的场景
MANUAL_INTERVENTION_REQUIRED = [
    'database_connection',
    'security_incident',
    'data_corruption',
    'cross_service_dependency'
]

实际效果

半年时间，从传统告警到AIOps自愈，效果还是明显的：

指标改善：

  • 平均故障恢复时间（MTTR）：从45分钟降到12分钟
  • 告警准确率：从60%提升到85%
  • 误报率：从40%降到15%
  • 人工介入次数：减少了65%

团队感受：

  • 凌晨被叫醒的次数少了
  • 告警不再被无视了
  • 自愈机制给了团队信心，但也没敢完全依赖它

架构示意

整个AIOps监控自愈系统大致这样：

graph TB A[应用服务] --> B[Prometheus] A --> C[应用日志] A --> D[业务指标] B --> E[告警规则] E --> F[Alertmanager] C --> G[日志分析] D --> H[业务监控] B --> I[异常检测] I --> J[决策引擎] F --> J G --> J H --> J J --> K[自动编排] K --> L[Ansible Playbook] L --> M[自愈动作] J --> N[人工决策] N --> O[工单系统] M --> P[恢复验证] P --> Q{成功?} Q -->|是| R[完成] Q -->|否| S[回滚] S --> N style J fill:#90EE90 style K fill:#87CEEB style N fill:#FFB6C1

写在最后

AIOps不是什么魔法，更多是经验和规则的工程化。

它解决的是：

  • 重复性工作自动化
  • 简单场景快速响应
  • 知识经验固化沉淀

它解决不了的是：

  • 根本性技术问题
  • 架构设计缺陷
  • 业务理解缺失

实际使用时，建议：

  1. 先把基础监控做好，指标齐全才有分析的基础
  2. 从小场景开始自愈，验证了再扩展
  3. 保留人工审批通道，特别是生产环境的关键动作
  4. 自愈失败时要有明确的回滚机制和通知路径

凌晨三点被电话叫醒，这感觉谁都不想再经历。但自愈机制更像是"应急包"，真正要少加班，还是得从架构设计、代码质量、容量规划这些地方下功夫。

这次AIOps监控实践用到了Prometheus 2.45、Grafana 10.1、Ansible 2.15，以及自研的异常检测和决策引擎。整个过程断断续续搞了半年，中间几次想放弃，觉得"还不如人工处理"，但熬过阵痛期后，确实省了不少事。

版权声明: 本文首发于 指尖魔法屋-AIOps监控实践笔记（https://blog.thinkmoon.cn/post/147-aiops-monitoring-alerting-self-healing-practice/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章

AI部署最佳实践：这次怎么落地的
回想这几年在 AI 项目上的折腾，真正让我在夜深人静时抱着电脑叹气的，从来不是模型推理本身——那通常是技术方案中最确定的一环——而是那些为了把一…
📁 技术实践
📅 2023-04-15