// process_manager.go // frpc-console 进程管理模块 // 2.6-preview: 端口检测 + 单实例锁定 + 状态自述 package main import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "net" "net/http" "os" "os/exec" "path/filepath" "strconv" "strings" "sync" "syscall" "time" "golang.org/x/sys/unix" ) // ================================================================ // 常量定义 // ================================================================ const ( LockFileName = ".frpc.lock" PortCheckTimeout = 500 * time.Millisecond StartWaitTime = 500 * time.Millisecond StopMaxWaitTime = 5 * time.Second LockAcquireTimeout = 30 * time.Second LockRetryInterval = 100 * time.Millisecond APITimeout = 2 * time.Second ) // ================================================================ // 数据结构 // ================================================================ 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"` } type ProcessStatus struct { State string `json:"state"` PID int `json:"pid"` Port int `json:"port"` Uptime string `json:"uptime,omitempty"` Version string `json:"version,omitempty"` Error string `json:"error,omitempty"` } 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 configPath string frpcBinPath string adminPort int lockFile *os.File locked bool } func NewProcessManager(dataDir, configPath, frpcBinPath string) *ProcessManager { return &ProcessManager{ dataDir: dataDir, configPath: configPath, frpcBinPath: frpcBinPath, adminPort: 0, } } // ================================================================ // 配置读取 // ================================================================ func (pm *ProcessManager) LoadConfig() error { content, err := os.ReadFile(pm.configPath) if err != nil { return fmt.Errorf("读取配置文件失败: %w", err) } if port := extractIntValue(string(content), "admin_port"); port > 0 { pm.adminPort = port return nil } if port := extractIntValueFromSection(string(content), "webServer", "port"); port > 0 { pm.adminPort = port return nil } return fmt.Errorf("未找到 admin_port 或 webServer.port 配置") } 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, "]") { 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 } // ================================================================ // 端口检测 // ================================================================ 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 } func (pm *ProcessManager) GetPortStatus() (*PortStatus, error) { status := &PortStatus{Port: pm.adminPort, Occupied: false, PID: 0, IsFRPC: false} occupied, err := pm.CheckPort() if err != nil { return status, err } status.Occupied = occupied if !occupied { return status, nil } 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, "" } // ================================================================ // 互斥锁 (flock) // ================================================================ 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) } start := time.Now() for { err := unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.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) } } func (pm *ProcessManager) Unlock() error { pm.mu.Lock() defer pm.mu.Unlock() if !pm.locked { return nil } if pm.lockFile != nil { unix.Flock(int(pm.lockFile.Fd()), unix.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 } // ================================================================ // 状态查询 // ================================================================ func (pm *ProcessManager) Status() (*ProcessStatus, error) { status := &ProcessStatus{State: "unknown", PID: 0, Port: pm.adminPort} if pm.adminPort <= 0 { status.Error = "admin_port 未配置" return status, nil } portStatus, err := pm.GetPortStatus() if err != nil { status.Error = err.Error() return status, nil } if !portStatus.Occupied { 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 } status.State = "running" status.PID = portStatus.PID pm.writePIDFile(portStatus.PID) if info := pm.getFRPCStatus(portStatus.PID); info != nil { status.Version = info.Version } return status, 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 } // ================================================================ // 操作执行 (Start / Stop / Restart) // ================================================================ func (pm *ProcessManager) Start(ctx context.Context) error { if err := pm.Lock(); err != nil { return fmt.Errorf("获取锁失败: %w", err) } defer pm.Unlock() return pm.startLocked(ctx) } func (pm *ProcessManager) startLocked(ctx context.Context) error { 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) } cmd := exec.CommandContext(ctx, pm.frpcBinPath, "-c", pm.configPath) // 根据平台设置 SysProcAttr setProcessAttributes(cmd) cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr if err := cmd.Start(); err != nil { return fmt.Errorf("启动 frpc 失败: %w", err) } time.Sleep(StartWaitTime) occupied, err := pm.CheckPort() if err != nil { return fmt.Errorf("验证启动状态失败: %w", err) } if !occupied { return fmt.Errorf("frpc 启动失败: 端口未监听") } pid, _ := pm.getPIDByPort(pm.adminPort) if pid > 0 { pm.writePIDFile(pid) } return nil } func (pm *ProcessManager) Stop(ctx context.Context) error { if err := pm.Lock(); err != nil { return fmt.Errorf("获取锁失败: %w", err) } defer pm.Unlock() return pm.stopLocked(ctx) } func (pm *ProcessManager) stopLocked(ctx context.Context) error { 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 { return fmt.Errorf("端口 %d 被非 frpc 进程占用, 无法安全停止", pm.adminPort) } if pid <= 0 { pid = pm.readPIDFile() if pid <= 0 { return fmt.Errorf("无法确定 frpc 进程 PID") } } proc, err := os.FindProcess(pid) if err != nil { pm.deletePIDFile() return nil } if err := proc.Signal(syscall.SIGTERM); err != nil { pm.deletePIDFile() return nil } start := time.Now() for time.Since(start) < StopMaxWaitTime { occupied, _ := pm.CheckPort() if !occupied { pm.deletePIDFile() return nil } time.Sleep(200 * time.Millisecond) } proc.Kill() time.Sleep(500 * time.Millisecond) if occupied, _ := pm.CheckPort(); occupied { return fmt.Errorf("强制停止失败: 端口仍被占用") } pm.deletePIDFile() 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) } if err := pm.startLocked(ctx); err != nil { return fmt.Errorf("启动失败: %w", err) } return nil } // ================================================================ // 健康检查 // ================================================================ func (pm *ProcessManager) HealthCheck() map[string]interface{} { result := map[string]interface{}{"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 }