Python项目监控怎么配置告警

wen python案例 2026-07-04 192

本文目录导读：

Python项目监控怎么配置告警

基础监控告警方案
使用专业的监控工具
使用Sentry进行错误监控
使用健康检查API
综合告警配置系统
最佳实践建议

我来详细介绍Python项目监控告警的配置方法,包括多种方案和实践建议。

基础监控告警方案

使用Python内置日志 + 邮件告警

import logging
import smtplib
from email.mime.text import MIMEText
from datetime import datetime
class AlertLogger:
    def __init__(self, email_config):
        self.email_config = email_config
        self.setup_logger()
    def setup_logger(self):
        self.logger = logging.getLogger('alert_logger')
        self.logger.setLevel(logging.INFO)
        # 文件日志
        fh = logging.FileHandler('app_monitor.log')
        fh.setLevel(logging.INFO)
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        fh.setFormatter(formatter)
        self.logger.addHandler(fh)
    def send_alert_email(self, subject, message, level='ERROR'):
        """发送告警邮件"""
        if level in ['ERROR', 'CRITICAL']:
            try:
                msg = MIMEText(message)
                msg['Subject'] = f"[{level}] {subject}"
                msg['From'] = self.email_config['from']
                msg['To'] = self.email_config['to']
                with smtplib.SMTP(self.email_config['smtp_server'], 
                                self.email_config['smtp_port']) as server:
                    server.starttls()
                    server.login(self.email_config['username'], 
                               self.email_config['password'])
                    server.send_message(msg)
            except Exception as e:
                self.logger.error(f"发送告警邮件失败: {e}")
    def alert(self, message, level='ERROR'):
        """记录并发送告警"""
        self.logger.log(getattr(logging, level), message)
        self.send_alert_email(
            subject="应用告警",
            message=message,
            level=level
        )
# 使用示例
email_config = {
    'smtp_server': 'smtp.gmail.com',
    'smtp_port': 587,
    'username': 'your-email@gmail.com',
    'password': 'your-password',
    'from': 'your-email@gmail.com',
    'to': 'alert-recipient@example.com'
}
alert_logger = AlertLogger(email_config)
alert_logger.alert("数据库连接失败", 'CRITICAL')

使用专业的监控工具

Prometheus + Grafana方案

# app/metrics.py
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from flask import Flask, Response
import time
app = Flask(__name__)
# 定义指标
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', 
                       ['method', 'endpoint', 'status'])
REQUEST_DURATION = Histogram('http_request_duration_seconds', 
                            'HTTP request duration in seconds',
                            ['method', 'endpoint'])
ACTIVE_USERS = Gauge('active_users', 'Current number of active users')
@app.route('/metrics')
def metrics():
    """暴露Prometheus指标"""
    return Response(generate_latest(), mimetype='text/plain')
@app.before_request
def before_request():
    request.start_time = time.time()
@app.after_request
def after_request(response):
    # 记录请求计数
    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.path,
        status=response.status_code
    ).inc()
    # 记录请求耗时
    duration = time.time() - request.start_time
    REQUEST_DURATION.labels(
        method=request.method,
        endpoint=request.path
    ).observe(duration)
    return response
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000)

Prometheus告警规则配置 (alert.rules.yml)：

groups:
- name: python_app_alerts
  rules:
  - alert: HighErrorRate
    expr: rate(http_requests_total{status=~"5.."}[5m]) > 0.1
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "High error rate detected"
      description: "Error rate is {{ $value }} requests/s for the last 5 minutes"
  - alert: SlowResponseTime
    expr: histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m])) > 2
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Slow response time detected"
      description: "P95 response time is {{ $value }} seconds"
  - alert: LowActiveUsers
    expr: active_users < 10
    for: 10m
    labels:
      severity: info
    annotations:
      summary: "Low active users"
      description: "Active users dropped to {{ $value }}"

Grafana告警配置

{
  "alert": {
    "conditions": [
      {
        "evaluator": {
          "params": [5],
          "type": "gt"
        },
        "operator": {
          "type": "and"
        },
        "query": {
          "params": ["A", "5m", "now"]
        },
        "reducer": {
          "params": [],
          "type": "avg"
        },
        "type": "query"
      }
    ],
    "executionErrorState": "alerting",
    "for": "5m",
    "frequency": "1m",
    "handler": 1,
    "name": "API Response Time Alert",
    "noDataState": "no_data",
    "notifications": [
      {
        "uid": "email_channel"
      }
    ]
  }
}

使用Sentry进行错误监控

# 安装: pip install sentry-sdk
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
sentry_sdk.init(
    dsn="https://your-dsn@sentry.io/project-id",
    integrations=[
        FlaskIntegration(),
        SqlalchemyIntegration(),
    ],
    # 设置发送告警的规则
    traces_sample_rate=1.0,
    # 环境配置
    environment="production",
    # 错误级别配置
    before_send=lambda event, hint: event if event.get('level') in ['error', 'fatal'] else None,
    # 用户信息
    with_locals=True,
)
# 设置告警规则示例
def configure_sentry_alerts():
    """配置Sentry告警"""
    alerts_config = {
        'error_count': {
            'threshold': 10,
            'interval': '1h',
            'channels': ['email', 'slack']
        },
        'error_rate': {
            'threshold': 5.0,  # 百分比
            'interval': '5m',
            'channels': ['pagerduty']
        }
    }
    return alerts_config
# 在代码中使用
try:
    risky_operation()
except Exception as e:
    sentry_sdk.capture_exception(e)
    raise

使用健康检查API

# app/health.py
from flask import Flask, jsonify
import psutil
import platform
from datetime import datetime
app = Flask(__name__)
@app.route('/health')
def health_check():
    """健康检查端点"""
    health_status = {
        'status': 'healthy',
        'timestamp': datetime.now().isoformat(),
        'services': {}
    }
    # 检查数据库连接
    health_status['services']['database'] = check_database()
    # 检查Redis连接
    health_status['services']['redis'] = check_redis()
    # 检查系统资源
    health_status['system'] = {
        'cpu_usage': psutil.cpu_percent(),
        'memory_usage': psutil.virtual_memory().percent,
        'disk_usage': psutil.disk_usage('/').percent
    }
    # 设置整体状态
    if any(s['status'] == 'unhealthy' for s in health_status['services'].values()):
        health_status['status'] = 'unhealthy'
    return jsonify(health_status)
@app.route('/readiness')
def readiness():
    """就绪检查"""
    return jsonify({
        'status': 'ready',
        'timestamp': datetime.now().isoformat()
    })
@app.route('/liveness')
def liveness():
    """存活检查"""
    return jsonify({
        'status': 'alive',
        'timestamp': datetime.now().isoformat()
    })
def check_database():
    """检查数据库连接"""
    try:
        # 测试数据库连接
        db.session.execute('SELECT 1')
        return {'status': 'healthy', 'message': 'Database connected'}
    except Exception as e:
        return {'status': 'unhealthy', 'message': str(e)}
def check_redis():
    """检查Redis连接"""
    try:
        redis_client.ping()
        return {'status': 'healthy', 'message': 'Redis connected'}
    except Exception as e:
        return {'status': 'unhealthy', 'message': str(e)}

综合告警配置系统

# alert_manager.py
import json
import requests
from typing import Dict, List, Optional
from datetime import datetime
class AlertManager:
    def __init__(self, config_file='alert_config.json'):
        self.config = self.load_config(config_file)
        self.alert_history = []
    def load_config(self, config_file):
        """加载告警配置"""
        with open(config_file, 'r') as f:
            return json.load(f)
    def send_alert(self, alert_type, message, severity='high'):
        """发送告警"""
        alert = {
            'type': alert_type,
            'message': message,
            'severity': severity,
            'timestamp': datetime.now().isoformat()
        }
        # 记录告警历史
        self.alert_history.append(alert)
        # 根据配置发送到不同渠道
        channels = self.config.get('alert_channels', {})
        if 'email' in channels:
            self.send_email_alert(alert)
        if 'slack' in channels:
            self.send_slack_alert(alert)
        if 'webhook' in channels:
            self.send_webhook_alert(alert)
        if 'pagerduty' in channels:
            self.send_pagerduty_alert(alert)
    def send_slack_alert(self, alert):
        """发送Slack告警"""
        webhook_url = self.config['slack_webhook_url']
        payload = {
            'text': f"[{alert['severity'].upper()}] {alert['type']}: {alert['message']}",
            'attachments': [
                {
                    'color': 'danger' if alert['severity'] == 'high' else 'warning',
                    'fields': [
                        {'title': 'Type', 'value': alert['type'], 'short': True},
                        {'title': 'Time', 'value': alert['timestamp'], 'short': True}
                    ]
                }
            ]
        }
        requests.post(webhook_url, json=payload)
    def send_webhook_alert(self, alert):
        """发送Webhook告警"""
        webhooks = self.config.get('webhooks', [])
        for webhook in webhooks:
            try:
                requests.post(
                    webhook['url'],
                    json=alert,
                    headers={'Authorization': f"Bearer {webhook['token']}"}
                )
            except Exception as e:
                print(f"发送Webhook告警失败: {e}")
    def check_alert_thresholds(self, metrics):
        """检查告警阈值"""
        alerts = []
        # 检查错误率
        if metrics.get('error_rate', 0) > self.config['error_rate_threshold']:
            alerts.append({
                'type': 'error_rate',
                'message': f"Error rate {metrics['error_rate']:.2f}% exceeds threshold",
                'severity': 'high'
            })
        # 检查响应时间
        if metrics.get('response_time', 0) > self.config['response_time_threshold']:
            alerts.append({
                'type': 'response_time',
                'message': f"Response time {metrics['response_time']:.2f}s exceeds threshold",
                'severity': 'medium'
            })
        # 检查CPU使用率
        if metrics.get('cpu_usage', 0) > self.config['cpu_threshold']:
            alerts.append({
                'type': 'cpu_usage',
                'message': f"CPU usage {metrics['cpu_usage']:.1f}% exceeds threshold",
                'severity': 'high'
            })
        return alerts
# 配置文件示例 (alert_config.json)
config_example = {
    "alert_channels": ["email", "slack", "webhook"],
    "slack_webhook_url": "https://hooks.slack.com/services/xxx/yyy/zzz",
    "webhooks": [
        {
            "url": "https://api.example.com/alerts",
            "token": "your-token-here"
        }
    ],
    "thresholds": {
        "error_rate": 5.0,
        "response_time": 3.0,
        "cpu_threshold": 80.0,
        "memory_threshold": 85.0
    },
    "notification_rules": {
        "high": {
            "channels": ["email", "slack", "pagerduty"],
            "repeat_interval": 300
        },
        "medium": {
            "channels": ["slack"],
            "repeat_interval": 600
        },
        "low": {
            "channels": ["email"],
            "repeat_interval": 3600
        }
    }
}

最佳实践建议

告警配置清单

# Python项目告警配置清单
## 基础告警
- [ ] 应用崩溃/退出告警
- [ ] 数据库连接失败告警
- [ ] 内存使用率过高告警
- [ ] CPU使用率过高告警
## API监控
- [ ] 响应时间超阈值告警
- [ ] 错误率过高告警
- [ ] 请求量异常告警
- [ ] 5xx错误告警
## 业务监控
- [ ] 关键业务流程失败告警
- [ ] 数据一致性检查告警
- [ ] 任务队列积压告警
## 基础设施
- [ ] 磁盘空间不足告警
- [ ] 网络连接问题告警
- [ ] 证书过期告警
## 告警配置
- [ ] 配置告警级别（P0-P4）
- [ ] 配置告警渠道（邮件/短信/IM）
- [ ] 配置静默时间
- [ ] 配置升级机制

Docker Compose监控栈

# docker-compose.yml
version: '3.8'
services:
  prometheus:
    image: prom/prometheus
    volumes:
      - ./prometheus/alert.rules.yml:/etc/prometheus/alert.rules.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--alertmanager.url=http://alertmanager:9093'
    ports:
      - "9090:9090"
  alertmanager:
    image: prom/alertmanager
    volumes:
      - ./alertmanager/config.yml:/etc/alertmanager/config.yml
    ports:
      - "9093:9093"
    environment:
      - SLACK_WEBHOOK_URL=${SLACK_WEBHOOK_URL}
  grafana:
    image: grafana/grafana
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
      - GF_INSTALL_PLUGINS=grafana-piechart-panel
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources:/etc/grafana/provisioning/datasources
    ports:
      - "3000:3000"
    depends_on:
      - prometheus
  sentry:
    image: sentry
    environment:
      - SENTRY_SECRET_KEY=${SENTRY_SECRET_KEY}
      - SENTRY_POSTGRES_HOST=postgres
      - SENTRY_DB_USER=sentry
      - SENTRY_DB_PASSWORD=${SENTRY_DB_PASSWORD}
      - SENTRY_REDIS_HOST=redis
    ports:
      - "9000:9000"
    depends_on:
      - postgres
      - redis
volumes:
  prometheus_data:
  grafana_data:

配置Python项目监控告警的关键步骤：