Python数据API监控告警怎么配

wen python案例 2026-07-05 208

本文目录导读：

Python数据API监控告警怎么配

基础监控框架搭建
数据质量监控
集成告警通知
配置管理
最佳实践建议

我来帮你详细介绍Python数据API监控告警的配置方法。

基础监控框架搭建

使用Prometheus + Grafana方案

# requirements.txt
prometheus-client==0.17.1
requests==2.31.0
schedule==1.2.0
# monitor_api.py
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from flask import Flask, Response
import requests
import time
import threading
app = Flask(__name__)
# 定义指标
API_REQUESTS = Counter('api_requests_total', 'Total API requests', ['endpoint', 'status'])
API_LATENCY = Histogram('api_latency_seconds', 'API response time', ['endpoint'])
API_ERRORS = Counter('api_errors_total', 'Total API errors', ['endpoint', 'error_type'])
API_HEALTH = Gauge('api_health_status', 'API health status', ['endpoint'])
def monitor_api(endpoint, url):
    """监控单个API"""
    start_time = time.time()
    try:
        response = requests.get(url, timeout=10)
        latency = time.time() - start_time
        # 记录指标
        API_LATENCY.labels(endpoint=endpoint).observe(latency)
        if response.status_code == 200:
            API_REQUESTS.labels(endpoint=endpoint, status='success').inc()
            API_HEALTH.labels(endpoint=endpoint).set(1)
        else:
            API_REQUESTS.labels(endpoint=endpoint, status='failed').inc()
            API_HEALTH.labels(endpoint=endpoint).set(0)
    except Exception as e:
        API_ERRORS.labels(endpoint=endpoint, error_type=type(e).__name__).inc()
        API_HEALTH.labels(endpoint=endpoint).set(0)
@app.route('/metrics')
def metrics():
    return Response(generate_latest(), mimetype='text/plain')
if __name__ == '__main__':
    # 定时监控任务
    def run_monitor():
        endpoints = {
            'user_api': 'https://api.example.com/users',
            'order_api': 'https://api.example.com/orders',
            'product_api': 'https://api.example.com/products'
        }
        for name, url in endpoints.items():
            monitor_api(name, url)
    # 每30秒监控一次
    import schedule
    schedule.every(30).seconds.do(run_monitor)
    # 启动监控线程
    thread = threading.Thread(target=lambda: app.run(port=8000))
    thread.daemon = True
    thread.start()
    while True:
        schedule.run_pending()
        time.sleep(1)

自定义告警系统

# alert_system.py
import requests
import time
from collections import deque
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
import json
class APIMonitor:
    def __init__(self, config_file='monitor_config.json'):
        self.config = self.load_config(config_file)
        self.alert_history = deque(maxlen=100)
        self.stats = {}
    def load_config(self, config_file):
        """加载监控配置"""
        with open(config_file, 'r') as f:
            return json.load(f)
    def check_api_health(self, api_name, api_config):
        """检查API健康状态"""
        url = api_config['url']
        method = api_config.get('method', 'GET')
        timeout = api_config.get('timeout', 10)
        start_time = time.time()
        try:
            if method.upper() == 'GET':
                response = requests.get(url, timeout=timeout)
            else:
                response = requests.post(url, json=api_config.get('data'), timeout=timeout)
            latency = time.time() - start_time
            status_code = response.status_code
            # 检查响应时间
            max_latency = api_config.get('max_latency', 2)
            if latency > max_latency:
                self.trigger_alert(api_name, 'high_latency', 
                                 f'Response time {latency:.2f}s exceeds {max_latency}s')
            # 检查状态码
            if status_code >= 500:
                self.trigger_alert(api_name, 'server_error', 
                                 f'Server error: {status_code}')
            elif status_code >= 400:
                self.trigger_alert(api_name, 'client_error',
                                 f'Client error: {status_code}')
            # 检查响应内容
            if api_config.get('check_content'):
                content = response.text
                for keyword in api_config['check_content'].get('must_contain', []):
                    if keyword not in content:
                        self.trigger_alert(api_name, 'content_error',
                                         f'Missing required content: {keyword}')
            # 更新统计信息
            self.update_stats(api_name, 'success' if status_code == 200 else 'error')
            return {
                'status': 'healthy' if status_code == 200 else 'unhealthy',
                'latency': latency,
                'status_code': status_code
            }
        except requests.exceptions.Timeout:
            self.trigger_alert(api_name, 'timeout', f'API timeout after {timeout}s')
            self.update_stats(api_name, 'timeout')
            return {'status': 'timeout', 'error': 'Request timeout'}
        except requests.exceptions.ConnectionError:
            self.trigger_alert(api_name, 'connection_error', 'Cannot connect to API')
            self.update_stats(api_name, 'connection_error')
            return {'status': 'unreachable', 'error': 'Connection failed'}
        except Exception as e:
            self.trigger_alert(api_name, 'unknown', str(e))
            self.update_stats(api_name, 'unknown_error')
            return {'status': 'error', 'error': str(e)}
    def update_stats(self, api_name, status):
        """更新API统计信息"""
        if api_name not in self.stats:
            self.stats[api_name] = {
                'total_checks': 0,
                'success_count': 0,
                'error_count': 0,
                'error_rate': 0,
                'last_check_time': None
            }
        stats = self.stats[api_name]
        stats['total_checks'] += 1
        stats['last_check_time'] = datetime.now().isoformat()
        if status == 'success':
            stats['success_count'] += 1
        else:
            stats['error_count'] += 1
        # 计算错误率
        stats['error_rate'] = stats['error_count'] / stats['total_checks'] * 100
        # 触发错误率告警
        if stats['error_rate'] > self.config.get('error_rate_threshold', 10):
            self.trigger_alert(api_name, 'high_error_rate',
                             f'Error rate: {stats["error_rate"]:.2f}%')
    def trigger_alert(self, api_name, alert_type, message):
        """触发告警"""
        alert = {
            'time': datetime.now().isoformat(),
            'api': api_name,
            'type': alert_type,
            'message': message
        }
        self.alert_history.append(alert)
        print(f"[ALERT] {alert}")
        # 发送告警通知
        if self.config.get('email_alerts'):
            self.send_email_alert(alert)
        if self.config.get('webhook_alerts'):
            self.send_webhook_alert(alert)
        if self.config.get('slack_alerts'):
            self.send_slack_alert(alert)
    def send_email_alert(self, alert):
        """发送邮件告警"""
        msg = MIMEText(f"""
API监控告警
时间: {alert['time']}
API: {alert['api']}
类型: {alert['type']}
消息: {alert['message']}
        """)
        msg['Subject'] = f"[API Alert] {alert['type']} - {alert['api']}"
        msg['From'] = self.config['email']['from']
        msg['To'] = self.config['email']['to']
        try:
            with smtplib.SMTP(self.config['email']['smtp_server'], 
                            self.config['email']['smtp_port']) as server:
                server.starttls()
                server.login(self.config['email']['username'],
                           self.config['email']['password'])
                server.send_message(msg)
        except Exception as e:
            print(f"Failed to send email alert: {e}")
    def send_webhook_alert(self, alert):
        """发送Webhook告警"""
        webhook_url = self.config.get('webhook_url')
        if webhook_url:
            try:
                requests.post(webhook_url, json=alert, timeout=5)
            except Exception as e:
                print(f"Failed to send webhook: {e}")
    def send_slack_alert(self, alert):
        """发送Slack告警"""
        slack_webhook = self.config.get('slack_webhook_url')
        if slack_webhook:
            message = {
                'text': f"🚨 *API Alert*\n"
                       f"• API: {alert['api']}\n"
                       f"• Type: {alert['type']}\n"
                       f"• Message: {alert['message']}"
            }
            try:
                requests.post(slack_webhook, json=message, timeout=5)
            except Exception as e:
                print(f"Failed to send Slack alert: {e}")
    def run(self):
        """运行监控循环"""
        while True:
            for api_name, api_config in self.config['apis'].items():
                status = self.check_api_health(api_name, api_config)
                print(f"[{datetime.now().isoformat()}] {api_name}: {status}")
            time.sleep(self.config.get('check_interval', 60))
# 配置文件示例: monitor_config.json
"""
{
    "apis": {
        "user_service": {
            "url": "https://api.example.com/v1/users/health",
            "method": "GET",
            "timeout": 10,
            "max_latency": 2,
            "check_content": {
                "must_contain": ["status", "healthy"]
            }
        },
        "order_service": {
            "url": "https://api.example.com/v1/orders/status",
            "method": "POST",
            "data": {"check": "health"},
            "timeout": 5,
            "max_latency": 1
        }
    },
    "check_interval": 30,
    "error_rate_threshold": 10,
    "email_alerts": true,
    "email": {
        "smtp_server": "smtp.gmail.com",
        "smtp_port": 587,
        "from": "monitor@example.com",
        "to": "admin@example.com",
        "username": "your_email",
        "password": "your_password"
    },
    "slack_alerts": true,
    "slack_webhook_url": "https://hooks.slack.com/services/xxx",
    "webhook_alerts": true,
    "webhook_url": "https://your-webhook.com/alert"
}
"""
if __name__ == '__main__':
    monitor = APIMonitor('monitor_config.json')
    monitor.run()

数据质量监控

# data_quality_monitor.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import json
class DataQualityMonitor:
    def __init__(self):
        self.quality_rules = []
        self.violations = []
    def add_null_check(self, columns, threshold=0.05):
        """添加空值检查规则"""
        def check_null(df, dataset_name):
            violations = []
            for col in columns:
                null_rate = df[col].isnull().mean()
                if null_rate > threshold:
                    violations.append({
                        'type': 'null_value',
                        'column': col,
                        'actual_rate': null_rate,
                        'threshold': threshold,
                        'dataset': dataset_name
                    })
            return violations
        self.quality_rules.append(check_null)
    def add_range_check(self, column, min_val, max_val):
        """添加数值范围检查"""
        def check_range(df, dataset_name):
            violations = []
            outliers = df[(df[column] < min_val) | (df[column] > max_val)]
            if len(outliers) > 0:
                violations.append({
                    'type': 'out_of_range',
                    'column': column,
                    'outlier_count': len(outliers),
                    'min': min_val,
                    'max': max_val,
                    'dataset': dataset_name
                })
            return violations
        self.quality_rules.append(check_range)
    def add_uniqueness_check(self, columns):
        """添加唯一性检查"""
        def check_unique(df, dataset_name):
            violations = []
            for col in columns:
                if df[col].duplicated().any():
                    violations.append({
                        'type': 'duplicate',
                        'column': col,
                        'duplicate_count': df[col].duplicated().sum(),
                        'dataset': dataset_name
                    })
            return violations
        self.quality_rules.append(check_unique)
    def add_data_type_check(self, column, expected_type):
        """添加数据类型检查"""
        def check_type(df, dataset_name):
            violations = []
            actual_type = df[column].dtype
            if actual_type != expected_type:
                violations.append({
                    'type': 'data_type_mismatch',
                    'column': column,
                    'expected': str(expected_type),
                    'actual': str(actual_type),
                    'dataset': dataset_name
                })
            return violations
        self.quality_rules.append(check_type)
    def monitor_data(self, df, dataset_name='dataset'):
        """监控数据质量"""
        all_violations = []
        for rule in self.quality_rules:
            violations = rule(df, dataset_name)
            all_violations.extend(violations)
        self.violations.extend(all_violations)
        if all_violations:
            self.alert_data_quality_issues(all_violations)
        return {
            'dataset': dataset_name,
            'total_records': len(df),
            'violations_found': len(all_violations),
            'violations': all_violations,
            'timestamp': datetime.now().isoformat()
        }
    def alert_data_quality_issues(self, violations):
        """数据质量问题告警"""
        for violation in violations:
            print(f"[DATA QUALITY] {violation['type']} in {violation['dataset']}: "
                  f"{violation}")
# 使用示例
def monitor_api_data_quality(api_response):
    """监控API返回数据质量"""
    monitor = DataQualityMonitor()
    # 添加各种检查规则
    monitor.add_null_check(['id', 'name', 'email'], threshold=0.01)
    monitor.add_range_check('age', 0, 150)
    monitor.add_uniqueness_check(['id', 'email'])
    monitor.add_data_type_check('amount', np.float64)
    # 模拟API返回数据
    data = pd.DataFrame(json.loads(api_response))
    # 执行监控
    result = monitor.monitor_data(data, 'user_api')
    return result

集成告警通知

# alert_integration.py
import requests
from typing import Dict, List, Optional
import logging
class AlertManager:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.channels = []
    def add_email_channel(self, smtp_config):
        self.channels.append(EmailChannel(smtp_config))
    def add_slack_channel(self, webhook_url):
        self.channels.append(SlackChannel(webhook_url))
    def add_dingtalk_channel(self, webhook_url, secret=None):
        self.channels.append(DingTalkChannel(webhook_url, secret))
    def add_wechat_channel(self, corpid, corpsecret, agentid):
        self.channels.append(WeChatChannel(corpid, corpsecret, agentid))
    def send_alert(self, alert_type: str, message: str, 
                   severity: str = 'info', metadata: Optional[Dict] = None):
        """发送告警到所有配置的渠道"""
        alert = Alert(alert_type, message, severity, metadata)
        for channel in self.channels:
            try:
                channel.send(alert)
            except Exception as e:
                self.logger.error(f"Failed to send alert via {channel.__class__.__name__}: {e}")
class Alert:
    def __init__(self, alert_type: str, message: str, 
                 severity: str = 'info', metadata: Optional[Dict] = None):
        self.type = alert_type
        self.message = message
        self.severity = severity
        self.metadata = metadata or {}
        self.timestamp = datetime.now().isoformat()
class SlackChannel:
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url
    def send(self, alert: Alert):
        color = {
            'info': '#36a64f',
            'warning': '#ffcc00',
            'error': '#ff0000',
            'critical': '#8b0000'
        }.get(alert.severity, '#36a64f')
        payload = {
            "attachments": [{
                "color": color,
                "title": f"[{alert.severity.upper()}] {alert.type}",
                "text": alert.message,
                "fields": [
                    {"title": "Severity", "value": alert.severity, "short": True},
                    {"title": "Time", "value": alert.timestamp, "short": True}
                ],
                "footer": "API Monitor"
            }]
        }
        for key, value in alert.metadata.items():
            payload["attachments"][0]["fields"].append({
                "title": key, "value": str(value), "short": True
            })
        requests.post(self.webhook_url, json=payload)
# 使用示例
alert_manager = AlertManager()
alert_manager.add_slack_channel('https://hooks.slack.com/services/xxx')
alert_manager.add_email_channel({
    'smtp_server': 'smtp.gmail.com',
    'port': 587,
    'username': 'your_email',
    'password': 'your_password',
    'from': 'monitor@example.com',
    'to': ['admin@example.com']
})
# 发送告警
alert_manager.send_alert(
    alert_type='api_down',
    message='User service API is unreachable',
    severity='critical',
    metadata={
        'endpoint': '/v1/users',
        'error': 'Connection timeout',
        'duration': '5 minutes'
    }
)

配置管理

# config_manager.py
import yaml
import json
from pathlib import Path
from typing import Dict, Any
class MonitorConfig:
    def __init__(self, config_path: str = 'monitor_config.yaml'):
        self.config_path = Path(config_path)
        self.config = self.load_config()
    def load_config(self) -> Dict[str, Any]:
        """加载配置文件"""
        if self.config_path.suffix in ['.yaml', '.yml']:
            with open(self.config_path, 'r') as f:
                return yaml.safe_load(f)
        elif self.config_path.suffix == '.json':
            with open(self.config_path, 'r') as f:
                return json.load(f)
        else:
            raise ValueError(f"Unsupported config format: {self.config_path.suffix}")
    def get_api_config(self, api_name: str) -> Dict[str, Any]:
        """获取特定API的配置"""
        return self.config.get('apis', {}).get(api_name, {})
    def get_alert_config(self) -> Dict[str, Any]:
        """获取告警配置"""
        return self.config.get('alert', {})
    def get_monitor_interval(self) -> int:
        """获取监控间隔"""
        return self.config.get('monitor', {}).get('interval', 60)
    def validate_config(self) -> bool:
        """验证配置有效性"""
        required_fields = ['apis', 'alert']
        for field in required_fields:
            if field not in self.config:
                raise ValueError(f"Missing required config field: {field}")
        return True
# 配置文件示例: monitor_config.yaml
"""
# 全局配置
monitor:
  interval: 60  # 监控间隔（秒）
  timeout: 10    # 默认超时时间（秒）
# API监控配置
apis:
  user_service:
    url: "https://api.example.com/v1/users/health"
    method: GET
    timeout: 5
    max_latency: 2.0
    expected_status: [200]
    check_content:
      - "healthy"
      - "status"
  order_service:
    url: "https://api.example.com/v1/orders/health"
    method: POST
    data:
      action: health_check
    timeout: 3
    max_latency: 1.5
    expected_status: [200, 201]
    response_schema:
      type: object
      properties:
        status:
          type: string
        uptime:
          type: number
# 数据质量监控
data_quality:
  null_threshold: 0.01
  outlier_threshold: 3
  schema_validation: true
# 告警配置
alert:
  channels:
    - type: email
      enabled: true
      config:
        smtp_server: smtp.gmail.com
        port: 587
        username: "monitor@example.com"
        password: "your_password"
        from: "monitor@example.com"
        to: ["admin@example.com", "devops@example.com"]
    - type: slack
      enabled: true
      config:
        webhook_url: "https://hooks.slack.com/services/xxx"
        channel: "#monitoring"
    - type: webhook
      enabled: false
      config:
        url: "https://your-webhook.com/alert"
        method: POST
  thresholds:
    error_rate: 10  # 错误率超过10%触发告警
    latency: 5      # 延迟超过5秒触发告警
    consecutive_failures: 3  # 连续失败3次触发告警
  severity_levels:
    info:
      channels: [slack]
    warning:
      channels: [slack, email]
    error:
      channels: [slack, email, webhook]
    critical:
      channels: [slack, email, webhook, phone]
# 日志配置
logging:
  level: INFO
  file: /var/log/api_monitor.log
  max_size: 100MB
  backup_count: 10
"""

最佳实践建议

监控指标选择

响应时间：p50/p95/p99分位值
错误率：5XX错误、4XX错误
可用性：API健康检查
数据质量：空值率、异常值、数据一致性

告警级别定义

ALERT_SEVERITY = {
    'critical': {
        'name': '严重',
        'response_time': '立即响应',
        'examples': ['API完全不可用', '数据丢失']
    },
    'warning': {
        'name': '警告',
        'response_time': '15分钟内',
        'examples': ['响应时间升高', '错误率上升']
    },
    'info': {
        'name': '信息',
        'response_time': '工作时间处理',
        'examples': ['API版本更新', '流量变化']
    }
}

告警去重和分级

class AlertDeduplicator:
    def __init__(self):
        self.alert_history = {}
        self.cooldown_period = 300  # 5分钟冷却期
    def should_send_alert(self, alert_key: str) -> bool:
        """检查是否需要发送告警（去重）"""
        now = time.time()
        if alert_key in self.alert_history:
            last_sent = self.alert_history[alert_key]
            if now - last_sent < self.cooldown_period:
                return False
        self.alert_history[alert_key] = now
        return True