#!/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()