R-1247: the weekly upload line test — the agent's own iperf3 client to ep0's tunnel-only server, Tuesday 10:00-15:59 after 10 quiet minutes, under the heavy-op gate, best of 3; line_test stanza + GET /host/line-test; 15 tests red-proved; unreleased (next v0.164.0)
gates / gates (push) Successful in 1m35s
gates / gates (push) Successful in 1m35s
This commit is contained in:
@@ -34,6 +34,7 @@ type Config struct {
|
||||
WGTunnel WGTunnelConfig `json:"wg_tunnel"`
|
||||
GuestNet GuestNetConfig `json:"guest_net"`
|
||||
DiskTrim DiskTrimConfig `json:"disk_trim"`
|
||||
LineTest LineTestConfig `json:"line_test"`
|
||||
OOB OOBConfig `json:"oob"`
|
||||
SelfUpdate SelfUpdateConfig `json:"selfupdate"`
|
||||
LogLevel string `json:"log_level"` // debug|info|warn|error (default info)
|
||||
@@ -150,6 +151,16 @@ type DiskTrimConfig struct {
|
||||
// Enabled reports whether the weekly trim should run.
|
||||
func (d DiskTrimConfig) Enabled() bool { return !d.Disable }
|
||||
|
||||
// LineTestConfig configures the R-1247 weekly upload line test (internal/linetest). DEFAULT-ON: it reaches only ep0's
|
||||
// tunnel-only iperf3 server, so it runs only on a box that already has an operational offsite tunnel (operator
|
||||
// ruling 2026-10-24 07:26: every box with a tunnel measures its line weekly).
|
||||
type LineTestConfig struct {
|
||||
Disable bool `json:"disable"`
|
||||
}
|
||||
|
||||
// Enabled reports whether the weekly line test should run.
|
||||
func (l LineTestConfig) Enabled() bool { return !l.Disable }
|
||||
|
||||
// GuestNetConfig configures the R-54 guest-network watchdog (internal/guestnet).
|
||||
//
|
||||
// **This is the repo's first DEFAULT-ON feature gate, and the inversion is deliberate.** Every other
|
||||
|
||||
@@ -104,6 +104,12 @@ type GuestDiskTrimReporter interface {
|
||||
GuestDiskTrimStatus(ctx context.Context) *GuestDiskTrimStatus
|
||||
}
|
||||
|
||||
// LineTestReporter is the R-1247 seam the weekly line test plugs into (hub does not import linetest). nil → no
|
||||
// line_test stanza.
|
||||
type LineTestReporter interface {
|
||||
LineTestStatus(ctx context.Context) *LineTestStatus
|
||||
}
|
||||
|
||||
// Collector builds a HostReport from read-only sources. All deps are behind narrow
|
||||
// interfaces for unit testing.
|
||||
type Collector struct {
|
||||
@@ -124,6 +130,7 @@ type Collector struct {
|
||||
ctrlSup ControllerSupervisorReporter // R-523: in-guest controller supervisor (nil → stanza omitted)
|
||||
guestNet GuestNetReporter // R-54: per-guest network watchdog (nil → stanza omitted)
|
||||
diskTrim GuestDiskTrimReporter // R-444: weekly guest disk trim (nil → stanza omitted)
|
||||
lineTest LineTestReporter // R-1247: weekly upload line test (nil → stanza omitted)
|
||||
deliberateOff func() *DeliberateOff // R-998: household power-off record of this boot (nil → omitted)
|
||||
foreignKey ForeignKeyArchiveReporter // R-366 slice 2 (nil → omitted)
|
||||
selfUpdate SelfUpdateReporter // D1: agent self-update pending status (nil → false)
|
||||
@@ -259,6 +266,12 @@ func (c *Collector) SetGuestDiskTrimReporter(r GuestDiskTrimReporter) *Collector
|
||||
return c
|
||||
}
|
||||
|
||||
// SetLineTestReporter wires the R-1247 weekly line test as a report source (nil-safe → stanza omitted).
|
||||
func (c *Collector) SetLineTestReporter(r LineTestReporter) *Collector {
|
||||
c.lineTest = r
|
||||
return c
|
||||
}
|
||||
|
||||
// SetDeliberateOffSource wires the R-998 power-off record (poweroff.State.Current, adapted).
|
||||
func (c *Collector) SetDeliberateOffSource(f func() *DeliberateOff) *Collector {
|
||||
c.deliberateOff = f
|
||||
@@ -427,6 +440,10 @@ func (c *Collector) Collect(ctx context.Context) (*HostReport, error) {
|
||||
if c.diskTrim != nil {
|
||||
report.GuestDiskTrim = c.diskTrim.GuestDiskTrimStatus(ctx)
|
||||
}
|
||||
// R-1247: the last weekly line test (nil reporter = not wired → stanza omitted).
|
||||
if c.lineTest != nil {
|
||||
report.LineTest = c.lineTest.LineTestStatus(ctx)
|
||||
}
|
||||
// R-366 slice 2: archives the restore-test skipped as another key's (nil → not evaluated yet → omitted).
|
||||
if c.foreignKey != nil {
|
||||
report.ForeignKeyArchives = c.foreignKey.ForeignKeyArchives(ctx)
|
||||
|
||||
@@ -129,6 +129,11 @@ type HostReport struct {
|
||||
// wired; an empty `guests` list means the job runs and no guest has been trimmed yet. No secret.
|
||||
GuestDiskTrim *GuestDiskTrimStatus `json:"guest_disk_trim,omitempty"`
|
||||
|
||||
// LineTest is the weekly upload line test stanza (R-1247; operator ruling 2026-10-24): the schedule, the last
|
||||
// measured upload line (best of the runs, Mbit/s, as ep0's iperf3 server counted it) and the last attempt. Present
|
||||
// only when the job is wired. No secret.
|
||||
LineTest *LineTestStatus `json:"line_test,omitempty"`
|
||||
|
||||
// ForeignKeyArchives (R-366 slice 2, `09` §3 decision 168): per backup tier, the whole-guest archives the
|
||||
// restore-test SKIPPED because they were written with another key (an earlier install of this box). This box
|
||||
// cannot open them; the hub turns a CHANGE of this list into one operator event. Absent = not evaluated yet since
|
||||
@@ -232,6 +237,19 @@ type GuestDiskTrimStatus struct {
|
||||
Guests []GuestDiskTrim `json:"guests,omitempty"`
|
||||
}
|
||||
|
||||
// LineTestStatus is the R-1247 weekly upload line test stanza. `last_ok_at` + `mbps` are the last SUCCESSFUL
|
||||
// measurement ("" / 0 = never); `last_attempt_at` + `last_error` the last attempt, whatever its verdict — never read
|
||||
// the attempt time as a result.
|
||||
type LineTestStatus struct {
|
||||
Schedule string `json:"schedule"`
|
||||
LastOKAt string `json:"last_ok_at,omitempty"`
|
||||
Mbps float64 `json:"mbps,omitempty"`
|
||||
RunsMbps []float64 `json:"runs_mbps,omitempty"`
|
||||
Server string `json:"server,omitempty"`
|
||||
LastAttemptAt string `json:"last_attempt_at,omitempty"`
|
||||
LastError string `json:"last_error,omitempty"`
|
||||
}
|
||||
|
||||
// GuestDiskTrim is one guest's LAST trim attempt. `ok` with `last_attempt_at` is the verdict of that attempt — never
|
||||
// read the time alone as success; `last_ok_at` is the last attempt that succeeded ("" = never). `bytes_trimmed` is
|
||||
// the sum of the "(N bytes) trimmed" lines `pct fstrim` printed, over `mounts` mount points.
|
||||
|
||||
@@ -0,0 +1,294 @@
|
||||
package linetest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"strconv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A minimal iperf3 CLIENT for one TCP upload test (client → server), speaking the iperf3 control protocol (3.x) to the
|
||||
// stock `iperf3 --server` on ep0 (operator ruling 2026-10-24: the agent speaks iperf3 itself, nothing is installed on
|
||||
// the host). The number that counts is the SERVER's: the bytes it received, from its results JSON — bytes the client
|
||||
// handed to its socket include what still sat in the send buffer when the timer ended.
|
||||
//
|
||||
// The protocol, as the server drives it (one state byte at a time on the control connection):
|
||||
//
|
||||
// client → 37-byte cookie
|
||||
// server → PARAM_EXCHANGE(9) client → params JSON (4-byte big-endian length + bytes)
|
||||
// server → CREATE_STREAMS(10) client opens N data connections, each starts with the same cookie
|
||||
// server → TEST_START(1), TEST_RUNNING(2) client sends for the test time, then → TEST_END(4)
|
||||
// server → EXCHANGE_RESULTS(13) client → its results JSON; server → its results JSON
|
||||
// server → DISPLAY_RESULTS(14) client → IPERF_DONE(16)
|
||||
//
|
||||
// A server already running a test answers ACCESS_DENIED(-1) and closes: that is ErrBusy (another box is testing; try
|
||||
// again later the same day). SERVER_ERROR(-3) / SERVER_TERMINATE(-2) end the test with an error.
|
||||
|
||||
const (
|
||||
stParamExchange = 9
|
||||
stCreateStreams = 10
|
||||
stTestStart = 1
|
||||
stTestRunning = 2
|
||||
stTestEnd = 4
|
||||
stExchangeResults = 13
|
||||
stDisplayResults = 14
|
||||
stIperfDone = 16
|
||||
stAccessDenied = -1
|
||||
stServerTerminate = -2
|
||||
stServerError = -3
|
||||
|
||||
cookieSize = 37 // 36 characters + NUL
|
||||
blockSize = 128 * 1024
|
||||
maxJSONBytes = 1 << 20
|
||||
)
|
||||
|
||||
// ErrBusy: the server is running another client's test.
|
||||
var ErrBusy = errors.New("iperf3 server busy")
|
||||
|
||||
// Result is one upload test.
|
||||
type Result struct {
|
||||
Mbps float64 // the server's received bits per second / 1e6
|
||||
ServerBytes int64 // bytes the server received, all streams
|
||||
ClientBytes int64 // bytes the client wrote (diagnostics only)
|
||||
ServerSeconds float64 // the server's measured duration
|
||||
}
|
||||
|
||||
// Dialer opens a TCP connection (net.Dialer in production; a seam for tests).
|
||||
type Dialer interface {
|
||||
DialContext(ctx context.Context, network, address string) (net.Conn, error)
|
||||
}
|
||||
|
||||
// Client runs one test. The whole run is bounded by ctx; Run sets every connection's deadline from it.
|
||||
type Client struct {
|
||||
Dialer Dialer
|
||||
Streams int // parallel streams (-P)
|
||||
Time time.Duration // test time (-t)
|
||||
}
|
||||
|
||||
func (c Client) dialer() Dialer {
|
||||
if c.Dialer != nil {
|
||||
return c.Dialer
|
||||
}
|
||||
return &net.Dialer{Timeout: 5 * time.Second}
|
||||
}
|
||||
|
||||
func newCookie() ([]byte, error) {
|
||||
const alphabet = "abcdefghijklmnopqrstuvwxyz234567"
|
||||
b := make([]byte, cookieSize)
|
||||
if _, err := rand.Read(b[:cookieSize-1]); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for i := 0; i < cookieSize-1; i++ {
|
||||
b[i] = alphabet[int(b[i])%len(alphabet)]
|
||||
}
|
||||
b[cookieSize-1] = 0
|
||||
return b, nil
|
||||
}
|
||||
|
||||
func writeJSON(w io.Writer, v any) error {
|
||||
data, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var hdr [4]byte
|
||||
binary.BigEndian.PutUint32(hdr[:], uint32(len(data)))
|
||||
if _, err := w.Write(append(hdr[:], data...)); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func readJSON(r io.Reader, v any) error {
|
||||
var hdr [4]byte
|
||||
if _, err := io.ReadFull(r, hdr[:]); err != nil {
|
||||
return err
|
||||
}
|
||||
n := binary.BigEndian.Uint32(hdr[:])
|
||||
if n == 0 || n > maxJSONBytes {
|
||||
return fmt.Errorf("iperf3: results JSON of %d bytes refused", n)
|
||||
}
|
||||
buf := make([]byte, n)
|
||||
if _, err := io.ReadFull(r, buf); err != nil {
|
||||
return err
|
||||
}
|
||||
return json.Unmarshal(buf, v)
|
||||
}
|
||||
|
||||
type streamResult struct {
|
||||
ID int `json:"id"`
|
||||
Bytes int64 `json:"bytes"`
|
||||
Retransmits int64 `json:"retransmits"`
|
||||
Jitter float64 `json:"jitter"`
|
||||
Errors int64 `json:"errors"`
|
||||
OmittedErrors int64 `json:"omitted_errors"`
|
||||
Packets int64 `json:"packets"`
|
||||
OmittedPackets int64 `json:"omitted_packets"`
|
||||
StartTime float64 `json:"start_time"`
|
||||
EndTime float64 `json:"end_time"`
|
||||
}
|
||||
|
||||
type results struct {
|
||||
CPUTotal float64 `json:"cpu_util_total"`
|
||||
CPUUser float64 `json:"cpu_util_user"`
|
||||
CPUSystem float64 `json:"cpu_util_system"`
|
||||
SenderHasRetransmits int `json:"sender_has_retransmits"`
|
||||
CongestionUsed string `json:"congestion_used,omitempty"`
|
||||
Streams []streamResult `json:"streams"`
|
||||
}
|
||||
|
||||
// streamIDs are the ids iperf3 gives its streams: the first is 1, every later one is (streams so far) + 2 — so 1, 3,
|
||||
// 4, 5 … (iperf_add_stream). The server matches our results by these ids.
|
||||
func streamIDs(n int) []int {
|
||||
ids := make([]int, n)
|
||||
for i := range ids {
|
||||
if i == 0 {
|
||||
ids[i] = 1
|
||||
} else {
|
||||
ids[i] = i + 2
|
||||
}
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
// Run performs one upload test to addr ("10.77.0.1:5201").
|
||||
func (c Client) Run(ctx context.Context, addr string) (Result, error) {
|
||||
var res Result
|
||||
streams, dur := c.Streams, c.Time
|
||||
if streams < 1 {
|
||||
streams = 1
|
||||
}
|
||||
if dur <= 0 {
|
||||
dur = 10 * time.Second
|
||||
}
|
||||
deadline, ok := ctx.Deadline()
|
||||
if !ok {
|
||||
return res, errors.New("iperf3: Run needs a context with a deadline")
|
||||
}
|
||||
cookie, err := newCookie()
|
||||
if err != nil {
|
||||
return res, err
|
||||
}
|
||||
ctrl, err := c.dialer().DialContext(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("iperf3: connect: %w", err)
|
||||
}
|
||||
defer ctrl.Close()
|
||||
ctrl.SetDeadline(deadline)
|
||||
if _, err := ctrl.Write(cookie); err != nil {
|
||||
return res, fmt.Errorf("iperf3: cookie: %w", err)
|
||||
}
|
||||
|
||||
var data []net.Conn
|
||||
defer func() {
|
||||
for _, d := range data {
|
||||
d.Close()
|
||||
}
|
||||
}()
|
||||
var sent atomic.Int64
|
||||
var server results
|
||||
var wg sync.WaitGroup
|
||||
for {
|
||||
var st [1]byte
|
||||
if _, err := io.ReadFull(ctrl, st[:]); err != nil {
|
||||
return res, fmt.Errorf("iperf3: control read: %w", err)
|
||||
}
|
||||
switch int8(st[0]) {
|
||||
case stParamExchange:
|
||||
params := map[string]any{"tcp": true, "omit": 0, "time": int(dur / time.Second), "num": 0, "blockcount": 0,
|
||||
"parallel": streams, "len": blockSize, "pacing_timer": 1000, "client_version": "3.18"}
|
||||
if err := writeJSON(ctrl, params); err != nil {
|
||||
return res, fmt.Errorf("iperf3: params: %w", err)
|
||||
}
|
||||
case stCreateStreams:
|
||||
for i := 0; i < streams; i++ {
|
||||
d, err := c.dialer().DialContext(ctx, "tcp", addr)
|
||||
if err != nil {
|
||||
return res, fmt.Errorf("iperf3: data stream %d: %w", i+1, err)
|
||||
}
|
||||
d.SetDeadline(deadline)
|
||||
data = append(data, d)
|
||||
if _, err := d.Write(cookie); err != nil {
|
||||
return res, fmt.Errorf("iperf3: data stream %d cookie: %w", i+1, err)
|
||||
}
|
||||
}
|
||||
case stTestStart:
|
||||
case stTestRunning:
|
||||
stopAt := time.Now().Add(dur)
|
||||
if stopAt.After(deadline) {
|
||||
return res, errors.New("iperf3: no time left for the test inside the deadline")
|
||||
}
|
||||
buf := make([]byte, blockSize)
|
||||
for _, d := range data {
|
||||
d.SetWriteDeadline(stopAt)
|
||||
wg.Add(1)
|
||||
go func(d net.Conn) {
|
||||
defer wg.Done()
|
||||
for {
|
||||
n, err := d.Write(buf)
|
||||
sent.Add(int64(n))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}(d)
|
||||
}
|
||||
wg.Wait()
|
||||
if _, err := ctrl.Write([]byte{stTestEnd}); err != nil {
|
||||
return res, fmt.Errorf("iperf3: test end: %w", err)
|
||||
}
|
||||
case stExchangeResults:
|
||||
ids := streamIDs(len(data))
|
||||
mine := results{SenderHasRetransmits: -1, CongestionUsed: "cubic"}
|
||||
per := sent.Load() / int64(max(len(data), 1))
|
||||
for _, id := range ids {
|
||||
mine.Streams = append(mine.Streams, streamResult{ID: id, Bytes: per, Retransmits: -1, EndTime: dur.Seconds()})
|
||||
}
|
||||
if err := writeJSON(ctrl, mine); err != nil {
|
||||
return res, fmt.Errorf("iperf3: send results: %w", err)
|
||||
}
|
||||
if err := readJSON(ctrl, &server); err != nil {
|
||||
return res, fmt.Errorf("iperf3: read server results: %w", err)
|
||||
}
|
||||
case stDisplayResults:
|
||||
ctrl.Write([]byte{stIperfDone})
|
||||
return summarize(server, sent.Load(), dur)
|
||||
case stAccessDenied:
|
||||
return res, ErrBusy
|
||||
case stServerTerminate, stServerError:
|
||||
var codes [8]byte
|
||||
io.ReadFull(ctrl, codes[:])
|
||||
return res, fmt.Errorf("iperf3: server error (state %d, iperf errno %d, errno %d)", int8(st[0]),
|
||||
int32(binary.BigEndian.Uint32(codes[:4])), int32(binary.BigEndian.Uint32(codes[4:])))
|
||||
default:
|
||||
return res, fmt.Errorf("iperf3: unexpected state %s", strconv.Itoa(int(int8(st[0]))))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// summarize turns the server's results into a Result. No stream bytes is an error, never a zero line.
|
||||
func summarize(server results, clientBytes int64, dur time.Duration) (Result, error) {
|
||||
res := Result{ClientBytes: clientBytes}
|
||||
for _, s := range server.Streams {
|
||||
res.ServerBytes += s.Bytes
|
||||
if s.EndTime > res.ServerSeconds {
|
||||
res.ServerSeconds = s.EndTime
|
||||
}
|
||||
}
|
||||
if res.ServerBytes <= 0 {
|
||||
return res, errors.New("iperf3: the server reported no received bytes")
|
||||
}
|
||||
secs := res.ServerSeconds
|
||||
if secs <= 0 {
|
||||
secs = dur.Seconds()
|
||||
}
|
||||
res.Mbps = float64(res.ServerBytes) * 8 / secs / 1e6
|
||||
return res, nil
|
||||
}
|
||||
@@ -0,0 +1,439 @@
|
||||
// Package linetest is the weekly upload line-speed test (R-1247; operator ruling 2026-10-24 07:26).
|
||||
//
|
||||
// Why: the controller's upload watch and qBittorrent's default limit (50 % of the line, D41) need the household's
|
||||
// upload line. The passive reading — the nightly off-site push — is only a low floor (2026-10-24 night: demo-hp „at
|
||||
// least 10 Mbit/s"; this test read 90.7 on the same line). So once a week the HOST (the tunnel lives on the host)
|
||||
// runs a 10-second iperf3 upload to ep0's tunnel-only iperf3 server (felhom.eu scripts/ep0-iperf3). No file is left
|
||||
// anywhere and nothing counts against the household's storage.
|
||||
//
|
||||
// The rule, each part pinned by a test in linetest_test.go:
|
||||
// - Weekly: due from Tuesday 10:00 local until one measurement succeeded since then (a box that was off catches up
|
||||
// on a later day, inside the window).
|
||||
// - Daytime only: a test starts only 10:00–15:59 local — never in the night chain's window (01:00–06:59), and each
|
||||
// host starts from its own minute (SlotMinute, spread over 10:00–13:59) so the fleet does not test at once.
|
||||
// - Never beside a backup, a restore-test, an OS update or a trim: the test holds the host-wide one-heavy-op gate.
|
||||
// - Only after QuietMinutes of quiet uplink: every one-minute transmit rate of the host's physical uplink below
|
||||
// QuietMbps — so neither the family's upload nor a restore is measured against itself.
|
||||
// - Up to RunsPerMeasurement runs of TestTime each; the highest counts. A run is bounded by RunTimeout.
|
||||
// - „Server busy" (another box is testing) → tried again at the next tick the same day; at most MaxTriesPerDay.
|
||||
// - No tunnel (no operational wg-felhom with a server address) → no test and no value; the controller then keeps its
|
||||
// passive floor and sets no qBittorrent limit.
|
||||
// - The last result is persisted, so a restart neither loses it nor re-tests.
|
||||
package linetest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"hash/fnv"
|
||||
"log/slog"
|
||||
"net/netip"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
|
||||
)
|
||||
|
||||
const (
|
||||
Weekday = time.Tuesday
|
||||
StartHour = 10 // first eligible local hour (inclusive)
|
||||
EndHour = 16 // first NOT-eligible local hour: the last start is 15:59
|
||||
SlotSpreadMinutes = 240
|
||||
QuietMinutes = 10
|
||||
QuietMbps = 2.0
|
||||
RunsPerMeasurement = 3
|
||||
TestTime = 10 * time.Second
|
||||
Streams = 4
|
||||
RunTimeout = 25 * time.Second
|
||||
RunGap = 30 * time.Second
|
||||
MaxTriesPerDay = 6
|
||||
ServerPort = 5201
|
||||
TickInterval = time.Minute
|
||||
FirstTickDelay = 5 * time.Minute
|
||||
GateName = "line-test"
|
||||
)
|
||||
|
||||
// ScheduleText is carried on the host report.
|
||||
const ScheduleText = "weekly, due Tuesday from 10:00 host-local time; starts 10:00-15:59 after 10 quiet minutes; never beside a backup, restore-test or update"
|
||||
|
||||
// Gate is the host-wide one-heavy-operation gate (*backup.InFlight).
|
||||
type Gate interface {
|
||||
TryAcquire(what string) (release func(), busy string, ok bool)
|
||||
}
|
||||
|
||||
// Target answers the iperf3 server's tunnel address, ok=false when the tunnel is not operational.
|
||||
type Target func() (netip.Addr, bool)
|
||||
|
||||
// Runner runs one upload test (Client in production).
|
||||
type Runner interface {
|
||||
Run(ctx context.Context, addr string) (Result, error)
|
||||
}
|
||||
|
||||
// UplinkTx reads the host's physical-uplink transmit byte counter.
|
||||
type UplinkTx func() (uint64, error)
|
||||
|
||||
// State is what is persisted (line-test.json).
|
||||
type State struct {
|
||||
LastOKAt time.Time `json:"last_ok_at,omitempty"`
|
||||
Mbps float64 `json:"mbps,omitempty"`
|
||||
RunsMbps []float64 `json:"runs_mbps,omitempty"`
|
||||
Server string `json:"server,omitempty"`
|
||||
LastAttemptAt time.Time `json:"last_attempt_at,omitempty"`
|
||||
LastError string `json:"last_error,omitempty"`
|
||||
TriesDay string `json:"tries_day,omitempty"` // local date of TriesToday
|
||||
TriesToday int `json:"tries_today,omitempty"`
|
||||
}
|
||||
|
||||
type txSample struct {
|
||||
at time.Time
|
||||
tx uint64
|
||||
}
|
||||
|
||||
// Tester is the weekly job.
|
||||
type Tester struct {
|
||||
runner Runner
|
||||
target Target
|
||||
gate Gate
|
||||
uplink UplinkTx
|
||||
statePath string
|
||||
slot int // minutes after StartHour:00 this host may first start
|
||||
logger *slog.Logger
|
||||
loc *time.Location
|
||||
now func() time.Time
|
||||
sleep func(context.Context, time.Duration) bool
|
||||
|
||||
mu sync.Mutex
|
||||
st State
|
||||
samples []txSample
|
||||
}
|
||||
|
||||
// New builds the job and loads the persisted state (missing = empty; corrupt = logged, empty — the cost is one extra
|
||||
// test, never a lost value elsewhere).
|
||||
func New(runner Runner, target Target, gate Gate, uplink UplinkTx, hostID, statePath string, logger *slog.Logger) *Tester {
|
||||
if logger == nil {
|
||||
logger = slog.Default()
|
||||
}
|
||||
t := &Tester{runner: runner, target: target, gate: gate, uplink: uplink, statePath: statePath,
|
||||
slot: SlotMinute(hostID), logger: logger, loc: time.Local, now: time.Now, sleep: sleepCtx}
|
||||
t.load()
|
||||
return t
|
||||
}
|
||||
|
||||
func sleepCtx(ctx context.Context, d time.Duration) bool {
|
||||
tm := time.NewTimer(d)
|
||||
defer tm.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-tm.C:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// SlotMinute spreads hosts over the first SlotSpreadMinutes of the window, stable per host id.
|
||||
func SlotMinute(hostID string) int {
|
||||
h := fnv.New32a()
|
||||
h.Write([]byte(hostID))
|
||||
return int(h.Sum32() % SlotSpreadMinutes)
|
||||
}
|
||||
|
||||
func (t *Tester) load() {
|
||||
data, err := os.ReadFile(t.statePath)
|
||||
if err != nil {
|
||||
if !os.IsNotExist(err) {
|
||||
t.logger.Warn("linetest: state read failed — starting empty", "path", t.statePath, "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err := json.Unmarshal(data, &t.st); err != nil {
|
||||
t.logger.Warn("linetest: state file corrupt — starting empty", "path", t.statePath, "err", err)
|
||||
t.st = State{}
|
||||
}
|
||||
}
|
||||
|
||||
func (t *Tester) saveLocked() error {
|
||||
data, err := json.MarshalIndent(t.st, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(t.statePath), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
tmp := t.statePath + ".tmp"
|
||||
if err := os.WriteFile(tmp, data, 0o600); err != nil {
|
||||
os.Remove(tmp)
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, t.statePath)
|
||||
}
|
||||
|
||||
// EligibleAt reports whether a test may START at local time lt for a host whose slot is slot minutes after
|
||||
// StartHour:00 on the due day. On a catch-up day (anchorDay=false) the slot is not waited for again.
|
||||
func EligibleAt(lt time.Time, slot int, anchorDay bool) bool {
|
||||
h := lt.Hour()
|
||||
if h < StartHour || h >= EndHour {
|
||||
return false
|
||||
}
|
||||
if anchorDay {
|
||||
mins := (h-StartHour)*60 + lt.Minute()
|
||||
return mins >= slot
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// weekAnchor is the most recent Tuesday StartHour:00 at or before lt.
|
||||
func weekAnchor(lt time.Time) time.Time {
|
||||
daysBack := (int(lt.Weekday()) - int(Weekday) + 7) % 7
|
||||
d := lt.AddDate(0, 0, -daysBack)
|
||||
a := time.Date(d.Year(), d.Month(), d.Day(), StartHour, 0, 0, 0, lt.Location())
|
||||
if a.After(lt) {
|
||||
d = d.AddDate(0, 0, -7)
|
||||
a = time.Date(d.Year(), d.Month(), d.Day(), StartHour, 0, 0, 0, lt.Location())
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
// Due reports whether a measurement is due at lt given the state.
|
||||
func Due(st State, lt time.Time) bool {
|
||||
return st.LastOKAt.IsZero() || st.LastOKAt.In(lt.Location()).Before(weekAnchor(lt))
|
||||
}
|
||||
|
||||
// sample records the uplink counter (once a minute) and keeps QuietMinutes+1 samples.
|
||||
func (t *Tester) sample() {
|
||||
if t.uplink == nil {
|
||||
return
|
||||
}
|
||||
tx, err := t.uplink()
|
||||
if err != nil {
|
||||
t.logger.Debug("linetest: uplink counter unreadable", "err", err)
|
||||
t.mu.Lock()
|
||||
t.samples = nil // a gap is not quiet
|
||||
t.mu.Unlock()
|
||||
return
|
||||
}
|
||||
t.mu.Lock()
|
||||
t.samples = append(t.samples, txSample{at: t.now(), tx: tx})
|
||||
if len(t.samples) > QuietMinutes+1 {
|
||||
t.samples = t.samples[len(t.samples)-(QuietMinutes+1):]
|
||||
}
|
||||
t.mu.Unlock()
|
||||
}
|
||||
|
||||
// quiet reports whether the last QuietMinutes minutes each sent below QuietMbps, and the highest minute seen.
|
||||
func quiet(samples []txSample) (bool, float64) {
|
||||
if len(samples) < QuietMinutes+1 {
|
||||
return false, 0
|
||||
}
|
||||
worst := 0.0
|
||||
for i := 1; i < len(samples); i++ {
|
||||
dt := samples[i].at.Sub(samples[i-1].at).Seconds()
|
||||
if dt <= 0 || dt > 3*TickInterval.Seconds() || samples[i].tx < samples[i-1].tx {
|
||||
return false, worst // a gap or a counter reset is not proof of quiet
|
||||
}
|
||||
mbps := float64(samples[i].tx-samples[i-1].tx) * 8 / dt / 1e6
|
||||
if mbps > worst {
|
||||
worst = mbps
|
||||
}
|
||||
}
|
||||
return worst < QuietMbps, worst
|
||||
}
|
||||
|
||||
// Run ticks every TickInterval until ctx ends. It never returns an error: a failed test is a reported fact.
|
||||
func (t *Tester) Run(ctx context.Context) {
|
||||
t.logger.Info("linetest: weekly upload line test starting", "schedule", ScheduleText, "slot_minute", t.slot)
|
||||
if !t.sleep(ctx, FirstTickDelay) {
|
||||
return
|
||||
}
|
||||
for {
|
||||
t.sample()
|
||||
t.Tick(ctx)
|
||||
if !t.sleep(ctx, TickInterval) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Tick is one look.
|
||||
func (t *Tester) Tick(ctx context.Context) {
|
||||
lt := t.now().In(t.loc)
|
||||
t.mu.Lock()
|
||||
st := t.st
|
||||
samples := append([]txSample(nil), t.samples...)
|
||||
t.mu.Unlock()
|
||||
if !Due(st, lt) {
|
||||
return
|
||||
}
|
||||
anchor := weekAnchor(lt)
|
||||
anchorDay := lt.YearDay() == anchor.YearDay() && lt.Year() == anchor.Year()
|
||||
if !EligibleAt(lt, t.slot, anchorDay) {
|
||||
return
|
||||
}
|
||||
day := lt.Format("2006-01-02")
|
||||
if st.TriesDay == day && st.TriesToday >= MaxTriesPerDay {
|
||||
return
|
||||
}
|
||||
ip, ok := t.target()
|
||||
if !ok {
|
||||
t.logger.Debug("linetest: no operational tunnel — no test")
|
||||
return
|
||||
}
|
||||
if q, worst := quiet(samples); !q {
|
||||
t.logger.Debug("linetest: uplink not quiet — waiting", "worst_minute_mbps", fmt.Sprintf("%.1f", worst), "samples", len(samples))
|
||||
return
|
||||
}
|
||||
release, busy, ok := t.gate.TryAcquire(GateName)
|
||||
if !ok {
|
||||
t.logger.Info("linetest: deferred — a heavy operation is in flight", "busy", busy)
|
||||
return
|
||||
}
|
||||
defer release()
|
||||
t.measure(ctx, netip.AddrPortFrom(ip, ServerPort).String(), day)
|
||||
}
|
||||
|
||||
func (t *Tester) measure(ctx context.Context, addr, day string) {
|
||||
start := t.now()
|
||||
var runs []float64
|
||||
var lastErr error
|
||||
busy := false
|
||||
for i := 0; i < RunsPerMeasurement; i++ {
|
||||
if i > 0 && !t.sleep(ctx, RunGap) {
|
||||
break
|
||||
}
|
||||
rctx, cancel := context.WithTimeout(ctx, RunTimeout)
|
||||
r, err := t.runner.Run(rctx, addr)
|
||||
cancel()
|
||||
if errors.Is(err, ErrBusy) {
|
||||
busy, lastErr = true, err
|
||||
break
|
||||
}
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
t.logger.Warn("linetest: run failed", "run", i+1, "server", addr, "err", err)
|
||||
continue
|
||||
}
|
||||
runs = append(runs, r.Mbps)
|
||||
t.logger.Info(fmt.Sprintf("linetest: run %d: %.1f Mbit/s", i+1, r.Mbps), "server", addr,
|
||||
"server_bytes", r.ServerBytes, "server_seconds", r.ServerSeconds)
|
||||
}
|
||||
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if t.st.TriesDay != day {
|
||||
t.st.TriesDay, t.st.TriesToday = day, 0
|
||||
}
|
||||
t.st.TriesToday++
|
||||
t.st.LastAttemptAt = start.UTC()
|
||||
if len(runs) > 0 {
|
||||
best := runs[0]
|
||||
for _, r := range runs {
|
||||
if r > best {
|
||||
best = r
|
||||
}
|
||||
}
|
||||
t.st.LastOKAt, t.st.Mbps, t.st.RunsMbps, t.st.Server, t.st.LastError = start.UTC(), best, runs, addr, ""
|
||||
t.logger.Info(fmt.Sprintf("linetest: upload line measured: %.1f Mbit/s (best of %d)", best, len(runs)),
|
||||
"server", addr, "runs", len(runs), "duration_s", t.now().Sub(start).Seconds())
|
||||
} else {
|
||||
msg := "no run succeeded"
|
||||
if lastErr != nil {
|
||||
msg = lastErr.Error()
|
||||
}
|
||||
if len(msg) > 300 {
|
||||
msg = msg[:300]
|
||||
}
|
||||
t.st.LastError = msg
|
||||
if busy {
|
||||
t.logger.Info("linetest: the test server is busy — trying again later today", "server", addr, "tries_today", t.st.TriesToday)
|
||||
} else {
|
||||
t.logger.Warn("linetest: measurement failed", "server", addr, "tries_today", t.st.TriesToday, "err", msg)
|
||||
}
|
||||
}
|
||||
if err := t.saveLocked(); err != nil {
|
||||
t.logger.Warn("linetest: state save failed — the result will not survive a restart", "path", t.statePath, "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Value is the measured line (ok=false: never measured).
|
||||
func (t *Tester) Value() (mbps float64, at time.Time, ok bool) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if t.st.LastOKAt.IsZero() || t.st.Mbps <= 0 {
|
||||
return 0, time.Time{}, false
|
||||
}
|
||||
return t.st.Mbps, t.st.LastOKAt, true
|
||||
}
|
||||
|
||||
// LineTestStatus implements hub.LineTestReporter: a pure read of the persisted state.
|
||||
func (t *Tester) LineTestStatus(context.Context) *hub.LineTestStatus {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
out := &hub.LineTestStatus{Schedule: ScheduleText, Mbps: t.st.Mbps, RunsMbps: t.st.RunsMbps, Server: t.st.Server,
|
||||
LastError: t.st.LastError}
|
||||
if !t.st.LastOKAt.IsZero() {
|
||||
out.LastOKAt = t.st.LastOKAt.UTC().Format(time.RFC3339)
|
||||
}
|
||||
if !t.st.LastAttemptAt.IsZero() {
|
||||
out.LastAttemptAt = t.st.LastAttemptAt.UTC().Format(time.RFC3339)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ── the host's physical uplink counter ──────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// SysUplinkTx reads the transmit bytes of the host's physical uplink under root (normally "/"): the interface of the
|
||||
// default route in /proc/net/route, and — when it is a bridge (vmbr0), as on every Proxmox host — the sum of its
|
||||
// PHYSICAL member ports (those with a `device` link). The bridge's own counter does not see the guests' bridged
|
||||
// traffic, which is the traffic this check exists for.
|
||||
func SysUplinkTx(root string) UplinkTx {
|
||||
return func() (uint64, error) {
|
||||
ifc, err := defaultRouteIface(filepath.Join(root, "proc/net/route"))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
sys := filepath.Join(root, "sys/class/net")
|
||||
ports := []string{ifc}
|
||||
if ents, err := os.ReadDir(filepath.Join(sys, ifc, "brif")); err == nil {
|
||||
ports = ports[:0]
|
||||
for _, e := range ents {
|
||||
if _, err := os.Stat(filepath.Join(sys, e.Name(), "device")); err == nil {
|
||||
ports = append(ports, e.Name())
|
||||
}
|
||||
}
|
||||
if len(ports) == 0 {
|
||||
return 0, fmt.Errorf("linetest: bridge %s has no physical port", ifc)
|
||||
}
|
||||
}
|
||||
var sum uint64
|
||||
for _, p := range ports {
|
||||
b, err := os.ReadFile(filepath.Join(sys, p, "statistics/tx_bytes"))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
n, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
sum += n
|
||||
}
|
||||
return sum, nil
|
||||
}
|
||||
}
|
||||
|
||||
func defaultRouteIface(path string) (string, error) {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
for _, line := range strings.Split(string(b), "\n")[1:] {
|
||||
f := strings.Fields(line)
|
||||
if len(f) >= 8 && f[1] == "00000000" && f[7] == "00000000" {
|
||||
return f[0], nil
|
||||
}
|
||||
}
|
||||
return "", errors.New("linetest: no default route")
|
||||
}
|
||||
@@ -0,0 +1,433 @@
|
||||
package linetest
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"net/netip"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
|
||||
)
|
||||
|
||||
// ── a fake iperf3 server (the server half of the protocol, as `iperf3 --server` drives it) ─────────────────────────
|
||||
|
||||
type fakeServer struct {
|
||||
ln net.Listener
|
||||
busy bool // answer ACCESS_DENIED
|
||||
stall bool // accept the cookie, then never answer
|
||||
recvSeconds float64 // what the server reports as its duration
|
||||
gotIDs []int // the stream ids the client reported
|
||||
gotParams map[string]any
|
||||
mu sync.Mutex
|
||||
received int64
|
||||
}
|
||||
|
||||
func newFakeServer(t *testing.T) *fakeServer {
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s := &fakeServer{ln: ln, recvSeconds: 1}
|
||||
t.Cleanup(func() { ln.Close() })
|
||||
go s.serve()
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *fakeServer) addr() string { return s.ln.Addr().String() }
|
||||
|
||||
func wjson(w io.Writer, v any) {
|
||||
b, _ := json.Marshal(v)
|
||||
var h [4]byte
|
||||
binary.BigEndian.PutUint32(h[:], uint32(len(b)))
|
||||
w.Write(append(h[:], b...))
|
||||
}
|
||||
|
||||
func (s *fakeServer) serve() {
|
||||
ctrl, err := s.ln.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer ctrl.Close()
|
||||
cookie := make([]byte, cookieSize)
|
||||
if _, err := io.ReadFull(ctrl, cookie); err != nil {
|
||||
return
|
||||
}
|
||||
if s.busy {
|
||||
ctrl.Write([]byte{0xff})
|
||||
return
|
||||
}
|
||||
if s.stall {
|
||||
time.Sleep(5 * time.Second)
|
||||
return
|
||||
}
|
||||
ctrl.Write([]byte{stParamExchange})
|
||||
var params map[string]any
|
||||
if readJSON(ctrl, ¶ms) != nil {
|
||||
return
|
||||
}
|
||||
s.gotParams = params
|
||||
n := int(params["parallel"].(float64))
|
||||
ctrl.Write([]byte{stCreateStreams})
|
||||
var wg sync.WaitGroup
|
||||
var conns []net.Conn
|
||||
for i := 0; i < n; i++ {
|
||||
d, err := s.ln.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
c := make([]byte, cookieSize)
|
||||
io.ReadFull(d, c)
|
||||
conns = append(conns, d)
|
||||
wg.Add(1)
|
||||
go func(d net.Conn) {
|
||||
defer wg.Done()
|
||||
k, _ := io.Copy(io.Discard, d)
|
||||
s.mu.Lock()
|
||||
s.received += k
|
||||
s.mu.Unlock()
|
||||
}(d)
|
||||
}
|
||||
ctrl.Write([]byte{stTestStart, stTestRunning})
|
||||
var st [1]byte
|
||||
if _, err := io.ReadFull(ctrl, st[:]); err != nil || st[0] != stTestEnd {
|
||||
return
|
||||
}
|
||||
ctrl.Write([]byte{stExchangeResults})
|
||||
var theirs results
|
||||
if readJSON(ctrl, &theirs) != nil {
|
||||
return
|
||||
}
|
||||
for _, x := range theirs.Streams {
|
||||
s.gotIDs = append(s.gotIDs, x.ID)
|
||||
}
|
||||
// The server counts what it RECEIVED; report a fixed figure so the test can tell it from the client's own count.
|
||||
wjson(ctrl, results{Streams: []streamResult{{ID: 1, Bytes: 12_500_000, EndTime: s.recvSeconds}}})
|
||||
ctrl.Write([]byte{stDisplayResults})
|
||||
io.ReadFull(ctrl, st[:])
|
||||
for _, d := range conns {
|
||||
d.Close()
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func TestClientSpeaksTheProtocolAndCountsTheServersBytes(t *testing.T) {
|
||||
s := newFakeServer(t)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
r, err := Client{Streams: 4, Time: 300 * time.Millisecond}.Run(ctx, s.addr())
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if r.ServerBytes != 12_500_000 || r.Mbps != 100 {
|
||||
t.Fatalf("result = %+v; want the SERVER's 12.5 MB in 1 s = 100 Mbit/s", r)
|
||||
}
|
||||
if want := []int{1, 3, 4, 5}; len(s.gotIDs) != 4 || s.gotIDs[0] != want[0] || s.gotIDs[1] != want[1] || s.gotIDs[3] != want[3] {
|
||||
t.Fatalf("stream ids sent = %v, want %v (iperf3 numbers its streams 1, 3, 4, 5)", s.gotIDs, want)
|
||||
}
|
||||
if s.gotParams["tcp"] != true || s.gotParams["parallel"].(float64) != 4 {
|
||||
t.Fatalf("params = %v", s.gotParams)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientBusyServerIsErrBusy(t *testing.T) {
|
||||
s := newFakeServer(t)
|
||||
s.busy = true
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if _, err := (Client{Streams: 4, Time: time.Second}).Run(ctx, s.addr()); !errors.Is(err, ErrBusy) {
|
||||
t.Fatalf("err = %v, want ErrBusy", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientHonoursTheDeadlineAgainstAStalledServer(t *testing.T) {
|
||||
s := newFakeServer(t)
|
||||
s.stall = true
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 400*time.Millisecond)
|
||||
defer cancel()
|
||||
start := time.Now()
|
||||
_, err := Client{Streams: 4, Time: time.Second}.Run(ctx, s.addr())
|
||||
if err == nil {
|
||||
t.Fatal("a stalled server gave no error")
|
||||
}
|
||||
if took := time.Since(start); took > 1500*time.Millisecond {
|
||||
t.Fatalf("Run took %v against a stalled server; the deadline is 400 ms", took)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientRefusesAContextWithoutDeadline(t *testing.T) {
|
||||
if _, err := (Client{}).Run(context.Background(), "127.0.0.1:1"); err == nil {
|
||||
t.Fatal("Run without a deadline must refuse (an unbounded test is the thing RunTimeout exists to prevent)")
|
||||
}
|
||||
}
|
||||
|
||||
// ── the schedule ────────────────────────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
func TestEligibleWindowNeverInTheNightChain(t *testing.T) {
|
||||
loc := time.FixedZone("CET", 3600)
|
||||
start := time.Date(2026, 10, 19, 0, 0, 0, 0, loc) // a Monday
|
||||
for m := 0; m < 7*24*60; m++ {
|
||||
lt := start.Add(time.Duration(m) * time.Minute)
|
||||
for _, anchorDay := range []bool{true, false} {
|
||||
if !EligibleAt(lt, 0, anchorDay) {
|
||||
continue
|
||||
}
|
||||
if h := lt.Hour(); h < StartHour || h >= EndHour || (h >= 1 && h <= 6) {
|
||||
t.Fatalf("eligible at %s — outside 10:00-15:59 or inside the night chain (01:00-06:59)", lt.Format("Mon 15:04"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSlotDelaysTheFirstStartOnTheDueDayOnly(t *testing.T) {
|
||||
loc := time.UTC
|
||||
tue := time.Date(2026, 10, 20, 10, 30, 0, 0, loc)
|
||||
if EligibleAt(tue, 45, true) {
|
||||
t.Fatal("10:30 with slot 45 must wait until 10:45 on the due day")
|
||||
}
|
||||
if !EligibleAt(tue.Add(15*time.Minute), 45, true) {
|
||||
t.Fatal("10:45 with slot 45 must be eligible")
|
||||
}
|
||||
if !EligibleAt(tue.Add(24*time.Hour), 45, false) {
|
||||
t.Fatal("a catch-up day does not wait for the slot again")
|
||||
}
|
||||
if a, b := SlotMinute("demo-hp-bb76ea"), SlotMinute("demo-felhom-8363b5"); a < 0 || a >= SlotSpreadMinutes || b < 0 || b >= SlotSpreadMinutes {
|
||||
t.Fatalf("slots out of range: %d %d", a, b)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDueWeeklyFromTuesday(t *testing.T) {
|
||||
loc := time.UTC
|
||||
mon := time.Date(2026, 10, 19, 12, 0, 0, 0, loc)
|
||||
tue := time.Date(2026, 10, 20, 12, 0, 0, 0, loc)
|
||||
if !Due(State{}, mon) {
|
||||
t.Fatal("never measured is due")
|
||||
}
|
||||
st := State{LastOKAt: time.Date(2026, 10, 14, 11, 0, 0, 0, loc)} // last Wednesday
|
||||
if Due(st, mon) {
|
||||
t.Fatal("measured after last Tuesday 10:00 is not due on Monday")
|
||||
}
|
||||
if !Due(st, tue) {
|
||||
t.Fatal("from Tuesday 10:00 a new measurement is due")
|
||||
}
|
||||
}
|
||||
|
||||
// ── the job ─────────────────────────────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
type fakeRunner struct {
|
||||
mu sync.Mutex
|
||||
calls int
|
||||
out []runOut
|
||||
}
|
||||
type runOut struct {
|
||||
mbps float64
|
||||
err error
|
||||
}
|
||||
|
||||
func (f *fakeRunner) Run(ctx context.Context, addr string) (Result, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
i := f.calls
|
||||
f.calls++
|
||||
if i >= len(f.out) {
|
||||
return Result{}, errors.New("no more scripted runs")
|
||||
}
|
||||
return Result{Mbps: f.out[i].mbps, ServerBytes: 1}, f.out[i].err
|
||||
}
|
||||
|
||||
type fakeGate struct{ busy string }
|
||||
|
||||
func (g *fakeGate) TryAcquire(what string) (func(), string, bool) {
|
||||
if g.busy != "" {
|
||||
return nil, g.busy, false
|
||||
}
|
||||
return func() {}, "", true
|
||||
}
|
||||
|
||||
type rig struct {
|
||||
t *Tester
|
||||
run *fakeRunner
|
||||
gate *fakeGate
|
||||
now time.Time
|
||||
tx uint64
|
||||
tunnel bool
|
||||
}
|
||||
|
||||
// newRig: Tuesday 12:00 UTC (eligible, any slot), tunnel up, 11 quiet samples taken.
|
||||
func newRig(t *testing.T, outs ...runOut) *rig {
|
||||
r := &rig{run: &fakeRunner{out: outs}, gate: &fakeGate{}, now: time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC), tunnel: true}
|
||||
r.t = New(r.run, func() (netip.Addr, bool) { return netip.MustParseAddr("10.77.0.1"), r.tunnel }, r.gate,
|
||||
func() (uint64, error) { return r.tx, nil }, "host-x", filepath.Join(t.TempDir(), "line-test.json"), nil)
|
||||
r.t.loc = time.UTC
|
||||
r.t.now = func() time.Time { return r.now }
|
||||
r.t.sleep = func(context.Context, time.Duration) bool { return true }
|
||||
r.quietMinutes(QuietMinutes + 1)
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *rig) minute(bytes uint64) {
|
||||
r.now = r.now.Add(time.Minute)
|
||||
r.tx += bytes
|
||||
r.t.sample()
|
||||
}
|
||||
|
||||
func (r *rig) quietMinutes(n int) {
|
||||
for i := 0; i < n; i++ {
|
||||
r.minute(1_000_000) // 1 MB/min = 0.13 Mbit/s
|
||||
}
|
||||
}
|
||||
|
||||
func TestTickMeasuresTheHighestOfThreeRuns(t *testing.T) {
|
||||
r := newRig(t, runOut{mbps: 88}, runOut{mbps: 91.5}, runOut{mbps: 90})
|
||||
r.t.Tick(context.Background())
|
||||
mbps, _, ok := r.t.Value()
|
||||
if !ok || mbps != 91.5 || r.run.calls != RunsPerMeasurement {
|
||||
t.Fatalf("value %.1f ok=%v calls=%d; want the highest of 3 runs = 91.5", mbps, ok, r.run.calls)
|
||||
}
|
||||
r.t.Tick(context.Background())
|
||||
if r.run.calls != RunsPerMeasurement {
|
||||
t.Fatal("a measured week must not test again")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusyServerIsRetriedLaterTheSameDay(t *testing.T) {
|
||||
r := newRig(t, runOut{err: ErrBusy}, runOut{mbps: 90}, runOut{mbps: 89}, runOut{mbps: 92})
|
||||
r.t.Tick(context.Background())
|
||||
if _, _, ok := r.t.Value(); ok || r.run.calls != 1 {
|
||||
t.Fatalf("busy: ok=%v calls=%d; want no value and the measurement stopped after the busy answer", ok, r.run.calls)
|
||||
}
|
||||
if r.t.st.TriesToday != 1 || r.t.st.LastError == "" {
|
||||
t.Fatalf("busy attempt not recorded: %+v", r.t.st)
|
||||
}
|
||||
r.quietMinutes(1)
|
||||
r.t.Tick(context.Background())
|
||||
if mbps, _, ok := r.t.Value(); !ok || mbps != 92 {
|
||||
t.Fatalf("retry later the same day: value %.1f ok=%v", mbps, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTriesPerDayAreBounded(t *testing.T) {
|
||||
outs := make([]runOut, 20)
|
||||
for i := range outs {
|
||||
outs[i] = runOut{err: ErrBusy}
|
||||
}
|
||||
r := newRig(t, outs...)
|
||||
for i := 0; i < 10; i++ {
|
||||
r.quietMinutes(1)
|
||||
r.t.Tick(context.Background())
|
||||
}
|
||||
if r.run.calls != MaxTriesPerDay {
|
||||
t.Fatalf("calls = %d in one day, want at most %d", r.run.calls, MaxTriesPerDay)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoTunnelNoTest(t *testing.T) {
|
||||
r := newRig(t, runOut{mbps: 90})
|
||||
r.tunnel = false
|
||||
r.t.Tick(context.Background())
|
||||
if r.run.calls != 0 {
|
||||
t.Fatal("tested without an operational tunnel")
|
||||
}
|
||||
if _, _, ok := r.t.Value(); ok {
|
||||
t.Fatal("a value without a test")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusyUplinkOrHeavyOpWaits(t *testing.T) {
|
||||
r := newRig(t, runOut{mbps: 90}, runOut{mbps: 90}, runOut{mbps: 90})
|
||||
r.minute(60_000_000) // one minute at 8 Mbit/s: the family is uploading
|
||||
r.t.Tick(context.Background())
|
||||
if r.run.calls != 0 {
|
||||
t.Fatal("tested while the uplink was busy in the last 10 minutes")
|
||||
}
|
||||
r.quietMinutes(QuietMinutes)
|
||||
r.gate.busy = "backup:felhom-pbs"
|
||||
r.t.Tick(context.Background())
|
||||
if r.run.calls != 0 {
|
||||
t.Fatal("tested beside a heavy operation")
|
||||
}
|
||||
r.gate.busy = ""
|
||||
r.t.Tick(context.Background())
|
||||
if r.run.calls == 0 {
|
||||
t.Fatal("never tested once the uplink was quiet and the gate free")
|
||||
}
|
||||
}
|
||||
|
||||
func TestNeverTestsInTheNightOrBeforeTuesday(t *testing.T) {
|
||||
r := newRig(t, runOut{mbps: 90}, runOut{mbps: 90}, runOut{mbps: 90})
|
||||
r.now = time.Date(2026, 10, 20, 3, 0, 0, 0, time.UTC) // Tuesday 03:00 — the night chain
|
||||
r.quietMinutes(QuietMinutes + 1)
|
||||
r.t.Tick(context.Background())
|
||||
r.t.st.LastOKAt = time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
|
||||
r.now = time.Date(2026, 10, 19, 12, 0, 0, 0, time.UTC) // Monday noon, measured last week
|
||||
r.quietMinutes(QuietMinutes + 1)
|
||||
r.t.Tick(context.Background())
|
||||
if r.run.calls != 0 {
|
||||
t.Fatalf("tested %d times at night or when not due", r.run.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStateSurvivesARestartAndReportsIt(t *testing.T) {
|
||||
r := newRig(t, runOut{mbps: 90.7}, runOut{err: errors.New("reset")}, runOut{mbps: 80})
|
||||
r.t.Tick(context.Background())
|
||||
again := New(r.run, nil, r.gate, nil, "host-x", r.t.statePath, nil)
|
||||
mbps, at, ok := again.Value()
|
||||
if !ok || mbps != 90.7 || at.IsZero() {
|
||||
t.Fatalf("after a restart: %.1f %v %v", mbps, at, ok)
|
||||
}
|
||||
b, _ := json.Marshal(again.LineTestStatus(context.Background()))
|
||||
var back hub.LineTestStatus
|
||||
json.Unmarshal(b, &back)
|
||||
if back.Mbps != 90.7 || back.LastOKAt == "" || len(back.RunsMbps) != 2 || back.Schedule == "" {
|
||||
t.Fatalf("report stanza = %s", b)
|
||||
}
|
||||
}
|
||||
|
||||
// ── the uplink counter ──────────────────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
func TestUplinkCountsTheBridgesPhysicalPortsOnly(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
w := func(p, s string) {
|
||||
os.MkdirAll(filepath.Dir(filepath.Join(root, p)), 0o755)
|
||||
os.WriteFile(filepath.Join(root, p), []byte(s), 0o644)
|
||||
}
|
||||
w("proc/net/route", "Iface\tDestination\tGateway\tFlags\tRefCnt\tUse\tMetric\tMask\n"+
|
||||
"wg-felhom\t0000004D\t00000000\t0001\t0\t0\t0\tFFFFFFFF\n"+
|
||||
"vmbr0\t00000000\t0100A8C0\t0003\t0\t0\t0\t00000000\n")
|
||||
w("sys/class/net/vmbr0/statistics/tx_bytes", "5\n")
|
||||
w("sys/class/net/vmbr0/brif/enp1s0/.keep", "")
|
||||
w("sys/class/net/vmbr0/brif/veth9201i0/.keep", "")
|
||||
w("sys/class/net/enp1s0/device/.keep", "")
|
||||
w("sys/class/net/enp1s0/statistics/tx_bytes", "1000\n")
|
||||
w("sys/class/net/veth9201i0/statistics/tx_bytes", "777777\n")
|
||||
got, err := SysUplinkTx(root)()
|
||||
if err != nil || got != 1000 {
|
||||
t.Fatalf("uplink tx = %d, %v; want 1000 (the physical port only — not the bridge, not the guest's veth)", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQuietNeedsTenFullMinutes(t *testing.T) {
|
||||
base := time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC)
|
||||
var s []txSample
|
||||
for i := 0; i <= QuietMinutes; i++ {
|
||||
s = append(s, txSample{at: base.Add(time.Duration(i) * time.Minute), tx: uint64(i) * 1_000_000})
|
||||
}
|
||||
if q, _ := quiet(s); !q {
|
||||
t.Fatal("ten quiet minutes must read quiet")
|
||||
}
|
||||
if q, _ := quiet(s[1:]); q {
|
||||
t.Fatal("nine minutes of samples must not read quiet")
|
||||
}
|
||||
gap := append([]txSample(nil), s...)
|
||||
gap[5].at = gap[5].at.Add(-10 * time.Minute)
|
||||
if q, _ := quiet(gap); q {
|
||||
t.Fatal("a gap in the samples must not read quiet")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package localapi
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
// LineTestReader is the R-1247 weekly upload line test (internal/linetest.Tester) as the local API reads it.
|
||||
type LineTestReader interface {
|
||||
Value() (mbps float64, at time.Time, ok bool)
|
||||
}
|
||||
|
||||
// LineTestResponse is GET /host/line-test. Present=false: never measured, or the job is not wired (no tunnel, disabled,
|
||||
// or an agent before the route answers 404 — the controller reads all three as „unknown").
|
||||
type LineTestResponse struct {
|
||||
Present bool `json:"present"`
|
||||
Mbps float64 `json:"mbps,omitempty"`
|
||||
MeasuredAt string `json:"measured_at,omitempty"` // RFC3339 UTC
|
||||
Method string `json:"method,omitempty"` // "iperf3-tunnel": a 10-second upload to ep0 inside the tunnel
|
||||
}
|
||||
|
||||
// SetLineTest wires the weekly line test (main.go builds it after the local API server).
|
||||
func (s *Server) SetLineTest(r LineTestReader) { s.lineTest = r }
|
||||
|
||||
func (s *Server) handleLineTest(w http.ResponseWriter, r *http.Request, vmid int) {
|
||||
if s.lineTest == nil {
|
||||
writeOK(w, LineTestResponse{Present: false})
|
||||
return
|
||||
}
|
||||
mbps, at, ok := s.lineTest.Value()
|
||||
if !ok {
|
||||
writeOK(w, LineTestResponse{Present: false})
|
||||
return
|
||||
}
|
||||
s.logger.Debug("local-api: /host/line-test served", "vmid", vmid, "mbps", mbps, "measured_at", at.UTC().Format(time.RFC3339))
|
||||
writeOK(w, LineTestResponse{Present: true, Mbps: mbps, MeasuredAt: at.UTC().Format(time.RFC3339), Method: "iperf3-tunnel"})
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package localapi
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type fixedLine struct {
|
||||
mbps float64
|
||||
ok bool
|
||||
}
|
||||
|
||||
func (f fixedLine) Value() (float64, time.Time, bool) {
|
||||
return f.mbps, time.Date(2026, 10, 24, 6, 31, 0, 0, time.UTC), f.ok
|
||||
}
|
||||
|
||||
func lineTestBody(t *testing.T, s *Server) LineTestResponse {
|
||||
t.Helper()
|
||||
rec := httptest.NewRecorder()
|
||||
s.handleLineTest(rec, httptest.NewRequest("GET", "/host/line-test", nil), 9201)
|
||||
var env struct {
|
||||
OK bool `json:"ok"`
|
||||
Data LineTestResponse `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil {
|
||||
t.Fatalf("decode %q: %v", rec.Body.String(), err)
|
||||
}
|
||||
return env.Data
|
||||
}
|
||||
|
||||
// R-1247: the controller sets qBittorrent's limit only from a PRESENT value; never measured and not wired must both
|
||||
// read present:false, never a zero line.
|
||||
func TestLineTestRoute(t *testing.T) {
|
||||
s := &Server{logger: slog.Default()}
|
||||
if got := lineTestBody(t, s); got.Present {
|
||||
t.Fatalf("not wired: %+v, want present:false", got)
|
||||
}
|
||||
s.SetLineTest(fixedLine{ok: false})
|
||||
if got := lineTestBody(t, s); got.Present || got.Mbps != 0 {
|
||||
t.Fatalf("never measured: %+v, want present:false", got)
|
||||
}
|
||||
s.SetLineTest(fixedLine{mbps: 90.7, ok: true})
|
||||
got := lineTestBody(t, s)
|
||||
if !got.Present || got.Mbps != 90.7 || got.Method != "iperf3-tunnel" || !strings.HasPrefix(got.MeasuredAt, "2026-10-24T06:31") {
|
||||
t.Fatalf("measured: %+v", got)
|
||||
}
|
||||
}
|
||||
@@ -336,6 +336,7 @@ type Server struct {
|
||||
formatJobs *FormatJobStore // F20-BUG3 detached-format job record (optional)
|
||||
staleLock StaleLockController // F2-b startup stale-lock recovery (optional)
|
||||
partialSweep func(ctx context.Context, vmid int) // R-1006 start-up sweep of dead vzdump leftovers (optional)
|
||||
lineTest LineTestReader // R-1247 weekly line test (optional; nil → present:false)
|
||||
// guestPower (F-REBOOT) is per-guest start-attempt state for the guest-power watchdog.
|
||||
// Guarded by guestPowerMu in guestpower.go; in-memory on purpose (see guestPowerState).
|
||||
guestPower map[int]guestPowerState
|
||||
@@ -573,6 +574,9 @@ func (s *Server) Handler() http.Handler {
|
||||
// R-856 (`09` §3 decision 143): the host crash guard's record of the last HOST boot — the controller
|
||||
// waits ~15 min with app mails after an unclean one. Read-only; a missing/garbled file = present:false.
|
||||
mux.HandleFunc("GET /host/crash-guard", s.withGuest(s.handleCrashGuard))
|
||||
// R-1247: the weekly upload line test's last measurement — the controller's qBittorrent limit and upload watch
|
||||
// read it. Read-only; never measured (or the job not wired) = present:false.
|
||||
mux.HandleFunc("GET /host/line-test", s.withGuest(s.handleLineTest))
|
||||
// Disk management (slice 8C) — self-scoped; format routes through the data-bearing classifier+gate.
|
||||
mux.HandleFunc("GET /disks", s.withGuest(s.handleDisks))
|
||||
mux.HandleFunc("GET /disks/candidates", s.withGuest(s.handleDiskCandidates))
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
package wgtunnel
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// R-1247: the line test's server is the desired block's pbs_tunnel_ip, and ONLY while the tunnel is operational
|
||||
// (registered + unit active) — „no tunnel → no test" rests on this answer.
|
||||
func TestLineTestTargetOnlyWhileOperational(t *testing.T) {
|
||||
m, _, active, _ := testManager(t, &fakeHub{})
|
||||
ctx := context.Background()
|
||||
if _, ok := m.LineTestTarget(); ok {
|
||||
t.Fatal("a target before any Apply")
|
||||
}
|
||||
m.Apply(ctx, false, nil) // registers
|
||||
pub := localPub(t, m)
|
||||
m.Apply(ctx, true, testBlock(pub)) // installs; the unit is not active yet
|
||||
if _, ok := m.LineTestTarget(); ok {
|
||||
t.Fatal("a target while the unit is not active")
|
||||
}
|
||||
*active = true
|
||||
m.Apply(ctx, true, testBlock(pub))
|
||||
ip, ok := m.LineTestTarget()
|
||||
if !ok || ip.String() != "10.77.0.1" {
|
||||
t.Fatalf("operational tunnel: target %v ok=%v, want 10.77.0.1", ip, ok)
|
||||
}
|
||||
m.Apply(ctx, true, nil) // revoked
|
||||
if _, ok := m.LineTestTarget(); ok {
|
||||
t.Fatal("a target after the hub revoked the tunnel")
|
||||
}
|
||||
}
|
||||
@@ -128,6 +128,7 @@ type Manager struct {
|
||||
type convSnapshot struct {
|
||||
wgBlockDesired bool
|
||||
operational bool
|
||||
serverIP netip.Addr // the desired block's pbs_tunnel_ip (R-1247 line test target); invalid when none
|
||||
}
|
||||
|
||||
// TunnelConvergence returns the cached (desired, operational) snapshot without any exec (fast-tick
|
||||
@@ -139,6 +140,17 @@ func (m *Manager) TunnelConvergence() (desired, operational bool) {
|
||||
return m.conv.wgBlockDesired, m.conv.operational
|
||||
}
|
||||
|
||||
// LineTestTarget is the R-1247 line test's server: ep0's tunnel address, ok only while the tunnel is operational
|
||||
// (registered + unit active, as of the last Apply). Cached — no exec.
|
||||
func (m *Manager) LineTestTarget() (netip.Addr, bool) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if !m.conv.operational || !m.conv.serverIP.IsValid() {
|
||||
return netip.Addr{}, false
|
||||
}
|
||||
return m.conv.serverIP, true
|
||||
}
|
||||
|
||||
// NewManager builds a Manager. stateDir is the agent state dir (default /var/lib/felhom-agent —
|
||||
// note the FELHOM_WG sudoers install entry hard-codes the staged path under it).
|
||||
func NewManager(runner proxmox.Runner, registrar Registrar, stateDir string, logger *slog.Logger) *Manager {
|
||||
@@ -348,7 +360,13 @@ func (m *Manager) Apply(ctx context.Context, fetched bool, block *hub.WireWiregu
|
||||
operational = m.isActive(ctx)
|
||||
}
|
||||
}
|
||||
m.conv = convSnapshot{wgBlockDesired: desired, operational: operational}
|
||||
var server netip.Addr
|
||||
if desired {
|
||||
if a, err := netip.ParseAddr(block.Endpoint.PBSTunnelIP); err == nil && a.Is4() {
|
||||
server = a
|
||||
}
|
||||
}
|
||||
m.conv = convSnapshot{wgBlockDesired: desired, operational: operational, serverIP: server}
|
||||
}()
|
||||
|
||||
keyExists := false
|
||||
|
||||
Reference in New Issue
Block a user