Settings: one window for how far back every graph reaches, 1 minute to 24 hours
Each machine keeps 300 points per curve whatever the window, so memory doesn't follow it; a wider window means wider points, each the average of its bucket rather than one instant out of it. The hub passes the setting to every machine it manages. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Mitchell-Johnson
co-authored by
Claude Opus 5
Claude Opus 5
parent
885b050d02
commit
04dd98e2cd
@@ -0,0 +1,141 @@
|
||||
package metrics
|
||||
|
||||
// One kept history point is the average of the samples behind it. At the default ten-minute
|
||||
// window that bucket is a single sample and nothing is averaged; at a day it is five minutes
|
||||
// of them, and the difference matters — a graph of the last day drawn from whichever instant
|
||||
// landed on each five-minute mark would show spikes that never happened and miss the ones
|
||||
// that did.
|
||||
//
|
||||
// The newest sample of the bucket is the template: its shape, labels, totals and slow
|
||||
// sections (filesystems, UPS) are the current ones. Only the moving numbers are averaged,
|
||||
// and only where an entry is present in that sample — a disk plugged in halfway through the
|
||||
// bucket is averaged over the samples that have it.
|
||||
|
||||
func fold(b []Sample) Sample {
|
||||
out := b[len(b)-1]
|
||||
if len(b) == 1 {
|
||||
return out
|
||||
}
|
||||
n := float64(len(b))
|
||||
sum := func(get func(Sample) float64) float64 {
|
||||
var t float64
|
||||
for _, s := range b {
|
||||
t += get(s)
|
||||
}
|
||||
return t / n
|
||||
}
|
||||
sumU := func(get func(Sample) uint64) uint64 {
|
||||
var t uint64
|
||||
for _, s := range b {
|
||||
t += get(s)
|
||||
}
|
||||
return t / uint64(len(b))
|
||||
}
|
||||
out.CPU = sum(func(s Sample) float64 { return s.CPU })
|
||||
out.Load[0] = sum(func(s Sample) float64 { return s.Load[0] })
|
||||
out.Load[1] = sum(func(s Sample) float64 { return s.Load[1] })
|
||||
out.Load[2] = sum(func(s Sample) float64 { return s.Load[2] })
|
||||
|
||||
out.Cores = append([]float64(nil), out.Cores...)
|
||||
for i := range out.Cores {
|
||||
var t, c float64
|
||||
for _, s := range b {
|
||||
if i < len(s.Cores) {
|
||||
t, c = t+s.Cores[i], c+1
|
||||
}
|
||||
}
|
||||
if c > 0 {
|
||||
out.Cores[i] = t / c
|
||||
}
|
||||
}
|
||||
|
||||
out.Mem.Used = sumU(func(s Sample) uint64 { return s.Mem.Used })
|
||||
out.Mem.Avail = sumU(func(s Sample) uint64 { return s.Mem.Avail })
|
||||
out.Mem.Free = sumU(func(s Sample) uint64 { return s.Mem.Free })
|
||||
out.Mem.Cached = sumU(func(s Sample) uint64 { return s.Mem.Cached })
|
||||
out.Swap.Used = sumU(func(s Sample) uint64 { return s.Swap.Used })
|
||||
|
||||
// the rest are lists whose entries are matched by name across the bucket
|
||||
temps := avgBy(b, func(s Sample) map[string][]float64 {
|
||||
m := make(map[string][]float64, len(s.Temps))
|
||||
for _, x := range s.Temps {
|
||||
m[x.Chip+"/"+x.Label] = []float64{x.C}
|
||||
}
|
||||
return m
|
||||
})
|
||||
out.Temps = append([]Temp(nil), out.Temps...)
|
||||
for i, x := range out.Temps {
|
||||
if v, ok := temps[x.Chip+"/"+x.Label]; ok {
|
||||
out.Temps[i].C = v[0]
|
||||
}
|
||||
}
|
||||
|
||||
fans := avgBy(b, func(s Sample) map[string][]float64 {
|
||||
m := make(map[string][]float64, len(s.Fans))
|
||||
for _, x := range s.Fans {
|
||||
m[x.Chip+"/"+x.Label] = []float64{float64(x.RPM)}
|
||||
}
|
||||
return m
|
||||
})
|
||||
out.Fans = append([]Fan(nil), out.Fans...)
|
||||
for i, x := range out.Fans {
|
||||
if v, ok := fans[x.Chip+"/"+x.Label]; ok {
|
||||
out.Fans[i].RPM = int(v[0] + .5)
|
||||
}
|
||||
}
|
||||
|
||||
nets := avgBy(b, func(s Sample) map[string][]float64 {
|
||||
m := make(map[string][]float64, len(s.Net))
|
||||
for _, x := range s.Net {
|
||||
m[x.Iface] = []float64{x.RxBps, x.TxBps}
|
||||
}
|
||||
return m
|
||||
})
|
||||
out.Net = append([]Net(nil), out.Net...)
|
||||
for i, x := range out.Net {
|
||||
if v, ok := nets[x.Iface]; ok {
|
||||
out.Net[i].RxBps, out.Net[i].TxBps = v[0], v[1]
|
||||
}
|
||||
}
|
||||
|
||||
disks := avgBy(b, func(s Sample) map[string][]float64 {
|
||||
m := make(map[string][]float64, len(s.Disks))
|
||||
for _, x := range s.Disks {
|
||||
m[x.Name] = []float64{x.ReadBps, x.WriteBps, x.UtilPct}
|
||||
}
|
||||
return m
|
||||
})
|
||||
out.Disks = append([]Disk(nil), out.Disks...)
|
||||
for i, x := range out.Disks {
|
||||
if v, ok := disks[x.Name]; ok {
|
||||
out.Disks[i].ReadBps, out.Disks[i].WriteBps, out.Disks[i].UtilPct = v[0], v[1], v[2]
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// avgBy averages each sample's named numbers across the bucket, per name.
|
||||
func avgBy(b []Sample, of func(Sample) map[string][]float64) map[string][]float64 {
|
||||
sums := map[string][]float64{}
|
||||
count := map[string]float64{}
|
||||
for _, s := range b {
|
||||
for k, vs := range of(s) {
|
||||
acc := sums[k]
|
||||
if acc == nil {
|
||||
acc = make([]float64, len(vs))
|
||||
}
|
||||
for i, v := range vs {
|
||||
if i < len(acc) {
|
||||
acc[i] += v
|
||||
}
|
||||
}
|
||||
sums[k], count[k] = acc, count[k]+1
|
||||
}
|
||||
}
|
||||
for k, acc := range sums {
|
||||
for i := range acc {
|
||||
acc[i] /= count[k]
|
||||
}
|
||||
}
|
||||
return sums
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestFoldAverages(t *testing.T) {
|
||||
b := []Sample{
|
||||
{T: 1, CPU: 10, Cores: []float64{10, 0}, Mem: Mem{Total: 100, Used: 10},
|
||||
Temps: []Temp{{Chip: "c", Label: "l", C: 40}}, Net: []Net{{Iface: "eth0", RxBps: 100}},
|
||||
Disks: []Disk{{Name: "sda", ReadBps: 10, UtilPct: 5}}},
|
||||
{T: 2, CPU: 30, Cores: []float64{30, 100}, Mem: Mem{Total: 100, Used: 30},
|
||||
Temps: []Temp{{Chip: "c", Label: "l", C: 60}}, Net: []Net{{Iface: "eth0", RxBps: 300}, {Iface: "new", RxBps: 7}},
|
||||
Disks: []Disk{{Name: "sda", ReadBps: 30, UtilPct: 15}}},
|
||||
}
|
||||
f := fold(b)
|
||||
if f.T != 2 {
|
||||
t.Errorf("a folded point is stamped with the newest sample in it, got %d", f.T)
|
||||
}
|
||||
if f.CPU != 20 || f.Cores[0] != 20 || f.Cores[1] != 50 {
|
||||
t.Errorf("cpu averaged: %v %v", f.CPU, f.Cores)
|
||||
}
|
||||
if f.Mem.Used != 20 || f.Mem.Total != 100 {
|
||||
t.Errorf("memory: used averaged, total taken from the newest, got %+v", f.Mem)
|
||||
}
|
||||
if f.Temps[0].C != 50 || f.Net[0].RxBps != 200 || f.Disks[0].ReadBps != 20 || f.Disks[0].UtilPct != 10 {
|
||||
t.Errorf("lists matched by name: %v %v %v", f.Temps, f.Net, f.Disks)
|
||||
}
|
||||
if len(f.Net) != 2 || f.Net[1].RxBps != 7 {
|
||||
t.Errorf("an interface that appeared inside the bucket is averaged over the samples that have it: %v", f.Net)
|
||||
}
|
||||
// the template's own slices must not be touched: Latest() shares them
|
||||
if b[1].Cores[0] != 30 || b[1].Temps[0].C != 60 {
|
||||
t.Error("fold wrote through to the sample it templated from")
|
||||
}
|
||||
}
|
||||
|
||||
func TestStrideAndTrim(t *testing.T) {
|
||||
c := &Collector{interval: 2 * time.Second, window: DefaultWindow}
|
||||
if got := c.stride(); got != 2*time.Second {
|
||||
t.Errorf("a window of %v over %d points is finer than the interval, so the interval wins; got %v", DefaultWindow, History, got)
|
||||
}
|
||||
c.window = 24 * time.Hour
|
||||
if got := c.stride(); got != 24*time.Hour/History {
|
||||
t.Errorf("stride at a day: %v", got)
|
||||
}
|
||||
// a full day of samples at the day stride keeps exactly the budget
|
||||
c.interval, c.window = time.Second, time.Hour
|
||||
stride := c.stride().Milliseconds()
|
||||
for i := 0; i < 5*History; i++ {
|
||||
c.keep(Sample{T: int64(i) * stride})
|
||||
}
|
||||
if len(c.ring) != History {
|
||||
t.Errorf("the point count is the budget, got %d", len(c.ring))
|
||||
}
|
||||
if span := c.ring[len(c.ring)-1].T - c.ring[0].T; span > c.window.Milliseconds() {
|
||||
t.Errorf("the kept history reaches back further than the window: %d ms", span)
|
||||
}
|
||||
// narrowing the window drops what is now too old, at once
|
||||
c.SetWindow(MinWindow)
|
||||
if span := c.ring[len(c.ring)-1].T - c.ring[0].T; span > MinWindow.Milliseconds() {
|
||||
t.Errorf("narrowing the window left %d ms of history", span)
|
||||
}
|
||||
c.SetWindow(365 * 24 * time.Hour)
|
||||
if c.Window() != MaxWindow {
|
||||
t.Errorf("a window past the maximum is the maximum, got %v", c.Window())
|
||||
}
|
||||
}
|
||||
@@ -32,7 +32,12 @@ const (
|
||||
DefaultInterval = 2 * time.Second
|
||||
MinInterval = 100 * time.Millisecond
|
||||
MaxInterval = 10 * time.Second
|
||||
History = 300 // samples kept; the window is History × interval (30 s at 100 ms, 10 min at 2 s)
|
||||
History = 300 // points kept per curve; the window below sets how far apart they are
|
||||
// How far back the kept history reaches. The point count is fixed, so this sets the spacing:
|
||||
// at a minute every sample is its own point, at a day each point is the average of ~5 minutes.
|
||||
DefaultWindow = 10 * time.Minute
|
||||
MinWindow = 1 * time.Minute
|
||||
MaxWindow = 24 * time.Hour
|
||||
fsEvery = 30 * time.Second // filesystem usage changes slowly and statfs on network mounts isn't free
|
||||
upsEvery = 10 * time.Second // UPS daemons are an exec each; don't hammer them at fast intervals
|
||||
sectorSize = 512 // /proc/diskstats always counts 512-byte sectors
|
||||
@@ -138,8 +143,11 @@ type Sample struct {
|
||||
|
||||
type Collector struct {
|
||||
mu sync.RWMutex
|
||||
ring []Sample
|
||||
ring []Sample // the kept history: one point per stride, each the average of its bucket
|
||||
last Sample // the newest raw sample, whatever the stride
|
||||
bucket []Sample // samples since the last point was kept
|
||||
interval time.Duration
|
||||
window time.Duration
|
||||
wake chan struct{} // nudges Run when the interval changes
|
||||
|
||||
prevCPU []cpuTimes
|
||||
@@ -162,7 +170,7 @@ type Collector struct {
|
||||
|
||||
func NewCollector() *Collector {
|
||||
c := &Collector{prevNet: map[string][2]uint64{}, prevDisk: map[string][3]uint64{},
|
||||
interval: DefaultInterval, wake: make(chan struct{}, 1)}
|
||||
interval: DefaultInterval, window: DefaultWindow, wake: make(chan struct{}, 1)}
|
||||
switch {
|
||||
case lookPath("pwrstat"):
|
||||
c.upsProbe = upsPwrstat
|
||||
@@ -223,13 +231,80 @@ func (c *Collector) Interval() time.Duration {
|
||||
return c.interval
|
||||
}
|
||||
|
||||
// SetWindow sets how far back the kept history reaches, clamped to [MinWindow, MaxWindow].
|
||||
func (c *Collector) SetWindow(d time.Duration) {
|
||||
if d < MinWindow {
|
||||
d = MinWindow
|
||||
}
|
||||
if d > MaxWindow {
|
||||
d = MaxWindow
|
||||
}
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.window = d
|
||||
c.trim()
|
||||
}
|
||||
|
||||
func (c *Collector) Window() time.Duration {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
return c.window
|
||||
}
|
||||
|
||||
// Stride is how far apart the kept points are: the sampling interval until the window is
|
||||
// wider than History samples of it, then the window split History ways.
|
||||
func (c *Collector) Stride() time.Duration {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
return c.stride()
|
||||
}
|
||||
|
||||
func (c *Collector) stride() time.Duration {
|
||||
if s := c.window / History; s > c.interval {
|
||||
return s
|
||||
}
|
||||
return c.interval
|
||||
}
|
||||
|
||||
// keep folds samples into the history, one point per stride. Averaged, not sampled: a CPU
|
||||
// graph of the last six hours should show the load through each five minutes, not whichever
|
||||
// two-second instant happened to land on the mark.
|
||||
func (c *Collector) keep(s Sample) {
|
||||
c.bucket = append(c.bucket, s)
|
||||
if s.T-c.bucket[0].T < c.stride().Milliseconds() && len(c.bucket) > 1 {
|
||||
return
|
||||
}
|
||||
c.ring = append(c.ring, fold(c.bucket))
|
||||
c.bucket = c.bucket[:0]
|
||||
c.trim()
|
||||
}
|
||||
|
||||
// trim drops points past the count budget and past the window. Caller holds c.mu.
|
||||
func (c *Collector) trim() {
|
||||
if len(c.ring) > History {
|
||||
c.ring = c.ring[len(c.ring)-History:]
|
||||
}
|
||||
n := len(c.ring)
|
||||
if n == 0 {
|
||||
return
|
||||
}
|
||||
cut := c.ring[n-1].T - c.window.Milliseconds()
|
||||
i := 0
|
||||
for i < n && c.ring[i].T < cut {
|
||||
i++
|
||||
}
|
||||
if i > 0 {
|
||||
c.ring = c.ring[i:]
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Collector) Latest() (Sample, bool) {
|
||||
c.mu.RLock()
|
||||
defer c.mu.RUnlock()
|
||||
if len(c.ring) == 0 {
|
||||
if c.last.T == 0 {
|
||||
return Sample{}, false
|
||||
}
|
||||
return c.ring[len(c.ring)-1], true
|
||||
return c.last, true
|
||||
}
|
||||
|
||||
// ---- sampling ----
|
||||
@@ -265,10 +340,8 @@ func (c *Collector) sample() {
|
||||
return // deltas are meaningless on the priming pass
|
||||
}
|
||||
c.mu.Lock()
|
||||
c.ring = append(c.ring, s)
|
||||
if len(c.ring) > History {
|
||||
c.ring = c.ring[len(c.ring)-History:]
|
||||
}
|
||||
c.last = s
|
||||
c.keep(s)
|
||||
c.mu.Unlock()
|
||||
|
||||
c.subMu.Lock()
|
||||
|
||||
Reference in New Issue
Block a user