From e03ceaa41fb2c0362a08637173fdfeb5b07d7057 Mon Sep 17 00:00:00 2001 From: Muse Date: Thu, 22 Oct 2026 21:57:23 +0000 Subject: [PATCH] B4: probe task store, panel API, heartbeat ingestion, agent prober - probe package: probe_ping_task/probe_ping_node tables; task CRUD with hub save semantics (id high-water, base-diff assignment, 64-per-agent crowding refusal, auto_join cap); filing with assignment-gated insert; PingHistory weighted-median chart read over minute+hour tiers; auto-join on first Record (ensureRows). - controller: /panel/api/agent/probe-tasks* management endpoints (node id space, hub validation chain verbatim); series=ping now serves real rows from probe.PingHistory; heartbeat ingests probeResults (missing latency skipped, batch capped); config response carries probe_tasks (always present, empty = clear). - agent: per-task probe loops (immediate first probe, ticker, 900ms resolve/dial budgets, <=3 addresses, resolve overrun = no sample), result buffer drained on heartbeat and requeued on failure; task set applied before the config-hash early return. --- internal/agent/agent.go | 40 ++ internal/agent/ping.go | 267 +++++++++ internal/web/controller/agent.go | 41 ++ internal/web/controller/api.go | 3 + internal/web/controller/probe_public.go | 24 +- internal/web/controller/probe_tasks.go | 333 +++++++++++ internal/web/service/probe/meta.go | 1 + internal/web/service/probe/ping.go | 739 ++++++++++++++++++++++++ internal/web/service/probe/probe.go | 100 ++++ internal/web/service/probe/store.go | 23 + internal/web/service/probe/traffic.go | 17 +- 11 files changed, 1582 insertions(+), 6 deletions(-) create mode 100644 internal/agent/ping.go create mode 100644 internal/web/controller/probe_tasks.go create mode 100644 internal/web/service/probe/ping.go diff --git a/internal/agent/agent.go b/internal/agent/agent.go index a1830d0..000569c 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -84,6 +84,10 @@ type Agent struct { updateFailedAt time.Time updateReport *agentUpdateStatus exePathOverride string + + // Latency probe runner (item 7 B4, see ping.go): task set pushed + // on the config poll, samples returned on the heartbeat. + probes *probeRunner } // New creates a new agent instance @@ -100,6 +104,7 @@ func New(cfg *Config) *Agent { }, }, stopCh: make(chan struct{}), + probes: newProbeRunner(), } } @@ -260,6 +265,9 @@ func (a *Agent) Run() error { // Stop stops the agent func (a *Agent) Stop() { + if a.probes != nil { + a.probes.stop() + } close(a.stopCh) } @@ -292,9 +300,20 @@ func (a *Agent) sendHeartbeat() error { 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") @@ -302,15 +321,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) @@ -335,6 +364,11 @@ func (a *Agent) fetchAndApplyConfig() error { ConfigHash string `json:"config_hash"` XrayConfig json.RawMessage `json:"xray_config"` 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 @@ -344,6 +378,12 @@ func (a *Agent) fetchAndApplyConfig() error { // when the xray config itself is unchanged. a.maybeHandleUpdate(result.AgentUpdate) + // 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 diff --git a/internal/agent/ping.go b/internal/agent/ping.go new file mode 100644 index 0000000..a40828e --- /dev/null +++ b/internal/agent/ping.go @@ -0,0 +1,267 @@ +package agent + +import ( + "context" + "net" + "sync" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/logger" +) + +// Item 7 B4: latency probes, the agent side. The master pushes the +// probe task set on the config poll (probe_tasks); the agent runs one +// loop per task — resolve, TCP-connect, time the handshake — and the +// samples ride the next heartbeat back (probeResults). A port of +// monitor-probe's agent prober (agent/src/main.rs tcp_probe and its +// per-task scheduler), with the same numbers: at most 64 tasks, a +// 900 ms budget each for resolution and for every dial, at most three +// addresses tried, and a resolution overrun producing NO sample at +// all rather than a recorded miss. + +// ProbeTask is one probe target as pushed by the master. +type ProbeTask struct { + ID int64 `json:"id"` + Target string `json:"target"` + Interval int64 `json:"interval"` +} + +// ProbeResult is one filed sample: the task and its handshake latency +// in milliseconds; negative means the target did not answer. +type ProbeResult struct { + TaskID int64 `json:"taskId"` + LatencyMs int64 `json:"latencyMs"` +} + +const ( + // maxProbeTasksRun mirrors the master's per-agent assignment cap + // (probe.MaxProbesPerNode); an over-long list is truncated, never + // run whole, with the loss logged once per apply. + maxProbeTasksRun = 64 + // probeDialBudget bounds name resolution and every single dial. + probeDialBudget = 900 * time.Millisecond + // probeMaxAddrs caps how many resolved addresses one probe tries. + probeMaxAddrs = 3 + // probeResultCap bounds the unsent-sample buffer; past it the + // oldest samples drop (they describe a master outage, and their + // timestamps would be discarded on arrival anyway). + probeResultCap = 8192 + // Probe intervals are clamped into the master's legal range on + // arrival as a backstop (the panel refuses out-of-range values). + minProbeIntervalSec = 5 + maxProbeIntervalSec = 3600 +) + +type runningProbe struct { + target string + interval int64 + cancel context.CancelFunc +} + +// probeRunner owns the running probe loops and the result buffer. +// All state is behind mu; probe loops only touch it through methods. +type probeRunner struct { + mu sync.Mutex + tasks map[int64]*runningProbe + results []ProbeResult + dropped int64 + resolveLogged map[int64]bool + stopped bool +} + +func newProbeRunner() *probeRunner { + return &probeRunner{ + tasks: map[int64]*runningProbe{}, + resolveLogged: map[int64]bool{}, + } +} + +// clampProbeInterval folds an interval into the legal range. +func clampProbeInterval(sec int64) int64 { + if sec < minProbeIntervalSec { + return minProbeIntervalSec + } + if sec > maxProbeIntervalSec { + return maxProbeIntervalSec + } + return sec +} + +// applyTasks reconciles the running loops with a freshly pushed set: +// a task whose (target, interval) is unchanged keeps its loop and its +// phase; anything else is cancelled and, if still wanted, restarted. +func (r *probeRunner) applyTasks(tasks []ProbeTask) { + if len(tasks) > maxProbeTasksRun { + logger.Warningf("probe: master pushed %d tasks, running only the first %d", len(tasks), maxProbeTasksRun) + tasks = tasks[:maxProbeTasksRun] + } + want := make(map[int64]ProbeTask, len(tasks)) + for _, t := range tasks { + t.Interval = clampProbeInterval(t.Interval) + want[t.ID] = t + } + + r.mu.Lock() + defer r.mu.Unlock() + if r.stopped { + return + } + for id, running := range r.tasks { + t, ok := want[id] + if !ok || t.Target != running.target || t.Interval != running.interval { + running.cancel() + delete(r.tasks, id) + } + } + for id, t := range want { + if _, ok := r.tasks[id]; ok { + continue + } + ctx, cancel := context.WithCancel(context.Background()) + r.tasks[id] = &runningProbe{target: t.Target, interval: t.Interval, cancel: cancel} + go r.runLoop(ctx, t) + } +} + +// stop cancels every probe loop (agent shutdown). +func (r *probeRunner) stop() { + r.mu.Lock() + defer r.mu.Unlock() + r.stopped = true + for id, running := range r.tasks { + running.cancel() + delete(r.tasks, id) + } +} + +// runLoop probes immediately, then on the task's interval. A tick +// that arrives while a probe is still running is dropped rather than +// queued (the hub's MissedTickBehavior::Delay; a Go ticker behaves +// the same way). +func (r *probeRunner) runLoop(ctx context.Context, task ProbeTask) { + r.probeOnce(task) + ticker := time.NewTicker(time.Duration(task.Interval) * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + r.probeOnce(task) + } + } +} + +func (r *probeRunner) probeOnce(task ProbeTask) { + ms, sampled := probeTCP(task.Target, func() { + r.noteResolveTimeout(task.ID, task.Target) + }) + if !sampled { + return + } + r.addResult(ProbeResult{TaskID: task.ID, LatencyMs: ms}) +} + +// noteResolveTimeout logs a skipped round once per task per process: +// a target whose name stops resolving would otherwise be silent +// forever (nothing is filed for it), and one line per interval per +// target would flood the log on a DNS outage (hub discipline). +func (r *probeRunner) noteResolveTimeout(taskID int64, target string) { + r.mu.Lock() + defer r.mu.Unlock() + if r.resolveLogged[taskID] { + return + } + r.resolveLogged[taskID] = true + logger.Warningf("probe: resolving %s overran %s; round skipped (further rounds log nothing)", target, probeDialBudget) +} + +func (r *probeRunner) addResult(res ProbeResult) { + r.mu.Lock() + defer r.mu.Unlock() + if len(r.results) >= probeResultCap { + r.results = r.results[1:] + r.dropped++ + } + r.results = append(r.results, res) +} + +// drainResults takes the buffered samples for one heartbeat. The +// caller requeues them when the heartbeat fails, so a master outage +// loses nothing that still fits the buffer. +func (r *probeRunner) drainResults() []ProbeResult { + r.mu.Lock() + defer r.mu.Unlock() + if len(r.results) == 0 { + return nil + } + out := r.results + r.results = nil + return out +} + +// requeueResults puts a failed heartbeat's batch back ahead of +// anything probed since, trimming the oldest if the buffer overflows. +func (r *probeRunner) requeueResults(batch []ProbeResult) { + if len(batch) == 0 { + return + } + r.mu.Lock() + defer r.mu.Unlock() + merged := make([]ProbeResult, 0, len(batch)+len(r.results)) + merged = append(merged, batch...) + merged = append(merged, r.results...) + if len(merged) > probeResultCap { + merged = merged[len(merged)-probeResultCap:] + } + r.results = merged +} + +// probeTCP measures one target's TCP handshake latency in +// milliseconds. It returns (ms, true) for a sample to file — the +// handshake time, or -1 when the target did not answer — and +// (0, false) when the round produced no sample: name resolution +// overran its budget, which the hub treats as "nothing happened" +// rather than a miss, because filing it would blame the target for +// the resolver's state. A malformed target is -1: it is re-probed +// every interval and reads as total loss, which is what it is. +func probeTCP(target string, onResolveTimeout func()) (int64, bool) { + host, port, err := net.SplitHostPort(target) + if err != nil { + return -1, true + } + rctx, cancel := context.WithTimeout(context.Background(), probeDialBudget) + addrs, err := (&net.Resolver{}).LookupIPAddr(rctx, host) + timedOut := rctx.Err() == context.DeadlineExceeded + cancel() + if err != nil { + if timedOut { + if onResolveTimeout != nil { + onResolveTimeout() + } + return 0, false + } + return -1, true + } + if len(addrs) == 0 { + return -1, true + } + dialer := &net.Dialer{} + tried := 0 + for _, addr := range addrs { + if tried >= probeMaxAddrs { + break + } + tried++ + dctx, dcancel := context.WithTimeout(context.Background(), probeDialBudget) + start := time.Now() + conn, derr := dialer.DialContext(dctx, "tcp", net.JoinHostPort(addr.IP.String(), port)) + elapsed := time.Since(start) + dcancel() + if derr == nil { + conn.Close() + return elapsed.Milliseconds(), true + } + } + return -1, true +} diff --git a/internal/web/controller/agent.go b/internal/web/controller/agent.go index 046580c..58ac44a 100644 --- a/internal/web/controller/agent.go +++ b/internal/web/controller/agent.go @@ -511,6 +511,23 @@ func (a *AgentController) confirmAgentUpdate(id uint, reportedVersion string) { }).Error } +// heartbeatProbeResult is one latency sample an agent piggybacks on +// its heartbeat (item 7 B4). LatencyMs is a pointer: an absent field +// means "no sample" (a resolution overrun produces none) and the row +// is skipped, while a present negative value is a genuine recorded +// miss — the two must never collapse into each other. +type heartbeatProbeResult struct { + TaskID int64 `json:"taskId"` + LatencyMs *int64 `json:"latencyMs"` +} + +// maxProbeResultsPerHeartbeat caps one heartbeat's probe batch. The +// agent's buffer holds far less than this in practice (64 tasks × 60 s +// at the fastest legal interval is under a thousand samples); the cap +// exists so a buggy or hostile reporter cannot make one request file +// an unbounded batch. +const maxProbeResultsPerHeartbeat = 2048 + func (a *AgentController) heartbeat(c *gin.Context) { agent, ok := a.findAgent(bearerToken(c)) if !ok { @@ -525,6 +542,7 @@ func (a *AgentController) heartbeat(c *gin.Context) { UpdateStatus *heartbeatUpdateStatus `json:"updateStatus"` ProbeFacts *heartbeatProbeFacts `json:"probeFacts"` ProbeMetrics *heartbeatProbeMetrics `json:"probeMetrics"` + ProbeResults []heartbeatProbeResult `json:"probeResults"` } _ = c.ShouldBindJSON(&req) // The last-seen stamp as it stood before this heartbeat: the probe @@ -542,6 +560,23 @@ func (a *AgentController) heartbeat(c *gin.Context) { // probe package logs and swallows its own failures. probe.Record(int64(agent.Id), probeSample(req.ProbeMetrics), prevLastSeen) } + if len(req.ProbeResults) > 0 { + // Item 7 B4: file the latency batch, stamped at arrival (hub + // semantics). Results ride the heartbeat rather than the + // traffic report on purpose: probing must keep working while + // xray is down, and the traffic report early-returns then. + results := make([]probe.PingResult, 0, len(req.ProbeResults)) + for _, r := range req.ProbeResults { + if r.TaskID <= 0 || r.LatencyMs == nil { + continue + } + results = append(results, probe.PingResult{TaskID: r.TaskID, LatencyMs: *r.LatencyMs}) + if len(results) >= maxProbeResultsPerHeartbeat { + break + } + } + probe.InsertPings(int64(agent.Id), results) + } if req.Telemetry != nil { a.updateAgentTelemetry(uint(agent.Id), req.Telemetry) if req.Telemetry.AgentVersion != nil && *req.Telemetry.AgentVersion != "" { @@ -581,6 +616,12 @@ func (a *AgentController) getConfig(c *gin.Context) { payload := gin.H{ "config_hash": hex.EncodeToString(sum[:]), "xray_config": json.RawMessage(raw), + // Item 7 B4: the agent's probe task set rides the existing + // config poll (no new channel). Always present from this + // master version — an empty list means "run no probes", while + // an absent field (older master) leaves the agent's current + // set untouched. + "probe_tasks": probe.PingTasksFor(int64(agent.Id)), } // Item 20: a pending one-click update intent rides along on the config // poll (no new channel — the agent already polls this endpoint). diff --git a/internal/web/controller/api.go b/internal/web/controller/api.go index a1d358d..917e4e2 100644 --- a/internal/web/controller/api.go +++ b/internal/web/controller/api.go @@ -198,6 +198,9 @@ func (a *APIController) initRouter(g *gin.RouterGroup) { // Item 7 B3: probe metadata editing rides the same admin group // (/panel/api/agent/probe-meta/:nodeId). NewProbeMetaController(agentAdmin) + // Item 7 B4: probe task (latency target) management, same group + // (/panel/api/agent/probe-tasks*). + NewProbeTasksController(agentAdmin) // Hosts API — per-inbound override endpoints for subscription links hosts := api.Group("/hosts") diff --git a/internal/web/controller/probe_public.go b/internal/web/controller/probe_public.go index 83acf74..a1e6a42 100644 --- a/internal/web/controller/probe_public.go +++ b/internal/web/controller/probe_public.go @@ -120,7 +120,7 @@ func (pc *PublicProbeController) nodes(c *gin.Context) { type metricsResponse struct { Metrics []probe.MetricRow `json:"metrics"` - Ping []any `json:"ping"` + Ping []probe.PingRow `json:"ping"` Probes map[string]string `json:"probes"` Loss map[string]float64 `json:"loss"` Step int64 `json:"step"` @@ -171,12 +171,11 @@ func (pc *PublicProbeController) nodeMetrics(c *gin.Context) { series := c.Query("series") wantMetrics := series == "" || series == "metrics" wantPing := series == "" || series == "ping" - _ = wantPing // ping rows are written by B4; until then the series is empty by construction span := probe.ComputeSpan(hours, points, time.Now().Unix()) resp := metricsResponse{ Metrics: []probe.MetricRow{}, - Ping: []any{}, + Ping: []probe.PingRow{}, Probes: map[string]string{}, Loss: map[string]float64{}, Step: span.Step, @@ -190,6 +189,25 @@ func (pc *PublicProbeController) nodeMetrics(c *gin.Context) { } resp.Metrics = rows } + if wantPing { + // B4: the real latency series over the same window/bucket as + // the metrics above (hub metrics handler series=ping). + pingRows, pingLoss, err := probe.PingHistory(id, span) + if err != nil { + logger.Warningf("probe public: ping %d: %v", id, err) + answerText(c, http.StatusInternalServerError, "服务内部出错") + return + } + names, err := probe.PingTaskNames(id) + if err != nil { + logger.Warningf("probe public: ping names %d: %v", id, err) + answerText(c, http.StatusInternalServerError, "服务内部出错") + return + } + resp.Ping = pingRows + resp.Probes = names + resp.Loss = pingLoss + } c.JSON(http.StatusOK, resp) } diff --git a/internal/web/controller/probe_tasks.go b/internal/web/controller/probe_tasks.go new file mode 100644 index 0000000..e89af45 --- /dev/null +++ b/internal/web/controller/probe_tasks.go @@ -0,0 +1,333 @@ +package controller + +import ( + "errors" + "fmt" + "strconv" + "strings" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/web/service/probe" + + "github.com/gin-gonic/gin" +) + +// Item 7 B4: probe task management — the operator-maintained latency +// target set (ITEM7-SCOPE §5.7, user ruling Q4: no preset list; the +// operator keeps targets, grouped by the free-form region label, from +// the panel). Endpoints ride the panel-authenticated agent admin +// group like B3's probe-meta, and validation mirrors hub api.rs's +// save_ping_task chain refusal for refusal — including refusing (not +// clamping) an out-of-range interval. +// +// The panel-facing id space is the panel NODE id, as on the +// probe-meta surface; assignments are stored against agent ids and +// translated at this layer, where the binding lives. + +type ProbeTasksController struct{} + +func NewProbeTasksController(g *gin.RouterGroup) *ProbeTasksController { + pc := &ProbeTasksController{} + g.GET("/probe-tasks", pc.listTasks) + g.PUT("/probe-tasks", pc.saveTask) + g.DELETE("/probe-tasks/:id", pc.deleteTask) + g.POST("/probe-tasks/reorder", pc.reorderTasks) + g.PUT("/probe-tasks/by-node/:nodeId", pc.setNodeTasks) + return pc +} + +// probeTaskForm is the wire shape of one task as the panel edits it. +type probeTaskForm struct { + ID int64 `json:"id"` + Name string `json:"name"` + Target string `json:"target"` + Interval int64 `json:"interval"` + AutoJoin bool `json:"autoJoin"` + Sort int64 `json:"sort"` + Region string `json:"region"` + NodeIDs []int64 `json:"nodeIds"` + BaseNodeIDs *[]int64 `json:"baseNodeIds"` +} + +// agentsByNode maps panel node id → agent row for every bound agent. +func agentsByNode() (map[int64]model.AgentNode, error) { + var agents []model.AgentNode + if err := database.GetDB().Where("node_id IS NOT NULL").Find(&agents).Error; err != nil { + return nil, err + } + out := make(map[int64]model.AgentNode, len(agents)) + for _, ag := range agents { + out[int64(*ag.NodeID)] = ag + } + return out, nil +} + +// agentDisplayName renders an agent for operator-facing messages: the +// bound node's name where there is one (the panel identifies nodes by +// name and never shows an id, as the hub does). +func agentDisplayName(agentID int64) string { + var ag model.AgentNode + if err := database.GetDB().Where("id = ?", agentID).First(&ag).Error; err == nil { + if ag.NodeID != nil { + var node model.Node + if err := database.GetDB().Where("id = ?", *ag.NodeID).First(&node).Error; err == nil && node.Name != "" { + return node.Name + } + } + if ag.Name != "" { + return ag.Name + } + } + return fmt.Sprintf("#%d", agentID) +} + +// nodeName renders a panel node for messages, "" when it is gone. +func nodeName(nodeID int64) string { + var node model.Node + if err := database.GetDB().Where("id = ?", nodeID).First(&node).Error; err != nil { + return "" + } + return node.Name +} + +// resolveAgentIDs translates the form's node ids into agent ids, +// refusing the whole save when one names a node that is gone or has +// no agent bound (the hub's foreign key does this by construction; +// the naming keeps the refusal readable). +func resolveAgentIDs(nodeIDs []int64, bound map[int64]model.AgentNode) ([]int64, string) { + out := make([]int64, 0, len(nodeIDs)) + for _, nodeID := range nodeIDs { + ag, ok := bound[nodeID] + if !ok { + if name := nodeName(nodeID); name != "" { + return nil, fmt.Sprintf("节点「%s」未绑定 agent，无法指派探测任务", name) + } + return nil, fmt.Sprintf("节点 %d 不存在，可能已被删除", nodeID) + } + out = append(out, int64(ag.Id)) + } + return out, "" +} + +func (pc *ProbeTasksController) listTasks(c *gin.Context) { + tasks, err := probe.PingTasks() + if err != nil { + jsonMsg(c, "load probe tasks failed", err) + return + } + bound, err := agentsByNode() + if err != nil { + jsonMsg(c, "load probe tasks failed", err) + return + } + nodeByAgent := make(map[int64]int64, len(bound)) + for nodeID, ag := range bound { + nodeByAgent[int64(ag.Id)] = nodeID + } + forms := make([]probeTaskForm, 0, len(tasks)) + for _, t := range tasks { + f := probeTaskForm{ + ID: t.ID, Name: t.Name, Target: t.Target, Interval: t.Interval, + AutoJoin: t.AutoJoin, Sort: t.Sort, Region: t.Region, NodeIDs: []int64{}, + } + for _, agentID := range t.AgentIDs { + if nodeID, ok := nodeByAgent[agentID]; ok { + f.NodeIDs = append(f.NodeIDs, nodeID) + } + } + forms = append(forms, f) + } + jsonObj(c, forms, nil) +} + +// validProbeTarget mirrors hub valid_target: a target the agent can +// resolve — host:port, an IPv6 literal bracketed as a URL writes one. +// A bare contains(':') would admit a bare IPv6 address, ":443" with no +// host and "host:" with no port, each of which probes at 100% loss +// forever with nothing identifying the target as the cause. +func validProbeTarget(target string) bool { + var host, port string + if rest, ok := strings.CutPrefix(target, "["); ok { + h, p, found := strings.Cut(rest, "]:") + if !found { + return false + } + host, port = h, p + } else { + // rsplit_once(':'): host:port splits on the last colon. + i := strings.LastIndex(target, ":") + if i < 0 { + return false + } + host, port = target[:i], target[i+1:] + } + if host == "" || strings.Contains(host, ":") && !strings.HasPrefix(target, "[") { + return false + } + n, err := strconv.Atoi(port) + return err == nil && n > 0 && n <= 65535 +} + +// validateProbeTaskForm normalizes the form in place (hub order: +// trim, presence, name length, target shape, interval range) and +// returns the first refusal's message, "" when it may be stored. The +// region rule is ours (the hub has no region): the meta group rule. +func validateProbeTaskForm(f *probeTaskForm) string { + // Trimmed into the stored value rather than a discarded copy: + // what reaches the agent is the stored target, and a trailing + // space from a paste would leave the probe reporting loss + // indefinitely (hub discipline, same for the name). + f.Name = strings.TrimSpace(f.Name) + f.Target = strings.TrimSpace(f.Target) + f.Region = strings.TrimSpace(f.Region) + if f.Name == "" || f.Target == "" { + return "请填写名称和目标" + } + if len([]rune(f.Name)) > 64 { + return "名称最多 64 个字" + } + if !validProbeTarget(f.Target) { + return "目标要写成「主机:端口」，例如 1.1.1.1:443 或 [2606:4700:4700::1111]:443" + } + // Refused rather than clamped: the agent clamps again on + // arrival, so an out-of-range value would silently become a + // different number while the panel still displays what was + // entered (hub). + if f.Interval < probe.MinProbeInterval || f.Interval > 3600 { + return fmt.Sprintf("间隔要在 %d 到 3600 秒之间", probe.MinProbeInterval) + } + if len([]rune(f.Region)) > 13 || hasControl(f.Region) { + return "地区最多 13 个字，不能含控制字符" + } + return "" +} + +// probeTaskError renders a store refusal in the hub's words. +func probeTaskError(c *gin.Context, err error) { + var crowded *probe.PingCrowdedError + var autoJoin *probe.PingAutoJoinError + switch { + case errors.Is(err, probe.ErrPingTaskGone): + jsonMsg(c, err.Error(), errRequest) + case errors.As(err, &crowded): + jsonMsg(c, fmt.Sprintf("节点「%s」会被分配超过 %d 个探测任务，agent 最多只跑这么多，多出来的会被静默丢掉", + agentDisplayName(crowded.AgentID), probe.MaxProbesPerNode), errRequest) + case errors.As(err, &autoJoin): + jsonMsg(c, fmt.Sprintf("新节点会自动加入 %d 个探测任务，agent 最多只跑 %d 个", + autoJoin.Count, probe.MaxProbesPerNode), errRequest) + default: + jsonMsg(c, err.Error(), errRequest) + } +} + +func (pc *ProbeTasksController) saveTask(c *gin.Context) { + var form probeTaskForm + if err := c.ShouldBindJSON(&form); err != nil { + jsonMsg(c, "Invalid request", err) + return + } + if form.NodeIDs == nil { + form.NodeIDs = []int64{} + } + if msg := validateProbeTaskForm(&form); msg != "" { + jsonMsg(c, msg, errRequest) + return + } + bound, err := agentsByNode() + if err != nil { + jsonMsg(c, "save probe task failed", err) + return + } + agentIDs, refusal := resolveAgentIDs(form.NodeIDs, bound) + if refusal != "" { + jsonMsg(c, refusal, errRequest) + return + } + var base *[]int64 + if form.BaseNodeIDs != nil { + baseIDs, refusal := resolveAgentIDs(*form.BaseNodeIDs, bound) + if refusal != "" { + jsonMsg(c, refusal, errRequest) + return + } + base = &baseIDs + } + id, err := probe.SavePingTask(probe.PingTask{ + ID: form.ID, + Name: form.Name, + Target: form.Target, + Interval: form.Interval, + AutoJoin: form.AutoJoin, + Region: form.Region, + AgentIDs: agentIDs, + }, base) + if err != nil { + probeTaskError(c, err) + return + } + jsonObj(c, gin.H{"id": id}, nil) +} + +func (pc *ProbeTasksController) deleteTask(c *gin.Context) { + id, err := strconv.ParseInt(c.Param("id"), 10, 64) + if err != nil { + jsonMsg(c, "Invalid request", err) + return + } + if err := probe.DeletePingTask(id); err != nil { + jsonMsg(c, "delete probe task failed", err) + return + } + jsonObj(c, gin.H{"message": "ok"}, nil) +} + +func (pc *ProbeTasksController) reorderTasks(c *gin.Context) { + var form struct { + IDs []int64 `json:"ids"` + } + if err := c.ShouldBindJSON(&form); err != nil { + jsonMsg(c, "Invalid request", err) + return + } + if err := probe.ReorderPingTasks(form.IDs); err != nil { + jsonMsg(c, err.Error(), errRequest) + return + } + jsonObj(c, gin.H{"message": "ok"}, nil) +} + +// setNodeTasks edits one node's probe assignments from the node's +// side (hub set_node_ping_tasks): tasks/base are task ids, only the +// difference from base is applied. +func (pc *ProbeTasksController) setNodeTasks(c *gin.Context) { + nodeID, err := strconv.Atoi(c.Param("nodeId")) + if err != nil { + c.JSON(404, gin.H{"message": "agent not found"}) + return + } + var agent model.AgentNode + if err := database.GetDB().Where("node_id = ?", nodeID).First(&agent).Error; err != nil { + c.JSON(404, gin.H{"message": "agent not found"}) + return + } + var form struct { + Tasks []int64 `json:"tasks"` + Base []int64 `json:"base"` + } + if err := c.ShouldBindJSON(&form); err != nil { + jsonMsg(c, "监控列表格式不对", errRequest) + return + } + if form.Tasks == nil { + form.Tasks = []int64{} + } + if form.Base == nil { + form.Base = []int64{} + } + if err := probe.SetNodePingTasks(int64(agent.Id), form.Tasks, form.Base); err != nil { + probeTaskError(c, err) + return + } + jsonObj(c, gin.H{"message": "ok"}, nil) +} diff --git a/internal/web/service/probe/meta.go b/internal/web/service/probe/meta.go index 4ab6d17..aa17ce0 100644 --- a/internal/web/service/probe/meta.go +++ b/internal/web/service/probe/meta.go @@ -143,6 +143,7 @@ func (s *Store) DeleteAgentData(agentID int64) error { for _, table := range []string{ "probe_meta", "probe_traffic", "probe_metric", "probe_metric_hour", "probe_ping_record", "probe_ping_hour", + "probe_ping_node", } { if _, err := tx.Exec(`DELETE FROM `+table+` WHERE agent_id = ?`, agentID); err != nil { return err diff --git a/internal/web/service/probe/ping.go b/internal/web/service/probe/ping.go new file mode 100644 index 0000000..8818656 --- /dev/null +++ b/internal/web/service/probe/ping.go @@ -0,0 +1,739 @@ +package probe + +import ( + "database/sql" + "errors" + "fmt" + "sort" + "strconv" +) + +// Item 7 B4: probe (latency) tasks — the operator-maintained target +// set, its per-agent assignment, the filing of agent-reported results +// and the chart read. A port of hub db.rs's ping section and api.rs's +// validation chain (ITEM7-SCOPE §5.7): no preset targets exist, the +// operator keeps the list (grouped by the free-form region label). +// +// The one structural difference from the hub: hub nodes live in the +// same database, so its foreign keys police assignments. Here the +// agents live in the panel database and probe.db cannot reference +// them; the controller validates agent existence against the panel +// database before saving, and filing still re-checks the assignment +// inside the insert statement, exactly like hub insert_pings. + +// MaxProbesPerNode bounds how many probe tasks one agent may be +// assigned; the agent enforces the same cap as a backstop (hub +// MAX_PROBES_PER_NODE). The master refuses the write that would +// exceed it, because a silent truncation would draw empty charts. +const MaxProbesPerNode = 64 + +// MinProbeInterval is the fastest probe cadence, in seconds; the +// editing surface refuses anything outside [MinProbeInterval, 3600] +// rather than clamping it (hub save_ping_task). +const MinProbeInterval = 5 + +// pingTaskIDIssuedKey is the probe_setting high-water mark of issued +// ping task ids (hub "{table}_id_issued"). +const pingTaskIDIssuedKey = "ping_task_id_issued" + +// PingTask is one probe target as the panel keeps it. AgentIDs is the +// assignment side (probe_ping_node), filled by reads. +type PingTask struct { + ID int64 + Name string + Target string + Interval int64 + AutoJoin bool + Sort int64 + Region string + AgentIDs []int64 +} + +// PingTaskRef is the wire form of one task as pushed to an agent: +// id, target and interval, nothing else (hub ping_tasks_for). +type PingTaskRef struct { + ID int64 `json:"id"` + Target string `json:"target"` + Interval int64 `json:"interval"` +} + +// PingResult is one probe sample an agent reports: the task it ran +// and the handshake latency in milliseconds (negative = the target +// did not answer; a resolution overrun produces no sample at all and +// never reaches this type). +type PingResult struct { + TaskID int64 + LatencyMs int64 +} + +// ErrPingTaskGone reports an update against a task another session +// deleted first (hub save_ping_task's refusal). +var ErrPingTaskGone = errors.New("监控不存在，可能已被删除") + +// PingCrowdedError reports a save that would leave one agent over +// MaxProbesPerNode; the controller renders the hub message with the +// agent's display name. +type PingCrowdedError struct{ AgentID int64 } + +func (e *PingCrowdedError) Error() string { + return fmt.Sprintf("agent %d would exceed %d probe tasks", e.AgentID, MaxProbesPerNode) +} + +// PingAutoJoinError reports a save under which the next agent to +// appear would auto-join more tasks than an agent can run. +type PingAutoJoinError struct{ Count int64 } + +func (e *PingAutoJoinError) Error() string { + return fmt.Sprintf("new nodes would auto-join %d probe tasks", e.Count) +} + +const pingTaskColumns = `id, name, target, interval, auto_join, sort, region` + +func scanPingTask(sc interface{ Scan(...any) error }) (PingTask, error) { + var t PingTask + var autoJoin int64 + if err := sc.Scan(&t.ID, &t.Name, &t.Target, &t.Interval, &autoJoin, &t.Sort, &t.Region); err != nil { + return PingTask{}, err + } + t.AutoJoin = autoJoin != 0 + return t, nil +} + +// PingTasks lists every probe task in panel order (sort, then id), +// each with its assigned agent ids. +func (s *Store) PingTasks() ([]PingTask, error) { + rows, err := s.db.Query(`SELECT ` + pingTaskColumns + ` FROM probe_ping_task ORDER BY sort, id`) + if err != nil { + return nil, err + } + var tasks []PingTask + for rows.Next() { + t, err := scanPingTask(rows) + if err != nil { + rows.Close() + return nil, err + } + tasks = append(tasks, t) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + if tasks == nil { + tasks = []PingTask{} + } + assignRows, err := s.db.Query(`SELECT task_id, agent_id FROM probe_ping_node ORDER BY agent_id`) + if err != nil { + return nil, err + } + defer assignRows.Close() + byTask := make(map[int64][]int64) + for assignRows.Next() { + var taskID, agentID int64 + if err := assignRows.Scan(&taskID, &agentID); err != nil { + return nil, err + } + byTask[taskID] = append(byTask[taskID], agentID) + } + if err := assignRows.Err(); err != nil { + return nil, err + } + for i := range tasks { + if ids := byTask[tasks[i].ID]; ids != nil { + tasks[i].AgentIDs = ids + } else { + tasks[i].AgentIDs = []int64{} + } + } + return tasks, nil +} + +// nextPingTaskID issues the next task id from the high-water mark, +// falling back to the table's own maximum for databases that predate +// the mark (hub next_id/issued). Runs inside the save transaction. +func nextPingTaskID(tx *sql.Tx) (int64, error) { + var issued, maxID int64 + var raw sql.NullString + if err := tx.QueryRow(`SELECT value FROM probe_setting WHERE key = ?`, pingTaskIDIssuedKey).Scan(&raw); err != nil && err != sql.ErrNoRows { + return 0, err + } else if raw.Valid { + if n, perr := strconv.ParseInt(raw.String, 10, 64); perr == nil { + issued = n + } + } + if err := tx.QueryRow(`SELECT COALESCE(MAX(id), 0) FROM probe_ping_task`).Scan(&maxID); err != nil { + return 0, err + } + next := issued + if maxID > next { + next = maxID + } + next++ + if _, err := tx.Exec(`INSERT INTO probe_setting (key, value) VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value`, + pingTaskIDIssuedKey, fmt.Sprint(next)); err != nil { + return 0, err + } + return next, nil +} + +// SavePingTask creates (ID == 0) or replaces one probe task and its +// assignments, in one transaction (hub save_ping_task). base, when +// non-nil on an update, is the assignment set the editor started +// from: only the difference from it is applied, so an assignment +// made while the editor was open survives. Without base the +// assignment is replaced wholesale. The caller validates field +// values and agent existence beforehand (hub api layer). +func (s *Store) SavePingTask(t PingTask, base *[]int64) (int64, error) { + tx, err := s.db.Begin() + if err != nil { + return 0, err + } + defer tx.Rollback() + + id := t.ID + if id > 0 { + res, err := tx.Exec(`UPDATE probe_ping_task SET name=?2, target=?3, interval=?4, auto_join=?5, region=?6 WHERE id=?1`, + id, t.Name, t.Target, t.Interval, boolToInt(t.AutoJoin), t.Region) + if err != nil { + return 0, err + } + if n, err := res.RowsAffected(); err != nil { + return 0, err + } else if n == 0 { + // Deleted from another session; without this the save + // would report success against nothing. + return 0, ErrPingTaskGone + } + } else { + next, err := nextPingTaskID(tx) + if err != nil { + return 0, err + } + if _, err := tx.Exec(`INSERT INTO probe_ping_task (id, name, target, interval, auto_join, sort, region) + VALUES (?1, ?2, ?3, ?4, ?5, (SELECT COALESCE(MAX(sort), -1) + 1 FROM probe_ping_task), ?6)`, + next, t.Name, t.Target, t.Interval, boolToInt(t.AutoJoin), t.Region); err != nil { + return 0, err + } + id = next + } + + var added []int64 + if base != nil && t.ID > 0 { + baseSet := int64Set(*base) + newSet := int64Set(t.AgentIDs) + for _, agentID := range *base { + if !newSet[agentID] { + if _, err := tx.Exec(`DELETE FROM probe_ping_node WHERE task_id=?1 AND agent_id=?2`, id, agentID); err != nil { + return 0, err + } + } + } + for _, agentID := range t.AgentIDs { + if !baseSet[agentID] { + added = append(added, agentID) + } + } + } else { + if _, err := tx.Exec(`DELETE FROM probe_ping_node WHERE task_id=?1`, id); err != nil { + return 0, err + } + added = t.AgentIDs + } + for _, agentID := range added { + if _, err := tx.Exec(`INSERT OR IGNORE INTO probe_ping_node (task_id, agent_id) VALUES (?1, ?2)`, id, agentID); err != nil { + return 0, err + } + } + + if err := refuseCrowded(tx, nil); err != nil { + return 0, err + } + var joining int64 + if err := tx.QueryRow(`SELECT COUNT(*) FROM probe_ping_task WHERE auto_join`).Scan(&joining); err != nil { + return 0, err + } + if joining > MaxProbesPerNode { + return 0, &PingAutoJoinError{Count: joining} + } + if err := tx.Commit(); err != nil { + return 0, err + } + return id, nil +} + +// refuseCrowded fails the transaction when some agent is assigned +// more tasks than an agent will run — queried from the table after +// the rows are in, inside the write's own transaction (hub +// refuse_crowded). onlyAgent, when set, scopes the check to that +// agent: another agent already over the limit is not this edit's +// problem to refuse. +func refuseCrowded(tx *sql.Tx, onlyAgent *int64) error { + var scope any + if onlyAgent != nil { + scope = *onlyAgent + } + var agentID sql.NullInt64 + err := tx.QueryRow(`SELECT agent_id FROM probe_ping_node + WHERE ?1 IS NULL OR agent_id = ?1 + GROUP BY agent_id HAVING COUNT(*) > ?2 LIMIT 1`, scope, MaxProbesPerNode).Scan(&agentID) + if err == sql.ErrNoRows { + return nil + } + if err != nil { + return err + } + return &PingCrowdedError{AgentID: agentID.Int64} +} + +// SetNodePingTasks edits one agent's assignments from the agent's +// side (hub set_node_ping_tasks): only the difference from base is +// applied, inside one transaction. The caller checks the agent exists. +func (s *Store) SetNodePingTasks(agentID int64, tasks, base []int64) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + taskSet := int64Set(tasks) + baseSet := int64Set(base) + for _, taskID := range base { + if !taskSet[taskID] { + if _, err := tx.Exec(`DELETE FROM probe_ping_node WHERE task_id=?1 AND agent_id=?2`, taskID, agentID); err != nil { + return err + } + } + } + for _, taskID := range tasks { + if !baseSet[taskID] { + // A task deleted since the editor opened fails its foreign + // key on the hub; probe.db has no cross-database key, so + // existence is tested in the statement instead — naming + // the probe is all the panel needs. + res, err := tx.Exec(`INSERT OR IGNORE INTO probe_ping_node (task_id, agent_id) + SELECT ?1, ?2 WHERE EXISTS (SELECT 1 FROM probe_ping_task WHERE id = ?1)`, taskID, agentID) + if err != nil { + return err + } + if n, err := res.RowsAffected(); err != nil { + return err + } else if n == 0 { + return errors.New("有监控已被删除，刷新后重试") + } + } + } + if err := refuseCrowded(tx, &agentID); err != nil { + return err + } + return tx.Commit() +} + +// DeletePingTask removes a task and everything filed under it +// (hub delete_ping_task): the record table is cleared explicitly, +// the node's assignments and the hourly tier go with it. +func (s *Store) DeletePingTask(id int64) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + for _, stmt := range []string{ + `DELETE FROM probe_ping_record WHERE task_id = ?`, + `DELETE FROM probe_ping_hour WHERE task_id = ?`, + `DELETE FROM probe_ping_node WHERE task_id = ?`, + `DELETE FROM probe_ping_task WHERE id = ?`, + } { + if _, err := tx.Exec(stmt, id); err != nil { + return err + } + } + return tx.Commit() +} + +// ReorderPingTasks rewrites the panel order (hub reorder): the id +// list must name every task exactly once, checked against the table +// inside the transaction, so a list edited elsewhere cannot sort +// against a stale view. +func (s *Store) ReorderPingTasks(ids []int64) error { + seen := make(map[int64]bool, len(ids)) + for _, id := range ids { + if seen[id] { + return errors.New("排序里有重复的条目") + } + seen[id] = true + } + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + var count int64 + if err := tx.QueryRow(`SELECT COUNT(*) FROM probe_ping_task`).Scan(&count); err != nil { + return err + } + if count != int64(len(ids)) { + return errors.New("列表已在别处改动，刷新后再排序") + } + for i, id := range ids { + res, err := tx.Exec(`UPDATE probe_ping_task SET sort=?2 WHERE id=?1`, id, int64(i)) + if err != nil { + return err + } + if n, err := res.RowsAffected(); err != nil { + return err + } else if n != 1 { + return errors.New("列表已在别处改动，刷新后再排序") + } + } + return tx.Commit() +} + +// PingTasksFor returns the task list pushed to one agent: id order, +// because the agent's own cap keeps the first MaxProbesPerNode and an +// unordered list at that boundary could yield a different subset on +// each push (hub ping_tasks_for). +func (s *Store) PingTasksFor(agentID int64) ([]PingTaskRef, error) { + rows, err := s.db.Query(`SELECT t.id, t.target, t.interval FROM probe_ping_task t + JOIN probe_ping_node n ON n.task_id = t.id WHERE n.agent_id = ?1 ORDER BY t.id`, agentID) + if err != nil { + return nil, err + } + defer rows.Close() + out := []PingTaskRef{} + for rows.Next() { + var r PingTaskRef + if err := rows.Scan(&r.ID, &r.Target, &r.Interval); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +// PingTaskNames returns the chart labels of one agent's probes +// (id → name), restricted to the tasks assigned to it (hub +// ping_task_names): names are operator text and the rest of the +// table belongs to agents this caller may not be able to see. +func (s *Store) PingTaskNames(agentID int64) (map[string]string, error) { + rows, err := s.db.Query(`SELECT id, name FROM probe_ping_task + WHERE id IN (SELECT task_id FROM probe_ping_node WHERE agent_id = ?1)`, agentID) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[string]string{} + for rows.Next() { + var id int64 + var name string + if err := rows.Scan(&id, &name); err != nil { + return nil, err + } + out[fmt.Sprint(id)] = name + } + return out, rows.Err() +} + +// InsertPings files one agent's reported results at the arrival +// stamp ts, one transaction for the batch (hub insert_pings). A +// result only lands under a task the agent is assigned: the test is +// inside the statement, the only place it is atomic with the write. +// Results for anything else — a probe deleted while its results were +// in flight, a task id the reporter invented — are dropped, not +// errors, since the agent can do nothing useful with the distinction. +func (s *Store) InsertPings(agentID int64, results []PingResult, ts int64) error { + if len(results) == 0 { + return nil + } + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + stmt, err := tx.Prepare(`INSERT OR REPLACE INTO probe_ping_record (agent_id, ts, task_id, latency) + SELECT ?1, ?2, ?3, ?4 + WHERE EXISTS (SELECT 1 FROM probe_ping_node WHERE task_id = ?3 AND agent_id = ?1)`) + if err != nil { + return err + } + defer stmt.Close() + for _, r := range results { + if _, err := stmt.Exec(agentID, ts, r.TaskID, r.LatencyMs); err != nil { + return err + } + } + return tx.Commit() +} + +// PingRow is one thinned latency point of the contract chart (hub +// ping_records rows): the bucket's median answer, present keys only +// — band appears when the bucket actually varied, loss when it lost +// anything (absence reads as 0, the themes' documented semantics). +type PingRow struct { + TaskID int64 `json:"task_id"` + Ts int64 `json:"ts"` + Latency *int64 `json:"latency"` + Band *[2]int64 `json:"band,omitempty"` + Loss *int64 `json:"loss,omitempty"` +} + +type pingWeightedMedian struct { + value int64 + weight int64 +} + +// pingReadTally is one task's samples within one bucket of the chart +// read (hub Tally): hourly rows enter as their hour's median weighted +// by their answer count, single results as themselves. +type pingReadTally struct { + medians []pingWeightedMedian + lost int64 + lo, hi *int64 +} + +func (t *pingReadTally) add(answered, lost int64, median, lo, hi *int64) { + if median != nil { + t.medians = append(t.medians, pingWeightedMedian{value: *median, weight: answered}) + } + t.lost += lost + if lo != nil && (t.lo == nil || *lo < *t.lo) { + v := *lo + t.lo = &v + } + if hi != nil && (t.hi == nil || *hi > *t.hi) { + v := *hi + t.hi = &v + } +} + +func (t *pingReadTally) answered() int64 { + var total int64 + for _, m := range t.medians { + total += m.weight + } + return total +} + +// median is the weighted median of the answers: the middle answer's +// value, or the average of the middle two (hub Tally::median). +func (t *pingReadTally) median() *int64 { + if len(t.medians) == 0 { + return nil + } + sorted := append([]pingWeightedMedian(nil), t.medians...) + sort.Slice(sorted, func(i, j int) bool { + if sorted[i].value != sorted[j].value { + return sorted[i].value < sorted[j].value + } + return sorted[i].weight < sorted[j].weight + }) + total := t.answered() + at := func(rank int64) (int64, bool) { + var seen int64 + for _, m := range sorted { + seen += m.weight + if seen >= rank { + return m.value, true + } + } + return 0, false + } + lo, ok1 := at((total + 1) / 2) + hi, ok2 := at(total/2 + 1) + if !ok1 || !ok2 { + return nil + } + v := (lo + hi) / 2 + return &v +} + +// PingHistory answers one agent's latency window (hub ping_records): +// one row per task per step-sized bucket holding the bucket median, +// its range and its loss, plus the per-task loss over the whole +// window (unrounded — the caller renders it). Only tasks still +// assigned to the agent contribute. Hourly windows merge the folded +// hourly tier (up to the watermark) with the minute rows after it, +// one read snapshot throughout. +func (s *Store) PingHistory(agentID int64, sp Span) ([]PingRow, map[string]float64, error) { + tx, err := s.db.Begin() + if err != nil { + return nil, nil, err + } + defer tx.Rollback() + + out := []PingRow{} + totals := make(map[int64][2]int64) // task → [lost, samples] + type openTally struct { + task int64 + tally *pingReadTally + } + var open []openTally + var bucket int64 + flush := func() { + for _, o := range open { + row := PingRow{TaskID: o.task, Ts: bucket * sp.Step, Latency: o.tally.median()} + if o.tally.lo != nil && o.tally.hi != nil && *o.tally.hi > *o.tally.lo { + row.Band = &[2]int64{*o.tally.lo, *o.tally.hi} + } + if o.tally.lost > 0 { + total := o.tally.answered() + o.tally.lost + loss := (100*o.tally.lost + total - 1) / total + row.Loss = &loss + } + out = append(out, row) + } + open = open[:0] + } + feed := func(b, task int64, answered, lost int64, median, lo, hi *int64) { + if b != bucket { + flush() + bucket = b + } + tot := totals[task] + tot[0] += lost + tot[1] += answered + lost + totals[task] = tot + for i := range open { + if open[i].task == task { + open[i].tally.add(answered, lost, median, lo, hi) + return + } + } + tally := &pingReadTally{} + tally.add(answered, lost, median, lo, hi) + open = append(open, openTally{task, tally}) + } + feedRow := func(b, task, latency int64) { + if latency >= 0 { + feed(b, task, 1, 0, &latency, &latency, &latency) + } else { + feed(b, task, 0, 1, nil, nil, nil) + } + } + + minutesFrom := sp.Since + if sp.Hourly { + rolled := int64(-1) << 62 // hub: unwrap_or(i64::MIN) + var raw sql.NullString + if qerr := tx.QueryRow(`SELECT value FROM probe_setting WHERE key = ?`, rolledKey).Scan(&raw); qerr != nil && qerr != sql.ErrNoRows { + return nil, nil, qerr + } else if raw.Valid { + if n, perr := parseInt64(raw.String); perr == nil { + rolled = n + } + } + var newest sql.NullInt64 + if err := tx.QueryRow(`SELECT MAX(ts) FROM probe_ping_record WHERE agent_id=?1`, agentID).Scan(&newest); err != nil { + return nil, nil, err + } + if newest.Valid && newest.Int64-DetailDays*86400 > minutesFrom { + minutesFrom = newest.Int64 - DetailDays*86400 + } + if rolled > minutesFrom { + minutesFrom = rolled + } + hours, err := tx.Query(`SELECT ts/?3, task_id, answered, lost, latency, lo, hi FROM probe_ping_hour + WHERE agent_id=?1 AND ts>=?2 AND ts=?2 + AND task_id IN (SELECT task_id FROM probe_ping_node WHERE agent_id=?1) + ORDER BY ts`, agentID, minutesFrom, sp.Step) + if err != nil { + return nil, nil, err + } + for rows.Next() { + var b, task, latency int64 + if err := rows.Scan(&b, &task, &latency); err != nil { + rows.Close() + return nil, nil, err + } + feedRow(b, task, latency) + } + if err := rows.Err(); err != nil { + rows.Close() + return nil, nil, err + } + rows.Close() + flush() + + // Panel order decides each task's series order (hub: themes take + // colours and legend from first appearance across the window, + // which per-bucket feeding would leave to chance). + rank := make(map[int64]int) + rankRows, err := tx.Query(`SELECT id FROM probe_ping_task ORDER BY sort, id`) + if err != nil { + return nil, nil, err + } + for rankRows.Next() { + var id int64 + if err := rankRows.Scan(&id); err != nil { + rankRows.Close() + return nil, nil, err + } + rank[id] = len(rank) + } + rankRows.Close() + if err := rankRows.Err(); err != nil { + return nil, nil, err + } + const unranked = int(^uint(0) >> 1) + sort.SliceStable(out, func(i, j int) bool { + ri, ok := rank[out[i].TaskID] + if !ok { + ri = unranked + } + rj, ok := rank[out[j].TaskID] + if !ok { + rj = unranked + } + return ri < rj + }) + + loss := map[string]float64{} + for task, tot := range totals { + if tot[0] > 0 && tot[1] > 0 { + loss[fmt.Sprint(task)] = 100.0 * float64(tot[0]) / float64(tot[1]) + } + } + return out, loss, nil +} + +func int64Set(ids []int64) map[int64]bool { + out := make(map[int64]bool, len(ids)) + for _, id := range ids { + out[id] = true + } + return out +} + +func nullIntPtr(v sql.NullInt64) *int64 { + if !v.Valid { + return nil + } + return &v.Int64 +} + +func parseInt64(s string) (int64, error) { + return strconv.ParseInt(s, 10, 64) +} diff --git a/internal/web/service/probe/probe.go b/internal/web/service/probe/probe.go index 5522ae1..4aa17be 100644 --- a/internal/web/service/probe/probe.go +++ b/internal/web/service/probe/probe.go @@ -136,3 +136,103 @@ func DeleteAgentData(agentID int64) { logger.Warningf("probe: delete agent %d data: %v", agentID, err) } } + +// ---- B4: probe tasks, results and the latency read ---- + +// PingTasks lists every probe task with its assignments. With no +// store the list is empty. +func PingTasks() ([]PingTask, error) { + st := Default() + if st == nil { + return []PingTask{}, nil + } + return st.PingTasks() +} + +// SavePingTask creates or replaces one probe task (see Store). +func SavePingTask(t PingTask, base *[]int64) (int64, error) { + st := Default() + if st == nil { + return 0, nil + } + return st.SavePingTask(t, base) +} + +// DeletePingTask removes one probe task and its filed history. +func DeletePingTask(id int64) error { + st := Default() + if st == nil { + return nil + } + return st.DeletePingTask(id) +} + +// SetNodePingTasks applies one agent's assignment difference. +func SetNodePingTasks(agentID int64, tasks, base []int64) error { + st := Default() + if st == nil { + return nil + } + return st.SetNodePingTasks(agentID, tasks, base) +} + +// ReorderPingTasks rewrites the panel order of the task list. +func ReorderPingTasks(ids []int64) error { + st := Default() + if st == nil { + return nil + } + return st.ReorderPingTasks(ids) +} + +// PingTasksFor reads the task list pushed to one agent on its config +// poll. Best-effort like the rest of the ingest side: a store failure +// reads as "no tasks" and is logged, so a broken probe.db never +// breaks config delivery — the agent keeps its current list until a +// poll succeeds. +func PingTasksFor(agentID int64) []PingTaskRef { + st := Default() + if st == nil { + return []PingTaskRef{} + } + refs, err := st.PingTasksFor(agentID) + if err != nil { + logger.Warningf("probe: ping tasks for agent %d: %v", agentID, err) + return []PingTaskRef{} + } + return refs +} + +// PingTaskNames reads one agent's probe chart labels. +func PingTaskNames(agentID int64) (map[string]string, error) { + st := Default() + if st == nil { + return map[string]string{}, nil + } + return st.PingTaskNames(agentID) +} + +// InsertPings files one agent's probe results at the arrival stamp. +// Best-effort (heartbeat path): failures are logged and swallowed. +func InsertPings(agentID int64, results []PingResult) { + if len(results) == 0 { + return + } + st := Default() + if st == nil { + return + } + if err := st.InsertPings(agentID, results, time.Now().Unix()); err != nil { + logger.Warningf("probe: insert pings agent %d: %v", agentID, err) + } +} + +// PingHistory is the B4 read port for one agent's latency window. +// With no store the window is simply empty. +func PingHistory(agentID int64, sp Span) ([]PingRow, map[string]float64, error) { + st := Default() + if st == nil { + return []PingRow{}, map[string]float64{}, nil + } + return st.PingHistory(agentID, sp) +} diff --git a/internal/web/service/probe/store.go b/internal/web/service/probe/store.go index 1e44acb..79a4b8b 100644 --- a/internal/web/service/probe/store.go +++ b/internal/web/service/probe/store.go @@ -150,6 +150,29 @@ CREATE TABLE IF NOT EXISTS probe_ping_hour ( hi INTEGER, PRIMARY KEY (agent_id, ts, task_id) ) WITHOUT ROWID; + +-- Probe tasks and their per-agent assignment (hub "ping_task" / +-- "ping_node"). B4 adds both. A task is one operator-maintained +-- target — there is no preset list; region is the operator's own +-- grouping label (empty = ungrouped). Ids come from a high-water +-- mark in probe_setting so a deleted task's id is never reissued +-- (hub next_id discipline): in-flight results and an editor opened +-- before the delete must not land on a different task. +CREATE TABLE IF NOT EXISTS probe_ping_task ( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL, + target TEXT NOT NULL, + interval INTEGER NOT NULL DEFAULT 60, + auto_join INTEGER NOT NULL DEFAULT 0, + sort INTEGER NOT NULL DEFAULT 0, + region TEXT NOT NULL DEFAULT '' +); + +CREATE TABLE IF NOT EXISTS probe_ping_node ( + task_id INTEGER NOT NULL, + agent_id INTEGER NOT NULL, + PRIMARY KEY (task_id, agent_id) +); ` // Store is one open probe.db. The zero value is not usable; call Open. diff --git a/internal/web/service/probe/traffic.go b/internal/web/service/probe/traffic.go index 3beee41..bed6172 100644 --- a/internal/web/service/probe/traffic.go +++ b/internal/web/service/probe/traffic.go @@ -87,12 +87,23 @@ func dateOnly(t time.Time) time.Time { } // ensureRows lazily creates the agent's meta + traffic rows so booking -// and reading never special-case a first report. +// and reading never special-case a first report. A meta row created +// here is also the moment the agent joins the probe system, so the +// auto-join probe tasks are assigned then — and only then, matching +// the hub, where auto_join applies to nodes created after the task +// was flagged, never retroactively to existing ones (B4). func (s *Store) ensureRows(agentID int64) error { - if _, err := s.db.Exec(`INSERT OR IGNORE INTO probe_meta (agent_id) VALUES (?)`, agentID); err != nil { + res, err := s.db.Exec(`INSERT OR IGNORE INTO probe_meta (agent_id) VALUES (?)`, agentID) + if err != nil { return err } - _, err := s.db.Exec(`INSERT OR IGNORE INTO probe_traffic (agent_id) VALUES (?)`, agentID) + if n, err := res.RowsAffected(); err == nil && n > 0 { + if _, err := s.db.Exec(`INSERT OR IGNORE INTO probe_ping_node (task_id, agent_id) + SELECT id, ? FROM probe_ping_task WHERE auto_join`, agentID); err != nil { + return err + } + } + _, err = s.db.Exec(`INSERT OR IGNORE INTO probe_traffic (agent_id) VALUES (?)`, agentID) return err }