Fix connection logs: add background status.log sync task
ConnLogs were never populated because AppendConnLog was never called. Added StartStatusSync() goroutine that runs every 10 seconds: 1. Iterates all running instances 2. Reads status.log via ParseStatus (status-version 3) 3. For each CLIENT_LIST entry: - If no active ConnLog exists → AppendConnLog (new connection) - If active ConnLog exists → UpdateActiveConnStats (traffic) 4. For active ConnLogs with no matching CLIENT_LIST → CloseActiveConn Store additions: ListActiveConns(instanceID) — returns all unclosed connections UpdateActiveConnStats() — updates bytes in/out without flush StartStatusSync is called from main.go right after Service creation. UpdateActiveConnStats deliberately skips flush() to avoid writing db.json every 10 seconds; stats are persisted on disconnect.
This commit is contained in:
@@ -56,6 +56,7 @@ func main() {
|
|||||||
log.Printf("warn: ensure CA: %v", err)
|
log.Printf("warn: ensure CA: %v", err)
|
||||||
}
|
}
|
||||||
svc := service.New(cfg, st, ovm)
|
svc := service.New(cfg, st, ovm)
|
||||||
|
svc.StartStatusSync() // 后台定时同步连接日志
|
||||||
|
|
||||||
addr := cfg.Host + ":" + itoa(cfg.Port)
|
addr := cfg.Host + ":" + itoa(cfg.Port)
|
||||||
log.Printf("openvpn-manager listening on %s, data=%s, dist=%s", addr, cfg.DataDir, *distDir)
|
log.Printf("openvpn-manager listening on %s, data=%s, dist=%s", addr, cfg.DataDir, *distDir)
|
||||||
|
|||||||
@@ -36,6 +36,59 @@ func New(cfg *config.Config, st *store.Store, ovm *openvpn.Manager) *Service {
|
|||||||
return &Service{Cfg: cfg, Store: st, Ovm: ovm}
|
return &Service{Cfg: cfg, Store: st, Ovm: ovm}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// StartStatusSync 启动后台定时任务,每 10 秒读取所有实例的 status.log,
|
||||||
|
// 解析 CLIENT_LIST 并同步到 ConnLogs。应在 main.go 中以 goroutine 启动。
|
||||||
|
func (s *Service) StartStatusSync() {
|
||||||
|
go func() {
|
||||||
|
ticker := time.NewTicker(10 * time.Second)
|
||||||
|
defer ticker.Stop()
|
||||||
|
for range ticker.C {
|
||||||
|
s.syncAllInstanceStatus()
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Service) syncAllInstanceStatus() {
|
||||||
|
instances := s.Store.ListInstances()
|
||||||
|
for _, in := range instances {
|
||||||
|
if in.Status != "running" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
statusPath := filepath.Join(s.Cfg.InstanceDir(in.Name), "status.log")
|
||||||
|
entries, err := s.Ovm.ParseStatus(statusPath)
|
||||||
|
if err != nil {
|
||||||
|
continue // 文件不存在或解析失败,跳过
|
||||||
|
}
|
||||||
|
// 构建当前在线 CN 集合
|
||||||
|
onlineCNs := map[string]bool{}
|
||||||
|
for _, e := range entries {
|
||||||
|
onlineCNs[e.CommonName] = true
|
||||||
|
// 如果没有活跃连接记录,则新建
|
||||||
|
if s.Store.FindActiveConn(in.ID, e.CommonName) == nil {
|
||||||
|
_ = s.Store.AppendConnLog(model.ConnectionLog{
|
||||||
|
InstanceID: in.ID,
|
||||||
|
CommonName: e.CommonName,
|
||||||
|
RealIP: e.RealAddress,
|
||||||
|
VPNIP: e.VPNAddress,
|
||||||
|
BytesIn: e.BytesRecv,
|
||||||
|
BytesOut: e.BytesSent,
|
||||||
|
ConnectedAt: e.ConnectedAt,
|
||||||
|
})
|
||||||
|
} else {
|
||||||
|
// 更新流量统计
|
||||||
|
_ = s.Store.UpdateActiveConnStats(in.ID, e.CommonName, e.BytesRecv, e.BytesSent)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// 标记已断开的连接
|
||||||
|
for _, c := range s.Store.ListActiveConns(in.ID) {
|
||||||
|
if !onlineCNs[c.CommonName] {
|
||||||
|
now := time.Now()
|
||||||
|
_ = s.Store.CloseActiveConn(in.ID, c.CommonName, now)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (s *Service) audit(c context.Context, action, target, detail, result, ip string) {
|
func (s *Service) audit(c context.Context, action, target, detail, result, ip string) {
|
||||||
username, _ := c.Value("user").(string)
|
username, _ := c.Value("user").(string)
|
||||||
if username == "" {
|
if username == "" {
|
||||||
|
|||||||
@@ -287,6 +287,35 @@ func (s *Store) CloseActiveConn(instanceID, commonName string, at time.Time) err
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ListActiveConns 返回指定实例所有未断开的连接。
|
||||||
|
func (s *Store) ListActiveConns(instanceID string) []model.ConnectionLog {
|
||||||
|
s.mu.RLock()
|
||||||
|
defer s.mu.RUnlock()
|
||||||
|
var out []model.ConnectionLog
|
||||||
|
for i := len(s.data.ConnLogs) - 1; i >= 0; i-- {
|
||||||
|
c := s.data.ConnLogs[i]
|
||||||
|
if c.InstanceID == instanceID && c.DisconnectedAt == nil {
|
||||||
|
out = append(out, c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// UpdateActiveConnStats 更新活跃连接的流量统计。
|
||||||
|
func (s *Store) UpdateActiveConnStats(instanceID, commonName string, bytesIn, bytesOut int64) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
for i := len(s.data.ConnLogs) - 1; i >= 0; i-- {
|
||||||
|
c := &s.data.ConnLogs[i]
|
||||||
|
if c.InstanceID == instanceID && c.CommonName == commonName && c.DisconnectedAt == nil {
|
||||||
|
c.BytesIn = bytesIn
|
||||||
|
c.BytesOut = bytesOut
|
||||||
|
return nil // 不 flush,避免高频写盘
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// ---- Backups ----
|
// ---- Backups ----
|
||||||
|
|
||||||
func (s *Store) ListBackups() []model.Backup {
|
func (s *Store) ListBackups() []model.Backup {
|
||||||
|
|||||||
Reference in New Issue
Block a user