pengxiao
ddbb79b9f6
单容器 Docker 架构的国土空间规划课程智能问答系统，集成 FastAPI 后端与 Next.js 前端。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
373 lines
13 KiB
Python
373 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
服务监控脚本 - 监控多服务架构的健康状态
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
import logging
|
|
import requests
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional
|
|
from pathlib import Path
|
|
|
|
# 配置日志
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler('service_monitor.log'),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class ServiceMonitor:
|
|
"""服务监控器"""
|
|
|
|
def __init__(self):
|
|
self.services = {
|
|
"backend": {
|
|
"name": "FastAPI后端服务",
|
|
"url": "http://localhost:8000/health",
|
|
"timeout": 5,
|
|
"required": True
|
|
},
|
|
"frontend": {
|
|
"name": "Next.js前端服务",
|
|
"url": "http://localhost:8001",
|
|
"timeout": 5,
|
|
"required": True
|
|
},
|
|
"database": {
|
|
"name": "PostgreSQL数据库",
|
|
"type": "internal", # 内部服务，通过后端检查
|
|
"required": True
|
|
},
|
|
"vectorizer": {
|
|
"name": "向量化服务",
|
|
"type": "process", # 进程检查
|
|
"process_name": "vectorizer_service.py",
|
|
"required": False
|
|
}
|
|
}
|
|
|
|
self.check_interval = 30 # 检查间隔（秒）
|
|
self.alert_threshold = 3 # 报警阈值（连续失败次数）
|
|
self.status_history = []
|
|
self.max_history = 100
|
|
|
|
def check_service(self, service_id: str, service_config: Dict) -> Dict:
|
|
"""检查单个服务"""
|
|
service_status = {
|
|
"service_id": service_id,
|
|
"name": service_config["name"],
|
|
"timestamp": datetime.now().isoformat(),
|
|
"status": "unknown",
|
|
"response_time": None,
|
|
"error": None
|
|
}
|
|
|
|
try:
|
|
start_time = time.time()
|
|
|
|
if service_config.get("type") == "internal":
|
|
# 内部服务，通过后端检查
|
|
backend_status = self._check_backend_service(service_id)
|
|
service_status.update(backend_status)
|
|
|
|
elif service_config.get("type") == "process":
|
|
# 进程检查
|
|
process_status = self._check_process_service(service_config)
|
|
service_status.update(process_status)
|
|
|
|
else:
|
|
# HTTP服务检查
|
|
response = requests.get(
|
|
service_config["url"],
|
|
timeout=service_config.get("timeout", 5)
|
|
)
|
|
|
|
response_time = (time.time() - start_time) * 1000 # 毫秒
|
|
|
|
if response.status_code == 200:
|
|
service_status["status"] = "healthy"
|
|
service_status["response_time"] = response_time
|
|
|
|
# 解析响应内容
|
|
try:
|
|
data = response.json()
|
|
service_status["details"] = data
|
|
except:
|
|
pass
|
|
else:
|
|
service_status["status"] = "unhealthy"
|
|
service_status["error"] = f"HTTP {response.status_code}"
|
|
|
|
except requests.exceptions.Timeout:
|
|
service_status["status"] = "timeout"
|
|
service_status["error"] = "请求超时"
|
|
|
|
except requests.exceptions.ConnectionError:
|
|
service_status["status"] = "unreachable"
|
|
service_status["error"] = "连接失败"
|
|
|
|
except Exception as e:
|
|
service_status["status"] = "error"
|
|
service_status["error"] = str(e)
|
|
|
|
return service_status
|
|
|
|
def _check_backend_service(self, service_id: str) -> Dict:
|
|
"""检查后端内部服务"""
|
|
try:
|
|
# 通过后端API检查数据库状态
|
|
response = requests.get(
|
|
"http://localhost:8000/system/status",
|
|
timeout=5
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
|
|
if service_id == "database":
|
|
db_status = data.get("database", {})
|
|
db_type = db_status.get("type", "unknown")
|
|
|
|
return {
|
|
"status": "healthy" if db_type != "unknown" else "unhealthy",
|
|
"details": db_status
|
|
}
|
|
|
|
return {"status": "unhealthy", "error": "后端检查失败"}
|
|
|
|
except Exception as e:
|
|
return {"status": "error", "error": str(e)}
|
|
|
|
def _check_process_service(self, service_config: Dict) -> Dict:
|
|
"""检查进程服务"""
|
|
process_name = service_config.get("process_name")
|
|
|
|
if not process_name:
|
|
return {"status": "unknown", "error": "未配置进程名"}
|
|
|
|
try:
|
|
# 检查进程是否存在（Linux/Mac）
|
|
if sys.platform != "win32":
|
|
import subprocess
|
|
result = subprocess.run(
|
|
["pgrep", "-f", process_name],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
return {"status": "healthy", "details": {"processes": result.stdout.strip().split()}}
|
|
else:
|
|
return {"status": "stopped", "error": "进程未运行"}
|
|
|
|
else:
|
|
# Windows平台检查
|
|
import psutil
|
|
process_count = 0
|
|
|
|
for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
|
|
try:
|
|
cmdline = proc.info.get('cmdline')
|
|
if cmdline and process_name in ' '.join(cmdline):
|
|
process_count += 1
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
|
pass
|
|
|
|
if process_count > 0:
|
|
return {"status": "healthy", "details": {"process_count": process_count}}
|
|
else:
|
|
return {"status": "stopped", "error": "进程未运行"}
|
|
|
|
except Exception as e:
|
|
return {"status": "error", "error": str(e)}
|
|
|
|
def check_all_services(self) -> Dict:
|
|
"""检查所有服务"""
|
|
logger.info("开始检查所有服务...")
|
|
|
|
results = {}
|
|
overall_status = "healthy"
|
|
failed_services = []
|
|
|
|
for service_id, service_config in self.services.items():
|
|
status = self.check_service(service_id, service_config)
|
|
results[service_id] = status
|
|
|
|
if status["status"] != "healthy":
|
|
if service_config.get("required", False):
|
|
overall_status = "unhealthy"
|
|
failed_services.append(service_id)
|
|
|
|
logger.warning(f"❌ {service_config['name']}: {status['status']} - {status.get('error', '')}")
|
|
else:
|
|
logger.info(f"✅ {service_config['name']}: 健康")
|
|
|
|
# 保存到历史记录
|
|
check_result = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"overall_status": overall_status,
|
|
"services": results,
|
|
"failed_services": failed_services
|
|
}
|
|
|
|
self.status_history.append(check_result)
|
|
if len(self.status_history) > self.max_history:
|
|
self.status_history = self.status_history[-self.max_history:]
|
|
|
|
# 保存到文件
|
|
self._save_status_history()
|
|
|
|
logger.info(f"检查完成: 总体状态 - {overall_status}")
|
|
|
|
return check_result
|
|
|
|
def _save_status_history(self):
|
|
"""保存状态历史"""
|
|
try:
|
|
history_file = Path("service_status_history.json")
|
|
|
|
# 只保存最近24小时的数据
|
|
cutoff_time = datetime.now() - timedelta(hours=24)
|
|
recent_history = [
|
|
h for h in self.status_history
|
|
if datetime.fromisoformat(h["timestamp"].replace("Z", "+00:00")) > cutoff_time
|
|
]
|
|
|
|
with open(history_file, "w", encoding="utf-8") as f:
|
|
json.dump(recent_history, f, ensure_ascii=False, indent=2)
|
|
|
|
except Exception as e:
|
|
logger.error(f"保存状态历史失败: {e}")
|
|
|
|
def load_status_history(self) -> List[Dict]:
|
|
"""加载状态历史"""
|
|
try:
|
|
history_file = Path("service_status_history.json")
|
|
if history_file.exists():
|
|
with open(history_file, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception as e:
|
|
logger.error(f"加载状态历史失败: {e}")
|
|
|
|
return []
|
|
|
|
def get_service_stats(self, hours: int = 24) -> Dict:
|
|
"""获取服务统计信息"""
|
|
cutoff_time = datetime.now() - timedelta(hours=hours)
|
|
|
|
relevant_history = [
|
|
h for h in self.status_history
|
|
if datetime.fromisoformat(h["timestamp"].replace("Z", "+00:00")) > cutoff_time
|
|
]
|
|
|
|
if not relevant_history:
|
|
return {}
|
|
|
|
stats = {
|
|
"total_checks": len(relevant_history),
|
|
"healthy_checks": sum(1 for h in relevant_history if h["overall_status"] == "healthy"),
|
|
"uptime_percentage": 0,
|
|
"service_stats": {}
|
|
}
|
|
|
|
if stats["total_checks"] > 0:
|
|
stats["uptime_percentage"] = (stats["healthy_checks"] / stats["total_checks"]) * 100
|
|
|
|
# 计算每个服务的统计
|
|
for service_id in self.services.keys():
|
|
service_checks = []
|
|
for check in relevant_history:
|
|
if service_id in check["services"]:
|
|
service_checks.append(check["services"][service_id])
|
|
|
|
if service_checks:
|
|
healthy_count = sum(1 for s in service_checks if s["status"] == "healthy")
|
|
total_count = len(service_checks)
|
|
|
|
stats["service_stats"][service_id] = {
|
|
"name": self.services[service_id]["name"],
|
|
"total_checks": total_count,
|
|
"healthy_checks": healthy_count,
|
|
"availability": (healthy_count / total_count * 100) if total_count > 0 else 0,
|
|
"last_status": service_checks[-1]["status"] if service_checks else "unknown"
|
|
}
|
|
|
|
return stats
|
|
|
|
def run_monitor(self, duration_minutes: Optional[int] = None):
|
|
"""运行监控器"""
|
|
logger.info("=" * 60)
|
|
logger.info("服务监控器启动")
|
|
logger.info("=" * 60)
|
|
|
|
start_time = time.time()
|
|
|
|
try:
|
|
while True:
|
|
# 检查是否达到运行时长限制
|
|
if duration_minutes:
|
|
elapsed_minutes = (time.time() - start_time) / 60
|
|
if elapsed_minutes >= duration_minutes:
|
|
logger.info(f"达到运行时长限制 ({duration_minutes} 分钟)，停止监控")
|
|
break
|
|
|
|
# 执行检查
|
|
self.check_all_services()
|
|
|
|
# 显示统计信息
|
|
stats = self.get_service_stats(hours=1)
|
|
if stats:
|
|
logger.info(f"最近1小时可用性: {stats.get('uptime_percentage', 0):.1f}%")
|
|
|
|
# 等待下一次检查
|
|
logger.info(f"等待 {self.check_interval} 秒...")
|
|
time.sleep(self.check_interval)
|
|
|
|
except KeyboardInterrupt:
|
|
logger.info("收到中断信号，停止监控")
|
|
except Exception as e:
|
|
logger.error(f"监控器运行异常: {e}")
|
|
finally:
|
|
logger.info("服务监控器停止")
|
|
|
|
def main():
|
|
"""主函数"""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="服务监控工具")
|
|
parser.add_argument("action", choices=["check", "monitor", "stats"],
|
|
help="执行的操作")
|
|
parser.add_argument("--duration", type=int, default=60,
|
|
help="监控持续时间（分钟），仅用于monitor模式")
|
|
parser.add_argument("--hours", type=int, default=24,
|
|
help="统计小时数，用于stats模式")
|
|
|
|
args = parser.parse_args()
|
|
|
|
monitor = ServiceMonitor()
|
|
|
|
if args.action == "check":
|
|
# 单次检查
|
|
result = monitor.check_all_services()
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
|
elif args.action == "monitor":
|
|
# 持续监控
|
|
monitor.run_monitor(args.duration)
|
|
|
|
elif args.action == "stats":
|
|
# 显示统计信息
|
|
stats = monitor.get_service_stats(args.hours)
|
|
print(json.dumps(stats, ensure_ascii=False, indent=2))
|
|
|
|
if __name__ == "__main__":
|
|
main()
|