Skip to content
Merged
18 changes: 6 additions & 12 deletions backend/api/handlers/cftunnel.go
Original file line number Diff line number Diff line change
Expand Up @@ -68,10 +68,8 @@ func (h *CfTunnelHandler) List(c *gin.Context) {
resp := make([]tunnelResponse, 0, len(tunnels))
for _, t := range tunnels {
// 实时校正运行状态
if st, err := h.mgr.GetStatus(t.ID); err == nil {
if running, _ := st["running"].(bool); running {
t.Status = "running"
}
if h.mgr.GetStatus(t.ID) == "running" {
t.Status = "running"
}
resp = append(resp, toTunnelResponse(t))
}
Expand Down Expand Up @@ -184,7 +182,7 @@ func (h *CfTunnelHandler) Update(c *gin.Context) {
// DELETE /api/v1/cftunnel/:id
func (h *CfTunnelHandler) Delete(c *gin.Context) {
id, _ := strconv.ParseUint(c.Param("id"), 10, 64)
_ = h.mgr.Stop(uint(id))
h.mgr.Stop(uint(id))
h.db.Delete(&model.CloudflareTunnel{}, id)
logger.WriteLog("info", "cftunnel", fmt.Sprintf("删除CF隧道 [%d]", id))
c.JSON(http.StatusOK, gin.H{"code": 200, "message": "删除成功"})
Expand All @@ -205,20 +203,16 @@ func (h *CfTunnelHandler) Start(c *gin.Context) {
// POST /api/v1/cftunnel/:id/stop
func (h *CfTunnelHandler) Stop(c *gin.Context) {
id, _ := strconv.ParseUint(c.Param("id"), 10, 64)
_ = h.mgr.Stop(uint(id))
h.mgr.Stop(uint(id))
c.JSON(http.StatusOK, gin.H{"code": 200, "message": "已停止"})
}

// GetStatus 获取隧道状态
// GET /api/v1/cftunnel/:id/status
func (h *CfTunnelHandler) GetStatus(c *gin.Context) {
id, _ := strconv.ParseUint(c.Param("id"), 10, 64)
st, err := h.mgr.GetStatus(uint(id))
if err != nil {
c.JSON(http.StatusNotFound, gin.H{"code": 404, "message": "隧道不存在"})
return
}
c.JSON(http.StatusOK, gin.H{"code": 200, "data": st})
status := h.mgr.GetStatus(uint(id))
c.JSON(http.StatusOK, gin.H{"code": 200, "data": gin.H{"status": status}})
}

// GetLogs 获取隧道日志
Expand Down
72 changes: 63 additions & 9 deletions backend/api/handlers/linereg.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ const (
cfgKeyFailureThreshold = "probe_failure_threshold"
cfgKeyToleranceMs = "probe_tolerance_ms"
cfgKeyMaxConcurrent = "probe_max_concurrent"
cfgKeyToolFilter = "probe_tool_filter"
cfgKeyRebindMode = "port_rebind_mode"
)

// 探测参数默认值(与 linereg.DefaultInterval / selector 默认一致)。
Expand Down Expand Up @@ -57,18 +59,22 @@ func NewLineregHandler(db *gorm.DB, log *logrus.Logger, mgr *linereg.Manager) *L

// probeConfigResponse 前端回填用响应体。
type probeConfigResponse struct {
IntervalSec int `json:"interval_sec"`
FailureThreshold int `json:"failure_threshold"`
ToleranceMs int `json:"tolerance_ms"`
MaxConcurrent int `json:"max_concurrent"`
IntervalSec int `json:"interval_sec"`
FailureThreshold int `json:"failure_threshold"`
ToleranceMs int `json:"tolerance_ms"`
MaxConcurrent int `json:"max_concurrent"`
ToolFilter string `json:"tool_filter"` // 逗号分隔的工具名;空 = 全部参与
RebindMode string `json:"rebind_mode"` // auto / manual / off
}

// probeConfigRequest 前端提交的配置更新请求体。
type probeConfigRequest struct {
IntervalSec int `json:"interval_sec"`
FailureThreshold int `json:"failure_threshold"`
ToleranceMs int `json:"tolerance_ms"`
MaxConcurrent int `json:"max_concurrent"`
IntervalSec int `json:"interval_sec"`
FailureThreshold int `json:"failure_threshold"`
ToleranceMs int `json:"tolerance_ms"`
MaxConcurrent int `json:"max_concurrent"`
ToolFilter string `json:"tool_filter"`
RebindMode string `json:"rebind_mode"`
}

// getConfigInt 从 SystemConfig 读取一个整型参数;缺失或解析失败时返回 def。
Expand All @@ -84,17 +90,39 @@ func getConfigInt(db *gorm.DB, key string, def int) int {
return v
}

// GetConfig 读取探测策略四项参数(无记录返回默认值)。
// GetConfig 读取探测策略参数(无记录返回默认值)。
func (h *LineregHandler) GetConfig(c *gin.Context) {
resp := probeConfigResponse{
IntervalSec: getConfigInt(h.db, cfgKeyIntervalSec, defaultIntervalSec),
FailureThreshold: getConfigInt(h.db, cfgKeyFailureThreshold, defaultFailureThreshold),
ToleranceMs: getConfigInt(h.db, cfgKeyToleranceMs, defaultToleranceMs),
MaxConcurrent: getConfigInt(h.db, cfgKeyMaxConcurrent, defaultMaxConcurrent),
ToolFilter: getConfigString(h.db, cfgKeyToolFilter, ""),
RebindMode: h.mgr.RebindMode(),
}
c.JSON(http.StatusOK, gin.H{"code": 200, "data": resp})
}

// getConfigString 从 SystemConfig 读取一个字符串参数;缺失时返回 def。
func getConfigString(db *gorm.DB, key, def string) string {
var cfg model.SystemConfig
if err := db.Where("key = ?", key).First(&cfg).Error; err != nil {
return def
}
return cfg.Value
}

// setConfigUpsertString 写入(或更新)一条 SystemConfig 字符串参数。
func setConfigUpsertString(db *gorm.DB, key, value string) {
var cfg model.SystemConfig
if err := db.Where("key = ?", key).First(&cfg).Error; err == nil {
cfg.Value = value
db.Save(&cfg)
return
}
db.Create(&model.SystemConfig{Key: key, Value: value})
}

// setConfigUpsert 写入(或更新)一条 SystemConfig 整型参数。
func setConfigUpsert(db *gorm.DB, key string, value int) {
var cfg model.SystemConfig
Expand Down Expand Up @@ -135,15 +163,41 @@ func (h *LineregHandler) UpdateConfig(c *gin.Context) {
c.JSON(http.StatusBadRequest, gin.H{"code": 400, "message": field + " out of range"})
return
}
// 重绑模式校验
switch req.RebindMode {
case "", linereg.RebindModeAuto, linereg.RebindModeManual, linereg.RebindModeOff:
default:
c.JSON(http.StatusBadRequest, gin.H{"code": 400, "message": "rebind_mode must be auto/manual/off"})
return
}
// 持久化
setConfigUpsert(h.db, cfgKeyIntervalSec, req.IntervalSec)
setConfigUpsert(h.db, cfgKeyFailureThreshold, req.FailureThreshold)
setConfigUpsert(h.db, cfgKeyToleranceMs, req.ToleranceMs)
setConfigUpsert(h.db, cfgKeyMaxConcurrent, req.MaxConcurrent)
// 立即应用(间隔需在下一轮探测循环生效,其余透传到 selector 即时生效)
setConfigUpsertString(h.db, cfgKeyToolFilter, req.ToolFilter)
setConfigUpsertString(h.db, cfgKeyRebindMode, req.RebindMode)
h.mgr.SetInterval(time.Duration(req.IntervalSec) * time.Second)
h.mgr.SetFailureThreshold(req.FailureThreshold)
h.mgr.SetTolerance(time.Duration(req.ToleranceMs) * time.Millisecond)
h.mgr.SetMaxConcurrent(req.MaxConcurrent)
h.mgr.SetToolFilter(req.ToolFilter)
h.mgr.SetRebindMode(req.RebindMode)
c.JSON(http.StatusOK, gin.H{"code": 200, "message": "已更新"})
}

// PendingRebinds 返回 manual 模式下待重绑的服务清单(svcID -> 目标线路)。
func (h *LineregHandler) PendingRebinds(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"code": 200, "data": h.mgr.PendingRebinds()})
}

// ApplyRebinds 手动触发全部待重绑服务(manual 模式使用)。
func (h *LineregHandler) ApplyRebinds(c *gin.Context) {
applied, err := h.mgr.ApplyPendingRebinds()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"code": 500, "message": err.Error(), "applied": applied})
return
}
c.JSON(http.StatusOK, gin.H{"code": 200, "message": "已重绑", "applied": applied})
}
2 changes: 2 additions & 0 deletions backend/api/router.go
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,8 @@ func NewRouter(opts RouterOptions) *gin.Engine {
lineHandler := handlers.NewLineregHandler(opts.DB, opts.Log, opts.LineregMgr)
auth.GET("/linereg/config", lineHandler.GetConfig)
auth.PUT("/linereg/config", lineHandler.UpdateConfig)
auth.GET("/linereg/rebind-pending", lineHandler.PendingRebinds)
auth.POST("/linereg/rebind-apply", lineHandler.ApplyRebinds)

// WireGuard
wgHandler := handlers.NewWireguardHandler(opts.DB, opts.Log, opts.WireguardMgr)
Expand Down
130 changes: 124 additions & 6 deletions backend/service/linereg/linereg.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ const (
cfgKeyFailureThreshold = "probe_failure_threshold"
cfgKeyToleranceMs = "probe_tolerance_ms"
cfgKeyMaxConcurrent = "probe_max_concurrent"
cfgKeyToolFilter = "probe_tool_filter"
cfgKeyRebindMode = "port_rebind_mode"
)

// 探测参数默认值(与 selector 默认一致)。
Expand Down Expand Up @@ -68,6 +70,14 @@ type Manager struct {
lastUpstream map[uint]string
lastMu sync.Mutex

// rebindMode 端口层重绑模式:"auto"(选线变化自动重绑,默认)/
// "manual"(半自动:只记录待重绑清单,由用户手动触发)/ "off"(关闭重绑)。
rebindMode string
// pendingRebinds manual 模式下记录待重绑的服务(svcID -> 目标线路)。
// 每次选线变化时更新;用户手动触发后清空。
pendingRebinds map[uint]string
pendingMu sync.Mutex

cancel context.CancelFunc
wg sync.WaitGroup
}
Expand All @@ -78,12 +88,78 @@ func NewManager(db *gorm.DB, log *logrus.Logger, tolerance time.Duration) *Manag
log = logrus.New()
}
return &Manager{
db: db,
log: log,
interval: DefaultInterval,
selector: selector.NewSelector(nil, tolerance),
lastUpstream: make(map[uint]string),
db: db,
log: log,
interval: DefaultInterval,
selector: selector.NewSelector(nil, tolerance),
lastUpstream: make(map[uint]string),
rebindMode: RebindModeAuto,
pendingRebinds: make(map[uint]string),
}
}

// 端口层重绑模式取值。
const (
RebindModeAuto = "auto" // 选线变化自动重绑(默认)
RebindModeManual = "manual" // 半自动:只记录待重绑清单,由用户手动触发
RebindModeOff = "off" // 关闭端口层重绑
)

// SetRebindMode 设置端口层重绑模式。非法值忽略。
func (m *Manager) SetRebindMode(mode string) {
switch mode {
case RebindModeAuto, RebindModeManual, RebindModeOff:
m.rebindMode = mode
}
}

// RebindMode 返回当前端口层重绑模式。
func (m *Manager) RebindMode() string {
if m.rebindMode == "" {
return RebindModeAuto
}
return m.rebindMode
}

// PendingRebinds 返回 manual 模式下待重绑的服务(svcID -> 目标线路)快照。
func (m *Manager) PendingRebinds() map[uint]string {
m.pendingMu.Lock()
defer m.pendingMu.Unlock()
out := make(map[uint]string, len(m.pendingRebinds))
for k, v := range m.pendingRebinds {
out[k] = v
}
return out
}

// ApplyPendingRebinds 手动触发所有待重绑服务(manual 模式使用)。
// 逐个调用 portRebinder;全部成功后清空待重绑清单,返回处理数量。
func (m *Manager) ApplyPendingRebinds() (int, error) {
if m.portRebinder == nil {
return 0, fmt.Errorf("端口层重绑回调未注入")
}
pending := m.PendingRebinds()
applied := 0
var firstErr error
for svcID, lineID := range pending {
if err := m.portRebinder(svcID, lineID); err != nil {
m.log.Errorf("[线路选择] 手动重绑端口层服务 %d 到线路 %s 失败: %v", svcID, lineID, err)
m.db.Model(&model.TunService{}).Where("id = ?", svcID).Update("last_error", err.Error())
if firstErr == nil {
firstErr = err
}
continue
}
m.db.Model(&model.TunService{}).Where("id = ?", svcID).Update("last_error", "")
m.log.Infof("[线路选择] 手动重绑端口层服务 %d 到线路 %s 成功", svcID, lineID)
applied++
}
if firstErr == nil {
m.pendingMu.Lock()
m.pendingRebinds = make(map[uint]string)
m.pendingMu.Unlock()
}
return applied, firstErr
}

// Selector 返回内部选择器,供 API / UI 读取状态或手动锁线。
Expand Down Expand Up @@ -113,13 +189,27 @@ func (m *Manager) SetTolerance(d time.Duration) {
m.selector.SetTolerance(d)
}

// LoadProbeConfig 从 SystemConfig 读取探测策略四项参数并应用。
// SetToolFilter 设置参与自动选线的工具集合(逗号分隔的工具名;空 = 全部)。
// 透传到 selector,仅 Tool 命中的线路参与自动选线,其余线路仍可展示/手动锁定。
func (m *Manager) SetToolFilter(filter string) {
var tools []string
for _, t := range strings.Split(filter, ",") {
if t = strings.TrimSpace(t); t != "" {
tools = append(tools, t)
}
}
m.selector.SetToolFilter(tools)
}

// LoadProbeConfig 从 SystemConfig 读取探测策略参数并应用。
// 缺失的键使用默认值。应在 Start 前调用(间隔与并发在首轮探测前生效)。
func (m *Manager) LoadProbeConfig() error {
intervalSec := defaultIntervalSec
failureThreshold := defaultFailureThreshold
toleranceMs := defaultToleranceMs
maxConcurrent := defaultMaxConcurrent
toolFilter := ""
rebindMode := RebindModeAuto

var cfg model.SystemConfig
if err := m.db.Where("key = ?", cfgKeyIntervalSec).First(&cfg).Error; err == nil {
Expand All @@ -142,11 +232,26 @@ func (m *Manager) LoadProbeConfig() error {
maxConcurrent = v
}
}
if err := m.db.Where("key = ?", cfgKeyToolFilter).First(&cfg).Error; err == nil {
toolFilter = cfg.Value
}
if err := m.db.Where("key = ?", cfgKeyRebindMode).First(&cfg).Error; err == nil && cfg.Value != "" {
rebindMode = cfg.Value
}

m.SetInterval(time.Duration(intervalSec) * time.Second)
m.SetFailureThreshold(failureThreshold)
m.SetTolerance(time.Duration(toleranceMs) * time.Millisecond)
m.SetMaxConcurrent(maxConcurrent)
m.SetRebindMode(rebindMode)
// 工具过滤:逗号分隔的工具名列表;空 = 全部工具参与自动选线。
var tools []string
for _, t := range strings.Split(toolFilter, ",") {
if t = strings.TrimSpace(t); t != "" {
tools = append(tools, t)
}
}
m.selector.SetToolFilter(tools)
return nil
}

Expand Down Expand Up @@ -379,6 +484,10 @@ func (m *Manager) applyPortSwitch(lineID string) {
if lineID == "" || m.portRebinder == nil {
return
}
// off 模式:关闭端口层重绑,直接返回。
if m.RebindMode() == RebindModeOff {
return
}
var services []model.TunService
if err := m.db.Where("caddy_site_id = ? AND (domain IS NULL OR domain = '')", 0).Find(&services).Error; err != nil {
m.log.Warnf("[线路选择] 查询端口层服务失败: %v", err)
Expand All @@ -405,6 +514,15 @@ func (m *Manager) applyPortSwitch(lineID string) {
if !found {
continue
}
// manual 模式:只记录待重绑清单(svcID -> 目标线路),不自动停起;
// 由用户通过 ApplyPendingRebinds 手动触发(避免自动重建带来抖动)。
if m.RebindMode() == RebindModeManual {
m.pendingMu.Lock()
m.pendingRebinds[svc.ID] = effLineID
m.pendingMu.Unlock()
m.log.Infof("[线路选择] manual 模式:端口层服务 %d 待重绑到线路 %s(等待手动触发)", svc.ID, effLineID)
continue
}
if err := m.portRebinder(svc.ID, effLineID); err != nil {
m.log.Errorf("[线路选择] 端口层服务 %d 重绑线路 %s 失败: %v", svc.ID, effLineID, err)
// 记录错误供 UI 排查
Expand Down
Loading