| 1 | package applog |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "strings" |
| 6 | "sync" |
| 7 | "time" |
| 8 | ) |
| 9 | |
| 10 | // AppKey names the (account, app) a buffered record belongs to — the unit |
| 11 | // both budgets and bucket keys are scoped by. |
| 12 | type AppKey struct{ Acct, App string } |
| 13 | |
| 14 | // Sink buffers tagged records between ship-loop flushes: a bounded ring per |
| 15 | // app plus a token-bucket rate cap per app, both with drop honesty — when |
| 16 | // either binds, oldest lines go first and the next Drain carries one |
| 17 | // synthetic platform record naming the count, never silent thinning |
| 18 | // (spec §§4–5). Safe for concurrent use; producers (exec tees, event emit |
| 19 | // sites) and the ship loop share one Sink per agent process. |
| 20 | type Sink struct { |
| 21 | // Budgets are fields, not constructor params, so tests and future |
| 22 | // per-app overrides set them directly; the zero Sink is unusable — |
| 23 | // NewSink applies the spec defaults. |
| 24 | MaxBytesPerApp int // ring cap per app (default 4 MiB) |
| 25 | RatePerMin float64 // sustained app-line budget per app (default 1 MiB/min) |
| 26 | Burst float64 // token bucket depth (default 4 MiB) |
| 27 | |
| 28 | box string |
| 29 | now func() time.Time |
| 30 | |
| 31 | mu sync.Mutex |
| 32 | apps map[AppKey]*appBuf |
| 33 | seq uint64 // monotonic across apps AND restarts — the follow cursor's spine |
| 34 | |
| 35 | notify chan struct{} // poked (non-blocking) on Event; the ship loop's fast-flush signal |
| 36 | } |
| 37 | |
| 38 | type appBuf struct { |
| 39 | recs []Record |
| 40 | bytes int |
| 41 | tokens float64 |
| 42 | lastRef time.Time |
| 43 | dropped int |
| 44 | } |
| 45 | |
| 46 | // NewSink returns a Sink stamping records with box and timestamps from now |
| 47 | // (a seam so tests control the clock). The sequence counter starts at the |
| 48 | // construction instant's UnixMilli rather than zero: chunk keys embed the |
| 49 | // first record's seq, and per-box seq order is a promise the cursor relies |
| 50 | // on — a restart that re-based to zero would write keys that sort BEFORE |
| 51 | // last boot's and break exact per-box ordering. Milliseconds always outrun |
| 52 | // the admitted line rate (the rate cap is ~17 KiB/s), so a rebase lands |
| 53 | // above everything the previous process numbered. |
| 54 | func NewSink(box string, now func() time.Time) *Sink { |
| 55 | return &Sink{ |
| 56 | MaxBytesPerApp: 4 << 20, |
| 57 | RatePerMin: 1 << 20, |
| 58 | Burst: 4 << 20, |
| 59 | box: box, |
| 60 | now: now, |
| 61 | apps: make(map[AppKey]*appBuf), |
| 62 | seq: uint64(now().UnixMilli()), |
| 63 | notify: make(chan struct{}, 1), |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | // Wake is the ship loop's fast-flush signal: it receives after an Event has |
| 68 | // been buffered, so a wake failure reaches the bucket in seconds rather |
| 69 | // than at the next tick (spec §5). Level-triggered and lossy by design — |
| 70 | // one pending poke summarizes any number of events. |
| 71 | func (s *Sink) Wake() <-chan struct{} { return s.notify } |
| 72 | |
| 73 | // Known lists every app this sink has ever buffered for, in no particular |
| 74 | // order — the ship loop's retention worklist. |
| 75 | func (s *Sink) Known() []AppKey { |
| 76 | s.mu.Lock() |
| 77 | defer s.mu.Unlock() |
| 78 | out := make([]AppKey, 0, len(s.apps)) |
| 79 | for k := range s.apps { |
| 80 | out = append(out, k) |
| 81 | } |
| 82 | return out |
| 83 | } |
| 84 | |
| 85 | // App buffers one line of app output for (acct, app), parsing structured |
| 86 | // JSON lines at the door (ParseAppLine) so the bucket holds fields and |
| 87 | // honored levels rather than opaque text. Subject to both budgets. |
| 88 | func (s *Sink) App(acct, app, host, line string) { |
| 89 | msg, level, fields := ParseAppLine(line) |
| 90 | s.mu.Lock() |
| 91 | defer s.mu.Unlock() |
| 92 | b := s.buf(AppKey{acct, app}) |
| 93 | cost := float64(recCost(msg)) |
| 94 | b.refill(s.now(), s.RatePerMin, s.Burst) |
| 95 | if b.tokens < cost { |
| 96 | b.dropped++ |
| 97 | return |
| 98 | } |
| 99 | b.tokens -= cost |
| 100 | s.append(b, Record{ |
| 101 | Host: host, |
| 102 | Stream: StreamApp, |
| 103 | Level: level, |
| 104 | Msg: msg, |
| 105 | Fields: fields, |
| 106 | }) |
| 107 | } |
| 108 | |
| 109 | // Event buffers one platform event — member-safe text composed at the emit |
| 110 | // site (spec §5), never a relayed journal line. Events bypass the rate cap: |
| 111 | // they are the diagnosis, and a chatty app must not crowd out its own |
| 112 | // wake-failure line. The ring cap still applies (a platform-event storm is |
| 113 | // a bug, not a budget). |
| 114 | func (s *Sink) Event(acct, app, host, level, msg string) { |
| 115 | s.mu.Lock() |
| 116 | s.append(s.buf(AppKey{acct, app}), Record{ |
| 117 | Host: host, |
| 118 | Stream: StreamPlatform, |
| 119 | Level: level, |
| 120 | Msg: msg, |
| 121 | }) |
| 122 | s.mu.Unlock() |
| 123 | select { // non-blocking: a pending poke already says everything |
| 124 | case s.notify <- struct{}{}: |
| 125 | default: |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | // dropNoticeSuffix ends the synthetic line Drain appends for an app whose |
| 130 | // lines hit the rate cap. |
| 131 | const dropNoticeSuffix = " (rate cap)" |
| 132 | |
| 133 | // IsDropNotice reports whether r is that synthetic line: a platform event |
| 134 | // about the logging budget, not about the instance, so Availability (which |
| 135 | // takes the rest of the platform stream) leaves it out. |
| 136 | func IsDropNotice(r Record) bool { |
| 137 | return r.Stream == StreamPlatform && strings.HasPrefix(r.Msg, "dropped ") && strings.HasSuffix(r.Msg, dropNoticeSuffix) |
| 138 | } |
| 139 | |
| 140 | // Drain returns everything buffered, keyed by app, and resets the buffers |
| 141 | // (sequence numbers and rate-bucket state persist — they are the process's, |
| 142 | // not the batch's). An app whose lines were dropped since the last drain |
| 143 | // gets one trailing synthetic record naming the count. Empty apps are |
| 144 | // omitted; an empty sink drains to an empty map. |
| 145 | func (s *Sink) Drain() map[AppKey][]Record { |
| 146 | s.mu.Lock() |
| 147 | defer s.mu.Unlock() |
| 148 | out := make(map[AppKey][]Record) |
| 149 | for k, b := range s.apps { |
| 150 | if b.dropped > 0 { |
| 151 | s.append(b, Record{ |
| 152 | Stream: StreamPlatform, |
| 153 | Level: LevelWarn, |
| 154 | Msg: fmt.Sprintf("dropped %d app log line(s)"+dropNoticeSuffix, b.dropped), |
| 155 | }) |
| 156 | b.dropped = 0 |
| 157 | } |
| 158 | if len(b.recs) == 0 { |
| 159 | continue |
| 160 | } |
| 161 | out[k] = b.recs |
| 162 | b.recs = nil |
| 163 | b.bytes = 0 |
| 164 | } |
| 165 | return out |
| 166 | } |
| 167 | |
| 168 | // buf returns (creating if needed) the app's buffer. Callers hold s.mu. |
| 169 | func (s *Sink) buf(k AppKey) *appBuf { |
| 170 | b := s.apps[k] |
| 171 | if b == nil { |
| 172 | b = &appBuf{tokens: s.Burst, lastRef: s.now()} |
| 173 | s.apps[k] = b |
| 174 | } |
| 175 | return b |
| 176 | } |
| 177 | |
| 178 | // append stamps and stores one record, evicting oldest-first past the ring |
| 179 | // cap. Callers hold s.mu. |
| 180 | func (s *Sink) append(b *appBuf, r Record) { |
| 181 | s.seq++ |
| 182 | r.TS = s.now().UTC() |
| 183 | r.Box = s.box |
| 184 | r.Seq = s.seq |
| 185 | b.recs = append(b.recs, r) |
| 186 | b.bytes += recCost(r.Msg) |
| 187 | for b.bytes > s.MaxBytesPerApp && len(b.recs) > 1 { |
| 188 | b.bytes -= recCost(b.recs[0].Msg) |
| 189 | b.recs = b.recs[1:] |
| 190 | b.dropped++ |
| 191 | } |
| 192 | } |
| 193 | |
| 194 | // refill advances the token bucket to now. Callers hold s.mu. |
| 195 | func (b *appBuf) refill(now time.Time, perMin, burst float64) { |
| 196 | elapsed := now.Sub(b.lastRef) |
| 197 | b.lastRef = now |
| 198 | b.tokens += perMin * elapsed.Minutes() |
| 199 | if b.tokens > burst { |
| 200 | b.tokens = burst |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | // recCost is the budget cost of one record: its message plus a flat |
| 205 | // overhead standing in for the envelope. Both budgets use the same measure |
| 206 | // so "4 MiB of ring" and "1 MiB/min of rate" describe the same bytes. |
| 207 | func recCost(msg string) int { return len(msg) + 64 } |
| 208 | |
amadan