feat: per-forward worker 模型 — 每条转发独立 frpc 配置+独立进程

核心改动(ring 协议不变,Task 本来就是 {Local,Remote,Link} 三元组):
- process.WorkerKey: worker 键改为 'local~remote~port' 三元组
- renderForward: 每条 forward 渲染只含 1 个 proxy 的独立 frpc 配置
  (替代 renderRemote 把该 remote 全部 forwards 塞进一个进程)
- ClaimFn/RevokeFn: 认领/撤销只操作这一条 forward 自己的进程
- SyncWorkers/auto-start: 只拉起 localOnly 转发的独立进程,
  集群转发由 ring 认领节点拉起 (消除三台争抢 proxy already exists)
- RemoteStatus/StartRemote/StopRemote/RestartRemote: remote 级聚合辅助,
  保持 /profiles API 形状不变, 前端零改动
- handleStatus/logs: 按 worker key 解析真实 remote 名

效果: 三台节点的 frpc 各自只注册自己拥有的 proxy, 单条转发故障不再波及兄弟
This commit is contained in:
JianFeeeee
2026-08-24 01:42:58 +08:00
parent 2292ee7f3a
commit e3a978888a
9 changed files with 340 additions and 101 deletions

View File

@ -8,6 +8,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"flag" "flag"
"fmt"
"log" "log"
"net" "net"
"net/http" "net/http"
@ -90,7 +91,7 @@ func main() {
} }
return fallbackBin return fallbackBin
}, },
Render: renderRemote(st), Render: renderForward(st),
AutoRestart: func(name string) bool { AutoRestart: func(name string) bool {
s, _ := st.Settings() s, _ := st.Settings()
if !s.RestartOnExit { if !s.RestartOnExit {
@ -172,44 +173,32 @@ func main() {
// disabled (by RevokeFn); clear it so renderRemote renders the // disabled (by RevokeFn); clear it so renderRemote renders the
// proxy back in. No-op for a fresh claim. // proxy back in. No-op for a fresh claim.
_ = st.SetLinkDisabled(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort, false) _ = st.SetLinkDisabled(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort, false)
// Start (or restart) the worker for the remote we just claimed. // Start (or restart) the per-forward worker for exactly this link.
// Restart re-renders the frpc config with all of this remote's // Each forward has its own frpc process (keyed by the forward
// links (including the just-upserted one) and respawns the process. // triple); restarting only this key leaves sibling forwards'
// We deliberately do NOT iterate every stored remote here: this // processes untouched.
// node's store may hold remotes for cluster forwards it does NOT
// own (the submitter persists the whole canvas), and starting
// those here would duplicate workers across the cluster.
if tk.Remote.Enabled { if tk.Remote.Enabled {
if _, has := pm.Status(tk.Remote.Name); has { key := process.WorkerKey(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort)
_ = pm.Restart(tk.Remote.Name) if _, has := pm.Status(key); has {
_ = pm.Restart(key)
} else { } else {
_ = pm.Start(tk.Remote.Name) _ = pm.Start(key)
} }
} }
log.Printf("ring[%s] claimed task %s: %s→%s:%d", selfID, tk.ID, tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort) log.Printf("ring[%s] claimed task %s: %s→%s:%d", selfID, tk.ID, tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort)
return nil return nil
}, },
// Revoke cancels the forward NON-destructively: flag the exact link // Revoke cancels the forward NON-destructively: stop THIS forward's
// disabled (so renderRemote drops just its proxy and a later forward- // own worker process (per-forward model — no siblings share it) and
// centric start can re-enable it) and right-size the worker — stop it // flag the exact link disabled so a later forward-centric start can
// when the remote has no remaining enabled forwards, else restart so // re-enable it. The old delete-local/remote/drop-all-matching-links
// sibling proxies survive. The old delete-local/remote/drop-all- // form served the topology-derived-canvas model but broke per-forward
// matching-links form served the topology-derived-canvas model but // stop/start: it wiped sibling forwards sharing the local or remote.
// broke per-forward stop/start: it wiped sibling forwards sharing the
// local or remote (group stop) and made re-start fail (link gone).
RevokeFn: func(ctx context.Context, tk *cluster.Task) error { RevokeFn: func(ctx context.Context, tk *cluster.Task) error {
_ = st.SetLinkDisabled(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort, true) _ = st.SetLinkDisabled(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort, true)
fwd, _ := st.LinksForRemote(tk.Remote.Name) key := process.WorkerKey(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort)
enabled := 0 if _, running := pm.Status(key); running {
for _, f := range fwd { _ = pm.Stop(key)
if !f.Disabled {
enabled++
}
}
if enabled == 0 {
_ = pm.Stop(tk.Remote.Name)
} else if _, running := pm.Status(tk.Remote.Name); running {
_ = pm.Restart(tk.Remote.Name)
} }
log.Printf("ring[%s] revoked task %s: %s→%s:%d", selfID, tk.ID, tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort) log.Printf("ring[%s] revoked task %s: %s→%s:%d", selfID, tk.ID, tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort)
return nil return nil
@ -297,53 +286,72 @@ func main() {
return path, ver, nil return path, ver, nil
}, },
SyncWorkers: func() { SyncWorkers: func() {
// Only forwards that stay on this node (localOnly) get their worker // PER-FORWARD worker model: every enabled, non-disabled link on THIS
// started here. Cluster-distributed forwards (non-localOnly) are claimed // node gets its own frpc process keyed by the forward triple.
// and spawned by the owning cluster node — starting them locally too // - localOnly forwards: always owned here (loopback stays local).
// would duplicate the worker. // - cluster forwards: only started by the ring claim (ClaimFn) —
remotes, err := st.ListRemotes() // SyncWorkers deliberately skips them so a node that merely holds
// a canvas copy does not spawn workers for forwards it doesn't own
// (the old "every node registers every proxy" bug).
links, err := st.ListLinks()
if err != nil { if err != nil {
return return
} }
for _, r := range remotes { for _, ln := range links {
if !r.Enabled { loc, ok := st.GetLocal(ln.Local)
if !ok || !loc.LocalOnly {
continue // cluster-owned or unknown local
}
rem, ok := st.GetRemote(ln.Remote)
if !ok || !rem.Enabled {
continue continue
} }
// worker starts here only if every link of this remote is localOnly key := process.WorkerKey(ln.Local, ln.Remote, ln.RemotePort)
fwd, err := st.LinksForRemote(r.Name) if ln.Disabled {
if err != nil { _ = pm.Stop(key) // per-forward stop kills exactly its process
continue continue
} }
allLocal := len(fwd) > 0 if _, has := pm.Status(key); has {
for _, f := range fwd { _ = pm.Restart(key)
loc, ok := st.GetLocal(f.Service)
if !ok || !loc.LocalOnly {
allLocal = false
break
}
}
if !allLocal {
// has at least one cluster-distributed forward; cluster owns it
continue
}
if _, has := pm.Status(r.Name); has {
_ = pm.Restart(r.Name)
} else { } else {
_ = pm.Start(r.Name) _ = pm.Start(key)
} }
} }
}, },
} }
// Auto-start enabled remotes on boot. // Auto-start localOnly forwards on boot. Cluster forwards are NOT started
// here — they are re-submitted by the leader's startup auto-submit below
// and claimed (spawning their own worker) wherever the ring places them.
s, _ := st.Settings() s, _ := st.Settings()
if s.AutoStartProfiles { if s.AutoStartProfiles {
remotes, _ := st.ListRemotes() links, _ := st.ListLinks()
for _, r := range remotes { remotesByName := map[string]store.Remote{}
if r.Enabled { if remotes, err := st.ListRemotes(); err == nil {
_ = pm.Start(r.Name) for _, r := range remotes {
remotesByName[r.Name] = r
} }
} }
localsByName := map[string]store.Local{}
if locals, err := st.ListLocals(); err == nil {
for _, l := range locals {
localsByName[l.Name] = l
}
}
for _, ln := range links {
if ln.Disabled {
continue
}
loc, ok := localsByName[ln.Local]
if !ok || !loc.LocalOnly {
continue
}
rem, ok := remotesByName[ln.Remote]
if !ok || !rem.Enabled {
continue
}
_ = pm.Start(process.WorkerKey(ln.Local, ln.Remote, ln.RemotePort))
}
} }
handler, err := httpapi.NewServeMux(h) handler, err := httpapi.NewServeMux(h)
@ -633,35 +641,44 @@ func splitCSV(s string) []string {
return out return out
} }
// renderRemote builds the process.Render closure from store data. // renderForward builds the process.Render closure for the PER-FORWARD worker
func renderRemote(st *store.Store) func(string) ([]byte, error) { // model: each (local, remote, remotePort) link gets its OWN frpc config
return func(remoteName string) ([]byte, error) { // containing exactly ONE proxy. The worker key is the forward triple
// (process.WorkerKey), so every node only ever registers its own proxies on
// frps — this eliminates the multi-node "proxy already exists" fight of the
// old per-remote model, where every node rendered ALL forwards of a remote.
// A disabled link renders an empty proxy list: frpc exits cleanly with no
// proxies (the supervisor sees a clean exit and stops restarting it), which
// is exactly the per-forward stop semantics.
func renderForward(st *store.Store) func(string) ([]byte, error) {
return func(key string) ([]byte, error) {
localName, remoteName, port, ok := process.ParseWorkerKey(key)
if !ok {
return nil, fmt.Errorf("invalid worker key %q", key)
}
rem, ok := st.GetRemote(remoteName) rem, ok := st.GetRemote(remoteName)
if !ok { if !ok {
return nil, store.ErrNotFound return nil, store.ErrNotFound
} }
forwards, err := st.LinksForRemote(remoteName) loc, ok := st.GetLocal(localName)
if err != nil { if !ok {
return nil, err return nil, store.ErrNotFound
} }
proxies := make([]render.Proxy, 0, len(forwards)) disabled := false
for _, f := range forwards { for _, ln := range mustListLinks(st) {
// A disabled forward is omitted from the generated frpc config so if ln.Local == localName && ln.Remote == remoteName && ln.RemotePort == port {
// a per-forward stop drops just this proxy on worker restart, disabled = ln.Disabled
// leaving sibling forwards on the same remote untouched. break
if f.Disabled {
continue
}
loc, ok := st.GetLocal(f.Service)
if !ok {
continue
} }
}
proxies := []render.Proxy(nil)
if !disabled {
proxies = append(proxies, render.Proxy{ proxies = append(proxies, render.Proxy{
Name: loc.Name, Name: loc.Name,
Type: loc.Protocol, Type: loc.Protocol,
LocalIP: loc.IP, LocalIP: loc.IP,
LocalPort: loc.Port, LocalPort: loc.Port,
RemotePort: f.RemotePort, RemotePort: port,
UseEncryption: loc.UseEncryption, UseEncryption: loc.UseEncryption,
UseCompression: loc.UseCompression, UseCompression: loc.UseCompression,
BandwidthLimit: loc.BandwidthLimit, BandwidthLimit: loc.BandwidthLimit,
@ -690,6 +707,12 @@ func renderRemote(st *store.Store) func(string) ([]byte, error) {
} }
} }
// mustListLinks is ListLinks with errors swallowed (best-effort reads).
func mustListLinks(st *store.Store) []store.Link {
links, _ := st.ListLinks()
return links
}
// multiFlag collects repeated string flags (-peer a -peer b ...). // multiFlag collects repeated string flags (-peer a -peer b ...).
type multiFlag []string type multiFlag []string

View File

@ -2,11 +2,13 @@ package httpapi
import ( import (
"encoding/json" "encoding/json"
"log"
"net/http" "net/http"
"os" "os"
"strings" "strings"
"webui4frpc/internal/cluster" "webui4frpc/internal/cluster"
"webui4frpc/internal/process"
"webui4frpc/internal/store" "webui4frpc/internal/store"
) )
@ -127,8 +129,11 @@ func (h *Handler) applyCanvas(w http.ResponseWriter, r *http.Request, canvas *ca
if !keep[old.Name] { if !keep[old.Name] {
if old.LocalOnly { if old.LocalOnly {
if h.Process != nil { if h.Process != nil {
// Per-forward model: stop every localOnly worker of this local.
if fwd, _ := s.LinksForLocal(old.Name); len(fwd) > 0 { if fwd, _ := s.LinksForLocal(old.Name); len(fwd) > 0 {
_ = h.Process.Stop(fwd[0].Remote) for _, f := range fwd {
_ = h.Process.Stop(process.WorkerKey(old.Name, f.Remote, f.RemotePort))
}
} }
} }
} else if h.Ring != nil { } else if h.Ring != nil {
@ -324,9 +329,13 @@ func (h *Handler) handleRemoteUpsert(w http.ResponseWriter, r *http.Request) {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)
return return
} }
// Start the worker if enabled. // Start the per-forward workers for this remote if enabled.
if rem.Enabled { if rem.Enabled {
_ = h.Process.Start(rem.Name) // In the per-forward model, merely upserting a remote does not start
// any workers — the actual links control which forwards run. The
// caller is expected to save the canvas (PUT /canvas) or start
// individual forwards via the forwards page.
log.Printf("remote %s upserted (enabled=%v); use status page forwards to start", rem.Name, rem.Enabled)
} }
writeJSON(w, http.StatusOK, rem) writeJSON(w, http.StatusOK, rem)
} }
@ -341,7 +350,7 @@ func (h *Handler) handleRemoteDelete(w http.ResponseWriter, r *http.Request) {
http.Error(w, "remote name required", http.StatusBadRequest) http.Error(w, "remote name required", http.StatusBadRequest)
return return
} }
_ = h.Process.Stop(name) _ = h.Process.StopRemote(name)
if err := h.Store.DeleteRemote(name); err != nil { if err := h.Store.DeleteRemote(name); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)
return return
@ -367,8 +376,10 @@ func (h *Handler) handleProfile(w http.ResponseWriter, r *http.Request) {
switch action { switch action {
case "": case "":
// GET profile status // GET profile status — aggregate per-forward workers into a single
st, has := h.Process.Status(name) // remote-centric view. "hasProcess" is true when ANY forward worker
// for this remote is running.
st, has := h.Process.RemoteStatus(name)
forwards, _ := h.Store.LinksForRemote(name) forwards, _ := h.Store.LinksForRemote(name)
writeJSON(w, http.StatusOK, map[string]any{ writeJSON(w, http.StatusOK, map[string]any{
"name": name, "process": st, "hasProcess": has, "forwards": forwards, "name": name, "process": st, "hasProcess": has, "forwards": forwards,
@ -384,11 +395,11 @@ func (h *Handler) handleProfile(w http.ResponseWriter, r *http.Request) {
} }
var err error var err error
if action == "start" { if action == "start" {
err = h.Process.Start(name) err = h.Process.StartRemote(name)
} else if action == "stop" { } else if action == "stop" {
err = h.Process.Stop(name) err = h.Process.StopRemote(name)
} else { } else {
err = h.Process.Restart(name) err = h.Process.RestartRemote(name)
} }
if err != nil { if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)
@ -399,7 +410,15 @@ func (h *Handler) handleProfile(w http.ResponseWriter, r *http.Request) {
methodNotAllowed(w) methodNotAllowed(w)
} }
case "config": case "config":
data, err := os.ReadFile(h.Process.ConfigPath(name)) // Per-forward model: no single config file for a remote. Return the
// first forward's config as a hint; the status page already shows
// per-forward configs.
workers := h.Process.RemoteWorkers(name)
if len(workers) == 0 {
http.NotFound(w, r)
return
}
data, err := os.ReadFile(h.Process.ConfigPath(workers[0]))
if err != nil { if err != nil {
http.NotFound(w, r) http.NotFound(w, r)
return return
@ -407,7 +426,13 @@ func (h *Handler) handleProfile(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
_, _ = w.Write(data) _, _ = w.Write(data)
case "logs": case "logs":
path := h.Process.LogPath(name) // Per-forward model: return the first per-forward log for this remote.
workers := h.Process.RemoteWorkers(name)
if len(workers) == 0 {
http.Error(w, "no workers for this remote", http.StatusNotFound)
return
}
path := h.Process.LogPath(workers[0])
data, err := tailFile(path, 64*1024) data, err := tailFile(path, 64*1024)
if err != nil { if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)

View File

@ -9,6 +9,7 @@ import (
"time" "time"
"webui4frpc/internal/cluster" "webui4frpc/internal/cluster"
"webui4frpc/internal/process"
"webui4frpc/internal/store" "webui4frpc/internal/store"
) )
@ -76,8 +77,11 @@ func (h *Handler) handleLocalByName(w http.ResponseWriter, r *http.Request) {
case http.MethodDelete: case http.MethodDelete:
if l.LocalOnly { if l.LocalOnly {
if h.Process != nil { if h.Process != nil {
// Per-forward model: stop every localOnly worker of this local.
if fwd, _ := h.Store.LinksForLocal(name); len(fwd) > 0 { if fwd, _ := h.Store.LinksForLocal(name); len(fwd) > 0 {
_ = h.Process.Stop(fwd[0].Remote) for _, f := range fwd {
_ = h.Process.Stop(process.WorkerKey(name, f.Remote, f.RemotePort))
}
} }
} }
} else if h.Ring != nil { } else if h.Ring != nil {

View File

@ -4,6 +4,7 @@ import (
"encoding/json" "encoding/json"
"net/http" "net/http"
"webui4frpc/internal/process"
"webui4frpc/internal/store" "webui4frpc/internal/store"
) )
@ -74,10 +75,11 @@ func (h *Handler) startForward(local, remote string, port int) error {
ln.Disabled = false ln.Disabled = false
if loc.LocalOnly { if loc.LocalOnly {
if h.Process != nil { if h.Process != nil {
if _, has := h.Process.Status(remote); has { key := process.WorkerKey(local, remote, port)
_ = h.Process.Restart(remote) if _, has := h.Process.Status(key); has {
_ = h.Process.Restart(key)
} else { } else {
_ = h.Process.Start(remote) _ = h.Process.Start(key)
} }
} }
return nil return nil
@ -109,8 +111,9 @@ func (h *Handler) stopForward(local, remote string, port int) error {
} }
if loc.LocalOnly { if loc.LocalOnly {
if h.Process != nil { if h.Process != nil {
if _, has := h.Process.Status(remote); has { key := process.WorkerKey(local, remote, port)
_ = h.Process.Restart(remote) if _, has := h.Process.Status(key); has {
_ = h.Process.Restart(key)
} }
} }
return nil return nil

View File

@ -6,6 +6,8 @@ import (
"net/http" "net/http"
"sync" "sync"
"time" "time"
"webui4frpc/internal/process"
) )
// workerLog is one frpc worker's log tail on a node. // workerLog is one frpc worker's log tail on a node.
@ -34,8 +36,13 @@ func (h *Handler) localWorkerLogs() nodeLogsResp {
for _, name := range h.Process.ListWorkers() { for _, name := range h.Process.ListWorkers() {
st, _ := h.Process.Status(name) st, _ := h.Process.Status(name)
lines, _ := tailFile(h.Process.LogPath(name), 64*1024) lines, _ := tailFile(h.Process.LogPath(name), 64*1024)
// Parse the worker key to get the real remote name for display.
remoteName := name
if _, r, _, ok := process.ParseWorkerKey(name); ok {
remoteName = r
}
out.Workers = append(out.Workers, workerLog{ out.Workers = append(out.Workers, workerLog{
Name: name, Remote: name, State: st.State, Lines: lines, Name: name, Remote: remoteName, State: st.State, Lines: lines,
}) })
} }
return out return out

View File

@ -257,7 +257,7 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) {
binary = h.BinaryPath() binary = h.BinaryPath()
} }
for _, rv := range remotes { for _, rv := range remotes {
st, has := h.Process.Status(rv.Name) st, has := h.Process.RemoteStatus(rv.Name)
fwd, _ := h.Store.LinksForRemote(rv.Name) fwd, _ := h.Store.LinksForRemote(rv.Name)
p := profileStatus{ p := profileStatus{
Name: rv.Name, Enabled: rv.Enabled, Status: st, HasProc: has, Forwards: fwd, Name: rv.Name, Enabled: rv.Enabled, Status: st, HasProc: has, Forwards: fwd,
@ -271,10 +271,12 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) {
} }
} }
// Prefer admin-derived conn state; fall back to log inference. // Prefer admin-derived conn state; fall back to log inference.
// In the per-forward model multiple workers may exist for one remote;
// use the first worker's log when available.
if p.AdminEnabled && len(p.ProxyStates) > 0 { if p.AdminEnabled && len(p.ProxyStates) > 0 {
p.ConnState = connStateAdmin(p.ProxyStates) p.ConnState = connStateAdmin(p.ProxyStates)
} else { } else {
p.ConnState = connStateOf(h.Process.LogPath(rv.Name), rv.Enabled, st) p.ConnState = connStateOf(firstWorkerLog(h.Process, rv.Name), rv.Enabled, st)
} }
profiles = append(profiles, p) profiles = append(profiles, p)
} }
@ -300,11 +302,17 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) {
} }
ts := make([]localTargetStatus, 0, len(targets)) ts := make([]localTargetStatus, 0, len(targets))
for _, tg := range targets { for _, tg := range targets {
st, _ := h.Process.Status(tg.Remote) // Per-forward model: this target's worker is the specific
// (local,remote,port) forward; active = its own process running.
st, has := h.Process.Status(process.WorkerKey(l.Name, tg.Remote, tg.RemotePort))
state := "stopped"
if has {
state = st.State
}
ts = append(ts, localTargetStatus{ ts = append(ts, localTargetStatus{
Remote: tg.Remote, Remote: tg.Remote,
RemotePort: tg.RemotePort, RemotePort: tg.RemotePort,
WorkerState: st.State, WorkerState: state,
}) })
} }
localStatuses = append(localStatuses, localStatus{Local: l, Targets: ts}) localStatuses = append(localStatuses, localStatus{Local: l, Targets: ts})
@ -346,7 +354,8 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) {
if loc.LocalOnly { if loc.LocalOnly {
fs.Kind = "local" fs.Kind = "local"
fs.OwnerID = selfID fs.OwnerID = selfID
if st, has := h.Process.Status(ln.Remote); has { // Per-forward model: check this exact forward's worker process.
if st, has := h.Process.Status(process.WorkerKey(ln.Local, ln.Remote, ln.RemotePort)); has {
fs.Active = !ln.Disabled && st.State == "running" fs.Active = !ln.Disabled && st.State == "running"
} }
} else { } else {
@ -727,6 +736,16 @@ func frpcVersionOf(binPath string) string {
return "" return ""
} }
// firstWorkerLog returns the log path of the first per-forward worker of a
// remote (worker keys are "local~remote~port"), or "" when none exist. Used
// by status aggregation to infer frps connectivity from worker logs.
func firstWorkerLog(pm *process.Manager, remote string) string {
for _, key := range pm.RemoteWorkers(remote) {
return pm.LogPath(key)
}
return ""
}
// connStateOf derives the frps connection state from the local worker process // connStateOf derives the frps connection state from the local worker process
// state plus a probe of its log. We never control the remote frps: the worker is // state plus a probe of its log. We never control the remote frps: the worker is
// our local frpc; "connected" means frpc has successfully logged in to frps. // our local frpc; "connected" means frpc has successfully logged in to frps.

View File

@ -287,6 +287,88 @@ func (m *Manager) ListWorkers() []string {
return out return out
} }
// RemoteWorkers returns the worker keys of all per-forward workers whose key
// parses to the given remote name. In the per-forward model every link has
// its own process keyed "local~remote~port"; this aggregates them so remote-
// centric callers (status page, profile endpoints) can report an overall
// view without knowing the key format.
func (m *Manager) RemoteWorkers(remote string) []string {
var out []string
for _, key := range m.ListWorkers() {
if _, r, _, ok := ParseWorkerKey(key); ok && r == remote {
out = append(out, key)
}
}
return out
}
// RemoteStatus summarizes a remote's per-forward workers into one Status:
// running when ANY forward worker runs; crashed/starting/restarting take
// priority in that order; stopped (has=false) when no worker exists.
// This keeps the legacy per-remote status shape working on top of the
// per-forward worker model.
func (m *Manager) RemoteStatus(remote string) (Status, bool) {
keys := m.RemoteWorkers(remote)
if len(keys) == 0 {
return Status{State: "stopped"}, false
}
summary := Status{State: "stopped"}
worst := -1 // rank: running=0 < others=1 < crashed=2
for _, k := range keys {
st, _ := m.Status(k)
rank := 1
switch st.State {
case "running":
rank = 0
case "crashed":
rank = 2
}
if rank > worst || (rank == worst && summary.State == "stopped") {
if rank != worst || st.State != "stopped" {
summary = st
worst = rank
}
}
}
return summary, true
}
// StartRemote starts every per-forward worker of one remote. Missing workers
// are skipped; the first error wins.
func (m *Manager) StartRemote(remote string) error {
var firstErr error
for _, key := range m.RemoteWorkers(remote) {
if err := m.Start(key); err != nil && firstErr == nil {
firstErr = err
}
}
return firstErr
}
// RestartRemote restarts every per-forward worker belonging to one remote
// (used where the old model had a single per-remote worker). Missing workers
// are skipped; the first error wins.
func (m *Manager) RestartRemote(remote string) error {
var firstErr error
for _, key := range m.RemoteWorkers(remote) {
if err := m.Restart(key); err != nil && firstErr == nil {
firstErr = err
}
}
return firstErr
}
// StopRemote stops every per-forward worker of one remote.
func (m *Manager) StopRemote(remote string) error {
var firstErr error
for _, key := range m.RemoteWorkers(remote) {
if err := m.Stop(key); err != nil && firstErr == nil {
firstErr = err
}
}
return firstErr
}
// StopAll stops all workers (used on manager shutdown). // StopAll stops all workers (used on manager shutdown).
func (m *Manager) StopAll() { func (m *Manager) StopAll() {
m.mu.Lock() m.mu.Lock()

View File

@ -0,0 +1,39 @@
package process
import (
"fmt"
"strconv"
"strings"
)
// WorkerKey is the unique identity of ONE forward's worker process:
//
// "<local>~<remote>~<port>"
//
// The per-forward worker model gives every link its own frpc process, config
// file and log file. This fixes the multi-node proxy-name fight of the old
// per-remote model (every node rendered the SAME proxy list and raced to
// register them on frps -> "proxy already exists"), and isolates faults: one
// crashing forward can no longer take its siblings down.
//
// '~' is a legal filename character on every supported OS (including Windows,
// unlike ':') and is rare in user-chosen node names. ParseWorkerKey rejects
// malformed keys so callers fail loudly instead of silently mis-rendering.
func WorkerKey(local, remote string, port int) string {
return fmt.Sprintf("%s~%s~%d", local, remote, port)
}
// ParseWorkerKey splits a worker key back into its (local, remote, port)
// forward triple. ok=false for anything that is not a well-formed key
// (e.g. legacy per-remote worker names).
func ParseWorkerKey(key string) (local, remote string, port int, ok bool) {
parts := strings.Split(key, "~")
if len(parts) != 3 {
return "", "", 0, false
}
p, err := strconv.Atoi(parts[2])
if err != nil || parts[0] == "" || parts[1] == "" || p <= 0 {
return "", "", 0, false
}
return parts[0], parts[1], p, true
}

View File

@ -0,0 +1,37 @@
package process
import "testing"
func TestWorkerKeyRoundTrip(t *testing.T) {
cases := []struct {
local, remote string
port int
}{
{"web", "frps1", 18080},
{"db", "aliyun-frps", 5432},
{"svc-1", "node.a.example", 80},
}
for _, c := range cases {
key := WorkerKey(c.local, c.remote, c.port)
l, r, p, ok := ParseWorkerKey(key)
if !ok || l != c.local || r != c.remote || p != c.port {
t.Fatalf("roundtrip %q -> (%q,%q,%d,%v)", key, l, r, p, ok)
}
}
}
func TestParseWorkerKeyRejectsBad(t *testing.T) {
bad := []string{
"", "srv-a", // legacy per-remote names / empty
"a~b", // missing port
"a~b~x", // non-numeric port
"a~b~0", // port out of range
"~b~1", // empty local
"a~~1", // empty remote
}
for _, key := range bad {
if _, _, _, ok := ParseWorkerKey(key); ok {
t.Fatalf("ParseWorkerKey(%q) accepted, want reject", key)
}
}
}