| 1 | // Package relay: Meter accumulates per-app byte and packet counts and |
| 2 | // flushes them as an hourly time series to the deployment bucket (spec §3 |
| 3 | // metering). |
| 4 | package relay |
| 5 | |
| 6 | import ( |
| 7 | "context" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "os" |
| 11 | "strconv" |
| 12 | "sync" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/metrics" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | // Meter satisfies ServerConfig.Meter (internal/relay/server.go). |
| 20 | var _ interface { |
| 21 | Add(id Identity, in, out, rejected int64) |
| 22 | } = (*Meter)(nil) |
| 23 | |
| 24 | // MeterKey is the bucket key for a box's metering chunk covering the given |
| 25 | // hour (metrics.HourOf's format, e.g. 2026-09-07-12). |
| 26 | func MeterKey(box, hour string) string { return "metrics/relay/" + box + "/" + hour + ".json" } |
| 27 | |
| 28 | // MeterApp is one app's counters for one sample interval. In and Out are |
| 29 | // bytes forwarded on the two relay legs, counted at the socket writes — Out |
| 30 | // includes the STUN Data-indication framing (about 36 bytes per packet), so |
| 31 | // it is larger than the raw media payload. Rejected is every peer packet |
| 32 | // that arrived at a relay socket and was not forwarded (no permission, over |
| 33 | // a limit, etc). Allocs is a gauge: the number of open allocations at |
| 34 | // sample time, not a delta. |
| 35 | type MeterApp struct { |
| 36 | In int64 `json:"in"` |
| 37 | Out int64 `json:"out"` |
| 38 | Rejected int64 `json:"rejected"` |
| 39 | Allocs int `json:"allocs"` |
| 40 | } |
| 41 | |
| 42 | // MeterSample is one flush's worth of per-app counters, keyed |
| 43 | // "<account>/<app>". |
| 44 | type MeterSample struct { |
| 45 | At time.Time `json:"at"` |
| 46 | Apps map[string]MeterApp `json:"apps"` |
| 47 | } |
| 48 | |
| 49 | // MeterChunk is the hourly object at MeterKey: every sample flushed for |
| 50 | // that box and hour, in order. |
| 51 | type MeterChunk struct { |
| 52 | V int `json:"v"` |
| 53 | Box string `json:"box"` |
| 54 | Hour string `json:"hour"` |
| 55 | Samples []MeterSample `json:"samples"` |
| 56 | } |
| 57 | |
| 58 | // Meter accumulates per-app counters between flushes and appends them as a |
| 59 | // sample to the box's hourly object. Safe for concurrent use. |
| 60 | type Meter struct { |
| 61 | mu sync.Mutex |
| 62 | box string |
| 63 | now func() time.Time |
| 64 | // Log receives diagnostics (a corrupt chunk found, etc); nil discards. |
| 65 | Log func(string, ...any) |
| 66 | cur map[Identity]*MeterApp |
| 67 | } |
| 68 | |
| 69 | // NewMeter returns a Meter for box, using now for both sample timestamps |
| 70 | // and hour selection (a func so tests control it), and logf for |
| 71 | // diagnostics (nil discards them). |
| 72 | func NewMeter(box string, now func() time.Time, logf func(string, ...any)) *Meter { |
| 73 | return &Meter{box: box, now: now, Log: logf, cur: map[Identity]*MeterApp{}} |
| 74 | } |
| 75 | |
| 76 | func (m *Meter) log(format string, args ...any) { |
| 77 | if m.Log != nil { |
| 78 | m.Log(format, args...) |
| 79 | } |
| 80 | } |
| 81 | |
| 82 | func (m *Meter) get(id Identity) *MeterApp { |
| 83 | a, ok := m.cur[id] |
| 84 | if !ok { |
| 85 | a = &MeterApp{} |
| 86 | m.cur[id] = a |
| 87 | } |
| 88 | return a |
| 89 | } |
| 90 | |
| 91 | // Add accumulates in/out bytes and rejected packets for id since the last |
| 92 | // flush. |
| 93 | func (m *Meter) Add(id Identity, in, out, rejected int64) { |
| 94 | m.mu.Lock() |
| 95 | a := m.get(id) |
| 96 | a.In += in |
| 97 | a.Out += out |
| 98 | a.Rejected += rejected |
| 99 | m.mu.Unlock() |
| 100 | } |
| 101 | |
| 102 | // SetAllocs sets id's current open-allocation gauge. |
| 103 | func (m *Meter) SetAllocs(id Identity, n int) { |
| 104 | m.mu.Lock() |
| 105 | m.get(id).Allocs = n |
| 106 | m.mu.Unlock() |
| 107 | } |
| 108 | |
| 109 | // SetAllocsAll sets the gauge from a complete census of live allocations: |
| 110 | // every id in live gets its count, and every id the meter is carrying that |
| 111 | // live does NOT name gets zero. |
| 112 | // |
| 113 | // The zeroing half is the point. Flush keeps an app whose gauge is above |
| 114 | // zero and evicts one whose four counters are all zero, so an app that is |
| 115 | // only ever set and never cleared reports a phantom allocation in every |
| 116 | // sample from then on and is never evicted — a slow leak in a process the |
| 117 | // unit caps at 256 MiB, and a lie in the numbers the bill is made of. |
| 118 | // Setting the gauge one id at a time cannot do this, because the ids that |
| 119 | // need zeroing are precisely the ones the caller no longer sees. |
| 120 | func (m *Meter) SetAllocsAll(live map[Identity]int) { |
| 121 | m.mu.Lock() |
| 122 | defer m.mu.Unlock() |
| 123 | for id, a := range m.cur { |
| 124 | a.Allocs = live[id] // absent from live: zero |
| 125 | } |
| 126 | for id, n := range live { |
| 127 | m.get(id).Allocs = n |
| 128 | } |
| 129 | } |
| 130 | |
| 131 | // Flush writes one sample of deltas since the previous flush, appended to |
| 132 | // this box's hourly object by read-append-CAS. The relay is the only |
| 133 | // writer of these objects, so the CAS only ever races with itself. |
| 134 | // |
| 135 | // An app with nothing to report — every delta zero and no open |
| 136 | // allocations — is dropped from the sample and from the meter's live set, |
| 137 | // so a burst of short-lived apps does not grow m.cur without bound. An app |
| 138 | // with open allocations is kept (with zero deltas) even while idle, so its |
| 139 | // allocation gauge keeps being reported. And a sample left naming no app at |
| 140 | // all is not written: an idle relay writes nothing rather than a minute-by- |
| 141 | // minute record of having nothing to say. |
| 142 | func (m *Meter) Flush(ctx context.Context, s release.Store) error { |
| 143 | m.mu.Lock() |
| 144 | now := m.now() |
| 145 | sample := MeterSample{At: now, Apps: map[string]MeterApp{}} |
| 146 | for id, a := range m.cur { |
| 147 | if a.In == 0 && a.Out == 0 && a.Rejected == 0 && a.Allocs == 0 { |
| 148 | delete(m.cur, id) |
| 149 | continue |
| 150 | } |
| 151 | sample.Apps[id.Account+"/"+id.App] = *a |
| 152 | m.cur[id] = &MeterApp{Allocs: a.Allocs} // deltas reset, gauge kept |
| 153 | } |
| 154 | m.mu.Unlock() |
| 155 | // Nothing to say, so say nothing. Almost every relay box is idle almost |
| 156 | // all of the time, and a sample with no apps in it is a read, a write and |
| 157 | // a row of JSON that carries no information — 1,440 of them a day per |
| 158 | // box, growing the hourly object a reader has to walk and the CAS chain |
| 159 | // that guards it. An hour in which nothing happened is best represented |
| 160 | // by an hour with no object. |
| 161 | if len(sample.Apps) == 0 { |
| 162 | return nil |
| 163 | } |
| 164 | hour := metrics.HourOf(now) |
| 165 | key := MeterKey(m.box, hour) |
| 166 | var lastErr error |
| 167 | for attempt := 0; attempt < 3; attempt++ { |
| 168 | body, etag, err := s.Get(ctx, key) |
| 169 | var c MeterChunk |
| 170 | switch { |
| 171 | case errors.Is(err, os.ErrNotExist): |
| 172 | c = MeterChunk{V: 1, Box: m.box, Hour: hour} |
| 173 | etag = "" |
| 174 | case err != nil: |
| 175 | return err |
| 176 | default: |
| 177 | if uerr := json.Unmarshal(body, &c); uerr != nil { |
| 178 | m.log("relay: meter chunk %s unreadable (%v); starting a fresh chunk, %d bytes discarded", key, uerr, len(body)) |
| 179 | corruptKey := key + ".corrupt-" + strconv.FormatInt(now.Unix(), 10) |
| 180 | if _, perr := s.Put(ctx, corruptKey, body, ""); perr != nil { |
| 181 | m.log("relay: could not preserve corrupt meter chunk %s as %s: %v", key, corruptKey, perr) |
| 182 | } |
| 183 | c = MeterChunk{V: 1, Box: m.box, Hour: hour} |
| 184 | } |
| 185 | } |
| 186 | c.Samples = append(c.Samples, sample) |
| 187 | out, _ := json.Marshal(c) |
| 188 | if _, err = s.Put(ctx, key, out, etag); err == nil { |
| 189 | return nil |
| 190 | } else if !errors.Is(err, release.ErrCASConflict) { |
| 191 | return err |
| 192 | } else { |
| 193 | lastErr = err |
| 194 | } |
| 195 | } |
| 196 | return lastErr |
| 197 | } |
| 198 | |
| 199 | // LoadWindow returns every sample from the previous and current hour's |
| 200 | // objects, for rebuilding rolling-hour budgets after a restart. An hour |
| 201 | // whose chunk is corrupt is logged via logf (nil discards) and skipped |
| 202 | // rather than failing the whole load. |
| 203 | func LoadWindow(ctx context.Context, s release.Store, box string, now time.Time, logf func(string, ...any)) ([]MeterSample, error) { |
| 204 | var out []MeterSample |
| 205 | for _, h := range []string{metrics.HourOf(now.Add(-time.Hour)), metrics.HourOf(now)} { |
| 206 | key := MeterKey(box, h) |
| 207 | body, _, err := s.Get(ctx, key) |
| 208 | if errors.Is(err, os.ErrNotExist) { |
| 209 | continue |
| 210 | } |
| 211 | if err != nil { |
| 212 | return nil, err |
| 213 | } |
| 214 | var c MeterChunk |
| 215 | if uerr := json.Unmarshal(body, &c); uerr != nil { |
| 216 | if logf != nil { |
| 217 | logf("relay: meter chunk %s unreadable (%v); budgets rebuilt without it", key, uerr) |
| 218 | } |
| 219 | continue |
| 220 | } |
| 221 | out = append(out, c.Samples...) |
| 222 | } |
| 223 | return out, nil |
| 224 | } |
| 225 | |
amadan