ITEM6 B3: agent sing-box runtime (parallel path)
Agent consumes singbox_config alongside xray: candidate->check-> flush->swap activation with .previous rollback, wire-level v2ray_api stats client with NexusGate epoch discipline (mid-query flip voids the round, new epoch counts in full, agent-first snapshot baselines), access-log online tracker (authenticated-connection lines, ANSI-safe, merged with snapshot-active users), per-core traffic baselines with the sing-box one carrying its epoch. install-agent.sh installs the self-built sing-box v1.13.21 next to xray (xray/geo removal deferred to the B5 switchover). Env-gated real-core E2E covers vless-reality, ss2022, mixed, trojan and the forwarding leg end to end; hysteria2 is probe-gated (sandbox blocks UDP replies) and stays a B5 pilot item.
This commit is contained in:
Muse
+178
-5
@@ -14,6 +14,7 @@ import (
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
@@ -84,6 +85,37 @@ type Agent struct {
|
||||
updateFailedAt time.Time
|
||||
updateReport *agentUpdateStatus
|
||||
exePathOverride string
|
||||
|
||||
// sing-box runtime (ITEM6 B3, see singbox.go). sbMu serializes
|
||||
// config activation, the sing-box traffic cycle and shutdown: a
|
||||
// flush-before-change must never race a periodic snapshot.
|
||||
sbMu sync.Mutex
|
||||
sbProc *singboxProcess
|
||||
sbStats *singboxStatsClient
|
||||
sbOnline *singboxOnlineTracker
|
||||
sbStatsAddr string
|
||||
lastSingboxHash string
|
||||
sbBaselineLoaded bool
|
||||
|
||||
// sing-box version cache (telemetry), mirroring the xray one.
|
||||
sbVerMu sync.Mutex
|
||||
sbVer string
|
||||
sbVerFor string
|
||||
sbVerMod time.Time
|
||||
}
|
||||
|
||||
// ensureSingboxState lazily creates the sing-box runtime state.
|
||||
// Callers hold sbMu (or are in single-threaded setup).
|
||||
func (a *Agent) ensureSingboxState() {
|
||||
if a.sbProc == nil {
|
||||
a.sbProc = &singboxProcess{}
|
||||
}
|
||||
if a.sbStats == nil {
|
||||
a.sbStats = newSingboxStatsClient(a.sbProc.epoch)
|
||||
}
|
||||
if a.sbOnline == nil {
|
||||
a.sbOnline = newSingboxOnlineTracker(singboxLogPath())
|
||||
}
|
||||
}
|
||||
|
||||
// New creates a new agent instance
|
||||
@@ -239,6 +271,7 @@ func (a *Agent) Run() error {
|
||||
case <-a.stopCh:
|
||||
logger.Info("Agent stopping...")
|
||||
a.stopXray()
|
||||
a.stopSingbox()
|
||||
return nil
|
||||
case <-heartbeatTicker.C:
|
||||
if err := a.sendHeartbeat(); err != nil {
|
||||
@@ -324,9 +357,11 @@ func (a *Agent) fetchAndApplyConfig() error {
|
||||
}
|
||||
|
||||
var result struct {
|
||||
ConfigHash string `json:"config_hash"`
|
||||
XrayConfig json.RawMessage `json:"xray_config"`
|
||||
AgentUpdate *updateIntent `json:"agent_update"`
|
||||
ConfigHash string `json:"config_hash"`
|
||||
XrayConfig json.RawMessage `json:"xray_config"`
|
||||
SingboxConfig json.RawMessage `json:"singbox_config"`
|
||||
SingboxConfigHash string `json:"singbox_config_hash"`
|
||||
AgentUpdate *updateIntent `json:"agent_update"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
return err
|
||||
@@ -336,6 +371,18 @@ func (a *Agent) fetchAndApplyConfig() error {
|
||||
// when the xray config itself is unchanged.
|
||||
a.maybeHandleUpdate(result.AgentUpdate)
|
||||
|
||||
// sing-box runtime (ITEM6 B3): an independent, parallel path. Its
|
||||
// failures are logged and retried on the next poll; they never fail
|
||||
// the xray apply below or mark the poll failed.
|
||||
if result.SingboxConfigHash != "" && result.SingboxConfigHash != a.lastSingboxHash {
|
||||
if err := a.applySingboxConfig(result.SingboxConfig); err != nil {
|
||||
logger.Warningf("Failed to apply sing-box config: %v", err)
|
||||
} else {
|
||||
a.lastSingboxHash = result.SingboxConfigHash
|
||||
logger.Info("sing-box config applied successfully")
|
||||
}
|
||||
}
|
||||
|
||||
// Skip if config hasn't changed
|
||||
if result.ConfigHash == a.lastConfigHash && result.ConfigHash != "" {
|
||||
return nil
|
||||
@@ -613,11 +660,25 @@ func (a *Agent) reportTrafficWith(sc statsClient, post func(trafficReport) error
|
||||
return nil
|
||||
}
|
||||
|
||||
// collectAndReportTraffic snapshots the local xray counters and reports
|
||||
// collectAndReportTraffic runs one reporting cycle per running core
|
||||
// (xray and, since ITEM6 B3, sing-box). The two cores keep separate
|
||||
// clients and baselines; a failure in one never blocks the other.
|
||||
func (a *Agent) collectAndReportTraffic() error {
|
||||
var firstErr error
|
||||
if err := a.collectAndReportXrayTraffic(); err != nil {
|
||||
firstErr = err
|
||||
}
|
||||
if err := a.collectAndReportSingboxTraffic(); err != nil && firstErr == nil {
|
||||
firstErr = err
|
||||
}
|
||||
return firstErr
|
||||
}
|
||||
|
||||
// collectAndReportXrayTraffic snapshots the local xray counters and reports
|
||||
// the deltas since the last master-acknowledged baseline. The XrayAPI
|
||||
// instance is kept across polls so baselines survive; xray restarts are
|
||||
// handled (counters resetting to zero are treated as fresh).
|
||||
func (a *Agent) collectAndReportTraffic() error {
|
||||
func (a *Agent) collectAndReportXrayTraffic() error {
|
||||
if a.xrayProc == nil || !a.xrayProc.IsRunning() {
|
||||
return nil
|
||||
}
|
||||
@@ -632,6 +693,118 @@ func (a *Agent) collectAndReportTraffic() error {
|
||||
return a.reportTrafficWith(&a.xrayAPI, a.postTrafficReport)
|
||||
}
|
||||
|
||||
// singboxBaselineFile is the on-disk form of the sing-box tracker's
|
||||
// last committed baseline. The epoch travels with the values: after
|
||||
// an agent restart against a still-running (or since-restarted) core,
|
||||
// a baseline from another core instance must not be restored.
|
||||
type singboxBaselineFile struct {
|
||||
Version int `json:"version"`
|
||||
Epoch string `json:"epoch"`
|
||||
Values map[string]int64 `json:"values"`
|
||||
}
|
||||
|
||||
// singboxBaselinePath derives the sing-box baseline location from the
|
||||
// xray one (agent-traffic-baseline.json -> agent-traffic-baseline-singbox.json),
|
||||
// so main's placement next to agent.json carries over.
|
||||
func (a *Agent) singboxBaselinePath() string {
|
||||
p := a.baselinePath()
|
||||
if strings.HasSuffix(p, ".json") {
|
||||
return strings.TrimSuffix(p, ".json") + "-singbox.json"
|
||||
}
|
||||
return p + "-singbox"
|
||||
}
|
||||
|
||||
func loadSingboxBaseline(path string) (*singboxBaselineFile, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var f singboxBaselineFile
|
||||
if err := json.Unmarshal(data, &f); err != nil {
|
||||
return nil, fmt.Errorf("parse sing-box traffic baseline: %w", err)
|
||||
}
|
||||
return &f, nil
|
||||
}
|
||||
|
||||
func saveSingboxBaseline(path, epoch string, values map[string]int64) error {
|
||||
data, err := json.Marshal(singboxBaselineFile{Version: 1, Epoch: epoch, Values: values})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tmp := path + ".tmp"
|
||||
if err := os.WriteFile(tmp, data, 0o600); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, path)
|
||||
}
|
||||
|
||||
// collectAndReportSingboxTraffic runs one sing-box reporting cycle.
|
||||
func (a *Agent) collectAndReportSingboxTraffic() error {
|
||||
a.sbMu.Lock()
|
||||
defer a.sbMu.Unlock()
|
||||
if a.sbProc == nil || !a.sbProc.isRunning() {
|
||||
return nil
|
||||
}
|
||||
return a.reportSingboxTrafficLocked()
|
||||
}
|
||||
|
||||
// reportSingboxTrafficLocked is the sing-box counterpart of
|
||||
// reportTrafficWith, with the same contract — commit only after the
|
||||
// master acknowledges (HTTP 200), persist after every commit — plus
|
||||
// the epoch carried through restore/persist. The online set merges
|
||||
// the access-log tracker with the users whose counters just moved.
|
||||
// Callers hold sbMu.
|
||||
func (a *Agent) reportSingboxTrafficLocked() error {
|
||||
a.ensureSingboxState()
|
||||
if a.sbStatsAddr == "" {
|
||||
// The applied config carries no v2ray_api stats block (the
|
||||
// panel-side stats switches are off): nothing exists to read.
|
||||
return nil
|
||||
}
|
||||
a.sbStats.SetAddr(a.sbStatsAddr)
|
||||
|
||||
if !a.sbBaselineLoaded {
|
||||
a.sbBaselineLoaded = true
|
||||
if f, err := loadSingboxBaseline(a.singboxBaselinePath()); err != nil {
|
||||
if !os.IsNotExist(err) {
|
||||
logger.Warningf("sing-box traffic baseline at %s unusable (%v); starting from a fresh baseline", a.singboxBaselinePath(), err)
|
||||
}
|
||||
} else {
|
||||
a.sbStats.RestoreTrafficBaseline(f.Values, f.Epoch)
|
||||
}
|
||||
}
|
||||
|
||||
a.sbOnline.Poll(time.Now())
|
||||
|
||||
snap, err := a.sbStats.GetTrafficSnapshot()
|
||||
if err != nil {
|
||||
if errors.Is(err, errSingboxEpochChanged) {
|
||||
// The core restarted around this poll (config change or
|
||||
// crash): this round establishes nothing; the next poll
|
||||
// rebases onto the new instance. Not a fault.
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
report := buildTrafficReport(snap, mergeOnlineUsers(a.sbOnline.Online(time.Now()), snap))
|
||||
if !report.empty() {
|
||||
if err := a.postTrafficReport(report); err != nil {
|
||||
// No commit: deltas are recomputed against the same
|
||||
// baseline next poll, exactly like the xray path.
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
a.sbStats.CommitTrafficSnapshot(snap)
|
||||
if err := saveSingboxBaseline(a.singboxBaselinePath(), a.sbProc.epoch(), snap.BaselineValues()); err != nil {
|
||||
// The in-memory commit stands (see the xray path's reasoning).
|
||||
logger.Warningf("Failed to persist sing-box traffic baseline: %v", err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// LoadConfig loads agent config from file
|
||||
func LoadConfig(path string) (*Config, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
|
||||
Reference in New Issue
Block a user