amadan

carlos / platform Public

host: usage flush loop rides the agent, 5min cadence

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 6441e1a334c258238fa989fdd58a4ba9f56e309b parent ba1d922
3 files changed, +106 −0
  • internal/host/agent.go +3 −0
  • internal/host/usage_loop.go +55 −0
  • internal/host/usage_loop_test.go +48 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 23fb23f62..09b9aaf10 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -315,6 +315,9 @@ func RunAgent(args []string) {
// calling it directly would block the front door forever.
edge.RunWithActivatorReady(args, act, cache, func(e *edge.Edge) {
go tunnelLoop(e)
+ m := edge.NewUsageMeter()
+ e.EnableUsage(m)
+ go usageLoop(m.Drain)
})
// The edge has drained; take the children down too (they're this process's
diff --git a/internal/host/usage_loop.go b/internal/host/usage_loop.go
new file mode 100644
index 000000000..b3af91669
--- /dev/null
+++ b/internal/host/usage_loop.go
@@ -0,0 +1,55 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "log"
+ "os"
+ "time"
+
+ "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())
+}
+
+// usageLoop flushes edge usage every 5 minutes, forever — the metricsLoop
+// conventions exactly: recover guard, disabled with a log line if the box
+// has no status store (mirroring metricsLoop's errNoStatusStore vs. other-error
+// 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) {
+ defer func() {
+ if r := recover(); r != nil {
+ log.Printf("agent: usage loop panic, edge metering stopped for this process: %v", r)
+ }
+ }()
+ ctx := context.Background()
+ // Same selection rule statusLoop/metricsLoop use (agent.go's
+ // openStatusStore): a status/deployment bucket wins and keeps direct S3;
+ // a fleet box with neither set publishes through the edge's store door.
+ raw, target, err := openStatusStore(ctx, os.Getenv)
+ if err != nil {
+ if errors.Is(err, errNoStatusStore) {
+ log.Printf("agent: usage loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
+ } else {
+ log.Printf("agent: usage loop disabled: %v", err)
+ }
+ return
+ }
+ s := release.NewPrefixStore(raw, resolveStatusPrefix(os.Getenv))
+ box := metricsBoxLabel(os.Getenv, readMachineID())
+ 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 {
+ log.Printf("agent: usage flush: %v", err)
+ }
+ }
+}
diff --git a/internal/host/usage_loop_test.go b/internal/host/usage_loop_test.go
new file mode 100644
index 000000000..9208f6f64
--- /dev/null
+++ b/internal/host/usage_loop_test.go
@@ -0,0 +1,48 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/usage"
+)
+
+func TestUsageFlushWritesAndReflushesMerge(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+ at := time.Date(2026, 8, 11, 9, 0, 0, 0, time.UTC)
+ 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 {
+ t.Fatal(err)
+ }
+ if err := usageFlush(ctx, s, "box1", func() map[string]usage.EdgeRow { return rows }, at.Add(5*time.Minute)); err != nil {
+ t.Fatal(err)
+ }
+ raw, _, err := s.Get(ctx, usage.EdgeDayKey("box1", "2026-08-11"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var day usage.EdgeDay
+ if err := json.Unmarshal(raw, &day); err != nil {
+ t.Fatal(err)
+ }
+ if got := day.Hours["09"]["a.example.com"].Requests; got != 6 {
+ t.Fatalf("requests = %d, want 6", got)
+ }
+}
+
+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())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if keys, _ := s.List(context.Background(), "usage/"); len(keys) != 0 {
+ t.Fatalf("wrote %v", keys)
+ }
+}