监控API错误率的脚本如何写

wen 实用脚本 2026-07-10 101

本文目录导读：

监控API错误率的脚本如何写

基础Shell脚本（适用于简单场景）
Python脚本（功能更完善）
生产级Prometheus指标暴露
Docker Compose部署
使用方式

我来提供一个监控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"

选择哪种实现取决于你的具体需求：