mirror of
https://github.com/hanzoai/o11y.git
synced 2026-10-15 12:58:33 +00:00
zeekay
bc2c3be4a4
HIGH-1 amplification: cap events/envelope (1000); module.Ingest pre-aggregates occurrences by fingerprint and store.UpsertIssues writes the whole batch in ONE tx (a flood of identical events => one upsert, count+=N); per-org issue ceiling (10000) drops NEW fingerprints past the cap; per-org token-bucket rate limit (50/s burst 100) after DSN verify; retention/TTL sweep (DeleteStale) gated by O11Y_ERRORTRACKING_RETENTION_DAYS (default 90). MEDIUM-1: envelope item length is bounds-checked (>=0 && <= remaining) before it is added to pos — a MaxInt64/negative length no longer overflows the slice bound; falls back to newline framing. MEDIUM-2: default-on secret redaction (sk-/AKIA/bearer/JWT/conn-string/PAN, always) + PII scrub (email/IP) unless O11Y_ERRORTRACKING_CAPTURE_PII=true (fail-secure, mirrors llmobs capture-messages). Applied before the value enters the fingerprint. MEDIUM-3: DSN keys are versioned '<v>:<hmac>' with a per-org revocation watermark (o11y_ingest_revocations, wholesale-cached); rotate ONE org's min-version to revoke only its below-min DSNs — no global secret roll. LOW-1: lifecycle UpdateIssue is optimistic-concurrency guarded (version column, WHERE version=expected, bump on write; stale => 409 conflict, not clobber). Ingest never touches version. LOW-2: orgFromContext uses NewUUID+error (no MustNewUUID panic on a bad claim). INFO-1: corrected the occurrence-persistence claim + removed the dangling chsink.go reference (only NoopSink exists; logs sink is an honest fast-follow). Tests: 67 pass incl amplification-bounded (5000 events -> 1 upsert), event cap, per-org ceiling, MaxInt64/negative length no-panic, rate-limit, secret/PII scrub, versioned-key + isolated revocation, optimistic-update conflict, DeleteStale. Crown-jewel isolation/parity/DSN tests unchanged and still green.
56 lines
1.4 KiB
Go
56 lines
1.4 KiB
Go
package implerrortracking
|
|
|
|
import (
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/hanzoai/o11y/pkg/valuer"
|
|
)
|
|
|
|
// Per-org token-bucket rate limit on the public ingest path — sustained-flood
|
|
// backpressure that complements the per-request event cap and per-org issue ceiling
|
|
// (which bound a SINGLE request). It is applied AFTER DSN verification, so a bucket
|
|
// is only ever created for an authenticated org (a forged project fails HMAC first
|
|
// and never allocates state). In-process/per-replica by design; a cross-replica
|
|
// distributed quota is a fast-follow.
|
|
const (
|
|
ingestRatePerSec = 50 // steady-state events/requests per org per replica
|
|
ingestBurst = 100 // burst allowance
|
|
)
|
|
|
|
type tokenBucket struct {
|
|
mu sync.Mutex
|
|
tokens float64
|
|
last time.Time
|
|
}
|
|
|
|
func (b *tokenBucket) allow(rate, burst float64, now time.Time) bool {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.tokens += now.Sub(b.last).Seconds() * rate
|
|
if b.tokens > burst {
|
|
b.tokens = burst
|
|
}
|
|
b.last = now
|
|
if b.tokens >= 1 {
|
|
b.tokens--
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
type rateLimiter struct {
|
|
rate float64
|
|
burst float64
|
|
buckets sync.Map // orgID string -> *tokenBucket
|
|
}
|
|
|
|
func newRateLimiter(rate, burst float64) *rateLimiter {
|
|
return &rateLimiter{rate: rate, burst: burst}
|
|
}
|
|
|
|
func (l *rateLimiter) allow(org valuer.UUID) bool {
|
|
v, _ := l.buckets.LoadOrStore(org.String(), &tokenBucket{tokens: l.burst, last: time.Now()})
|
|
return v.(*tokenBucket).allow(l.rate, l.burst, time.Now())
|
|
}
|