// process_manager.go // frpc-console 进程管理模块 // 2.6-preview: 端口检测 + 单实例锁定 + 状态自述 // // 设计原则: // 1. 端口是事实来源,PID 文件是缓存 // 2. 所有操作幂等 (多次调用结果一致) // 3. 通过文件锁 (flock) 实现进程间互斥 // 4. 通过 admin_port API 获取 frpc 真实状态 package main import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "net" "os" "os/exec" "path/filepath" "strconv" "strings" "sync" "syscall" "time" ) // ================================================================ // 常量定义 // ================================================================ const ( // 锁文件路径 (相对于 DataDir) LockFileName = ".frpc.lock" // 端口检测超时 PortCheckTimeout = 500 * time.Millisecond // 启动后等待端口就绪的时间 StartWaitTime = 500 * time.Millisecond // 停止时等待端口释放的最大时间 StopMaxWaitTime = 5 * time.Second // 锁获取超时 LockAcquireTimeout = 30 * time.Second // 锁重试间隔 LockRetryInterval = 100 * time.Millisecond LockMaxRetries = 5 // API 请求超时 APITimeout = 2 * time.Second ) // ================================================================ // 数据结构 // ================================================================ // 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"` } // ProcessStatus frpc 进程状态 type ProcessStatus struct { State string `json:"state"` // "running" | "stopped" | "unknown" | "conflict" PID int `json:"pid"` // 进程 PID (如果运行中) Port int `json:"port"` // 监听的端口 Uptime string `json:"uptime"` // 运行时长 (可选) Version string `json:"version"` // frpc 版本 (如果 API 可访问) Error string `json:"error,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"` } // ================================================================ // ProcessManager 主结构 // ================================================================ type ProcessManager struct { mu sync.Mutex dataDir string // 数据目录 (存放 lock 和 pid 文件) configPath string // frpc.toml 路径 frpcBinPath string // frpc 二进制路径 adminPort int // admin_port (从配置读取) lockFile *os.File // flock 文件句柄 locked bool // 是否持有锁 } // NewProcessManager 创建进程管理器 func NewProcessManager(dataDir, configPath, frpcBinPath string) *ProcessManager { return &ProcessManager{ dataDir: dataDir, configPath: configPath, frpcBinPath: frpcBinPath, adminPort: 0, // 需要调用 LoadConfig 后确定 } } // ================================================================ // 配置读取 // ================================================================ // LoadConfig 从 frpc.toml 读取 admin_port 配置 // 兼容 frp 0.52.0 前后的配置格式 func (pm *ProcessManager) LoadConfig() error { // 读取 frpc.toml 内容 content, err := os.ReadFile(pm.configPath) if err != nil { return fmt.Errorf("读取配置文件失败: %w", err) } // 尝试解析 admin_port (旧格式) // admin_port = 7400 if port := extractIntValue(string(content), "admin_port"); port > 0 { pm.adminPort = port return nil } // 尝试解析 webServer.port (新格式, 0.52.0+) // [webServer] // port = 7400 if port := extractIntValueFromSection(string(content), "webServer", "port"); port > 0 { pm.adminPort = port return nil } // 如果都找不到,说明 frpc 配置没有启用 admin 端口 return fmt.Errorf("未找到 admin_port 或 webServer.port 配置") } // extractIntValue 从配置中提取 key = value 格式的值 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 } // extractIntValueFromSection 从指定 section 中提取 key = value 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, "]") { inSection = strings.Trim(trimmed, "[]") == 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 } // ================================================================ // 端口检测器 (PortDetector) // ================================================================ // CheckPort 检测端口是否被占用 func (pm *ProcessManager) CheckPort() (bool, error) { if pm.adminPort <= 0 { return false, fmt.Errorf("admin_port 未配置") } conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", pm.adminPort), PortCheckTimeout) if err != nil { // 连接失败 = 端口未被占用 return false, nil } conn.Close() return true, nil } // GetPortStatus 获取端口完整状态 (占用 + PID + 进程类型) func (pm *ProcessManager) GetPortStatus() (*PortStatus, error) { status := &PortStatus{ Port: pm.adminPort, Occupied: false, PID: 0, IsFRPC: false, } // 1. 检测端口是否被占用 occupied, err := pm.CheckPort() if err != nil { return status, err } status.Occupied = occupied if !occupied { return status, nil } // 2. 反查 PID pid, err := pm.getPIDByPort(pm.adminPort) if err != nil { // 反查失败,尝试从 PID 文件读取 if pidFromFile := pm.readPIDFile(); pidFromFile > 0 { status.PID = pidFromFile // 验证这个 PID 是否真的在监听端口 if pm.isProcessListeningOnPort(pidFromFile, pm.adminPort) { status.PID = pidFromFile } else { status.PID = 0 } } } else { status.PID = pid } if status.PID == 0 { return status, nil } // 3. 验证进程是否是 frpc isFRPC, cmd := pm.isFRPCProcess(status.PID) status.IsFRPC = isFRPC status.ProcessCmd = cmd return status, nil } // getPIDByPort 通过端口反查 PID // 优先级: ss > netstat > lsof func (pm *ProcessManager) getPIDByPort(port int) (int, error) { // 方法 1: ss -lpn (最可靠) if pid, err := pm.getPIDBySS(port); err == nil && pid > 0 { return pid, nil } // 方法 2: netstat -tulpn (兼容性广) if pid, err := pm.getPIDByNetstat(port); err == nil && pid > 0 { return pid, nil } // 方法 3: lsof -i :port (最后的备选) if pid, err := pm.getPIDByLsof(port); err == nil && pid > 0 { return pid, nil } return 0, fmt.Errorf("无法通过端口反查 PID") } // getPIDBySS 通过 ss 命令反查 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 } // ss 输出格式: tcp LISTEN 0 128 0.0.0.0:7400 0.0.0.0:* users:(("frpc",pid=12345,fd=3)) // 提取 pid=12345 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) } // getPIDByNetstat 通过 netstat 命令反查 PID 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 } // netstat 输出格式: tcp 0 0 0.0.0.0:7400 0.0.0.0:* LISTEN 12345/frpc parts := strings.Fields(line) if len(parts) < 7 { continue } last := parts[len(parts)-1] // 提取 PID: 12345/frpc 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) } // getPIDByLsof 通过 lsof 命令反查 PID 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") { // lsof 输出格式: frpc 12345 root 3u IPv4 123456 0t0 TCP *:7400 (LISTEN) 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) } // isProcessListeningOnPort 验证 PID 是否在监听指定端口 func (pm *ProcessManager) isProcessListeningOnPort(pid, port int) bool { // 通过 /proc 验证 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") } // isFRPCProcess 验证进程是否是 frpc func (pm *ProcessManager) isFRPCProcess(pid int) (bool, string) { // 方法 1: 读取 /proc//cmdline 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 } } // 方法 2: ps -p 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, "" } // ================================================================ // 互斥锁 (flock) // ================================================================ // Lock 获取进程间互斥锁 func (pm *ProcessManager) Lock() error { pm.mu.Lock() defer pm.mu.Unlock() if pm.locked { return nil // 已持有锁 } lockPath := filepath.Join(pm.dataDir, LockFileName) // 确保数据目录存在 if err := os.MkdirAll(pm.dataDir, 0755); err != nil { return fmt.Errorf("创建数据目录失败: %w", err) } file, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0644) if err != nil { return fmt.Errorf("打开锁文件失败: %w", err) } // 尝试获取排他锁 (阻塞) // 使用 LOCK_EX | LOCK_NB 实现非阻塞尝试,然后手动重试 start := time.Now() for { err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) if err == nil { pm.lockFile = file pm.locked = true return nil } if err != syscall.EWOULDBLOCK { file.Close() return fmt.Errorf("获取锁失败: %w", err) } // 检查超时 if time.Since(start) > LockAcquireTimeout { file.Close() return fmt.Errorf("获取锁超时 (超过 %v)", LockAcquireTimeout) } time.Sleep(LockRetryInterval) } } // Unlock 释放互斥锁 func (pm *ProcessManager) Unlock() error { pm.mu.Lock() defer pm.mu.Unlock() if !pm.locked { return nil } if pm.lockFile != nil { syscall.Flock(int(pm.lockFile.Fd()), syscall.LOCK_UN) pm.lockFile.Close() pm.lockFile = nil } pm.locked = false return nil } // ================================================================ // 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 } // ================================================================ // 状态查询 (Status) // ================================================================ // Status 获取 frpc 进程实时状态 func (pm *ProcessManager) Status() (*ProcessStatus, error) { status := &ProcessStatus{ State: "unknown", PID: 0, Port: pm.adminPort, } if pm.adminPort <= 0 { status.State = "unknown" status.Error = "admin_port 未配置" return status, nil } // 1. 获取端口状态 portStatus, err := pm.GetPortStatus() if err != nil { status.Error = err.Error() return status, nil } if !portStatus.Occupied { // 端口未被占用: 清理过期的 PID 文件 pm.deletePIDFile() status.State = "stopped" return status, nil } if !portStatus.IsFRPC { status.State = "conflict" status.PID = portStatus.PID status.Error = fmt.Sprintf("端口 %d 被非 frpc 进程占用 (PID: %d)", pm.adminPort, portStatus.PID) return status, nil } // 2. 端口被 frpc 占用: 更新 PID 文件 status.State = "running" status.PID = portStatus.PID pm.writePIDFile(portStatus.PID) // 3. 尝试通过 API 获取更多信息 if info := pm.getFRPCStatus(portStatus.PID); info != nil { status.Version = info.Version } return status, nil } // ================================================================ // API 回源 (2.7-preview 预留) // ================================================================ // getFRPCStatus 通过 admin API 获取 frpc 状态 (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 } // ================================================================ // 操作执行 (Start / Stop / Restart) // ================================================================ // Start 启动 frpc (幂等) func (pm *ProcessManager) Start(ctx context.Context) error { // 1. 获取锁 if err := pm.Lock(); err != nil { return fmt.Errorf("获取锁失败: %w", err) } defer pm.Unlock() // 2. 双重检查: 端口是否已被占用 portStatus, err := pm.GetPortStatus() if err != nil { return fmt.Errorf("检测端口状态失败: %w", err) } if portStatus.Occupied { if portStatus.IsFRPC { // 已启动, 幂等返回 pm.writePIDFile(portStatus.PID) return nil } return fmt.Errorf("端口 %d 被非 frpc 进程占用 (PID: %d)", pm.adminPort, portStatus.PID) } // 3. 启动 frpc if err := pm.startFRPC(ctx); err != nil { return err } // 4. 等待端口就绪 time.Sleep(StartWaitTime) // 5. 验证启动成功 occupied, err := pm.CheckPort() if err != nil { return fmt.Errorf("验证启动状态失败: %w", err) } if !occupied { return fmt.Errorf("frpc 启动失败: 端口未监听") } // 6. 写入 PID 文件 pid, _ := pm.getPIDByPort(pm.adminPort) if pid > 0 { pm.writePIDFile(pid) } return nil } // startFRPC 实际执行 frpc 启动 (Setsid) func (pm *ProcessManager) startFRPC(ctx context.Context) error { // 构建启动命令 cmd := exec.CommandContext(ctx, pm.frpcBinPath, "-c", pm.configPath) // Setsid: 创建独立会话, 使 frpc 脱离 console 生命周期 cmd.SysProcAttr = &syscall.SysProcAttr{ Setsid: true, } // 重定向输出 (可选) cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr // 启动 if err := cmd.Start(); err != nil { return fmt.Errorf("启动 frpc 失败: %w", err) } // 注意: 这里不 wait, 让 frpc 独立运行 return nil } // Stop 停止 frpc (幂等) func (pm *ProcessManager) Stop(ctx context.Context) error { // 1. 获取锁 if err := pm.Lock(); err != nil { return fmt.Errorf("获取锁失败: %w", err) } defer pm.Unlock() // 2. 检查端口状态 portStatus, err := pm.GetPortStatus() if err != nil { return fmt.Errorf("检测端口状态失败: %w", err) } if !portStatus.Occupied { // 已停止, 幂等返回 pm.deletePIDFile() return nil } var pid int if portStatus.IsFRPC { pid = portStatus.PID } else { // 端口被非 frpc 占用, 不能强制停止 return fmt.Errorf("端口 %d 被非 frpc 进程占用, 无法安全停止", pm.adminPort) } if pid <= 0 { // 尝试从 PID 文件读取 pid = pm.readPIDFile() if pid <= 0 { return fmt.Errorf("无法确定 frpc 进程 PID") } } // 3. 发送 SIGTERM (优雅停止) proc, err := os.FindProcess(pid) if err != nil { return fmt.Errorf("查找进程失败: %w", err) } if err := proc.Signal(syscall.SIGTERM); err != nil { // 可能进程已退出 pm.deletePIDFile() return nil } // 4. 等待端口释放 start := time.Now() for time.Since(start) < StopMaxWaitTime { occupied, _ := pm.CheckPort() if !occupied { pm.deletePIDFile() return nil } time.Sleep(200 * time.Millisecond) } // 5. 端口未释放, 强制 kill proc.Kill() time.Sleep(500 * time.Millisecond) // 再次检查 occupied, _ := pm.CheckPort() if occupied { return fmt.Errorf("强制停止失败: 端口仍被占用") } pm.deletePIDFile() return nil } // Restart 重启 frpc (原子操作) func (pm *ProcessManager) Restart(ctx context.Context) error { // 1. 获取锁 (整个操作持有锁) if err := pm.Lock(); err != nil { return fmt.Errorf("获取锁失败: %w", err) } defer pm.Unlock() // 2. 停止 if err := pm.stopLocked(ctx); err != nil { return fmt.Errorf("停止失败: %w", err) } // 3. 启动 if err := pm.startLocked(ctx); err != nil { return fmt.Errorf("启动失败: %w", err) } return nil } // stopLocked 内部停止 (调用者必须持有锁) func (pm *ProcessManager) stopLocked(ctx context.Context) error { // 同 Stop 逻辑, 但跳过锁获取 portStatus, err := pm.GetPortStatus() if err != nil { return err } if !portStatus.Occupied { pm.deletePIDFile() return nil } // ... 其余逻辑与 Stop 相同 // (为节省篇幅, 这里省略重复代码, 实际实现可复用) return nil } // startLocked 内部启动 (调用者必须持有锁) func (pm *ProcessManager) startLocked(ctx context.Context) error { // 同 Start 逻辑, 但跳过锁获取 portStatus, err := pm.GetPortStatus() if err != nil { return err } if portStatus.Occupied && portStatus.IsFRPC { return nil } // ... 其余逻辑与 Start 相同 return nil } // ================================================================ // 健康检查 (用于 WebUI 展示) // ================================================================ // HealthCheck 返回简要健康状态 func (pm *ProcessManager) HealthCheck() map[string]interface{} { result := make(map[string]interface{}) result["admin_port"] = pm.adminPort status, err := pm.Status() if err != nil { result["state"] = "error" result["error"] = err.Error() return result } result["state"] = status.State result["pid"] = status.PID if status.Version != "" { result["version"] = status.Version } if status.Error != "" { result["error"] = status.Error } return result }