"""监控与统计服务。""" import copy import logging import os import psutil from collections import deque from datetime import datetime, timezone, timedelta from models import TrafficSnapshot, Connection, AuditLog, Instance, User from database import db log = logging.getLogger("socks.stats") # 实时指标缓冲区（内存） _metrics_buffer = deque(maxlen=600) # 10分钟 @ 10秒间隔 _metrics_lock = __import__("threading").Lock() # 网络流量基线（用于计算会话增量） _net_baseline = None _net_baseline_lock = __import__("threading").Lock() def _get_net_delta(): global _net_baseline try: net = psutil.net_io_counters() current = {"bytes_sent": net.bytes_sent, "bytes_recv": net.bytes_recv} with _net_baseline_lock: if _net_baseline is None: _net_baseline = copy.deepcopy(current) return {"bytes_sent": 0, "bytes_recv": 0} return { "bytes_sent": current["bytes_sent"] - _net_baseline["bytes_sent"], "bytes_recv": current["bytes_recv"] - _net_baseline["bytes_recv"], } except Exception: return {"bytes_sent": 0, "bytes_recv": 0} def _record_metrics(): """记录系统级指标。""" try: cpu = psutil.cpu_percent(interval=0.5) mem = psutil.virtual_memory() net_delta = _get_net_delta() # 计算活跃连接 with db.app.app_context(): active = db.session.query(Connection).filter_by( state="active" ).count() item = { "ts": datetime.now(timezone.utc).isoformat(), "cpu": cpu, "mem_percent": mem.percent, "mem_used_mb": mem.used // 1024 // 1024, "mem_total_mb": mem.total // 1024 // 1024, "net_bytes_sent": net_delta["bytes_sent"], "net_bytes_recv": net_delta["bytes_recv"], "active_connections": active, } with _metrics_lock: _metrics_buffer.append(item) except Exception as e: log.error("记录指标失败: %s", e) def get_realtime_metrics(): """获取实时系统指标。""" _record_metrics() with _metrics_lock: items = list(_metrics_buffer) if not items: _record_metrics() items = list(_metrics_buffer) return { "latest": items[-1] if items else None, "history": items[-60:], # 最近60个点（10分钟） } def get_traffic_trend(days=30): """获取流量趋势（每日快照）。""" with db.app.app_context(): start = datetime.now(timezone.utc).date() - timedelta(days=days) snapshots = db.session.query( TrafficSnapshot.date, db.func.sum(TrafficSnapshot.bytes_in).label("total_in"), db.func.sum(TrafficSnapshot.bytes_out).label("total_out"), db.func.sum(TrafficSnapshot.connections).label("total_conn"), ).filter( TrafficSnapshot.date >= start ).group_by(TrafficSnapshot.date).order_by( TrafficSnapshot.date ).all() return [{ "date": s.date.isoformat(), "bytes_in_mb": round((s.total_in or 0) / 1024 / 1024, 1), "bytes_out_mb": round((s.total_out or 0) / 1024 / 1024, 1), "connections": s.total_conn or 0, } for s in snapshots] def get_user_stats(): """获取各用户流量排名。""" from sqlalchemy import func with db.app.app_context(): users = db.session.query( User.username, User.bytes_in, User.bytes_out, User.max_concurrent, User.total_traffic_mb, (User.bytes_in + User.bytes_out).label("total_bytes"), ).filter_by(enabled=True).order_by( (User.bytes_in + User.bytes_out).desc() ).limit(20).all() return [{ "username": u.username, "bytes_in_mb": round(u.bytes_in / 1024 / 1024, 1), "bytes_out_mb": round(u.bytes_out / 1024 / 1024, 1), "total_mb": round(u.total_bytes / 1024 / 1024, 1), "limit_mb": u.total_traffic_mb, "usage_pct": round( u.total_bytes / 1024 / 1024 / u.total_traffic_mb * 100, 1 ) if u.total_traffic_mb else None, } for u in users] def get_instance_stats(): """获取各实例流量统计。""" with db.app.app_context(): instances = Instance.query.all() result = [] for inst in instances: conns = db.session.query( db.func.sum(Connection.bytes_in).label("total_in"), db.func.sum(Connection.bytes_out).label("total_out"), db.func.count(Connection.id).label("total_conn"), ).filter( Connection.instance_id == inst.id ).first() result.append({ "id": inst.id, "name": inst.name, "host": inst.listen_host, "port": inst.listen_port, "bytes_in_mb": round((conns.total_in or 0) / 1024 / 1024, 1), "bytes_out_mb": round((conns.total_out or 0) / 1024 / 1024, 1), "total_connections": conns.total_conn or 0, }) return result def get_active_connections(): """获取当前活跃连接列表。""" with db.app.app_context(): conns = db.session.query(Connection).filter_by( state="active" ).order_by(Connection.started_at.desc()).limit(50).all() return [{ "id": c.id, "instance": db.session.query(Instance).get(c.instance_id).name if c.instance_id else "", "username": c.username, "src": f"{c.src_ip}:{c.src_port}" if c.src_ip else "", "dst": f"{c.dst_addr}:{c.dst_port}" if c.dst_addr else "", "protocol": c.protocol, "started": c.started_at.isoformat() if c.started_at else "", "bytes_in": c.bytes_in, "bytes_out": c.bytes_out, } for c in conns] def get_dashboard_summary(): """仪表盘汇总。""" metrics = get_realtime_metrics() with db.app.app_context(): total_instances = db.session.query(Instance).count() running_instances = db.session.query(Instance).filter_by( enabled=True ).count() total_users = db.session.query(User).count() active_users = db.session.query(User).filter_by( enabled=True, banned=False ).count() active_conns = db.session.query(Connection).filter_by( state="active" ).count() today_logs = db.session.query(AuditLog).filter( AuditLog.timestamp >= datetime.now(timezone.utc).replace( hour=0, minute=0, second=0 ) ).count() return { "system": metrics["latest"], "total_instances": total_instances, "running_instances": running_instances, "total_users": total_users, "active_users": active_users, "active_connections": active_conns, "today_logs": today_logs, } def prune_old_data(): """清理旧数据（运行在后台）。""" with db.app.app_context(): # 清理 30 天前的连接记录（closed的） cutoff = datetime.now(timezone.utc) - timedelta(days=30) db.session.query(Connection).filter( Connection.state == "closed", Connection.started_at < cutoff ).delete(synchronize_session=False) # 清理 90 天前的审计日志 cutoff2 = datetime.now(timezone.utc) - timedelta(days=90) db.session.query(AuditLog).filter(AuditLog.timestamp < cutoff2).delete( synchronize_session=False ) db.session.commit() log.info("数据清理完成") def record_traffic_snapshot(): """记录每日流量快照（用于流量趋势图表）。""" from datetime import date today = date.today() with db.app.app_context(): from sqlalchemy import func today_start = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0) result = db.session.query( func.coalesce(func.sum(Connection.bytes_in), 0), func.coalesce(func.sum(Connection.bytes_out), 0), func.count(Connection.id), ).filter(Connection.started_at >= today_start).first() total_in = result[0] if result else 0 total_out = result[1] if result else 0 total_conns = result[2] if result else 0 snap = db.session.query(TrafficSnapshot).filter_by(date=today).first() if snap: snap.bytes_in = total_in snap.bytes_out = total_out snap.connections = total_conns else: snap = TrafficSnapshot( date=today, bytes_in=total_in, bytes_out=total_out, connections=total_conns, ) db.session.add(snap) db.session.commit() log.info("流量快照已记录: %s", today)