amadan

carlos / platform Public

host: resolve analytics switches per pass; flush the analytics meter on the usage tick

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 9161c6df13c828463594e292493d9b47ae43acc9 parent f682770
5 files changed, +180 −10
  • internal/host/agent.go +11 −1
  • internal/host/telemetry_pass.go +69 −0
  • internal/host/telemetry_pass_test.go +38 −0
  • internal/host/usage_loop.go +31 −6
  • internal/host/usage_loop_test.go +31 −3
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 14d1385a2..79b9b90a4 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -165,6 +165,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/analytics"
"github.com/carlosframework/platform/internal/appconfig"
"github.com/carlosframework/platform/internal/apperr"
"github.com/carlosframework/platform/internal/applog"
@@ -446,7 +447,14 @@ func RunAgent(args []string) {
if errSink != nil {
e.EnableErrors(errSink)
}
- go usageLoop(m.Drain)
+ // The analytics meter (spec 2026-09-07 §3.3): the edge counts
+ // pageviews per (account, app) while telemetryPass (fullPass and
+ // instancesLoop) keeps its switch map current, and usageLoop
+ // drains it into day objects on the same 5-minute tick as usage.
+ am := analytics.NewMeter(dataDir(), status.Label(readMachineID()), time.Now)
+ e.EnableAnalytics(am)
+ setAnalyticsSink(e.SetAnalyticsOn)
+ go usageLoop(m.Drain, am)
},
// Certificate issuance is a deployment-global duty, so it runs where
// the housekeeper runs (housekeeper.go). The gate goes DOWN into the
@@ -1506,6 +1514,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
if err := reconcileSerialized(ctx, store, reg, provisionTickConfig(os.Getenv)); err != nil {
log.Printf("agent: fast converge reconcile: %v", err)
}
+ telemetryPass(ctx, store, reg)
runMovesPass(ctx, store)
// Folded once per pass and handed in, never re-read inside the
// loop: this leg runs on every serial bump (a seconds cadence),
@@ -2005,6 +2014,7 @@ func instancesLoop(reg *registry.Registry) {
if err := reconcileSerialized(ctx, store, reg, cfg); err != nil {
log.Printf("agent: instances pass: %v", err)
}
+ telemetryPass(ctx, store, reg)
runMovesPass(ctx, store)
deploymentRoutes := edgesfile.DeploymentRoutes(ctx, store, time.Now())
if err := domains.ReconcileAliases(ctx, store, reg, time.Now, deploymentRoutes, &aliasClaimCache); err != nil {
diff --git a/internal/host/telemetry_pass.go b/internal/host/telemetry_pass.go
new file mode 100644
index 000000000..1891c8d19
--- /dev/null
+++ b/internal/host/telemetry_pass.go
@@ -0,0 +1,69 @@
+// The analytics switch pass (spec 2026-09-07 §3.3): resolve
+// telemetry.json app → account → default for every (account, app) this
+// box serves and hand the edge a fresh switch map. Rides the serial-
+// triggered full pass and the five-minute instances loop, so a console
+// write lands within seconds and the hourly floor stands underneath.
+package host
+
+import (
+ "context"
+ "log"
+ "sync/atomic"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/telemetry"
+)
+
+var analyticsSink atomic.Pointer[func(map[string]bool)]
+
+// setAnalyticsSink registers where a resolved switch map goes (the edge's
+// SetAnalyticsOn), the way setUpstreamSource publishes the prober table.
+func setAnalyticsSink(fn func(map[string]bool)) { analyticsSink.Store(&fn) }
+
+// telemetrySwitches resolves the count switch for each distinct
+// (account, app) among the box's routes. Status routes carry no app and
+// are skipped. One Resolve (two Gets) per pair; a pair that fails to
+// resolve is left OFF and logged — off is the safe side of this switch.
+func telemetrySwitches(ctx context.Context, s release.Store, reg *registry.Registry) (map[string]bool, error) {
+ routes, err := reg.List()
+ if err != nil {
+ return nil, err
+ }
+ on := map[string]bool{}
+ seen := map[string]bool{}
+ for _, rt := range routes {
+ if rt.Kind == "status" || rt.App == "" {
+ continue
+ }
+ sqid := release.RouteAccount(rt)
+ key := sqid + "/" + rt.App
+ if seen[key] {
+ continue
+ }
+ seen[key] = true
+ res, err := telemetry.Resolve(ctx, s, sqid, rt.App)
+ if err != nil {
+ log.Printf("agent: telemetry resolve %s: %v (analytics stays off)", key, err)
+ continue
+ }
+ if res.Count == telemetry.CountEdge {
+ on[key] = true
+ }
+ }
+ return on, nil
+}
+
+// telemetryPass resolves and pushes; errors are one log line each.
+func telemetryPass(ctx context.Context, s release.Store, reg *registry.Registry) {
+ sink := analyticsSink.Load()
+ if sink == nil {
+ return
+ }
+ on, err := telemetrySwitches(ctx, s, reg)
+ if err != nil {
+ log.Printf("agent: telemetry pass: %v", err)
+ return
+ }
+ (*sink)(on)
+}
diff --git a/internal/host/telemetry_pass_test.go b/internal/host/telemetry_pass_test.go
new file mode 100644
index 000000000..04b43afda
--- /dev/null
+++ b/internal/host/telemetry_pass_test.go
@@ -0,0 +1,38 @@
+package host
+
+import (
+ "context"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/telemetry"
+)
+
+func TestTelemetrySwitchesResolvePerRoute(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ reg := edgesTestRegistry(t)
+ for _, rt := range []registry.Route{
+ {Host: "a.example", Kind: "instance", Account: "bab", App: "web", Status: "ready"},
+ {Host: "b.example", Kind: "service", Account: "bab", App: "api", Status: "ready"},
+ {Host: "c.example", Kind: "static", Account: "zzz", App: "site", Status: "ready"},
+ {Host: "status.example", Kind: "status", Status: "ready"},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ telemetry.Update(ctx, s, telemetry.AccountKey("bab"), func(st *telemetry.Settings) { st.Analytics = &telemetry.Analytics{Count: telemetry.CountEdge} })
+ telemetry.Update(ctx, s, telemetry.AppKey("bab", "api"), func(st *telemetry.Settings) { st.Analytics = &telemetry.Analytics{Count: telemetry.CountOff} })
+ on, err := telemetrySwitches(ctx, s, reg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !on["bab/web"] || on["bab/api"] || on["zzz/site"] {
+ t.Fatalf("switches = %v", on)
+ }
+ if _, ok := on["/"]; ok {
+ t.Fatal("status routes have no app and must not appear")
+ }
+}
diff --git a/internal/host/usage_loop.go b/internal/host/usage_loop.go
index b3af91669..9ec1d81d8 100644
--- a/internal/host/usage_loop.go
+++ b/internal/host/usage_loop.go
@@ -7,14 +7,38 @@ import (
"os"
"time"
+ "github.com/carlosframework/platform/internal/analytics"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/usage"
)
// usageFlush drains the edge meter and CAS-merges the deltas into the box's
-// usage/edge day object. One flush; the loop below owns cadence.
-func usageFlush(ctx context.Context, s release.Store, box string, drain func() map[string]usage.EdgeRow, now time.Time) error {
- return usage.UpsertEdgeDay(ctx, s, box, now, drain())
+// usage/edge day object, then (when am is non-nil) drains the analytics
+// meter's deltas into their app-day objects (spec §3.3: the same tick, a
+// second CAS write per app). Usage writes go to s (rebased under
+// CARLOS_STATUS_PREFIX on a fleet box); analytics writes go to raw — a day
+// file lives under apps/<sqid>/<app>/analytics/, the app's own data, and
+// must NOT be rebased (same reasoning as log shipping — logship.go:54-61).
+// A failed analytics batch is re-queued for the next tick; a failed usage
+// write is returned but does not block the analytics flush.
+func usageFlush(ctx context.Context, s, raw release.Store, box string, drain func() map[string]usage.EdgeRow, am *analytics.Meter, now time.Time) error {
+ err := usage.UpsertEdgeDay(ctx, s, box, now, drain())
+ if am == nil {
+ return err
+ }
+ failed := map[analytics.AppKey][]analytics.Delta{}
+ for k, ds := range am.Drain() {
+ for _, d := range ds {
+ if uerr := analytics.UpsertDay(ctx, raw, k.Acct, k.App, d); uerr != nil {
+ log.Printf("agent: analytics flush %s/%s %s: %v (re-queued)", k.Acct, k.App, d.Date, uerr)
+ failed[k] = append(failed[k], d)
+ }
+ }
+ }
+ if len(failed) > 0 {
+ am.Requeue(failed)
+ }
+ return err
}
// usageLoop flushes edge usage every 5 minutes, forever — the metricsLoop
@@ -23,8 +47,9 @@ func usageFlush(ctx context.Context, s release.Store, box string, drain func() m
// distinction at metrics_loop.go:773-779, rather than the brief's single log
// line, so the two loops read the same way in the logs), errors logged and
// never fatal (the drained window is lost, the next one is not; modelling
-// grade tolerates the gap).
-func usageLoop(drain func() map[string]usage.EdgeRow) {
+// grade tolerates the gap). am is the analytics meter (nil disables the
+// analytics half of the flush); its deltas are folded in on the same tick.
+func usageLoop(drain func() map[string]usage.EdgeRow, am *analytics.Meter) {
defer func() {
if r := recover(); r != nil {
log.Printf("agent: usage loop panic, edge metering stopped for this process: %v", r)
@@ -48,7 +73,7 @@ func usageLoop(drain func() map[string]usage.EdgeRow) {
log.Printf("agent: usage loop armed (5min flush), publishing to %s", target)
for {
time.Sleep(5 * time.Minute)
- if err := usageFlush(ctx, s, box, drain, time.Now()); err != nil {
+ if err := usageFlush(ctx, s, raw, box, drain, am, time.Now()); err != nil {
log.Printf("agent: usage flush: %v", err)
}
}
diff --git a/internal/host/usage_loop_test.go b/internal/host/usage_loop_test.go
index 9208f6f64..d8b5fabb3 100644
--- a/internal/host/usage_loop_test.go
+++ b/internal/host/usage_loop_test.go
@@ -6,6 +6,7 @@ import (
"testing"
"time"
+ "github.com/carlosframework/platform/internal/analytics"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/usage"
)
@@ -17,10 +18,10 @@ func TestUsageFlushWritesAndReflushesMerge(t *testing.T) {
rows := map[string]usage.EdgeRow{
"a.example.com": {Account: "bab", App: "hello", Requests: 3, BytesOut: 300},
}
- if err := usageFlush(ctx, s, "box1", func() map[string]usage.EdgeRow { return rows }, at); err != nil {
+ if err := usageFlush(ctx, s, s, "box1", func() map[string]usage.EdgeRow { return rows }, nil, at); err != nil {
t.Fatal(err)
}
- if err := usageFlush(ctx, s, "box1", func() map[string]usage.EdgeRow { return rows }, at.Add(5*time.Minute)); err != nil {
+ if err := usageFlush(ctx, s, s, "box1", func() map[string]usage.EdgeRow { return rows }, nil, at.Add(5*time.Minute)); err != nil {
t.Fatal(err)
}
raw, _, err := s.Get(ctx, usage.EdgeDayKey("box1", "2026-08-11"))
@@ -38,7 +39,7 @@ func TestUsageFlushWritesAndReflushesMerge(t *testing.T) {
func TestUsageFlushEmptyDrainWritesNothing(t *testing.T) {
s := release.NewDirStore(t.TempDir())
- err := usageFlush(context.Background(), s, "box1", func() map[string]usage.EdgeRow { return nil }, time.Now().UTC())
+ err := usageFlush(context.Background(), s, s, "box1", func() map[string]usage.EdgeRow { return nil }, nil, time.Now().UTC())
if err != nil {
t.Fatal(err)
}
@@ -46,3 +47,30 @@ func TestUsageFlushEmptyDrainWritesNothing(t *testing.T) {
t.Fatalf("wrote %v", keys)
}
}
+
+// TestUsageFlushAlsoFlushesAnalytics: the same tick that flushes edge usage
+// also drains the analytics meter into its app-day objects (spec §3.3). A
+// batch that fails to write is re-queued and lands on the next flush.
+func TestUsageFlushAlsoFlushesAnalytics(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ now := time.Date(2026, 9, 7, 9, 30, 0, 0, time.UTC)
+ am := analytics.NewMeter(t.TempDir(), "box1", func() time.Time { return now })
+ am.Observe(analytics.Obs{Acct: "bab", App: "web", Host: "h", Method: "GET", Status: 200, ContentType: "text/html", Path: "/", UA: "Mozilla/5.0 Safari", ClientIP: "1.2.3.4", Bytes: 10, When: now})
+ if err := usageFlush(ctx, s, s, "box1", func() map[string]usage.EdgeRow { return nil }, am, now); err != nil {
+ t.Fatal(err)
+ }
+ sum, err := analytics.ReadRange(ctx, s, "bab", "web", now, now)
+ if err != nil || sum.PV != 1 {
+ t.Fatalf("flushed day: %+v %v", sum, err)
+ }
+ // A failed analytics flush re-queues; the next flush lands it.
+ am.Observe(analytics.Obs{Acct: "bab", App: "web", Host: "h", Method: "GET", Status: 200, ContentType: "text/html", Path: "/", UA: "Mozilla/5.0 Safari", ClientIP: "1.2.3.4", Bytes: 10, When: now})
+ _ = usageFlush(ctx, s, failingStore{Store: s}, "box1", func() map[string]usage.EdgeRow { return nil }, am, now)
+ if err := usageFlush(ctx, s, s, "box1", func() map[string]usage.EdgeRow { return nil }, am, now); err != nil {
+ t.Fatal(err)
+ }
+ if sum, _ := analytics.ReadRange(ctx, s, "bab", "web", now, now); sum.PV != 2 {
+ t.Fatalf("after requeue: %d", sum.PV)
+ }
+}