| 1 | package usage |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | // EdgeRow is one route's traffic within one hour on one box. |
| 15 | type EdgeRow struct { |
| 16 | Account string `json:"account"` |
| 17 | App string `json:"app"` |
| 18 | Requests int64 `json:"requests"` |
| 19 | BytesOut int64 `json:"bytes_out"` |
| 20 | // HopRequests/HopBytesOut are the SUBSET of Requests/BytesOut that |
| 21 | // arrived over a verified backhaul hop (multi-region spec Part 3) — an |
| 22 | // annotation inside the totals, never an addition to them: a consumer |
| 23 | // that sums BytesOut already includes every hop byte, and must never add |
| 24 | // HopBytesOut on top. They exist so the platform can see which traffic |
| 25 | // crossed a region (its own egress is doubled by a hop) and so backhaul |
| 26 | // can be priced later without a shape change; nothing bills them today. |
| 27 | // |
| 28 | // Additive and omitempty: a day object written before these existed |
| 29 | // parses with both zero, one written after them is ignored field-for- |
| 30 | // field by an older binary, and a single-region deployment's objects |
| 31 | // stay byte-identical to what it wrote yesterday. |
| 32 | HopRequests int64 `json:"hop_requests,omitempty"` |
| 33 | HopBytesOut int64 `json:"hop_bytes_out,omitempty"` |
| 34 | } |
| 35 | |
| 36 | // EdgeDay is usage/edge/<box>/days/<date>.json — a box's edge traffic for |
| 37 | // one UTC day, hour-bucketed like metrics.DayRollup. Unlike metrics there is |
| 38 | // no separate raw-chunk family: the 5-minute flush CAS-merges deltas |
| 39 | // straight into this object, which is both the durability write and the |
| 40 | // rollup. |
| 41 | type EdgeDay struct { |
| 42 | V int `json:"v"` // 1 |
| 43 | Box string `json:"box"` |
| 44 | Date string `json:"date"` |
| 45 | Hours map[string]map[string]EdgeRow `json:"hours"` // "09" -> host -> row |
| 46 | } |
| 47 | |
| 48 | // EdgeDayKey is the bucket key for a box's day object. |
| 49 | func EdgeDayKey(box, date string) string { |
| 50 | return "usage/edge/" + box + "/days/" + date + ".json" |
| 51 | } |
| 52 | |
| 53 | // EdgeDaysPrefix is the List prefix for a box's day objects. |
| 54 | func EdgeDaysPrefix(box string) string { return "usage/edge/" + box + "/days/" } |
| 55 | |
| 56 | // EdgeBoxesPrefix is the ListDirs prefix that discovers metering boxes. |
| 57 | const EdgeBoxesPrefix = "usage/edge/" |
| 58 | |
| 59 | // AddDeltas folds drained per-host deltas into the day's hour bucket, |
| 60 | // summing counters and refreshing the route identity (a host re-pointed to |
| 61 | // a new app mid-hour lands on the later identity — modelling grade). |
| 62 | func (d *EdgeDay) AddDeltas(hour string, deltas map[string]EdgeRow) { |
| 63 | if d.Hours == nil { |
| 64 | d.Hours = map[string]map[string]EdgeRow{} |
| 65 | } |
| 66 | rows := d.Hours[hour] |
| 67 | if rows == nil { |
| 68 | rows = map[string]EdgeRow{} |
| 69 | d.Hours[hour] = rows |
| 70 | } |
| 71 | for host, delta := range deltas { |
| 72 | row := rows[host] |
| 73 | row.Account, row.App = delta.Account, delta.App |
| 74 | row.Requests += delta.Requests |
| 75 | row.BytesOut += delta.BytesOut |
| 76 | row.HopRequests += delta.HopRequests |
| 77 | row.HopBytesOut += delta.HopBytesOut |
| 78 | rows[host] = row |
| 79 | } |
| 80 | } |
| 81 | |
| 82 | // UpsertEdgeDay CAS-merges deltas into the box's day object for now's UTC |
| 83 | // day and hour: read (etag), add, conditional put; on ErrCASConflict re-read |
| 84 | // and retry, three attempts. Empty deltas write nothing. |
| 85 | func UpsertEdgeDay(ctx context.Context, s release.Store, box string, now time.Time, deltas map[string]EdgeRow) error { |
| 86 | if len(deltas) == 0 { |
| 87 | return nil |
| 88 | } |
| 89 | date, hour := DayOf(now), HourLabel(now) |
| 90 | key := EdgeDayKey(box, date) |
| 91 | for attempt := 0; attempt < 3; attempt++ { |
| 92 | day := EdgeDay{V: 1, Box: box, Date: date} |
| 93 | raw, etag, err := s.Get(ctx, key) |
| 94 | switch { |
| 95 | case err == nil: |
| 96 | if uerr := json.Unmarshal(raw, &day); uerr != nil { |
| 97 | return fmt.Errorf("usage: parse %s: %w", key, uerr) |
| 98 | } |
| 99 | case errors.Is(err, os.ErrNotExist): |
| 100 | etag = "" // create-only Put |
| 101 | default: |
| 102 | return err |
| 103 | } |
| 104 | day.AddDeltas(hour, deltas) |
| 105 | out, err := json.Marshal(day) |
| 106 | if err != nil { |
| 107 | return err |
| 108 | } |
| 109 | if _, err := s.Put(ctx, key, out, etag); err == nil { |
| 110 | return nil |
| 111 | } else if !errors.Is(err, release.ErrCASConflict) { |
| 112 | return err |
| 113 | } |
| 114 | } |
| 115 | return fmt.Errorf("usage: %s: lost CAS race three times", key) |
| 116 | } |
| 117 | |
amadan