本文目录导读：
我来提供一个监控API错误率的脚本方案,包含多种实现方式：
基础Shell脚本（适用于简单场景）
#!/bin/bash
# API监控配置
API_URL="https://api.example.com/endpoint"
LOG_FILE="/var/log/api_monitor.log"
ALERT_EMAIL="admin@example.com"
# 错误率阈值（%）
ERROR_THRESHOLD=5
# 检测次数
CHECK_COUNT=100
# 监控函数
monitor_api() {
local total=0
local errors=0
echo "开始监控API: $API_URL"
echo "时间: $(date '+%Y-%m-%d %H:%M:%S')" >> $LOG_FILE
for ((i=1; i<=$CHECK_COUNT; i++)); do
# 发送请求并获取HTTP状态码
response=$(curl -o /dev/null -s -w "%{http_code}" --max-time 10 $API_URL)
if [ $? -ne 0 ] || [ $response -ge 400 ]; then
errors=$((errors + 1))
echo "请求 #$i 失败 - HTTP状态码: $response" >> $LOG_FILE
fi
total=$((total + 1))
# 间隔1秒
sleep 1
done
# 计算错误率
error_rate=$((errors * 100 / total))
echo "总请求数: $total" >> $LOG_FILE
echo "错误数: $errors" >> $LOG_FILE
echo "错误率: ${error_rate}%" >> $LOG_FILE
# 判断是否需要告警
if [ $error_rate -gt $ERROR_THRESHOLD ]; then
echo "告警: API错误率 ${error_rate}% 超过阈值 ${ERROR_THRESHOLD}%"
echo "API错误率 ${error_rate}% 超过阈值 ${ERROR_THRESHOLD}%" | mail -s "API错误率告警" $ALERT_EMAIL
fi
echo "----------------------------------------" >> $LOG_FILE
}
# 执行监控
monitor_api
Python脚本（功能更完善）
#!/usr/bin/env python3
"""
API错误率监控脚本
支持：多个端点、滑动窗口、多种告警方式
"""
import requests
import time
import logging
import json
from datetime import datetime, timedelta
from collections import deque
import smtplib
from email.mime.text import MIMEText
class APIMonitor:
def __init__(self, config):
self.config = config
self.request_history = deque(maxlen=config['window_size'])
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(config['log_file']),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def check_endpoint(self, endpoint):
"""检查单个端点"""
try:
start_time = time.time()
response = requests.get(
endpoint['url'],
headers=endpoint.get('headers', {}),
timeout=endpoint.get('timeout', 10)
)
response_time = time.time() - start_time
status_code = response.status_code
# 记录请求结果
result = {
'timestamp': datetime.now(),
'url': endpoint['url'],
'status_code': status_code,
'response_time': response_time,
'error': False
}
# 判断是否错误
if status_code >= 400:
result['error'] = True
self.logger.warning(f"请求失败: {endpoint['url']} - HTTP {status_code}")
self.request_history.append(result)
return result
except Exception as e:
error_result = {
'timestamp': datetime.now(),
'url': endpoint['url'],
'status_code': 0,
'response_time': 0,
'error': True,
'error_message': str(e)
}
self.request_history.append(error_result)
self.logger.error(f"请求异常: {endpoint['url']} - {str(e)}")
return error_result
def calculate_error_rate(self):
"""计算错误率"""
if not self.request_history:
return 0
total_requests = len(self.request_history)
error_requests = sum(1 for r in self.request_history if r['error'])
return (error_requests / total_requests) * 100
def send_alert(self, error_rate, threshold):
"""发送告警"""
alert_message = f"""
API错误率告警
时间: {datetime.now()}
当前错误率: {error_rate:.2f}%
阈值: {threshold}%
需要立即关注!
"""
self.logger.warning(alert_message)
# 发送邮件告警
if self.config.get('email_alerts'):
self.send_email_alert(alert_message)
# Webhook通知
if self.config.get('webhook_url'):
self.send_webhook_alert(error_rate)
def send_email_alert(self, message):
"""发送邮件告警"""
try:
msg = MIMEText(message)
msg['Subject'] = 'API错误率告警'
msg['From'] = self.config['email_from']
msg['To'] = self.config['email_to']
with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
server.starttls()
server.login(self.config['email_user'], self.config['email_password'])
server.send_message(msg)
except Exception as e:
self.logger.error(f"发送邮件告警失败: {str(e)}")
def send_webhook_alert(self, error_rate):
"""发送Webhook通知"""
try:
payload = {
'error_rate': error_rate,
'timestamp': str(datetime.now()),
'message': f'API错误率超出阈值: {error_rate:.2f}%'
}
requests.post(self.config['webhook_url'], json=payload)
except Exception as e:
self.logger.error(f"发送Webhook失败: {str(e)}")
def run(self):
"""运行监控"""
self.logger.info("开始API错误率监控")
while True:
for endpoint in self.config['endpoints']:
self.check_endpoint(endpoint)
# 计算错误率
error_rate = self.calculate_error_rate()
current_threshold = self.config.get('error_threshold', 5)
if error_rate > current_threshold and len(self.request_history) >= 10:
self.send_alert(error_rate, current_threshold)
self.logger.info(f"当前错误率: {error_rate:.2f}% (窗口大小: {len(self.request_history)})")
# 等待下次检查
time.sleep(self.config['check_interval'])
def main():
# 配置示例
config = {
'endpoints': [
{
'url': 'https://api.example.com/v1/users',
'timeout': 10,
'headers': {'Authorization': 'Bearer token123'}
},
{
'url': 'https://api.example.com/v1/products',
'timeout': 10
}
],
'window_size': 100, # 滑动窗口大小
'error_threshold': 5, # 错误率阈值(%)
'check_interval': 60, # 检查间隔(秒)
'log_file': '/var/log/api_monitor.log',
'email_alerts': False, # 是否启用邮件告警
'webhook_url': None, # Webhook地址
# 邮件配置（如果需要）
'email_from': 'monitor@example.com',
'email_to': 'admin@example.com',
'smtp_server': 'smtp.example.com',
'smtp_port': 587,
'email_user': 'monitor@example.com',
'email_password': 'password'
}
monitor = APIMonitor(config)
try:
monitor.run()
except KeyboardInterrupt:
print("\n停止监控")
monitor.cleanup()
if __name__ == "__main__":
main()
生产级Prometheus指标暴露
#!/usr/bin/env python3
"""
Prometheus格式的API错误率监控
适用于Kubernetes和Prometheus监控
"""
from prometheus_client import start_http_server, Counter, Gauge, Histogram
import requests
import time
import random
# 定义Prometheus指标
api_requests_total = Counter(
'api_requests_total',
'Total API requests',
['endpoint', 'method', 'status']
)
api_request_duration_seconds = Histogram(
'api_request_duration_seconds',
'API request duration in seconds',
['endpoint'],
buckets=(0.1, 0.5, 1, 2, 5, 10)
)
api_error_rate = Gauge(
'api_error_rate_percent',
'Current API error rate in percent',
['endpoint']
)
class PrometheusAPIMonitor:
def __init__(self, endpoints, port=8000):
self.endpoints = endpoints
self.port = port
self.error_counts = {ep['url']: 0 for ep in endpoints}
self.total_counts = {ep['url']: 0 for ep in endpoints}
def monitor_endpoint(self, endpoint):
"""监控单个端点"""
try:
start_time = time.time()
response = requests.get(
endpoint['url'],
timeout=endpoint.get('timeout', 10)
)
duration = time.time() - start_time
# 记录指标
status_group = '2xx' if 200 <= response.status_code < 300 else \
'4xx' if 400 <= response.status_code < 500 else \
'5xx' if 500 <= response.status_code < 600 else 'other'
api_requests_total.labels(
endpoint=endpoint['url'],
method=response.request.method,
status=status_group
).inc()
api_request_duration_seconds.labels(
endpoint=endpoint['url']
).observe(duration)
# 更新错误计数
self.total_counts[endpoint['url']] += 1
if response.status_code >= 400:
self.error_counts[endpoint['url']] += 1
# 计算并更新错误率
if self.total_counts[endpoint['url']] > 0:
rate = (self.error_counts[endpoint['url']] /
self.total_counts[endpoint['url']]) * 100
api_error_rate.labels(endpoint=endpoint['url']).set(rate)
except Exception as e:
api_requests_total.labels(
endpoint=endpoint['url'],
method='GET',
status='error'
).inc()
def run(self):
"""启动监控"""
print(f"Starting Prometheus metrics server on port {self.port}")
start_http_server(self.port)
while True:
for endpoint in self.endpoints:
self.monitor_endpoint(endpoint)
time.sleep(30) # 每30秒检查一次
# 使用示例
if __name__ == "__main__":
endpoints = [
{'url': 'https://api.example.com/health', 'timeout': 5},
{'url': 'https://api.example.com/v1/data', 'timeout': 10}
]
monitor = PrometheusAPIMonitor(endpoints, port=8000)
monitor.run()
Docker Compose部署
version: '3.8'
services:
api-monitor:
build: .
container_name: api-error-monitor
volumes:
- ./logs:/var/log
- ./config:/app/config
environment:
- PYTHONUNBUFFERED=1
restart: unless-stopped
ports:
- "8000:8000" # Prometheus指标端口
logging:
driver: "json-file"
options:
max-size: "100m"
max-file: "3"
使用方式
执行基本监控：
# 运行Shell脚本 chmod +x api_monitor.sh ./api_monitor.sh # 运行Python脚本 python3 api_monitor.py # 运行Prometheus版本 python3 prometheus_monitor.py
查看监控数据（Prometheus）：
# 访问Prometheus指标 curl http://localhost:8000/metrics # 查询错误率 curl http://localhost:8000/metrics | grep api_error_rate
配置示例（YAML）：
# config.yaml
endpoints:
- url: "https://api.example.com/v1/users"
timeout: 10
headers:
Authorization: "Bearer token123"
- url: "https://api.example.com/v1/products"
timeout: 10
monitoring:
window_size: 100
error_threshold: 5
check_interval: 60
alerting:
email:
enabled: false
from: "monitor@example.com"
to: "admin@example.com"
webhook:
enabled: true
url: "https://hooks.slack.com/services/xxx/yyy/zzz"
选择哪种实现取决于你的具体需求：
- 简单场景：使用Shell脚本
- 功能完善：使用Python脚本
- 生产环境：使用Prometheus指标版本
- 容器化部署：配合Docker使用
