merge feat/probe-b6: item 7 B1-B6 (probe collector, history, contract API, ping, themes, notify)
# Conflicts: # internal/agent/agent.go
This commit is contained in:
+49
-1
@@ -102,6 +102,10 @@ type Agent struct {
|
||||
sbVer string
|
||||
sbVerFor string
|
||||
sbVerMod time.Time
|
||||
|
||||
// Latency probe runner (item 7 B4, see ping.go): task set pushed
|
||||
// on the config poll, samples returned on the heartbeat.
|
||||
probes *probeRunner
|
||||
}
|
||||
|
||||
// ensureSingboxState lazily creates the sing-box runtime state.
|
||||
@@ -132,6 +136,7 @@ func New(cfg *Config) *Agent {
|
||||
},
|
||||
},
|
||||
stopCh: make(chan struct{}),
|
||||
probes: newProbeRunner(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -150,9 +155,11 @@ func (e *statusError) Error() string {
|
||||
// Register registers the agent with the master
|
||||
func (a *Agent) Register() error {
|
||||
url := fmt.Sprintf("%s/panel/api/agent/register", a.config.MasterURL)
|
||||
payload := map[string]string{
|
||||
payload := map[string]any{
|
||||
"token": a.config.Token,
|
||||
"name": a.config.NodeName,
|
||||
// Probe facts ride along (B1): old masters ignore the extra field.
|
||||
"probeFacts": collectFacts(),
|
||||
}
|
||||
body, _ := json.Marshal(payload)
|
||||
req, err := http.NewRequest("POST", url, bytes.NewReader(body))
|
||||
@@ -291,6 +298,9 @@ func (a *Agent) Run() error {
|
||||
|
||||
// Stop stops the agent
|
||||
func (a *Agent) Stop() {
|
||||
if a.probes != nil {
|
||||
a.probes.stop()
|
||||
}
|
||||
close(a.stopCh)
|
||||
}
|
||||
|
||||
@@ -313,13 +323,30 @@ func (a *Agent) sendHeartbeat() error {
|
||||
"token": a.config.Token,
|
||||
"timestamp": time.Now().Unix(),
|
||||
"telemetry": a.collectTelemetry(),
|
||||
// Probe snapshot (B1): facts are process-cached so re-sending them
|
||||
// costs nothing and heals a master that missed the register-time
|
||||
// copy; metrics are sampled fresh each heartbeat. Old masters
|
||||
// ignore both fields.
|
||||
"probeFacts": collectFacts(),
|
||||
"probeMetrics": collectMetrics(),
|
||||
}
|
||||
if rep := a.currentUpdateReport(); rep != nil {
|
||||
payload["updateStatus"] = rep
|
||||
}
|
||||
// Latency probe samples (B4) ride the heartbeat — deliberately not
|
||||
// the traffic report, which stops when xray is down while probing
|
||||
// must not. Drained here, requeued on any failure below.
|
||||
var probeBatch []ProbeResult
|
||||
if a.probes != nil {
|
||||
probeBatch = a.probes.drainResults()
|
||||
if len(probeBatch) > 0 {
|
||||
payload["probeResults"] = probeBatch
|
||||
}
|
||||
}
|
||||
body, _ := json.Marshal(payload)
|
||||
req, err := http.NewRequest("POST", url, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
a.requeueProbeBatch(probeBatch)
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
@@ -327,15 +354,25 @@ func (a *Agent) sendHeartbeat() error {
|
||||
|
||||
resp, err := a.do(req)
|
||||
if err != nil {
|
||||
a.requeueProbeBatch(probeBatch)
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
a.requeueProbeBatch(probeBatch)
|
||||
return fmt.Errorf("heartbeat failed with status %d", resp.StatusCode)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// requeueProbeBatch returns a heartbeat's probe samples to the
|
||||
// runner buffer after a failed send.
|
||||
func (a *Agent) requeueProbeBatch(batch []ProbeResult) {
|
||||
if a.probes != nil {
|
||||
a.probes.requeueResults(batch)
|
||||
}
|
||||
}
|
||||
|
||||
// fetchAndApplyConfig fetches the xray config from master and applies it if changed
|
||||
func (a *Agent) fetchAndApplyConfig() error {
|
||||
url := fmt.Sprintf("%s/panel/api/agent/config", a.config.MasterURL)
|
||||
@@ -362,6 +399,11 @@ func (a *Agent) fetchAndApplyConfig() error {
|
||||
SingboxConfig json.RawMessage `json:"singbox_config"`
|
||||
SingboxConfigHash string `json:"singbox_config_hash"`
|
||||
AgentUpdate *updateIntent `json:"agent_update"`
|
||||
// ProbeTasks is a pointer on purpose (B4): a master too old
|
||||
// to know probes omits the field and the agent must keep its
|
||||
// current set, while a present-but-empty list means "run no
|
||||
// probes" and clears it.
|
||||
ProbeTasks *[]ProbeTask `json:"probe_tasks"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
return err
|
||||
@@ -383,6 +425,12 @@ func (a *Agent) fetchAndApplyConfig() error {
|
||||
}
|
||||
}
|
||||
|
||||
// Probe tasks change independently of the xray config, so they
|
||||
// are applied before the hash early-return below.
|
||||
if result.ProbeTasks != nil && a.probes != nil {
|
||||
a.probes.applyTasks(*result.ProbeTasks)
|
||||
}
|
||||
|
||||
// Skip if config hasn't changed
|
||||
if result.ConfigHash == a.lastConfigHash && result.ConfigHash != "" {
|
||||
return nil
|
||||
|
||||
Reference in New Issue
Block a user