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