// internal/process/manager.go // frpc-console 进程管理模块 // 2.6-preview: 状态机驱动 + 冲突检测 + 生命周期管理 // 2.7-preview: 软化版 - 精确控制 + CONFLICT/STOPPING 状态 package process import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "log" "net" "net/http" "os" "os/exec" "path/filepath" "runtime" "strconv" "strings" "sync" "syscall" "time" ) // ================================================================ // 常量定义 // ================================================================ const ( LockFileName = ".frpc.lock" PortCheckTimeout = 500 * time.Millisecond StopMaxWaitTime = 5 * time.Second LockAcquireTimeout = 30 * time.Second LockRetryInterval = 100 * time.Millisecond APITimeout = 2 * time.Second StartupMaxAttempts = 50 StartupRetryDelay = 200 * time.Millisecond StartupTimeout = 10 * time.Second HealthCheckInterval = 15 * time.Second ) // ================================================================ // 进程状态定义 // ================================================================ 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" ) // ================================================================ // 数据结构 // ================================================================ type PortCheckResult struct { Ready bool Err error } type FrpcInstance struct { PID int ParentPID int ExecPath string CmdLine string Owned bool } type ProcessState struct { Phase ProcessPhase `json:"phase"` PID int `json:"pid"` Port int `json:"port"` StartedAt time.Time `json:"started_at"` ExitCode int `json:"exit_code,omitempty"` Alive bool `json:"alive"` PortReady bool `json:"port_ready"` PortError string `json:"port_error,omitempty"` FRPReady bool `json:"frp_ready"` Version string `json:"version,omitempty"` ExpectedStop bool `json:"expected_stop"` LastOutput string `json:"last_output,omitempty"` LastError string `json:"last_error,omitempty"` Error string `json:"error,omitempty"` Conflicts []FrpcInstance `json:"conflicts,omitempty"` } // PortStatus 兼容旧接口 type PortStatus struct { Port int `json:"port"` Occupied bool `json:"occupied"` PID int `json:"pid"` IsFRPC bool `json:"is_frpc"` ProcessCmd string `json:"process_cmd,omitempty"` } // FRPCStatus 来自 frpc admin API 的状态响应 type FRPCStatus struct { Version string `json:"version"` RunID string `json:"run_id"` Proxies []struct { Name string `json:"name"` Type string `json:"type"` Status string `json:"status"` LocalAddr string `json:"local_addr"` } `json:"proxies"` } // ================================================================ // 全局变量 // ================================================================ var ( globalManager *ProcessManager globalMu sync.Mutex ) // ================================================================ // ProcessManager 主结构 // ================================================================ type ProcessManager struct { mu sync.Mutex dataDir string configPath string frpcBinPath string adminPort int startTime time.Time expectedStop bool exitCode int exitMu sync.Mutex lastOutput string lastError string outputMu sync.Mutex lockFile *os.File locked bool currentPhase ProcessPhase statusMu sync.RWMutex } // ================================================================ // 构造函数 // ================================================================ func NewManager(dataDir, configPath, frpcBinPath string) *ProcessManager { pm := &ProcessManager{ dataDir: dataDir, configPath: configPath, frpcBinPath: frpcBinPath, adminPort: 0, currentPhase: PhaseUnknown, } return pm } func SetGlobalManager(pm *ProcessManager) { globalMu.Lock() defer globalMu.Unlock() globalManager = pm } func GetGlobalManager() *ProcessManager { globalMu.Lock() defer globalMu.Unlock() return globalManager } func (pm *ProcessManager) AdminPort() int { return pm.adminPort } func (pm *ProcessManager) CurrentPhase() ProcessPhase { pm.statusMu.RLock() defer pm.statusMu.RUnlock() return pm.currentPhase } func (pm *ProcessManager) setPhase(phase ProcessPhase) { pm.statusMu.Lock() defer pm.statusMu.Unlock() if pm.currentPhase != phase { log.Printf("[STATUS] %s → %s", pm.currentPhase, phase) pm.currentPhase = phase } } // ================================================================ // 配置读取 // ================================================================ func (pm *ProcessManager) LoadConfig() error { content, err := os.ReadFile(pm.configPath) if err != nil { return fmt.Errorf("读取配置文件失败: %w", err) } log.Printf("[DEBUG] LoadConfig 读取到文件,长度: %d 字节", len(content)) preview := string(content) if len(preview) > 600 { preview = preview[:600] + "\n... (截断)" } log.Printf("[DEBUG] 文件内容预览:\n%s", preview) if port := extractIntValue(string(content), "admin_port"); port > 0 { pm.adminPort = port log.Printf("[DEBUG] ✅ 从 admin_port 解析到端口: %d", port) return nil } log.Printf("[DEBUG] ❌ admin_port 未找到") 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 未找到") 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 未找到或解析失败") return fmt.Errorf("未找到 admin_port 或 webServer.port/addr 配置") } func extractIntValue(content, key string) int { lines := strings.Split(content, "\n") for _, line := range lines { trimmed := strings.TrimSpace(line) if strings.HasPrefix(trimmed, key) { parts := strings.SplitN(trimmed, "=", 2) if len(parts) == 2 { val := strings.TrimSpace(parts[1]) val = strings.Trim(val, `"`) if port, err := strconv.Atoi(val); err == nil && port > 0 { return port } } } } return 0 } func extractIntValueFromSection(content, section, key string) int { lines := strings.Split(content, "\n") inSection := false for _, line := range lines { trimmed := strings.TrimSpace(line) if strings.HasPrefix(trimmed, "[") && strings.HasSuffix(trimmed, "]") { sectionName := strings.TrimSpace(strings.Trim(trimmed, "[]")) inSection = strings.EqualFold(sectionName, section) continue } if inSection && strings.HasPrefix(trimmed, key) { parts := strings.SplitN(trimmed, "=", 2) if len(parts) == 2 { val := strings.TrimSpace(parts[1]) val = strings.Trim(val, `"`) if port, err := strconv.Atoi(val); err == nil && port > 0 { return port } } } } return 0 } func extractPortFromAddrSection(content, section, key string) int { lines := strings.Split(content, "\n") inSection := false for _, line := range lines { trimmed := strings.TrimSpace(line) if strings.HasPrefix(trimmed, "[") && strings.HasSuffix(trimmed, "]") { sectionName := strings.TrimSpace(strings.Trim(trimmed, "[]")) inSection = strings.EqualFold(sectionName, section) continue } if inSection && strings.HasPrefix(trimmed, key) { parts := strings.SplitN(trimmed, "=", 2) if len(parts) == 2 { val := strings.TrimSpace(parts[1]) val = strings.Trim(val, `"`) if idx := strings.LastIndex(val, ":"); idx != -1 { portStr := val[idx+1:] if port, err := strconv.Atoi(portStr); err == nil && port > 0 { return port } } } } } return 0 } // ================================================================ // 端口检测 // ================================================================ func (pm *ProcessManager) CheckPort() PortCheckResult { if pm.adminPort <= 0 { return PortCheckResult{Ready: false, Err: fmt.Errorf("admin_port 未配置")} } conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", pm.adminPort), PortCheckTimeout) if err != nil { return PortCheckResult{Ready: false, Err: err} } conn.Close() return PortCheckResult{Ready: true, Err: nil} } func (pm *ProcessManager) GetPortStatus() (*PortStatus, error) { status := &PortStatus{Port: pm.adminPort, Occupied: false, PID: 0, IsFRPC: false} result := pm.CheckPort() if result.Err != nil { return status, result.Err } if !result.Ready { return status, nil } status.Occupied = true pid, err := pm.getPIDByPort(pm.adminPort) if err != nil { if pidFromFile := pm.readPIDFile(); pidFromFile > 0 { if pm.isProcessListeningOnPort(pidFromFile, pm.adminPort) { status.PID = pidFromFile } } } else { status.PID = pid } if status.PID == 0 { return status, nil } isFRPC, cmd := pm.isFRPCProcess(status.PID) status.IsFRPC = isFRPC status.ProcessCmd = cmd return status, nil } func (pm *ProcessManager) getPIDByPort(port int) (int, error) { if pid, err := pm.getPIDBySS(port); err == nil && pid > 0 { return pid, nil } if pid, err := pm.getPIDByNetstat(port); err == nil && pid > 0 { return pid, nil } if pid, err := pm.getPIDByLsof(port); err == nil && pid > 0 { return pid, nil } return 0, fmt.Errorf("无法通过端口反查 PID") } func (pm *ProcessManager) getPIDBySS(port int) (int, error) { cmd := exec.Command("ss", "-lpn", "state", "listening") out, err := cmd.Output() if err != nil { return 0, err } scanner := bufio.NewScanner(bytes.NewReader(out)) for scanner.Scan() { line := scanner.Text() if !strings.Contains(line, fmt.Sprintf(":%d", port)) { continue } if idx := strings.Index(line, "pid="); idx != -1 { end := strings.Index(line[idx:], ",") if end == -1 { end = strings.Index(line[idx:], ")") } if end == -1 { continue } pidStr := line[idx+4 : idx+end] pidStr = strings.TrimSpace(pidStr) if pid, err := strconv.Atoi(pidStr); err == nil && pid > 0 { return pid, nil } } } return 0, fmt.Errorf("未找到监听端口 %d 的进程", port) } func (pm *ProcessManager) getPIDByNetstat(port int) (int, error) { cmd := exec.Command("netstat", "-tulpn") out, err := cmd.Output() if err != nil { return 0, err } scanner := bufio.NewScanner(bytes.NewReader(out)) for scanner.Scan() { line := scanner.Text() if !strings.Contains(line, fmt.Sprintf(":%d", port)) { continue } parts := strings.Fields(line) if len(parts) < 7 { continue } last := parts[len(parts)-1] if idx := strings.Index(last, "/"); idx != -1 { pidStr := last[:idx] if pid, err := strconv.Atoi(pidStr); err == nil && pid > 0 { return pid, nil } } } return 0, fmt.Errorf("未找到监听端口 %d 的进程", port) } func (pm *ProcessManager) getPIDByLsof(port int) (int, error) { cmd := exec.Command("lsof", "-i", fmt.Sprintf(":%d", port), "-sTCP:LISTEN") out, err := cmd.Output() if err != nil { return 0, err } scanner := bufio.NewScanner(bytes.NewReader(out)) for scanner.Scan() { line := scanner.Text() if strings.Contains(line, "frpc") { parts := strings.Fields(line) if len(parts) >= 2 { if pid, err := strconv.Atoi(parts[1]); err == nil && pid > 0 { return pid, nil } } } } return 0, fmt.Errorf("未找到监听端口 %d 的 frpc 进程", port) } func (pm *ProcessManager) isProcessListeningOnPort(pid, port int) bool { cmd := exec.Command("lsof", "-p", strconv.Itoa(pid), "-a", "-i", fmt.Sprintf(":%d", port), "-sTCP:LISTEN") out, err := cmd.Output() if err != nil { return false } return strings.Contains(string(out), "LISTEN") } func (pm *ProcessManager) isFRPCProcess(pid int) (bool, string) { cmdlinePath := fmt.Sprintf("/proc/%d/cmdline", pid) if data, err := os.ReadFile(cmdlinePath); err == nil { cmd := strings.ReplaceAll(string(data), "\x00", " ") if strings.Contains(cmd, "frpc") { return true, cmd } } cmd := exec.Command("ps", "-p", strconv.Itoa(pid), "-o", "args=") out, err := cmd.Output() if err == nil { args := strings.TrimSpace(string(out)) if strings.Contains(args, "frpc") { return true, args } } return false, "" } // ================================================================ // 进程存活检测 // ================================================================ func (pm *ProcessManager) isProcessAlive(pid int) bool { if pid <= 0 { return false } if runtime.GOOS == "windows" { cmd := exec.Command("tasklist", "/FI", "PID eq", strconv.Itoa(pid)) output, err := cmd.CombinedOutput() if err != nil { return false } return strings.Contains(string(output), strconv.Itoa(pid)) } process, err := os.FindProcess(pid) if err != nil { return false } return process.Signal(syscall.Signal(0)) == nil } // ================================================================ // 实例检测(精确扫描) // ================================================================ // DetectFrpcInstances 检测系统中所有 frpc 实例 // 三重确认:PID存在 + 进程存活 + Executable/CmdLine匹配 func (pm *ProcessManager) DetectFrpcInstances() []FrpcInstance { var instances []FrpcInstance // 获取当前 frpc 可执行文件的绝对路径 targetExe := pm.frpcBinPath if !filepath.IsAbs(targetExe) { if abs, err := filepath.Abs(targetExe); err == nil { targetExe = abs } } if runtime.GOOS == "windows" { // Windows: 使用 tasklist + wmic 获取进程信息 cmd := exec.Command("tasklist", "/FO", "CSV", "/FI", "IMAGENAME eq frpc.exe") out, err := cmd.Output() if err != nil { log.Printf("[WARN] 检测 frpc 实例失败: %v", err) return instances } lines := strings.Split(string(out), "\n") for _, line := range lines { if !strings.Contains(line, "frpc.exe") { continue } parts := strings.Split(line, ",") if len(parts) < 2 { continue } pidStr := strings.Trim(parts[1], `"`) pid, err := strconv.Atoi(pidStr) if err != nil || pid <= 0 { continue } if !pm.isProcessAlive(pid) { continue } // 获取 cmdline(Windows 较难获取,先用进程名匹配) inst := FrpcInstance{ PID: pid, ExecPath: "frpc.exe", CmdLine: "frpc.exe", Owned: false, } // 检查是否 Owned ownedPid := pm.readPIDFile() if ownedPid == pid { inst.Owned = true } instances = append(instances, inst) } return instances } // Linux: 使用 pgrep + /proc 精确匹配 cmd := exec.Command("pgrep", "-f", "frpc") out, err := cmd.Output() if err != nil { // 没有找到进程,pgrep 返回非0,属于正常情况 return instances } pids := strings.Split(strings.TrimSpace(string(out)), "\n") for _, pidStr := range pids { pidStr = strings.TrimSpace(pidStr) if pidStr == "" { continue } pid, err := strconv.Atoi(pidStr) if err != nil || pid <= 0 { continue } if !pm.isProcessAlive(pid) { continue } // 获取命令行 cmdline, _ := os.ReadFile(fmt.Sprintf("/proc/%d/cmdline", pid)) cmdLineStr := strings.ReplaceAll(string(cmdline), "\x00", " ") // 获取可执行文件路径 exePath, _ := os.Readlink(fmt.Sprintf("/proc/%d/exe", pid)) // 获取父进程 PID stat, _ := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)) var parentPID int if len(stat) > 0 { // stat 格式: pid (comm) state ppid ... parts := strings.Fields(string(stat)) if len(parts) > 3 { parentPID, _ = strconv.Atoi(parts[3]) } } inst := FrpcInstance{ PID: pid, ParentPID: parentPID, ExecPath: exePath, CmdLine: cmdLineStr, Owned: false, } // 三重确认 Owned ownedPid := pm.readPIDFile() if ownedPid == pid { // 条件1: PID 匹配 if pm.isProcessAlive(pid) { // 条件2: 进程存在 // 条件3: 执行路径匹配 if exePath != "" && (strings.Contains(exePath, "frpc") || strings.Contains(cmdLineStr, "frpc")) { inst.Owned = true } } } instances = append(instances, inst) } return instances } // filterOwned 返回 Owned 实例 func (pm *ProcessManager) filterOwned(instances []FrpcInstance) []FrpcInstance { var result []FrpcInstance for _, inst := range instances { if inst.Owned { result = append(result, inst) } } return result } // filterUnknown 返回 Unknown 实例 func (pm *ProcessManager) filterUnknown(instances []FrpcInstance) []FrpcInstance { var result []FrpcInstance for _, inst := range instances { if !inst.Owned { result = append(result, inst) } } return result } // ================================================================ // 冲突检测 // ================================================================ type ConflictInfo struct { HasConflict bool Count int Owned []FrpcInstance Unknown []FrpcInstance } func (pm *ProcessManager) DetectConflict() ConflictInfo { instances := pm.DetectFrpcInstances() owned := pm.filterOwned(instances) unknown := pm.filterUnknown(instances) return ConflictInfo{ HasConflict: len(unknown) > 0 || len(owned) > 1, Count: len(instances), Owned: owned, Unknown: unknown, } } // ================================================================ // 实例健康检查 // ================================================================ func (pm *ProcessManager) isInstanceHealthy(inst FrpcInstance) bool { // 1. 进程必须存在 if !pm.isProcessAlive(inst.PID) { return false } // 2. 端口必须可连接 result := pm.CheckPort() if !result.Ready { return false } // 3. FRP 就绪(通过 stdout 或 admin API) stdout := pm.getLastOutput() if pm.isFRPReady(stdout) { return true } // 4. 尝试 admin API if info := pm.getFRPCStatus(inst.PID); info != nil && info.Version != "" { return true } return false } // ================================================================ // 进程清理 // ================================================================ func (pm *ProcessManager) killProcess(pid int) error { if pid <= 0 { return nil } if runtime.GOOS == "windows" { cmd := exec.Command("taskkill", "/F", "/PID", strconv.Itoa(pid)) return cmd.Run() } process, err := os.FindProcess(pid) if err != nil { return err } return process.Kill() } func (pm *ProcessManager) cleanupOrphans() { result := pm.CheckPort() if !result.Ready { return } pid := pm.readPIDFile() if pid > 0 && pm.isProcessAlive(pid) { return } log.Printf("[WARN] 检测到孤儿 frpc 进程 (端口 %d 被占用但无有效 PID),正在清理...", pm.adminPort) if runtime.GOOS == "windows" { exec.Command("taskkill", "/F", "/IM", "frpc.exe").Run() } else { exec.Command("pkill", "-f", "frpc").Run() } pm.deletePIDFile() time.Sleep(1 * time.Second) } // ================================================================ // PID 文件操作 // ================================================================ func (pm *ProcessManager) pidFilePath() string { return filepath.Join(pm.dataDir, "frpc.pid") } func (pm *ProcessManager) readPIDFile() int { data, err := os.ReadFile(pm.pidFilePath()) if err != nil { return 0 } pid, err := strconv.Atoi(strings.TrimSpace(string(data))) if err != nil || pid <= 0 { return 0 } return pid } func (pm *ProcessManager) writePIDFile(pid int) error { return os.WriteFile(pm.pidFilePath(), []byte(strconv.Itoa(pid)), 0644) } func (pm *ProcessManager) deletePIDFile() error { err := os.Remove(pm.pidFilePath()) if os.IsNotExist(err) { return nil } return err } // ================================================================ // stdout/stderr 捕获 // ================================================================ func (pm *ProcessManager) captureOutput(reader io.ReadCloser, isError bool) { defer reader.Close() scanner := bufio.NewScanner(reader) buf := make([]byte, 64*1024) scanner.Buffer(buf, 256*1024) var lines []string const maxLines = 20 for scanner.Scan() { line := scanner.Text() pm.outputMu.Lock() if isError { pm.lastError = line } else { pm.lastOutput = line } pm.outputMu.Unlock() if isError { lines = append(lines, line) if len(lines) > maxLines { lines = lines[1:] } pm.outputMu.Lock() pm.lastError = strings.Join(lines, "\n") pm.outputMu.Unlock() } else { lines = append(lines, line) if len(lines) > maxLines { lines = lines[1:] } pm.outputMu.Lock() pm.lastOutput = strings.Join(lines, "\n") pm.outputMu.Unlock() } } } func (pm *ProcessManager) getLastOutput() string { pm.outputMu.Lock() defer pm.outputMu.Unlock() return pm.lastOutput } func (pm *ProcessManager) getLastError() string { pm.outputMu.Lock() defer pm.outputMu.Unlock() return pm.lastError } // ================================================================ // FRPReady 检测 // ================================================================ func (pm *ProcessManager) isFRPReady(stdout string) bool { if strings.Contains(stdout, "start proxy success") || strings.Contains(stdout, "login to server success") { return true } return false } // ================================================================ // 状态计算 // ================================================================ func (pm *ProcessManager) computeState(pid int, result PortCheckResult) ProcessState { alive := pm.isProcessAlive(pid) portReady := result.Ready portErr := result.Err stdout := pm.getLastOutput() stderr := pm.getLastError() frpReady := pm.isFRPReady(stdout) pm.exitMu.Lock() exitCode := pm.exitCode pm.exitMu.Unlock() state := ProcessState{ PID: pid, Port: pm.adminPort, Alive: alive, PortReady: portReady, PortError: "", FRPReady: frpReady, StartedAt: pm.startTime, ExpectedStop: pm.expectedStop, LastOutput: stdout, LastError: stderr, ExitCode: exitCode, } if portErr != nil { state.PortError = portErr.Error() } // ============================================================ // 状态判定逻辑 // ============================================================ // 1. 如果是主动停止,直接返回 STOPPING if pm.expectedStop && alive { state.Phase = PhaseStopping return state } // 2. 进程不存在 if !alive { if pm.expectedStop { state.Phase = PhaseStopped return state } if exitCode != 0 && exitCode != -1 { state.Error = fmt.Sprintf("进程异常退出 (exit code: %d)", exitCode) if stderr != "" { state.Error += ": " + stderr } } else { state.Error = "进程意外退出" if stderr != "" { state.Error += ": " + stderr } } state.Phase = PhaseFailed return state } // 3. 进程存在,检查超时 if time.Since(pm.startTime) > StartupTimeout { state.Error = fmt.Sprintf("启动超时 (超过 %v)", StartupTimeout) if portErr != nil { state.Error += ": " + portErr.Error() } state.Phase = PhaseFailed return state } // 4. 冲突检测 conflict := pm.DetectConflict() if conflict.HasConflict { state.Phase = PhaseConflict state.Conflicts = append(conflict.Owned, conflict.Unknown...) if len(conflict.Unknown) > 0 { state.Error = fmt.Sprintf("检测到 %d 个未知 frpc 实例", len(conflict.Unknown)) } if len(conflict.Owned) > 1 { state.Error = fmt.Sprintf("检测到 %d 个 Owned frpc 实例(异常)", len(conflict.Owned)) } return state } // 5. 正常运行状态 if portReady && frpReady { state.Phase = PhaseRunning return state } // 6. 端口就绪但 FRP 未就绪 if portReady && !frpReady { state.Phase = PhaseDegraded state.Error = "端口已监听,但 frpc 未报告就绪" return state } // 7. 端口未就绪 if !portReady { if portErr != nil && strings.Contains(portErr.Error(), "permission denied") { state.Phase = PhaseDegraded state.Error = "端口检测权限不足: " + portErr.Error() return state } state.Phase = PhaseStarting if portErr != nil { state.Error = "等待端口就绪: " + portErr.Error() } return state } state.Phase = PhaseUnknown return state } // ================================================================ // 状态查询 // ================================================================ func (pm *ProcessManager) Status() (*ProcessState, error) { pid := pm.readPIDFile() result := pm.CheckPort() state := pm.computeState(pid, result) pm.setPhase(state.Phase) return &state, nil } // ================================================================ // API 回源 (2.7-preview 预留) // ================================================================ func (pm *ProcessManager) getFRPCStatus(pid int) *FRPCStatus { if pid <= 0 || pm.adminPort <= 0 { return nil } 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 { return nil } client := &http.Client{Timeout: APITimeout} resp, err := client.Do(req) if err != nil { return nil } defer resp.Body.Close() if resp.StatusCode != 200 { return nil } body, err := io.ReadAll(resp.Body) if err != nil { return nil } var status FRPCStatus if err := json.Unmarshal(body, &status); err != nil { return nil } return &status } // ================================================================ // 操作执行 // ================================================================ // Lock 由平台文件 (lock_*.go) 实现 func (pm *ProcessManager) Start(ctx context.Context) error { if err := pm.Lock(); err != nil { return fmt.Errorf("获取锁失败: %w", err) } defer pm.Unlock() pm.expectedStop = false // 1. 检测冲突 conflict := pm.DetectConflict() if conflict.HasConflict { log.Printf("[WARN] 检测到 %d 个 frpc 实例 (Owned: %d, Unknown: %d),进入 CONFLICT 状态", conflict.Count, len(conflict.Owned), len(conflict.Unknown)) // 清理 Unknown 实例 for _, inst := range conflict.Unknown { log.Printf("[INFO] 清理 Unknown frpc 实例 (PID: %d)", inst.PID) pm.killProcess(inst.PID) } // 如果有且仅有一个 Owned 实例,保留它 if len(conflict.Owned) == 1 { pid := conflict.Owned[0].PID log.Printf("[INFO] 保留 Owned frpc 实例 (PID: %d),清理完成", pid) pm.writePIDFile(pid) pm.setPhase(PhaseRunning) return nil } // 多个 Owned 或没有 Owned:全部清理,重新启动 if len(conflict.Owned) > 1 { log.Printf("[WARN] 多个 Owned 实例冲突,全部清理") for _, inst := range conflict.Owned { pm.killProcess(inst.PID) } } 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) { log.Printf("[INFO] frpc 已在运行 (PID: %d)", pid) pm.setPhase(PhaseRunning) return nil } pm.cleanupOrphans() } pm.startTime = time.Now() pm.expectedStop = false pm.exitMu.Lock() pm.exitCode = -1 pm.exitMu.Unlock() cmd := exec.CommandContext(ctx, pm.frpcBinPath, "-c", pm.configPath) setProcessAttributes(cmd) stdoutPipe, err := cmd.StdoutPipe() if err != nil { return fmt.Errorf("创建 stdout pipe 失败: %w", err) } stderrPipe, err := cmd.StderrPipe() if err != nil { return fmt.Errorf("创建 stderr pipe 失败: %w", err) } go pm.captureOutput(stdoutPipe, false) go pm.captureOutput(stderrPipe, true) if err := cmd.Start(); err != nil { return fmt.Errorf("启动 frpc 失败: %w", err) } if err := pm.writePIDFile(cmd.Process.Pid); err != nil { log.Printf("[WARN] 写入 PID 文件失败: %v", err) } log.Printf("[DEBUG] frpc 进程已启动,PID: %d", cmd.Process.Pid) pm.setPhase(PhaseStarting) go func() { err := cmd.Wait() var code int if err != nil { if exitErr, ok := err.(*exec.ExitError); ok { code = exitErr.ExitCode() } else { code = -1 } } else { code = 0 } pm.exitMu.Lock() pm.exitCode = code pm.exitMu.Unlock() log.Printf("[INFO] frpc 进程 (PID: %d) 已退出,退出码: %d", cmd.Process.Pid, code) if !pm.expectedStop && code != 0 { log.Printf("[WARN] frpc 进程异常退出 (exit code: %d)", code) pm.setPhase(PhaseFailed) } if !pm.expectedStop { pm.deletePIDFile() } }() // 等待端口就绪 for attempt := 0; attempt < StartupMaxAttempts; attempt++ { r := pm.CheckPort() if r.Ready { stdout := pm.getLastOutput() if pm.isFRPReady(stdout) { log.Printf("[INFO] frpc 启动成功 (PID: %d, 端口: %d),耗时 %dms", cmd.Process.Pid, pm.adminPort, attempt*int(StartupRetryDelay/time.Millisecond)) pm.setPhase(PhaseRunning) return nil } log.Printf("[DEBUG] 端口已就绪,等待 frpc 初始化...") } time.Sleep(StartupRetryDelay) } // 超时:检查进程状态 if pm.isProcessAlive(cmd.Process.Pid) { stderr := pm.getLastError() log.Printf("[WARN] frpc 启动超时 (PID: %d),当前 stderr: %s", cmd.Process.Pid, stderr) if info := pm.getFRPCStatus(cmd.Process.Pid); info != nil && info.Version != "" { log.Printf("[INFO] frpc admin API 可访问,版本: %s", info.Version) pm.writePIDFile(cmd.Process.Pid) pm.setPhase(PhaseRunning) return nil } pm.killProcess(cmd.Process.Pid) pm.deletePIDFile() pm.setPhase(PhaseFailed) return fmt.Errorf("frpc 启动超时: 进程存在但端口未就绪") } pm.deletePIDFile() pm.setPhase(PhaseFailed) return fmt.Errorf("frpc 启动失败: 进程已退出") } func (pm *ProcessManager) Stop(ctx context.Context) error { if err := pm.Lock(); err != nil { return fmt.Errorf("获取锁失败: %w", err) } defer pm.Unlock() pm.expectedStop = true pm.setPhase(PhaseStopping) return pm.stopLocked(ctx) } func (pm *ProcessManager) stopLocked(_ context.Context) error { pid := pm.readPIDFile() if pid <= 0 { if runtime.GOOS == "windows" { exec.Command("taskkill", "/F", "/IM", "frpc.exe").Run() } else { exec.Command("pkill", "-f", "frpc").Run() } pm.deletePIDFile() pm.setPhase(PhaseStopped) return nil } if !pm.isProcessAlive(pid) { pm.deletePIDFile() pm.setPhase(PhaseStopped) return nil } // 发送 SIGTERM process, err := os.FindProcess(pid) if err != nil { pm.deletePIDFile() pm.setPhase(PhaseStopped) return nil } if err := process.Signal(syscall.SIGTERM); err != nil { pm.deletePIDFile() pm.setPhase(PhaseStopped) return nil } // 等待进程退出 start := time.Now() for time.Since(start) < StopMaxWaitTime { if !pm.isProcessAlive(pid) { pm.deletePIDFile() pm.setPhase(PhaseStopped) return nil } time.Sleep(200 * time.Millisecond) } // 强制 kill if runtime.GOOS == "windows" { exec.Command("taskkill", "/F", "/PID", strconv.Itoa(pid)).Run() } else { process.Kill() } time.Sleep(500 * time.Millisecond) if pm.isProcessAlive(pid) { pm.setPhase(PhaseFailed) return fmt.Errorf("强制停止失败: 进程仍存活 (PID: %d)", pid) } pm.deletePIDFile() pm.setPhase(PhaseStopped) return nil } func (pm *ProcessManager) Restart(ctx context.Context) error { if err := pm.Lock(); err != nil { return fmt.Errorf("获取锁失败: %w", err) } defer pm.Unlock() if err := pm.stopLocked(ctx); err != nil { return fmt.Errorf("停止失败: %w", err) } time.Sleep(1 * time.Second) if err := pm.startLocked(ctx); err != nil { return fmt.Errorf("启动失败: %w", err) } return nil } // ================================================================ // 健康检查看门狗(由 main.go 调用) // ================================================================ func (pm *ProcessManager) StartHealthMonitor(ctx context.Context) { ticker := time.NewTicker(HealthCheckInterval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: pm.runHealthCheck() } } } func (pm *ProcessManager) runHealthCheck() { state, err := pm.Status() if err != nil { log.Printf("[WARN] 健康检查失败: %v", err) return } // 根据状态执行自动恢复 switch state.Phase { case PhaseConflict: log.Printf("[WARN] 检测到冲突,尝试自动恢复...") // 冲突恢复:清理未知实例 for _, inst := range state.Conflicts { if !inst.Owned { pm.killProcess(inst.PID) } } pm.setPhase(PhaseStarting) case PhaseDegraded: log.Printf("[WARN] frpc 处于降级状态,尝试恢复...") // 降级恢复:重启 pm.Restart(context.Background()) case PhaseFailed: if !state.ExpectedStop { log.Printf("[WARN] frpc 已失败,自动重启...") pm.Start(context.Background()) } } } // ================================================================ // 健康检查 // ================================================================ func (pm *ProcessManager) HealthCheck() map[string]interface{} { result := make(map[string]interface{}) result["admin_port"] = pm.adminPort result["phase"] = pm.CurrentPhase() state, err := pm.Status() if err != nil { result["error"] = err.Error() return result } result["pid"] = state.PID result["port"] = state.Port result["alive"] = state.Alive result["port_ready"] = state.PortReady result["frp_ready"] = state.FRPReady if state.PortError != "" { result["port_error"] = state.PortError } if state.Error != "" { result["error"] = state.Error } if state.Version != "" { result["version"] = state.Version } if len(state.Conflicts) > 0 { result["conflicts"] = state.Conflicts } return result }