TEMP_spec_system_noad/monitor.py

#!/usr/bin/env python3
"""
生產環境監控腳本
監控系統效能、Redis狀態、資料庫連接等
"""
import time
import requests
import redis
import json
from datetime import datetime
import argparse
import sys

def check_app_health():
    """檢查應用程式健康狀態"""
    try:
        response = requests.get('http://localhost:12010/', timeout=10)
        return {
            'status': 'healthy' if response.status_code == 200 else 'unhealthy',
            'status_code': response.status_code,
            'response_time': response.elapsed.total_seconds()
        }
    except Exception as e:
        return {
            'status': 'unhealthy',
            'error': str(e),
            'response_time': None
        }

def check_redis_health():
    """檢查 Redis 健康狀態"""
    try:
        r = redis.from_url('redis://localhost:6379/0')
        r.ping()
        info = r.info()
        return {
            'status': 'healthy',
            'used_memory': info.get('used_memory_human'),
            'connected_clients': info.get('connected_clients'),
            'total_commands_processed': info.get('total_commands_processed'),
            'keyspace_hits': info.get('keyspace_hits'),
            'keyspace_misses': info.get('keyspace_misses')
        }
    except Exception as e:
        return {
            'status': 'unhealthy',
            'error': str(e)
        }

def check_onlyoffice_health():
    """檢查 OnlyOffice 健康狀態"""
    try:
        response = requests.get('http://localhost:12011/healthcheck', timeout=10)
        return {
            'status': 'healthy' if response.status_code == 200 else 'unhealthy',
            'status_code': response.status_code,
            'response_time': response.elapsed.total_seconds()
        }
    except Exception as e:
        return {
            'status': 'unhealthy',
            'error': str(e),
            'response_time': None
        }

def get_docker_stats():
    """獲取 Docker 容器統計資訊"""
    import subprocess
    try:
        result = subprocess.run(['docker', 'stats', '--no-stream', '--format', 
                               'table {{.Container}}\t{{.CPUPerc}}\t{{.MemUsage}}\t{{.MemPerc}}'],
                               capture_output=True, text=True, timeout=10)
        if result.returncode == 0:
            return result.stdout
        else:
            return f"Error: {result.stderr}"
    except Exception as e:
        return f"Error getting docker stats: {str(e)}"

def main():
    parser = argparse.ArgumentParser(description='暫時規範系統監控工具')
    parser.add_argument('--json', action='store_true', help='以JSON格式輸出')
    parser.add_argument('--watch', '-w', type=int, metavar='SECONDS', 
                       help='持續監控，指定刷新間隔秒數')
    args = parser.parse_args()

    def run_checks():
        timestamp = datetime.now().isoformat()
        
        # 執行各項健康檢查
        app_health = check_app_health()
        redis_health = check_redis_health()
        onlyoffice_health = check_onlyoffice_health()
        
        results = {
            'timestamp': timestamp,
            'app': app_health,
            'redis': redis_health,
            'onlyoffice': onlyoffice_health
        }
        
        if args.json:
            print(json.dumps(results, indent=2, ensure_ascii=False))
        else:
            # 格式化輸出
            print(f"\n🕒 監控時間: {timestamp}")
            print("=" * 60)
            
            # 應用程式狀態
            status_icon = "✅" if app_health['status'] == 'healthy' else "❌"
            print(f"{status_icon} 應用程式: {app_health['status']}")
            if app_health.get('response_time'):
                print(f"   響應時間: {app_health['response_time']:.3f}s")
            if app_health.get('error'):
                print(f"   錯誤: {app_health['error']}")
            
            # Redis 狀態
            status_icon = "✅" if redis_health['status'] == 'healthy' else "❌"
            print(f"{status_icon} Redis: {redis_health['status']}")
            if redis_health['status'] == 'healthy':
                print(f"   記憶體使用: {redis_health['used_memory']}")
                print(f"   連接數: {redis_health['connected_clients']}")
                if redis_health['keyspace_hits'] and redis_health['keyspace_misses']:
                    hit_rate = redis_health['keyspace_hits'] / (redis_health['keyspace_hits'] + redis_health['keyspace_misses']) * 100
                    print(f"   快取命中率: {hit_rate:.2f}%")
            elif redis_health.get('error'):
                print(f"   錯誤: {redis_health['error']}")
            
            # OnlyOffice 狀態
            status_icon = "✅" if onlyoffice_health['status'] == 'healthy' else "❌"
            print(f"{status_icon} OnlyOffice: {onlyoffice_health['status']}")
            if onlyoffice_health.get('response_time'):
                print(f"   響應時間: {onlyoffice_health['response_time']:.3f}s")
            if onlyoffice_health.get('error'):
                print(f"   錯誤: {onlyoffice_health['error']}")
            
            # Docker 統計
            print("\n📊 容器資源使用:")
            print(get_docker_stats())
        
        return results

    try:
        if args.watch:
            while True:
                try:
                    if not args.json:
                        print("\033[H\033[J")  # 清空終端
                    run_checks()
                    if args.json:
                        print()  # JSON輸出間的分隔
                    time.sleep(args.watch)
                except KeyboardInterrupt:
                    print("\n👋 監控已停止")
                    sys.exit(0)
        else:
            run_checks()
    except Exception as e:
        print(f"❌ 監控執行錯誤: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()