| 1 | // The log shipping loop — the box-side pump of `carlos logs` (spec |
| 2 | // 2026-08-10-carlos-logs-design.md §5). Producers (the activator's exec |
| 3 | // tees and platform events) buffer into one applog.Sink; every 10s — or |
| 4 | // within ~2s of a platform event, which is what lets `carlos deploy`'s |
| 5 | // timeout window see the wake error — this loop drains it and writes one |
| 6 | // immutable gzipped chunk per app into the deployment bucket. On each hour |
| 7 | // boundary it prunes ITS OWN box subtree per app: hours beyond the age |
| 8 | // floor, then oldest-first beyond the byte cap (sizes ride in the chunk |
| 9 | // keys — applog.ChunkKeySize — so no Store interface change). The |
| 10 | // sampler/flush split mirrors metricsLoop for the same reason: the loop |
| 11 | // shares a process with the box's traffic, so it recovers panics and |
| 12 | // disables itself without a bucket rather than taking the box down. |
| 13 | package host |
| 14 | |
| 15 | import ( |
| 16 | "context" |
| 17 | "errors" |
| 18 | "log" |
| 19 | "os" |
| 20 | "sync" |
| 21 | "time" |
| 22 | |
| 23 | "github.com/carlosframework/platform/internal/applog" |
| 24 | "github.com/carlosframework/platform/internal/release" |
| 25 | ) |
| 26 | |
| 27 | const ( |
| 28 | logshipTick = 10 * time.Second |
| 29 | logshipDebounce = 2 * time.Second // event-poke settle: batch a wake's burst into one PUT |
| 30 | logshipMaxAge = 7 * 24 * time.Hour |
| 31 | logshipMaxBytes = 50 << 20 // per app per box |
| 32 | // logshipMaxPending caps records held across failed flushes, per app. |
| 33 | // The bucket being unwritable must not grow the agent's heap forever |
| 34 | // (samplerMaxBuf's reasoning); oldest go first, silently — a box-wide |
| 35 | // bucket outage is loudly visible elsewhere, and the gap in seq numbers |
| 36 | // records that something is missing. |
| 37 | logshipMaxPending = 4096 |
| 38 | ) |
| 39 | |
| 40 | func logshipLoop(sink *applog.Sink, box string) { |
| 41 | // Same rule as metricsLoop: this loop shares a process with the agent |
| 42 | // serving traffic — log shipping stopping is a bad day, the box |
| 43 | // stopping is an outage. |
| 44 | defer func() { |
| 45 | if r := recover(); r != nil { |
| 46 | log.Printf("agent: logship loop panic, log shipping stopped for this process: %v", r) |
| 47 | } |
| 48 | }() |
| 49 | |
| 50 | ctx := context.Background() |
| 51 | // Same store selection as statusLoop/metricsLoop (agent.go's |
| 52 | // openStatusStore): a bucket wins, and a cut-over fleet box with no bucket |
| 53 | // ships through the edge's store door instead of going silent. |
| 54 | // |
| 55 | // The keys stay UNPREFIXED — deliberately unlike status and metrics, which |
| 56 | // rebase under CARLOS_STATUS_PREFIX. A log chunk is the app's own data |
| 57 | // (applog.Prefix = release.AppPrefix + "logs/"), and the console reads it |
| 58 | // straight from the platform layout (applog.ReadWindow over |
| 59 | // apps/<sqid>/<app>/logs/**); rebasing under fleets/<name>/ would file a |
| 60 | // fleet box's logs somewhere `carlos logs` never looks. The door's write |
| 61 | // ACL admits exactly this shape for the box's own account (aclWrite's |
| 62 | // segment-exact apps/<sqid>/<app>/logs/** scope). |
| 63 | s, target, err := openStatusStore(ctx, os.Getenv) |
| 64 | if err != nil { |
| 65 | if errors.Is(err, errNoStatusStore) { |
| 66 | log.Printf("agent: logship loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set") |
| 67 | } else { |
| 68 | log.Printf("agent: logship loop disabled: %v", err) |
| 69 | } |
| 70 | return |
| 71 | } |
| 72 | log.Printf("agent: logship loop armed (10s ticks, event fast-flush), publishing to %s", target) |
| 73 | |
| 74 | pending := make(map[applog.AppKey][]applog.Record) |
| 75 | pruned := "" // last hour a prune ran for, "" = not yet |
| 76 | ticker := time.NewTicker(logshipTick) |
| 77 | defer ticker.Stop() |
| 78 | for { |
| 79 | select { |
| 80 | case <-ticker.C: |
| 81 | case <-sink.Wake(): |
| 82 | time.Sleep(logshipDebounce) // let the event's burst finish |
| 83 | } |
| 84 | logshipGather(sink, pending) |
| 85 | logshipFlush(ctx, s, pending) |
| 86 | |
| 87 | if hour := applog.HourOf(time.Now()); hour != pruned { |
| 88 | // Prune covers every app this box is CURRENTLY holding or has |
| 89 | // just flushed; an app that stops logging entirely ages out via |
| 90 | // other boxes' passes or the app's own deletion sweep. Sub-cap |
| 91 | // apps pay one ListDirs each per hour. |
| 92 | for _, k := range logshipKnownApps(pending, sink) { |
| 93 | chunkPrune(ctx, s, applog.FamilyLogs, box, k, time.Now()) |
| 94 | } |
| 95 | pruned = hour |
| 96 | } |
| 97 | } |
| 98 | } |
| 99 | |
| 100 | // logshipGather drains the sink into pending, enforcing the per-app cap — |
| 101 | // oldest first, so what survives a long bucket outage is the freshest tail. |
| 102 | func logshipGather(sink *applog.Sink, pending map[applog.AppKey][]applog.Record) { |
| 103 | for k, recs := range sink.Drain() { |
| 104 | merged := append(pending[k], recs...) |
| 105 | if over := len(merged) - logshipMaxPending; over > 0 { |
| 106 | merged = merged[over:] |
| 107 | } |
| 108 | pending[k] = merged |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | // logshipFlush writes one chunk per pending app — create-only, keyed by the |
| 113 | // batch's first seq, so a retried batch can never overwrite a delivered one |
| 114 | // — and forgets what landed. A failed app keeps its records for the next |
| 115 | // round; other apps' flushes proceed regardless. |
| 116 | func logshipFlush(ctx context.Context, s release.Store, pending map[applog.AppKey][]applog.Record) { |
| 117 | for k, recs := range pending { |
| 118 | if len(recs) == 0 { |
| 119 | delete(pending, k) |
| 120 | continue |
| 121 | } |
| 122 | body, err := applog.EncodeChunk(recs) |
| 123 | if err != nil { // marshal of our own records — a bug, not a retry case |
| 124 | log.Printf("agent: logship encode %s/%s: %v (dropping %d records)", k.Acct, k.App, err, len(recs)) |
| 125 | delete(pending, k) |
| 126 | continue |
| 127 | } |
| 128 | key, err := applog.ChunkKey(k.Acct, k.App, recs[0].Box, applog.HourOf(recs[0].TS), recs[0].Seq, len(body)) |
| 129 | if err != nil { |
| 130 | unshippable(applog.FamilyLogs, k, err) |
| 131 | delete(pending, k) |
| 132 | continue |
| 133 | } |
| 134 | if _, err := s.Put(ctx, key, body, ""); err != nil { |
| 135 | log.Printf("agent: logship flush %s/%s: %v (%d records held)", k.Acct, k.App, err, len(recs)) |
| 136 | continue |
| 137 | } |
| 138 | delete(pending, k) |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | // logshipKnownApps is the union of apps with pending records and apps the |
| 143 | // sink has ever buffered for — the prune worklist. |
| 144 | func logshipKnownApps(pending map[applog.AppKey][]applog.Record, sink *applog.Sink) []applog.AppKey { |
| 145 | seen := make(map[applog.AppKey]bool) |
| 146 | var out []applog.AppKey |
| 147 | for k := range pending { |
| 148 | if !seen[k] { |
| 149 | seen[k] = true |
| 150 | out = append(out, k) |
| 151 | } |
| 152 | } |
| 153 | for _, k := range sink.Known() { |
| 154 | if !seen[k] { |
| 155 | seen[k] = true |
| 156 | out = append(out, k) |
| 157 | } |
| 158 | } |
| 159 | return out |
| 160 | } |
| 161 | |
| 162 | // chunkPrune enforces retention for ONE app's ONE family on THIS box's |
| 163 | // subtree only — each box owns what it wrote (spec §5). Hour granularity: an |
| 164 | // hour goes when it is older than the age floor, or (oldest first) while the |
| 165 | // subtree's key-embedded sizes total over the byte cap. Errors are logged |
| 166 | // and left for the next hourly pass — retention is idempotent. |
| 167 | func chunkPrune(ctx context.Context, s release.Store, fam applog.Family, box string, k applog.AppKey, now time.Time) { |
| 168 | prefix, err := applog.FamilyBoxPrefix(fam, k.Acct, k.App, box) |
| 169 | if err != nil { |
| 170 | unshippable(fam, k, err) |
| 171 | return |
| 172 | } |
| 173 | hourDirs, err := s.ListDirs(ctx, prefix) |
| 174 | if err != nil { |
| 175 | log.Printf("agent: logship prune %s: %v", prefix, err) |
| 176 | return |
| 177 | } |
| 178 | // ListDirs returns full prefixes ending in "/", sorted is not promised — |
| 179 | // the hour names sort chronologically, so order them ourselves via the |
| 180 | // keys' lexical shape. Collect per-hour keys+sizes as we go. |
| 181 | type hourEnt struct { |
| 182 | dir string |
| 183 | keys []string |
| 184 | bytes int64 |
| 185 | } |
| 186 | var hours []hourEnt |
| 187 | var total int64 |
| 188 | for _, dir := range hourDirs { |
| 189 | keys, err := s.List(ctx, dir) |
| 190 | if err != nil { |
| 191 | log.Printf("agent: logship prune %s: %v", dir, err) |
| 192 | return |
| 193 | } |
| 194 | var b int64 |
| 195 | for _, key := range keys { |
| 196 | if n, ok := applog.ChunkKeySize(key); ok { |
| 197 | b += n |
| 198 | } |
| 199 | } |
| 200 | hours = append(hours, hourEnt{dir: dir, keys: keys, bytes: b}) |
| 201 | total += b |
| 202 | } |
| 203 | // hourDirs from both implementations come lexically sorted (S3 lists in |
| 204 | // key order; DirStore sorts), and hour names are zero-padded UTC — so |
| 205 | // hours[0] is oldest. |
| 206 | cutoff := now.Add(-logshipMaxAge) |
| 207 | for _, h := range hours { |
| 208 | hourName := applog.HourNameOfDir(h.dir) |
| 209 | drop := applog.HourBefore(hourName, cutoff) || total > logshipMaxBytes |
| 210 | if !drop { |
| 211 | break // hours are oldest-first; nothing further is droppable |
| 212 | } |
| 213 | for _, key := range h.keys { |
| 214 | if err := s.Delete(ctx, key); err != nil { |
| 215 | log.Printf("agent: logship prune %s: %v", key, err) |
| 216 | return |
| 217 | } |
| 218 | } |
| 219 | total -= h.bytes |
| 220 | } |
| 221 | } |
| 222 | |
| 223 | // unshippableSeen remembers which (family, app) pairs unshippable has |
| 224 | // already reported, so a route that can never ship says so once per |
| 225 | // process rather than every tick. Bounded by the routes a box serves. |
| 226 | var unshippableSeen sync.Map |
| 227 | |
| 228 | type unshippableKey struct { |
| 229 | fam applog.Family |
| 230 | app applog.AppKey |
| 231 | } |
| 232 | |
| 233 | // unshippable reports, once per family and app, that the family's key |
| 234 | // builder refused the app (applog.ErrBadKeySegment): the records are |
| 235 | // dropped, and no write or prune happens for it. The case that put this |
| 236 | // here is a route with no app, filed under its account with an empty app |
| 237 | // name; writing it would make apps/<sqid>//..., which hostops.Apps refuses |
| 238 | // to walk. |
| 239 | func unshippable(fam applog.Family, k applog.AppKey, err error) { |
| 240 | if _, dup := unshippableSeen.LoadOrStore(unshippableKey{fam, k}, true); dup { |
| 241 | return |
| 242 | } |
| 243 | log.Printf("agent: %s: not shipping for account %q app %q, dropping its records (logged once): %v", fam, k.Acct, k.App, err) |
| 244 | } |
| 245 | |
amadan