amadan

carlos / platform Public

edge: per-route request/egress metering behind EnableUsage

Adds UsageMeter (internal/edge/usagemeter.go): a mutex'd map of
usage.EdgeRow keyed by host, Record()'d per request and Drain()'d by
the (later) agent flush loop. countingWriter wraps the ResponseWriter
to count body bytes without buffering; deliberately omits ReaderFrom
so io.Copy falls back to Write.

Edge.EnableUsage arms metering atomically. ServeHTTP wraps w in a
countingWriter once, right after route lookup succeeds and before the
upgrade/cache branching, so all three dispatch shapes (hibernating,
cacheable, plain) plus the upgrade path count through one site via a
single defer. Kind=="status" routes are excluded (not billable); the
tunnel-registration hostname and store door return before route
lookup and were already out of scope.

Extends TestRoutesHostToTCPUpstream's e2e shape with
TestServeHTTPMetersRequestsAndBytesExcludingStatus: 2 GETs to a
TCP-backed route show Requests==2 and BytesOut equal to served body
bytes, and a Kind=="status" route never appears in Drain().
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell ba1d9229439788335ce2a4262863ea1e8816a881 parent e0d761a
4 files changed, +247 −0
  • internal/edge/edge.go +27 −0
  • internal/edge/edge_test.go +70 −0
  • internal/edge/usagemeter.go +84 −0
  • internal/edge/usagemeter_test.go +66 −0
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 219e8697e..3a6cd1bf3 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -101,6 +101,12 @@ type Edge struct {
staticPub ed25519.PublicKey
staticCacheDir string
staticRecheck time.Duration
+ // usageMeter, when non-nil, is where ServeHTTP records the request/byte
+ // count of every route it serves (see EnableUsage). Atomic because it is
+ // armed by the agent role after Edge construction, mirroring how
+ // tunnelLoop attaches via the ready hook — serving may already be
+ // underway.
+ usageMeter atomic.Pointer[UsageMeter]
// tunnels, when non-nil, is the reverse-channel hub (tunnel.go): remote
// hosts register on its reserved hostname and routes with Tunnel set are
// proxied over the channel they name. nil = tunnel serving not configured
@@ -198,6 +204,12 @@ func (e *Edge) EnableTunnels(hub *TunnelHub) {
e.armStoreDoor()
}
+// EnableUsage arms per-route request/byte metering. Called by the agent
+// role after construction (mirroring how tunnelLoop attaches via the ready
+// hook); the bare `carlos edge` role never calls it, so pure-proxy edges
+// meter nothing. Safe to call after serving has started (atomic).
+func (e *Edge) EnableUsage(m *UsageMeter) { e.usageMeter.Store(m) }
+
// armStoreDoor builds (or clears) the fleet store door from whatever is
// configured now. Called from both EnableStatic and EnableTunnels because the
// door needs one thing from each and the two are wired in either order at
@@ -548,6 +560,21 @@ func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return
}
+ // Meter every route this box serves except the platform's own status
+ // host (not billable) — wrapping w here, once, before the upgrade/cache
+ // branching below, is what makes all three dispatch shapes (hibernating,
+ // cacheable, plain) plus the upgrade path count through this single
+ // site: the defer fires after whichever branch actually served the
+ // request. Deliberately placed after the tunnel-hostname and store-door
+ // returns above (edge.go:531-544) — that hostname isn't a registry
+ // route, so up.route isn't defined yet there, and neither one is
+ // billable traffic.
+ if m := e.usageMeter.Load(); m != nil && up.route.Kind != "status" {
+ var bytes int64
+ w = &countingWriter{ResponseWriter: w, n: &bytes}
+ defer func() { m.Record(up.route, 1, bytes) }()
+ }
+
// A Connection: Upgrade request (a WebSocket handshake, typically GET)
// must reach the activator/proxy with the real ResponseWriter, completely
// untouched by caching: the stdlib reverse proxy's upgrade path type-
diff --git a/internal/edge/edge_test.go b/internal/edge/edge_test.go
index 80d00b007..0e2197a38 100644
--- a/internal/edge/edge_test.go
+++ b/internal/edge/edge_test.go
@@ -1527,3 +1527,73 @@ func TestNoVersionHeaderOnUnversionedRoute(t *testing.T) {
t.Fatalf("X-Carlos-Version present (%v), want absent for an unversioned route", got)
}
}
+
+// TestServeHTTPMetersRequestsAndBytesExcludingStatus drives a full Edge over
+// HTTP (mirroring TestRoutesHostToTCPUpstream) with EnableUsage armed, and
+// proves the single wrap site in ServeHTTP counts every plain-path request
+// while a Kind=="status" route (the platform's own status page — not
+// billable, per the placement rules) never shows up in Drain().
+func TestServeHTTPMetersRequestsAndBytesExcludingStatus(t *testing.T) {
+ body := "metered response body"
+ backend, _ := hostEchoBackend(t, body)
+ addr := backend.Listener.Addr().String()
+
+ reg := openReg(t)
+ if err := reg.Upsert(registry.Route{Host: "metered.example.com", Kind: "service", Addr: addr, Account: "bab", App: "hello", Status: "ready"}); err != nil {
+ t.Fatalf("upsert metered route: %v", err)
+ }
+ if err := reg.Upsert(registry.Route{Host: "status.example.com", Kind: "status", Status: "ready"}); err != nil {
+ t.Fatalf("upsert status route: %v", err)
+ }
+
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ // Static/status serving needs a store wired in for the status route to
+ // answer at all (see testEdgeWithStore); an empty pubkey is fine since
+ // the status class never verifies signatures.
+ store := release.NewDirStore(filepath.Join(t.TempDir(), "store"))
+ _, pubB64, err := release.GenerateKey()
+ if err != nil {
+ t.Fatalf("generate key: %v", err)
+ }
+ pub, err := release.ParsePublicKey(pubB64, "test")
+ if err != nil {
+ t.Fatalf("parse public key: %v", err)
+ }
+ e.EnableStatic(store, pub, filepath.Join(t.TempDir(), "cache"))
+ store.Put(context.Background(), "status/index.html", []byte("<html>status</html>"), "")
+ if err := e.Reload(); err != nil {
+ t.Fatalf("Reload: %v", err)
+ }
+
+ meter := NewUsageMeter()
+ e.EnableUsage(meter)
+
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+
+ for i := 0; i < 2; i++ {
+ code, got := getHost(t, srv.URL, "metered.example.com")
+ if code != http.StatusOK || got != body {
+ t.Fatalf("request %d: %d %q, want 200 %q", i, code, got, body)
+ }
+ }
+ if code, _ := getHost(t, srv.URL, "status.example.com"); code != http.StatusOK {
+ t.Fatalf("status route: %d, want 200", code)
+ }
+
+ rows := meter.Drain()
+ row := rows["metered.example.com"]
+ wantBytes := int64(2 * len(body))
+ if row.Requests != 2 || row.BytesOut != wantBytes {
+ t.Fatalf("metered.example.com row = %+v, want Requests=2 BytesOut=%d", row, wantBytes)
+ }
+ if row.Account != "bab" || row.App != "hello" {
+ t.Fatalf("metered.example.com row = %+v, want Account=bab App=hello", row)
+ }
+ if _, ok := rows["status.example.com"]; ok {
+ t.Fatalf("status route must not be metered, got %+v", rows["status.example.com"])
+ }
+}
diff --git a/internal/edge/usagemeter.go b/internal/edge/usagemeter.go
new file mode 100644
index 000000000..b82caa55a
--- /dev/null
+++ b/internal/edge/usagemeter.go
@@ -0,0 +1,84 @@
+package edge
+
+import (
+ "bufio"
+ "fmt"
+ "net"
+ "net/http"
+ "sync"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/usage"
+)
+
+// UsageMeter accumulates per-route request/byte counters between flushes.
+// Record sits on the serving path: one mutex'd map update per request —
+// no allocation, no I/O. Drain swaps the map out for the usage flush loop
+// (internal/host), so a slow bucket write never blocks serving.
+type UsageMeter struct {
+ mu sync.Mutex
+ m map[string]usage.EdgeRow
+}
+
+// NewUsageMeter returns an empty meter.
+func NewUsageMeter() *UsageMeter {
+ return &UsageMeter{m: map[string]usage.EdgeRow{}}
+}
+
+// Record adds one served exchange for rt.
+func (u *UsageMeter) Record(rt registry.Route, requests, bytes int64) {
+ u.mu.Lock()
+ row := u.m[rt.Host]
+ row.Account = release.RouteAccount(rt)
+ row.App = rt.App
+ row.Requests += requests
+ row.BytesOut += bytes
+ u.m[rt.Host] = row
+ u.mu.Unlock()
+}
+
+// Drain returns the accumulated rows and resets the meter.
+func (u *UsageMeter) Drain() map[string]usage.EdgeRow {
+ u.mu.Lock()
+ out := u.m
+ u.m = map[string]usage.EdgeRow{}
+ u.mu.Unlock()
+ return out
+}
+
+// countingWriter counts response body bytes. Deliberately does NOT
+// implement io.ReaderFrom: io.Copy falls back to Write, which counts.
+// Hijack/Flush forward via the interfaces the wrapped writer provides,
+// mirroring recordingResponseWriter (edge.go).
+type countingWriter struct {
+ http.ResponseWriter
+ n *int64
+}
+
+func (c *countingWriter) Write(p []byte) (int, error) {
+ n, err := c.ResponseWriter.Write(p)
+ *c.n += int64(n)
+ return n, err
+}
+
+// Hijack forwards to the underlying ResponseWriter's http.Hijacker, when it
+// implements one — the same optional-interface pattern the stdlib itself
+// uses, and the same forwarding recordingResponseWriter provides.
+func (c *countingWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+ hj, ok := c.ResponseWriter.(http.Hijacker)
+ if !ok {
+ return nil, nil, fmt.Errorf("edge: underlying ResponseWriter does not support http.Hijacker")
+ }
+ return hj.Hijack()
+}
+
+// Flush forwards to the underlying ResponseWriter's http.Flusher, when it
+// implements one — needed so a metered (non-upgrade) response can still
+// stream incrementally to the real client, e.g. Server-Sent Events, instead
+// of buffering until the handler returns.
+func (c *countingWriter) Flush() {
+ if f, ok := c.ResponseWriter.(http.Flusher); ok {
+ f.Flush()
+ }
+}
diff --git a/internal/edge/usagemeter_test.go b/internal/edge/usagemeter_test.go
new file mode 100644
index 000000000..983174211
--- /dev/null
+++ b/internal/edge/usagemeter_test.go
@@ -0,0 +1,66 @@
+package edge
+
+import (
+ "net/http"
+ "net/http/httptest"
+ "sync"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestUsageMeterRecordAndDrain(t *testing.T) {
+ m := NewUsageMeter()
+ rt := registry.Route{Host: "a.example.com", App: "hello", Account: "bab"}
+ m.Record(rt, 1, 100)
+ m.Record(rt, 1, 50)
+ m.Record(registry.Route{Host: "b.example.com", App: "site"}, 1, 10) // Account "" -> release.OpsAccount
+
+ got := m.Drain()
+ a := got["a.example.com"]
+ if a.Requests != 2 || a.BytesOut != 150 || a.Account != "bab" || a.App != "hello" {
+ t.Fatalf("a row = %+v", a)
+ }
+ if got["b.example.com"].Account != release.OpsAccount {
+ t.Fatalf("b row = %+v", got["b.example.com"])
+ }
+ if again := m.Drain(); len(again) != 0 {
+ t.Fatalf("drain did not reset: %+v", again)
+ }
+}
+
+func TestUsageMeterConcurrent(t *testing.T) {
+ m := NewUsageMeter()
+ rt := registry.Route{Host: "a.example.com", App: "hello", Account: "bab"}
+ var wg sync.WaitGroup
+ for i := 0; i < 50; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for j := 0; j < 100; j++ {
+ m.Record(rt, 1, 10)
+ }
+ }()
+ }
+ wg.Wait()
+ row := m.Drain()["a.example.com"]
+ if row.Requests != 5000 || row.BytesOut != 50000 {
+ t.Fatalf("row = %+v", row)
+ }
+}
+
+func TestCountingWriterCountsBytes(t *testing.T) {
+ var n int64
+ rec := httptest.NewRecorder()
+ w := &countingWriter{ResponseWriter: rec, n: &n}
+ w.WriteHeader(http.StatusTeapot)
+ w.Write([]byte("hello "))
+ w.Write([]byte("world"))
+ if n != 11 {
+ t.Fatalf("counted %d bytes", n)
+ }
+ if rec.Code != http.StatusTeapot || rec.Body.String() != "hello world" {
+ t.Fatalf("forwarding broken: %d %q", rec.Code, rec.Body.String())
+ }
+}