amadan

carlos / platform Public

reqrec: Recorder samples not-found and match records, counts per minute

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude 04e702e74437a852a1cf4a2b1b5437a2211a8be3 parent 6a4fcb1
2 files changed, +546 −0
  • internal/reqrec/recorder.go +305 −0
  • internal/reqrec/recorder_test.go +241 −0
diff --git a/internal/reqrec/recorder.go b/internal/reqrec/recorder.go
new file mode 100644
index 000000000..4ab57a771
--- /dev/null
+++ b/internal/reqrec/recorder.go
@@ -0,0 +1,305 @@
+package reqrec
+
+import (
+ "sort"
+ "strconv"
+ "sync"
+ "time"
+)
+
+// Sink receives finished records. The agent's log sink implements it; the
+// recorder never holds its own lock while calling it.
+type Sink interface {
+ Edge(acct, app, host, level, msg string, fields map[string]string)
+}
+
+// Kind says what a record is about.
+type Kind string
+
+const (
+ KindNotFound Kind = "notfound"
+ KindBlock Kind = "block"
+ KindWatch Kind = "watch"
+ KindMinute Kind = "minute"
+)
+
+// Per app and minute, at most this many records of each class are written. The
+// minute counters still count everything, and say how many records were left
+// out, so a flood costs a bounded number of log lines and nothing is hidden.
+const (
+ NotFoundPerMin = 60
+ MatchPerMin = 300
+)
+
+// Response is one response the edge served. It carries raw request data; the
+// recorder redacts it before anything leaves.
+type Response struct {
+ Acct, App, Host, Method, Path, RawQuery, UA, ClientIP, Country string
+ Status int
+ // Woke is whether the request woke the app, and only means something when
+ // WokeKnown is set.
+ Woke, WokeKnown bool
+ When time.Time
+}
+
+// Match is a Response that a protection rule matched, either blocking it or
+// only watching it.
+type Match struct {
+ Response
+ Kind Kind
+ Tier Tier
+ RuleID string
+ // Fragment arrives already passed through Fragment by the caller.
+ Fragment string
+ Ref string
+}
+
+type minuteKey struct {
+ acct, app string
+ minute time.Time
+}
+
+type minuteCounts struct {
+ s2xx, s3xx, s4xx, s404, s5xx int
+ wakes, wakes404 int
+ block, watch int
+ notFoundRecs, matchRecs int
+ droppedNotFound, droppedMatch int
+}
+
+// Recorder turns the edge's responses into sampled records and per-minute
+// counts. A nil Recorder ignores every call.
+type Recorder struct {
+ sink Sink
+ salt func(day string) []byte
+ now func() time.Time
+
+ mu sync.Mutex
+ minutes map[minuteKey]*minuteCounts
+ salts map[string][]byte
+}
+
+// NewRecorder returns a Recorder writing to sink. salt gives the secret for a
+// UTC day (YYYY-MM-DD) and is asked at most once per day; it may be nil, which
+// leaves the client hash out of records. now is used to age out cached salts.
+func NewRecorder(sink Sink, salt func(day string) []byte, now func() time.Time) *Recorder {
+ if now == nil {
+ now = time.Now
+ }
+ return &Recorder{
+ sink: sink,
+ salt: salt,
+ now: now,
+ minutes: map[minuteKey]*minuteCounts{},
+ salts: map[string][]byte{},
+ }
+}
+
+// saltFor returns the salt for day. r.mu must be held. Only today and
+// yesterday are kept, so the cache cannot grow, while a response stamped just
+// before midnight still finds its day.
+func (r *Recorder) saltFor(day string) []byte {
+ if r.salt == nil {
+ return nil
+ }
+ if s, ok := r.salts[day]; ok {
+ return s
+ }
+ s := r.salt(day)
+ if len(r.salts) >= 2 {
+ keep := map[string]bool{
+ r.now().UTC().Format("2006-01-02"): true,
+ r.now().UTC().Add(-24 * time.Hour).Format("2006-01-02"): true,
+ }
+ for d := range r.salts {
+ if !keep[d] {
+ delete(r.salts, d)
+ }
+ }
+ }
+ r.salts[day] = s
+ return s
+}
+
+// count adds x to its minute's counters. r.mu must be held.
+func (c *minuteCounts) count(x Response) {
+ switch {
+ case x.Status == 404:
+ c.s404++
+ case x.Status >= 200 && x.Status < 300:
+ c.s2xx++
+ case x.Status >= 300 && x.Status < 400:
+ c.s3xx++
+ case x.Status >= 400 && x.Status < 500:
+ c.s4xx++
+ case x.Status >= 500 && x.Status < 600:
+ c.s5xx++
+ }
+ if x.WokeKnown && x.Woke {
+ c.wakes++
+ if x.Status == 404 {
+ c.wakes404++
+ }
+ }
+}
+
+func (r *Recorder) counts(x Response) *minuteCounts {
+ k := minuteKey{x.Acct, x.App, x.When.UTC().Truncate(time.Minute)}
+ c := r.minutes[k]
+ if c == nil {
+ c = &minuteCounts{}
+ r.minutes[k] = c
+ }
+ return c
+}
+
+// Response counts a response and, for a 404 under the per-minute cap, writes a
+// not-found record.
+func (r *Recorder) Response(x Response) {
+ if r == nil {
+ return
+ }
+ r.mu.Lock()
+ c := r.counts(x)
+ c.count(x)
+ var salt []byte
+ emit := false
+ if x.Status == 404 {
+ if c.notFoundRecs < NotFoundPerMin {
+ c.notFoundRecs++
+ emit = true
+ salt = r.saltFor(x.When.UTC().Format("2006-01-02"))
+ } else {
+ c.droppedNotFound++
+ }
+ }
+ r.mu.Unlock()
+ if !emit {
+ return
+ }
+ f := r.fields(x, KindNotFound, salt)
+ r.sink.Edge(x.Acct, x.App, x.Host, "info", "404 "+x.Method+" "+f["path"], f)
+}
+
+// Match counts a response a rule matched, as a block or a watch and by status
+// class like any other response, and writes a match record under the
+// per-minute cap. It makes no not-found record, so a matched 404 is one record.
+func (r *Recorder) Match(m Match) {
+ if r == nil {
+ return
+ }
+ r.mu.Lock()
+ c := r.counts(m.Response)
+ c.count(m.Response)
+ switch m.Kind {
+ case KindBlock:
+ c.block++
+ case KindWatch:
+ c.watch++
+ }
+ var salt []byte
+ emit := false
+ if c.matchRecs < MatchPerMin {
+ c.matchRecs++
+ emit = true
+ salt = r.saltFor(m.When.UTC().Format("2006-01-02"))
+ } else {
+ c.droppedMatch++
+ }
+ r.mu.Unlock()
+ if !emit {
+ return
+ }
+ f := r.fields(m.Response, m.Kind, salt)
+ f["tier"] = m.Tier.String()
+ set(f, "ref", m.Ref)
+ set(f, "rule", m.RuleID)
+ set(f, "frag", m.Fragment)
+ level := "info"
+ if m.Kind == KindBlock {
+ level = "warn"
+ }
+ r.sink.Edge(m.Acct, m.App, m.Host, level,
+ string(m.Kind)+" "+m.Tier.String()+" "+m.RuleID+" "+m.Method+" "+f["path"], f)
+}
+
+// fields builds the redacted record body shared by not-found and match records.
+// Empty values are left out.
+func (r *Recorder) fields(x Response, kind Kind, salt []byte) map[string]string {
+ f := map[string]string{
+ "kind": string(kind),
+ "method": x.Method,
+ "path": MaskPath(x.Path),
+ "ua": UAFamily(x.UA),
+ "status": strconv.Itoa(x.Status),
+ }
+ set(f, "qnames", QueryNames(x.RawQuery))
+ set(f, "country", x.Country)
+ set(f, "client", ClientPrefix(x.ClientIP))
+ set(f, "cid", ClientHash(salt, ClientKey(x.ClientIP)))
+ if x.WokeKnown {
+ f["woke"] = "0"
+ if x.Woke {
+ f["woke"] = "1"
+ }
+ }
+ return f
+}
+
+func set(f map[string]string, k, v string) {
+ if v != "" {
+ f[k] = v
+ }
+}
+
+// Flush writes one minute record for every app and minute that ended before
+// now, oldest first, and forgets them. The minute still open is left alone so
+// late responses land in it.
+func (r *Recorder) Flush(now time.Time) {
+ if r == nil {
+ return
+ }
+ cutoff := now.UTC().Truncate(time.Minute)
+ type done struct {
+ k minuteKey
+ c minuteCounts
+ }
+ var out []done
+ r.mu.Lock()
+ for k, c := range r.minutes {
+ if k.minute.Before(cutoff) {
+ out = append(out, done{k, *c})
+ delete(r.minutes, k)
+ }
+ }
+ r.mu.Unlock()
+ sort.Slice(out, func(i, j int) bool {
+ a, b := out[i].k, out[j].k
+ if !a.minute.Equal(b.minute) {
+ return a.minute.Before(b.minute)
+ }
+ if a.acct != b.acct {
+ return a.acct < b.acct
+ }
+ return a.app < b.app
+ })
+ for _, d := range out {
+ minute := d.k.minute.Format(time.RFC3339)
+ n := strconv.Itoa
+ r.sink.Edge(d.k.acct, d.k.app, "", "info", "minute "+minute, map[string]string{
+ "kind": string(KindMinute),
+ "minute": minute,
+ "s2xx": n(d.c.s2xx),
+ "s3xx": n(d.c.s3xx),
+ "s4xx": n(d.c.s4xx),
+ "s404": n(d.c.s404),
+ "s5xx": n(d.c.s5xx),
+ "wakes": n(d.c.wakes),
+ "wakes404": n(d.c.wakes404),
+ "block": n(d.c.block),
+ "watch": n(d.c.watch),
+ "dropped_notfound": n(d.c.droppedNotFound),
+ "dropped_match": n(d.c.droppedMatch),
+ })
+ }
+}
diff --git a/internal/reqrec/recorder_test.go b/internal/reqrec/recorder_test.go
new file mode 100644
index 000000000..03199e718
--- /dev/null
+++ b/internal/reqrec/recorder_test.go
@@ -0,0 +1,241 @@
+package reqrec
+
+import (
+ "bytes"
+ "fmt"
+ "sync"
+ "testing"
+ "time"
+)
+
+type memSink struct {
+ recs []map[string]string
+ msgs, levels []string
+}
+
+func (m *memSink) Edge(acct, app, host, level, msg string, f map[string]string) {
+ m.recs, m.msgs, m.levels = append(m.recs, f), append(m.msgs, msg), append(m.levels, level)
+}
+
+func TestRecorderNotFoundSampledAndCounted(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, func(string) []byte { return bytes.Repeat([]byte{7}, 32) }, func() time.Time { return now })
+ for i := 0; i < 10000; i++ {
+ r.Response(Response{Acct: "a1", App: "shop", Host: "shop.example.com", Method: "GET",
+ Path: "/wp-login.php", UA: "curl/8.7.1", ClientIP: "203.0.113.7", Status: 404, Woke: i == 0, WokeKnown: true, When: now})
+ }
+ r.Response(Response{Acct: "a1", App: "shop", Method: "GET", Path: "/", Status: 200, WokeKnown: true, When: now})
+ r.Flush(now.Add(time.Minute))
+ var notfound, minute int
+ for _, f := range s.recs {
+ switch f["kind"] {
+ case "notfound":
+ notfound++
+ if f["client"] != "203.0.113.0/24" || f["ua"] != "tool" || f["status"] != "404" || len(f["cid"]) != 12 {
+ t.Fatalf("bad notfound fields %v", f)
+ }
+ case "minute":
+ minute++
+ if f["s404"] != "10000" || f["s2xx"] != "1" || f["wakes"] != "1" || f["wakes404"] != "1" || f["dropped_notfound"] != "9940" {
+ t.Fatalf("bad minute %v", f)
+ }
+ }
+ }
+ if notfound != NotFoundPerMin || minute != 1 {
+ t.Fatalf("notfound=%d minute=%d", notfound, minute)
+ }
+}
+
+func TestRecorderMatchRecordsAndCap(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, nil, func() time.Time { return now })
+ m := Match{Response: Response{Acct: "a1", App: "shop", Method: "GET", Path: "/.env", Status: 404, When: now},
+ Kind: KindBlock, Tier: TierHard, RuleID: "dotenv", Ref: "ref1"}
+ for i := 0; i < MatchPerMin+5; i++ {
+ r.Match(m)
+ }
+ r.Flush(now.Add(time.Minute))
+ first := s.recs[0]
+ if first["kind"] != "block" || first["tier"] != "hard" || first["rule"] != "dotenv" || first["ref"] != "ref1" || s.levels[0] != "warn" {
+ t.Fatalf("bad match record %v level %s", first, s.levels[0])
+ }
+ if _, ok := first["woke"]; ok {
+ t.Error("woke written while unknown")
+ }
+ if first["cid"] != "" {
+ t.Error("cid written without a salt")
+ }
+ last := s.recs[len(s.recs)-1]
+ if last["kind"] != "minute" || last["block"] != fmt.Sprint(MatchPerMin+5) || last["dropped_match"] != "5" {
+ t.Fatalf("bad minute %v", last)
+ }
+}
+
+func TestRecorderFlushOnlyCompletedMinutes(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 30, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, nil, func() time.Time { return now })
+ r.Response(Response{Acct: "a1", App: "shop", Status: 200, When: now})
+ r.Flush(now) // minute 21:07 not complete
+ if len(s.recs) != 0 {
+ t.Fatal("flushed an open minute")
+ }
+ r.Flush(now.Add(31 * time.Second))
+ if len(s.recs) != 1 || s.recs[0]["minute"] != "2026-10-24T21:07:00Z" {
+ t.Fatalf("got %v", s.recs)
+ }
+}
+
+func TestRecorderStatusClassesAndMatchCounts(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, nil, func() time.Time { return now })
+ for _, st := range []int{200, 204, 301, 400, 403, 404, 500, 503} {
+ r.Response(Response{Acct: "a1", App: "shop", Method: "GET", Path: "/x", Status: st, When: now})
+ }
+ // A watch match counts as a watch and in its status class, and makes no
+ // not-found record of its own even for a 404.
+ r.Match(Match{Response: Response{Acct: "a1", App: "shop", Method: "GET", Path: "/y", Status: 404, Woke: true, WokeKnown: true, When: now},
+ Kind: KindWatch, Tier: TierProbe, RuleID: "r"})
+ r.Flush(now.Add(time.Minute))
+ var min map[string]string
+ var kinds []string
+ for _, f := range s.recs {
+ kinds = append(kinds, f["kind"])
+ if f["kind"] == "minute" {
+ min = f
+ }
+ }
+ want := map[string]string{"s2xx": "2", "s3xx": "1", "s4xx": "2", "s404": "2", "s5xx": "2",
+ "wakes": "1", "wakes404": "1", "block": "0", "watch": "1", "dropped_notfound": "0", "dropped_match": "0"}
+ for k, v := range want {
+ if min[k] != v {
+ t.Errorf("%s = %q, want %q (%v)", k, min[k], v, min)
+ }
+ }
+ if min["minute"] != "2026-10-24T21:07:00Z" || s.msgs[len(s.msgs)-1] != "minute 2026-10-24T21:07:00Z" {
+ t.Errorf("bad minute label %v %q", min, s.msgs[len(s.msgs)-1])
+ }
+ nf, w := 0, 0
+ for _, k := range kinds {
+ if k == "notfound" {
+ nf++
+ }
+ if k == "watch" {
+ w++
+ }
+ }
+ // The plain 404 gets a record, the watch match gets one, the 404 under a
+ // match does not get a second.
+ if nf != 1 || w != 1 {
+ t.Errorf("kinds %v", kinds)
+ }
+}
+
+func TestRecorderNotFoundMessageAndOmittedFields(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &memSink{}
+ r := NewRecorder(s, nil, func() time.Time { return now })
+ r.Response(Response{Acct: "a1", App: "shop", Method: "GET", Path: "/a", Status: 404, When: now})
+ f := s.recs[0]
+ if s.msgs[0] != "404 GET /a" || s.levels[0] != "info" {
+ t.Errorf("msg %q level %q", s.msgs[0], s.levels[0])
+ }
+ for _, k := range []string{"country", "cid", "qnames", "woke", "client"} {
+ if v, ok := f[k]; ok {
+ t.Errorf("%s present as %q, want omitted", k, v)
+ }
+ }
+ m := Match{Response: Response{Acct: "a1", App: "shop", Method: "POST", Path: "/b", Status: 403, Woke: false, WokeKnown: true, When: now},
+ Kind: KindWatch, Tier: TierProbe, RuleID: "r1", Fragment: "x"}
+ r.Match(m)
+ if s.msgs[1] != "watch probe r1 POST /b" || s.levels[1] != "info" || s.recs[1]["woke"] != "0" || s.recs[1]["frag"] != "x" {
+ t.Errorf("match record %q %q %v", s.msgs[1], s.levels[1], s.recs[1])
+ }
+}
+
+func TestRecorderSaltOncePerDayAndFlushOrder(t *testing.T) {
+ now := time.Date(2026, 10, 24, 23, 59, 0, 0, time.UTC)
+ s := &memSink{}
+ var days []string
+ r := NewRecorder(s, func(day string) []byte {
+ days = append(days, day)
+ return []byte("salt-" + day)
+ }, func() time.Time { return now })
+ for i := 0; i < 5; i++ {
+ r.Response(Response{Acct: "a1", App: "shop", Method: "GET", Path: "/", ClientIP: "198.51.100.9", Status: 404, When: now})
+ }
+ next := now.Add(2 * time.Minute) // 00:01 the next day
+ r.Response(Response{Acct: "a1", App: "shop", Method: "GET", Path: "/", ClientIP: "198.51.100.9", Status: 404, When: next})
+ r.Response(Response{Acct: "a1", App: "blog", Method: "GET", Path: "/", Status: 200, When: now})
+ if len(days) != 2 || days[0] != "2026-10-24" || days[1] != "2026-10-25" {
+ t.Fatalf("salt days %v", days)
+ }
+ if s.recs[0]["cid"] == s.recs[5]["cid"] {
+ t.Error("cid did not change across days")
+ }
+ s.recs, s.msgs = nil, nil
+ r.Flush(next.Add(time.Minute))
+ if len(s.msgs) != 3 || s.msgs[0] != "minute 2026-10-24T23:59:00Z" || s.msgs[2] != "minute 2026-10-25T00:01:00Z" {
+ t.Fatalf("flush order %v", s.msgs)
+ }
+ r.Flush(next.Add(time.Hour))
+ if len(s.msgs) != 3 {
+ t.Fatal("flushed minutes were not forgotten")
+ }
+}
+
+func TestRecorderNil(t *testing.T) {
+ var r *Recorder
+ r.Response(Response{})
+ r.Match(Match{})
+ r.Flush(time.Now())
+}
+
+func TestRecorderConcurrentResponses(t *testing.T) {
+ now := time.Date(2026, 10, 24, 21, 7, 0, 0, time.UTC)
+ s := &lockedSink{}
+ r := NewRecorder(s, func(string) []byte { return []byte("salt") }, func() time.Time { return now })
+ var wg sync.WaitGroup
+ for g := 0; g < 16; g++ {
+ wg.Add(1)
+ go func(g int) {
+ defer wg.Done()
+ for i := 0; i < 200; i++ {
+ r.Response(Response{Acct: "a1", App: fmt.Sprint("app", g%3), Method: "GET", Path: "/p",
+ ClientIP: "203.0.113.7", Status: 404, When: now})
+ if i%50 == 0 {
+ r.Flush(now.Add(time.Minute))
+ }
+ }
+ }(g)
+ }
+ wg.Wait()
+ r.Flush(now.Add(time.Minute))
+ var total int
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ for _, f := range s.recs {
+ if f["kind"] == "minute" {
+ var n int
+ fmt.Sscan(f["s404"], &n)
+ total += n
+ }
+ }
+ if total != 16*200 {
+ t.Fatalf("counted %d 404s, want %d", total, 16*200)
+ }
+}
+
+type lockedSink struct {
+ mu sync.Mutex
+ recs []map[string]string
+}
+
+func (l *lockedSink) Edge(acct, app, host, level, msg string, f map[string]string) {
+ l.mu.Lock()
+ l.recs = append(l.recs, f)
+ l.mu.Unlock()
+}