| 1 | package apperr |
| 2 | |
| 3 | import ( |
| 4 | "sync" |
| 5 | "time" |
| 6 | |
| 7 | "github.com/carlosframework/platform/internal/applog" |
| 8 | ) |
| 9 | |
| 10 | // MaxGroupsPerAppHour caps coalesced records per app per UTC hour, counting |
| 11 | // distinct (fp, host, release) combinations admitted within the hour, across |
| 12 | // every drain (spec §3.2) — not admissions since the last drain. A key once |
| 13 | // admitted this hour stays admitted; only genuinely new keys count against |
| 14 | // the cap. Per app, not per box: one tenant's storm must not suppress a |
| 15 | // neighbour's. |
| 16 | const MaxGroupsPerAppHour = 1000 |
| 17 | |
| 18 | // OverflowFP names the one row the excess beyond the cap folds into. Its |
| 19 | // Count is occurrences, and the tab labels it as such. |
| 20 | const OverflowFP = "_overflow" |
| 21 | |
| 22 | // Sink coalesces errors between ship-loop flushes: per app, one record per |
| 23 | // (fp, host, release) carrying count/first/last and the first sample. An |
| 24 | // error storm is the moment this must work — ten thousand identical panics |
| 25 | // in a minute become one record. Safe for concurrent use; the edge, the |
| 26 | // activator and the ship loop share one Sink per agent process. |
| 27 | type Sink struct { |
| 28 | box string |
| 29 | now func() time.Time |
| 30 | |
| 31 | mu sync.Mutex |
| 32 | apps map[applog.AppKey]*appBuf |
| 33 | seq uint64 |
| 34 | } |
| 35 | |
| 36 | type groupKey struct{ fp, host, release string } |
| 37 | |
| 38 | type appBuf struct { |
| 39 | groups map[groupKey]*Record |
| 40 | order []groupKey // insertion order, for a deterministic drain |
| 41 | overflow [2]*Record // one per family, indexed by IsAvailability: Errors' excess never lands on Availability's row, nor the reverse |
| 42 | hour string // the UTC hour `seen` counts for |
| 43 | seen map[groupKey]bool // keys admitted this hour, across every drain — the cap counter; Drain must not touch it |
| 44 | } |
| 45 | |
| 46 | // NewSink returns a Sink stamping records with box and timestamps from now. |
| 47 | // seq starts at UnixMilli for the reason applog.NewSink gives: chunk keys |
| 48 | // embed the first record's seq, and a restart must sort after last boot. |
| 49 | func NewSink(box string, now func() time.Time) *Sink { |
| 50 | return &Sink{box: box, now: now, apps: make(map[applog.AppKey]*appBuf), seq: uint64(now().UnixMilli())} |
| 51 | } |
| 52 | |
| 53 | // Add buffers one occurrence for (acct, app). r is sanitised here (§3.1), |
| 54 | // fingerprinted when it carries no FP, and folded into its group. |
| 55 | func (s *Sink) Add(acct, app string, r Record) { |
| 56 | now := s.now().UTC() |
| 57 | r.Name = Sanitize(r.Name) |
| 58 | r.Path = SanitizePath(r.Path) |
| 59 | if r.Path == "/" && r.Status == 0 && r.Source != SourceEdge { |
| 60 | r.Path = "" |
| 61 | } |
| 62 | if len(r.Stack) > MaxStack { |
| 63 | r.Stack = r.Stack[:MaxStack] |
| 64 | } |
| 65 | r.Stack = Sanitize(r.Stack) |
| 66 | if len(r.Fields) > 0 { |
| 67 | f := make(map[string]string, len(r.Fields)) |
| 68 | for k, v := range r.Fields { |
| 69 | f[k] = Sanitize(v) |
| 70 | } |
| 71 | r.Fields = f |
| 72 | } |
| 73 | if r.FP == "" { |
| 74 | r.FP = Fingerprint(r.Source, r.Kind, r.Name, r.Stack) |
| 75 | r.FPV = FPVersion |
| 76 | } |
| 77 | if len(r.FP) > 64 { |
| 78 | r.FP = r.FP[:64] |
| 79 | } |
| 80 | |
| 81 | s.mu.Lock() |
| 82 | defer s.mu.Unlock() |
| 83 | b := s.apps[applog.AppKey{Acct: acct, App: app}] |
| 84 | if b == nil { |
| 85 | b = &appBuf{groups: make(map[groupKey]*Record)} |
| 86 | s.apps[applog.AppKey{Acct: acct, App: app}] = b |
| 87 | } |
| 88 | if h := applog.HourOf(now); h != b.hour { |
| 89 | b.hour, b.seen = h, make(map[groupKey]bool) |
| 90 | } |
| 91 | k := groupKey{r.FP, r.Host, r.Release} |
| 92 | if g := b.groups[k]; g != nil { |
| 93 | g.Count++ |
| 94 | g.Last = now |
| 95 | return |
| 96 | } |
| 97 | if !b.seen[k] && len(b.seen) >= MaxGroupsPerAppHour { |
| 98 | fam := 0 |
| 99 | if IsAvailability(r) { |
| 100 | fam = 1 |
| 101 | } |
| 102 | if b.overflow[fam] == nil { |
| 103 | o := r |
| 104 | o.FP = OverflowFP |
| 105 | o.TS, o.First, o.Last, o.Count = now, now, now, 0 |
| 106 | o.Box = s.box |
| 107 | s.seq++ |
| 108 | o.Seq = s.seq |
| 109 | b.overflow[fam] = &o |
| 110 | } |
| 111 | b.overflow[fam].Count++ |
| 112 | b.overflow[fam].Last = now |
| 113 | return |
| 114 | } |
| 115 | b.seen[k] = true |
| 116 | s.seq++ |
| 117 | r.TS, r.First, r.Last, r.Count = now, now, now, 1 |
| 118 | r.Box = s.box |
| 119 | r.Seq = s.seq |
| 120 | b.groups[k] = &r |
| 121 | b.order = append(b.order, k) |
| 122 | } |
| 123 | |
| 124 | // Drain returns everything buffered, keyed by app, and resets the groups |
| 125 | // (the per-hour cap counter `seen` and seq persist — they are the hour's and |
| 126 | // the process's, not the batch's; a key seen in an earlier drain this hour |
| 127 | // stays admitted without counting against the cap again). Empty apps are |
| 128 | // omitted. |
| 129 | func (s *Sink) Drain() map[applog.AppKey][]Record { |
| 130 | s.mu.Lock() |
| 131 | defer s.mu.Unlock() |
| 132 | out := make(map[applog.AppKey][]Record) |
| 133 | for k, b := range s.apps { |
| 134 | if len(b.order) == 0 && b.overflow == [2]*Record{} { |
| 135 | continue |
| 136 | } |
| 137 | recs := make([]Record, 0, len(b.order)+2) |
| 138 | for _, gk := range b.order { |
| 139 | recs = append(recs, *b.groups[gk]) |
| 140 | } |
| 141 | for _, o := range b.overflow { |
| 142 | if o != nil { |
| 143 | recs = append(recs, *o) |
| 144 | } |
| 145 | } |
| 146 | out[k] = recs |
| 147 | b.groups = make(map[groupKey]*Record) |
| 148 | b.order = nil |
| 149 | b.overflow = [2]*Record{} |
| 150 | } |
| 151 | return out |
| 152 | } |
| 153 | |
| 154 | // Known lists every app this sink has ever buffered for — the prune |
| 155 | // worklist, as applog.Sink.Known is for logs. |
| 156 | func (s *Sink) Known() []applog.AppKey { |
| 157 | s.mu.Lock() |
| 158 | defer s.mu.Unlock() |
| 159 | out := make([]applog.AppKey, 0, len(s.apps)) |
| 160 | for k := range s.apps { |
| 161 | out = append(out, k) |
| 162 | } |
| 163 | return out |
| 164 | } |
| 165 | |
amadan