本文目录导读：
我来帮你详细介绍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
监控Dashboard
使用Grafana配置可视化dashboard,展示：
- API健康状态热力图
- 响应时间趋势图
- 错误率统计
- 各API可用性百分比
- 数据质量指标
这个监控系统可以根据实际需求灵活扩展,支持多种告警渠道和数据质量检查，建议先从基础的健康检查和响应时间监控开始，逐步添加更复杂的监控规则。
