监控API网关状态的脚本如何编写

wen 实用脚本 2026-07-06 176

本文目录导读：

监控API网关状态的脚本如何编写

Shell脚本（Linux系统）
Python脚本（更完善）
使用Prometheus + Grafana（推荐）
Docker-Compose部署
使用说明

我来为您提供监控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网关的健康状态、响应时间、端点可用性等关键指标,并支持邮件告警和日志记录。

本文地址： https://www.bmjup.com/post/17570.html