| new file mode 100644 |
| index 000000000..559a8a477 |
| --- /dev/null |
| +++ b/internal/relay/meter.go |
| @@ -0,0 +1,159 @@ |
| +// Package relay: Meter accumulates per-app byte and packet counts and |
| +// flushes them as an hourly time series to the deployment bucket (spec §3 |
| +// metering). |
| +package relay |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "os" |
| + "sync" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/metrics" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// Meter satisfies ServerConfig.Meter (internal/relay/server.go). |
| +var _ interface { |
| + Add(id Identity, in, out, rejected int64) |
| +} = (*Meter)(nil) |
| + |
| +// MeterKey is the bucket key for a box's metering chunk covering the given |
| +// hour (metrics.HourOf's format, e.g. 2026-09-07-12). |
| +func MeterKey(box, hour string) string { return "metrics/relay/" + box + "/" + hour + ".json" } |
| + |
| +// MeterApp is one app's counters for one sample interval. In and Out are |
| +// bytes forwarded on the two relay legs, counted at the socket writes — Out |
| +// includes the STUN Data-indication framing (about 36 bytes per packet), so |
| +// it is larger than the raw media payload. Rejected is every peer packet |
| +// that arrived at a relay socket and was not forwarded (no permission, over |
| +// a limit, etc). Allocs is a gauge: the number of open allocations at |
| +// sample time, not a delta. |
| +type MeterApp struct { |
| + In int64 `json:"in"` |
| + Out int64 `json:"out"` |
| + Rejected int64 `json:"rejected"` |
| + Allocs int `json:"allocs"` |
| +} |
| + |
| +// MeterSample is one flush's worth of per-app counters, keyed |
| +// "<account>/<app>". |
| +type MeterSample struct { |
| + At time.Time `json:"at"` |
| + Apps map[string]MeterApp `json:"apps"` |
| +} |
| + |
| +// MeterChunk is the hourly object at MeterKey: every sample flushed for |
| +// that box and hour, in order. |
| +type MeterChunk struct { |
| + V int `json:"v"` |
| + Box string `json:"box"` |
| + Hour string `json:"hour"` |
| + Samples []MeterSample `json:"samples"` |
| +} |
| + |
| +// Meter accumulates per-app counters between flushes and appends them as a |
| +// sample to the box's hourly object. Safe for concurrent use. |
| +type Meter struct { |
| + mu sync.Mutex |
| + box string |
| + now func() time.Time |
| + cur map[Identity]*MeterApp |
| +} |
| + |
| +// NewMeter returns a Meter for box, using now for both sample timestamps |
| +// and hour selection (a func so tests control it). |
| +func NewMeter(box string, now func() time.Time) *Meter { |
| + return &Meter{box: box, now: now, cur: map[Identity]*MeterApp{}} |
| +} |
| + |
| +func (m *Meter) get(id Identity) *MeterApp { |
| + a, ok := m.cur[id] |
| + if !ok { |
| + a = &MeterApp{} |
| + m.cur[id] = a |
| + } |
| + return a |
| +} |
| + |
| +// Add accumulates in/out bytes and rejected packets for id since the last |
| +// flush. |
| +func (m *Meter) Add(id Identity, in, out, rejected int64) { |
| + m.mu.Lock() |
| + a := m.get(id) |
| + a.In += in |
| + a.Out += out |
| + a.Rejected += rejected |
| + m.mu.Unlock() |
| +} |
| + |
| +// SetAllocs sets id's current open-allocation gauge. |
| +func (m *Meter) SetAllocs(id Identity, n int) { |
| + m.mu.Lock() |
| + m.get(id).Allocs = n |
| + m.mu.Unlock() |
| +} |
| + |
| +// Flush writes one sample of deltas since the previous flush, appended to |
| +// this box's hourly object by read-append-CAS. The relay is the only |
| +// writer of these objects, so the CAS only ever races with itself. |
| +func (m *Meter) Flush(ctx context.Context, s release.Store) error { |
| + m.mu.Lock() |
| + now := m.now() |
| + sample := MeterSample{At: now, Apps: map[string]MeterApp{}} |
| + for id, a := range m.cur { |
| + sample.Apps[id.Account+"/"+id.App] = *a |
| + m.cur[id] = &MeterApp{Allocs: a.Allocs} // deltas reset, gauge kept |
| + } |
| + m.mu.Unlock() |
| + hour := metrics.HourOf(now) |
| + key := MeterKey(m.box, hour) |
| + var lastErr error |
| + for attempt := 0; attempt < 3; attempt++ { |
| + body, etag, err := s.Get(ctx, key) |
| + var c MeterChunk |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + c = MeterChunk{V: 1, Box: m.box, Hour: hour} |
| + etag = "" |
| + case err != nil: |
| + return err |
| + default: |
| + if json.Unmarshal(body, &c) != nil { |
| + c = MeterChunk{V: 1, Box: m.box, Hour: hour} |
| + } |
| + } |
| + c.Samples = append(c.Samples, sample) |
| + out, _ := json.Marshal(c) |
| + if _, err = s.Put(ctx, key, out, etag); err == nil { |
| + return nil |
| + } else if !errors.Is(err, release.ErrCASConflict) { |
| + return err |
| + } else { |
| + lastErr = err |
| + } |
| + } |
| + return lastErr |
| +} |
| + |
| +// LoadWindow returns every sample from the previous and current hour's |
| +// objects, for rebuilding rolling-hour budgets after a restart. |
| +func LoadWindow(ctx context.Context, s release.Store, box string, now time.Time) ([]MeterSample, error) { |
| + var out []MeterSample |
| + for _, h := range []string{metrics.HourOf(now.Add(-time.Hour)), metrics.HourOf(now)} { |
| + body, _, err := s.Get(ctx, MeterKey(box, h)) |
| + if errors.Is(err, os.ErrNotExist) { |
| + continue |
| + } |
| + if err != nil { |
| + return nil, err |
| + } |
| + var c MeterChunk |
| + if json.Unmarshal(body, &c) == nil { |
| + out = append(out, c.Samples...) |
| + } |
| + } |
| + return out, nil |
| +} |
amadan