amadan

carlos / platform Public

metrics: hourly min/avg/max rollups into day objects

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 231963ac6e2b1609d1e0cc7c02d731fc36c3bbec parent d247d40
2 files changed, +318 −0
  • internal/metrics/rollup.go +185 −0
  • internal/metrics/rollup_test.go +133 −0
diff --git a/internal/metrics/rollup.go b/internal/metrics/rollup.go
new file mode 100644
index 000000000..032562492
--- /dev/null
+++ b/internal/metrics/rollup.go
@@ -0,0 +1,185 @@
+package metrics
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Agg is a min/avg/max/count summary of one metric over some window. N
+// counts only the samples that actually carried the metric (nil readings —
+// a parser that couldn't read a value — are excluded, never treated as
+// zero).
+type Agg struct {
+ Min float64 `json:"min"`
+ Avg float64 `json:"avg"`
+ Max float64 `json:"max"`
+ N int `json:"n"`
+}
+
+// HourAgg is one hour's rollup: whole-host metrics plus per-unit metrics.
+// A metric that no sample in the hour carried is simply absent from the
+// map — never a zero-valued Agg standing in for "no data".
+type HourAgg struct {
+ Host map[string]Agg `json:"host"` // "cpu","mem","disk","load1"
+ Units map[string]map[string]Agg `json:"units"` // unit -> "cpu","mem_bytes" -> Agg
+}
+
+// DayRollup is a box's rolled-up day, keyed by UTC hour ("00".."23") —
+// the object stored under metrics/boxes/<box>/days/<date>.json.
+type DayRollup struct {
+ V int `json:"v"` // 1
+ Box string `json:"box"`
+ Date string `json:"date"` // UTC 2006-01-02
+ Hours map[string]HourAgg `json:"hours"`
+}
+
+// aggOf reduces vals to an Agg, reporting ok=false for an empty slice — the
+// "this metric was never carried" case, which callers turn into "key
+// absent from the map" rather than a zero-valued Agg.
+func aggOf(vals []float64) (Agg, bool) {
+ if len(vals) == 0 {
+ return Agg{}, false
+ }
+ min, max, sum := vals[0], vals[0], 0.0
+ for _, v := range vals {
+ if v < min {
+ min = v
+ }
+ if v > max {
+ max = v
+ }
+ sum += v
+ }
+ return Agg{Min: min, Avg: sum / float64(len(vals)), Max: max, N: len(vals)}, true
+}
+
+// RollupHour aggregates a slice of raw samples (typically an hour's worth)
+// into one HourAgg. Nil metric readings are excluded from N and, when a
+// metric is nil in every sample, its key is absent from the returned map
+// entirely (both at the host level and per-unit). Host metrics are
+// "cpu","mem","disk","load1"; unit metrics are "cpu" (from
+// UnitSample.CPUFrac) and "mem_bytes" (from MemBytes, cast to float64).
+func RollupHour(samples []Sample) HourAgg {
+ var hostCPU, hostMem, hostDisk, hostLoad1 []float64
+ for _, s := range samples {
+ if v := s.Host.CPUFrac; v != nil {
+ hostCPU = append(hostCPU, *v)
+ }
+ if v := s.Host.MemFrac; v != nil {
+ hostMem = append(hostMem, *v)
+ }
+ if v := s.Host.DiskFrac; v != nil {
+ hostDisk = append(hostDisk, *v)
+ }
+ if v := s.Host.Load1; v != nil {
+ hostLoad1 = append(hostLoad1, *v)
+ }
+ }
+
+ host := map[string]Agg{}
+ for key, vals := range map[string][]float64{
+ "cpu": hostCPU, "mem": hostMem, "disk": hostDisk, "load1": hostLoad1,
+ } {
+ if agg, ok := aggOf(vals); ok {
+ host[key] = agg
+ }
+ }
+
+ unitCPU := map[string][]float64{}
+ unitMem := map[string][]float64{}
+ unitNames := map[string]bool{}
+ for _, s := range samples {
+ for name, u := range s.Units {
+ unitNames[name] = true
+ if u.CPUFrac != nil {
+ unitCPU[name] = append(unitCPU[name], *u.CPUFrac)
+ }
+ if u.MemBytes != nil {
+ unitMem[name] = append(unitMem[name], float64(*u.MemBytes))
+ }
+ }
+ }
+
+ units := map[string]map[string]Agg{}
+ for name := range unitNames {
+ inner := map[string]Agg{}
+ if agg, ok := aggOf(unitCPU[name]); ok {
+ inner["cpu"] = agg
+ }
+ if agg, ok := aggOf(unitMem[name]); ok {
+ inner["mem_bytes"] = agg
+ }
+ if len(inner) > 0 {
+ units[name] = inner
+ }
+ }
+
+ return HourAgg{Host: host, Units: units}
+}
+
+// UpsertRollup reads box's day object for hour's UTC date (a missing
+// object is not an error — it starts a fresh DayRollup), sets the entry
+// for hour's UTC hour bucket ("00".."23") to agg, and writes the day back
+// with the same read-etag, Put, retry-once-on-ErrCASConflict pattern
+// WriteRaw uses: a second conflict is not retried again and is returned to
+// the caller.
+func UpsertRollup(ctx context.Context, s release.Store, box string, hour time.Time, agg HourAgg) error {
+ date := hour.UTC().Format("2006-01-02")
+ hourKey := hour.UTC().Format("15")
+ key := DayKey(box, date)
+
+ day, etag, err := currentDayRollup(ctx, s, key, box, date)
+ if err != nil {
+ return err
+ }
+ day.Hours[hourKey] = agg
+ body, err := json.MarshalIndent(day, "", " ")
+ if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, body, etag); err != nil {
+ if !errors.Is(err, release.ErrCASConflict) {
+ return err
+ }
+ day, etag, err = currentDayRollup(ctx, s, key, box, date)
+ if err != nil {
+ return err
+ }
+ day.Hours[hourKey] = agg
+ body, err = json.MarshalIndent(day, "", " ")
+ if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, body, etag); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// currentDayRollup reads key's current DayRollup and etag for a
+// replace-only Put, folding "doesn't exist yet" into a fresh DayRollup at
+// ifMatch "" (create-only) — the same absent-means-empty convention
+// currentRawEtag uses for raw chunks.
+func currentDayRollup(ctx context.Context, s release.Store, key, box, date string) (DayRollup, string, error) {
+ body, etag, err := s.Get(ctx, key)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return DayRollup{V: 1, Box: box, Date: date, Hours: map[string]HourAgg{}}, "", nil
+ }
+ return DayRollup{}, "", err
+ }
+ var day DayRollup
+ if err := json.Unmarshal(body, &day); err != nil {
+ return DayRollup{}, "", err
+ }
+ if day.Hours == nil {
+ day.Hours = map[string]HourAgg{}
+ }
+ return day, etag, nil
+}
diff --git a/internal/metrics/rollup_test.go b/internal/metrics/rollup_test.go
new file mode 100644
index 000000000..9711ad3bf
--- /dev/null
+++ b/internal/metrics/rollup_test.go
@@ -0,0 +1,133 @@
+package metrics
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func rollupFloat(v float64) *float64 { return &v }
+func rollupInt(v int64) *int64 { return &v }
+
+func TestRollupHourHostCPUNilExcluded(t *testing.T) {
+ samples := []Sample{
+ {Host: HostSample{CPUFrac: rollupFloat(0.2), MemFrac: rollupFloat(0.5), DiskFrac: rollupFloat(0.1), Load1: rollupFloat(1.0)}},
+ {Host: HostSample{CPUFrac: nil, MemFrac: rollupFloat(0.6), DiskFrac: rollupFloat(0.2), Load1: rollupFloat(1.5)}},
+ {Host: HostSample{CPUFrac: rollupFloat(0.8), MemFrac: rollupFloat(0.4), DiskFrac: rollupFloat(0.3), Load1: rollupFloat(0.5)}},
+ }
+
+ agg := RollupHour(samples)
+
+ cpu, ok := agg.Host["cpu"]
+ if !ok {
+ t.Fatalf("Host[\"cpu\"] absent, want present with N=2")
+ }
+ if cpu.N != 2 {
+ t.Fatalf("cpu.N = %d, want 2 (nil sample excluded)", cpu.N)
+ }
+ if cpu.Min != 0.2 || cpu.Max != 0.8 || cpu.Avg != 0.5 {
+ t.Fatalf("cpu = %+v, want min=0.2 max=0.8 avg=0.5", cpu)
+ }
+}
+
+func TestRollupHourUnitPresentInOneSample(t *testing.T) {
+ samples := []Sample{
+ {Units: map[string]UnitSample{
+ "a.service": {CPUFrac: rollupFloat(0.3), MemBytes: rollupInt(1000)},
+ }},
+ {Units: map[string]UnitSample{}},
+ }
+
+ agg := RollupHour(samples)
+
+ unit, ok := agg.Units["a.service"]
+ if !ok {
+ t.Fatalf("Units[\"a.service\"] absent, want present")
+ }
+ cpu, ok := unit["cpu"]
+ if !ok || cpu.N != 1 {
+ t.Fatalf("unit cpu = %+v (ok=%v), want N=1", cpu, ok)
+ }
+ if cpu.Min != 0.3 || cpu.Max != 0.3 || cpu.Avg != 0.3 {
+ t.Fatalf("unit cpu = %+v, want min=avg=max=0.3", cpu)
+ }
+ mem, ok := unit["mem_bytes"]
+ if !ok || mem.N != 1 {
+ t.Fatalf("unit mem_bytes = %+v (ok=%v), want N=1", mem, ok)
+ }
+ if mem.Min != 1000 || mem.Max != 1000 || mem.Avg != 1000 {
+ t.Fatalf("unit mem_bytes = %+v, want min=avg=max=1000", mem)
+ }
+}
+
+func TestRollupHourAllNilMetricAbsent(t *testing.T) {
+ samples := []Sample{
+ {Host: HostSample{CPUFrac: rollupFloat(0.1)}},
+ {Host: HostSample{CPUFrac: rollupFloat(0.2)}},
+ }
+
+ agg := RollupHour(samples)
+
+ if _, ok := agg.Host["load1"]; ok {
+ t.Fatalf("Host[\"load1\"] present, want absent (never carried in any sample)")
+ }
+ if _, ok := agg.Host["mem"]; ok {
+ t.Fatalf("Host[\"mem\"] present, want absent")
+ }
+ if _, ok := agg.Host["disk"]; ok {
+ t.Fatalf("Host[\"disk\"] present, want absent")
+ }
+ if len(agg.Units) != 0 {
+ t.Fatalf("Units = %+v, want empty (no unit samples at all)", agg.Units)
+ }
+}
+
+func TestUpsertRollupTwoHoursSameDay(t *testing.T) {
+ store := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+
+ hour10 := time.Date(2026, 8, 2, 10, 30, 0, 0, time.UTC)
+ hour11 := time.Date(2026, 8, 2, 11, 15, 0, 0, time.UTC)
+
+ agg10 := HourAgg{Host: map[string]Agg{"cpu": {Min: 0.1, Avg: 0.1, Max: 0.1, N: 1}}}
+ agg11 := HourAgg{Host: map[string]Agg{"cpu": {Min: 0.2, Avg: 0.2, Max: 0.2, N: 1}}}
+
+ if err := UpsertRollup(ctx, store, "b", hour10, agg10); err != nil {
+ t.Fatalf("UpsertRollup hour10: %v", err)
+ }
+ if err := UpsertRollup(ctx, store, "b", hour11, agg11); err != nil {
+ t.Fatalf("UpsertRollup hour11: %v", err)
+ }
+
+ keys, err := store.List(ctx, "metrics/boxes/b/days/")
+ if err != nil {
+ t.Fatalf("List: %v", err)
+ }
+ if len(keys) != 1 {
+ t.Fatalf("got %d day keys, want exactly one: %v", len(keys), keys)
+ }
+
+ body, _, err := store.Get(ctx, DayKey("b", "2026-08-02"))
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ var got DayRollup
+ if err := json.Unmarshal(body, &got); err != nil {
+ t.Fatalf("Unmarshal: %v", err)
+ }
+ if got.V != 1 || got.Box != "b" || got.Date != "2026-08-02" {
+ t.Fatalf("got %+v, want V=1 Box=b Date=2026-08-02", got)
+ }
+ if len(got.Hours) != 2 {
+ t.Fatalf("got %d hours, want 2 (10 and 11): %+v", len(got.Hours), got.Hours)
+ }
+ if got.Hours["10"].Host["cpu"].N != 1 || got.Hours["10"].Host["cpu"].Avg != 0.1 {
+ t.Fatalf("Hours[10] = %+v, want cpu avg 0.1", got.Hours["10"])
+ }
+ if got.Hours["11"].Host["cpu"].N != 1 || got.Hours["11"].Host["cpu"].Avg != 0.2 {
+ t.Fatalf("Hours[11] = %+v, want cpu avg 0.2", got.Hours["11"])
+ }
+}