amadan

carlos / platform Public

metrics: raw chunk shapes, bucket keys, CAS overwrite, prune selection

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell d247d40db7bd10752b7c7b1bd27768575149bb5a parent 514f507
3 files changed, +295 −0
  • internal/metrics/chunk.go +129 −0
  • internal/metrics/chunk_test.go +120 −0
  • internal/metrics/sample.go +46 −0
diff --git a/internal/metrics/chunk.go b/internal/metrics/chunk.go
new file mode 100644
index 000000000..bcb72011d
--- /dev/null
+++ b/internal/metrics/chunk.go
@@ -0,0 +1,129 @@
+package metrics
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// hourFormat is the UTC hour format used both in RawChunk.Hour and in raw
+// chunk keys: 2006-01-02-15.
+const hourFormat = "2006-01-02-15"
+
+// HourOf formats t as the UTC hour bucket a raw chunk belongs to.
+func HourOf(t time.Time) string {
+ return t.UTC().Format(hourFormat)
+}
+
+// RawKey is the bucket key for a box's raw chunk covering the given hour
+// (itself formatted like HourOf's output, e.g. 2026-08-02-21).
+func RawKey(box, hour string) string {
+ return "metrics/boxes/" + box + "/raw/" + hour + ".json"
+}
+
+// DayKey is the bucket key for a box's rolled-up day chunk (date formatted
+// 2006-01-02).
+func DayKey(box, date string) string {
+ return "metrics/boxes/" + box + "/days/" + date + ".json"
+}
+
+// WriteRaw marshals c indented and writes it to its hour-derived key,
+// last-writer-wins. A box appends samples to the same hour's chunk
+// repeatedly as the hour progresses, so a bare create-only Put (ifMatch "")
+// would fail every write after the first with ErrCASConflict — this mirrors
+// status.WriteBox's read-etag-then-Put dance to express "replace whatever is
+// there, or create it" on top of release.Store's CAS-only Put: read the
+// current etag (a not-found just means "absent", so ifMatch stays "" for a
+// create), Put with that etag, and if the Put itself loses the race
+// (something else wrote in between), re-read once and retry with the fresh
+// etag. A second conflict is NOT retried again and is returned to the
+// caller: this box is the only writer that should ever be targeting its own
+// raw key, so two conflicts in a row means something else collided
+// (clock-skewed duplicate machine id, a bug elsewhere) — that deserves a
+// loud failure, not a silent third attempt.
+func WriteRaw(ctx context.Context, s release.Store, c RawChunk) error {
+ body, err := json.MarshalIndent(c, "", " ")
+ if err != nil {
+ return err
+ }
+ key := RawKey(c.Box, c.Hour)
+
+ etag, err := currentRawEtag(ctx, s, key)
+ if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, body, etag); err != nil {
+ if !errors.Is(err, release.ErrCASConflict) {
+ return err
+ }
+ etag, err = currentRawEtag(ctx, s, key)
+ if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, body, etag); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// currentRawEtag reads a key's etag for a replace-only Put, folding
+// "doesn't exist yet" into "" (create-only) — the same absent-means-empty
+// convention release.Store's own CAS contract uses.
+func currentRawEtag(ctx context.Context, s release.Store, key string) (string, error) {
+ _, etag, err := s.Get(ctx, key)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return "", nil
+ }
+ return "", err
+ }
+ return etag, nil
+}
+
+// StaleRawKeys returns the subset of keys (raw chunk keys, as produced by
+// RawKey) whose hour is older than now.Add(-keep) — candidates for pruning.
+// It is pure: no I/O, and it MUST NEVER return a key it cannot parse — never
+// delete what you don't understand. A key is parsed by taking the text
+// between the last "/" and ".json" and parsing it with the same UTC hour
+// format RawKey/HourOf use; anything that doesn't fit that shape (a
+// different key layout, a corrupted or hand-edited name) is silently
+// skipped rather than treated as either stale or fresh.
+func StaleRawKeys(keys []string, now time.Time, keep time.Duration) []string {
+ cutoff := now.Add(-keep)
+ var out []string
+ for _, key := range keys {
+ hour, ok := rawKeyHour(key)
+ if !ok {
+ continue
+ }
+ t, err := time.Parse(hourFormat, hour)
+ if err != nil {
+ continue
+ }
+ if t.Before(cutoff) {
+ out = append(out, key)
+ }
+ }
+ return out
+}
+
+// rawKeyHour extracts the "<hour>" portion of a "…/<hour>.json" key.
+func rawKeyHour(key string) (string, bool) {
+ const suffix = ".json"
+ if !strings.HasSuffix(key, suffix) {
+ return "", false
+ }
+ base := strings.TrimSuffix(key, suffix)
+ idx := strings.LastIndex(base, "/")
+ hour := base[idx+1:]
+ if hour == "" {
+ return "", false
+ }
+ return hour, true
+}
diff --git a/internal/metrics/chunk_test.go b/internal/metrics/chunk_test.go
new file mode 100644
index 000000000..6dafc5268
--- /dev/null
+++ b/internal/metrics/chunk_test.go
@@ -0,0 +1,120 @@
+package metrics
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestHourOf(t *testing.T) {
+ got := HourOf(time.Date(2026, 8, 2, 21, 45, 0, 0, time.UTC))
+ if got != "2026-08-02-21" {
+ t.Fatalf("got %q, want 2026-08-02-21", got)
+ }
+}
+
+func TestRawKeyDayKey(t *testing.T) {
+ if got, want := RawKey("b", "2026-08-02-21"), "metrics/boxes/b/raw/2026-08-02-21.json"; got != want {
+ t.Fatalf("RawKey got %q, want %q", got, want)
+ }
+ if got, want := DayKey("b", "2026-08-02"), "metrics/boxes/b/days/2026-08-02.json"; got != want {
+ t.Fatalf("DayKey got %q, want %q", got, want)
+ }
+}
+
+func sampleFloat(v float64) *float64 { return &v }
+
+func TestWriteRawRoundTrip(t *testing.T) {
+ store := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+
+ c := RawChunk{
+ V: 1,
+ Box: "b",
+ Hour: "2026-08-02-21",
+ Samples: []Sample{
+ {
+ At: time.Date(2026, 8, 2, 21, 0, 0, 0, time.UTC),
+ Host: HostSample{CPUFrac: sampleFloat(0.5)},
+ Units: map[string]UnitSample{
+ "carlos-app@x.example.com.service": {CPUFrac: sampleFloat(0.1)},
+ },
+ },
+ },
+ }
+
+ if err := WriteRaw(ctx, store, c); err != nil {
+ t.Fatalf("WriteRaw: %v", err)
+ }
+
+ body, _, err := store.Get(ctx, RawKey(c.Box, c.Hour))
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ var got RawChunk
+ if err := json.Unmarshal(body, &got); err != nil {
+ t.Fatalf("Unmarshal: %v", err)
+ }
+ if got.V != c.V || got.Box != c.Box || got.Hour != c.Hour || len(got.Samples) != 1 {
+ t.Fatalf("got %+v, want %+v", got, c)
+ }
+ if got.Samples[0].Host.CPUFrac == nil || *got.Samples[0].Host.CPUFrac != 0.5 {
+ t.Fatalf("host cpu round-trip mismatch: %+v", got.Samples[0].Host)
+ }
+}
+
+func TestWriteRawOverwriteSameHour(t *testing.T) {
+ store := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+
+ first := RawChunk{V: 1, Box: "b", Hour: "2026-08-02-21", Samples: []Sample{
+ {At: time.Date(2026, 8, 2, 21, 0, 0, 0, time.UTC), Host: HostSample{CPUFrac: sampleFloat(0.1)}},
+ }}
+ second := RawChunk{V: 1, Box: "b", Hour: "2026-08-02-21", Samples: []Sample{
+ {At: time.Date(2026, 8, 2, 21, 5, 0, 0, time.UTC), Host: HostSample{CPUFrac: sampleFloat(0.2)}},
+ {At: time.Date(2026, 8, 2, 21, 10, 0, 0, time.UTC), Host: HostSample{CPUFrac: sampleFloat(0.3)}},
+ }}
+
+ if err := WriteRaw(ctx, store, first); err != nil {
+ t.Fatalf("WriteRaw first: %v", err)
+ }
+ if err := WriteRaw(ctx, store, second); err != nil {
+ t.Fatalf("WriteRaw second: %v", err)
+ }
+
+ keys, err := store.List(ctx, "metrics/boxes/b/raw/")
+ if err != nil {
+ t.Fatalf("List: %v", err)
+ }
+ if len(keys) != 1 {
+ t.Fatalf("got %d keys, want exactly one (overwrite, not a new object): %v", len(keys), keys)
+ }
+
+ body, _, err := store.Get(ctx, RawKey("b", "2026-08-02-21"))
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ var got RawChunk
+ if err := json.Unmarshal(body, &got); err != nil {
+ t.Fatalf("Unmarshal: %v", err)
+ }
+ if len(got.Samples) != 2 {
+ t.Fatalf("got %d samples, want the latest write's 2 (latest content wins)", len(got.Samples))
+ }
+}
+
+func TestStaleRawKeys(t *testing.T) {
+ now := time.Date(2026, 8, 2, 22, 30, 0, 0, time.UTC)
+ keys := []string{
+ "metrics/boxes/b/raw/2026-07-31-10.json", // stale
+ "metrics/boxes/b/raw/2026-08-02-21.json", // fresh
+ "metrics/boxes/b/raw/garbage.json", // unparsable — never returned
+ }
+ got := StaleRawKeys(keys, now, 48*time.Hour)
+ if len(got) != 1 || got[0] != keys[0] {
+ t.Fatalf("got %v, want only the stale key", got)
+ }
+}
diff --git a/internal/metrics/sample.go b/internal/metrics/sample.go
new file mode 100644
index 000000000..99dcef827
--- /dev/null
+++ b/internal/metrics/sample.go
@@ -0,0 +1,46 @@
+package metrics
+
+import "time"
+
+// HostSample is one point-in-time reading of whole-host resource usage.
+// Every field is a pointer because a missing reading (a parser in parse.go
+// returning nil) must stay distinguishable from a real zero all the way out
+// to whatever renders or alerts on it.
+type HostSample struct {
+ CPUFrac *float64 `json:"cpu"`
+ Load1 *float64 `json:"load1"`
+ Load5 *float64 `json:"load5"`
+ Load15 *float64 `json:"load15"`
+ MemFrac *float64 `json:"mem"`
+ DiskFrac *float64 `json:"disk"`
+}
+
+// UnitSample is one point-in-time reading for a single systemd unit's
+// cgroup. ReadBytes and WriteBytes are cumulative counters (not deltas) —
+// callers wanting a rate diff consecutive samples themselves.
+type UnitSample struct {
+ CPUFrac *float64 `json:"cpu"`
+ MemBytes *int64 `json:"mem_bytes"`
+ MemPeakBytes *int64 `json:"mem_peak_bytes"`
+ ReadBytes *int64 `json:"read_bytes"` // cumulative
+ WriteBytes *int64 `json:"write_bytes"` // cumulative
+}
+
+// Sample is one point-in-time reading of a box: the host as a whole plus
+// every monitored unit, keyed by full unit name (e.g.
+// carlos-app@x.example.com.service).
+type Sample struct {
+ At time.Time `json:"at"`
+ Host HostSample `json:"host"`
+ Units map[string]UnitSample `json:"units"`
+}
+
+// RawChunk is an hour's worth of raw samples for one box, the unit of
+// storage under metrics/boxes/<box>/raw/. V is the wire-shape version (1
+// today); Hour is UTC, formatted 2006-01-02-15.
+type RawChunk struct {
+ V int `json:"v"` // 1
+ Box string `json:"box"`
+ Hour string `json:"hour"` // UTC, 2006-01-02-15
+ Samples []Sample `json:"samples"`
+}