lxh2875931338/frpc-console
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

首次优化的进程管理工具

Browse Source
This commit is contained in:
lxh2875931338
2026-08-10 21:47:24 +08:00
parent 399c13b732
commit 2950e40153
1 changed files with 387 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/process/manager.go
+387 -57
View File
@@ -1,10 +1,7 @@
// internal/process/manager.go
// frpc-console 进程管理模块
// 2.6-preview: 状态机驱动 + 端口检测 + 单实例锁定
// 作者留：这个版本可以用了，已经明显看到进程出来发现端口不对立马被杀掉的状态了
// 但是稍显粗暴，差点意思，到时候让Deepseek稍微软化下就好了
// 但是现阶段Preview还不能用，再等等
// 2.6-preview: 状态机驱动 + 冲突检测 + 生命周期管理
// 2.7-preview: 软化版 - 精确控制 + CONFLICT/STOPPING 状态
package process
@@ -44,6 +41,8 @@ const (
StartupMaxAttempts = 50
StartupRetryDelay = 200 * time.Millisecond
StartupTimeout = 10 * time.Second
HealthCheckInterval = 15 * time.Second
)
// ================================================================
@@ -59,6 +58,8 @@ const (
PhaseDegraded ProcessPhase = "DEGRADED"
PhaseFailed ProcessPhase = "FAILED"
PhaseStopped ProcessPhase = "STOPPED"
PhaseStopping ProcessPhase = "STOPPING" // 正在停止中
PhaseConflict ProcessPhase = "CONFLICT" // 检测到多实例冲突
)
// ================================================================
@@ -70,6 +71,14 @@ type PortCheckResult struct {
Err error
}
type FrpcInstance struct {
PID int
ParentPID int
ExecPath string
CmdLine string
Owned bool // 是否由当前 ProcessManager 管理
}
type ProcessState struct {
Phase ProcessPhase `json:"phase"`
PID int `json:"pid"`
@@ -88,8 +97,16 @@ type ProcessState struct {
LastOutput string `json:"last_output,omitempty"`
LastError string `json:"last_error,omitempty"`
Error string `json:"error,omitempty"`
// 冲突检测信息
Conflicts []FrpcInstance `json:"conflicts,omitempty"`
}
var (
globalManager *ProcessManager
globalMu sync.Mutex
)
type PortStatus struct {
Port int `json:"port"`
Occupied bool `json:"occupied"`
@@ -131,24 +148,28 @@ type ProcessManager struct {
lockFile *os.File
locked bool
}
var (
globalManager *ProcessManager
globalMu sync.Mutex
)
// 当前状态缓存
currentPhase ProcessPhase
statusMu sync.RWMutex
// 取消函数（用于停止健康检查）
cancelFunc context.CancelFunc
}
// ================================================================
// 构造函数
// ================================================================
func NewManager(dataDir, configPath, frpcBinPath string) *ProcessManager {
return &ProcessManager{
dataDir: dataDir,
configPath: configPath,
frpcBinPath: frpcBinPath,
adminPort: 0,
pm := &ProcessManager{
dataDir: dataDir,
configPath: configPath,
frpcBinPath: frpcBinPath,
adminPort: 0,
currentPhase: PhaseUnknown,
}
return pm
}
func SetGlobalManager(pm *ProcessManager) {
@@ -167,6 +188,21 @@ 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
}
}
// ================================================================
// 配置读取
// ================================================================
@@ -469,6 +505,205 @@ func (pm *ProcessManager) isProcessAlive(pid int) bool {
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
}
// ================================================================
// 进程清理
// ================================================================
@@ -490,34 +725,6 @@ func (pm *ProcessManager) killProcess(pid int) error {
return process.Kill()
}
// killAllFrpc 强制杀死所有 frpc 进程（包括孤儿）
func (pm *ProcessManager) killAllFrpc() {
log.Printf("[INFO] 清理所有 frpc 进程...")
if runtime.GOOS == "windows" {
exec.Command("taskkill", "/F", "/IM", "frpc.exe").Run()
} else {
exec.Command("pkill", "-9", "-f", "frpc").Run()
}
time.Sleep(1 * time.Second)
pm.deletePIDFile()
}
// countFrpcProcesses 统计当前 frpc 进程数量
func (pm *ProcessManager) countFrpcProcesses() int {
if runtime.GOOS == "windows" {
cmd := exec.Command("tasklist", "/FI", "IMAGENAME eq frpc.exe")
out, _ := cmd.CombinedOutput()
return strings.Count(string(out), "frpc.exe")
}
cmd := exec.Command("pgrep", "-c", "-f", "frpc")
out, err := cmd.Output()
if err != nil {
return 0
}
count, _ := strconv.Atoi(strings.TrimSpace(string(out)))
return count
}
func (pm *ProcessManager) cleanupOrphans() {
result := pm.CheckPort()
if !result.Ready {
@@ -673,7 +880,17 @@ func (pm *ProcessManager) computeState(pid int, result PortCheckResult) ProcessS
state.PortError = portErr.Error()
}
// ---- 状态判定逻辑 ----
// ============================================================
// 状态判定逻辑
// ============================================================
// 1. 如果是主动停止，直接返回 STOPPING
if pm.expectedStop && alive {
state.Phase = PhaseStopping
return state
}
// 2. 进程不存在
if !alive {
if pm.expectedStop {
state.Phase = PhaseStopped
@@ -695,7 +912,7 @@ func (pm *ProcessManager) computeState(pid int, result PortCheckResult) ProcessS
return state
}
// ---- alive == true ----
// 3. 进程存在，检查超时
if time.Since(pm.startTime) > StartupTimeout {
state.Error = fmt.Sprintf("启动超时 (超过 %v)", StartupTimeout)
if portErr != nil {
@@ -705,22 +922,34 @@ func (pm *ProcessManager) computeState(pid int, result PortCheckResult) ProcessS
return state
}
if portReady && frpReady {
state.Phase = PhaseRunning
// 检测多进程
if count := pm.countFrpcProcesses(); count > 1 {
state.Phase = PhaseDegraded
state.Error = fmt.Sprintf("检测到 %d 个 frpc 进程，可能存在多开", count)
// 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
@@ -746,6 +975,7 @@ func (pm *ProcessManager) Status() (*ProcessState, error) {
pid := pm.readPIDFile()
result := pm.CheckPort()
state := pm.computeState(pid, result)
pm.setPhase(state.Phase)
return &state, nil
}
@@ -796,12 +1026,43 @@ func (pm *ProcessManager) Start(ctx context.Context) error {
}
defer pm.Unlock()
// 先杀所有 frpc 进程（避免多开）
pm.killAllFrpc()
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)
}
@@ -811,6 +1072,7 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error {
pid := pm.readPIDFile()
if pid > 0 && pm.isProcessAlive(pid) {
log.Printf("[INFO] frpc 已在运行 (PID: %d)", pid)
pm.setPhase(PhaseRunning)
return nil
}
pm.cleanupOrphans()
@@ -847,6 +1109,8 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error {
}
log.Printf("[DEBUG] frpc 进程已启动，PID: %d", cmd.Process.Pid)
pm.setPhase(PhaseStarting)
go func() {
err := cmd.Wait()
var code int
@@ -866,12 +1130,14 @@ 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.expectedStop {
pm.deletePIDFile()
}
}()
// 等待端口就绪
for attempt := 0; attempt < StartupMaxAttempts; attempt++ {
r := pm.CheckPort()
if r.Ready {
@@ -879,6 +1145,7 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error {
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 初始化...")
@@ -893,15 +1160,17 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error {
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 启动失败: 进程已退出")
}
@@ -912,6 +1181,7 @@ func (pm *ProcessManager) Stop(ctx context.Context) error {
defer pm.Unlock()
pm.expectedStop = true
pm.setPhase(PhaseStopping)
return pm.stopLocked(ctx)
}
@@ -924,33 +1194,41 @@ func (pm *ProcessManager) stopLocked(_ context.Context) error {
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 {
@@ -959,10 +1237,12 @@ func (pm *ProcessManager) stopLocked(_ context.Context) error {
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
}
@@ -982,6 +1262,54 @@ func (pm *ProcessManager) Restart(ctx context.Context) error {
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())
}
}
}
// ================================================================
// 健康检查
// ================================================================
@@ -989,15 +1317,14 @@ func (pm *ProcessManager) Restart(ctx context.Context) error {
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["phase"] = "ERROR"
result["error"] = err.Error()
return result
}
result["phase"] = state.Phase
result["pid"] = state.PID
result["port"] = state.Port
result["alive"] = state.Alive
@@ -1012,6 +1339,9 @@ func (pm *ProcessManager) HealthCheck() map[string]interface{} {
if state.Version != "" {
result["version"] = state.Version
}
if len(state.Conflicts) > 0 {
result["conflicts"] = state.Conflicts
}
return result
}