diff --git a/internal/web/controller/agent.go b/internal/web/controller/agent.go index cb8b502..7d66570 100644 --- a/internal/web/controller/agent.go +++ b/internal/web/controller/agent.go @@ -17,6 +17,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/logger" "github.com/mhsanaei/3x-ui/v3/internal/selfupdate" "github.com/mhsanaei/3x-ui/v3/internal/web/service" + "github.com/mhsanaei/3x-ui/v3/internal/web/service/probe" "github.com/mhsanaei/3x-ui/v3/internal/xray" "github.com/gin-gonic/gin" @@ -276,6 +277,7 @@ type heartbeatProbeFacts struct { type heartbeatProbeMetrics struct { BootID *string `json:"bootId"` Iface *string `json:"iface"` + Uptime *uint64 `json:"uptime"` Load *[3]float32 `json:"load"` MemTotal *uint64 `json:"memTotal"` MemUsed *uint64 `json:"memUsed"` @@ -393,6 +395,55 @@ func (a *AgentController) updateAgentProbe(id uint, f *heartbeatProbeFacts, m *h _ = database.GetDB().Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error } +// probeSample flattens a heartbeat metrics block into the probe +// package's Sample. Absent fields become zero values — the agent sends +// a full block whenever it sends one, and the probe store treats the +// block's presence (not per-field presence) as the signal, mirroring +// the hub's report frames. +func probeSample(m *heartbeatProbeMetrics) probe.Sample { + var s probe.Sample + if m.BootID != nil { + s.BootID = *m.BootID + } + if m.Uptime != nil { + s.Uptime = *m.Uptime + } + if m.CPU != nil { + s.CPU = *m.CPU + } + if m.MemUsed != nil { + s.MemUsed = *m.MemUsed + } + if m.SwapUsed != nil { + s.SwapUsed = *m.SwapUsed + } + if m.DiskUsed != nil { + s.DiskUsed = *m.DiskUsed + } + if m.NetRxTotal != nil { + s.NetRxTotal = *m.NetRxTotal + } + if m.NetTxTotal != nil { + s.NetTxTotal = *m.NetTxTotal + } + if m.NetRx != nil { + s.NetRx = *m.NetRx + } + if m.NetTx != nil { + s.NetTx = *m.NetTx + } + if m.TCP != nil { + s.TCP = *m.TCP + } + if m.UDP != nil { + s.UDP = *m.UDP + } + if m.Procs != nil { + s.Procs = *m.Procs + } + return s +} + // heartbeatUpdateStatus is the agent's report about a one-click update // intent (item 20), piggybacked on the heartbeat like telemetry. type heartbeatUpdateStatus struct { @@ -476,11 +527,21 @@ func (a *AgentController) heartbeat(c *gin.Context) { ProbeMetrics *heartbeatProbeMetrics `json:"probeMetrics"` } _ = c.ShouldBindJSON(&req) + // The last-seen stamp as it stood before this heartbeat: the probe + // ledger's unseen-boot rule measures a reboot against it, so it + // must be captured before touchAgent overwrites it. + prevLastSeen := agent.LastSeen a.touchAgent(uint(agent.Id), "") reportedVersion := agent.AgentVersion if req.ProbeFacts != nil || req.ProbeMetrics != nil { a.updateAgentProbe(uint(agent.Id), req.ProbeFacts, req.ProbeMetrics) } + if req.ProbeMetrics != nil { + // Item 7 B2: file the sample into probe.db history + the + // whole-machine traffic ledger. Best-effort by design — the + // probe package logs and swallows its own failures. + probe.Record(int64(agent.Id), probeSample(req.ProbeMetrics), prevLastSeen) + } if req.Telemetry != nil { a.updateAgentTelemetry(uint(agent.Id), req.Telemetry) if req.Telemetry.AgentVersion != nil && *req.Telemetry.AgentVersion != "" { diff --git a/internal/web/job/probe_history_job.go b/internal/web/job/probe_history_job.go new file mode 100644 index 0000000..916fbb4 --- /dev/null +++ b/internal/web/job/probe_history_job.go @@ -0,0 +1,22 @@ +package job + +import ( + "github.com/mhsanaei/3x-ui/v3/internal/web/service/probe" +) + +// ProbeHistoryJob is the hourly housekeeping of the item 7 probe +// history (roll_up + prune against probe.db). The heavy lifting lives +// in the probe package; the job is a thin cron shell, and every failure +// is logged there rather than propagated — a broken history store must +// never disturb the panel. +type ProbeHistoryJob struct{} + +// NewProbeHistoryJob creates a probe history maintenance job instance. +func NewProbeHistoryJob() *ProbeHistoryJob { + return new(ProbeHistoryJob) +} + +// Run executes one maintenance pass (roll_up, then prune). +func (j *ProbeHistoryJob) Run() { + probe.Maintain() +} diff --git a/internal/web/service/probe/history.go b/internal/web/service/probe/history.go new file mode 100644 index 0000000..05e8588 --- /dev/null +++ b/internal/web/service/probe/history.go @@ -0,0 +1,291 @@ +package probe + +import ( + "math" + "time" +) + +// Sample is one heartbeat's probe metrics block, as the master received +// it (item 7 B1 wire shape). A Sample is only built when the heartbeat +// carried a metrics block at all; individual zero values mirror the +// agent's softened reads, exactly like a hub report frame. +type Sample struct { + BootID string + Uptime uint64 // seconds; 0 = unreadable, i.e. no boot time + CPU float32 + MemUsed uint64 + SwapUsed uint64 + DiskUsed uint64 + NetRxTotal uint64 // kernel lifetime counters of the counted ifaces + NetTxTotal uint64 + NetRx uint64 // agent-reported per-second rates (peaks, fallback) + NetTx uint64 + TCP uint32 + UDP uint32 + Procs uint32 +} + +// reading is one report's pair of kernel byte counters, the boot epoch +// they belong to, and when the master received them (hub Reading). +type reading struct { + epoch string + rx, tx int64 + at time.Time // wall clock: stamps and calendar dates + tick time.Time // monotonic anchor for rate durations + booted *int64 // unix seconds the machine booted, by master's clock + booked bool +} + +// continuedBy reports whether next continues this reading's span. A new +// epoch or a counter that moved backwards leaves nothing to subtract. +func (r *reading) continuedBy(next *reading) bool { + return r.epoch == next.epoch && next.rx >= r.rx && next.tx >= r.tx +} + +// minuteAcc is the running mean/peak accumulator for the minute in +// progress (hub Minute): mean fields are summed, peaks are maxed. +type minuteAcc struct { + n float64 + cpuSum float64 + memSum float64 + swapSum float64 + diskSum float64 + tcpSum float64 + udpSum float64 + procsSum float64 + cpuPeak float64 + rxPeak uint64 + txPeak uint64 +} + +func (a *minuteAcc) add(s *Sample) { + a.n++ + a.cpuSum += float64(s.CPU) + a.memSum += float64(s.MemUsed) + a.swapSum += float64(s.SwapUsed) + a.diskSum += float64(s.DiskUsed) + a.tcpSum += float64(s.TCP) + a.udpSum += float64(s.UDP) + a.procsSum += float64(s.Procs) + if float64(s.CPU) > a.cpuPeak { + a.cpuPeak = float64(s.CPU) + } + if s.NetRx > a.rxPeak { + a.rxPeak = s.NetRx + } + if s.NetTx > a.txPeak { + a.txPeak = s.NetTx + } +} + +// counterMark is where a history row's network rate starts: the kernel +// counters as an earlier report carried them (hub Mark). +type counterMark struct { + tick time.Time + epoch string + rx int64 + tx int64 +} + +// nodeState is one agent's in-memory ingest state (hub Agent + the held +// reading of file()). It is rebuilt within one heartbeat of a master +// restart; nothing here is the only copy of anything booked. +type nodeState struct { + lastMinute *int64 // minute bucket the stored-row cadence is on + acc minuteAcc + mark *counterMark + held *reading +} + +func u64toI64(v uint64) (int64, bool) { + if v > math.MaxInt64 { + return 0, false + } + return int64(v), true +} + +// Record files one heartbeat sample: it books the traffic reading (hub +// file()) and, on a minute change, stores the aggregated minute row +// (hub report()). at is the arrival time on the master's clock; +// prevLastSeenSec is the agent's last-seen stamp from before this +// heartbeat (seconds), which the unseen-boot rule measures against. +// +// Errors are returned for tests; the facade swallows them. +func (s *Store) Record(agentID int64, smp Sample, at time.Time, prevLastSeenSec int64) error { + s.mu.Lock() + defer s.mu.Unlock() + st := s.nodes[agentID] + if st == nil { + st = &nodeState{} + s.nodes[agentID] = st + } + + // The traffic reading exists only with an epoch and a valid pair of + // counters: no reading is not a reading of zero — booking zero would + // align the baseline to it and book the next report's lifetime + // counter as one delta. + var rd *reading + if smp.BootID != "" { + if rx, ok1 := u64toI64(smp.NetRxTotal); ok1 { + if tx, ok2 := u64toI64(smp.NetTxTotal); ok2 { + rd = &reading{epoch: smp.BootID, rx: rx, tx: tx, at: at, tick: at} + if smp.Uptime > 0 { + b := at.Unix() - int64(smp.Uptime) + rd.booted = &b + } + } + } + } + if rd != nil { + if err := s.file(st, agentID, rd, prevLastSeenSec); err != nil { + return err + } + } + + // Minute row: the accumulator takes every sample; a sample arriving + // in a new minute flushes a row stamped with that minute, summarizing + // the interval since the previous row (hub semantics — the means + // include the flushing sample). + minute := at.Unix() / 60 + st.acc.add(&smp) + store := false + if st.lastMinute == nil { + st.lastMinute = &minute + } else if *st.lastMinute != minute { + store = true + } + if store { + if err := s.storeRow(st, agentID, smp, rd, at, minute); err != nil { + return err + } + st.acc = minuteAcc{} + st.lastMinute = &minute + } + if rd != nil { + if store || st.mark == nil || st.mark.epoch != rd.epoch { + st.mark = &counterMark{tick: rd.tick, epoch: rd.epoch, rx: rd.rx, tx: rd.tx} + } + } + return nil +} + +// file books readings into the traffic row about once a minute rather +// than with every report (hub file()): each booking is a DB write, and +// skipping the readings in between loses nothing because the kernel +// counter keeps counting — provided every break is booked on both +// sides: +// +// - no reading held: book this one; +// - a break (epoch change / counter backwards): book the held reading +// as the last of its span, then this one as the first of the next; +// - a new minute: book the held reading, dated by its own arrival, so +// bytes moved before midnight count toward that day; where it was +// already booked, book this one instead. +func (s *Store) file(st *nodeState, agentID int64, rd *reading, prevLastSeenSec int64) error { + book := func(r *reading) error { + _, err := s.accumulate(agentID, r.epoch, r.rx, r.tx, r.at, r.booted, prevLastSeenSec) + return err + } + var bookLast, bookThis bool + switch { + case st.held == nil: + bookThis = true + case !st.held.continuedBy(rd): + bookLast, bookThis = !st.held.booked, true + case st.held.at.Unix()/60 != rd.at.Unix()/60: + bookLast, bookThis = !st.held.booked, st.held.booked + default: + bookLast, bookThis = false, false + } + if bookLast { + if err := book(st.held); err != nil { + return err + } + } + if bookThis { + if err := book(rd); err != nil { + return err + } + } + rd.booked = bookThis + st.held = rd + return nil +} + +// storeRow writes the flushed minute row (hub insert_metric via +// Minute.write_into): mean fields from the accumulator (integer means +// rounded), peaks from the accumulator, and the network rates derived +// from the counter mark when the span is unbroken — so the stored +// rates integrate to the booked traffic — else the agent's own rates +// from the flushing sample. +func (s *Store) storeRow(st *nodeState, agentID int64, smp Sample, rd *reading, at time.Time, minute int64) error { + acc := &st.acc + row := struct { + cpu, cpuMax float64 + mem, swap, disk int64 + netRx, netTx int64 + netRxMax, netTxMax int64 + tcp, udp, procs int64 + }{ + cpu: float64(smp.CPU), cpuMax: float64(smp.CPU), + mem: int64(smp.MemUsed), swap: int64(smp.SwapUsed), disk: int64(smp.DiskUsed), + netRx: int64(smp.NetRx), netTx: int64(smp.NetTx), + netRxMax: int64(smp.NetRx), netTxMax: int64(smp.NetTx), + tcp: int64(smp.TCP), udp: int64(smp.UDP), procs: int64(smp.Procs), + } + if acc.n > 0 { + row.cpu = acc.cpuSum / acc.n + row.mem = int64(math.Round(acc.memSum / acc.n)) + row.swap = int64(math.Round(acc.swapSum / acc.n)) + row.disk = int64(math.Round(acc.diskSum / acc.n)) + row.tcp = int64(math.Round(acc.tcpSum / acc.n)) + row.udp = int64(math.Round(acc.udpSum / acc.n)) + row.procs = int64(math.Round(acc.procsSum / acc.n)) + row.cpuMax = acc.cpuPeak + row.netRxMax = int64(acc.rxPeak) + row.netTxMax = int64(acc.txPeak) + } + if rd != nil && st.mark != nil && st.mark.epoch == rd.epoch { + elapsed := at.Sub(st.mark.tick).Seconds() + if elapsed < 1.0 { + elapsed = 1.0 + } + if d := rd.rx - st.mark.rx; d > 0 { + row.netRx = int64(float64(d) / elapsed) + } else { + row.netRx = 0 + } + if d := rd.tx - st.mark.tx; d > 0 { + row.netTx = int64(float64(d) / elapsed) + } else { + row.netTx = 0 + } + } + _, err := s.db.Exec(` + INSERT OR REPLACE INTO probe_metric + (agent_id, ts, cpu, mem_used, swap_used, disk_used, net_rx, net_tx, + tcp, udp, procs, net_rx_max, net_tx_max, cpu_max) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + agentID, minute*60, row.cpu, row.mem, row.swap, row.disk, row.netRx, row.netTx, + row.tcp, row.udp, row.procs, row.netRxMax, row.netTxMax, row.cpuMax) + return err +} + +// BookHeld books every node's held, not-yet-booked reading (hub +// book_held): a node that reboots before reporting again would otherwise +// take the bytes since its last booking with it. Called on master +// shutdown; best-effort. prevLastSeen is no longer meaningful at +// shutdown, so the unseen-boot exception cannot fire from here (the +// readings being booked continue spans the master already saw). +func (s *Store) BookHeld() { + s.mu.Lock() + defer s.mu.Unlock() + for id, st := range s.nodes { + if st.held != nil && !st.held.booked { + r := st.held + if _, err := s.accumulate(id, r.epoch, r.rx, r.tx, r.at, r.booted, 0); err == nil { + r.booked = true + } + } + } +} diff --git a/internal/web/service/probe/history_test.go b/internal/web/service/probe/history_test.go new file mode 100644 index 0000000..98a2a0b --- /dev/null +++ b/internal/web/service/probe/history_test.go @@ -0,0 +1,175 @@ +package probe + +import ( + "testing" + "time" +) + +type metricRow struct { + ts int64 + cpu, cpuMax float64 + mem, swap, disk int64 + netRx, netTx, netRxMax, netTxMax int64 + tcp, udp, procs int64 +} + +func readMetricRows(t *testing.T, st *Store, agent int64) []metricRow { + t.Helper() + rows, err := st.db.Query(` + SELECT ts, cpu, mem_used, swap_used, disk_used, net_rx, net_tx, + tcp, udp, procs, net_rx_max, net_tx_max, cpu_max + FROM probe_metric WHERE agent_id = ? ORDER BY ts`, agent) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []metricRow + for rows.Next() { + var r metricRow + if err := rows.Scan(&r.ts, &r.cpu, &r.mem, &r.swap, &r.disk, &r.netRx, &r.netTx, + &r.tcp, &r.udp, &r.procs, &r.netRxMax, &r.netTxMax, &r.cpuMax); err != nil { + t.Fatal(err) + } + out = append(out, r) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + return out +} + +func rec(t *testing.T, st *Store, agent int64, smp Sample, at time.Time) { + t.Helper() + if err := st.Record(agent, smp, at, 0); err != nil { + t.Fatalf("record: %v", err) + } +} + +// Samples landing in the same minute bucket merge into one row when +// the next minute's sample flushes it: mean fields average (ints +// rounded), peaks max, and the stored net rate derives from the +// kernel counters between the row marks, not from the agent's rates. +func TestMinuteBucketMergeAndDerivedRates(t *testing.T) { + st := newTestStore(t) + base := time.Now().Add(-10 * time.Minute).Truncate(time.Minute) + mk := func(cpu float32, rxTot uint64, rate uint64, mem uint64) Sample { + return Sample{BootID: "k/fp", CPU: cpu, MemUsed: mem, NetRxTotal: rxTot, NetTxTotal: rxTot * 2, + NetRx: rate, NetTx: rate * 2, TCP: 3, UDP: 2, Procs: 10} + } + rec(t, st, 1, mk(10, 60_000, 1000, 100), base) // minute M, first sample + rec(t, st, 1, mk(20, 90_000, 1000, 200), base.Add(30*time.Second)) // minute M again + rec(t, st, 1, mk(30, 150_000, 1000, 300), base.Add(time.Minute)) // minute M+1: flushes + rows := readMetricRows(t, st, 1) + if len(rows) != 1 { + t.Fatalf("want 1 row, got %d", len(rows)) + } + r := rows[0] + if r.ts != base.Add(time.Minute).Unix() { + t.Fatalf("row stamped %d, want flushing minute %d", r.ts, base.Add(time.Minute).Unix()) + } + if r.cpu != 20 { // mean of 10,20,30 (flushing sample included) + t.Fatalf("cpu mean=%v, want 20", r.cpu) + } + if r.mem != 200 { + t.Fatalf("mem mean=%d, want 200", r.mem) + } + if r.cpuMax != 30 { + t.Fatalf("cpu_max=%v, want 30", r.cpuMax) + } + if r.netRx != 1500 { // (150000-60000) / 60s between marks + t.Fatalf("net_rx=%d, want 1500 (counter-derived)", r.netRx) + } + if r.netRxMax != 1000 { + t.Fatalf("net_rx_max=%d, want agent peak 1000", r.netRxMax) + } + // Next minute: single-sample row; rate from the new mark. + rec(t, st, 1, mk(40, 210_000, 700, 400), base.Add(2*time.Minute)) + rows = readMetricRows(t, st, 1) + if len(rows) != 2 { + t.Fatalf("want 2 rows, got %d", len(rows)) + } + r = rows[1] + if r.cpu != 40 || r.netRx != 1000 || r.netRxMax != 700 { + t.Fatalf("second row wrong: %+v", r) + } +} + +// At an epoch break the counter mark no longer applies, so the row +// keeps the agent's own reported rates. +func TestMinuteRowEpochBreakKeepsAgentRates(t *testing.T) { + st := newTestStore(t) + base := time.Now().Add(-10 * time.Minute).Truncate(time.Minute) + rec(t, st, 1, Sample{BootID: "k1/fp", NetRxTotal: 1000, NetTxTotal: 1000, NetRx: 111, NetTx: 222}, base) + rec(t, st, 1, Sample{BootID: "k2/fp", NetRxTotal: 500, NetTxTotal: 500, NetRx: 333, NetTx: 444}, base.Add(time.Minute)) + rows := readMetricRows(t, st, 1) + if len(rows) != 1 { + t.Fatalf("want 1 row, got %d", len(rows)) + } + if rows[0].netRx != 333 || rows[0].netTx != 444 { + t.Fatalf("epoch break must keep agent rates: %+v", rows[0]) + } +} + +// Samples without a boot_id still file metric rows (hub files rows +// from any well-formed report) but never touch the traffic ledger. +func TestRecordWithoutBootFilesRowOnly(t *testing.T) { + st := newTestStore(t) + base := time.Now().Add(-10 * time.Minute).Truncate(time.Minute) + rec(t, st, 1, Sample{CPU: 5, MemUsed: 42}, base) + rec(t, st, 1, Sample{CPU: 7, MemUsed: 43}, base.Add(time.Minute)) + if rows := readMetricRows(t, st, 1); len(rows) != 1 { + t.Fatalf("want 1 row, got %d", len(rows)) + } + if _, ok, err := st.TrafficFor(1); err != nil || ok { + t.Fatalf("no traffic row expected: ok=%v err=%v", ok, err) + } +} + +// The held-reading rhythm: readings inside a minute are not booked +// one by one; the minute change books the held one (dated by its own +// arrival), and BookHeld books the tail — so totals equal the full +// counter climb exactly once. +func TestHeldReadingRhythmAndBookHeld(t *testing.T) { + st := newTestStore(t) + base := time.Now().Add(-10 * time.Minute).Truncate(time.Minute) + mk := func(rx uint64) Sample { return Sample{BootID: "k/fp", NetRxTotal: rx, NetTxTotal: rx} } + rec(t, st, 1, mk(1000), base) // booked at once (first), delta 0 + rec(t, st, 1, mk(1300), base.Add(20*time.Second)) // held + rec(t, st, 1, mk(1600), base.Add(40*time.Second)) // held (replaces) + rec(t, st, 1, mk(1900), base.Add(time.Minute)) // new minute: books held@40s (delta 600) + tr, ok, _ := st.TrafficFor(1) + if !ok || tr.TotalRx != 600 { + t.Fatalf("after minute change total=%d ok=%v, want 600", tr.TotalRx, ok) + } + rec(t, st, 1, mk(2200), base.Add(2*time.Minute)) // books held@60s (delta 300) + tr, _, _ = st.TrafficFor(1) + if tr.TotalRx != 900 { + t.Fatalf("total=%d, want 900", tr.TotalRx) + } + st.BookHeld() // books the @120s reading (delta 300) + tr, _, _ = st.TrafficFor(1) + if tr.TotalRx != 1200 { + t.Fatalf("after BookHeld total=%d, want 1200 (=2200-1000)", tr.TotalRx) + } +} + +// A break (epoch change) books both sides at once: the held reading +// closes its span, the new one aligns the new baseline with delta 0. +func TestRecordBreakBooksBothSides(t *testing.T) { + st := newTestStore(t) + base := time.Now().Add(-10 * time.Minute).Truncate(time.Minute) + mk := func(boot string, rx uint64) Sample { return Sample{BootID: boot, NetRxTotal: rx, NetTxTotal: rx} } + rec(t, st, 1, mk("k1/fp", 1000), base) + rec(t, st, 1, mk("k1/fp", 1400), base.Add(20*time.Second)) // held, unbooked + rec(t, st, 1, mk("k2/fp", 50), base.Add(40*time.Second)) // break: books held (400) + new (0) + tr, _, _ := st.TrafficFor(1) + if tr.TotalRx != 400 { + t.Fatalf("break booking wrong: total=%d, want 400", tr.TotalRx) + } + rec(t, st, 1, mk("k2/fp", 150), base.Add(80*time.Second)) // new minute books held@40s (0), holds this + st.BookHeld() + tr, _, _ = st.TrafficFor(1) + if tr.TotalRx != 500 { + t.Fatalf("total=%d, want 500", tr.TotalRx) + } +} diff --git a/internal/web/service/probe/maintenance.go b/internal/web/service/probe/maintenance.go new file mode 100644 index 0000000..0384e02 --- /dev/null +++ b/internal/web/service/probe/maintenance.go @@ -0,0 +1,348 @@ +package probe + +import ( + "database/sql" + "math" + "sort" + "strconv" + "time" +) + +// watermark reads the roll-up watermark (unix seconds, hour-aligned): +// the next hour not yet folded into the hourly tier. +func (s *Store) watermark() (int64, bool, error) { + var v string + err := s.db.QueryRow(`SELECT value FROM probe_setting WHERE key = ?`, rolledKey).Scan(&v) + if err == sql.ErrNoRows { + return 0, false, nil + } + if err != nil { + return 0, false, err + } + n, perr := strconv.ParseInt(v, 10, 64) + if perr != nil { + return 0, false, nil + } + return n, true, nil +} + +func (s *Store) setWatermark(tx *sql.Tx, hour int64) error { + _, err := tx.Exec(`INSERT OR REPLACE INTO probe_setting (key, value) VALUES (?, ?)`, + rolledKey, strconv.FormatInt(hour, 10)) + return err +} + +func hourFloor(ts int64) int64 { + r := ts % 3600 + if r < 0 { + r += 3600 + } + return ts - r +} + +// oldestStamp is the earliest ts across the minute-tier tables, used +// to initialize the watermark on first roll_up (hub `oldest`). +func (s *Store) oldestStamp() (int64, bool, error) { + var oldest sql.NullInt64 + err := s.db.QueryRow(` + SELECT MIN(ts) FROM ( + SELECT ts FROM probe_metric + UNION ALL SELECT ts FROM probe_ping_record + )`).Scan(&oldest) + if err != nil { + return 0, false, err + } + return oldest.Int64, oldest.Valid, nil +} + +// RollUp folds closed hours from the minute tier into the hourly tier +// (hub roll_up): the watermark advances one hour per transaction, so a +// failure leaves the hour to be folded again rather than half folded, +// and INSERT OR REPLACE makes a second fold of the same hour a rewrite. +// Hours older than the retention floor are skipped (the watermark +// jumps to the floor), and the most recent closed hour waits out the +// late grace. Returns the number of hours folded. +func (s *Store) RollUp(now int64, keepDays int64) (int, error) { + if _, ok, err := s.watermark(); err != nil { + return 0, err + } else if !ok { + from := now + if oldest, valid, err := s.oldestStamp(); err != nil { + return 0, err + } else if valid { + from = oldest + } + tx, err := s.db.Begin() + if err != nil { + return 0, err + } + if err := s.setWatermark(tx, hourFloor(from)); err != nil { + tx.Rollback() + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, err + } + } + floor := hourFloor(now - keepDays*86400) + folded := 0 + for { + ok, err := s.foldNext(floor, now-3600-lateSeconds) + if err != nil { + return folded, err + } + if !ok { + return folded, nil + } + folded++ + } +} + +// foldNext folds the hour at the watermark (or at floor when the +// watermark is older) and moves the watermark past it, in one +// transaction. False once the hour is past last. +func (s *Store) foldNext(floor, last int64) (bool, error) { + tx, err := s.db.Begin() + if err != nil { + return false, err + } + defer tx.Rollback() + + var wmVal string + err = tx.QueryRow(`SELECT value FROM probe_setting WHERE key = ?`, rolledKey).Scan(&wmVal) + if err == sql.ErrNoRows { + return false, nil + } + if err != nil { + return false, err + } + wm, perr := strconv.ParseInt(wmVal, 10, 64) + if perr != nil { + return false, nil + } + hour := wm + if floor > hour { + hour = floor + } + if hour > last { + return false, nil + } + + // Metric fold: means weighted naturally by COUNT, byte columns + // truncate their mean (a loss under one byte), the count columns + // round (hub fold_next). + if _, err := tx.Exec(` + INSERT OR REPLACE INTO probe_metric_hour + (agent_id, ts, minutes, cpu, cpu_max, mem_used, swap_used, disk_used, + net_rx, net_tx, net_rx_max, net_tx_max, tcp, udp, procs) + SELECT agent_id, ?, COUNT(*), AVG(cpu), MAX(MAX(cpu, cpu_max)), + CAST(AVG(mem_used) AS INTEGER), CAST(AVG(swap_used) AS INTEGER), + CAST(AVG(disk_used) AS INTEGER), CAST(AVG(net_rx) AS INTEGER), + CAST(AVG(net_tx) AS INTEGER), MAX(MAX(net_rx, net_rx_max)), + MAX(MAX(net_tx, net_tx_max)), CAST(ROUND(AVG(tcp)) AS INTEGER), + CAST(ROUND(AVG(udp)) AS INTEGER), CAST(ROUND(AVG(procs)) AS INTEGER) + FROM probe_metric WHERE ts >= ? AND ts < ? + 3600 + GROUP BY agent_id`, hour, hour, hour); err != nil { + return false, err + } + + // Ping fold: per agent and task, the hour's median/lo/hi over the + // answered results and the answered/lost counts (hub Tally). + pingRows, err := tx.Query(` + SELECT agent_id, task_id, latency FROM probe_ping_record + WHERE ts >= ? AND ts < ? + 3600 ORDER BY agent_id, task_id`, hour, hour) + if err != nil { + return false, err + } + type tallyKey struct{ agent, task int64 } + tallies := make(map[tallyKey]*pingTally) + var order []tallyKey + for pingRows.Next() { + var agent, task, latency int64 + if err := pingRows.Scan(&agent, &task, &latency); err != nil { + pingRows.Close() + return false, err + } + k := tallyKey{agent, task} + t, ok := tallies[k] + if !ok { + t = &pingTally{} + tallies[k] = t + order = append(order, k) + } + t.add(latency) + } + if err := pingRows.Err(); err != nil { + pingRows.Close() + return false, err + } + pingRows.Close() + for _, k := range order { + t := tallies[k] + median, lo, hi := t.medianLoHi() + if _, err := tx.Exec(` + INSERT OR REPLACE INTO probe_ping_hour + (agent_id, ts, task_id, answered, lost, latency, lo, hi) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + k.agent, hour, k.task, t.answered, t.lost, + nullOr(median), nullOr(lo), nullOr(hi)); err != nil { + return false, err + } + } + + if err := s.setWatermark(tx, hour+3600); err != nil { + return false, err + } + if err := tx.Commit(); err != nil { + return false, err + } + return true, nil +} + +func nullOr(v *int64) any { + if v == nil { + return nil + } + return *v +} + +// pingTally accumulates one hour of probe results for one task +// (hub Tally over single results). +type pingTally struct { + answers []int64 + lost int64 + answered int64 + lo, hi *int64 +} + +func (t *pingTally) add(latency int64) { + if latency < 0 { + t.lost++ + return + } + t.answered++ + t.answers = append(t.answers, latency) + if t.lo == nil || latency < *t.lo { + v := latency + t.lo = &v + } + if t.hi == nil || latency > *t.hi { + v := latency + t.hi = &v + } +} + +// medianLoHi returns the median of the answers (the middle two +// averaged) and the lo/hi bounds; all nil when nothing answered. +func (t *pingTally) medianLoHi() (median, lo, hi *int64) { + if t.answered == 0 { + return nil, nil, nil + } + sort.Slice(t.answers, func(i, j int) bool { return t.answers[i] < t.answers[j] }) + n := int64(len(t.answers)) + mid := (t.answers[(n-1)/2] + t.answers[n/2]) / 2 + return &mid, t.lo, t.hi +} + +// Prune drops history beyond the retention window (hub prune): minute +// rows past DetailDays (or the window itself when shorter), hourly +// rows past keepDays. Minute rows are only deleted up to the roll-up +// watermark — a row outlives its window until its hour is folded, so a +// roll-up that has fallen behind costs disk rather than history. +// Traffic totals live in their own table precisely so history can be +// pruned freely. Deletes are per agent and chunked (a day per +// statement for the minute tier), releasing the write lock in between. +// Returns the number of rows deleted. +func (s *Store) Prune(keepDays int64) (int, error) { + now := time.Now().Unix() + detail := keepDays + if detail > DetailDays { + detail = DetailDays + } + minutesCutoff := now - detail*86400 + hoursCutoff := now - keepDays*86400 + + ids, err := s.agentIDs() + if err != nil { + return 0, err + } + pruned := 0 + type tableRule struct { + table string + cutoff int64 + guarded bool + } + rules := []tableRule{ + {"probe_metric", minutesCutoff, true}, + {"probe_ping_record", minutesCutoff, true}, + {"probe_metric_hour", hoursCutoff, false}, + {"probe_ping_hour", hoursCutoff, false}, + } + for _, id := range ids { + for _, rule := range rules { + for { + before := rule.cutoff + if rule.guarded { + // Guarded by the watermark: with none set, nothing + // is deletable (hub reads it as the minimum). + wm, ok, err := s.watermark() + if err != nil { + return pruned, err + } + if !ok { + before = math.MinInt64 + } else if wm < before { + before = wm + } + } + var first sql.NullInt64 + err := s.db.QueryRow( + `SELECT MIN(ts) FROM `+rule.table+` WHERE agent_id = ?`, id).Scan(&first) + if err != nil { + return pruned, err + } + if !first.Valid || first.Int64 >= before { + break + } + limit := before + if rule.guarded && first.Int64+86400 < limit { + limit = first.Int64 + 86400 + } + res, err := s.db.Exec( + `DELETE FROM `+rule.table+` WHERE agent_id = ? AND ts < ?`, id, limit) + if err != nil { + return pruned, err + } + n, _ := res.RowsAffected() + pruned += int(n) + if n == 0 { + break + } + } + } + } + return pruned, nil +} + +// agentIDs lists every agent with rows in any history table. +func (s *Store) agentIDs() ([]int64, error) { + rows, err := s.db.Query(` + SELECT agent_id FROM probe_metric + UNION SELECT agent_id FROM probe_ping_record + UNION SELECT agent_id FROM probe_metric_hour + UNION SELECT agent_id FROM probe_ping_hour + ORDER BY 1`) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} diff --git a/internal/web/service/probe/maintenance_test.go b/internal/web/service/probe/maintenance_test.go new file mode 100644 index 0000000..9cf5e19 --- /dev/null +++ b/internal/web/service/probe/maintenance_test.go @@ -0,0 +1,199 @@ +package probe + +import ( + "strconv" + "testing" + "time" +) + +func strconvFormat(n int64) string { return strconv.FormatInt(n, 10) } + +func seedMetric(t *testing.T, st *Store, agent, ts int64, cpu, cpuMax float64, mem, netRx, netRxMax, tcpC int64) { + t.Helper() + _, err := st.db.Exec(` + INSERT INTO probe_metric (agent_id, ts, cpu, mem_used, swap_used, disk_used, + net_rx, net_tx, tcp, udp, procs, net_rx_max, net_tx_max, cpu_max) + VALUES (?, ?, ?, ?, 0, 0, ?, 0, ?, 0, 0, ?, 0, ?)`, + agent, ts, cpu, mem, netRx, tcpC, netRxMax, cpuMax) + if err != nil { + t.Fatal(err) + } +} + +func seedPing(t *testing.T, st *Store, agent, ts, task, latency int64) { + t.Helper() + if _, err := st.db.Exec(`INSERT INTO probe_ping_record (agent_id, ts, task_id, latency) VALUES (?, ?, ?, ?)`, + agent, ts, task, latency); err != nil { + t.Fatal(err) + } +} + +// One closed hour folds into one hourly row per agent: means over the +// minute rows (byte columns truncated, counts rounded), peaks maxed. +func TestRollUpFoldsHour(t *testing.T) { + st := newTestStore(t) + now := time.Now().Unix() + hourA := hourFloor(now) - 3*3600 // closed and past the late grace + seedMetric(t, st, 1, hourA+60, 10, 15, 100, 100, 500, 1) + seedMetric(t, st, 1, hourA+120, 20, 25, 200, 100, 500, 2) + seedMetric(t, st, 1, hourA+180, 30, 35, 300, 100, 500, 4) + seedMetric(t, st, 2, hourA+60, 99, 99, 999, 9, 9, 9) // agent 2, same hour + // The most recent hour must not fold yet (late grace). + recent := hourFloor(now) - 3600 + seedMetric(t, st, 1, recent+60, 77, 77, 777, 7, 7, 7) + + folded, err := st.RollUp(now, 7) + if err != nil { + t.Fatal(err) + } + if folded == 0 { + t.Fatal("expected at least one folded hour") + } + var minutes, mem, netRx, netRxMax, tcpC int64 + var cpu, cpuMax float64 + err = st.db.QueryRow(` + SELECT minutes, cpu, cpu_max, mem_used, net_rx, net_rx_max, tcp + FROM probe_metric_hour WHERE agent_id = 1 AND ts = ?`, hourA). + Scan(&minutes, &cpu, &cpuMax, &mem, &netRx, &netRxMax, &tcpC) + if err != nil { + t.Fatal(err) + } + if minutes != 3 || cpu != 20 || cpuMax != 35 || mem != 200 || netRx != 100 || netRxMax != 500 || tcpC != 2 { + t.Fatalf("folded row wrong: minutes=%d cpu=%v cpuMax=%v mem=%d netRx=%d netRxMax=%d tcp=%d", + minutes, cpu, cpuMax, mem, netRx, netRxMax, tcpC) + } + var recentCount int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM probe_metric_hour WHERE ts = ?`, recent).Scan(&recentCount); err != nil { + t.Fatal(err) + } + if recentCount != 0 { + t.Fatal("recent hour must still be unfolded") + } + wm, ok, err := st.watermark() + if err != nil || !ok { + t.Fatalf("watermark: ok=%v err=%v", ok, err) + } + if wm < hourA+3600 || wm > recent+3600 { + t.Fatalf("watermark %d must sit past hour A (%d) and no later than past the recent hour (%d)", wm, hourA, recent) + } + // Folding again is a rewrite, not a duplicate: same values. + if _, err := st.RollUp(now, 7); err != nil { + t.Fatal(err) + } + var count int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM probe_metric_hour WHERE agent_id = 1 AND ts = ?`, hourA).Scan(&count); err != nil { + t.Fatal(err) + } + if count != 1 { + t.Fatalf("hour rows for A = %d, want 1", count) + } +} + +// Ping hours: answered/lost counts, the answered median, and lo/hi; +// an all-lost task keeps NULL latency fields. +func TestRollUpPingTallies(t *testing.T) { + st := newTestStore(t) + now := time.Now().Unix() + hourA := hourFloor(now) - 3*3600 + seedPing(t, st, 1, hourA+10, 7, 10) + seedPing(t, st, 1, hourA+20, 7, 20) + seedPing(t, st, 1, hourA+30, 7, -1) + seedPing(t, st, 1, hourA+40, 7, 40) + seedPing(t, st, 1, hourA+10, 8, -1) + seedPing(t, st, 1, hourA+20, 8, -1) + if _, err := st.RollUp(now, 7); err != nil { + t.Fatal(err) + } + var answered, lost, latency, lo, hi int64 + err := st.db.QueryRow(` + SELECT answered, lost, latency, lo, hi FROM probe_ping_hour + WHERE agent_id = 1 AND ts = ? AND task_id = 7`, hourA).Scan(&answered, &lost, &latency, &lo, &hi) + if err != nil { + t.Fatal(err) + } + if answered != 3 || lost != 1 || latency != 20 || lo != 10 || hi != 40 { + t.Fatalf("tally wrong: answered=%d lost=%d latency=%d lo=%d hi=%d", answered, lost, latency, lo, hi) + } + var latNull bool + err = st.db.QueryRow(` + SELECT latency IS NULL FROM probe_ping_hour + WHERE agent_id = 1 AND ts = ? AND task_id = 8`, hourA).Scan(&latNull) + if err != nil { + t.Fatal(err) + } + if !latNull { + t.Fatal("all-lost task must keep NULL latency") + } +} + +// Prune respects the watermark: minute rows past the detail window +// survive until their hour is folded; hourly rows past the retention +// window go; the traffic ledger is never pruned. +func TestPruneBoundaries(t *testing.T) { + st := newTestStore(t) + now := time.Now() + nowU := now.Unix() + oldMinute := hourFloor(nowU) - 8*86400 // past the 7-day detail window + recentMinute := hourFloor(nowU) - 3600 + seedMetric(t, st, 1, oldMinute+60, 1, 1, 1, 1, 1, 1) + seedMetric(t, st, 1, recentMinute+60, 1, 1, 1, 1, 1, 1) + // Watermark only covers up to oldMinute: the old row is NOT yet + // deletable even though it is past the window. + if _, err := st.db.Exec(`INSERT OR REPLACE INTO probe_setting (key, value) VALUES (?, ?)`, + rolledKey, strconvFormat(oldMinute)); err != nil { + t.Fatal(err) + } + if _, err := st.Prune(7); err != nil { + t.Fatal(err) + } + var n int + if err := st.db.QueryRow(`SELECT COUNT(*) FROM probe_metric WHERE agent_id = 1`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 2 { + t.Fatalf("unfolded old row must survive prune, rows=%d", n) + } + // Roll forward past the old row's hour, making the watermark cover + // it; prune must now drop it (the recent row stays). + if _, err := st.db.Exec(`UPDATE probe_setting SET value = ? WHERE key = ?`, + strconvFormat(oldMinute+3600), rolledKey); err != nil { + t.Fatal(err) + } + if _, err := st.Prune(7); err != nil { + t.Fatal(err) + } + if err := st.db.QueryRow(`SELECT COUNT(*) FROM probe_metric WHERE agent_id = 1`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("folded old row must be pruned, rows=%d", n) + } + // Hourly rows past retention go; within stay. + if _, err := st.db.Exec(`INSERT INTO probe_metric_hour (agent_id, ts, minutes, cpu, cpu_max, mem_used, swap_used, disk_used, net_rx, net_tx, net_rx_max, net_tx_max, tcp, udp, procs) VALUES (1, ?, 60, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)`, + hourFloor(nowU)-8*86400); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`INSERT INTO probe_metric_hour (agent_id, ts, minutes, cpu, cpu_max, mem_used, swap_used, disk_used, net_rx, net_tx, net_rx_max, net_tx_max, tcp, udp, procs) VALUES (1, ?, 60, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)`, + hourFloor(nowU)-6*86400); err != nil { + t.Fatal(err) + } + if _, err := st.Prune(7); err != nil { + t.Fatal(err) + } + if err := st.db.QueryRow(`SELECT COUNT(*) FROM probe_metric_hour WHERE agent_id = 1`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("hour rows=%d, want 1 (only the in-window one)", n) + } + // Traffic bookkeeping is never pruned. + book(t, st, 1, "k/fp", 100, 100, now.Add(-30*24*time.Hour), nil, 0) + book(t, st, 1, "k/fp", 300, 300, now.Add(-30*24*time.Hour).Add(time.Minute), nil, 0) + if _, err := st.Prune(7); err != nil { + t.Fatal(err) + } + tr, ok, _ := st.TrafficFor(1) + if !ok || tr.TotalRx != 200 { + t.Fatalf("traffic must survive prune: %+v ok=%v", tr, ok) + } +} diff --git a/internal/web/service/probe/probe.go b/internal/web/service/probe/probe.go new file mode 100644 index 0000000..abf0c67 --- /dev/null +++ b/internal/web/service/probe/probe.go @@ -0,0 +1,85 @@ +package probe + +import ( + "strconv" + "strings" + "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/logger" +) + +// RetentionDays reads the history retention window (days) from the +// panel settings (key retention_days, as in hub). Default 7 — the +// window the user settled on — clamped to [1, 365] (hub's cap). +func RetentionDays() int64 { + var row model.Setting + err := database.GetDB().Where("key = ?", "retention_days").First(&row).Error + if err != nil { + return DetailDays + } + n, perr := strconv.ParseInt(strings.TrimSpace(row.Value), 10, 64) + if perr != nil { + return DetailDays + } + if n < 1 { + return 1 + } + if n > 365 { + return 365 + } + return n +} + +// Record files one heartbeat's probe sample into the default store. +// prevLastSeenMs is the agent row's last-seen stamp from before this +// heartbeat (milliseconds, as stored on agent_nodes). Best-effort: any +// failure is logged and swallowed so a heartbeat never fails over +// history. +func Record(agentID int64, smp Sample, prevLastSeenMs int64) { + st := Default() + if st == nil { + return + } + if err := st.Record(agentID, smp, time.Now(), prevLastSeenMs/1000); err != nil { + logger.Warningf("probe: record agent %d: %v", agentID, err) + } +} + +// Maintain runs one housekeeping pass against the default store: +// roll_up first, then prune — prune runs even when roll_up fails, as +// it keeps minute rows until their hour is folded and skipping it +// would leave the database growing (hub housekeeping discipline). +func Maintain() { + st := Default() + if st == nil { + return + } + keep := RetentionDays() + if folded, err := st.RollUp(time.Now().Unix(), keep); err != nil { + logger.Warningf("probe: roll_up failed: %v", err) + } else if folded > 1 { + logger.Infof("probe: folded %d hours of history into the hourly tier", folded) + } + if _, err := st.Prune(keep); err != nil { + logger.Warningf("probe: prune failed: %v", err) + } +} + +// BookHeld books every node's held, not-yet-booked reading on the +// default store; called as the master shuts down. +func BookHeld() { + if st := Default(); st != nil { + st.BookHeld() + } +} + +// TrafficSnapshot is the B3 read port: every agent's gated ledger. +func TrafficSnapshot() (map[int64]Traffic, error) { + st := Default() + if st == nil { + return map[int64]Traffic{}, nil + } + return st.AllTraffic() +} diff --git a/internal/web/service/probe/store.go b/internal/web/service/probe/store.go new file mode 100644 index 0000000..1e44acb --- /dev/null +++ b/internal/web/service/probe/store.go @@ -0,0 +1,231 @@ +// Package probe is the master-side history store of item 7 (probe merge): +// it persists the per-minute metric rows and the whole-machine traffic +// ledger reported by agents (B1) into a dedicated SQLite file, probe.db, +// beside the panel database. The model, the booking rules and the +// roll-up/prune discipline are a port of monitor-probe's hub (src/db.rs, +// src/agent_ws.rs); see ITEM7-SCOPE.md §3.2/§3.3. +// +// probe.db is deliberately a separate file from x-ui.db: history writes, +// roll-ups and prunes then never contend for the panel database's write +// lock (ITEM7-SCOPE §7 risk 4). Everything here is best-effort by design: +// a broken or missing probe.db must never fail a heartbeat or a panel +// request, so the facade entry points log and swallow errors. +package probe + +import ( + "database/sql" + "fmt" + "os" + "path/filepath" + "sync" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/config" + "github.com/mhsanaei/3x-ui/v3/internal/logger" + + _ "github.com/mattn/go-sqlite3" +) + +// DetailDays is how long minute rows are kept before they may be pruned; +// it only ever bounds the minute tier — the hourly tier and the traffic +// ledger live by the retention window itself. Mirrors hub DETAIL_DAYS. +const DetailDays = 7 + +// lateSeconds is the grace a just-closed hour gets before roll_up folds +// it, so rows landing slightly late still make it into their hour. +// Mirrors hub LATE (3600 + 60). +const lateSeconds = 3600 + 60 + +// rolledKey is the probe_setting key holding the roll-up watermark: the +// start (unix seconds) of the next hour not yet folded into the hourly +// tier. Mirrors hub ROLLED ("history_rolled"). +const rolledKey = "history_rolled" + +const schema = ` +CREATE TABLE IF NOT EXISTS probe_setting ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); + +-- Per-node probe metadata (ITEM7-SCOPE §5.2). B2 creates the table and +-- reads traffic_reset_day; the editing surface lands in B3. +CREATE TABLE IF NOT EXISTS probe_meta ( + agent_id INTEGER PRIMARY KEY, + price REAL NOT NULL DEFAULT 0, + currency TEXT NOT NULL DEFAULT 'USD', + billing_cycle TEXT NOT NULL DEFAULT 'monthly', + expires_at TEXT, + traffic_limit INTEGER NOT NULL DEFAULT 0, + traffic_mode TEXT NOT NULL DEFAULT 'sum', + traffic_reset_day INTEGER NOT NULL DEFAULT 1, + public INTEGER NOT NULL DEFAULT 1, + sort INTEGER NOT NULL DEFAULT 0, + group_name TEXT NOT NULL DEFAULT '', + public_remark TEXT NOT NULL DEFAULT '', + notify INTEGER NOT NULL DEFAULT 0, + country_pin TEXT NOT NULL DEFAULT '', + ipv4_pin TEXT NOT NULL DEFAULT '', + ipv6_pin TEXT NOT NULL DEFAULT '' +); + +-- Monotonic byte counters that survive agent reboots and master +-- restarts; the whole-machine ledger (hub "traffic"). One row per agent. +CREATE TABLE IF NOT EXISTS probe_traffic ( + agent_id INTEGER PRIMARY KEY, + boot_id TEXT NOT NULL DEFAULT '', + last_rx INTEGER NOT NULL DEFAULT 0, + last_tx INTEGER NOT NULL DEFAULT 0, + total_rx INTEGER NOT NULL DEFAULT 0, + total_tx INTEGER NOT NULL DEFAULT 0, + month_rx INTEGER NOT NULL DEFAULT 0, + month_tx INTEGER NOT NULL DEFAULT 0, + month_start TEXT NOT NULL DEFAULT '', + day_rx INTEGER NOT NULL DEFAULT 0, + day_tx INTEGER NOT NULL DEFAULT 0, + day_start TEXT NOT NULL DEFAULT '' +); + +-- One row per agent per minute (hub "metric"). net_rx/net_tx are bytes +-- per second derived from the kernel counters, the *_max columns are the +-- minute's peaks of the agent-reported figures. +CREATE TABLE IF NOT EXISTS probe_metric ( + agent_id INTEGER NOT NULL, + ts INTEGER NOT NULL, + cpu REAL NOT NULL, + mem_used INTEGER NOT NULL, + swap_used INTEGER NOT NULL, + disk_used INTEGER NOT NULL, + net_rx INTEGER NOT NULL, + net_tx INTEGER NOT NULL, + tcp INTEGER NOT NULL, + udp INTEGER NOT NULL, + procs INTEGER NOT NULL, + net_rx_max INTEGER NOT NULL DEFAULT 0, + net_tx_max INTEGER NOT NULL DEFAULT 0, + cpu_max REAL NOT NULL DEFAULT 0, + PRIMARY KEY (agent_id, ts) +) WITHOUT ROWID; + +-- The hourly tier (hub "metric_hour"): every metric column plus the +-- count of minute rows folded in, which weights later merges. +CREATE TABLE IF NOT EXISTS probe_metric_hour ( + agent_id INTEGER NOT NULL, + ts INTEGER NOT NULL, + minutes INTEGER NOT NULL, + cpu REAL NOT NULL, + cpu_max REAL NOT NULL, + mem_used INTEGER NOT NULL, + swap_used INTEGER NOT NULL, + disk_used INTEGER NOT NULL, + net_rx INTEGER NOT NULL, + net_tx INTEGER NOT NULL, + net_rx_max INTEGER NOT NULL, + net_tx_max INTEGER NOT NULL, + tcp INTEGER NOT NULL, + udp INTEGER NOT NULL, + procs INTEGER NOT NULL, + PRIMARY KEY (agent_id, ts) +) WITHOUT ROWID; + +-- Probe results (hub "ping_record" / "ping_hour"). B2 creates the +-- tables so roll_up/prune cover them uniformly; B4 writes the rows. +-- latency < 0 is a lost probe; the hourly latency is the median of the +-- hour's answers and NULL when none arrived. +CREATE TABLE IF NOT EXISTS probe_ping_record ( + agent_id INTEGER NOT NULL, + ts INTEGER NOT NULL, + task_id INTEGER NOT NULL, + latency INTEGER NOT NULL, + PRIMARY KEY (agent_id, ts, task_id) +) WITHOUT ROWID; + +CREATE TABLE IF NOT EXISTS probe_ping_hour ( + agent_id INTEGER NOT NULL, + ts INTEGER NOT NULL, + task_id INTEGER NOT NULL, + answered INTEGER NOT NULL, + lost INTEGER NOT NULL, + latency INTEGER, + lo INTEGER, + hi INTEGER, + PRIMARY KEY (agent_id, ts, task_id) +) WITHOUT ROWID; +` + +// Store is one open probe.db. The zero value is not usable; call Open. +// A Store is safe for concurrent use: the single connection serializes +// writes, and the ingest state is mutex-guarded. +type Store struct { + db *sql.DB + + mu sync.Mutex // guards nodes + nodes map[int64]*nodeState +} + +// Open creates (if needed) and initializes the probe database at path. +func Open(path string) (*Store, error) { + if dir := filepath.Dir(path); dir != "" { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, err + } + } + dsn := path + "?_journal_mode=WAL&_busy_timeout=10000&_synchronous=NORMAL&_txlock=immediate" + db, err := sql.Open("sqlite3", dsn) + if err != nil { + return nil, err + } + // One connection: writes serialize on it (hub's single write + // connection), and WAL still lets the read side (B3) proceed. + db.SetMaxOpenConns(1) + if _, err := db.Exec(schema); err != nil { + db.Close() + return nil, fmt.Errorf("probe schema: %w", err) + } + return &Store{db: db, nodes: make(map[int64]*nodeState)}, nil +} + +// Close releases the database handle. +func (s *Store) Close() error { + if s == nil || s.db == nil { + return nil + } + return s.db.Close() +} + +// DB exposes the raw handle for the read side (B3) and tests. +func (s *Store) DB() *sql.DB { return s.db } + +// --- default store (lazy, failure-tolerant) --- + +var def struct { + mu sync.Mutex + store *Store + retryAt time.Time +} + +// DefaultPath is where probe.db lives: beside the panel database. +func DefaultPath() string { + return filepath.Join(config.GetDBFolderPath(), "probe.db") +} + +// Default returns the process-wide store, opening it on first use. A +// failed open is retried at most once a minute and reported as nil, so +// callers degrade to "no history" instead of failing. +func Default() *Store { + def.mu.Lock() + defer def.mu.Unlock() + if def.store != nil { + return def.store + } + if time.Now().Before(def.retryAt) { + return nil + } + st, err := Open(DefaultPath()) + if err != nil { + logger.Warningf("probe: open %s failed (history disabled until retry): %v", DefaultPath(), err) + def.retryAt = time.Now().Add(time.Minute) + return nil + } + def.store = st + return st +} diff --git a/internal/web/service/probe/traffic.go b/internal/web/service/probe/traffic.go new file mode 100644 index 0000000..3beee41 --- /dev/null +++ b/internal/web/service/probe/traffic.go @@ -0,0 +1,348 @@ +package probe + +import ( + "database/sql" + "math" + "strings" + "time" +) + +// Traffic is one agent's whole-machine ledger row (hub Traffic): the +// running totals booked from the kernel counters of its counted +// interfaces. This is the machine-wide account (close to what providers +// bill), separate from the panel's per-inbound/per-client xray traffic. +type Traffic struct { + TotalRx int64 `json:"totalRx"` + TotalTx int64 `json:"totalTx"` + MonthRx int64 `json:"monthRx"` + MonthTx int64 `json:"monthTx"` + MonthStart string `json:"monthStart"` + DayRx int64 `json:"dayRx"` + DayTx int64 `json:"dayTx"` + DayStart string `json:"-"` // read-side gating only, as in hub +} + +// MonthUsed is this period's usage as the node's plan meters it +// (hub month_used): "up" counts tx only, "down" rx only, "max" the +// larger direction, anything else (incl. "sum") both directions. +func (t Traffic) MonthUsed(mode string) int64 { + switch mode { + case "up": + return t.MonthTx + case "down": + return t.MonthRx + case "max": + if t.MonthRx > t.MonthTx { + return t.MonthRx + } + return t.MonthTx + default: + return satAdd(t.MonthRx, t.MonthTx) + } +} + +func satAdd(a, b int64) int64 { + if b > 0 && a > math.MaxInt64-b { + return math.MaxInt64 + } + if b < 0 && a < math.MinInt64-b { + return math.MinInt64 + } + return a + b +} + +// PeriodStart is the start of the billing period containing the given +// date for a reset day of month (hub period_start). A reset day past +// the end of a short month lands on that month's last day. +func PeriodStart(today time.Time, resetDay int64) time.Time { + if resetDay < 1 { + resetDay = 1 + } + if resetDay > 31 { + resetDay = 31 + } + clamped := func(y int, m time.Month) time.Time { + last := time.Date(y, m+1, 1, 0, 0, 0, 0, time.Local).AddDate(0, 0, -1).Day() + d := int(resetDay) + if d > last { + d = last + } + return time.Date(y, m, d, 0, 0, 0, 0, time.Local) + } + today = time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, time.Local) + this := clamped(today.Year(), today.Month()) + if !today.Before(this) { + return this + } + if today.Month() == time.January { + return clamped(today.Year()-1, time.December) + } + return clamped(today.Year(), today.Month()-1) +} + +const dateLayout = "2006-01-02" + +func dateOnly(t time.Time) time.Time { + return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.Local) +} + +// ensureRows lazily creates the agent's meta + traffic rows so booking +// and reading never special-case a first report. +func (s *Store) ensureRows(agentID int64) error { + if _, err := s.db.Exec(`INSERT OR IGNORE INTO probe_meta (agent_id) VALUES (?)`, agentID); err != nil { + return err + } + _, err := s.db.Exec(`INSERT OR IGNORE INTO probe_traffic (agent_id) VALUES (?)`, agentID) + return err +} + +// ResetDay reads the agent's billing reset day from probe_meta +// (default 1 when unset, hub's node default). +func (s *Store) ResetDay(agentID int64) (int64, error) { + var day int64 + err := s.db.QueryRow(`SELECT traffic_reset_day FROM probe_meta WHERE agent_id = ?`, agentID).Scan(&day) + if err == sql.ErrNoRows { + return 1, nil + } + if err != nil { + return 1, err + } + if day < 1 { + return 1, nil + } + if day > 31 { + return 31, nil + } + return day, nil +} + +// accumulate books one reading of an agent's raw kernel counters into +// its running totals — a faithful port of hub db.rs accumulate(). +// +// Only bytes the master observed a counter climb through are booked. +// Without a baseline under this exact boot there is nothing to subtract +// from: a first reading, a counter that shrank under the same boot, and +// a changed boot_id all book a zero delta and merely re-align the +// baseline (booking the bare reading would charge the machine's whole +// history again). The one exception is a machine that booted after the +// master last heard from it (booted > prevLastSeenSec): its counters +// started from zero while unobserved, so the reading is exactly the +// bytes since boot and is booked whole — a reboot during a master +// outage would otherwise lose the whole outage. +// +// at dates the reading for the day and billing period (hub local +// calendar); month/day boundaries never move a stored stamp backwards, +// so a panel correction stamped with the current period is not undone +// by a held-back reading. +func (s *Store) accumulate(agentID int64, bootID string, rx, tx int64, at time.Time, booted *int64, prevLastSeenSec int64) (Traffic, error) { + if err := s.ensureRows(agentID); err != nil { + return Traffic{}, err + } + resetDay, err := s.ResetDay(agentID) + if err != nil { + return Traffic{}, err + } + var ( + prevBoot, monthStart, dayStart string + lastRx, lastTx int64 + totalRx, totalTx, monthRx, monthTx, dayRx, dayTx int64 + ) + err = s.db.QueryRow(` + SELECT boot_id, last_rx, last_tx, total_rx, total_tx, month_rx, month_tx, + month_start, day_rx, day_tx, day_start + FROM probe_traffic WHERE agent_id = ?`, agentID). + Scan(&prevBoot, &lastRx, &lastTx, &totalRx, &totalTx, &monthRx, &monthTx, + &monthStart, &dayRx, &dayTx, &dayStart) + if err != nil { + return Traffic{}, err + } + + kernelPart := func(b string) string { + if i := strings.IndexByte(b, '/'); i >= 0 { + return b[:i] + } + return b + } + unseenBoot := prevBoot != "" && + kernelPart(prevBoot) != kernelPart(bootID) && + booted != nil && prevLastSeenSec > 0 && *booted > prevLastSeenSec + + var dRx, dTx int64 + switch { + case unseenBoot: + dRx, dTx = rx, tx + case prevBoot == "" || prevBoot != bootID: + dRx, dTx = 0, 0 + default: + if rx > lastRx { + dRx = rx - lastRx + } + if tx > lastTx { + dTx = tx - lastTx + } + } + + // Saturating adds: a total near MaxInt64 must not wrap negative. + totalRx = satAdd(totalRx, dRx) + totalTx = satAdd(totalTx, dTx) + monthRx = satAdd(monthRx, dRx) + monthTx = satAdd(monthTx, dTx) + dayRx = satAdd(dayRx, dRx) + dayTx = satAdd(dayTx, dTx) + + date := dateOnly(at) + today := dateOnly(time.Now()) + dated := func(stored string) time.Time { + if stamp, err := time.ParseInLocation(dateLayout, stored, time.Local); err == nil && !stamp.After(today) { + if stamp.After(date) { + return stamp + } + return date + } + return date + } + period := PeriodStart(dated(monthStart), resetDay) + if monthStart != period.Format(dateLayout) { + // A new billing period restarts the month counter, not the total. + monthRx, monthTx = dRx, dTx + } + day := dated(dayStart) + if dayStart != day.Format(dateLayout) { + dayRx, dayTx = dRx, dTx + } + + _, err = s.db.Exec(` + UPDATE probe_traffic SET boot_id = ?, last_rx = ?, last_tx = ?, + total_rx = ?, total_tx = ?, month_rx = ?, month_tx = ?, month_start = ?, + day_rx = ?, day_tx = ?, day_start = ? + WHERE agent_id = ?`, + bootID, rx, tx, totalRx, totalTx, monthRx, monthTx, period.Format(dateLayout), + dayRx, dayTx, day.Format(dateLayout), agentID) + if err != nil { + return Traffic{}, err + } + return Traffic{ + TotalRx: totalRx, TotalTx: totalTx, + MonthRx: monthRx, MonthTx: monthTx, MonthStart: period.Format(dateLayout), + DayRx: dayRx, DayTx: dayTx, DayStart: day.Format(dateLayout), + }, nil +} + +// SetTraffic corrects a ledger row from the panel side (hub +// set_traffic): month figures are re-stamped with the current period so +// the correction is not read back as belonging to an old period and +// discarded by the next booking. Nil fields keep their stored values. +// False when the agent has no meta row yet. +func (s *Store) SetTraffic(agentID int64, totalRx, totalTx, monthRx, monthTx *int64) (bool, error) { + resetDay, err := s.ResetDay(agentID) + if err != nil { + return false, err + } + var exists int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM probe_meta WHERE agent_id = ?`, agentID).Scan(&exists); err != nil { + return false, err + } + if exists == 0 { + return false, nil + } + if err := s.ensureRows(agentID); err != nil { + return false, err + } + period := PeriodStart(dateOnly(time.Now()), resetDay).Format(dateLayout) + // COALESCE keeps stored values for nil fields; month figures only + // move when the stored stamp already is the current period, and the + // stamp always advances to the current period (hub semantics). + _, err = s.db.Exec(` + UPDATE probe_traffic SET + total_rx = COALESCE(?, total_rx), total_tx = COALESCE(?, total_tx), + month_rx = COALESCE(?, CASE WHEN month_start = ? THEN month_rx ELSE 0 END), + month_tx = COALESCE(?, CASE WHEN month_start = ? THEN month_tx ELSE 0 END), + month_start = ? + WHERE agent_id = ?`, + ptrOrNil(totalRx), ptrOrNil(totalTx), ptrOrNil(monthRx), period, + ptrOrNil(monthTx), period, period, agentID) + if err != nil { + return false, err + } + return true, nil +} + +func ptrOrNil(v *int64) any { + if v == nil { + return nil + } + return *v +} + +// gate applies the read-side period gating of hub all_traffic: the +// period counters restart lazily in accumulate on the node's next +// report, so a node offline since before a boundary still holds the +// previous period's bytes on disk; readers see them as zero unless the +// stored stamp is the current period. +func gate(t Traffic, dayStart string, resetDay int64, now time.Time) Traffic { + period := PeriodStart(dateOnly(now), resetDay).Format(dateLayout) + if t.MonthStart != period { + t.MonthRx, t.MonthTx = 0, 0 + } + if dayStart != dateOnly(now).Format(dateLayout) { + t.DayRx, t.DayTx = 0, 0 + } + return t +} + +// TrafficFor returns one agent's ledger with read-side period gating — +// the read port B3's /api/nodes injects total/month figures from. +func (s *Store) TrafficFor(agentID int64) (Traffic, bool, error) { + var t Traffic + var dayStart string + err := s.db.QueryRow(` + SELECT total_rx, total_tx, month_rx, month_tx, month_start, day_rx, day_tx, day_start + FROM probe_traffic WHERE agent_id = ?`, agentID). + Scan(&t.TotalRx, &t.TotalTx, &t.MonthRx, &t.MonthTx, &t.MonthStart, &t.DayRx, &t.DayTx, &dayStart) + if err == sql.ErrNoRows { + return Traffic{}, false, nil + } + if err != nil { + return Traffic{}, false, err + } + resetDay, err := s.ResetDay(agentID) + if err != nil { + return Traffic{}, false, err + } + return gate(t, dayStart, resetDay, time.Now()), true, nil +} + +// AllTraffic returns every agent's ledger in one query, gated as in +// TrafficFor — the bulk read port for B3's node list. +func (s *Store) AllTraffic() (map[int64]Traffic, error) { + rows, err := s.db.Query(` + SELECT t.agent_id, t.total_rx, t.total_tx, t.month_rx, t.month_tx, t.month_start, + t.day_rx, t.day_tx, t.day_start, COALESCE(m.traffic_reset_day, 1) + FROM probe_traffic t LEFT JOIN probe_meta m ON m.agent_id = t.agent_id`) + if err != nil { + return nil, err + } + defer rows.Close() + out := make(map[int64]Traffic) + now := time.Now() + for rows.Next() { + var id, resetDay int64 + var t Traffic + var dayStart string + if err := rows.Scan(&id, &t.TotalRx, &t.TotalTx, &t.MonthRx, &t.MonthTx, &t.MonthStart, + &t.DayRx, &t.DayTx, &dayStart, &resetDay); err != nil { + return nil, err + } + out[id] = gate(t, dayStart, resetDay, now) + } + return out, rows.Err() +} + +// parseDay parses a stored "2006-01-02" stamp (used by tests). +func parseDay(s string) (time.Time, bool) { + t, err := time.ParseInLocation(dateLayout, s, time.Local) + if err != nil { + return time.Time{}, false + } + return t, true +} diff --git a/internal/web/service/probe/traffic_test.go b/internal/web/service/probe/traffic_test.go new file mode 100644 index 0000000..40d3a78 --- /dev/null +++ b/internal/web/service/probe/traffic_test.go @@ -0,0 +1,270 @@ +package probe + +import ( + "path/filepath" + "testing" + "time" +) + +func newTestStore(t *testing.T) *Store { + t.Helper() + st, err := Open(filepath.Join(t.TempDir(), "probe.db")) + if err != nil { + t.Fatalf("open: %v", err) + } + t.Cleanup(func() { st.Close() }) + return st +} + +func book(t *testing.T, st *Store, agent int64, boot string, rx, tx int64, at time.Time, booted *int64, lastSeen int64) Traffic { + t.Helper() + tr, err := st.accumulate(agent, boot, rx, tx, at, booted, lastSeen) + if err != nil { + t.Fatalf("accumulate: %v", err) + } + return tr +} + +func i64p(v int64) *int64 { return &v } + +func TestPeriodStart(t *testing.T) { + d := func(y int, m time.Month, day int) time.Time { + return time.Date(y, m, day, 0, 0, 0, 0, time.Local) + } + cases := []struct { + today time.Time + reset int64 + want time.Time + }{ + {d(2026, 10, 23), 1, d(2026, 10, 1)}, + {d(2026, 10, 23), 23, d(2026, 10, 23)}, + {d(2026, 10, 23), 24, d(2026, 9, 24)}, + {d(2026, 2, 15), 31, d(2026, 1, 31)}, // Jan has 31 + {d(2026, 2, 28), 31, d(2026, 2, 28)}, // short month: reset lands on last day + {d(2026, 3, 1), 31, d(2026, 2, 28)}, // Mar 1 with reset 31 is still in Feb's period + {d(2026, 3, 31), 31, d(2026, 3, 31)}, + {d(2026, 1, 1), 15, d(2025, 12, 15)}, // wraparound into previous year + {d(2026, 10, 23), 0, d(2026, 10, 1)}, // clamp low + {d(2026, 10, 23), 99, d(2026, 9, 30)}, // clamp high: Sep has 30 + } + for _, c := range cases { + if got := PeriodStart(c.today, c.reset); !got.Equal(c.want) { + t.Errorf("PeriodStart(%v, %d) = %v, want %v", c.today, c.reset, got, c.want) + } + } +} + +func TestAccumulateFirstReadingAlignsBaseline(t *testing.T) { + st := newTestStore(t) + now := time.Now() + tr := book(t, st, 1, "k1/fp", 1_000_000, 2_000_000, now, nil, 0) + if tr.TotalRx != 0 || tr.TotalTx != 0 || tr.MonthRx != 0 || tr.DayRx != 0 { + t.Fatalf("first reading must book 0, got %+v", tr) + } + // Same boot, counters climb: the delta books. + tr = book(t, st, 1, "k1/fp", 1_000_600, 2_000_900, now.Add(time.Minute), nil, now.Unix()) + if tr.TotalRx != 600 || tr.TotalTx != 900 || tr.MonthRx != 600 || tr.DayTx != 900 { + t.Fatalf("delta not booked: %+v", tr) + } +} + +func TestAccumulateCounterShrinkSameBoot(t *testing.T) { + st := newTestStore(t) + now := time.Now() + book(t, st, 1, "k1/fp", 5000, 5000, now, nil, 0) + book(t, st, 1, "k1/fp", 6000, 6000, now.Add(time.Minute), nil, now.Unix()) + // A counted interface vanishes: the sum shrinks. Booking it again + // would double-count the remainder, so the delta is 0 and the + // baseline re-aligns. + tr := book(t, st, 1, "k1/fp", 2000, 2000, now.Add(2*time.Minute), nil, now.Unix()) + if tr.TotalRx != 1000 { + t.Fatalf("shrink must book 0, total=%d", tr.TotalRx) + } + tr = book(t, st, 1, "k1/fp", 2500, 2600, now.Add(3*time.Minute), nil, now.Unix()) + if tr.TotalRx != 1500 || tr.TotalTx != 1600 { + t.Fatalf("post-shrink delta wrong: %+v", tr) + } +} + +func TestAccumulateBootChangeZeroDelta(t *testing.T) { + st := newTestStore(t) + now := time.Now() + book(t, st, 1, "k1/fpA", 1000, 1000, now, nil, 0) + book(t, st, 1, "k1/fpA", 2000, 2000, now.Add(time.Minute), nil, now.Unix()) + // Same kernel, different interface digest (a counted iface joined): + // epoch changed, delta must be 0 even though booted looks recent. + booted := now.Add(30 * time.Second).Unix() + tr := book(t, st, 1, "k1/fpB", 9000, 9000, now.Add(2*time.Minute), &booted, now.Add(time.Minute).Unix()) + if tr.TotalRx != 1000 { + t.Fatalf("epoch change must book 0, total=%d", tr.TotalRx) + } +} + +func TestAccumulateUnseenBootBooksWholeReading(t *testing.T) { + st := newTestStore(t) + t0 := time.Now().Add(-2 * time.Hour) + book(t, st, 1, "k1/fp", 500, 500, t0, nil, 0) + tr := book(t, st, 1, "k1/fp", 800, 900, t0.Add(time.Minute), nil, t0.Unix()) + if tr.TotalRx != 300 { + t.Fatalf("setup delta wrong: %+v", tr) + } + // The machine reboots while the master is not hearing from it: + // new kernel boot, booted after the master's last-seen stamp, so + // the whole counters-since-boot book. + booted := t0.Add(30 * time.Minute).Unix() + tr = book(t, st, 1, "k2/fp", 200, 100, t0.Add(31*time.Minute), &booted, t0.Add(time.Minute).Unix()) + if tr.TotalRx != 500 || tr.TotalTx != 500 { + t.Fatalf("unseen boot must book whole reading: %+v", tr) + } + // And it continues from the new baseline. + tr = book(t, st, 1, "k2/fp", 350, 100, t0.Add(32*time.Minute), &booted, t0.Add(31*time.Minute).Unix()) + if tr.TotalRx != 650 { + t.Fatalf("post-reboot delta wrong: %+v", tr) + } +} + +func TestAccumulateBootChangeSeenRecentlyBooksZero(t *testing.T) { + st := newTestStore(t) + t0 := time.Now().Add(-time.Hour) + book(t, st, 1, "k1/fp", 1000, 1000, t0, nil, 0) + book(t, st, 1, "k1/fp", 2000, 2000, t0.Add(time.Minute), nil, t0.Unix()) + // A different kernel boot that started BEFORE the master last saw + // the node (e.g. a second machine sharing the token): not the + // unseen-boot exception; books 0. + booted := t0.Add(30 * time.Second).Unix() // < lastSeen t0+60s + tr := book(t, st, 1, "k2/fp", 5000, 5000, t0.Add(2*time.Minute), &booted, t0.Add(time.Minute).Unix()) + if tr.TotalRx != 1000 { + t.Fatalf("seen boot change must book 0, total=%d", tr.TotalRx) + } +} + +func TestAccumulateCrossDay(t *testing.T) { + st := newTestStore(t) + day1 := dateOnly(time.Now()).Add(-48 * time.Hour) // midnight, two days ago + r1 := day1.Add(23*time.Hour + 58*time.Minute) // day D 23:58 + r2 := day1.Add(23*time.Hour + 59*time.Minute) // day D 23:59 + r3 := day1.Add(24*time.Hour + time.Minute) // day D+1 00:01 + book(t, st, 1, "k/fp", 1000, 1000, r1, nil, 0) + tr := book(t, st, 1, "k/fp", 1100, 1100, r2, nil, r1.Unix()) + if tr.DayRx != 100 { + t.Fatalf("day1 day_rx=%d, want 100", tr.DayRx) + } + tr = book(t, st, 1, "k/fp", 1150, 1150, r3, nil, r2.Unix()) + if tr.DayRx != 50 || tr.TotalRx != 150 || tr.MonthRx != 150 { + t.Fatalf("cross-day split wrong: %+v", tr) + } + if tr.DayStart != dateOnly(r3).Format(dateLayout) { + t.Fatalf("day stamp=%s", tr.DayStart) + } +} + +func TestAccumulateCrossPeriodWithResetDay(t *testing.T) { + st := newTestStore(t) + // reset day 15 on the meta row. + if err := st.ensureRows(7); err != nil { + t.Fatal(err) + } + if _, err := st.db.Exec(`UPDATE probe_meta SET traffic_reset_day = 15 WHERE agent_id = 7`); err != nil { + t.Fatal(err) + } + now := time.Now() + // Find a date in the current period and one in the next, both <= now + // when possible: use period arithmetic from PeriodStart itself. + periodStart := PeriodStart(dateOnly(now), 15) + inPeriod := periodStart.Add(24 * time.Hour) + if inPeriod.After(now) { + inPeriod = periodStart + } + book(t, st, 7, "k/fp", 1000, 1000, inPeriod, nil, 0) + tr := book(t, st, 7, "k/fp", 1300, 1300, inPeriod.Add(time.Minute), nil, inPeriod.Unix()) + if tr.MonthRx != 300 { + t.Fatalf("month before boundary=%d, want 300", tr.MonthRx) + } + nextPeriod := PeriodStart(periodStart.AddDate(0, 2, 0), 15) + at := nextPeriod.Add(time.Minute) + tr = book(t, st, 7, "k/fp", 1500, 1500, at, nil, inPeriod.Unix()) + // Delta 200 lands in the new period; the month counter restarts at + // the delta alone, the total keeps everything. (When the booking is + // future-dated relative to a held-back stamp, accumulate's dated() + // guard applies; both stamps here precede the booking date.) + if tr.TotalRx != 500 { + t.Fatalf("total=%d, want 500", tr.TotalRx) + } + if tr.MonthRx != 200 { + t.Fatalf("month after boundary=%d, want 200", tr.MonthRx) + } + if tr.MonthStart != nextPeriod.Format(dateLayout) { + t.Fatalf("month stamp=%s, want %s", tr.MonthStart, nextPeriod.Format(dateLayout)) + } +} + +func TestMonthUsedModes(t *testing.T) { + tr := Traffic{MonthRx: 100, MonthTx: 250} + if got := tr.MonthUsed("sum"); got != 350 { + t.Errorf("sum=%d", got) + } + if got := tr.MonthUsed("up"); got != 250 { + t.Errorf("up=%d", got) + } + if got := tr.MonthUsed("down"); got != 100 { + t.Errorf("down=%d", got) + } + if got := tr.MonthUsed("max"); got != 250 { + t.Errorf("max=%d", got) + } + if got := tr.MonthUsed(""); got != 350 { + t.Errorf("default=%d", got) + } +} + +func TestTrafficReadGating(t *testing.T) { + st := newTestStore(t) + // Book in an old period (two months ago): month figures belong to + // that period and must read back as zero, totals intact. + old := dateOnly(time.Now()).AddDate(0, -2, 0) + book(t, st, 3, "k/fp", 1000, 1000, old, nil, 0) + book(t, st, 3, "k/fp", 1400, 1500, old.Add(time.Minute), nil, old.Unix()) + got, ok, err := st.TrafficFor(3) + if err != nil || !ok { + t.Fatalf("TrafficFor: ok=%v err=%v", ok, err) + } + if got.TotalRx != 400 || got.TotalTx != 500 { + t.Fatalf("totals must survive gating: %+v", got) + } + if got.MonthRx != 0 || got.MonthTx != 0 || got.DayRx != 0 { + t.Fatalf("stale period must gate to zero: %+v", got) + } + all, err := st.AllTraffic() + if err != nil { + t.Fatal(err) + } + if all[3].TotalRx != 400 || all[3].MonthRx != 0 { + t.Fatalf("AllTraffic gating wrong: %+v", all[3]) + } + if _, ok, _ := st.TrafficFor(999); ok { + t.Fatal("unknown agent must report not-found") + } +} + +func TestSetTrafficStampsCurrentPeriod(t *testing.T) { + st := newTestStore(t) + now := time.Now() + book(t, st, 5, "k/fp", 1000, 1000, now, nil, 0) + ok, err := st.SetTraffic(5, i64p(10_000), nil, i64p(700), nil) + if err != nil || !ok { + t.Fatalf("SetTraffic: ok=%v err=%v", ok, err) + } + got, _, _ := st.TrafficFor(5) + if got.TotalRx != 10_000 || got.MonthRx != 700 { + t.Fatalf("correction not visible: %+v", got) + } + // The next booking must not discard the correction: month is + // stamped with the current period, so the delta adds onto it. + tr := book(t, st, 5, "k/fp", 1100, 1000, now.Add(time.Minute), nil, now.Unix()) + if tr.MonthRx != 800 || tr.TotalRx != 10_100 { + t.Fatalf("booking after correction wrong: %+v", tr) + } + if ok, _ := st.SetTraffic(4242, i64p(1), nil, nil, nil); ok { + t.Fatal("SetTraffic on unknown agent must be false") + } +} diff --git a/internal/web/web.go b/internal/web/web.go index 1a79cfe..65997ca 100644 --- a/internal/web/web.go +++ b/internal/web/web.go @@ -29,6 +29,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" "github.com/mhsanaei/3x-ui/v3/internal/web/service" "github.com/mhsanaei/3x-ui/v3/internal/web/service/panel" + "github.com/mhsanaei/3x-ui/v3/internal/web/service/probe" "github.com/mhsanaei/3x-ui/v3/internal/web/service/tgbot" "github.com/mhsanaei/3x-ui/v3/internal/web/websocket" @@ -331,6 +332,17 @@ func (s *Server) startTask(restartXray bool, loc *time.Location) { _, _ = s.cron.AddJob(cadenceReapOrphans, job.NewReapSyncOrphansJob()) + // Probe history housekeeping (item 7 B2): roll the closed hours of + // probe.db into the hourly tier and prune past the retention + // window. Hourly on the hour, plus one pass shortly after startup + // (the hub runs its housekeeping immediately at boot; the delayed + // kick follows the MemoryReleaseJob precedent below). + _, _ = s.cron.AddJob("@hourly", job.NewProbeHistoryJob()) + go func() { + time.Sleep(time.Minute) + job.NewProbeHistoryJob().Run() + }() + // check client ips from log file every day _, _ = s.cron.AddJob("@daily", job.NewClearLogsJob()) _, _ = s.cron.AddJob(cadenceXrayLogPrune, job.NewPruneXrayLogsJob()) @@ -596,7 +608,6 @@ func (s *Server) start(restartXray bool) (err error) { } }) - s.startTask(restartXray, loc) if isTgbotenabled, err := s.settingService.GetTgbotEnabled(); (err == nil) && isTgbotenabled { @@ -620,6 +631,10 @@ func (s *Server) StopPanelOnly() error { func (s *Server) stop(stopXray bool) error { s.cancel() + // Book the probe ledger's held readings before going down (item 7 + // B2): a node that reboots before reporting again would otherwise + // take the bytes since its last booking with it. + probe.BookHeld() if stopXray { } if s.cron != nil {