3x-ui-agent/internal/web/service/probe/history.go

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
			}
		}
	}
}