// Package monitor 周期性采集各专线运行状态、流量、会话与配额 // Logan.Li package monitor import ( "time" "vpn-platform/backend/internal/core" "vpn-platform/backend/internal/model" "vpn-platform/backend/internal/sessionmgr" "vpn-platform/backend/internal/store" "vpn-platform/backend/internal/trafficmgr" ) const collectInterval = 15 * time.Second // Start 启动状态采集循环(每 15 秒刷新一次) func Start(mgr *core.Manager, interval time.Duration) { if interval == 0 { interval = collectInterval } ticker := time.NewTicker(interval) go func() { collect(mgr, interval) // 启动即采集一次 for range ticker.C { collect(mgr, interval) } }() } func collect(mgr *core.Manager, interval time.Duration) { if store.DB == nil { return } var tunnels []model.Tunnel if err := store.DB.Find(&tunnels).Error; err != nil { return } for i := range tunnels { t := &tunnels[i] p, err := mgr.Get(t.Protocol) if err != nil { continue } st, err := p.Status(t) if err != nil { continue } mgr.SetStatus(t.ID, st) // 数据库状态同步 if st.State != "" && st.State != t.State { store.DB.Model(t).Update("state", st.State) } // 流量增量计算(用于配额/会话/异常检测) deltaRx := st.RxBytes - t.RxBytes deltaTx := st.TxBytes - t.TxBytes if deltaRx < 0 { deltaRx = st.RxBytes } if deltaTx < 0 { deltaTx = st.TxBytes } // 流量持久化(隧道级) if st.RxBytes > 0 || st.TxBytes > 0 { store.DB.Model(t).Updates(map[string]interface{}{"rx_bytes": st.RxBytes, "tx_bytes": st.TxBytes}) } // ---- 会话同步(阶段 1)---- if st.Running { // WireGuard 等多对端协议:从 Peers 同步 if len(st.Peers) > 0 { sessionmgr.SyncFromPeers(t.ID, t.Protocol, st.Peers) } // OpenVPN 的会话同步在 collectTraffic 内部完成(已直接写 VPNUser) } else { // 隧道未运行 → 标记所有 active 会话为 disconnected sessionmgr.SyncSessions(t.ID, t.Protocol, nil) } // ---- 流量配额检查(阶段 2)---- if deltaRx > 0 || deltaTx > 0 { action := trafficmgr.CheckQuota(t.ID, deltaRx, deltaTx) if action != "" { trafficmgr.HandleExceed(t.ID, action, t.Name) if action == "block" { _ = p.Stop(t) store.DB.Model(t).Update("state", "stopped") store.Audit("system", "quota_block", t.Name, "流量配额超额,自动停止") } } } // ---- 异常流量告警(阶段 3)---- if st.Running && (deltaRx > 0 || deltaTx > 0) { trafficmgr.CheckRateAbnormal(t.ID, deltaRx, deltaTx, interval) } // ---- 到期自动停止 ---- if t.ExpireAt != nil && time.Now().After(*t.ExpireAt) && t.State == "running" { _ = p.Stop(t) store.DB.Model(t).Update("state", "stopped") store.Audit("system", "expire_stop", t.Name, "专线到期自动停止") } } } // AutoStartAll 启动时拉起所有标记为自启的专线 func AutoStartAll(mgr *core.Manager) { var tunnels []model.Tunnel if err := store.DB.Where("auto_start = ?", true).Find(&tunnels).Error; err != nil { return } for i := range tunnels { t := &tunnels[i] if p, err := mgr.Get(t.Protocol); err == nil { if err := p.Start(t); err != nil { store.DB.Model(t).Updates(map[string]interface{}{"state": "error", "last_error": err.Error()}) } else { store.DB.Model(t).Update("state", "running") } } } }