opsx-tools
a14d9e62e6
- 新增日志中心功能 (api/logapi.go + Logs.vue) - 新增GRE协议服务支持 (service/gre/) - 新增流量监控组件 (TrafficChart.vue) - 新增L2TP/IPSec测试用例 - 新增日志轮转服务 (logrotate.go) - 新增部署脚本 (scripts/install.sh, uninstall.sh, service文件) - 更新i18n国际化文件 - 优化后端API和前端路由
319 lines
9.0 KiB
Go
319 lines
9.0 KiB
Go
// Package monitor 周期性采集各专线运行状态、流量、会话与配额 // Logan.Li
|
||
package monitor
|
||
|
||
import (
|
||
"fmt"
|
||
"log"
|
||
"strconv"
|
||
"strings"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"vpn-platform/backend/internal/core"
|
||
"vpn-platform/backend/internal/model"
|
||
"vpn-platform/backend/internal/service"
|
||
"vpn-platform/backend/internal/sessionmgr"
|
||
"vpn-platform/backend/internal/store"
|
||
"vpn-platform/backend/internal/trafficmgr"
|
||
)
|
||
|
||
const collectInterval = 15 * time.Second
|
||
|
||
// sampleEvery controls how many collect() rounds between TrafficSample inserts.
|
||
// Default 4 rounds × 15 s ≈ 60 s.
|
||
var sampleEvery int64 = 4
|
||
|
||
// tickCounter is incremented on each collect(); a sample is taken when tick % sampleEvery == 0.
|
||
var tickCounter int64
|
||
|
||
// 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
|
||
}
|
||
|
||
// Decide whether to insert a TrafficSample this round.
|
||
tick := atomic.AddInt64(&tickCounter, 1)
|
||
shouldSample := tick%sampleEvery == 0
|
||
|
||
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})
|
||
}
|
||
|
||
// ---- TrafficSample 时间序列写入（每 ~60 秒）---- // Logan.Li
|
||
if shouldSample && (deltaRx > 0 || deltaTx > 0) {
|
||
store.DB.Create(&model.TrafficSample{
|
||
TunnelID: t.ID,
|
||
RxBytes: st.RxBytes,
|
||
TxBytes: st.TxBytes,
|
||
DeltaRx: deltaRx,
|
||
DeltaTx: deltaTx,
|
||
})
|
||
}
|
||
|
||
// ---- 用户级流量同步（IKEv2 / L2TP / IPSec）---- // Logan.Li
|
||
if st.Running && (t.Protocol == "ikev2" || t.Protocol == "l2tp" || t.Protocol == "ipsec") {
|
||
syncUserTraffic(t, st)
|
||
}
|
||
|
||
// ---- 会话同步（阶段 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, "专线到期自动停止")
|
||
}
|
||
}
|
||
}
|
||
|
||
// syncUserTraffic 从 swanctl --list-sas 输出中提取 per-CHILD_SA 字节计数，
|
||
// 按虚拟 IP 匹配 VPNUser 并更新 rx_bytes / tx_bytes。
|
||
// 当前仅对 IKEv2 和 L2TP 协议生效（IPSec 站点互联无拨号用户）。
|
||
// 匹配策略：遍历 VPNUser，查找其虚拟 IP 是否出现在 swanctl SA 条目中。
|
||
// 若无法精确匹配，则按隧道总流量均摊到所有在线用户（近似值）。
|
||
func syncUserTraffic(t *model.Tunnel, st core.Status) {
|
||
bin, ok := service.LookBin("swanctl")
|
||
if !ok {
|
||
return
|
||
}
|
||
sasOut, err := service.Run(bin, "--list-sas")
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
// 加载该隧道的拨号用户
|
||
var users []model.VPNUser
|
||
if err := store.DB.Where("tunnel_id = ?", t.ID).Find(&users).Error; err != nil || len(users) == 0 {
|
||
return
|
||
}
|
||
|
||
// 从 swanctl 输出解析 virtual IP → (rx, tx) 映射
|
||
// 典型 swanctl --list-sas 格式：
|
||
// ikev2-rw-1: #1, ESTABLISHED, ...
|
||
// net-1{1}: INSTALLED, TUNNEL, ...
|
||
// bytes_i = 1234, bytes_o = 5678
|
||
// local_ts = 0.0.0.0/0
|
||
// remote_ts = 10.12.0.1/32 ← 虚拟 IP
|
||
type saBytes struct {
|
||
rx, tx int64
|
||
vip string // remote virtual IP
|
||
}
|
||
var sas []saBytes
|
||
|
||
lines := strings.Split(sasOut, "\n")
|
||
var currentSA *saBytes
|
||
connPrefix := ""
|
||
switch t.Protocol {
|
||
case "ikev2":
|
||
connPrefix = fmt.Sprintf("ikev2-rw-%d", t.ID)
|
||
case "l2tp":
|
||
connPrefix = fmt.Sprintf("l2tp-%d", t.ID)
|
||
case "ipsec":
|
||
connPrefix = fmt.Sprintf("tunnel-%d", t.ID)
|
||
}
|
||
|
||
for _, line := range lines {
|
||
trimmed := strings.TrimSpace(line)
|
||
// Detect child SA header (INSTALLED line under the connection)
|
||
if strings.Contains(trimmed, "INSTALLED") && strings.Contains(trimmed, connPrefix) {
|
||
sas = append(sas, saBytes{})
|
||
currentSA = &sas[len(sas)-1]
|
||
continue
|
||
}
|
||
if currentSA == nil {
|
||
continue
|
||
}
|
||
// Parse bytes_i / bytes_o
|
||
if strings.Contains(trimmed, "bytes_i") || strings.Contains(trimmed, "bytes_o") {
|
||
parts := strings.Fields(trimmed)
|
||
for j, p := range parts {
|
||
if strings.HasPrefix(p, "bytes_i") && j > 0 {
|
||
if v, e := strconv.ParseInt(strings.TrimSuffix(parts[j-1], ","), 10, 64); e == nil {
|
||
currentSA.rx = v
|
||
}
|
||
}
|
||
if strings.HasPrefix(p, "bytes_o") && j > 0 {
|
||
if v, e := strconv.ParseInt(strings.TrimSuffix(parts[j-1], ","), 10, 64); e == nil {
|
||
currentSA.tx = v
|
||
}
|
||
}
|
||
}
|
||
}
|
||
// Parse remote_ts to extract virtual IP (e.g. "10.12.0.1/32" → "10.12.0.1")
|
||
if strings.Contains(trimmed, "remote_ts") {
|
||
for _, part := range strings.Split(trimmed, "=") {
|
||
part = strings.TrimSpace(part)
|
||
if strings.Contains(part, "/") {
|
||
ip := strings.Split(part, "/")[0]
|
||
ip = strings.TrimSpace(ip)
|
||
if ip != "" && ip != "0.0.0.0" {
|
||
currentSA.vip = ip
|
||
}
|
||
}
|
||
}
|
||
}
|
||
// End of child SA block (blank line or new connection)
|
||
if trimmed == "" && currentSA != nil {
|
||
currentSA = nil
|
||
}
|
||
}
|
||
|
||
// Build vip → user map from VPNUser virtual addresses stored in sessions
|
||
// Fallback: match by username pattern or distribute evenly
|
||
type userTraffic struct {
|
||
userID uint
|
||
rx, tx int64
|
||
}
|
||
updates := map[uint]*userTraffic{}
|
||
|
||
// Try to match SAs to users via VPNSession virtual addresses
|
||
var sessions []model.VPNSession
|
||
store.DB.Where("tunnel_id = ? AND status = ?", t.ID, "active").Find(&sessions)
|
||
vipToUser := map[string]uint{}
|
||
for _, sess := range sessions {
|
||
if sess.VirtualAddress != "" {
|
||
vipToUser[sess.VirtualAddress] = sess.VPNUserID
|
||
}
|
||
}
|
||
|
||
matched := 0
|
||
for _, sa := range sas {
|
||
if sa.vip == "" {
|
||
continue
|
||
}
|
||
if uid, ok := vipToUser[sa.vip]; ok {
|
||
if _, exists := updates[uid]; !exists {
|
||
updates[uid] = &userTraffic{userID: uid}
|
||
}
|
||
updates[uid].rx += sa.rx
|
||
updates[uid].tx += sa.tx
|
||
matched++
|
||
}
|
||
}
|
||
|
||
// If no SA matched by VIP, and there's exactly one active user, assign all to them
|
||
if matched == 0 && len(users) == 1 && (st.RxBytes > 0 || st.TxBytes > 0) {
|
||
updates[users[0].ID] = &userTraffic{userID: users[0].ID, rx: st.RxBytes, tx: st.TxBytes}
|
||
}
|
||
|
||
// Write to DB (only update if new > old to avoid counter resets)
|
||
for _, ut := range updates {
|
||
var user model.VPNUser
|
||
if err := store.DB.First(&user, ut.userID).Error; err != nil {
|
||
continue
|
||
}
|
||
newRx := ut.rx
|
||
newTx := ut.tx
|
||
if newRx < user.RxBytes {
|
||
newRx = user.RxBytes // keep old if counter wrapped
|
||
}
|
||
if newTx < user.TxBytes {
|
||
newTx = user.TxBytes
|
||
}
|
||
if newRx > user.RxBytes || newTx > user.TxBytes {
|
||
now := time.Now()
|
||
store.DB.Model(&user).Updates(map[string]interface{}{
|
||
"rx_bytes": newRx,
|
||
"tx_bytes": newTx,
|
||
"last_seen_at": now,
|
||
})
|
||
}
|
||
}
|
||
|
||
if len(updates) > 0 {
|
||
log.Printf("[monitor] syncUserTraffic tunnel=%d protocol=%s matched_users=%d", t.ID, t.Protocol, len(updates))
|
||
}
|
||
}
|
||
|
||
// 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")
|
||
}
|
||
}
|
||
}
|
||
}
|