amadan

carlos / platform Public

edge: observe responses for switched-on apps (spec §3.3)

The counting writer already latches the final status and Content-Type for
every response shape, and the observer defer already fires once per served
route. One more line in that defer files an analytics.Obs for apps whose
analytics the telemetry pass has switched on — no new wrapper, no second
pass over the request, and status-host traffic stays excluded by the guard
that already wraps the defer.

clientIP believes X-Forwarded-For only on a request that arrived as a
verified mesh hop; otherwise it is RemoteAddr's host. A header a client set
never reaches the visitor hash.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev cc838ab880e9f2774c47eb06f66618a694a8f280 parent 008d40b
3 files changed, +215 −0
  • internal/edge/analytics.go +108 −0
  • internal/edge/analytics_test.go +96 −0
  • internal/edge/edge.go +11 −0
diff --git a/internal/edge/analytics.go b/internal/edge/analytics.go
new file mode 100644
index 000000000..09bd27195
--- /dev/null
+++ b/internal/edge/analytics.go
@@ -0,0 +1,108 @@
+package edge
+
+import (
+ "context"
+ "net"
+ "net/http"
+ "net/url"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/analytics"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// AnalyticsMeter is where the edge files one observation per response
+// for an app whose analytics are switched on (spec §3.3). analytics.Meter
+// satisfies it; the interface keeps the dependency on the contract.
+type AnalyticsMeter interface {
+ Observe(analytics.Obs)
+}
+
+type analyticsMeterRef struct{ AnalyticsMeter }
+
+// EnableAnalytics wires the meter in; nil is a valid no-op.
+func (e *Edge) EnableAnalytics(m AnalyticsMeter) {
+ if m == nil {
+ e.analyticsMeter.Store(nil)
+ return
+ }
+ e.analyticsMeter.Store(&analyticsMeterRef{m})
+}
+
+// SetAnalyticsOn replaces the switch map: sqid+"/"+app → counted. The
+// host's telemetry pass pushes it on every serial-triggered full pass and
+// every five minutes (spec §3.3, "converged by boxes on the next pass").
+// The map is stored, never copied and never mutated here: a caller must
+// hand over a map it will not write to again.
+func (e *Edge) SetAnalyticsOn(on map[string]bool) {
+ if on == nil {
+ e.analyticsOn.Store(nil)
+ return
+ }
+ e.analyticsOn.Store(&on)
+}
+
+func (e *Edge) analyticsEnabled(rt registry.Route) bool {
+ m := e.analyticsOn.Load()
+ if m == nil {
+ return false
+ }
+ return (*m)[release.RouteAccount(rt)+"/"+rt.App]
+}
+
+// withHopVerified marks ctx as a request that arrived over a proven mesh
+// hop, exactly as acceptHop does (homemap.go). Test-only in practice —
+// nothing on the serving path may put this mark on a request, which is the
+// whole property clientIP's X-Forwarded-For trust rests on.
+func withHopVerified(ctx context.Context) context.Context {
+ return context.WithValue(ctx, hopVerifiedKey{}, true)
+}
+
+// clientIP is the address the visitor hash uses (spec §3.3): the last
+// element of the verified inbound X-Forwarded-For chain on a proven hop,
+// otherwise RemoteAddr's host. A header the client set never reaches it —
+// on an unverified request acceptHop has already stripped the forwarded
+// headers, and this function does not look at them either way.
+func clientIP(r *http.Request) string {
+ if hopVerified(r.Context()) {
+ chain := strings.Join(r.Header.Values("X-Forwarded-For"), ", ")
+ if i := strings.LastIndexByte(chain, ','); i >= 0 {
+ chain = chain[i+1:]
+ }
+ if ip := strings.TrimSpace(chain); ip != "" {
+ return ip
+ }
+ }
+ host, _, err := net.SplitHostPort(r.RemoteAddr)
+ if err != nil {
+ return r.RemoteAddr
+ }
+ return host
+}
+
+// observeAnalytics files one Obs when the route's app is switched on. It
+// runs inside the observer defer in ServeHTTP, after the response has
+// completed, and reads only: the latched status and Content-Type off the
+// counting writer, the byte count, and the request's own headers. No
+// wrapper is added for it and no path here can wake an instance.
+func (e *Edge) observeAnalytics(rt registry.Route, r *http.Request, ob *countingWriter, bytes int64) {
+ ref := e.analyticsMeter.Load()
+ if ref == nil || !e.analyticsEnabled(rt) {
+ return
+ }
+ refHost := ""
+ if u, err := url.Parse(r.Header.Get("Referer")); err == nil {
+ refHost = u.Hostname()
+ }
+ prefetch := strings.EqualFold(r.Header.Get("Sec-Purpose"), "prefetch") || strings.EqualFold(r.Header.Get("Purpose"), "prefetch")
+ ref.Observe(analytics.Obs{
+ Acct: release.RouteAccount(rt), App: rt.App, Host: rt.Host,
+ Method: r.Method, Status: ob.status, ContentType: ob.ctype,
+ Path: r.URL.Path, RefererHost: refHost, UA: r.Header.Get("User-Agent"),
+ ClientIP: clientIP(r), Bytes: bytes,
+ Prefetch: prefetch, FetchDest: strings.ToLower(r.Header.Get("Sec-Fetch-Dest")),
+ When: time.Now(),
+ })
+}
diff --git a/internal/edge/analytics_test.go b/internal/edge/analytics_test.go
new file mode 100644
index 000000000..173b24cce
--- /dev/null
+++ b/internal/edge/analytics_test.go
@@ -0,0 +1,96 @@
+package edge
+
+import (
+ "net/http"
+ "net/http/httptest"
+ "sync"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/analytics"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type memAnalytics struct {
+ mu sync.Mutex
+ obs []analytics.Obs
+}
+
+func (m *memAnalytics) Observe(o analytics.Obs) { m.mu.Lock(); m.obs = append(m.obs, o); m.mu.Unlock() }
+func (m *memAnalytics) snap() []analytics.Obs {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ return append([]analytics.Obs(nil), m.obs...)
+}
+
+func TestClientIP(t *testing.T) {
+ r := httptest.NewRequest("GET", "http://h/", nil)
+ r.RemoteAddr = "203.0.113.5:4444"
+ r.Header.Set("X-Forwarded-For", "198.51.100.1")
+ if ip := clientIP(r); ip != "203.0.113.5" {
+ t.Fatalf("unverified: %q", ip)
+ }
+ r2 := r.WithContext(withHopVerified(r.Context()))
+ r2.Header.Set("X-Forwarded-For", "10.0.0.1, 198.51.100.7")
+ if ip := clientIP(r2); ip != "198.51.100.7" {
+ t.Fatalf("verified hop: %q", ip)
+ }
+}
+
+func TestEdgeObservesOnlyWhenSwitchedOn(t *testing.T) {
+ backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "text/html; charset=utf-8")
+ w.Write([]byte("<h1>hi</h1>"))
+ }))
+ t.Cleanup(backend.Close)
+ reg := openReg(t)
+ rt := registry.Route{Host: "app.example.com", Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready", App: "web"}
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ m := &memAnalytics{}
+ e.EnableAnalytics(m)
+ srv := httptest.NewServer(e)
+ t.Cleanup(srv.Close)
+ get := func(path string, hdr map[string]string) {
+ req, _ := http.NewRequest("GET", srv.URL+path, nil)
+ req.Host = "app.example.com"
+ for k, v := range hdr {
+ req.Header.Set(k, v)
+ }
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp.Body.Close()
+ }
+ get("/", nil)
+ if len(m.snap()) != 0 {
+ t.Fatal("nothing is observed while the app is off")
+ }
+ e.SetAnalyticsOn(map[string]bool{release.OpsAccount + "/web": true})
+ get("/pricing?x=1", map[string]string{"Referer": "https://news.ycombinator.com/item?id=1", "User-Agent": "Mozilla/5.0 Safari", "Sec-Fetch-Dest": "document"})
+ obs := m.snap()
+ if len(obs) != 1 {
+ t.Fatalf("observed %d", len(obs))
+ }
+ o := obs[0]
+ if o.Acct != release.OpsAccount || o.App != "web" || o.Host != "app.example.com" || o.Method != "GET" || o.Status != 200 ||
+ o.ContentType != "text/html; charset=utf-8" || o.Path != "/pricing" || o.RefererHost != "news.ycombinator.com" ||
+ o.FetchDest != "document" || o.Bytes == 0 || o.ClientIP == "" || o.When.IsZero() {
+ t.Fatalf("obs = %+v", o)
+ }
+ get("/x", map[string]string{"Purpose": "prefetch"})
+ if !m.snap()[1].Prefetch {
+ t.Fatal("prefetch header must be carried")
+ }
+ e.SetAnalyticsOn(nil)
+ get("/", nil)
+ if len(m.snap()) != 2 {
+ t.Fatal("switching off stops observation")
+ }
+}
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 1a372d47a..cbd40ad2f 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -125,6 +125,16 @@ type Edge struct {
// errSink, when set, receives edge-observed errors (an upstream 5xx)
// — spec 2026-09-07 §3.2. Nil disables emission; nothing else changes.
errSink atomic.Pointer[errorSinkRef]
+ // analyticsMeter, when set, receives one observation per completed
+ // response for an app whose analytics are switched on (analytics.go,
+ // spec 2026-09-07 §3.3), and analyticsOn is that switch map keyed
+ // sqid+"/"+app. Both are atomic for the same reason usageMeter is: the
+ // agent role arms the meter after construction, and the telemetry pass
+ // replaces the switch map while serving is underway. Nil meter or nil
+ // map = the edge observes nothing, which is every `carlos edge` box and
+ // every app nobody has switched on.
+ analyticsMeter atomic.Pointer[analyticsMeterRef]
+ analyticsOn atomic.Pointer[map[string]bool]
// 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
@@ -1305,6 +1315,7 @@ func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if ob.status >= 500 && (answered == nil || !*answered) {
e.emitEdgeError(up.route, r, ob.status)
}
+ e.observeAnalytics(up.route, r, ob, bytes)
}()
}