feat: 日志中心 + GRE协议支持 + 流量监控 + 部署脚本
- 新增日志中心功能 (api/logapi.go + Logs.vue) - 新增GRE协议服务支持 (service/gre/) - 新增流量监控组件 (TrafficChart.vue) - 新增L2TP/IPSec测试用例 - 新增日志轮转服务 (logrotate.go) - 新增部署脚本 (scripts/install.sh, uninstall.sh, service文件) - 更新i18n国际化文件 - 优化后端API和前端路由
This commit is contained in:
opsx-tools committed
1 parent
7de1f70905
commit
a14d9e62e6
25 files changed
+2285
-25
No files matched your search
@@ -2,10 +2,16 @@
|
||||
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"
|
||||
@@ -13,6 +19,13 @@ import (
|
||||
|
||||
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 {
|
||||
@@ -35,6 +48,11 @@ func collect(mgr *core.Manager, interval time.Duration) {
|
||||
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)
|
||||
@@ -67,6 +85,22 @@ func collect(mgr *core.Manager, interval time.Duration) {
|
||||
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 同步
|
||||
@@ -106,6 +140,165 @@ func collect(mgr *core.Manager, interval time.Duration) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
|
||||
Reference in new issue
Block a user