| index 3f337bdb2..19a3db172 100644 |
| --- a/internal/applog/sink.go |
| +++ b/internal/applog/sink.go |
| @@ -1,7 +1,9 @@ |
| package applog |
| |
| import ( |
| + "cmp" |
| "fmt" |
| + "slices" |
| "strconv" |
| "strings" |
| "sync" |
| @@ -14,9 +16,9 @@ type AppKey struct{ Acct, App string } |
| |
| // Sink buffers tagged records between ship-loop flushes: a bounded ring per |
| // app plus a token-bucket rate cap per app, both with drop honesty — when |
| -// either binds, oldest lines go first and the next Drain carries one |
| -// synthetic platform record naming the count, never silent thinning |
| -// (spec §§4–5). Safe for concurrent use; producers (exec tees, event emit |
| +// either binds, oldest lines go first (edge records before any app or |
| +// platform line; see edgeRequest) and the next Drain carries one synthetic |
| +// record naming the count, never silent thinning (spec §§4–5). Safe for concurrent use; producers (exec tees, event emit |
| // sites) and the ship loop share one Sink per agent process. |
| type Sink struct { |
| // Budgets are fields, not constructor params, so tests and future |
| @@ -43,11 +45,17 @@ type Sink struct { |
| } |
| |
| type appBuf struct { |
| - recs []Record |
| - bytes int |
| - tokens float64 |
| - lastRef time.Time |
| - dropped int |
| + // recs holds app and platform records and edge holds edge records by |
| + // eviction class, each list in seq order; Drain merges them back into |
| + // one. Keeping them apart is what lets the ring shed edge records first |
| + // without a scan. |
| + recs []Record |
| + edge [edgeClasses][]Record |
| + edgeBytes [edgeClasses]int |
| + bytes int // every list's records, by costOf |
| + tokens float64 |
| + lastRef time.Time |
| + dropped int |
| |
| // Edge bucket state. edgeInit is false until the app's first edge |
| // record: the bucket is made lazily, starting full at whatever |
| @@ -58,6 +66,24 @@ type appBuf struct { |
| edgeDropped int |
| } |
| |
| +// Edge records are evicted from the ring in class order: request records |
| +// first, then blocks, then summaries. A record may only displace edge records |
| +// of its own class or a lower one, and never an app or platform record, so an |
| +// edge flood cannot evict the app's own lines (spec §7), a not-found or watch |
| +// flood cannot evict the record of a block, and nothing but other summaries |
| +// can evict the minute counts that say how much was sampled away. |
| +const ( |
| + edgeRequest = iota // not-found and watch records, and any other kind |
| + edgeBlock // block records (kind EdgeKindBlock) |
| + edgeSummary // minute counts (EdgeSummary) and the edge drop notice |
| + edgeClasses |
| +) |
| + |
| +// EdgeKindBlock is the kind field the edge's request recorder (internal/reqrec, |
| +// KindBlock) writes on a block record. applog imports nothing from reqrec, so |
| +// the word is repeated here; internal/host tests that the two agree. |
| +const EdgeKindBlock = "block" |
| + |
| // NewSink returns a Sink stamping records with box and timestamps from now |
| // (a seam so tests control the clock). The sequence counter starts at the |
| // construction instant's UnixMilli rather than zero: chunk keys embed the |
| @@ -120,7 +146,7 @@ func (s *Sink) App(acct, app, host, line string) { |
| Level: level, |
| Msg: msg, |
| Fields: fields, |
| - }) |
| + }, -1) |
| } |
| |
| // Edge buffers one edge request record for (acct, app). It runs on the |
| @@ -128,7 +154,9 @@ func (s *Sink) App(acct, app, host, line string) { |
| // nothing else: no I/O, no blocking send, and it does not poke notify |
| // (edge records ride the ship tick; only wake failures need the fast |
| // flush). The edge bucket is separate from App's, and the fields map is |
| -// copied because callers may reuse it. |
| +// copied because callers may reuse it. In the ring the record may displace |
| +// only edge records (see edgeRequest); with none to displace it is refused |
| +// and counted in the edge drop notice. |
| func (s *Sink) Edge(acct, app, host, level, msg string, fields map[string]string) { |
| r := Record{ |
| Host: host, |
| @@ -156,7 +184,29 @@ func (s *Sink) Edge(acct, app, host, level, msg string, fields map[string]string |
| return |
| } |
| b.edgeTokens -= cost |
| - s.append(b, r) |
| + class := edgeRequest |
| + if r.Fields["kind"] == EdgeKindBlock { |
| + class = edgeBlock |
| + } |
| + s.append(b, r, class) |
| +} |
| + |
| +// EdgeSummary buffers one edge summary record (the per-minute counts) for |
| +// (acct, app). It bypasses the edge rate bucket, as Event bypasses the app |
| +// one: the counts are what explain the sampling, and a flood that has spent |
| +// the edge budget is exactly when they matter. The ring still bounds it, and |
| +// it is the last edge record evicted. |
| +func (s *Sink) EdgeSummary(acct, app, host, level, msg string, fields map[string]string) { |
| + r := Record{ |
| + Host: host, |
| + Stream: StreamEdge, |
| + Level: level, |
| + Msg: msg, |
| + Fields: copyFields(fields), |
| + } |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + s.append(s.buf(AppKey{acct, app}), r, edgeSummary) |
| } |
| |
| func copyFields(f map[string]string) map[string]string { |
| @@ -182,7 +232,7 @@ func (s *Sink) Event(acct, app, host, level, msg string) { |
| Stream: StreamPlatform, |
| Level: level, |
| Msg: msg, |
| - }) |
| + }, -1) |
| s.mu.Unlock() |
| select { // non-blocking: a pending poke already says everything |
| case s.notify <- struct{}{}: |
| @@ -216,27 +266,55 @@ func (s *Sink) Drain() map[AppKey][]Record { |
| Stream: StreamPlatform, |
| Level: LevelWarn, |
| Msg: fmt.Sprintf("dropped %d app log line(s)"+dropNoticeSuffix, b.dropped), |
| - }) |
| + }, -1) |
| b.dropped = 0 |
| } |
| - if b.edgeDropped > 0 { |
| + if n := b.edgeDropped; n > 0 { |
| // Not IsDropNotice: this is an edge-stream record about the |
| // edge budget, so Availability (platform stream) never sees it. |
| - s.append(b, Record{ |
| + // It is a summary for eviction. Records it evicts to fit are |
| + // counted afresh, for the next drain's notice; if it cannot fit |
| + // at all (the ring holds only app lines), the count waits. |
| + b.edgeDropped = 0 |
| + if !s.append(b, Record{ |
| Stream: StreamEdge, |
| Level: LevelWarn, |
| - Msg: fmt.Sprintf("dropped %d edge record(s)"+dropNoticeSuffix, b.edgeDropped), |
| - Fields: map[string]string{"kind": "dropped", "n": strconv.Itoa(b.edgeDropped)}, |
| - }) |
| - b.edgeDropped = 0 |
| + Msg: fmt.Sprintf("dropped %d edge record(s)"+dropNoticeSuffix, n), |
| + Fields: map[string]string{"kind": "dropped", "n": strconv.Itoa(n)}, |
| + }, edgeSummary) { |
| + b.edgeDropped = n |
| + } |
| } |
| - if len(b.recs) == 0 { |
| + recs := b.drainAll() |
| + if len(recs) == 0 { |
| continue |
| } |
| - out[k] = b.recs |
| - b.recs = nil |
| - b.bytes = 0 |
| + out[k] = recs |
| + } |
| + return out |
| +} |
| + |
| +// drainAll empties the buffer and returns its records in seq order. Callers |
| +// hold s.mu. |
| +func (b *appBuf) drainAll() []Record { |
| + n := len(b.recs) |
| + for _, l := range b.edge { |
| + n += len(l) |
| } |
| + if n == 0 { |
| + return nil |
| + } |
| + out := make([]Record, 0, n) |
| + out = append(out, b.recs...) |
| + for _, l := range b.edge { |
| + out = append(out, l...) |
| + } |
| + // Each list is already in seq order; one sort merges them. |
| + slices.SortFunc(out, func(x, y Record) int { return cmp.Compare(x.Seq, y.Seq) }) |
| + b.recs = nil |
| + b.edge = [edgeClasses][]Record{} |
| + b.edgeBytes = [edgeClasses]int{} |
| + b.bytes = 0 |
| return out |
| } |
| |
| @@ -250,27 +328,77 @@ func (s *Sink) buf(k AppKey) *appBuf { |
| return b |
| } |
| |
| -// append stamps and stores one record, evicting oldest-first past the ring |
| -// cap. Callers hold s.mu. |
| -func (s *Sink) append(b *appBuf, r Record) { |
| +// append stamps and stores one record, evicting past the ring cap, and |
| +// reports whether it was stored. class is the edge eviction class of an edge |
| +// record, or -1 for an app or platform record. |
| +// |
| +// An edge record may displace only edge records of its class or a lower one, |
| +// oldest first; if even all of those would not make room it is refused |
| +// (counted in edgeDropped) and nothing is evicted. An app or platform record |
| +// displaces edge records first, lowest class first, and only then the oldest |
| +// app or platform record, never itself: a single oversized line still lands, |
| +// as before. Callers hold s.mu. |
| +func (s *Sink) append(b *appBuf, r Record, class int) bool { |
| + cost := costOf(r) |
| + if class >= 0 { |
| + room := s.MaxBytesPerApp - b.bytes |
| + for c := 0; c <= class; c++ { |
| + room += b.edgeBytes[c] |
| + } |
| + if cost > room { |
| + b.edgeDropped++ |
| + return false |
| + } |
| + } |
| s.seq++ |
| r.TS = s.now().UTC() |
| r.Box = s.box |
| r.Seq = s.seq |
| - b.recs = append(b.recs, r) |
| - b.bytes += costOf(r) |
| - for b.bytes > s.MaxBytesPerApp && len(b.recs) > 1 { |
| - old := b.recs[0] |
| - b.bytes -= costOf(old) |
| - b.recs = b.recs[1:] |
| - // An evicted edge record is owed to the edge notice, not the app |
| - // one: "dropped N app log line(s)" would be untrue. |
| - if old.Stream == StreamEdge { |
| + if class >= 0 { |
| + b.edge[class] = append(b.edge[class], r) |
| + b.edgeBytes[class] += cost |
| + } else { |
| + b.recs = append(b.recs, r) |
| + } |
| + b.bytes += cost |
| + for b.bytes > s.MaxBytesPerApp { |
| + if !b.evictOne(class) { |
| + break |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// evictOne drops the oldest record the incoming record of class may displace |
| +// (see append) and reports whether there was one. The incoming record is the |
| +// newest in its list and is never the one dropped. Callers hold s.mu. |
| +func (b *appBuf) evictOne(class int) bool { |
| + top := class |
| + if class < 0 { |
| + top = edgeClasses - 1 |
| + } |
| + for c := 0; c <= top; c++ { |
| + keep := 0 |
| + if c == class { |
| + keep = 1 // the incoming record itself |
| + } |
| + if len(b.edge[c]) > keep { |
| + old := b.edge[c][0] |
| + b.edge[c] = b.edge[c][1:] |
| + b.edgeBytes[c] -= costOf(old) |
| + b.bytes -= costOf(old) |
| b.edgeDropped++ |
| - } else { |
| - b.dropped++ |
| + return true |
| } |
| } |
| + if class < 0 && len(b.recs) > 1 { |
| + old := b.recs[0] |
| + b.recs = b.recs[1:] |
| + b.bytes -= costOf(old) |
| + b.dropped++ |
| + return true |
| + } |
| + return false |
| } |
| |
| // refill advances the token bucket to now. Callers hold s.mu. |
amadan