hamlab/internal/metrics/fold.go

package metrics

import "strconv"

// 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 })
	// CPU power: over the samples that have a reading, so a helper restart is not a dip to zero
	var pw, pn float64
	for _, s := range b {
		if s.CPUPower > 0 {
			pw, pn = pw+s.CPUPower, pn+1
		}
	}
	out.CPUPower = 0
	if pn > 0 {
		out.CPUPower = round1(pw / pn)
	}
	// and each package's, the same way
	out.CPUPowerPkgs = nil
	for _, s := range b {
		for i, w := range s.CPUPowerPkgs {
			for len(out.CPUPowerPkgs) <= i {
				out.CPUPowerPkgs = append(out.CPUPowerPkgs, 0)
			}
			out.CPUPowerPkgs[i] += w
		}
	}
	if pn > 0 {
		for i := range out.CPUPowerPkgs {
			out.CPUPowerPkgs[i] = round1(out.CPUPowerPkgs[i] / pn)
		}
	}
	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+"/"+strconv.Itoa(x.Pkg)] = []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+"/"+strconv.Itoa(x.Pkg)]; 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]
		}
	}
	out.ZFSWrite = foldZFSWrites(b, out.ZFSWrite)
	return out
}

// foldZFSWrites averages the write path's rates and the buffer's fill over a bucket, per pool;
// everything else (limits, the mover's state, the last walk) is the newest sample's.
func foldZFSWrites(b []Sample, last *ZFSWrites) *ZFSWrites {
	if last == nil {
		return nil
	}
	w := *last
	w.Pools = append([]PoolWrites(nil), last.Pools...)
	var dirty, delayed, over, n float64
	for _, s := range b {
		if s.ZFSWrite != nil {
			dirty, delayed, over, n = dirty+float64(s.ZFSWrite.Dirty), delayed+s.ZFSWrite.DelayedPerSec, over+s.ZFSWrite.OverMaxPerSec, n+1
		}
	}
	if n > 0 {
		w.Dirty, w.DelayedPerSec, w.OverMaxPerSec = uint64(dirty/n), round1(delayed/n), round1(over/n)
	}
	per := avgBy(b, func(s Sample) map[string][]float64 {
		m := map[string][]float64{}
		if s.ZFSWrite == nil {
			return m
		}
		for _, p := range s.ZFSWrite.Pools {
			v := []float64{p.WriteBps, p.LogicalBps, 0, 0}
			if p.Landing != nil {
				v[2], v[3] = p.Landing.InBps, p.Landing.DrainBps
			}
			m[p.Pool] = v
		}
		return m
	})
	for i, p := range w.Pools {
		if v, ok := per[p.Pool]; ok {
			w.Pools[i].WriteBps, w.Pools[i].LogicalBps = round1(v[0]), round1(v[1])
			if p.Landing != nil {
				l := *p.Landing
				l.InBps, l.DrainBps = round1(v[2]), round1(v[3])
				w.Pools[i].Landing = &l
			}
		}
	}
	return &w
}

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