| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "log" |
| 7 | "os" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/analytics" |
| 11 | "github.com/carlosframework/platform/internal/applog" |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | "github.com/carlosframework/platform/internal/usage" |
| 14 | ) |
| 15 | |
| 16 | // usageFlush drains the edge meter and CAS-merges the deltas into the box's |
| 17 | // usage/edge day object, then (when am is non-nil) drains the analytics |
| 18 | // meter's deltas into their app-day objects (spec §3.3: the same tick, a |
| 19 | // second CAS write per app). Usage writes go to s (rebased under |
| 20 | // CARLOS_STATUS_PREFIX on a fleet box); analytics writes go to raw — a day |
| 21 | // file lives under apps/<sqid>/<app>/analytics/, the app's own data, and |
| 22 | // must NOT be rebased (same reasoning as log shipping — logship.go:54-61). |
| 23 | // A failed analytics batch is re-queued for the next tick; a failed usage |
| 24 | // write is returned but does not block the analytics flush. |
| 25 | func usageFlush(ctx context.Context, s, raw release.Store, box string, drain func() map[string]usage.EdgeRow, am *analytics.Meter, now time.Time) error { |
| 26 | err := usage.UpsertEdgeDay(ctx, s, box, now, drain()) |
| 27 | if am == nil { |
| 28 | return err |
| 29 | } |
| 30 | failed := map[analytics.AppKey][]analytics.Delta{} |
| 31 | for k, ds := range am.Drain() { |
| 32 | for _, d := range ds { |
| 33 | uerr := analytics.UpsertDay(ctx, raw, k.Acct, k.App, d) |
| 34 | if errors.Is(uerr, applog.ErrBadKeySegment) { |
| 35 | // No key, so no retry can ever land it: drop, never re-queue. |
| 36 | unshippable(applog.FamilyAnalytics, k, uerr) |
| 37 | continue |
| 38 | } |
| 39 | if uerr != nil { |
| 40 | log.Printf("agent: analytics flush %s/%s %s: %v (re-queued)", k.Acct, k.App, d.Date, uerr) |
| 41 | failed[k] = append(failed[k], d) |
| 42 | } |
| 43 | } |
| 44 | } |
| 45 | if len(failed) > 0 { |
| 46 | am.Requeue(failed) |
| 47 | } |
| 48 | return err |
| 49 | } |
| 50 | |
| 51 | // usageLoop flushes edge usage every 5 minutes, forever — the metricsLoop |
| 52 | // conventions exactly: recover guard, disabled with a log line if the box |
| 53 | // has no status store (mirroring metricsLoop's errNoStatusStore vs. other-error |
| 54 | // distinction at metrics_loop.go:773-779, rather than the brief's single log |
| 55 | // line, so the two loops read the same way in the logs), errors logged and |
| 56 | // never fatal (the drained window is lost, the next one is not; modelling |
| 57 | // grade tolerates the gap). am is the analytics meter (nil disables the |
| 58 | // analytics half of the flush); its deltas are folded in on the same tick. |
| 59 | func usageLoop(drain func() map[string]usage.EdgeRow, am *analytics.Meter) { |
| 60 | defer func() { |
| 61 | if r := recover(); r != nil { |
| 62 | log.Printf("agent: usage loop panic, edge metering stopped for this process: %v", r) |
| 63 | } |
| 64 | }() |
| 65 | ctx := context.Background() |
| 66 | // Same selection rule statusLoop/metricsLoop use (agent.go's |
| 67 | // openStatusStore): a status/deployment bucket wins and keeps direct S3; |
| 68 | // a fleet box with neither set publishes through the edge's store door. |
| 69 | raw, target, err := openStatusStore(ctx, os.Getenv) |
| 70 | if err != nil { |
| 71 | if errors.Is(err, errNoStatusStore) { |
| 72 | log.Printf("agent: usage loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set") |
| 73 | } else { |
| 74 | log.Printf("agent: usage loop disabled: %v", err) |
| 75 | } |
| 76 | return |
| 77 | } |
| 78 | s := release.NewPrefixStore(raw, resolveStatusPrefix(os.Getenv)) |
| 79 | box := metricsBoxLabel(os.Getenv, readMachineID()) |
| 80 | log.Printf("agent: usage loop armed (5min flush), publishing to %s", target) |
| 81 | for { |
| 82 | time.Sleep(5 * time.Minute) |
| 83 | if err := usageFlush(ctx, s, raw, box, drain, am, time.Now()); err != nil { |
| 84 | log.Printf("agent: usage flush: %v", err) |
| 85 | } |
| 86 | } |
| 87 | } |
| 88 | |
amadan