fix(webui): make metrics history immune to system-clock changes
Every sample was stamped with time.Now(), so a timezone switch or NTP step punched a multi-hour gap into the series: old points collapsed to the left edge, new points bunched at the right, joined by one diagonal. Stamp rows from a monotonic seqClock instead — seeded once from the wall clock (or the newest persisted row) and thereafter advanced only by the monotonic elapsed time between writes. Rebase Downsample/Prune on the newest sample rather than time.Now() so a clock step just before the hourly compaction can't drop fresh data. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GofKhuF9xQHaz3UFfncR6D
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
4953bb33b2
commit
f31971f440
@@ -4,11 +4,13 @@ import (
|
|||||||
"database/sql"
|
"database/sql"
|
||||||
"encoding/csv"
|
"encoding/csv"
|
||||||
"io"
|
"io"
|
||||||
|
"math"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"bee/audit/internal/platform"
|
"bee/audit/internal/platform"
|
||||||
@@ -20,6 +22,67 @@ const metricsDBPath = "/appdata/bee/metrics.db"
|
|||||||
// MetricsDB persists live metric samples to SQLite.
|
// MetricsDB persists live metric samples to SQLite.
|
||||||
type MetricsDB struct {
|
type MetricsDB struct {
|
||||||
db *sql.DB
|
db *sql.DB
|
||||||
|
clock seqClock
|
||||||
|
}
|
||||||
|
|
||||||
|
// seqClock generates the timestamps stored on metric rows. It is seeded once
|
||||||
|
// from the wall clock (or the newest persisted row) and thereafter only ever
|
||||||
|
// advances by the *monotonic* elapsed time between Write calls. A system-clock
|
||||||
|
// step — an NTP correction, a timezone change, a manual `date` — can therefore
|
||||||
|
// never punch a gap into the history or send timestamps backwards, which is
|
||||||
|
// what used to smear every chart into a single diagonal line after the clock
|
||||||
|
// was changed. Charts count backwards from the newest sample when rendering,
|
||||||
|
// so the absolute value of the seed does not matter; only the spacing does.
|
||||||
|
type seqClock struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
lastReal time.Time // wall reading from the previous tick (carries a monotonic component)
|
||||||
|
cur float64 // current device time, unix-seconds
|
||||||
|
lastInt int64 // last integer ts handed out, to keep the PK strictly increasing
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *seqClock) seed(base float64) {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
c.lastReal = time.Now()
|
||||||
|
c.cur = base
|
||||||
|
// One below floor(base) so the first tick (near-zero elapsed) can still
|
||||||
|
// hand out floor(base) itself instead of being bumped forward.
|
||||||
|
c.lastInt = int64(math.Floor(base)) - 1
|
||||||
|
}
|
||||||
|
|
||||||
|
// tick advances the clock by the monotonic time elapsed since the previous
|
||||||
|
// call and returns the next row timestamp (unix-seconds, strictly increasing).
|
||||||
|
func (c *seqClock) tick() int64 {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
c.advanceLocked()
|
||||||
|
ts := int64(math.Floor(c.cur)) // truncate, matching time.Time.Unix semantics
|
||||||
|
if ts <= c.lastInt {
|
||||||
|
ts = c.lastInt + 1
|
||||||
|
}
|
||||||
|
c.lastInt = ts
|
||||||
|
return ts
|
||||||
|
}
|
||||||
|
|
||||||
|
// deviceNow returns the current device time, advancing the clock so the
|
||||||
|
// wall↔device relationship stays anchored to the latest observation.
|
||||||
|
func (c *seqClock) deviceNow() float64 {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
c.advanceLocked()
|
||||||
|
return c.cur
|
||||||
|
}
|
||||||
|
|
||||||
|
// advanceLocked moves cur forward by the monotonic time elapsed since the
|
||||||
|
// previous observation. c.mu must be held.
|
||||||
|
func (c *seqClock) advanceLocked() {
|
||||||
|
now := time.Now()
|
||||||
|
elapsed := now.Sub(c.lastReal).Seconds() // monotonic: immune to wall-clock steps
|
||||||
|
if elapsed < 0 || elapsed > 24*3600 {
|
||||||
|
elapsed = 0
|
||||||
|
}
|
||||||
|
c.lastReal = now
|
||||||
|
c.cur += elapsed
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *MetricsDB) Close() error {
|
func (m *MetricsDB) Close() error {
|
||||||
@@ -43,7 +106,15 @@ func openMetricsDB(path string) (*MetricsDB, error) {
|
|||||||
_ = db.Close()
|
_ = db.Close()
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
return &MetricsDB{db: db}, nil
|
m := &MetricsDB{db: db}
|
||||||
|
var maxTS sql.NullInt64
|
||||||
|
_ = db.QueryRow(`SELECT MAX(ts) FROM sys_metrics`).Scan(&maxTS)
|
||||||
|
if maxTS.Valid && maxTS.Int64 > 0 {
|
||||||
|
m.clock.seed(float64(maxTS.Int64))
|
||||||
|
} else {
|
||||||
|
m.clock.seed(float64(time.Now().UnixNano()) / 1e9)
|
||||||
|
}
|
||||||
|
return m, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func initMetricsSchema(db *sql.DB) error {
|
func initMetricsSchema(db *sql.DB) error {
|
||||||
@@ -127,9 +198,16 @@ func ensureMetricsColumn(db *sql.DB, table, column, definition string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Write inserts one sample into all relevant tables.
|
// Write inserts one sample into all relevant tables. The row timestamp is
|
||||||
|
// taken from the monotonic seqClock, not from s.Timestamp, so a system-clock
|
||||||
|
// change never corrupts the stored series.
|
||||||
func (m *MetricsDB) Write(s platform.LiveMetricSample) error {
|
func (m *MetricsDB) Write(s platform.LiveMetricSample) error {
|
||||||
ts := s.Timestamp.Unix()
|
return m.writeAt(m.clock.tick(), s)
|
||||||
|
}
|
||||||
|
|
||||||
|
// writeAt inserts one sample at an explicit timestamp. Production code uses
|
||||||
|
// Write; writeAt exists so tests can lay down a series with known spacing.
|
||||||
|
func (m *MetricsDB) writeAt(ts int64, s platform.LiveMetricSample) error {
|
||||||
tx, err := m.db.Begin()
|
tx, err := m.db.Begin()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -174,19 +252,22 @@ func (m *MetricsDB) Write(s platform.LiveMetricSample) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Downsample reduces density of old metrics rows to 1 sample per minute.
|
// Downsample reduces density of old metrics rows to 1 sample per minute.
|
||||||
// Only rows in the half-open window [deleteOlderThan, downsampleBefore) are
|
// Rows older than downsampleAge (but within retain) are thinned; rows newer
|
||||||
// affected — rows newer than downsampleBefore keep full 5-second resolution.
|
// than downsampleAge keep full 5-second resolution. For each 60-second bucket
|
||||||
// For each 60-second bucket the row with the smallest ts is kept; the rest
|
// the row with the smallest ts is kept; the rest are deleted. This trims ~92 %
|
||||||
// are deleted. This trims ~92 % of rows in that window while preserving
|
// of rows in that window while preserving the overall shape of every chart.
|
||||||
// the overall shape of every chart.
|
//
|
||||||
|
// The window is measured back from the newest sample (the device clock), not
|
||||||
|
// from the wall clock, so it is unaffected by system-clock changes.
|
||||||
//
|
//
|
||||||
// Called hourly by the metrics collector background goroutine.
|
// Called hourly by the metrics collector background goroutine.
|
||||||
func (m *MetricsDB) Downsample(downsampleBefore, deleteOlderThan time.Time) error {
|
func (m *MetricsDB) Downsample(downsampleAge, retain time.Duration) error {
|
||||||
if m == nil || m.db == nil {
|
if m == nil || m.db == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
start := deleteOlderThan.Unix()
|
now := m.clock.deviceNow()
|
||||||
end := downsampleBefore.Unix()
|
start := int64(now - retain.Seconds())
|
||||||
|
end := int64(now - downsampleAge.Seconds())
|
||||||
if end <= start {
|
if end <= start {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -207,13 +288,14 @@ DELETE FROM `+table+` WHERE ts >= ? AND ts < ?
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Prune deletes all rows older than the given cutoff from every metrics table.
|
// Prune deletes rows older than retain (measured back from the newest sample,
|
||||||
|
// so it is unaffected by system-clock changes) from every metrics table.
|
||||||
// Called hourly by the metrics collector to keep the DB size bounded.
|
// Called hourly by the metrics collector to keep the DB size bounded.
|
||||||
func (m *MetricsDB) Prune(before time.Time) error {
|
func (m *MetricsDB) Prune(retain time.Duration) error {
|
||||||
if m == nil || m.db == nil {
|
if m == nil || m.db == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
cutTS := before.Unix()
|
cutTS := int64(m.clock.deviceNow() - retain.Seconds())
|
||||||
for _, table := range []string{"sys_metrics", "gpu_metrics", "fan_metrics", "temp_metrics"} {
|
for _, table := range []string{"sys_metrics", "gpu_metrics", "fan_metrics", "temp_metrics"} {
|
||||||
if _, err := m.db.Exec("DELETE FROM "+table+" WHERE ts < ?", cutTS); err != nil {
|
if _, err := m.db.Exec("DELETE FROM "+table+" WHERE ts < ?", cutTS); err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -244,6 +326,10 @@ func (m *MetricsDB) LoadBetween(start, end time.Time) ([]platform.LiveMetricSamp
|
|||||||
if end.Before(start) {
|
if end.Before(start) {
|
||||||
start, end = end, start
|
start, end = end, start
|
||||||
}
|
}
|
||||||
|
// Row timestamps track the monotonic device clock, which is seeded from the
|
||||||
|
// wall clock and advances at the real rate, so ts stays aligned with wall
|
||||||
|
// time for the task-window correlation done here. (System-clock *steps* are
|
||||||
|
// absorbed by the device clock rather than corrupting the series.)
|
||||||
return m.loadSamples(
|
return m.loadSamples(
|
||||||
`SELECT ts,cpu_load_pct,mem_load_pct,power_w,IFNULL(power_source,''),IFNULL(power_mode,''),IFNULL(power_reason,'') FROM sys_metrics WHERE ts>=? AND ts<=? ORDER BY ts`,
|
`SELECT ts,cpu_load_pct,mem_load_pct,power_w,IFNULL(power_source,''),IFNULL(power_mode,''),IFNULL(power_reason,'') FROM sys_metrics WHERE ts>=? AND ts<=? ORDER BY ts`,
|
||||||
start.Unix(), end.Unix(),
|
start.Unix(), end.Unix(),
|
||||||
|
|||||||
@@ -153,11 +153,12 @@ func TestMetricsDBLoadBetweenFiltersWindow(t *testing.T) {
|
|||||||
|
|
||||||
base := time.Unix(1_700_000_000, 0).UTC()
|
base := time.Unix(1_700_000_000, 0).UTC()
|
||||||
for i := 0; i < 5; i++ {
|
for i := 0; i < 5; i++ {
|
||||||
if err := db.Write(platform.LiveMetricSample{
|
ts := base.Add(time.Duration(i) * time.Minute)
|
||||||
Timestamp: base.Add(time.Duration(i) * time.Minute),
|
if err := db.writeAt(ts.Unix(), platform.LiveMetricSample{
|
||||||
|
Timestamp: ts,
|
||||||
CPULoadPct: float64(i),
|
CPULoadPct: float64(i),
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
t.Fatalf("Write(%d): %v", i, err)
|
t.Fatalf("writeAt(%d): %v", i, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -393,9 +393,8 @@ func (h *handler) startMetricsCollector() {
|
|||||||
h.setLatestMetric(sample)
|
h.setLatestMetric(sample)
|
||||||
case <-pruneTicker.C:
|
case <-pruneTicker.C:
|
||||||
if h.metricsDB != nil {
|
if h.metricsDB != nil {
|
||||||
now := time.Now().UTC()
|
_ = h.metricsDB.Downsample(metricsDownsampleAge, metricsRetainWindow)
|
||||||
_ = h.metricsDB.Downsample(now.Add(-metricsDownsampleAge), now.Add(-metricsRetainWindow))
|
_ = h.metricsDB.Prune(metricsRetainWindow)
|
||||||
_ = h.metricsDB.Prune(now.Add(-metricsRetainWindow))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -969,8 +969,8 @@ func TestTaskChartSVGUsesTaskTimeWindow(t *testing.T) {
|
|||||||
{Timestamp: base.Add(-1 * time.Minute), PowerW: 300},
|
{Timestamp: base.Add(-1 * time.Minute), PowerW: 300},
|
||||||
}
|
}
|
||||||
for _, sample := range samples {
|
for _, sample := range samples {
|
||||||
if err := db.Write(sample); err != nil {
|
if err := db.writeAt(sample.Timestamp.Unix(), sample); err != nil {
|
||||||
t.Fatalf("Write: %v", err)
|
t.Fatalf("writeAt: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
_ = db.Close()
|
_ = db.Close()
|
||||||
|
|||||||
Reference in New Issue
Block a user