本文目录导读：
我来为您提供监控API网关状态的脚本方案,包含几种常用语言的实现：
Shell脚本（Linux系统）
#!/bin/bash
# API网关监控脚本
API_GATEWAY_URL="https://your-api-gateway.com/health"
ALERT_EMAIL="admin@example.com"
LOG_FILE="/var/log/api-gateway-monitor.log"
# 检查API网关健康状态
check_gateway_health() {
local response=$(curl -s -o /dev/null -w "%{http_code}" --connect-timeout 10 \
--max-time 30 "$API_GATEWAY_URL")
local timestamp=$(date "+%Y-%m-%d %H:%M:%S")
if [ "$response" -eq 200 ]; then
echo "[$timestamp] OK - API网关响应正常 (HTTP $response)" >> "$LOG_FILE"
return 0
else
echo "[$timestamp] ERROR - API网关响应异常 (HTTP $response)" >> "$LOG_FILE"
# 发送告警
send_alert "API网关异常: HTTP $response"
return 1
fi
}
# 检查响应时间
check_response_time() {
local response_time=$(curl -s -o /dev/null -w "%{time_total}" \
--connect-timeout 10 --max-time 30 "$API_GATEWAY_URL")
local threshold=5.0 # 5秒阈值
if (( $(echo "$response_time > $threshold" | bc -l) )); then
echo "[$(date)] WARNING - 响应时间过长: ${response_time}s" >> "$LOG_FILE"
send_alert "API网关响应时间异常: ${response_time}s"
fi
}
# 发送告警
send_alert() {
local message="$1"
echo "$message" | mail -s "API网关告警" "$ALERT_EMAIL"
}
# 主函数
main() {
check_gateway_health
check_response_time
# 清理旧日志
find "$LOG_FILE" -mtime +30 -delete
}
main
Python脚本（更完善）
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import requests
import time
import logging
import smtplib
from email.mime.text import MIMEText
from datetime import datetime
import json
import os
class APIGatewayMonitor:
def __init__(self, config_file='config.json'):
self.config = self.load_config(config_file)
self.setup_logging()
self.statistics = {
'total_checks': 0,
'successful_checks': 0,
'failed_checks': 0,
'total_response_time': 0
}
def load_config(self, config_file):
"""加载配置文件"""
default_config = {
'gateway_url': 'https://your-api-gateway.com',
'health_endpoint': '/health',
'alert_email': 'admin@example.com',
'smtp_server': 'smtp.example.com',
'smtp_port': 587,
'smtp_user': 'user@example.com',
'smtp_password': 'password',
'response_time_threshold': 5.0, # 秒
'check_interval': 60, # 秒
'max_retries': 3,
'timeout': 30
}
if os.path.exists(config_file):
with open(config_file, 'r') as f:
config = json.load(f)
default_config.update(config)
return default_config
def setup_logging(self):
"""配置日志"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('api_gateway_monitor.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def check_health(self):
"""检查健康状态"""
url = f"{self.config['gateway_url']}{self.config['health_endpoint']}"
for attempt in range(self.config['max_retries']):
try:
start_time = time.time()
response = requests.get(
url,
timeout=self.config['timeout'],
headers={'User-Agent': 'API-Gateway-Monitor/1.0'}
)
response_time = time.time() - start_time
# 检查HTTP状态码
if response.status_code == 200:
self.logger.info(f"健康检查成功 - 响应时间: {response_time:.2f}s")
self.update_statistics(True, response_time)
# 检查响应时间
if response_time > self.config['response_time_threshold']:
self.logger.warning(f"响应时间过长: {response_time:.2f}s")
self.send_alert(f"API网关响应时间异常: {response_time:.2f}s")
return True
else:
self.logger.error(f"HTTP错误: {response.status_code}")
self.update_statistics(False, response_time)
except requests.exceptions.Timeout:
self.logger.error(f"请求超时 (尝试 {attempt + 1}/{self.config['max_retries']})")
except requests.exceptions.ConnectionError:
self.logger.error(f"连接失败 (尝试 {attempt + 1}/{self.config['max_retries']})")
except Exception as e:
self.logger.error(f"未知错误: {str(e)}")
if attempt < self.config['max_retries'] - 1:
time.sleep(5) # 重试前等待
self.send_alert(f"健康检查失败 - 已重试 {self.config['max_retries']} 次")
return False
def check_endpoints(self):
"""检查所有端点"""
endpoints = ['/api/v1', '/api/v2', '/api/v3']
for endpoint in endpoints:
url = f"{self.config['gateway_url']}{endpoint}"
status = self.check_single_endpoint(url)
self.logger.info(f"端点 {endpoint}: {'正常' if status else '异常'}")
def check_single_endpoint(self, url):
"""检查单个端点"""
try:
response = requests.head(url, timeout=10)
return response.status_code < 500
except:
return False
def check_metrics(self):
"""检查性能指标"""
metrics = {
'uptime': self.calculate_uptime(),
'avg_response_time': self.calculate_avg_response_time(),
'error_rate': self.calculate_error_rate()
}
self.logger.info(f"性能指标: {json.dumps(metrics, indent=2)}")
return metrics
def calculate_uptime(self):
"""计算正常运行时间"""
if self.statistics['total_checks'] == 0:
return 100.0
return (self.statistics['successful_checks'] / self.statistics['total_checks']) * 100
def calculate_avg_response_time(self):
"""计算平均响应时间"""
if self.statistics['successful_checks'] == 0:
return 0.0
return self.statistics['total_response_time'] / self.statistics['successful_checks']
def calculate_error_rate(self):
"""计算错误率"""
if self.statistics['total_checks'] == 0:
return 0.0
return (self.statistics['failed_checks'] / self.statistics['total_checks']) * 100
def update_statistics(self, success, response_time):
"""更新统计信息"""
self.statistics['total_checks'] += 1
if success:
self.statistics['successful_checks'] += 1
self.statistics['total_response_time'] += response_time
else:
self.statistics['failed_checks'] += 1
def send_alert(self, message):
"""发送告警邮件"""
try:
msg = MIMEText(f"API网关告警\n\n{message}\n\n时间: {datetime.now()}")
msg['Subject'] = 'API网关告警'
msg['From'] = self.config['smtp_user']
msg['To'] = self.config['alert_email']
with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
server.starttls()
server.login(self.config['smtp_user'], self.config['smtp_password'])
server.send_message(msg)
self.logger.info(f"告警邮件已发送至 {self.config['alert_email']}")
except Exception as e:
self.logger.error(f"发送告警失败: {str(e)}")
def run(self):
"""运行监控"""
self.logger.info("API网关监控启动")
while True:
try:
# 检查健康状态
self.check_health()
# 检查端点
self.check_endpoints()
# 检查性能指标
self.check_metrics()
# 等待下一次检查
time.sleep(self.config['check_interval'])
except KeyboardInterrupt:
self.logger.info("监控停止")
break
except Exception as e:
self.logger.error(f"监控异常: {str(e)}")
time.sleep(30)
# 配置文件示例 (config.json)
"""
{
"gateway_url": "https://your-api-gateway.com",
"health_endpoint": "/health",
"alert_email": "admin@example.com",
"smtp_server": "smtp.example.com",
"smtp_port": 587,
"smtp_user": "user@example.com",
"smtp_password": "password",
"response_time_threshold": 5.0,
"check_interval": 60,
"max_retries": 3,
"timeout": 30
}
"""
if __name__ == "__main__":
monitor = APIGatewayMonitor()
monitor.run()
使用Prometheus + Grafana（推荐）
# prometheus.yml
scrape_configs:
- job_name: 'api-gateway'
metrics_path: '/metrics'
static_configs:
- targets: ['api-gateway:9090']
scrape_interval: 15s
# 自定义指标导出
import prometheus_client
from prometheus_client import Counter, Histogram, Gauge
import time
# 定义指标
requests_total = Counter('api_gateway_requests_total', 'Total requests')
request_duration = Histogram('api_gateway_request_duration_seconds', 'Request duration')
health_status = Gauge('api_gateway_health_status', 'Health status')
# 启动HTTP服务器
prometheus_client.start_http_server(8000)
while True:
# 检查健康状态
health_status.set(check_health())
time.sleep(15)
Docker-Compose部署
version: '3'
services:
api-gateway-monitor:
build: .
volumes:
- ./logs:/var/log
- ./config:/app/config
environment:
- GATEWAY_URL=https://your-api-gateway.com
- CHECK_INTERVAL=60
restart: always
prometheus:
image: prom/prometheus
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
ports:
- "9090:9090"
grafana:
image: grafana/grafana
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
使用说明
-
基本使用：
# 运行Python脚本 python3 api_gateway_monitor.py # 运行Shell脚本 chmod +x monitor.sh ./monitor.sh
-
Docker部署：
docker-compose up -d
-
系统服务：
# /etc/systemd/system/api-gateway-monitor.service [Unit] Description=API Gateway Monitor After=network.target [Service] Type=simple User=monitor ExecStart=/usr/bin/python3 /opt/monitor/api_gateway_monitor.py Restart=always [Install] WantedBy=multi-user.target
脚本会监控API网关的健康状态、响应时间、端点可用性等关键指标,并支持邮件告警和日志记录。
