mirror of
https://gitcode.com/JianFeeeee/webui4frpc.git
synced 2026-09-19 16:38:31 +00:00
fix: track cmd/webui4frpc/main.go — .gitignore 'webui4frpc' was matching the cmd/ directory, not just the binary
Also includes the startup auto-submit goroutine: on full-cluster restart, the leader re-submits local non-disabled, non-localOnly forwards from SQLite so the ring topology repopulates. SubmitTask is idempotent via HasTask, so this is safe when tasks survived (single-node restart).
This commit is contained in:
4
.gitignore
vendored
4
.gitignore
vendored
@ -7,8 +7,8 @@ web/dist/
|
||||
# Go 二进制
|
||||
*.exe
|
||||
*.test
|
||||
webui-frpc
|
||||
webui4frpc
|
||||
/webui-frpc
|
||||
/webui4frpc
|
||||
|
||||
# 运行时数据
|
||||
*.db
|
||||
|
||||
671
cmd/webui4frpc/main.go
Normal file
671
cmd/webui4frpc/main.go
Normal file
@ -0,0 +1,671 @@
|
||||
// webui-frpc is a standalone web controller for frpc: configure local forward
|
||||
// items and remote servers on a visual canvas, and it spawns one frpc worker
|
||||
// per remote server. No frp source code is bundled; frpc binaries are
|
||||
// downloaded on demand or pointed at manually.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"webui4frpc/internal/cluster"
|
||||
"webui4frpc/internal/httpapi"
|
||||
"webui4frpc/internal/install"
|
||||
"webui4frpc/internal/process"
|
||||
"webui4frpc/internal/render"
|
||||
"webui4frpc/internal/store"
|
||||
)
|
||||
|
||||
func main() {
|
||||
addr := flag.String("addr", "127.0.0.1:7500", "listen address")
|
||||
user := flag.String("user", "admin", "basic auth user")
|
||||
pass := flag.String("password", "admin", "basic auth password")
|
||||
workdir := flag.String("workdir", "./webui-frpc", "data directory (db, configs, logs, bin)")
|
||||
bin := flag.String("bin", "", "initial frpc binary path (optional; defaults to manual/auto-install)")
|
||||
frpc := flag.String("frpc", "", "default worker frpc binary path (fallback when settings has none)")
|
||||
var peers multiFlag
|
||||
flag.Var(&peers, "peer", "cluster peer host:port (repeatable; same creds)")
|
||||
joinKey := flag.String("join-key", os.Getenv("W4F_JOIN_KEY"), "cluster join key = the sponsor node's nodeKey (required with -peer to join an existing cluster)")
|
||||
flag.Parse()
|
||||
|
||||
wd, err := filepath.Abs(*workdir)
|
||||
if err != nil {
|
||||
log.Fatalf("resolve workdir: %v", err)
|
||||
}
|
||||
for _, d := range []string{wd, filepath.Join(wd, "configs"), filepath.Join(wd, "logs"), filepath.Join(wd, "bin")} {
|
||||
if err := os.MkdirAll(d, 0o755); err != nil {
|
||||
log.Fatalf("create dir %s: %v", d, err)
|
||||
}
|
||||
}
|
||||
|
||||
st, err := store.New(filepath.Join(wd, "manager.db"))
|
||||
if err != nil {
|
||||
log.Fatalf("open store: %v", err)
|
||||
}
|
||||
defer st.Close()
|
||||
|
||||
// Sync the flag credentials into the users table as a system=1 admin account
|
||||
// on every startup, so -user/-password changes propagate without manual
|
||||
// account setup. The auth middleware's Basic branch still falls back to the
|
||||
// raw flag creds, so inter-node cluster traffic (which uses SetBasicAuth
|
||||
// with these flags) never breaks even before this row exists.
|
||||
if err := st.SyncSystemUser(*user, *pass); err != nil {
|
||||
log.Fatalf("sync system user: %v", err)
|
||||
}
|
||||
|
||||
// Seed settings with the initial binary path if provided.
|
||||
if *bin != "" {
|
||||
cfg, _ := st.Settings()
|
||||
cfg.BinaryPath = *bin
|
||||
_ = st.UpdateSettings(cfg)
|
||||
}
|
||||
|
||||
// Determine the fallback binary: explicit -bin, else this executable (if it
|
||||
// looks like a usable frpc) — but we never want to accidentally treat our
|
||||
// own webui binary as frpc. So fallback is empty unless -bin is set; the UI
|
||||
// then asks the user to install or provide frpc.
|
||||
fallbackBin := ""
|
||||
if *frpc != "" {
|
||||
fallbackBin = *frpc
|
||||
} else if *bin != "" {
|
||||
fallbackBin = *bin
|
||||
}
|
||||
|
||||
pm := process.NewManager(process.Options{
|
||||
ConfigsDir: filepath.Join(wd, "configs"),
|
||||
LogsDir: filepath.Join(wd, "logs"),
|
||||
BinaryPath: func() string {
|
||||
if s, _ := st.Settings(); s.BinaryPath != "" {
|
||||
return s.BinaryPath
|
||||
}
|
||||
return fallbackBin
|
||||
},
|
||||
Render: renderRemote(st),
|
||||
AutoRestart: func(name string) bool {
|
||||
s, _ := st.Settings()
|
||||
if !s.RestartOnExit {
|
||||
return false
|
||||
}
|
||||
rem, ok := st.GetRemote(name)
|
||||
return ok && rem.Enabled
|
||||
},
|
||||
RestartInterval: func() int {
|
||||
s, _ := st.Settings()
|
||||
if s.RestartIntervalSeconds > 0 {
|
||||
return s.RestartIntervalSeconds
|
||||
}
|
||||
return 5
|
||||
},
|
||||
})
|
||||
|
||||
reg := cluster.NewRegistry(filepath.Join(wd, "bin"))
|
||||
// M6: register cluster peers from -peer flags (same creds as this node).
|
||||
for _, p := range peers {
|
||||
reg.RegisterPeer(p, p, *user, *pass)
|
||||
}
|
||||
|
||||
// M6 token ring: this node is the initial leader of the cluster ring
|
||||
// (creation node). Engine drives two-phase cycles; transport POSTs the
|
||||
// token to the successor over HTTP.
|
||||
selfID := reachableAddr(*addr)
|
||||
ringTransport := &cluster.TokenTransport{User: *user, Pass: *pass}
|
||||
// nodeKey is this node's cluster admission key. It is NOT generated at
|
||||
// startup — a fresh node has no key until it creates or joins a cluster.
|
||||
// A creation node (no -peer) generates its key on CreateCluster; a joiner
|
||||
// (has -peer) generates its key on AdoptState after a successful join.
|
||||
// If the store has a persisted key (from a previous cluster membership),
|
||||
// restore it so a restart-in-cluster node keeps its identity.
|
||||
nodeKey := ""
|
||||
if s, _ := st.Settings(); s.NodeKey != "" {
|
||||
nodeKey = s.NodeKey
|
||||
}
|
||||
var ring *cluster.Engine
|
||||
ring = cluster.NewEngine(
|
||||
selfID, selfID, *user, *pass, "0.1.0", nil,
|
||||
&cluster.AppHandler{
|
||||
LoadFn: func() (float64, float64) {
|
||||
return cluster.SampleMemLoad(), 10
|
||||
},
|
||||
// Claim persists the forward locally (intermediate config -> store),
|
||||
// then syncs workers so the frpc worker for that remote starts.
|
||||
ClaimFn: func(ctx context.Context, tk *cluster.Task) error {
|
||||
if err := st.UpsertLocal(tk.Local); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := st.UpsertRemote(tk.Remote); err != nil {
|
||||
return err
|
||||
}
|
||||
links, err := st.ListLinks()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Upsert the task's link (dedup by local+remote+remotePort).
|
||||
// The submitter persists the canvas — including this link — via
|
||||
// saveCanvas, and when this same node is also the claimer (lowest
|
||||
// load) blindly re-appending would duplicate the proxy in the
|
||||
// rendered frpc config (two proxies, same remotePort) and the
|
||||
// worker would fail to register. Skip only if already present.
|
||||
dup := false
|
||||
for _, l := range links {
|
||||
if l.Local == tk.Link.Local && l.Remote == tk.Link.Remote && l.RemotePort == tk.Link.RemotePort {
|
||||
dup = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !dup {
|
||||
links = append(links, tk.Link)
|
||||
}
|
||||
if err := st.ReplaceLinks(links); err != nil {
|
||||
return err
|
||||
}
|
||||
// A re-claim after a forward-centric stop leaves the link flagged
|
||||
// disabled (by RevokeFn); clear it so renderRemote renders the
|
||||
// proxy back in. No-op for a fresh claim.
|
||||
_ = st.SetLinkDisabled(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort, false)
|
||||
// Start (or restart) the worker for the remote we just claimed.
|
||||
// Restart re-renders the frpc config with all of this remote's
|
||||
// links (including the just-upserted one) and respawns the process.
|
||||
// We deliberately do NOT iterate every stored remote here: this
|
||||
// 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 _, has := pm.Status(tk.Remote.Name); has {
|
||||
_ = pm.Restart(tk.Remote.Name)
|
||||
} else {
|
||||
_ = pm.Start(tk.Remote.Name)
|
||||
}
|
||||
}
|
||||
log.Printf("ring[%s] claimed task %s: %s→%s:%d", selfID, tk.ID, tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort)
|
||||
return nil
|
||||
},
|
||||
// Revoke cancels the forward NON-destructively: flag the exact link
|
||||
// disabled (so renderRemote drops just its proxy and a later forward-
|
||||
// centric start can re-enable it) and right-size the worker — stop it
|
||||
// when the remote has no remaining enabled forwards, else restart so
|
||||
// sibling proxies survive. The old delete-local/remote/drop-all-
|
||||
// matching-links form served the topology-derived-canvas model but
|
||||
// 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 {
|
||||
_ = st.SetLinkDisabled(tk.Local.Name, tk.Remote.Name, tk.Link.RemotePort, true)
|
||||
fwd, _ := st.LinksForRemote(tk.Remote.Name)
|
||||
enabled := 0
|
||||
for _, f := range fwd {
|
||||
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)
|
||||
return nil
|
||||
},
|
||||
},
|
||||
ringTransport.SendTo(func(nodeID string) string {
|
||||
if ring == nil {
|
||||
return nodeID
|
||||
}
|
||||
for _, n := range ring.State().Nodes {
|
||||
if n.ID == nodeID {
|
||||
return n.Addr
|
||||
}
|
||||
}
|
||||
return nodeID
|
||||
}),
|
||||
selfID, len(peers) == 0, nodeKey, // creation node (no -peer) is the initial leader
|
||||
)
|
||||
// Install the persistence callbacks so ensureNodeKey (called from
|
||||
// CreateCluster / AdoptState) and detachAsStandalone can persist/clear
|
||||
// the key in the settings store without the engine depending on store.
|
||||
ring.SetKeyPersist(func(key string) error { return st.SetNodeKey(key) })
|
||||
// peerPersist saves the cached peer list (addr+key) every token cycle so
|
||||
// a crashed node can auto-rejoin via any cached peer on restart. Cleared
|
||||
// on explicit detach (detachAsStandalone) so a node that intentionally
|
||||
// left does NOT auto-rejoin.
|
||||
ring.SetPeerPersist(func(peersJSON string) error { return st.SetClusterPeers(peersJSON) })
|
||||
|
||||
// Re-apply local store group labels onto the ring topology after each
|
||||
// state adoption. Without this, group changes made via HTTP handlers
|
||||
// (POST /forwards/assign) are overwritten by the next e.state = tk.State
|
||||
// and never propagate to other nodes.
|
||||
ring.SetTopologySync(func() {
|
||||
links, err := st.ListLinks()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, ln := range links {
|
||||
ring.UpdateTopologyGroup(ln.Local, ln.Remote, ln.RemotePort, ln.Group)
|
||||
}
|
||||
})
|
||||
|
||||
h := &httpapi.Handler{
|
||||
Store: st,
|
||||
Process: pm,
|
||||
WorkDir: wd,
|
||||
BinDir: filepath.Join(wd, "bin"),
|
||||
Cluster: reg,
|
||||
Ring: ring,
|
||||
// SelfAddr must be reachable by peers. 0.0.0.0/empty on a real host
|
||||
// would break peer reconnect, so fall back to the hostname (container
|
||||
// name inside docker is a routable addr on the compose network).
|
||||
SelfAddr: reachableAddr(*addr),
|
||||
User: *user,
|
||||
Password: *pass,
|
||||
BinaryPath: func() string {
|
||||
s, _ := st.Settings()
|
||||
if s.BinaryPath != "" {
|
||||
return s.BinaryPath
|
||||
}
|
||||
return fallbackBin
|
||||
},
|
||||
InstallBinary: func(version string) (string, string, error) {
|
||||
// M6: prefer cluster peer exchange (local cache -> peers), fall
|
||||
// back to the official GitHub URL only when no peer has it.
|
||||
ictx, icancel := context.WithTimeout(context.Background(), 90*time.Second)
|
||||
defer icancel()
|
||||
if p, err := reg.EnsureVersion(ictx, version); err == nil {
|
||||
s, _ := st.Settings()
|
||||
s.BinaryPath = p
|
||||
if err := st.UpdateSettings(s); err != nil {
|
||||
return p, version, err
|
||||
}
|
||||
return p, version, nil
|
||||
}
|
||||
path, ver, err := install.Install(context.Background(), filepath.Join(wd, "bin"), version)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
s, _ := st.Settings()
|
||||
s.BinaryPath = path
|
||||
if err := st.UpdateSettings(s); err != nil {
|
||||
return path, ver, err
|
||||
}
|
||||
return path, ver, nil
|
||||
},
|
||||
SyncWorkers: func() {
|
||||
// Only forwards that stay on this node (localOnly) get their worker
|
||||
// started here. Cluster-distributed forwards (non-localOnly) are claimed
|
||||
// and spawned by the owning cluster node — starting them locally too
|
||||
// would duplicate the worker.
|
||||
remotes, err := st.ListRemotes()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, r := range remotes {
|
||||
if !r.Enabled {
|
||||
continue
|
||||
}
|
||||
// worker starts here only if every link of this remote is localOnly
|
||||
fwd, err := st.LinksForRemote(r.Name)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
allLocal := len(fwd) > 0
|
||||
for _, f := range fwd {
|
||||
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 {
|
||||
_ = pm.Start(r.Name)
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
// Auto-start enabled remotes on boot.
|
||||
s, _ := st.Settings()
|
||||
if s.AutoStartProfiles {
|
||||
remotes, _ := st.ListRemotes()
|
||||
for _, r := range remotes {
|
||||
if r.Enabled {
|
||||
_ = pm.Start(r.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
handler, err := httpapi.NewServeMux(h)
|
||||
if err != nil {
|
||||
log.Fatalf("build handler: %v", err)
|
||||
}
|
||||
srv := &http.Server{Addr: *addr, Handler: handler}
|
||||
|
||||
go func() {
|
||||
log.Printf("webui-frpc listening on %s (workdir %s)", *addr, wd)
|
||||
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||
log.Fatalf("server error: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// M6: cluster heartbeat + new-node bootstrap auto-propagation. The
|
||||
// bootstrap loop retries every 10s for the first minutes so binaries are
|
||||
// pulled as soon as neighbor info appears (a fresh node has none at boot).
|
||||
clusterCtx, clusterCancel := context.WithCancel(context.Background())
|
||||
defer clusterCancel()
|
||||
go reg.Discover(clusterCtx, 30*time.Second)
|
||||
// M6 token ring startup. Cached peers (persisted on every token cycle)
|
||||
// take priority over the -peer flag — -peer is only for FIRST-TIME
|
||||
// bootstrap; on crash/restart the node always rejoin via cached peers.
|
||||
//
|
||||
// 1. Persisted ClusterPeers → crash/restart recovery: try to rejoin
|
||||
// via any cached peer (addr+key from the last token cycle before
|
||||
// the crash). First node up finds all peers down → falls back to
|
||||
// CreateCluster; others rejoin it via cached peers.
|
||||
// 2. No cached peers + -peer flag → first-time bootstrap: join the
|
||||
// specified peer with the given join key. Retry every 10s for 5 min.
|
||||
// 3. No cached peers, no -peer → first-time creation node: seed a
|
||||
// fresh standalone cluster (CreateCluster).
|
||||
//
|
||||
// Explicit detach (detachAsStandalone) clears ClusterPeers, so a node
|
||||
// that intentionally left falls into case 2 or 3 and does NOT auto-rejoin.
|
||||
persistedPeers := ""
|
||||
if s, _ := st.Settings(); s.ClusterPeers != "" {
|
||||
persistedPeers = s.ClusterPeers
|
||||
}
|
||||
if persistedPeers != "" {
|
||||
go retryRejoinCached(clusterCtx, ring, persistedPeers)
|
||||
} else if len(peers) > 0 {
|
||||
go retryJoinCluster(clusterCtx, ring, peers[0], *joinKey)
|
||||
} else {
|
||||
if err := ring.CreateCluster(); err != nil {
|
||||
log.Printf("ring[%s] create cluster at startup: %v", selfID, err)
|
||||
}
|
||||
go ring.StartRing(clusterCtx)
|
||||
}
|
||||
go ring.WatchTokenLoss(clusterCtx)
|
||||
go ring.WatchLeader(clusterCtx)
|
||||
|
||||
// Startup auto-submit: the leader re-submits local non-disabled,
|
||||
// non-localOnly forwards from its SQLite store so the ring repopulates
|
||||
// after a full-cluster restart (ring topology is in-memory only and is
|
||||
// lost when all nodes crash simultaneously). On single-node restart the
|
||||
// topology is recovered via AdoptState, and SubmitTask is idempotent
|
||||
// via HasTask, so re-submitting is a harmless no-op in that case.
|
||||
// Only the leader does this — the canvas is typically saved on the
|
||||
// leader, so its local store has the complete set of forwards.
|
||||
go func() {
|
||||
for {
|
||||
if ring.IsLeader() {
|
||||
break
|
||||
}
|
||||
select {
|
||||
case <-time.After(2 * time.Second):
|
||||
case <-clusterCtx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
// Let the ring stabilize after leader election / rejoin.
|
||||
time.Sleep(3 * time.Second)
|
||||
links, err := st.ListLinks()
|
||||
if err != nil {
|
||||
log.Printf("ring[%s] auto-submit: %v", ring.ID, err)
|
||||
return
|
||||
}
|
||||
count := 0
|
||||
for _, ln := range links {
|
||||
if ln.Disabled {
|
||||
continue
|
||||
}
|
||||
loc, ok := st.GetLocal(ln.Local)
|
||||
if !ok || loc.LocalOnly {
|
||||
continue
|
||||
}
|
||||
rem, ok := st.GetRemote(ln.Remote)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
ring.SubmitTask(loc, rem, ln)
|
||||
count++
|
||||
}
|
||||
if count > 0 {
|
||||
log.Printf("ring[%s] leader auto-submitted %d forwards on startup", ring.ID, count)
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
ticker := time.NewTicker(10 * time.Second)
|
||||
defer ticker.Stop()
|
||||
bootTimeout := time.NewTimer(90 * time.Second)
|
||||
defer bootTimeout.Stop()
|
||||
for {
|
||||
bootCtx, bootCancel := context.WithTimeout(clusterCtx, 20*time.Second)
|
||||
err := reg.SyncFromPeers(bootCtx)
|
||||
bootCancel()
|
||||
log.Printf("cluster bootstrap sync iteration (peers=%d caches=%d err=%v)", len(reg.PeerList()), len(reg.CachedVersions()), err)
|
||||
if err != nil {
|
||||
log.Printf("cluster bootstrap sync: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-ticker.C:
|
||||
case <-bootTimeout.C:
|
||||
return
|
||||
case <-clusterCtx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
sig := make(chan os.Signal, 1)
|
||||
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
|
||||
<-sig
|
||||
log.Println("shutting down...")
|
||||
clusterCancel()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||||
defer cancel()
|
||||
pm.StopAll()
|
||||
_ = srv.Shutdown(ctx)
|
||||
}
|
||||
|
||||
// retryJoinCluster joins the ring at targetAddr using joinKey, retrying every
|
||||
// 10s for up to 5 min in case the peer isn't up yet. On success AdoptState is
|
||||
// called by JoinRing (which persists peers). On exhaustion, falls back to
|
||||
// CreateCluster so the node is at least functional as standalone.
|
||||
func retryJoinCluster(ctx context.Context, ring *cluster.Engine, targetAddr, joinKey string) {
|
||||
selfNode := ring.State().Nodes[0]
|
||||
ji := cluster.JoinInfo{ID: selfNode.ID, Addr: selfNode.Addr, Version: "0.1.0", JoinKey: joinKey}
|
||||
deadline := time.NewTimer(5 * time.Minute)
|
||||
defer deadline.Stop()
|
||||
ticker := time.NewTicker(10 * time.Second)
|
||||
defer ticker.Stop()
|
||||
// First attempt immediately, then on each tick.
|
||||
for {
|
||||
// If another node already joined us (we're now multi-node), stop
|
||||
// trying to join — we're already in a ring.
|
||||
if ring.IsMember() {
|
||||
log.Printf("ring[%s] already a cluster member, skipping join", selfNode.ID)
|
||||
return
|
||||
}
|
||||
jc, cancel := context.WithTimeout(ctx, 8*time.Second)
|
||||
err := ring.JoinRing(jc, targetAddr, ji)
|
||||
cancel()
|
||||
if err == nil {
|
||||
log.Printf("ring[%s] joined cluster via %s", selfNode.ID, targetAddr)
|
||||
return
|
||||
}
|
||||
log.Printf("ring[%s] join %s failed: %v; retrying", selfNode.ID, targetAddr, err)
|
||||
select {
|
||||
case <-ticker.C:
|
||||
case <-deadline.C:
|
||||
log.Printf("ring[%s] join retry exhausted, falling back to standalone", selfNode.ID)
|
||||
_ = ring.CreateCluster()
|
||||
go ring.StartRing(ctx)
|
||||
return
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// retryRejoinCached is the crash/restart recovery path: parse the persisted
|
||||
// peer list (JSON of [{addr,key},...]) and try to rejoin via each peer in
|
||||
// order. The first peer that accepts wins. Retry the whole list every 10s
|
||||
// for up to 5 min. On exhaustion, fall back to CreateCluster (standalone).
|
||||
func retryRejoinCached(ctx context.Context, ring *cluster.Engine, peersJSON string) {
|
||||
type peerEntry struct {
|
||||
Addr string `json:"addr"`
|
||||
Key string `json:"key"`
|
||||
}
|
||||
var peers []peerEntry
|
||||
if err := json.Unmarshal([]byte(peersJSON), &peers); err != nil || len(peers) == 0 {
|
||||
log.Printf("ring[%s] cached peers parse error, falling back to standalone", ring.ID)
|
||||
_ = ring.CreateCluster()
|
||||
go ring.StartRing(ctx)
|
||||
return
|
||||
}
|
||||
selfNode := ring.State().Nodes[0]
|
||||
deadline := time.NewTimer(5 * time.Minute)
|
||||
defer deadline.Stop()
|
||||
for {
|
||||
// If another node already joined us, stop trying to rejoin.
|
||||
if ring.IsMember() {
|
||||
log.Printf("ring[%s] already a cluster member, skipping rejoin", ring.ID)
|
||||
return
|
||||
}
|
||||
for _, p := range peers {
|
||||
ji := cluster.JoinInfo{ID: selfNode.ID, Addr: selfNode.Addr, Version: "0.1.0", JoinKey: p.Key}
|
||||
jc, cancel := context.WithTimeout(ctx, 8*time.Second)
|
||||
err := ring.JoinRing(jc, p.Addr, ji)
|
||||
cancel()
|
||||
if err == nil {
|
||||
log.Printf("ring[%s] rejoined cluster via cached peer %s", ring.ID, p.Addr)
|
||||
return
|
||||
}
|
||||
log.Printf("ring[%s] rejoin %s failed: %v", ring.ID, p.Addr, err)
|
||||
}
|
||||
select {
|
||||
case <-time.After(10 * time.Second):
|
||||
case <-deadline.C:
|
||||
log.Printf("ring[%s] rejoin retry exhausted, falling back to standalone", ring.ID)
|
||||
_ = ring.CreateCluster()
|
||||
go ring.StartRing(ctx)
|
||||
return
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// reachableAddr returns an address peers can dial back: an explicit
|
||||
// W4F_HOST (compose DNS name) wins; otherwise a non-wildcard listen addr; a
|
||||
// wildcard falls back to the hostname (routable inside docker networks).
|
||||
func reachableAddr(addr string) string {
|
||||
if h := os.Getenv("W4F_HOST"); h != "" {
|
||||
if _, port, err := net.SplitHostPort(addr); err == nil {
|
||||
return net.JoinHostPort(h, port)
|
||||
}
|
||||
return h
|
||||
}
|
||||
host, port, err := net.SplitHostPort(addr)
|
||||
if err != nil {
|
||||
return addr
|
||||
}
|
||||
if host == "" || host == "0.0.0.0" || host == "::" || host == "[::]" {
|
||||
if h, herr := os.Hostname(); herr == nil && h != "" {
|
||||
return net.JoinHostPort(h, port)
|
||||
}
|
||||
}
|
||||
return addr
|
||||
}
|
||||
|
||||
// splitCSV splits a comma-separated list into a slice (empty -> nil).
|
||||
func splitCSV(s string) []string {
|
||||
if strings.TrimSpace(s) == "" {
|
||||
return nil
|
||||
}
|
||||
parts := strings.Split(s, ",")
|
||||
out := make([]string, 0, len(parts))
|
||||
for _, p := range parts {
|
||||
if p = strings.TrimSpace(p); p != "" {
|
||||
out = append(out, p)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// renderRemote builds the process.Render closure from store data.
|
||||
func renderRemote(st *store.Store) func(string) ([]byte, error) {
|
||||
return func(remoteName string) ([]byte, error) {
|
||||
rem, ok := st.GetRemote(remoteName)
|
||||
if !ok {
|
||||
return nil, store.ErrNotFound
|
||||
}
|
||||
forwards, err := st.LinksForRemote(remoteName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
proxies := make([]render.Proxy, 0, len(forwards))
|
||||
for _, f := range forwards {
|
||||
// A disabled forward is omitted from the generated frpc config so
|
||||
// a per-forward stop drops just this proxy on worker restart,
|
||||
// leaving sibling forwards on the same remote untouched.
|
||||
if f.Disabled {
|
||||
continue
|
||||
}
|
||||
loc, ok := st.GetLocal(f.Service)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
proxies = append(proxies, render.Proxy{
|
||||
Name: loc.Name,
|
||||
Type: loc.Protocol,
|
||||
LocalIP: loc.IP,
|
||||
LocalPort: loc.Port,
|
||||
RemotePort: f.RemotePort,
|
||||
UseEncryption: loc.UseEncryption,
|
||||
UseCompression: loc.UseCompression,
|
||||
BandwidthLimit: loc.BandwidthLimit,
|
||||
PoolCount: loc.PoolCount,
|
||||
Metadatas: loc.Metadatas,
|
||||
Annotations: loc.Annotations,
|
||||
CustomDomains: splitCSV(loc.CustomDomains),
|
||||
SubDomain: loc.SubDomain,
|
||||
Locations: loc.Locations,
|
||||
HostHeaderRewrite: loc.HostHeaderRewrite,
|
||||
HTTPHeaders: loc.HTTPHeaders,
|
||||
BasicAuthUser: loc.BasicAuthUser,
|
||||
BasicAuthPassword: loc.BasicAuthPassword,
|
||||
|
||||
// M3: load balancing group + health check.
|
||||
LBGroup: loc.LBGroup,
|
||||
LBGroupKey: loc.LBGroupKey,
|
||||
HealthCheckType: loc.HealthCheckType,
|
||||
HealthCheckPath: loc.HealthCheckPath,
|
||||
HealthCheckTimeout: loc.HealthCheckTimeout,
|
||||
HealthCheckMaxFailed: loc.HealthCheckMaxFailed,
|
||||
HealthCheckInterval: loc.HealthCheckInterval,
|
||||
})
|
||||
}
|
||||
return render.Render(rem, proxies)
|
||||
}
|
||||
}
|
||||
|
||||
// multiFlag collects repeated string flags (-peer a -peer b ...).
|
||||
type multiFlag []string
|
||||
|
||||
func (m *multiFlag) String() string { return strings.Join(*m, ",") }
|
||||
func (m *multiFlag) Set(v string) error {
|
||||
*m = append(*m, v)
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user