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.
99 lines
3.0 KiB
Go
99 lines
3.0 KiB
Go
package implerrortracking
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/hanzoai/o11y/pkg/sqlstore"
|
|
"github.com/hanzoai/o11y/pkg/valuer"
|
|
"github.com/uptrace/bun"
|
|
)
|
|
|
|
// RevocationStore resolves an org's minimum-acceptable DSN key version, enabling
|
|
// PER-ORG key rotation without a global secret roll: bump ONE org's min-version and
|
|
// only that org's older-version DSNs stop verifying. Default 0 = no revocation.
|
|
type RevocationStore interface {
|
|
MinVersion(ctx context.Context, orgID valuer.UUID) int
|
|
}
|
|
|
|
// NoopRevocations never revokes (every org's min-version is 0). Used where key
|
|
// rotation state isn't wired (tests, standalone).
|
|
type NoopRevocations struct{}
|
|
|
|
func (NoopRevocations) MinVersion(context.Context, valuer.UUID) int { return 0 }
|
|
|
|
const revocationCacheTTL = 30 * time.Second
|
|
|
|
// ingestRevocation is one org's rotation watermark.
|
|
type ingestRevocation struct {
|
|
bun.BaseModel `bun:"table:o11y_ingest_revocations,alias:o11y_ingest_revocations"`
|
|
|
|
OrgID valuer.UUID `bun:"org_id,pk,type:text"`
|
|
MinVersion int64 `bun:"min_version,notnull,default:0"`
|
|
UpdatedAt time.Time `bun:"updated_at,notnull"`
|
|
}
|
|
|
|
// sqlRevocations is the table-backed store with a wholesale in-memory cache (the
|
|
// table is small — one row per rotated org). A read miss or TTL lapse reloads the
|
|
// whole table; on a load error the last-good cache is kept (fail-open to the last
|
|
// known state, never crash the hot ingest path).
|
|
type sqlRevocations struct {
|
|
sqlstore sqlstore.SQLStore
|
|
ttl time.Duration
|
|
|
|
mu sync.RWMutex
|
|
cache map[string]int
|
|
loadedAt time.Time
|
|
}
|
|
|
|
func NewSQLRevocations(sqlstore sqlstore.SQLStore) *sqlRevocations {
|
|
return &sqlRevocations{sqlstore: sqlstore, ttl: revocationCacheTTL, cache: map[string]int{}}
|
|
}
|
|
|
|
func (r *sqlRevocations) MinVersion(ctx context.Context, orgID valuer.UUID) int {
|
|
r.mu.RLock()
|
|
fresh := !r.loadedAt.IsZero() && time.Since(r.loadedAt) < r.ttl
|
|
v := r.cache[orgID.String()]
|
|
r.mu.RUnlock()
|
|
if fresh {
|
|
return v
|
|
}
|
|
r.reload(ctx)
|
|
r.mu.RLock()
|
|
defer r.mu.RUnlock()
|
|
return r.cache[orgID.String()]
|
|
}
|
|
|
|
func (r *sqlRevocations) reload(ctx context.Context) {
|
|
var rows []ingestRevocation
|
|
if err := r.sqlstore.BunDBCtx(ctx).NewSelect().Model(&rows).Scan(ctx); err != nil {
|
|
return // keep stale cache on error
|
|
}
|
|
m := make(map[string]int, len(rows))
|
|
for _, row := range rows {
|
|
m[row.OrgID.String()] = int(row.MinVersion)
|
|
}
|
|
r.mu.Lock()
|
|
r.cache = m
|
|
r.loadedAt = time.Now()
|
|
r.mu.Unlock()
|
|
}
|
|
|
|
// Rotate raises an org's minimum key version, revoking every DSN issued below it.
|
|
// The operator then mints a fresh DSN at the new version.
|
|
func (r *sqlRevocations) Rotate(ctx context.Context, orgID valuer.UUID, minVersion int) error {
|
|
rev := &ingestRevocation{OrgID: orgID, MinVersion: int64(minVersion), UpdatedAt: time.Now().UTC()}
|
|
_, err := r.sqlstore.BunDBCtx(ctx).
|
|
NewInsert().
|
|
Model(rev).
|
|
On("CONFLICT (org_id) DO UPDATE").
|
|
Set("min_version = EXCLUDED.min_version").
|
|
Set("updated_at = EXCLUDED.updated_at").
|
|
Exec(ctx)
|
|
r.mu.Lock()
|
|
r.loadedAt = time.Time{} // force reload on next read
|
|
r.mu.Unlock()
|
|
return err
|
|
}
|