amadan

carlos / platform Public

usage: meter vocabulary and edge day objects

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 238c17a56db3ef6ae78a8fde222176d543c348e2 parent d28572d
3 files changed, +213 −0
  • internal/usage/edge.go +100 −0
  • internal/usage/edge_test.go +80 −0
  • internal/usage/usage.go +33 −0
diff --git a/internal/usage/edge.go b/internal/usage/edge.go
new file mode 100644
index 000000000..af4ba8f5f
--- /dev/null
+++ b/internal/usage/edge.go
@@ -0,0 +1,100 @@
+package usage
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// EdgeRow is one route's traffic within one hour on one box.
+type EdgeRow struct {
+ Account string `json:"account"`
+ App string `json:"app"`
+ Requests int64 `json:"requests"`
+ BytesOut int64 `json:"bytes_out"`
+}
+
+// EdgeDay is usage/edge/<box>/days/<date>.json — a box's edge traffic for
+// one UTC day, hour-bucketed like metrics.DayRollup. Unlike metrics there is
+// no separate raw-chunk family: the 5-minute flush CAS-merges deltas
+// straight into this object, which is both the durability write and the
+// rollup.
+type EdgeDay struct {
+ V int `json:"v"` // 1
+ Box string `json:"box"`
+ Date string `json:"date"`
+ Hours map[string]map[string]EdgeRow `json:"hours"` // "09" -> host -> row
+}
+
+// EdgeDayKey is the bucket key for a box's day object.
+func EdgeDayKey(box, date string) string {
+ return "usage/edge/" + box + "/days/" + date + ".json"
+}
+
+// EdgeDaysPrefix is the List prefix for a box's day objects.
+func EdgeDaysPrefix(box string) string { return "usage/edge/" + box + "/days/" }
+
+// EdgeBoxesPrefix is the ListDirs prefix that discovers metering boxes.
+const EdgeBoxesPrefix = "usage/edge/"
+
+// AddDeltas folds drained per-host deltas into the day's hour bucket,
+// summing counters and refreshing the route identity (a host re-pointed to
+// a new app mid-hour lands on the later identity — modelling grade).
+func (d *EdgeDay) AddDeltas(hour string, deltas map[string]EdgeRow) {
+ if d.Hours == nil {
+ d.Hours = map[string]map[string]EdgeRow{}
+ }
+ rows := d.Hours[hour]
+ if rows == nil {
+ rows = map[string]EdgeRow{}
+ d.Hours[hour] = rows
+ }
+ for host, delta := range deltas {
+ row := rows[host]
+ row.Account, row.App = delta.Account, delta.App
+ row.Requests += delta.Requests
+ row.BytesOut += delta.BytesOut
+ rows[host] = row
+ }
+}
+
+// UpsertEdgeDay CAS-merges deltas into the box's day object for now's UTC
+// day and hour: read (etag), add, conditional put; on ErrCASConflict re-read
+// and retry, three attempts. Empty deltas write nothing.
+func UpsertEdgeDay(ctx context.Context, s release.Store, box string, now time.Time, deltas map[string]EdgeRow) error {
+ if len(deltas) == 0 {
+ return nil
+ }
+ date, hour := DayOf(now), HourLabel(now)
+ key := EdgeDayKey(box, date)
+ for attempt := 0; attempt < 3; attempt++ {
+ day := EdgeDay{V: 1, Box: box, Date: date}
+ raw, etag, err := s.Get(ctx, key)
+ switch {
+ case err == nil:
+ if uerr := json.Unmarshal(raw, &day); uerr != nil {
+ return fmt.Errorf("usage: parse %s: %w", key, uerr)
+ }
+ case errors.Is(err, os.ErrNotExist):
+ etag = "" // create-only Put
+ default:
+ return err
+ }
+ day.AddDeltas(hour, deltas)
+ out, err := json.Marshal(day)
+ if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, out, etag); err == nil {
+ return nil
+ } else if !errors.Is(err, release.ErrCASConflict) {
+ return err
+ }
+ }
+ return fmt.Errorf("usage: %s: lost CAS race three times", key)
+}
diff --git a/internal/usage/edge_test.go b/internal/usage/edge_test.go
new file mode 100644
index 000000000..424f0c9e7
--- /dev/null
+++ b/internal/usage/edge_test.go
@@ -0,0 +1,80 @@
+package usage
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestEdgeDayKeyAndLabels(t *testing.T) {
+ if got := EdgeDayKey("box1", "2026-08-11"); got != "usage/edge/box1/days/2026-08-11.json" {
+ t.Fatalf("EdgeDayKey = %q", got)
+ }
+ at := time.Date(2026, 8, 11, 9, 4, 0, 0, time.UTC)
+ if got := DayOf(at); got != "2026-08-11" {
+ t.Fatalf("DayOf = %q", got)
+ }
+ if got := MonthOf(at); got != "2026-08" {
+ t.Fatalf("MonthOf = %q", got)
+ }
+ if got := HourLabel(at); got != "09" {
+ t.Fatalf("HourLabel = %q", got)
+ }
+}
+
+func TestUpsertEdgeDayMergesAcrossFlushes(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ at := time.Date(2026, 8, 11, 9, 10, 0, 0, time.UTC)
+
+ d1 := map[string]EdgeRow{
+ "a.example.com": {Account: "bab", App: "hello", Requests: 10, BytesOut: 1000},
+ }
+ if err := UpsertEdgeDay(ctx, s, "box1", at, d1); err != nil {
+ t.Fatal(err)
+ }
+ // Second flush, same hour: counts must ADD, not replace.
+ if err := UpsertEdgeDay(ctx, s, "box1", at.Add(5*time.Minute), d1); err != nil {
+ t.Fatal(err)
+ }
+ // Third flush, next hour and a second host.
+ d2 := map[string]EdgeRow{
+ "a.example.com": {Account: "bab", App: "hello", Requests: 1, BytesOut: 50},
+ "b.example.com": {Account: "ops", App: "site", Requests: 2, BytesOut: 20},
+ }
+ if err := UpsertEdgeDay(ctx, s, "box1", at.Add(time.Hour), d2); err != nil {
+ t.Fatal(err)
+ }
+
+ raw, _, err := s.Get(ctx, EdgeDayKey("box1", "2026-08-11"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var day EdgeDay
+ if err := json.Unmarshal(raw, &day); err != nil {
+ t.Fatal(err)
+ }
+ if day.V != 1 || day.Box != "box1" || day.Date != "2026-08-11" {
+ t.Fatalf("header = %+v", day)
+ }
+ h9 := day.Hours["09"]["a.example.com"]
+ if h9.Requests != 20 || h9.BytesOut != 2000 || h9.Account != "bab" || h9.App != "hello" {
+ t.Fatalf("hour 09 row = %+v", h9)
+ }
+ if day.Hours["10"]["b.example.com"].Requests != 2 {
+ t.Fatalf("hour 10 = %+v", day.Hours["10"])
+ }
+}
+
+func TestUpsertEdgeDayEmptyDeltasIsNoop(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ if err := UpsertEdgeDay(context.Background(), s, "box1", time.Now().UTC(), nil); err != nil {
+ t.Fatal(err)
+ }
+ if keys, _ := s.List(context.Background(), "usage/"); len(keys) != 0 {
+ t.Fatalf("noop wrote %v", keys)
+ }
+}
diff --git a/internal/usage/usage.go b/internal/usage/usage.go
new file mode 100644
index 000000000..70a7695e8
--- /dev/null
+++ b/internal/usage/usage.go
@@ -0,0 +1,33 @@
+// Package usage holds the platform's metering shapes: what a box or console
+// pass writes under usage/ in the deployment bucket, and the canonical meter
+// vocabulary every consumer — statements, scenarios, dashboards, a future
+// billing engine — speaks. Meter names are FROZEN once merged: renaming one
+// orphans every statement already written (spec: billing readiness).
+//
+// Everything here is decision-grade, not invoice-grade (spec: modelling
+// instrument): CAS-merge retries can in rare cases double-count one 5-minute
+// window, and days a writer was down are simply absent. Consumers must treat
+// missing days as reduced coverage, never as zero usage.
+package usage
+
+import "time"
+
+// Meter identifiers — the canonical billing vocabulary.
+const (
+ MeterCPUCoreHours = "cpu_core_hours"
+ MeterMemGBHours = "mem_gb_hours"
+ MeterEgressGB = "egress_gb"
+ MeterRequests1k = "requests_1k"
+ MeterStorageGBMonth = "storage_gb_month"
+ MeterAppMonths = "app_months"
+ MeterInstanceHours = "instance_hours"
+)
+
+// DayOf is the UTC date label used in every usage key.
+func DayOf(t time.Time) string { return t.UTC().Format("2006-01-02") }
+
+// MonthOf is the UTC month label used in statement and bill keys.
+func MonthOf(t time.Time) string { return t.UTC().Format("2006-01") }
+
+// HourLabel is the two-digit UTC hour used inside day objects.
+func HourLabel(t time.Time) string { return t.UTC().Format("15") }