回来更新新的进程管理模块了：Reload 竞态修复

2026-08-11 21:22:33 +08:00
parent 491de35fa3
commit aebc4f4e58
2 changed files with 196 additions and 149 deletions
@@ -1,11 +1,5 @@
 // internal/process/manager.go
-// frpc-console 进程管理模块
-// 2.7-preview: 状态机驱动 + 端口归属检测 + admin API 健康检查
-
-// 260810-作者留：现在的进程管理模块的框架其实已经基本成型了
-// 但是manager与frpc本身的进程生命周期对不上是个大问题呢
-// 现阶段已经不是猫猫自己能扛得住的了，接下来会试着能不能再收敛一下测试信息完成后续的状态机设计
-// 或许依旧是持久战呢喵
+// 新增 PhaseReloading 状态 + FRPReady 绑定 PID

 package process

@@ -55,38 +49,36 @@ const (
 type ProcessPhase string

 const (
-	PhaseUnknown  ProcessPhase = "UNKNOWN"
-	PhaseStarting ProcessPhase = "STARTING"
-	PhaseRunning  ProcessPhase = "RUNNING"
-	PhaseDegraded ProcessPhase = "DEGRADED"
-	PhaseFailed   ProcessPhase = "FAILED"
-	PhaseStopped  ProcessPhase = "STOPPED"
-	PhaseStopping ProcessPhase = "STOPPING"
-	PhaseConflict ProcessPhase = "CONFLICT"
+	PhaseUnknown   ProcessPhase = "UNKNOWN"
+	PhaseStarting  ProcessPhase = "STARTING"
+	PhaseRunning   ProcessPhase = "RUNNING"
+	PhaseDegraded  ProcessPhase = "DEGRADED"
+	PhaseFailed    ProcessPhase = "FAILED"
+	PhaseStopped   ProcessPhase = "STOPPED"
+	PhaseStopping  ProcessPhase = "STOPPING"
+	PhaseConflict  ProcessPhase = "CONFLICT"
+	PhaseReloading ProcessPhase = "RELOADING" // 新增：reload 中间态
 )

 // ================================================================
 // 数据结构
 // ================================================================

-// PortCheckResult 端口检测结果（含归属信息）
 type PortCheckResult struct {
 	Ready   bool
 	Err     error
-	PID     int    // 占用端口的进程 PID（0 表示无法获取）
-	Process string // 占用端口的进程名
+	PID     int
+	Process string
 }

-// FrpcInstance 系统中检测到的 frpc 实例
 type FrpcInstance struct {
 	PID       int
 	ParentPID int
 	ExecPath  string
 	CmdLine   string
-	Owned     bool // 是否由当前 ProcessManager 管理
+	Owned     bool
 }

-// ProcessState 进程完整状态
 type ProcessState struct {
 	Phase     ProcessPhase `json:"phase"`
 	PID       int          `json:"pid"`
@@ -96,8 +88,8 @@ type ProcessState struct {

 	Alive     bool   `json:"alive"`
 	PortReady bool   `json:"port_ready"`
-	PortPID   int    `json:"port_pid"`   // 实际占用端口的 PID
-	PortOwner string `json:"port_owner"` // 实际占用端口的进程名
+	PortPID   int    `json:"port_pid"`
+	PortOwner string `json:"port_owner"`
 	PortError string `json:"port_error,omitempty"`
 	FRPReady  bool   `json:"frp_ready"`
 	Version   string `json:"version,omitempty"`
@@ -111,7 +103,6 @@ type ProcessState struct {
 	Conflicts []FrpcInstance `json:"conflicts,omitempty"`
 }

-// PortStatus 兼容旧接口
 type PortStatus struct {
 	Port       int    `json:"port"`
 	Occupied   bool   `json:"occupied"`
@@ -120,7 +111,7 @@ type PortStatus struct {
 	ProcessCmd string `json:"process_cmd,omitempty"`
 }

-// FRPCStatus 来自 frpc admin API 的状态响应
+// FRPCStatus 匹配 frpc admin API /api/status 的实际返回结构
 type FRPCStatus struct {
 	TCP []struct {
 		Name       string `json:"name"`
@@ -138,7 +129,6 @@ type FRPCStatus struct {
 		LocalAddr  string `json:"local_addr"`
 		RemoteAddr string `json:"remote_addr"`
 	} `json:"udp"`
-	// frp 0.70.0 还支持其他协议类型，可根据需要扩展
 	HTTP []struct {
 		Name       string `json:"name"`
 		Type       string `json:"type"`
@@ -181,7 +171,6 @@ type FRPCStatus struct {
 	} `json:"sudp"`
 }

-// ConflictInfo 冲突检测结果
 type ConflictInfo struct {
 	HasConflict bool
 	Count       int
@@ -226,6 +215,9 @@ type ProcessManager struct {

 	// 用于 admin API 检测的 HTTP 客户端
 	httpClient *http.Client
+
+	// 当前实例的 run_id（从 admin API 获取）
+	runID string
 }

 // ================================================================
@@ -268,6 +260,12 @@ func (pm *ProcessManager) CurrentPhase() ProcessPhase {
 	return pm.currentPhase
 }

+func (pm *ProcessManager) CurrentPID() int {
+	pm.statusMu.RLock()
+	defer pm.statusMu.RUnlock()
+	return pm.readPIDFile()
+}
+
 func (pm *ProcessManager) setPhase(phase ProcessPhase) {
 	pm.statusMu.Lock()
 	defer pm.statusMu.Unlock()
@@ -300,21 +298,21 @@ func (pm *ProcessManager) LoadConfig() error {
 		log.Printf("[DEBUG] ✅ 从 admin_port 解析到端口: %d", port)
 		return nil
 	}
-	log.Printf("[DEBUG] ❌ admin_port 未找到")
+	log.Printf("[DEBUG] admin_port 未找到，尝试解析 webServer")

 	if port := extractIntValueFromSection(string(content), "webServer", "port"); port > 0 {
 		pm.adminPort = port
 		log.Printf("[DEBUG] ✅ 从 webServer.port 解析到端口: %d", port)
 		return nil
 	}
-	log.Printf("[DEBUG] ❌ webServer.port 未找到")
+	log.Printf("[DEBUG] webServer.port 未找到")

 	if port := extractPortFromAddrSection(string(content), "webServer", "addr"); port > 0 {
 		pm.adminPort = port
 		log.Printf("[DEBUG] ✅ 从 webServer.addr 解析到端口: %d", port)
 		return nil
 	}
-	log.Printf("[DEBUG] ❌ webServer.addr 未找到或解析失败")
+	log.Printf("[DEBUG] webServer.addr 未找到或解析失败")

 	return fmt.Errorf("未找到 admin_port 或 webServer.port/addr 配置")
 }
@@ -392,7 +390,6 @@ func extractPortFromAddrSection(content, section, key string) int {
 // 端口检测（含归属信息）
 // ================================================================

-// CheckPort 检测端口是否被占用，并返回占用者信息
 func (pm *ProcessManager) CheckPort() PortCheckResult {
 	if pm.adminPort <= 0 {
 		return PortCheckResult{Ready: false, Err: fmt.Errorf("admin_port 未配置")}
@@ -404,7 +401,6 @@ func (pm *ProcessManager) CheckPort() PortCheckResult {
 	}
 	conn.Close()

-	// 端口已就绪，获取占用者信息
 	pid, process := pm.getPortOwner(pm.adminPort)
 	return PortCheckResult{
 		Ready:   true,
@@ -413,17 +409,14 @@ func (pm *ProcessManager) CheckPort() PortCheckResult {
 	}
 }

-// getPortOwner 获取占用端口的进程 PID 和进程名
 func (pm *ProcessManager) getPortOwner(port int) (int, string) {
 	if runtime.GOOS != "linux" {
 		return 0, ""
 	}

-	// 使用 ss 获取占用端口的 PID
 	cmd := exec.Command("sh", "-c", fmt.Sprintf("ss -lntp | grep ':%d ' | grep -oP 'pid=\\K[0-9]+' | head -1", port))
 	out, err := cmd.Output()
 	if err != nil {
-		log.Printf("[DEBUG] getPortOwner: ss 执行失败: %v", err)
 		return 0, ""
 	}
 	pidStr := strings.TrimSpace(string(out))
@@ -435,7 +428,6 @@ func (pm *ProcessManager) getPortOwner(port int) (int, string) {
 		return 0, ""
 	}

-	// 获取进程名
 	cmd2 := exec.Command("sh", "-c", fmt.Sprintf("ps -p %d -o comm= 2>/dev/null | head -1", pid))
 	out2, err2 := cmd2.Output()
 	if err2 != nil {
@@ -444,7 +436,6 @@ func (pm *ProcessManager) getPortOwner(port int) (int, string) {
 	return pid, strings.TrimSpace(string(out2))
 }

-// GetPortStatus 兼容旧接口
 func (pm *ProcessManager) GetPortStatus() (*PortStatus, error) {
 	status := &PortStatus{Port: pm.adminPort, Occupied: false, PID: 0, IsFRPC: false}
 	result := pm.CheckPort()
@@ -463,10 +454,9 @@ func (pm *ProcessManager) GetPortStatus() (*PortStatus, error) {
 }

 // ================================================================
-// 实例检测（精确扫描）
+// 实例检测
 // ================================================================

-// DetectFrpcInstances 检测系统中所有 frpc 实例
 func (pm *ProcessManager) DetectFrpcInstances() []FrpcInstance {
 	var instances []FrpcInstance

@@ -540,10 +530,6 @@ func (pm *ProcessManager) filterUnknown(instances []FrpcInstance) []FrpcInstance
 	return result
 }

-// ================================================================
-// 冲突检测
-// ================================================================
-
 func (pm *ProcessManager) DetectConflict() ConflictInfo {
 	instances := pm.DetectFrpcInstances()
 	owned := pm.filterOwned(instances)
@@ -581,60 +567,68 @@ func (pm *ProcessManager) isProcessAlive(pid int) bool {
 }

 // ================================================================
-// FRPReady 检测（使用 admin API）
+// FRPReady 检测（绑定 PID）
 // ================================================================

-// isFRPReady 检测 frpc 是否已完全就绪
-// 判定标准：至少有一个代理处于 running 状态
-func (pm *ProcessManager) isFRPReady() bool {
-	if pm.adminPort <= 0 {
-		log.Printf("[DEBUG] FRPReady=false: admin_port 未配置")
+func (pm *ProcessManager) isFRPReady(pid int) bool {
+	if pid <= 0 || pm.adminPort <= 0 {
+		log.Printf("[DEBUG] FRPReady(pid=%d): pid 无效或 admin_port 未配置", pid)
 		return false
 	}

 	// 方式1: stdout 检测（快速通道）
 	if strings.Contains(pm.getLastOutput(), "start proxy success") ||
 		strings.Contains(pm.getLastOutput(), "login to server success") {
-		log.Printf("[DEBUG] FRPReady=true: 检测到 stdout 关键字")
+		log.Printf("[DEBUG] FRPReady(pid=%d): 检测到 stdout 关键字", pid)
 		return true
 	}

-	// 方式2: admin API 检测
+	// 方式2: admin API 检测（绑定 PID）
+	// 首先确认端口是否被当前 PID 占用
+	portResult := pm.CheckPort()
+	if !portResult.Ready {
+		log.Printf("[DEBUG] FRPReady(pid=%d): 端口 %d 未就绪", pid, pm.adminPort)
+		return false
+	}
+	if portResult.PID > 0 && portResult.PID != pid {
+		log.Printf("[DEBUG] FRPReady(pid=%d): 端口被进程 %d 占用，与期望 PID %d 不一致", pid, portResult.PID, pid)
+		return false
+	}
+
 	url := fmt.Sprintf("http://127.0.0.1:%d/api/status", pm.adminPort)
 	ctx, cancel := context.WithTimeout(context.Background(), APITimeout)
 	defer cancel()

 	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
 	if err != nil {
-		log.Printf("[DEBUG] FRPReady=false: 创建请求失败: %v", err)
+		log.Printf("[DEBUG] FRPReady(pid=%d): 创建请求失败: %v", pid, err)
 		return false
 	}
 	resp, err := pm.httpClient.Do(req)
 	if err != nil {
-		log.Printf("[DEBUG] FRPReady=false: admin API 请求失败: %v", err)
+		log.Printf("[DEBUG] FRPReady(pid=%d): admin API 请求失败: %v", pid, err)
 		return false
 	}
 	defer resp.Body.Close()

 	if resp.StatusCode != 200 {
-		log.Printf("[DEBUG] FRPReady=false: admin API 返回状态码 %d", resp.StatusCode)
+		log.Printf("[DEBUG] FRPReady(pid=%d): admin API 返回状态码 %d", pid, resp.StatusCode)
 		return false
 	}

 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
-		log.Printf("[DEBUG] FRPReady=false: 读取响应失败: %v", err)
+		log.Printf("[DEBUG] FRPReady(pid=%d): 读取响应失败: %v", pid, err)
 		return false
 	}

 	var status FRPCStatus
 	if err := json.Unmarshal(body, &status); err != nil {
-		log.Printf("[DEBUG] FRPReady=false: 解析 JSON 失败: %v", err)
+		log.Printf("[DEBUG] FRPReady(pid=%d): 解析 JSON 失败: %v", pid, err)
 		return false
 	}

-	// 检查是否有任何代理处于 running 状态
-	// 遍历所有已知的代理类型
+	// 检查是否有代理处于 running 状态
 	proxyLists := [][]struct {
 		Name       string `json:"name"`
 		Type       string `json:"type"`
@@ -655,13 +649,13 @@ func (pm *ProcessManager) isFRPReady() bool {
 	for _, proxies := range proxyLists {
 		for _, p := range proxies {
 			if p.Status == "running" {
-				log.Printf("[DEBUG] FRPReady=true: 代理 %s 状态为 running", p.Name)
+				log.Printf("[DEBUG] FRPReady(pid=%d): 代理 %s 状态为 running", pid, p.Name)
 				return true
 			}
 		}
 	}

-	log.Printf("[DEBUG] FRPReady=false: 没有代理处于 running 状态")
+	log.Printf("[DEBUG] FRPReady(pid=%d): 没有代理处于 running 状态", pid)
 	return false
 }

@@ -855,7 +849,7 @@ func (pm *ProcessManager) computeState(pid int, result PortCheckResult) ProcessS
 	}

 	// 3. 启动超时
-	if time.Since(pm.startTime) > StartupTimeout {
+	if time.Since(pm.startTime) > StartupTimeout && pm.currentPhase != PhaseReloading {
 		state.Error = fmt.Sprintf("启动超时 (超过 %v)", StartupTimeout)
 		if portErr != nil {
 			state.Error += ": " + portErr.Error()
@@ -889,17 +883,16 @@ func (pm *ProcessManager) computeState(pid int, result PortCheckResult) ProcessS
 	}

 	// 6. 端口被自己的进程占用，检查 FRPReady
-	frpReady := pm.isFRPReady()
-	state.FRPReady = frpReady
+	state.FRPReady = pm.isFRPReady(pid)

-	if frpReady {
+	if state.FRPReady {
 		state.Phase = PhaseRunning
 		return state
 	}

 	// 7. 端口就绪但 FRP 未就绪
 	state.Phase = PhaseDegraded
-	state.Error = "端口已监听，但 admin API 未就绪"
+	state.Error = "端口已监听，但 admin API 未就绪或代理未运行"
 	return state
 }

@@ -915,6 +908,109 @@ func (pm *ProcessManager) Status() (*ProcessState, error) {
 	return &state, nil
 }

+// ================================================================
+// Reload 操作（P0 核心）
+// ================================================================
+
+// ReloadConfig 执行 frpc 配置热加载，带状态管理
+func (pm *ProcessManager) ReloadConfig(ctx context.Context) error {
+	if err := pm.Lock(); err != nil {
+		return fmt.Errorf("获取锁失败: %w", err)
+	}
+	defer pm.Unlock()
+
+	// 获取当前状态
+	pid := pm.readPIDFile()
+	if pid <= 0 || !pm.isProcessAlive(pid) {
+		// 进程不存在，直接启动
+		log.Printf("[INFO] Reload: frpc 未运行，执行启动")
+		return pm.startLocked(ctx)
+	}
+
+	// 记录 reload 前的 PID 和 run_id
+	oldPID := pid
+	log.Printf("[INFO] Reload: 开始热加载 (当前 PID: %d)", oldPID)
+
+	// 进入 RELOADING 状态
+	pm.setPhase(PhaseReloading)
+
+	// 执行 frpc reload 命令
+	frpcPath, err := pm.getFrpcPath()
+	if err != nil {
+		pm.setPhase(PhaseDegraded)
+		return fmt.Errorf("获取 frpc 路径失败: %w", err)
+	}
+
+	cmd := exec.Command(frpcPath, "reload", "-c", pm.configPath)
+	output, err := cmd.CombinedOutput()
+
+	// 检查 reload 执行结果
+	if err != nil {
+		// reload 命令失败，检查进程是否还在
+		if !pm.isProcessAlive(oldPID) {
+			// 进程已退出，reload 失败且进程丢失
+			log.Printf("[WARN] Reload: frpc 进程在 reload 期间退出 (PID: %d)", oldPID)
+			pm.setPhase(PhaseFailed)
+			pm.deletePIDFile()
+			return fmt.Errorf("reload 失败，frpc 进程已退出: %w", err)
+		}
+
+		// 进程还在，但 reload 命令失败，可能是配置问题
+		log.Printf("[WARN] Reload: 命令失败但进程仍在运行 (PID: %d), 输出: %s", oldPID, string(output))
+		pm.setPhase(PhaseDegraded)
+		return fmt.Errorf("reload 命令执行失败: %w", err)
+	}
+
+	log.Printf("[INFO] Reload: 命令执行成功，输出: %s", string(output))
+
+	// 等待新进程就绪
+	time.Sleep(1 * time.Second)
+
+	// 获取新进程的 PID
+	newPID := pm.readPIDFile()
+	if newPID <= 0 || newPID == oldPID {
+		// PID 没变化，可能是 reload 没有触发进程切换
+		log.Printf("[INFO] Reload: PID 未变化 (PID: %d)，验证服务状态...", oldPID)
+		if pm.isFRPReady(oldPID) {
+			pm.setPhase(PhaseRunning)
+			log.Printf("[INFO] Reload: 服务仍健康，保持运行 (PID: %d)", oldPID)
+			return nil
+		}
+		pm.setPhase(PhaseDegraded)
+		return fmt.Errorf("reload 后服务未就绪 (PID: %d)", oldPID)
+	}
+
+	// PID 已变化，验证新进程
+	log.Printf("[INFO] Reload: PID 从 %d 变为 %d", oldPID, newPID)
+
+	// 等待新进程的 FRPReady
+	for attempt := 0; attempt < 20; attempt++ {
+		if pm.isFRPReady(newPID) {
+			pm.setPhase(PhaseRunning)
+			log.Printf("[INFO] Reload: 成功切换到新进程 (PID: %d)", newPID)
+			return nil
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+
+	// 新进程未就绪，回退状态
+	pm.setPhase(PhaseDegraded)
+	return fmt.Errorf("reload 后新进程未就绪 (PID: %d)", newPID)
+}
+
+// getFrpcPath 获取 frpc 二进制路径
+func (pm *ProcessManager) getFrpcPath() (string, error) {
+	// 如果 frpcBinPath 有效，直接返回
+	if pm.frpcBinPath != "" {
+		if _, err := os.Stat(pm.frpcBinPath); err == nil {
+			return pm.frpcBinPath, nil
+		}
+	}
+	// 否则使用 frp 模块的 GetFrpcPath
+	// 避免循环引用，从外部传入
+	return pm.frpcBinPath, nil
+}
+
 // ================================================================
 // 操作执行
 // ================================================================
@@ -929,7 +1025,6 @@ func (pm *ProcessManager) Start(ctx context.Context) error {

 	pm.expectedStop = false

-	// 1. 检测冲突
 	conflict := pm.DetectConflict()
 	if conflict.HasConflict {
 		log.Printf("[WARN] 检测到 %d 个 frpc 实例 (Owned: %d, Unknown: %d)，进入 CONFLICT 状态",
@@ -957,20 +1052,15 @@ func (pm *ProcessManager) Start(ctx context.Context) error {
 		pm.deletePIDFile()
 	}

-	// 2. 清理孤儿
 	pm.cleanupOrphans()
-
-	// 3. 正常启动
 	return pm.startLocked(ctx)
 }

 func (pm *ProcessManager) startLocked(ctx context.Context) error {
-	// 检查端口状态
 	result := pm.CheckPort()
 	if result.Ready {
 		pid := pm.readPIDFile()
 		if pid > 0 && pm.isProcessAlive(pid) {
-			// 端口就绪且有有效 PID，检查是否匹配
 			if result.PID == pid {
 				log.Printf("[INFO] frpc 已在运行 (PID: %d)", pid)
 				pm.setPhase(PhaseRunning)
@@ -983,6 +1073,7 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error {

 	pm.startTime = time.Now()
 	pm.expectedStop = false
+	pm.runID = ""

 	pm.exitMu.Lock()
 	pm.exitCode = -1
@@ -1033,24 +1124,24 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error {

 		if !pm.expectedStop && code != 0 {
 			log.Printf("[WARN] frpc 进程异常退出 (exit code: %d)", code)
-			pm.setPhase(PhaseFailed)
+			if pm.currentPhase != PhaseReloading {
+				pm.setPhase(PhaseFailed)
+			}
 		}
 		if !pm.expectedStop {
 			pm.deletePIDFile()
 		}
 	}()

-	// 等待端口就绪
 	for attempt := 0; attempt < StartupMaxAttempts; attempt++ {
 		r := pm.CheckPort()
 		if r.Ready {
 			if r.PID > 0 && r.PID != cmd.Process.Pid {
 				log.Printf("[WARN] 端口被进程 %d 占用，与当前进程 %d 不一致，可能存在冲突", r.PID, cmd.Process.Pid)
-				// 继续等待，看是否被接管
 				time.Sleep(StartupRetryDelay)
 				continue
 			}
-			if pm.isFRPReady() {
+			if pm.isFRPReady(cmd.Process.Pid) {
 				log.Printf("[INFO] frpc 启动成功 (PID: %d, 端口: %d)，耗时 %dms",
 					cmd.Process.Pid, pm.adminPort, attempt*int(StartupRetryDelay/time.Millisecond))
 				pm.setPhase(PhaseRunning)
@@ -1061,12 +1152,10 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error {
 		time.Sleep(StartupRetryDelay)
 	}

-	// 超时：检查进程状态
 	if pm.isProcessAlive(cmd.Process.Pid) {
 		stderr := pm.getLastError()
 		log.Printf("[WARN] frpc 启动超时 (PID: %d)，当前 stderr: %s", cmd.Process.Pid, stderr)
-		// 最后一次尝试 admin API
-		if pm.isFRPReady() {
+		if pm.isFRPReady(cmd.Process.Pid) {
 			log.Printf("[INFO] frpc 实际已就绪 (admin API 可访问)，超时误判，修正状态")
 			pm.setPhase(PhaseRunning)
 			return nil
@@ -1186,6 +1275,12 @@ func (pm *ProcessManager) StartHealthMonitor(ctx context.Context) {
 }

 func (pm *ProcessManager) runHealthCheck() {
+	// 如果正在 reloading，跳过健康检查
+	if pm.currentPhase == PhaseReloading {
+		log.Printf("[DEBUG] 健康检查跳过: 正在 RELOADING")
+		return
+	}
+
 	state, err := pm.Status()
 	if err != nil {
 		log.Printf("[WARN] 健康检查失败: %v", err)