From 0d1607c35887f3833d228251fbad4b6e1847183f Mon Sep 17 00:00:00 2001 From: Muse Date: Thu, 22 Oct 2026 22:09:11 +0000 Subject: [PATCH] B4 tests: probe store, agent prober, controller + wire MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Baseline note: TestMetricHistoryMinuteFold fails identically on the pristine B3 tip in the same wall-clock window (minute-parity sensitive seed), passes in others — pre-existing, untouched path. --- internal/agent/ping_test.go | 169 +++++++ internal/web/controller/probe_public_test.go | 1 + .../web/controller/probe_tasks_agent_test.go | 138 ++++++ internal/web/controller/probe_tasks_test.go | 281 ++++++++++++ internal/web/service/probe/ping_test.go | 411 ++++++++++++++++++ 5 files changed, 1000 insertions(+) create mode 100644 internal/agent/ping_test.go create mode 100644 internal/web/controller/probe_tasks_agent_test.go create mode 100644 internal/web/controller/probe_tasks_test.go create mode 100644 internal/web/service/probe/ping_test.go diff --git a/internal/agent/ping_test.go b/internal/agent/ping_test.go new file mode 100644 index 0000000..e15ffac --- /dev/null +++ b/internal/agent/ping_test.go @@ -0,0 +1,169 @@ +package agent + +import ( + "encoding/json" + "net" + "testing" + "time" +) + +// probeTCP against a live listener yields a real non-negative +// latency; a closed port and a malformed target both file -1; 0 is +// never a latency (contract: failures are negative or absent). +func TestProbeTCPOutcomes(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + conn.Close() + } + }() + + ms, sampled := probeTCP(ln.Addr().String(), nil) + if !sampled || ms < 0 { + t.Fatalf("live target: ms=%d sampled=%v", ms, sampled) + } + ms, sampled = probeTCP("127.0.0.1:1", nil) + if !sampled || ms != -1 { + t.Fatalf("closed port: ms=%d sampled=%v", ms, sampled) + } + ms, sampled = probeTCP("not-a-target", nil) + if !sampled || ms != -1 { + t.Fatalf("malformed: ms=%d sampled=%v", ms, sampled) + } + ms, sampled = probeTCP("256.256.256.256:80", nil) + if !sampled || ms != -1 { + t.Fatalf("unresolvable: ms=%d sampled=%v", ms, sampled) + } +} + +// applyTasks: an unchanged task keeps its loop; a changed target or +// interval restarts it; a removed task is cancelled; the set is +// truncated at the cap with intervals clamped into range. +func TestProbeRunnerApplyTasks(t *testing.T) { + r := newProbeRunner() + defer r.stop() + + r.applyTasks([]ProbeTask{{ID: 1, Target: "127.0.0.1:1", Interval: 60}}) + if len(r.tasks) != 1 { + t.Fatalf("tasks %v", r.tasks) + } + first := r.tasks[1] + + // Unchanged: same loop object survives. + r.applyTasks([]ProbeTask{{ID: 1, Target: "127.0.0.1:1", Interval: 60}}) + if r.tasks[1] != first { + t.Fatal("unchanged task restarted") + } + // Changed interval: replaced, clamped values are what runs. + r.applyTasks([]ProbeTask{{ID: 1, Target: "127.0.0.1:1", Interval: 120}}) + if r.tasks[1] == first || r.tasks[1].interval != 120 { + t.Fatal("changed task not restarted") + } + // Removed: cancelled. + r.applyTasks(nil) + if len(r.tasks) != 0 { + t.Fatalf("tasks after clear %v", r.tasks) + } + + // Clamp + cap. + many := make([]ProbeTask, 0, 70) + for i := 1; i <= 70; i++ { + many = append(many, ProbeTask{ID: int64(i), Target: "127.0.0.1:1", Interval: 1}) + } + r.applyTasks(many) + if len(r.tasks) != maxProbeTasksRun { + t.Fatalf("cap: %d running", len(r.tasks)) + } + if r.tasks[1].interval != minProbeIntervalSec { + t.Fatalf("clamp: interval %d", r.tasks[1].interval) + } +} + +// A running loop files samples into the buffer; drain takes them and +// requeue restores order ahead of newer samples. +func TestProbeRunnerResultFlow(t *testing.T) { + r := newProbeRunner() + defer r.stop() + r.applyTasks([]ProbeTask{{ID: 9, Target: "127.0.0.1:1", Interval: 3600}}) + // The loop probes immediately on start. + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if got := r.drainResults(); len(got) > 0 { + if got[0].TaskID != 9 || got[0].LatencyMs != -1 { + t.Fatalf("sample %+v", got[0]) + } + goto drained + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("no immediate sample from a fresh task") +drained: + + r.addResult(ProbeResult{TaskID: 1, LatencyMs: 5}) + batch := r.drainResults() + r.addResult(ProbeResult{TaskID: 1, LatencyMs: 7}) + r.requeueResults(batch) + got := r.drainResults() + if len(got) != 2 || got[0].LatencyMs != 5 || got[1].LatencyMs != 7 { + t.Fatalf("requeue order %+v", got) + } + if again := r.drainResults(); again != nil { + t.Fatalf("second drain %+v", again) + } +} + +// Buffer overflow drops the oldest samples, keeping the newest cap. +func TestProbeRunnerBufferCap(t *testing.T) { + r := newProbeRunner() + defer r.stop() + for i := 0; i < probeResultCap+10; i++ { + r.addResult(ProbeResult{TaskID: 1, LatencyMs: int64(i)}) + } + got := r.drainResults() + if len(got) != probeResultCap { + t.Fatalf("buffer %d", len(got)) + } + if got[0].LatencyMs != 10 || got[len(got)-1].LatencyMs != int64(probeResultCap+9) { + t.Fatalf("kept range %d..%d", got[0].LatencyMs, got[len(got)-1].LatencyMs) + } + if r.dropped != 10 { + t.Fatalf("dropped %d", r.dropped) + } +} + +// The wire shapes: a result marshals as taskId/latencyMs, and the +// config decode distinguishes an absent probe_tasks (old master, +// keep current set) from an empty one (clear all). +func TestProbeWireShapes(t *testing.T) { + body, err := json.Marshal(ProbeResult{TaskID: 3, LatencyMs: -1}) + if err != nil || string(body) != `{"taskId":3,"latencyMs":-1}` { + t.Fatalf("result wire: %s %v", body, err) + } + decode := func(s string) *[]ProbeTask { + var result struct { + ProbeTasks *[]ProbeTask `json:"probe_tasks"` + } + if err := json.Unmarshal([]byte(s), &result); err != nil { + t.Fatal(err) + } + return result.ProbeTasks + } + if got := decode(`{"config_hash":"x"}`); got != nil { + t.Fatalf("absent field decoded as %v", got) + } + if got := decode(`{"probe_tasks":[]}`); got == nil || len(*got) != 0 { + t.Fatalf("empty list decoded as %v", got) + } + got := decode(`{"probe_tasks":[{"id":2,"target":"1.1.1.1:443","interval":30}]}`) + if got == nil || len(*got) != 1 || (*got)[0].ID != 2 || (*got)[0].Target != "1.1.1.1:443" || (*got)[0].Interval != 30 { + t.Fatalf("task decode %+v", got) + } +} diff --git a/internal/web/controller/probe_public_test.go b/internal/web/controller/probe_public_test.go index a7eeab4..70af6e2 100644 --- a/internal/web/controller/probe_public_test.go +++ b/internal/web/controller/probe_public_test.go @@ -78,6 +78,7 @@ func newProbeTestRig(t *testing.T) *probeTestRig { agentAdmin := engine.Group("/panel/api/agent") agentAdmin.Use(a.checkAPIAuth) NewProbeMetaController(agentAdmin) + NewProbeTasksController(agentAdmin) engine.NoRoute(func(c *gin.Context) { if pc.HandleFallback(c) { return diff --git a/internal/web/controller/probe_tasks_agent_test.go b/internal/web/controller/probe_tasks_agent_test.go new file mode 100644 index 0000000..e55dbdd --- /dev/null +++ b/internal/web/controller/probe_tasks_agent_test.go @@ -0,0 +1,138 @@ +package controller + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + + "github.com/gin-gonic/gin" + + "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" +) + +// Item 7 B4: the agent-facing wire — probe_tasks rides the config +// response (always present, empty list clears), and probeResults on +// the heartbeat are filed stamped at arrival, with junk rows +// (unknown/absent latency, non-positive task ids) skipped and +// unassigned tasks dropped by the store's assignment gate. +func TestAgentProbeTasksWire(t *testing.T) { + gin.SetMode(gin.TestMode) + dbDir := t.TempDir() + t.Setenv("XUI_DB_FOLDER", dbDir) + if err := database.InitDB(filepath.Join(dbDir, "x-ui.db")); err != nil { + t.Fatalf("InitDB: %v", err) + } + t.Cleanup(func() { _ = database.CloseDB() }) + db := database.GetDB() + + node := &model.Node{Name: "b4-wire"} + if err := db.Create(node).Error; err != nil { + t.Fatal(err) + } + const token = "b4-wire-token" + agent := &model.AgentNode{ + TokenHash: hashAgentToken(token), TokenPrefix: "b4-", + Name: "b4-wire-agent", NodeID: &node.Id, + } + if err := db.Create(agent).Error; err != nil { + t.Fatal(err) + } + agentID := int64(agent.Id) + probe.DeleteAgentData(agentID) + + taskID, err := probe.SavePingTask(probe.PingTask{ + Name: "wire", Target: "1.1.1.1:443", Interval: 30, + AgentIDs: []int64{agentID}, + }, nil) + if err != nil { + t.Fatal(err) + } + + engine := gin.New() + NewAgentController(engine.Group("/panel/api/agent")) + do := func(method, path, body string) (int, []byte) { + var rd *bytes.Reader + if body != "" { + rd = bytes.NewReader([]byte(body)) + } else { + rd = bytes.NewReader(nil) + } + req := httptest.NewRequest(method, path, rd) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + engine.ServeHTTP(rec, req) + return rec.Code, rec.Body.Bytes() + } + + const cfgPath = "/panel/api/agent/config" + const hbPath = "/panel/api/agent/heartbeat" + + // Config carries the task set. + code, body := do(http.MethodGet, cfgPath, "") + if code != http.StatusOK { + t.Fatalf("config: %d %s", code, body) + } + var cfg struct { + ProbeTasks []probe.PingTaskRef `json:"probe_tasks"` + } + if err := json.Unmarshal(body, &cfg); err != nil { + t.Fatal(err) + } + if len(cfg.ProbeTasks) != 1 || cfg.ProbeTasks[0].ID != taskID || + cfg.ProbeTasks[0].Target != "1.1.1.1:443" || cfg.ProbeTasks[0].Interval != 30 { + t.Fatalf("config probe_tasks: %s", body) + } + + // Heartbeat with a mixed batch: only the valid, assigned sample + // lands. (All rows share the arrival stamp, so the two filings of + // the same task collapse — assert on the surviving value.) + hb := `{"probeResults":[` + + `{"taskId":` + itoa(taskID) + `,"latencyMs":42},` + + `{"taskId":999999,"latencyMs":5},` + + `{"taskId":-3,"latencyMs":5},` + + `{"taskId":` + itoa(taskID) + `}` + + `]}` + code, body = do(http.MethodPost, hbPath, hb) + if code != http.StatusOK { + t.Fatalf("heartbeat: %d %s", code, body) + } + st := probe.Default() + var count int64 + var latency int64 + if err := st.DB().QueryRow(`SELECT COUNT(*), COALESCE(MAX(latency), -999) FROM probe_ping_record WHERE agent_id = ?`, agentID).Scan(&count, &latency); err != nil { + t.Fatal(err) + } + if count != 1 || latency != 42 { + t.Fatalf("filed rows %d latency %d", count, latency) + } + + // After the task is unassigned, config reports an empty (present) + // list and further results are dropped by the assignment gate. + if err := probe.SetNodePingTasks(agentID, []int64{}, []int64{taskID}); err != nil { + t.Fatal(err) + } + code, body = do(http.MethodGet, cfgPath, "") + if code != http.StatusOK { + t.Fatalf("config 2: %d", code) + } + if !bytes.Contains(body, []byte(`"probe_tasks":[]`)) { + t.Fatalf("empty probe_tasks not present: %s", body) + } + code, _ = do(http.MethodPost, hbPath, `{"probeResults":[{"taskId":`+itoa(taskID)+`,"latencyMs":77}]}`) + if code != http.StatusOK { + t.Fatalf("heartbeat 2: %d", code) + } + if err := st.DB().QueryRow(`SELECT COUNT(*) FROM probe_ping_record WHERE agent_id = ? AND latency = 77`, agentID).Scan(&count); err != nil { + t.Fatal(err) + } + if count != 0 { + t.Fatal("unassigned result was filed") + } + probe.DeletePingTask(taskID) +} diff --git a/internal/web/controller/probe_tasks_test.go b/internal/web/controller/probe_tasks_test.go new file mode 100644 index 0000000..57dd444 --- /dev/null +++ b/internal/web/controller/probe_tasks_test.go @@ -0,0 +1,281 @@ +package controller + +import ( + "encoding/json" + "net/http" + "strings" + "testing" + "time" + + "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" +) + +// Item 7 B4 controller tests: probe task management endpoints (hub +// validation chain, node-id translation, base-diff, reorder), the +// metrics series=ping read going live, and heartbeat ingestion of +// probeResults + config delivery of probe_tasks. + +type envelope struct { + Success bool `json:"success"` + Msg string `json:"msg"` + Obj json.RawMessage `json:"obj"` +} + +func decodeEnvelope(t *testing.T, body []byte) envelope { + t.Helper() + var env envelope + if err := json.Unmarshal(body, &env); err != nil { + t.Fatalf("envelope: %v (%s)", err, body) + } + return env +} + +func cleanPingTasks(t *testing.T) { + t.Helper() + tasks, err := probe.PingTasks() + if err != nil { + t.Fatal(err) + } + for _, task := range tasks { + if err := probe.DeletePingTask(task.ID); err != nil { + t.Fatal(err) + } + } +} + +const tasksPath = "/panel/api/agent/probe-tasks" + +func TestProbeTasksAuthAndValidation(t *testing.T) { + rig := newProbeTestRig(t) + cleanPingTasks(t) + + // Anonymous callers are refused by the group auth. + rec := rig.do(http.MethodPut, tasksPath, []byte(`{"name":"x","target":"1.1.1.1:443","interval":60}`), false) + if rec.Code == http.StatusOK && decodeEnvelope(t, rec.Body.Bytes()).Success { + t.Fatal("anonymous save succeeded") + } + + cases := []struct { + name string + body string + want string + }{ + {"empty name", `{"name":" ","target":"1.1.1.1:443","interval":60}`, "请填写名称和目标"}, + {"empty target", `{"name":"x","target":"","interval":60}`, "请填写名称和目标"}, + {"long name", `{"name":"` + strings.Repeat("名", 65) + `","target":"1.1.1.1:443","interval":60}`, "名称最多 64 个字"}, + {"no port", `{"name":"x","target":"1.1.1.1","interval":60}`, "目标要写成"}, + {"empty port", `{"name":"x","target":"1.1.1.1:","interval":60}`, "目标要写成"}, + {"port range", `{"name":"x","target":"1.1.1.1:65536","interval":60}`, "目标要写成"}, + {"bare v6", `{"name":"x","target":"2606:4700:4700::1111:443","interval":60}`, "目标要写成"}, + {"interval low", `{"name":"x","target":"1.1.1.1:443","interval":3}`, "间隔要在 5 到 3600 秒之间"}, + {"interval high", `{"name":"x","target":"1.1.1.1:443","interval":3601}`, "间隔要在 5 到 3600 秒之间"}, + {"long region", `{"name":"x","target":"1.1.1.1:443","interval":60,"region":"` + strings.Repeat("区", 14) + `"}`, "地区最多 13 个字"}, + } + for _, tc := range cases { + rec := rig.do(http.MethodPut, tasksPath, []byte(tc.body), true) + env := decodeEnvelope(t, rec.Body.Bytes()) + if env.Success { + t.Errorf("%s: save succeeded, want refusal", tc.name) + continue + } + if !strings.Contains(env.Msg, tc.want) { + t.Errorf("%s: msg %q, want fragment %q", tc.name, env.Msg, tc.want) + } + } + if tasks, _ := probe.PingTasks(); len(tasks) != 0 { + t.Fatalf("refused saves leaked %d tasks", len(tasks)) + } + + // Valid forms: bracketed v6, boundary interval, 13-rune region. + for _, body := range []string{ + `{"name":"v6","target":"[2606:4700:4700::1111]:443","interval":5}`, + `{"name":"edge","target":"example.com:65535","interval":3600,"region":"` + strings.Repeat("区", 13) + `"}`, + } { + rec := rig.do(http.MethodPut, tasksPath, []byte(body), true) + if env := decodeEnvelope(t, rec.Body.Bytes()); !env.Success { + t.Fatalf("valid save refused: %s (%s)", env.Msg, body) + } + } + cleanPingTasks(t) +} + +func TestProbeTasksNodeMappingAndLifecycle(t *testing.T) { + rig := newProbeTestRig(t) + cleanPingTasks(t) + agentID, nodeID := rig.seedAgent(t, "b4node", true) + + // A node with no agent bound cannot take assignments. + db := database.GetDB() + lone := &model.Node{Name: "b4-lone"} + if err := db.Create(lone).Error; err != nil { + t.Fatal(err) + } + body := `{"name":"t1","target":"1.1.1.1:443","interval":60,"nodeIds":[` + itoa(int64(lone.Id)) + `]}` + rec := rig.do(http.MethodPut, tasksPath, []byte(body), true) + if env := decodeEnvelope(t, rec.Body.Bytes()); env.Success || !strings.Contains(env.Msg, "未绑定 agent") { + t.Fatalf("unbound node save: %+v", env) + } + + // Create assigned to the bound node. + body = `{"name":"t1","target":"1.1.1.1:443","interval":60,"region":"亚洲","nodeIds":[` + itoa(int64(nodeID)) + `]}` + rec = rig.do(http.MethodPut, tasksPath, []byte(body), true) + env := decodeEnvelope(t, rec.Body.Bytes()) + if !env.Success { + t.Fatalf("create: %s", env.Msg) + } + var created struct { + ID int64 `json:"id"` + } + if err := json.Unmarshal(env.Obj, &created); err != nil || created.ID == 0 { + t.Fatalf("create obj: %s", env.Obj) + } + refs := probe.PingTasksFor(agentID) + if len(refs) != 1 || refs[0].ID != created.ID { + t.Fatalf("agent view: %v", refs) + } + + // The list speaks node ids. + rec = rig.do(http.MethodGet, tasksPath, nil, true) + env = decodeEnvelope(t, rec.Body.Bytes()) + var list []probeTaskForm + if err := json.Unmarshal(env.Obj, &list); err != nil || len(list) != 1 { + t.Fatalf("list: %s", env.Obj) + } + if len(list[0].NodeIDs) != 1 || list[0].NodeIDs[0] != int64(nodeID) || list[0].Region != "亚洲" { + t.Fatalf("list form %+v", list[0]) + } + + // by-node edit: clear the assignment with a base diff. + byNode := `{"tasks":[],"base":[` + itoa64(created.ID) + `]}` + rec = rig.do(http.MethodPut, tasksPath+"/by-node/"+itoa(int64(nodeID)), []byte(byNode), true) + if env := decodeEnvelope(t, rec.Body.Bytes()); !env.Success { + t.Fatalf("by-node clear: %s", env.Msg) + } + if refs := probe.PingTasksFor(agentID); len(refs) != 0 { + t.Fatalf("after by-node clear: %v", refs) + } + // by-node on an unknown node 404s. + rec = rig.do(http.MethodPut, tasksPath+"/by-node/999999", []byte(`{"tasks":[],"base":[]}`), true) + if rec.Code != http.StatusNotFound { + t.Fatalf("by-node unknown node: %d", rec.Code) + } + + // Reorder: second task, flip, duplicates refused. + rec = rig.do(http.MethodPut, tasksPath, []byte(`{"name":"t2","target":"8.8.8.8:53","interval":30}`), true) + env = decodeEnvelope(t, rec.Body.Bytes()) + var created2 struct { + ID int64 `json:"id"` + } + _ = json.Unmarshal(env.Obj, &created2) + flip := `{"ids":[` + itoa64(created2.ID) + `,` + itoa64(created.ID) + `]}` + rec = rig.do(http.MethodPost, tasksPath+"/reorder", []byte(flip), true) + if env := decodeEnvelope(t, rec.Body.Bytes()); !env.Success { + t.Fatalf("reorder: %s", env.Msg) + } + dup := `{"ids":[` + itoa64(created.ID) + `,` + itoa64(created.ID) + `]}` + rec = rig.do(http.MethodPost, tasksPath+"/reorder", []byte(dup), true) + if env := decodeEnvelope(t, rec.Body.Bytes()); env.Success { + t.Fatal("duplicate reorder accepted") + } + tasks, _ := probe.PingTasks() + if len(tasks) != 2 || tasks[0].ID != created2.ID { + t.Fatalf("order after flip %+v", tasks) + } + + // Delete clears the agent's view too. + rec = rig.do(http.MethodDelete, tasksPath+"/"+itoa64(created.ID), nil, true) + if env := decodeEnvelope(t, rec.Body.Bytes()); !env.Success { + t.Fatalf("delete: %s", env.Msg) + } + if tasks, _ := probe.PingTasks(); len(tasks) != 1 { + t.Fatalf("after delete %v", tasks) + } + cleanPingTasks(t) +} + +// series=ping serves real rows now: medians from filed samples, the +// probes label map, and whole-window loss for a fully-missed task. +func TestProbeMetricsPingSeriesLive(t *testing.T) { + rig := newProbeTestRig(t) + cleanPingTasks(t) + rig.setSetting(t, "public_page", "on") + agentID, _ := rig.seedAgent(t, "b4metrics", true) + + good, err := probe.SavePingTask(probe.PingTask{Name: "good", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{agentID}}, nil) + if err != nil { + t.Fatal(err) + } + bad, err := probe.SavePingTask(probe.PingTask{Name: "bad", Target: "127.0.0.1:1", Interval: 60, AgentIDs: []int64{agentID}}, nil) + if err != nil { + t.Fatal(err) + } + st := probe.Default() + now := time.Now().Unix() + for i := 0; i < 3; i++ { + if err := st.InsertPings(agentID, []probe.PingResult{{TaskID: good, LatencyMs: 10}}, now-int64(i)*10); err != nil { + t.Fatal(err) + } + } + if err := st.InsertPings(agentID, []probe.PingResult{{TaskID: bad, LatencyMs: -1}}, now); err != nil { + t.Fatal(err) + } + + rec := rig.do(http.MethodGet, "/api/nodes/"+itoa64(agentID)+"/metrics?series=ping&hours=1", nil, false) + if rec.Code != http.StatusOK { + t.Fatalf("metrics: %d %s", rec.Code, rec.Body.String()) + } + var resp struct { + Ping []struct { + TaskID int64 `json:"task_id"` + Latency *int64 `json:"latency"` + Loss *int64 `json:"loss"` + } `json:"ping"` + Probes map[string]string `json:"probes"` + Loss map[string]float64 `json:"loss"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatal(err) + } + var goodRow, badRow bool + for _, row := range resp.Ping { + switch row.TaskID { + case good: + goodRow = true + if row.Latency == nil || *row.Latency != 10 { + t.Fatalf("good latency %v", row.Latency) + } + case bad: + badRow = true + if row.Latency != nil { + t.Fatalf("bad latency %v, want null", *row.Latency) + } + if row.Loss == nil || *row.Loss != 100 { + t.Fatalf("bad bucket loss %v", row.Loss) + } + } + } + if !goodRow || !badRow { + t.Fatalf("rows %+v", resp.Ping) + } + if resp.Probes[itoa64(good)] != "good" || resp.Probes[itoa64(bad)] != "bad" { + t.Fatalf("probes %v", resp.Probes) + } + if resp.Loss[itoa64(bad)] != 100.0 { + t.Fatalf("window loss %v", resp.Loss) + } + if _, ok := resp.Loss[itoa64(good)]; ok { + t.Fatalf("good task in loss map %v", resp.Loss) + } + cleanPingTasks(t) +} + +func itoa64(n int64) string { + return strings.TrimSpace(jsonNumber(n)) +} + +func jsonNumber(n int64) string { + b, _ := json.Marshal(n) + return string(b) +} diff --git a/internal/web/service/probe/ping_test.go b/internal/web/service/probe/ping_test.go new file mode 100644 index 0000000..530a47b --- /dev/null +++ b/internal/web/service/probe/ping_test.go @@ -0,0 +1,411 @@ +package probe + +import ( + "errors" + "strconv" + "testing" + "time" +) + +func saveTask(t *testing.T, st *Store, task PingTask) int64 { + t.Helper() + id, err := st.SavePingTask(task, nil) + if err != nil { + t.Fatal(err) + } + return id +} + +// Create assigns an id from the high-water mark, lists back with its +// assignments, and a deleted id is never reissued (hub next_id). +func TestPingTaskCRUDAndIDReuse(t *testing.T) { + st := newTestStore(t) + id1 := saveTask(t, st, PingTask{Name: "CF", Target: "1.1.1.1:443", Interval: 60, Region: "全球", AgentIDs: []int64{7}}) + id2 := saveTask(t, st, PingTask{Name: "G", Target: "8.8.8.8:53", Interval: 30}) + if id1 == 0 || id2 == 0 || id1 == id2 { + t.Fatalf("ids %d %d", id1, id2) + } + tasks, err := st.PingTasks() + if err != nil { + t.Fatal(err) + } + if len(tasks) != 2 || tasks[0].ID != id1 || tasks[1].ID != id2 { + t.Fatalf("list %+v", tasks) + } + if len(tasks[0].AgentIDs) != 1 || tasks[0].AgentIDs[0] != 7 || tasks[0].Region != "全球" { + t.Fatalf("task0 %+v", tasks[0]) + } + if tasks[1].Sort <= tasks[0].Sort { + t.Fatalf("sort not appended: %d %d", tasks[0].Sort, tasks[1].Sort) + } + if err := st.DeletePingTask(id1); err != nil { + t.Fatal(err) + } + id3 := saveTask(t, st, PingTask{Name: "Q", Target: "9.9.9.9:443", Interval: 60}) + if id3 == id1 || id3 == id2 { + t.Fatalf("id reissued: %d", id3) + } + // Updating a task another session deleted reports it (hub). + if _, err := st.SavePingTask(PingTask{ID: id1, Name: "x", Target: "1.1.1.1:443", Interval: 60}, nil); !errors.Is(err, ErrPingTaskGone) { + t.Fatalf("gone update: %v", err) + } +} + +// Base-diff saving: an assignment made after the editor's base snapshot +// survives the save; base members removed from the new set go away. +func TestPingTaskSaveBaseDiff(t *testing.T) { + st := newTestStore(t) + id := saveTask(t, st, PingTask{Name: "A", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1, 2}}) + // Another session assigns agent 3 meanwhile. + other := saveTaskPingAssign(t, st, id, []int64{1, 2, 3}) + _ = other + base := []int64{1, 2} + if _, err := st.SavePingTask(PingTask{ID: id, Name: "A", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{2}}, &base); err != nil { + t.Fatal(err) + } + tasks, _ := st.PingTasks() + got := map[int64]bool{} + for _, a := range tasks[0].AgentIDs { + got[a] = true + } + if got[1] || !got[2] || !got[3] { + t.Fatalf("base-diff result %v", tasks[0].AgentIDs) + } + // Without base the assignment is replaced wholesale. + if _, err := st.SavePingTask(PingTask{ID: id, Name: "A", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{9}}, nil); err != nil { + t.Fatal(err) + } + tasks, _ = st.PingTasks() + if len(tasks[0].AgentIDs) != 1 || tasks[0].AgentIDs[0] != 9 { + t.Fatalf("wholesale replace %v", tasks[0].AgentIDs) + } +} + +func saveTaskPingAssign(t *testing.T, st *Store, taskID int64, agents []int64) error { + t.Helper() + _, err := st.db.Exec(`DELETE FROM probe_ping_node WHERE task_id = ?`, taskID) + if err != nil { + return err + } + for _, a := range agents { + if _, err := st.db.Exec(`INSERT INTO probe_ping_node (task_id, agent_id) VALUES (?, ?)`, taskID, a); err != nil { + return err + } + } + return nil +} + +// Crowding: the 65th task on one agent refuses the whole save, and so +// does a 65th auto_join task (a fresh agent would be over its cap the +// moment it appears). +func TestPingTaskCrowding(t *testing.T) { + st := newTestStore(t) + for i := 0; i < MaxProbesPerNode; i++ { + saveTask(t, st, PingTask{Name: "t", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{5}}) + } + _, err := st.SavePingTask(PingTask{Name: "one too many", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{5}}, nil) + var crowded *PingCrowdedError + if !errors.As(err, &crowded) || crowded.AgentID != 5 { + t.Fatalf("crowded: %v", err) + } + if tasks, _ := st.PingTasks(); len(tasks) != MaxProbesPerNode { + t.Fatalf("refused save leaked: %d tasks", len(tasks)) + } + // The same count on a different agent is fine. + saveTask(t, st, PingTask{Name: "ok", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{6}}) + + st2 := newTestStore(t) + for i := 0; i < MaxProbesPerNode; i++ { + saveTask(t, st2, PingTask{Name: "t", Target: "1.1.1.1:443", Interval: 60, AutoJoin: true}) + } + _, err = st2.SavePingTask(PingTask{Name: "join too many", Target: "1.1.1.1:443", Interval: 60, AutoJoin: true}, nil) + var aj *PingAutoJoinError + if !errors.As(err, &aj) { + t.Fatalf("autoJoin crowded: %v", err) + } +} + +// SetNodePingTasks applies only the base difference and refuses when +// THIS agent would overflow; another agent's overflow is not its fault. +func TestSetNodePingTasks(t *testing.T) { + st := newTestStore(t) + a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60}) + b := saveTask(t, st, PingTask{Name: "b", Target: "1.1.1.1:443", Interval: 60}) + // Assigned meanwhile by the task-side editor: survives. + saveTaskPingAssign(t, st, b, []int64{3}) + if err := st.SetNodePingTasks(3, []int64{a}, []int64{}); err != nil { + t.Fatal(err) + } + if refs, _ := st.PingTasksFor(3); len(refs) != 2 { + t.Fatalf("node tasks %v", refs) + } + // A task deleted since the base makes the edit refuse. + if err := st.SetNodePingTasks(3, []int64{a, 9999}, []int64{a}); err == nil { + t.Fatal("deleted task accepted") + } +} + +// Reorder: duplicates and stale lists refuse; a full list renumbers +// sort and PingTasks reflects it. +func TestReorderPingTasks(t *testing.T) { + st := newTestStore(t) + a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60}) + b := saveTask(t, st, PingTask{Name: "b", Target: "1.1.1.1:443", Interval: 60}) + if err := st.ReorderPingTasks([]int64{a, a}); err == nil { + t.Fatal("duplicate accepted") + } + if err := st.ReorderPingTasks([]int64{a}); err == nil { + t.Fatal("short list accepted") + } + if err := st.ReorderPingTasks([]int64{b, a}); err != nil { + t.Fatal(err) + } + tasks, _ := st.PingTasks() + if tasks[0].ID != b || tasks[1].ID != a || tasks[0].Sort != 0 || tasks[1].Sort != 1 { + t.Fatalf("after reorder %+v", tasks) + } +} + +// PingTasksFor is id-ordered (the agent's cap keeps the first N) and +// PingTaskNames covers only assigned tasks. +func TestPingTasksForAndNames(t *testing.T) { + st := newTestStore(t) + a := saveTask(t, st, PingTask{Name: "first", Target: "1.1.1.1:443", Interval: 45, AgentIDs: []int64{2}}) + b := saveTask(t, st, PingTask{Name: "second", Target: "8.8.8.8:53", Interval: 90, AgentIDs: []int64{2}}) + saveTask(t, st, PingTask{Name: "elsewhere", Target: "9.9.9.9:443", Interval: 60, AgentIDs: []int64{3}}) + refs, err := st.PingTasksFor(2) + if err != nil { + t.Fatal(err) + } + if len(refs) != 2 || refs[0].ID != a || refs[1].ID != b || refs[0].Target != "1.1.1.1:443" || refs[0].Interval != 45 { + t.Fatalf("refs %+v", refs) + } + names, err := st.PingTaskNames(2) + if err != nil { + t.Fatal(err) + } + if len(names) != 2 || names[strconv.FormatInt(a, 10)] != "first" { + t.Fatalf("names %v", names) + } +} + +// InsertPings files only under live assignments: a result for an +// unassigned or deleted task is dropped silently; a second sample at +// the same stamp replaces the first (one session, one stamp). +func TestInsertPingsAssignmentGate(t *testing.T) { + st := newTestStore(t) + id := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{4}}) + ts := time.Now().Unix() + err := st.InsertPings(4, []PingResult{ + {TaskID: id, LatencyMs: 12}, + {TaskID: id + 100, LatencyMs: 5}, // never assigned + }, ts) + if err != nil { + t.Fatal(err) + } + // From an agent the task is not assigned to: dropped too. + if err := st.InsertPings(9, []PingResult{{TaskID: id, LatencyMs: 99}}, ts); err != nil { + t.Fatal(err) + } + var count int64 + var latency int64 + st.db.QueryRow(`SELECT COUNT(*), COALESCE(MAX(latency), -999) FROM probe_ping_record`).Scan(&count, &latency) + if count != 1 || latency != 12 { + t.Fatalf("rows %d latency %d", count, latency) + } + if err := st.InsertPings(4, []PingResult{{TaskID: id, LatencyMs: 20}}, ts); err != nil { + t.Fatal(err) + } + st.db.QueryRow(`SELECT COUNT(*), COALESCE(MAX(latency), -999) FROM probe_ping_record`).Scan(&count, &latency) + if count != 1 || latency != 20 { + t.Fatalf("after replace rows %d latency %d", count, latency) + } +} + +// PingHistory over minute rows: per-bucket medians, band only when the +// bucket varied, bucket loss rounded up, window loss unrounded and +// only for tasks that lost something. +func TestPingHistoryMinuteRows(t *testing.T) { + st := newTestStore(t) + a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}}) + b := saveTask(t, st, PingTask{Name: "b", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}}) + now := time.Now().Unix() + base := (now/60)*60 - 600 // ten minutes back, bucket aligned + span := Span{Since: base, Step: 60, Hourly: false} + // Task A: one bucket with answers 10, 20, 40 and one miss. + for i, lat := range []int64{10, 20, 40, -1} { + seedPing(t, st, 1, base+int64(i)*7, a, lat) + } + // Task B: one bucket, all misses. + seedPing(t, st, 1, base+5, b, -1) + seedPing(t, st, 1, base+15, b, -1) + + rows, loss, err := st.PingHistory(1, span) + if err != nil { + t.Fatal(err) + } + if len(rows) != 2 { + t.Fatalf("rows %+v", rows) + } + ra, rb := rows[0], rows[1] + if ra.TaskID != a || rb.TaskID != b { + t.Fatalf("order %+v", rows) + } + if ra.Latency == nil || *ra.Latency != 20 { + t.Fatalf("median %v", ra.Latency) + } + if ra.Band == nil || (*ra.Band)[0] != 10 || (*ra.Band)[1] != 40 { + t.Fatalf("band %v", ra.Band) + } + if ra.Loss == nil || *ra.Loss != 25 { // 1 of 4, exact + t.Fatalf("bucket loss %v", ra.Loss) + } + if rb.Latency != nil || rb.Band != nil { + t.Fatalf("all-miss row %+v", rb) + } + if rb.Loss == nil || *rb.Loss != 100 { + t.Fatalf("all-miss loss %v", rb.Loss) + } + if loss[strconv.FormatInt(a, 10)] != 25.0 { + t.Fatalf("window loss a %v", loss) + } + if loss[strconv.FormatInt(b, 10)] != 100.0 { + t.Fatalf("window loss b %v", loss) + } + // An agent without the assignment sees none of it. + rows2, loss2, err := st.PingHistory(2, span) + if err != nil { + t.Fatal(err) + } + if len(rows2) != 0 || len(loss2) != 0 { + t.Fatalf("unassigned view %v %v", rows2, loss2) + } +} + +// PingHistory over an hourly span merges folded hours (weighted by +// their answer counts) with the minute rows after the watermark. +func TestPingHistoryHourlyMerge(t *testing.T) { + st := newTestStore(t) + a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}}) + now := time.Now().Unix() + hourStart := hourFloor(now) - 3*3600 + span := Span{Since: hourStart, Step: 3600, Hourly: true} + // Folded hour: 3 answers (median 30, range 10..50), 1 lost. + if _, err := st.db.Exec(`INSERT INTO probe_ping_hour (agent_id, ts, task_id, answered, lost, latency, lo, hi) + VALUES (1, ?, ?, 3, 1, 30, 10, 50)`, hourStart, a); err != nil { + t.Fatal(err) + } + // Watermark at the hour boundary: minute rows at/after it count. + if _, err := st.db.Exec(`INSERT INTO probe_setting (key, value) VALUES (?, ?)`, rolledKey, strconv.FormatInt(hourStart+3600, 10)); err != nil { + t.Fatal(err) + } + minuteTs := hourStart + 3600 + 120 // inside the NEXT hour bucket + seedPing(t, st, 1, minuteTs, a, 100) + // A stale minute row below the watermark must not double count. + seedPing(t, st, 1, hourStart+60, a, 999) + + rows, loss, err := st.PingHistory(1, span) + if err != nil { + t.Fatal(err) + } + if len(rows) != 2 { + t.Fatalf("rows %+v", rows) + } + if rows[0].Ts != hourStart || rows[0].Latency == nil || *rows[0].Latency != 30 { + t.Fatalf("hour row %+v", rows[0]) + } + if rows[0].Band == nil || (*rows[0].Band)[0] != 10 || (*rows[0].Band)[1] != 50 { + t.Fatalf("hour band %+v", rows[0].Band) + } + if rows[0].Loss == nil || *rows[0].Loss != 25 { + t.Fatalf("hour loss %v", rows[0].Loss) + } + if rows[1].Ts != hourStart+3600 || rows[1].Latency == nil || *rows[1].Latency != 100 { + t.Fatalf("minute row %+v", rows[1]) + } + // Window loss counts the hour tally plus the minute row: 1 lost of 5. + if loss[strconv.FormatInt(a, 10)] != 20.0 { + t.Fatalf("window loss %v", loss) + } +} + +// Weighted median: with answers weighted by counts, the middle ranks +// decide — two middle values average (hub Tally::median). +func TestPingWeightedMedianAveragesMiddleTwo(t *testing.T) { + st := newTestStore(t) + a := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}}) + now := time.Now().Unix() + base := (now / 3600) * 3600 + span := Span{Since: base, Step: 3600, Hourly: true} + // Two folded hours inside one bucket: medians 10 (×1) and 21 (×1) + // → total 2, ranks 1 and 2 → (10+21)/2 = 15 (integer). + for i, m := range []int64{10, 21} { + if _, err := st.db.Exec(`INSERT INTO probe_ping_hour (agent_id, ts, task_id, answered, lost, latency, lo, hi) + VALUES (1, ?, ?, 1, 0, ?, ?, ?)`, base+int64(i)*60, a, m, m, m); err != nil { + t.Fatal(err) + } + } + if _, err := st.db.Exec(`INSERT INTO probe_setting (key, value) VALUES (?, ?)`, rolledKey, strconv.FormatInt(base+3600, 10)); err != nil { + t.Fatal(err) + } + rows, _, err := st.PingHistory(1, span) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].Latency == nil || *rows[0].Latency != 15 { + t.Fatalf("weighted median %+v", rows) + } +} + +// Auto-join: an agent's first recorded sample assigns it every +// auto_join task; agents already known are NOT retroactively joined +// when a task is flagged later (hub semantics). +func TestAutoJoinOnFirstRecord(t *testing.T) { + st := newTestStore(t) + auto := saveTask(t, st, PingTask{Name: "auto", Target: "1.1.1.1:443", Interval: 60, AutoJoin: true}) + saveTask(t, st, PingTask{Name: "manual", Target: "1.1.1.1:443", Interval: 60}) + + // Existing agent 1 records before any task existed... (tasks were + // created first here, so use a pre-existing agent row instead:) + st.Record(1, Sample{BootID: "b1"}, time.Now(), 0) + if refs, _ := st.PingTasksFor(1); len(refs) != 1 || refs[0].ID != auto { + t.Fatalf("first record join %v", refs) + } + // Agent 2's first record joins too. + st.Record(2, Sample{BootID: "b2"}, time.Now(), 0) + if refs, _ := st.PingTasksFor(2); len(refs) != 1 { + t.Fatalf("agent2 join %v", refs) + } + // A task flagged auto_join afterwards does not reach agent 1/2. + late := saveTask(t, st, PingTask{Name: "late", Target: "1.1.1.1:443", Interval: 60, AutoJoin: true}) + st.Record(1, Sample{BootID: "b1"}, time.Now(), time.Now().UnixMilli()) + if refs, _ := st.PingTasksFor(1); len(refs) != 1 { + t.Fatalf("retroactive join happened: %v", refs) + } + // But a brand-new agent 3 picks up both. + st.Record(3, Sample{BootID: "b3"}, time.Now(), 0) + if refs, _ := st.PingTasksFor(3); len(refs) != 2 || refs[1].ID != late { + t.Fatalf("agent3 join %v", refs) + } +} + +// DeletePingTask clears records, hours and assignments with the task. +func TestDeletePingTaskClearsAll(t *testing.T) { + st := newTestStore(t) + id := saveTask(t, st, PingTask{Name: "a", Target: "1.1.1.1:443", Interval: 60, AgentIDs: []int64{1}}) + seedPing(t, st, 1, time.Now().Unix(), id, 5) + if _, err := st.db.Exec(`INSERT INTO probe_ping_hour (agent_id, ts, task_id, answered, lost, latency, lo, hi) + VALUES (1, 0, ?, 1, 0, 5, 5, 5)`, id); err != nil { + t.Fatal(err) + } + if err := st.DeletePingTask(id); err != nil { + t.Fatal(err) + } + for _, table := range []string{"probe_ping_record", "probe_ping_hour", "probe_ping_node", "probe_ping_task"} { + var n int64 + st.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&n) + if n != 0 { + t.Fatalf("%s still has %d rows", table, n) + } + } +}