Compare commits

...
Author SHA1 Message Date
hanzo-dev b0d74a3217 fix(errortracking): harden ingest per red review (HIGH-1/2, MED-1/2/3, LOW-1/2)
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.
2026-07-10 14:18:11 -07:00
hanzo-dev 3a4fab400f feat(errortracking): Sentry-class error tracking folded into the o11y plane
New pkg/modules/errortracking + pkg/types/errortrackingtypes: grouped-error
Issues over the ONE o11y plane. Occurrences stay in the telemetry store; only
non-derivable lifecycle (status/assignee/first-last-seen/count/regression) is
the net-new o11y_issues table.

- Fingerprint grouping at ingest (exception.type + normalized crash frame,
  message fallback; honors client fingerprint + {{ default }}).
- Sentry-envelope + legacy /store/ ingest shim -> normalize -> upsert issue.
  Public routes (OpenAccess) authenticated by the DSN public key
  (HMAC-SHA256 over the org, KMS platform secret, constant-time, fail-closed);
  org resolved from the DSN project via iamidentn's exact UUIDv5 so ingest and
  the IAM read align. No gateway change: DSN path /v1/o11y/<org> makes the SDK
  POST /v1/o11y/api/<org>/envelope/, which the existing mount forwards.
- Read routes (ListIssues/GetIssue/UpdateIssue) behind Hanzo IAM authz,
  org-scoped in SQL EXACTLY like llmobs (Where org_id = ?, fail-closed).
- Occurrence sink seam (default no-op; ClickHouse o11y_logs writer gated).
- 100_add_error_tracking migration: o11y_issues + unique (org_id,fingerprint).
- Tests (54): two-org isolation, upsert grouping, regression reopen, DSN auth
  (accept/reject/cross-org/disabled), fingerprint stability, envelope/store
  parsing (gzip/deflate), normalize (ts/message/tags/exception polymorphism).
2026-07-10 10:34:00 -07:00
30 changed files with 3339 additions and 2 deletions
@@ -0,0 +1,83 @@
package o11yapiserver
import (
"net/http"
"github.com/gorilla/mux"
"github.com/hanzoai/o11y/pkg/http/handler"
"github.com/hanzoai/o11y/pkg/types"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
)
// addErrorTrackingRoutes serves error/crash tracking (Sentry-class Issues) under
// /v1/o11y. Two families:
//
// - INGEST (public, DSN-authenticated in-handler): the Sentry wire endpoints
// POST /api/{project}/envelope/ and POST /api/{project}/store/. They are wrapped
// with OpenAccess (no IAM) because the Sentry SDK presents a DSN key, not a Hanzo
// session; the handler verifies that key. A Sentry DSN of
// https://<key>@<host>/v1/o11y/<org> makes the SDK POST to
// /v1/o11y/api/<org>/envelope/, which the existing /v1/o11y mount forwards here
// — no gateway change. The literal /api/ segment is the fixed Sentry wire
// contract, not a Hanzo-designed route.
//
// - READ (Hanzo IAM authz, org-scoped): the Issues list/detail/update the console
// Errors tab consumes at /v1/o11y/errortracking/issues[/{id}].
func (provider *provider) addErrorTrackingRoutes(router *mux.Router) error {
h := provider.errorTrackingHandler
routes := []struct {
method string
path string
fn http.HandlerFunc
def handler.OpenAPIDef
}{
{http.MethodPost, "/api/{project_id}/envelope/", provider.authzMiddleware.OpenAccess(h.EnvelopeIngest), handler.OpenAPIDef{
ID: "IngestErrorEnvelope", Tags: []string{"errortracking"}, Summary: "Ingest a Sentry envelope",
Description: "Sentry-envelope-compatible ingest. Authenticated by the DSN public key (X-Sentry-Auth or ?sentry_key), not a Hanzo session.",
RequestContentType: "application/x-sentry-envelope",
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusServiceUnavailable},
SecuritySchemes: []handler.OpenAPISecurityScheme{},
}},
{http.MethodPost, "/api/{project_id}/store/", provider.authzMiddleware.OpenAccess(h.StoreIngest), handler.OpenAPIDef{
ID: "IngestErrorStore", Tags: []string{"errortracking"}, Summary: "Ingest a legacy Sentry store event",
Description: "Legacy single-event Sentry ingest. Authenticated by the DSN public key.",
RequestContentType: "application/json",
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusServiceUnavailable},
SecuritySchemes: []handler.OpenAPISecurityScheme{},
}},
{http.MethodGet, "/api/errortracking/issues", provider.authzMiddleware.ViewAccess(h.ListIssues), handler.OpenAPIDef{
ID: "ListIssues", Tags: []string{"errortracking"}, Summary: "List error issues",
Description: "Lists grouped error issues (by fingerprint) for the caller's org with status, level, counts and first/last-seen.",
RequestQuery: new(errortrackingtypes.IssuesQuery),
Response: new(errortrackingtypes.GettableIssues),
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest}, SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodGet, "/api/errortracking/issues/{id}", provider.authzMiddleware.ViewAccess(h.GetIssue), handler.OpenAPIDef{
ID: "GetIssue", Tags: []string{"errortracking"}, Summary: "Get an error issue",
Description: "Returns a single issue with its latest occurrence sample.",
Response: new(errortrackingtypes.GettableIssue),
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusNotFound}, SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodPost, "/api/errortracking/issues/{id}", provider.authzMiddleware.EditAccess(h.UpdateIssue), handler.OpenAPIDef{
ID: "UpdateIssue", Tags: []string{"errortracking"}, Summary: "Update an issue's lifecycle",
Description: "Resolve, ignore, reopen or assign an issue.",
Request: new(errortrackingtypes.UpdateIssue), RequestContentType: "application/json",
Response: new(errortrackingtypes.Issue),
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest, http.StatusNotFound}, SecuritySchemes: newSecuritySchemes(types.RoleEditor),
}},
}
for _, rt := range routes {
if err := router.Handle(rt.path, handler.New(rt.fn, rt.def)).Methods(rt.method).GetError(); err != nil {
return err
}
}
return nil
}
+10
View File
@@ -16,6 +16,7 @@ import (
"github.com/hanzoai/o11y/pkg/modules/authdomain"
"github.com/hanzoai/o11y/pkg/modules/cloudintegration"
"github.com/hanzoai/o11y/pkg/modules/dashboard"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/modules/fields"
"github.com/hanzoai/o11y/pkg/modules/inframonitoring"
"github.com/hanzoai/o11y/pkg/modules/llmobs"
@@ -75,6 +76,7 @@ type provider struct {
rulerHandler ruler.Handler
llmPricingRuleHandler llmpricingrule.Handler
llmObsHandler llmobs.Handler
errorTrackingHandler errortracking.Handler
statsHandler statsreporter.Handler
}
@@ -111,6 +113,7 @@ func NewFactory(
rulerHandler ruler.Handler,
statsHandler statsreporter.Handler,
llmObsHandler llmobs.Handler,
errorTrackingHandler errortracking.Handler,
) factory.ProviderFactory[apiserver.APIServer, apiserver.Config] {
return factory.NewProviderFactory(factory.MustNewName("o11y"), func(ctx context.Context, providerSettings factory.ProviderSettings, config apiserver.Config) (apiserver.APIServer, error) {
return newProvider(
@@ -149,6 +152,7 @@ func NewFactory(
rulerHandler,
statsHandler,
llmObsHandler,
errorTrackingHandler,
)
})
}
@@ -189,6 +193,7 @@ func newProvider(
rulerHandler ruler.Handler,
statsHandler statsreporter.Handler,
llmObsHandler llmobs.Handler,
errorTrackingHandler errortracking.Handler,
) (apiserver.APIServer, error) {
settings := factory.NewScopedProviderSettings(providerSettings, "github.com/hanzoai/o11y/pkg/apiserver/o11yapiserver")
router := mux.NewRouter().UseEncodedPath()
@@ -227,6 +232,7 @@ func newProvider(
rulerHandler: rulerHandler,
llmPricingRuleHandler: llmPricingRuleHandler,
llmObsHandler: llmObsHandler,
errorTrackingHandler: errorTrackingHandler,
statsHandler: statsHandler,
}
@@ -352,6 +358,10 @@ func (provider *provider) AddToRouter(router *mux.Router) error {
return err
}
if err := provider.addErrorTrackingRoutes(router); err != nil {
return err
}
if err := provider.addTraceDetailRoutes(router); err != nil {
return err
}
@@ -0,0 +1,41 @@
package errortracking
import (
"context"
"net/http"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
// Module is the native error/crash tracking surface (Sentry-class Issues) folded
// into the o11y plane. Occurrences are OTel exception data in the telemetry store;
// this module owns the grouped-Issue lifecycle over that data and the ingest that
// normalizes Sentry-SDK reports into it.
type Module interface {
// Ingest groups a BATCH of normalized occurrences into the caller's org (resolved
// from the DSN by the handler): occurrences are collapsed by fingerprint and
// upserted in one transaction under the per-org issue ceiling, bounding the write
// amplification of a single request. Returns issues written.
Ingest(ctx context.Context, orgID valuer.UUID, occs []*errortrackingtypes.Occurrence) (int, error)
ListIssues(ctx context.Context, orgID valuer.UUID, q *errortrackingtypes.IssuesQuery) ([]*errortrackingtypes.Issue, int, error)
GetIssue(ctx context.Context, orgID, id valuer.UUID) (*errortrackingtypes.GettableIssue, error)
UpdateIssue(ctx context.Context, orgID, id valuer.UUID, in *errortrackingtypes.UpdateIssue) (*errortrackingtypes.Issue, error)
}
// Handler is the HTTP surface. The ingest endpoints are PUBLIC (OpenAccess) and
// authenticate the Sentry DSN key in-handler; the read endpoints are behind the
// shared Hanzo IAM authz middleware and are org-scoped from the validated claims.
type Handler interface {
// EnvelopeIngest accepts the modern Sentry envelope wire format
// (POST /api/{project}/envelope/).
EnvelopeIngest(rw http.ResponseWriter, r *http.Request)
// StoreIngest accepts the legacy single-event wire format
// (POST /api/{project}/store/).
StoreIngest(rw http.ResponseWriter, r *http.Request)
ListIssues(rw http.ResponseWriter, r *http.Request)
GetIssue(rw http.ResponseWriter, r *http.Request)
UpdateIssue(rw http.ResponseWriter, r *http.Request)
}
@@ -0,0 +1,123 @@
package implerrortracking
import (
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"net/http"
"strconv"
"strings"
"github.com/google/uuid"
"github.com/hanzoai/o11y/pkg/valuer"
)
// The ingest endpoints authenticate with the Sentry-native DSN model: the caller
// presents a public key that proves it holds an org's ingest credential. We make
// that key STATELESS and KMS-backed rather than adding a per-key secret table:
//
// publicKey(org, v) = "<v>:" + hex(HMAC-SHA256(platformIngestSecret, "org:"+org+":v"+v))
//
// The platform secret comes from KMS (never plaintext, never committed). The key
// carries its VERSION so ONE org can be rotated in isolation: bump that org's
// min-version (RevocationStore.Rotate) and only its below-min DSNs stop verifying —
// no global secret roll. Verifying is a version check + a constant-time compare.
//
// The org travels in the DSN project segment; the key proves the caller may write
// to THAT org at THAT version. Resolution reuses iamidentn's exact UUIDv5 mapping so
// the row written here is read back by exactly that tenant.
const defaultKeyVersion = 1
// orgUUIDFromProject maps a DSN project segment to the o11y org UUID. It mirrors
// iamidentn.toUUID("org", …) BYTE-FOR-BYTE (a raw UUID passes through; a slug is
// UUIDv5 over the URL namespace with the "hanzo:o11y:org:" prefix) so ingest and
// the IAM read path resolve the SAME tenant id.
func orgUUIDFromProject(project string) (valuer.UUID, bool) {
project = strings.TrimSpace(project)
if project == "" {
return valuer.UUID{}, false
}
if u, err := valuer.NewUUID(project); err == nil {
return u, true
}
derived := uuid.NewSHA1(uuid.NameSpaceURL, []byte("hanzo:o11y:org:"+project))
return valuer.MustNewUUID(derived.String()), true
}
// publicKeyForVersion derives the versioned ingest public key for a project.
func publicKeyForVersion(secret []byte, project string, version int) string {
m := hmac.New(sha256.New, secret)
m.Write([]byte("org:" + strings.TrimSpace(project) + ":v" + strconv.Itoa(version)))
return strconv.Itoa(version) + ":" + hex.EncodeToString(m.Sum(nil))
}
// publicKeyFor derives the default (v1) key.
func publicKeyFor(secret []byte, project string) string {
return publicKeyForVersion(secret, project, defaultKeyVersion)
}
// verifyKey constant-time compares a presented "<v>:<hmac>" key against the expected
// one for its project, rejecting versions below the org's revocation watermark. An
// empty secret or key, a malformed version, or a below-min version never verify
// (fail closed).
func verifyKey(secret []byte, project, presented string, minVersion int) bool {
if len(secret) == 0 || presented == "" {
return false
}
i := strings.IndexByte(presented, ':')
if i <= 0 {
return false
}
version, err := strconv.Atoi(presented[:i])
if err != nil || version <= 0 {
return false
}
if version < minVersion {
return false // revoked by rotation
}
expected := publicKeyForVersion(secret, project, version)
return hmac.Equal([]byte(expected), []byte(presented))
}
// sentryKeyFromRequest extracts the presented public key from the Sentry auth
// surface, in precedence order: the X-Sentry-Auth header, then the ?sentry_key
// query param. (The envelope-header DSN is intentionally NOT trusted as an auth
// source — it is client body, not a credential channel.)
func sentryKeyFromRequest(r *http.Request) string {
if k := parseSentryAuthHeader(r.Header.Get("X-Sentry-Auth")); k != "" {
return k
}
return strings.TrimSpace(r.URL.Query().Get("sentry_key"))
}
// parseSentryAuthHeader pulls sentry_key out of a header like:
//
// Sentry sentry_version=7, sentry_key=1:abc123, sentry_client=sentry.python/1.2
func parseSentryAuthHeader(h string) string {
h = strings.TrimSpace(h)
if h == "" {
return ""
}
h = strings.TrimPrefix(h, "Sentry ")
for _, part := range strings.Split(h, ",") {
kv := strings.SplitN(strings.TrimSpace(part), "=", 2)
if len(kv) == 2 && strings.TrimSpace(kv[0]) == "sentry_key" {
return strings.TrimSpace(kv[1])
}
}
return ""
}
// MintDSN builds the default-version DSN an operator hands to an app to report into
// an org. host is the ingest origin (e.g. "o11y.hanzo.ai"); the SDK derives its
// endpoint as https://<host>/v1/o11y/api/<org>/envelope/, which the existing
// /v1/o11y mount forwards to this module's ingest route.
func MintDSN(secret []byte, host, org string) string {
return MintDSNVersion(secret, host, org, defaultKeyVersion)
}
// MintDSNVersion builds a DSN at a specific key version (used after rotating an org).
func MintDSNVersion(secret []byte, host, org string, version int) string {
return "https://" + publicKeyForVersion(secret, org, version) + "@" + host + "/v1/o11y/" + org
}
@@ -0,0 +1,82 @@
package implerrortracking
import (
"net/http/httptest"
"testing"
"github.com/google/uuid"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
var testSecret = []byte("kms-platform-ingest-secret")
func TestVerifyKey_RoundTrip(t *testing.T) {
key := publicKeyFor(testSecret, "acme")
assert.True(t, verifyKey(testSecret, "acme", key, 0), "the derived key must verify for its project")
}
func TestVerifyKey_RejectsWrongProject(t *testing.T) {
key := publicKeyFor(testSecret, "acme")
assert.False(t, verifyKey(testSecret, "evil", key, 0), "a key minted for acme must not verify for another project")
}
func TestVerifyKey_RejectsWrongSecret(t *testing.T) {
key := publicKeyFor(testSecret, "acme")
assert.False(t, verifyKey([]byte("different-secret"), "acme", key, 0))
}
func TestVerifyKey_FailsClosed(t *testing.T) {
assert.False(t, verifyKey(nil, "acme", "anything", 0), "no secret => fail closed")
assert.False(t, verifyKey(testSecret, "acme", "", 0), "no presented key => fail closed")
}
// The MOST important parity test: the org UUID the ingest path derives from a DSN
// project MUST equal iamidentn.toUUID("org", slug) — otherwise a row written by
// ingest would be invisible to the org's IAM-authenticated reads. This replicates
// iamidentn's exact formula and asserts equality.
func TestOrgUUIDFromProject_MatchesIAMMapping(t *testing.T) {
for _, slug := range []string{"hanzo", "acme", "zoo"} {
want := uuid.NewSHA1(uuid.NameSpaceURL, []byte("hanzo:o11y:org:"+slug))
got, ok := orgUUIDFromProject(slug)
require.True(t, ok)
assert.Equal(t, want.String(), got.String(), "ingest org UUID must match the IAM read-path UUID for slug %q", slug)
}
}
func TestOrgUUIDFromProject_RawUUIDPassthrough(t *testing.T) {
u := valuer.GenerateUUID()
got, ok := orgUUIDFromProject(u.String())
require.True(t, ok)
assert.Equal(t, u.String(), got.String(), "a project that is already a UUID is used as-is")
}
func TestOrgUUIDFromProject_EmptyRejected(t *testing.T) {
_, ok := orgUUIDFromProject("")
assert.False(t, ok)
_, ok = orgUUIDFromProject(" ")
assert.False(t, ok)
}
func TestSentryKeyFromRequest_Header(t *testing.T) {
r := httptest.NewRequest("POST", "/api/acme/envelope/", nil)
r.Header.Set("X-Sentry-Auth", "Sentry sentry_version=7, sentry_key=pubkey123, sentry_client=sentry.python/1.40")
assert.Equal(t, "pubkey123", sentryKeyFromRequest(r))
}
func TestSentryKeyFromRequest_QueryFallback(t *testing.T) {
r := httptest.NewRequest("POST", "/api/acme/envelope/?sentry_key=qkey456", nil)
assert.Equal(t, "qkey456", sentryKeyFromRequest(r))
}
func TestSentryKeyFromRequest_HeaderWins(t *testing.T) {
r := httptest.NewRequest("POST", "/api/acme/envelope/?sentry_key=qkey", nil)
r.Header.Set("X-Sentry-Auth", "Sentry sentry_key=hkey")
assert.Equal(t, "hkey", sentryKeyFromRequest(r))
}
func TestMintDSN(t *testing.T) {
dsn := MintDSN(testSecret, "o11y.hanzo.ai", "acme")
assert.Equal(t, "https://"+publicKeyFor(testSecret, "acme")+"@o11y.hanzo.ai/v1/o11y/acme", dsn)
}
@@ -0,0 +1,138 @@
package implerrortracking
import (
"bytes"
"compress/flate"
"compress/gzip"
"compress/zlib"
"encoding/json"
"io"
"strings"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
)
// maxDecodedBytes caps the decompressed payload so a small gzip bomb cannot
// exhaust memory on this public endpoint. 24 MiB comfortably fits large stack
// traces / batched envelopes while staying bounded.
const maxDecodedBytes = 24 << 20
// maxEventsPerEnvelope caps the events extracted from ONE request, so a single
// envelope cannot fan out into an unbounded number of issue upserts (amplification
// backpressure). Real SDKs send one event per envelope; batching senders stay well
// under this.
const maxEventsPerEnvelope = 1000
// decodeBody transparently inflates gzip / zlib / raw-deflate request bodies (the
// Content-Encodings Sentry SDKs use), bounded by maxDecodedBytes. Identity/unknown
// encodings pass through unchanged.
func decodeBody(body []byte, encoding string) ([]byte, error) {
switch strings.ToLower(strings.TrimSpace(encoding)) {
case "", "identity":
return body, nil
case "gzip", "x-gzip":
r, err := gzip.NewReader(bytes.NewReader(body))
if err != nil {
return nil, err
}
defer func() { _ = r.Close() }()
return io.ReadAll(io.LimitReader(r, maxDecodedBytes))
case "deflate":
if r, err := zlib.NewReader(bytes.NewReader(body)); err == nil {
defer func() { _ = r.Close() }()
return io.ReadAll(io.LimitReader(r, maxDecodedBytes))
}
// Some clients send headerless raw DEFLATE.
fr := flate.NewReader(bytes.NewReader(body))
defer func() { _ = fr.Close() }()
return io.ReadAll(io.LimitReader(fr, maxDecodedBytes))
default:
return body, nil
}
}
// parseStoreBody decodes a legacy `/store/` payload: a single event JSON.
func parseStoreBody(body []byte) ([]*errortrackingtypes.SentryEvent, error) {
body = bytes.TrimSpace(body)
if len(body) == 0 {
return nil, errors.Newf(errors.TypeInvalidInput, errortrackingtypes.ErrCodeErrorTrackingInvalidInput, "empty store payload")
}
var ev errortrackingtypes.SentryEvent
if err := json.Unmarshal(body, &ev); err != nil {
return nil, errors.Wrapf(err, errors.TypeInvalidInput, errortrackingtypes.ErrCodeErrorTrackingInvalidInput, "invalid store event json")
}
return []*errortrackingtypes.SentryEvent{&ev}, nil
}
// parseEnvelope decodes a Sentry envelope and returns every `event`-type item.
// The envelope is newline-framed: a header line, then repeating (item-header,
// payload) pairs where a payload is either length-delimited (per its header) or
// runs to the next newline. Non-event items (transaction/session/attachment/…)
// are skipped. Malformed tails are tolerated — we return what parsed cleanly.
func parseEnvelope(body []byte) ([]*errortrackingtypes.SentryEvent, error) {
pos := 0
readLine := func() ([]byte, bool) {
if pos >= len(body) {
return nil, false
}
if nl := bytes.IndexByte(body[pos:], '\n'); nl >= 0 {
line := body[pos : pos+nl]
pos += nl + 1
return line, true
}
line := body[pos:]
pos = len(body)
return line, true
}
// Envelope header (event_id / dsn / sent_at) — required to be present but not
// otherwise consumed here.
if _, ok := readLine(); !ok {
return nil, errors.Newf(errors.TypeInvalidInput, errortrackingtypes.ErrCodeErrorTrackingInvalidInput, "empty envelope")
}
var events []*errortrackingtypes.SentryEvent
for pos < len(body) {
if len(events) >= maxEventsPerEnvelope {
break // amplification cap: ignore the tail of an oversized envelope
}
hdrLine, ok := readLine()
if !ok {
break
}
if len(bytes.TrimSpace(hdrLine)) == 0 {
continue
}
var ih errortrackingtypes.EnvelopeItemHeader
if err := json.Unmarshal(hdrLine, &ih); err != nil {
break // corrupt framing; stop rather than misread payloads as headers
}
var payload []byte
// Use the declared length ONLY when it is in-bounds — never add an
// attacker-supplied int to pos (a huge/negative length would overflow past
// the clamp and panic the slice). Out-of-range → treat as newline-delimited.
if ih.Length != nil && *ih.Length >= 0 && *ih.Length <= len(body)-pos {
end := pos + *ih.Length
payload = body[pos:end]
pos = end
if pos < len(body) && body[pos] == '\n' {
pos++
}
} else {
payload, ok = readLine()
if !ok {
break
}
}
if ih.Type == "event" {
var ev errortrackingtypes.SentryEvent
if err := json.Unmarshal(payload, &ev); err == nil {
events = append(events, &ev)
}
}
}
return events, nil
}
@@ -0,0 +1,126 @@
package implerrortracking
import (
"bytes"
"compress/gzip"
"compress/zlib"
"fmt"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestParseEnvelope_NewlineDelimitedEvent(t *testing.T) {
body := []byte(`{"event_id":"9ec79c33","dsn":"https://k@h/1"}
{"type":"event"}
{"event_id":"9ec79c33","exception":{"values":[{"type":"ZeroDivisionError","value":"division by zero"}]}}
`)
events, err := parseEnvelope(body)
require.NoError(t, err)
require.Len(t, events, 1)
require.NotNil(t, events[0].Exception)
assert.Equal(t, "ZeroDivisionError", events[0].Exception.Values[0].Type)
}
func TestParseEnvelope_LengthDelimitedEvent(t *testing.T) {
payload := `{"event_id":"abc","exception":{"values":[{"type":"E","value":"boom"}]}}`
body := []byte(fmt.Sprintf("{\"event_id\":\"abc\"}\n{\"type\":\"event\",\"length\":%d}\n%s\n", len(payload), payload))
events, err := parseEnvelope(body)
require.NoError(t, err)
require.Len(t, events, 1)
assert.Equal(t, "boom", events[0].Exception.Values[0].Value)
}
func TestParseEnvelope_SkipsNonEventItems(t *testing.T) {
body := []byte(`{"event_id":"x"}
{"type":"session"}
{"sid":"s1","status":"ok"}
{"type":"event"}
{"event_id":"x","exception":{"values":[{"type":"E"}]}}
{"type":"transaction"}
{"spans":[]}
`)
events, err := parseEnvelope(body)
require.NoError(t, err)
require.Len(t, events, 1, "only the event item is extracted")
assert.Equal(t, "E", events[0].Exception.Values[0].Type)
}
func TestParseEnvelope_MultipleEvents(t *testing.T) {
body := []byte(`{"event_id":"x"}
{"type":"event"}
{"exception":{"values":[{"type":"A"}]}}
{"type":"event"}
{"exception":{"values":[{"type":"B"}]}}
`)
events, err := parseEnvelope(body)
require.NoError(t, err)
require.Len(t, events, 2)
assert.Equal(t, "A", events[0].Exception.Values[0].Type)
assert.Equal(t, "B", events[1].Exception.Values[0].Type)
}
func TestParseEnvelope_EmptyFails(t *testing.T) {
_, err := parseEnvelope(nil)
require.Error(t, err)
}
func TestParseStoreBody(t *testing.T) {
events, err := parseStoreBody([]byte(`{"event_id":"z","exception":{"values":[{"type":"RuntimeError","value":"nope"}]}}`))
require.NoError(t, err)
require.Len(t, events, 1)
assert.Equal(t, "RuntimeError", events[0].Exception.Values[0].Type)
}
func TestParseStoreBody_EmptyFails(t *testing.T) {
_, err := parseStoreBody([]byte(" "))
require.Error(t, err)
}
func TestDecodeBody_Identity(t *testing.T) {
got, err := decodeBody([]byte("hello"), "")
require.NoError(t, err)
assert.Equal(t, "hello", string(got))
}
func TestDecodeBody_Gzip(t *testing.T) {
var buf bytes.Buffer
w := gzip.NewWriter(&buf)
_, _ = w.Write([]byte("compressed payload"))
require.NoError(t, w.Close())
got, err := decodeBody(buf.Bytes(), "gzip")
require.NoError(t, err)
assert.Equal(t, "compressed payload", string(got))
}
func TestDecodeBody_Deflate(t *testing.T) {
var buf bytes.Buffer
w := zlib.NewWriter(&buf)
_, _ = w.Write([]byte("zlib payload"))
require.NoError(t, w.Close())
got, err := decodeBody(buf.Bytes(), "deflate")
require.NoError(t, err)
assert.Equal(t, "zlib payload", string(got))
}
// End-to-end through decode+parse: a gzipped envelope decodes then parses.
func TestDecodeThenParse_GzippedEnvelope(t *testing.T) {
raw := `{"event_id":"x"}
{"type":"event"}
{"exception":{"values":[{"type":"OutOfMemory"}]}}
`
var buf bytes.Buffer
w := gzip.NewWriter(&buf)
_, _ = w.Write([]byte(raw))
require.NoError(t, w.Close())
decoded, err := decodeBody(buf.Bytes(), "gzip")
require.NoError(t, err)
events, err := parseEnvelope(decoded)
require.NoError(t, err)
require.Len(t, events, 1)
assert.Equal(t, "OutOfMemory", events[0].Exception.Values[0].Type)
}
@@ -0,0 +1,184 @@
package implerrortracking
import (
"crypto/sha256"
"encoding/hex"
"regexp"
"strings"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
)
// The grouping algorithm is a from-scratch, deterministic reimplementation of the
// public Sentry grouping MODEL (exception type + normalized crash frame, with a
// message fallback), not a port of any upstream code. It runs at ingest so the
// Issues list is a plain org-scoped SELECT rather than an aggregation over an
// org-less exception table.
// defaultFingerprintToken is the Sentry sentinel that expands to the computed
// default fingerprint inside a client-supplied fingerprint array.
const defaultFingerprintToken = "{{ default }}"
var (
reUUID = regexp.MustCompile(`\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b`)
reHexAddr = regexp.MustCompile(`\b0x[0-9a-fA-F]+\b`)
reLongHex = regexp.MustCompile(`\b[0-9a-fA-F]{8,}\b`)
reNumber = regexp.MustCompile(`\b\d[\d.,_]*\b`)
reQuoted = regexp.MustCompile(`'[^']*'|"[^"]*"`)
reWhitespace = regexp.MustCompile(`\s+`)
reFuncNoise = regexp.MustCompile(`0x[0-9a-fA-F]+`)
)
// computeFingerprint returns the stable 64-hex-char group key for an occurrence.
// A client-supplied fingerprint is honored (the Sentry contract), with the
// "{{ default }}" token expanded to the computed default parts.
func computeFingerprint(occ *errortrackingtypes.Occurrence, custom []string) string {
def := defaultFingerprintParts(occ)
var parts []string
if len(custom) > 0 {
for _, c := range custom {
if strings.TrimSpace(c) == defaultFingerprintToken {
parts = append(parts, def...)
continue
}
parts = append(parts, c)
}
} else {
parts = def
}
return hashParts(parts)
}
// defaultFingerprintParts builds the canonical grouping components: exception
// type + the normalized crash frame, falling back to a normalized message and
// then the transaction, so an occurrence with no useful signal still groups
// stably instead of collapsing every error into one bucket.
func defaultFingerprintParts(occ *errortrackingtypes.Occurrence) []string {
parts := make([]string, 0, 2)
if occ.Type != "" {
parts = append(parts, "type:"+occ.Type)
}
if frame := pickCrashFrame(occ.Frames); frame != nil {
if sig := normalizeFrame(frame); sig != "" {
parts = append(parts, "frame:"+sig)
return parts
}
}
if occ.Value != "" {
parts = append(parts, "value:"+normalizeMessage(occ.Value))
return parts
}
if occ.Transaction != "" {
parts = append(parts, "txn:"+occ.Transaction)
}
if len(parts) == 0 {
parts = append(parts, "level:"+occ.Level)
}
return parts
}
// pickCrashFrame returns the frame the error occurred in: the innermost (last)
// in-app frame if any, else the innermost frame. Sentry orders frames caller→callee,
// so the crash site is the last element.
func pickCrashFrame(frames []errortrackingtypes.Frame) *errortrackingtypes.Frame {
if len(frames) == 0 {
return nil
}
for i := len(frames) - 1; i >= 0; i-- {
if frames[i].InApp {
return &frames[i]
}
}
return &frames[len(frames)-1]
}
// normalizeFrame renders a frame to a host-independent signature: normalized
// function name at its module (or normalized filename). Line/column numbers and
// absolute paths are dropped so the same logical crash site groups across
// deploys, releases and machines.
func normalizeFrame(f *errortrackingtypes.Frame) string {
fn := normalizeFunction(f.Function)
loc := f.Module
if loc == "" {
loc = normalizeFilename(f.Filename)
}
if loc == "" {
loc = normalizeFilename(f.AbsPath)
}
switch {
case fn != "" && loc != "":
return fn + "@" + loc
case fn != "":
return fn
default:
return loc
}
}
func normalizeFunction(fn string) string {
fn = strings.TrimSpace(fn)
if fn == "" {
return ""
}
// Drop runtime address noise inside anonymous/closure names.
fn = reFuncNoise.ReplaceAllString(fn, "")
return strings.TrimSpace(fn)
}
// normalizeFilename keeps the basename and its immediate parent (enough to be
// unique in practice) and masks content-hash / versioned segments so bundled
// asset names like app.9f3a2b1c.js group across builds.
func normalizeFilename(name string) string {
name = strings.TrimSpace(name)
if name == "" {
return ""
}
name = strings.ReplaceAll(name, "\\", "/")
if i := strings.IndexAny(name, "?#"); i >= 0 {
name = name[:i]
}
segs := strings.Split(strings.Trim(name, "/"), "/")
// Keep the last two path segments.
if len(segs) > 2 {
segs = segs[len(segs)-2:]
}
for i, s := range segs {
s = reLongHex.ReplaceAllString(s, "*")
s = reNumber.ReplaceAllString(s, "*")
segs[i] = s
}
return strings.Join(segs, "/")
}
// normalizeMessage collapses variadic detail (ids, numbers, hex, quoted literals)
// so "user 123 missing" and "user 456 missing" land in one issue.
func normalizeMessage(msg string) string {
msg = strings.TrimSpace(msg)
if msg == "" {
return ""
}
msg = reUUID.ReplaceAllString(msg, "<uuid>")
msg = reHexAddr.ReplaceAllString(msg, "<hex>")
msg = reQuoted.ReplaceAllString(msg, "<str>")
msg = reLongHex.ReplaceAllString(msg, "<hex>")
msg = reNumber.ReplaceAllString(msg, "<num>")
msg = reWhitespace.ReplaceAllString(msg, " ")
return strings.TrimSpace(msg)
}
func hashParts(parts []string) string {
h := sha256.New()
for i, p := range parts {
if i > 0 {
h.Write([]byte{0})
}
h.Write([]byte(p))
}
return hex.EncodeToString(h.Sum(nil))
}
@@ -0,0 +1,82 @@
package implerrortracking
import (
"testing"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func frame(fn, module, file string, inApp bool) errortrackingtypes.Frame {
return errortrackingtypes.Frame{Function: fn, Module: module, Filename: file, InApp: inApp}
}
// Same crash site (type + top frame) groups even when the message varies.
func TestFingerprint_GroupsBySameCrashFrame(t *testing.T) {
a := &errortrackingtypes.Occurrence{Type: "ValueError", Value: "id 12 bad", Frames: []errortrackingtypes.Frame{frame("handle", "app.svc", "svc.py", true)}}
b := &errortrackingtypes.Occurrence{Type: "ValueError", Value: "id 999 bad", Frames: []errortrackingtypes.Frame{frame("handle", "app.svc", "svc.py", true)}}
assert.Equal(t, computeFingerprint(a, nil), computeFingerprint(b, nil), "same type+frame must group regardless of message")
}
func TestFingerprint_DistinctForDifferentTypes(t *testing.T) {
a := &errortrackingtypes.Occurrence{Type: "ValueError", Frames: []errortrackingtypes.Frame{frame("handle", "app.svc", "svc.py", true)}}
b := &errortrackingtypes.Occurrence{Type: "KeyError", Frames: []errortrackingtypes.Frame{frame("handle", "app.svc", "svc.py", true)}}
assert.NotEqual(t, computeFingerprint(a, nil), computeFingerprint(b, nil))
}
// The crash frame is the innermost in-app frame (Sentry orders caller→callee).
func TestFingerprint_PicksInnermostInAppFrame(t *testing.T) {
frames := []errortrackingtypes.Frame{
frame("main", "app", "main.go", true),
frame("libcall", "vendor.lib", "lib.go", false),
frame("crashHere", "app.worker", "worker.go", true),
frame("runtimePanic", "runtime", "panic.go", false),
}
got := pickCrashFrame(frames)
require.NotNil(t, got)
assert.Equal(t, "crashHere", got.Function, "innermost in-app frame is the crash site, not the runtime frame")
}
// Message-only errors group by normalized message (numbers/uuids masked).
func TestFingerprint_MessageFallbackMasksVariadic(t *testing.T) {
a := &errortrackingtypes.Occurrence{Type: "Message", Value: "user 123 not found in shard 7"}
b := &errortrackingtypes.Occurrence{Type: "Message", Value: "user 456 not found in shard 9"}
assert.Equal(t, computeFingerprint(a, nil), computeFingerprint(b, nil))
c := &errortrackingtypes.Occurrence{Type: "Message", Value: "totally different failure"}
assert.NotEqual(t, computeFingerprint(a, nil), computeFingerprint(c, nil))
}
func TestFingerprint_CustomHonored(t *testing.T) {
a := &errortrackingtypes.Occurrence{Type: "ValueError", Value: "x"}
b := &errortrackingtypes.Occurrence{Type: "KeyError", Value: "y"}
// Same explicit fingerprint => same group despite different types.
assert.Equal(t, computeFingerprint(a, []string{"my-group"}), computeFingerprint(b, []string{"my-group"}))
assert.NotEqual(t, computeFingerprint(a, []string{"g1"}), computeFingerprint(a, []string{"g2"}))
}
// "{{ default }}" expands to the computed default, so ["{{ default }}", "tenant"]
// subdivides the default group by tenant.
func TestFingerprint_DefaultTokenExpands(t *testing.T) {
occ := &errortrackingtypes.Occurrence{Type: "ValueError", Frames: []errortrackingtypes.Frame{frame("h", "m", "f.go", true)}}
base := computeFingerprint(occ, nil)
withToken := computeFingerprint(occ, []string{defaultFingerprintToken})
assert.Equal(t, base, withToken, "bare {{ default }} equals the default fingerprint")
subdivided := computeFingerprint(occ, []string{defaultFingerprintToken, "tenant-a"})
assert.NotEqual(t, base, subdivided, "adding a component must change the group")
}
func TestNormalizeMessage_Masks(t *testing.T) {
assert.Equal(t, normalizeMessage("id 42 at 0xdeadbeef"), normalizeMessage("id 7 at 0xcafef00d"))
assert.Equal(t,
normalizeMessage("row 550e8400-e29b-41d4-a716-446655440000 gone"),
normalizeMessage("row 550e8400-e29b-41d4-a716-000000000000 gone"),
)
}
func TestFingerprint_IsHex(t *testing.T) {
fp := computeFingerprint(&errortrackingtypes.Occurrence{Type: "E"}, nil)
assert.Len(t, fp, 64)
}
@@ -0,0 +1,243 @@
package implerrortracking
import (
"context"
"io"
"net/http"
"time"
"github.com/gorilla/mux"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/http/binding"
"github.com/hanzoai/o11y/pkg/http/render"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/types/authtypes"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
const (
viewTimeout = 30 * time.Second
writeTimeout = 15 * time.Second
ingestTimeout = 15 * time.Second
// maxCompressedBody bounds the raw request body before decompression; the
// decoded payload is separately bounded by maxDecodedBytes and the event count
// by maxEventsPerEnvelope.
maxCompressedBody = 6 << 20
)
// eventParser turns a decoded ingest body into events; the two wire formats differ
// only here (envelope framing vs a single store event).
type eventParser func([]byte) ([]*errortrackingtypes.SentryEvent, error)
type handler struct {
module errortracking.Module
// ingestSecret is the KMS-sourced platform ingest secret used to verify DSN
// public keys. Empty => ingest is disabled (fail closed), reads still work.
ingestSecret []byte
capturePII bool
revocations RevocationStore
limiter *rateLimiter
}
// NewHandler builds the HTTP surface. ingestSecret is the KMS-synced platform
// error-ingest secret (empty => ingest fails closed 503); capturePII retains
// end-user PII when true (default false = scrub); revocations resolves per-org key
// rotation (nil => none).
func NewHandler(module errortracking.Module, ingestSecret []byte, capturePII bool, revocations RevocationStore) errortracking.Handler {
if revocations == nil {
revocations = NoopRevocations{}
}
return &handler{
module: module,
ingestSecret: ingestSecret,
capturePII: capturePII,
revocations: revocations,
limiter: newRateLimiter(ingestRatePerSec, ingestBurst),
}
}
// --- ingest (public, DSN-authenticated) ---
func (h *handler) EnvelopeIngest(rw http.ResponseWriter, r *http.Request) {
h.ingest(rw, r, parseEnvelope)
}
func (h *handler) StoreIngest(rw http.ResponseWriter, r *http.Request) {
h.ingest(rw, r, parseStoreBody)
}
// ingest is the shared pipeline: enabled-check → resolve org from the DSN project →
// verify the DSN key at its version (rejecting revoked versions, constant-time) →
// per-org rate limit → bounded read+decode → parse (event-count capped) → normalize
// (scrub) → group+upsert the whole batch in one transaction. Every failure fails
// closed and never leaks internal detail to the untrusted client.
func (h *handler) ingest(rw http.ResponseWriter, r *http.Request, parse eventParser) {
ctx, cancel := context.WithTimeout(r.Context(), ingestTimeout)
defer cancel()
if len(h.ingestSecret) == 0 {
http.Error(rw, "error ingest is not configured", http.StatusServiceUnavailable)
return
}
project := mux.Vars(r)["project_id"]
orgID, ok := orgUUIDFromProject(project)
if !ok {
http.Error(rw, "missing project", http.StatusBadRequest)
return
}
minVersion := h.revocations.MinVersion(ctx, orgID)
if !verifyKey(h.ingestSecret, project, sentryKeyFromRequest(r), minVersion) {
// Sentry SDKs treat 401 as "bad DSN" and drop the event (no retry storm).
http.Error(rw, "invalid ingest key", http.StatusUnauthorized)
return
}
if !h.limiter.allow(orgID) {
rw.Header().Set("Retry-After", "1")
http.Error(rw, "rate limited", http.StatusTooManyRequests)
return
}
r.Body = http.MaxBytesReader(rw, r.Body, maxCompressedBody)
raw, err := io.ReadAll(r.Body)
if err != nil {
http.Error(rw, "payload too large", http.StatusRequestEntityTooLarge)
return
}
decoded, err := decodeBody(raw, r.Header.Get("Content-Encoding"))
if err != nil {
http.Error(rw, "cannot decode body", http.StatusBadRequest)
return
}
events, err := parse(decoded)
if err != nil {
http.Error(rw, "invalid payload", http.StatusBadRequest)
return
}
opts := ingestOpts{capturePII: h.capturePII}
occs := make([]*errortrackingtypes.Occurrence, 0, len(events))
lastID := ""
for _, ev := range events {
occ := normalizeEvent(ev, opts)
if occ.Fingerprint == "" {
continue
}
occs = append(occs, occ)
lastID = occ.EventID
}
if _, err := h.module.Ingest(ctx, orgID, occs); err != nil {
// A store failure is ours, not the client's — 500 so the SDK retries.
http.Error(rw, "ingest failed", http.StatusInternalServerError)
return
}
// Sentry SDKs only require 200; echo the last event id for parity.
render.Success(rw, http.StatusOK, map[string]string{"id": lastID})
}
// --- reads (Hanzo IAM authz, org-scoped) ---
func (h *handler) ListIssues(rw http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), viewTimeout)
defer cancel()
orgID, err := orgFromContext(ctx)
if err != nil {
render.Error(rw, err)
return
}
var q errortrackingtypes.IssuesQuery
if err := binding.Query.BindQuery(r.URL.Query(), &q); err != nil {
render.Error(rw, err)
return
}
items, total, err := h.module.ListIssues(ctx, orgID, &q)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, &errortrackingtypes.GettableIssues{
Items: items, Total: total, Offset: clampOffset(q.Offset), Limit: clampLimit(q.Limit),
})
}
func (h *handler) GetIssue(rw http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), viewTimeout)
defer cancel()
orgID, err := orgFromContext(ctx)
if err != nil {
render.Error(rw, err)
return
}
id, err := idFromPath(r)
if err != nil {
render.Error(rw, err)
return
}
issue, err := h.module.GetIssue(ctx, orgID, id)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, issue)
}
func (h *handler) UpdateIssue(rw http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), writeTimeout)
defer cancel()
orgID, err := orgFromContext(ctx)
if err != nil {
render.Error(rw, err)
return
}
id, err := idFromPath(r)
if err != nil {
render.Error(rw, err)
return
}
req := new(errortrackingtypes.UpdateIssue)
if err := binding.JSON.BindBody(r.Body, req); err != nil {
render.Error(rw, err)
return
}
issue, err := h.module.UpdateIssue(ctx, orgID, id, req)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, issue)
}
// --- shared helpers ---
// orgFromContext resolves the caller's org UUID from the gateway-asserted claims.
// It never panics on a malformed claim (a non-UUID org id fails closed as an
// unauthenticated request rather than crashing the handler).
func orgFromContext(ctx context.Context) (valuer.UUID, error) {
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
return valuer.UUID{}, err
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
return valuer.UUID{}, errors.Wrapf(err, errors.TypeUnauthenticated, errortrackingtypes.ErrCodeErrorTrackingUnauthorized, "identity carries no valid org")
}
return orgID, nil
}
func idFromPath(r *http.Request) (valuer.UUID, error) {
id, err := valuer.NewUUID(mux.Vars(r)["id"])
if err != nil {
return valuer.UUID{}, errors.Wrapf(err, errors.TypeInvalidInput, errortrackingtypes.ErrCodeErrorTrackingInvalidInput, "id is not a valid uuid")
}
return id, nil
}
@@ -0,0 +1,120 @@
package implerrortracking
import (
"bytes"
"context"
"net/http"
"net/http/httptest"
"testing"
"github.com/gorilla/mux"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func newIngestFixture(t *testing.T, secret []byte) (errortracking.Handler, errortracking.Module) {
t.Helper()
mod := NewModule(NewStore(newTestStore(t)), NewNoopSink())
return NewHandler(mod, secret, false, nil), mod
}
func envelopeFor(project string) []byte {
return []byte(`{"event_id":"deadbeef","dsn":"https://k@h/` + project + `"}
{"type":"event"}
{"event_id":"deadbeef","platform":"python","exception":{"values":[{"type":"ValueError","value":"bad input","stacktrace":{"frames":[{"function":"handle","module":"app.svc","in_app":true}]}}]}}
`)
}
func ingestReq(project, key string, body []byte) *http.Request {
req := httptest.NewRequest(http.MethodPost, "/api/"+project+"/envelope/", bytes.NewReader(body))
if key != "" {
req.Header.Set("X-Sentry-Auth", "Sentry sentry_version=7, sentry_key="+key)
}
return mux.SetURLVars(req, map[string]string{"project_id": project})
}
func TestIngest_EndToEnd_ValidKeyStoresIssueUnderResolvedOrg(t *testing.T) {
secret := []byte("kms-secret")
h, mod := newIngestFixture(t, secret)
key := publicKeyFor(secret, "acme")
w := httptest.NewRecorder()
h.EnvelopeIngest(w, ingestReq("acme", key, envelopeFor("acme")))
require.Equal(t, http.StatusOK, w.Code)
orgID, _ := orgUUIDFromProject("acme")
list, total, err := mod.ListIssues(context.Background(), orgID, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
require.Equal(t, 1, total)
require.Len(t, list, 1)
assert.Equal(t, "ValueError", list[0].Type)
assert.Equal(t, "bad input", list[0].Value)
assert.Equal(t, "python", list[0].Platform)
}
func TestIngest_RejectsBadKey(t *testing.T) {
secret := []byte("kms-secret")
h, mod := newIngestFixture(t, secret)
w := httptest.NewRecorder()
h.EnvelopeIngest(w, ingestReq("acme", "wrong-key", envelopeFor("acme")))
assert.Equal(t, http.StatusUnauthorized, w.Code)
orgID, _ := orgUUIDFromProject("acme")
_, total, err := mod.ListIssues(context.Background(), orgID, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
assert.Equal(t, 0, total, "a rejected ingest must persist nothing")
}
func TestIngest_MissingKeyRejected(t *testing.T) {
secret := []byte("kms-secret")
h, _ := newIngestFixture(t, secret)
w := httptest.NewRecorder()
h.EnvelopeIngest(w, ingestReq("acme", "", envelopeFor("acme")))
assert.Equal(t, http.StatusUnauthorized, w.Code)
}
// A key minted for org "acme" must not authorize writing to a different project.
func TestIngest_CrossOrgKeyRejected(t *testing.T) {
secret := []byte("kms-secret")
h, mod := newIngestFixture(t, secret)
acmeKey := publicKeyFor(secret, "acme")
w := httptest.NewRecorder()
// Present acme's key but target project "victim".
h.EnvelopeIngest(w, ingestReq("victim", acmeKey, envelopeFor("victim")))
assert.Equal(t, http.StatusUnauthorized, w.Code)
victimOrg, _ := orgUUIDFromProject("victim")
_, total, err := mod.ListIssues(context.Background(), victimOrg, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
assert.Equal(t, 0, total, "acme's key must not write into victim's org")
}
func TestIngest_DisabledWithoutSecret(t *testing.T) {
h, _ := newIngestFixture(t, nil) // no KMS secret => ingest disabled
w := httptest.NewRecorder()
h.EnvelopeIngest(w, ingestReq("acme", "anything", envelopeFor("acme")))
assert.Equal(t, http.StatusServiceUnavailable, w.Code)
}
func TestIngest_LegacyStoreEndpoint(t *testing.T) {
secret := []byte("kms-secret")
h, mod := newIngestFixture(t, secret)
key := publicKeyFor(secret, "acme")
body := []byte(`{"event_id":"1","exception":{"values":[{"type":"KeyError","value":"missing"}]}}`)
req := httptest.NewRequest(http.MethodPost, "/api/acme/store/", bytes.NewReader(body))
req.Header.Set("X-Sentry-Auth", "Sentry sentry_key="+key)
req = mux.SetURLVars(req, map[string]string{"project_id": "acme"})
w := httptest.NewRecorder()
h.StoreIngest(w, req)
require.Equal(t, http.StatusOK, w.Code)
orgID, _ := orgUUIDFromProject("acme")
_, total, err := mod.ListIssues(context.Background(), orgID, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
assert.Equal(t, 1, total)
}
@@ -0,0 +1,257 @@
package implerrortracking
import (
"context"
"fmt"
"strings"
"testing"
"time"
"github.com/hanzoai/o11y/pkg/types"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// --- HIGH-1: ingest amplification is bounded ---
// A flood of identical events collapses to ONE issue with an incremented count —
// not one upsert (or transaction) per event.
func TestIngest_CollapsesDuplicateFingerprints(t *testing.T) {
ctx := context.Background()
mod, orgA, _ := newTestModule(t)
occs := make([]*errortrackingtypes.Occurrence, 0, 5000)
for i := 0; i < 5000; i++ {
occs = append(occs, occ("fp-flood", "TypeError", "boom", time.Now().UTC()))
}
written, err := mod.Ingest(ctx, orgA, occs)
require.NoError(t, err)
assert.Equal(t, 1, written, "5000 identical events must become ONE upsert, not 5000")
list, total, err := mod.ListIssues(ctx, orgA, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
require.Equal(t, 1, total)
assert.Equal(t, int64(5000), list[0].Count, "count still reflects every event")
}
// parseEnvelope refuses to extract more than the per-request cap, so one request
// cannot fan out into unbounded upserts.
func TestParseEnvelope_CapsEventCount(t *testing.T) {
var b strings.Builder
b.WriteString(`{"event_id":"x"}` + "\n")
for i := 0; i < maxEventsPerEnvelope+500; i++ {
b.WriteString(`{"type":"event"}` + "\n")
b.WriteString(`{"exception":{"values":[{"type":"E"}]}}` + "\n")
}
events, err := parseEnvelope([]byte(b.String()))
require.NoError(t, err)
assert.Len(t, events, maxEventsPerEnvelope, "event extraction is capped")
}
// The per-org issue ceiling admits only `ceiling` NEW fingerprints; existing ones
// keep bumping past the cap.
func TestStore_CeilingCapsNewFingerprints(t *testing.T) {
ctx := context.Background()
s := NewStore(newTestStore(t))
org := valuer.GenerateUUID()
mk := func(fp string) *errortrackingtypes.Issue {
now := time.Now().UTC()
return &errortrackingtypes.Issue{
Fingerprint: fp, OrgID: org, Type: "E", Level: "error", Status: errortrackingtypes.StatusUnresolved,
FirstSeen: now, LastSeen: now, Count: 1,
}
}
batch := make([]*errortrackingtypes.Issue, 0, 10)
for i := 0; i < 10; i++ {
iss := mk(fmt.Sprintf("fp-%d", i))
iss.ID = valuer.GenerateUUID()
batch = append(batch, iss)
}
written, err := s.UpsertIssues(ctx, org, batch, 5)
require.NoError(t, err)
assert.Equal(t, 5, written, "only ceiling-many NEW fingerprints are admitted")
_, total, err := s.ListIssues(ctx, org, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
assert.Equal(t, 5, total)
// Re-ingesting the SAME batch: existing 5 bump (no new rows past the cap).
for _, iss := range batch {
iss.ID = valuer.GenerateUUID()
}
_, err = s.UpsertIssues(ctx, org, batch, 5)
require.NoError(t, err)
_, total2, err := s.ListIssues(ctx, org, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
assert.Equal(t, 5, total2, "ceiling still holds; existing issues just bumped")
}
// --- MEDIUM-1: a hostile envelope item length never panics ---
func TestParseEnvelope_HugeLengthNoPanic(t *testing.T) {
body := []byte(`{"event_id":"x"}` + "\n" +
`{"type":"event","length":9223372036854775807}` + "\n" +
`{"exception":{"values":[{"type":"E"}]}}` + "\n")
assert.NotPanics(t, func() {
_, _ = parseEnvelope(body)
}, "a MaxInt64 length must not overflow the slice bound")
}
func TestParseEnvelope_NegativeLengthNoPanic(t *testing.T) {
body := []byte(`{"event_id":"x"}` + "\n" +
`{"type":"event","length":-1}` + "\n" +
`{"exception":{"values":[{"type":"E"}]}}` + "\n")
assert.NotPanics(t, func() {
events, _ := parseEnvelope(body)
// Falls back to newline-delimited framing, so the event is still read.
require.Len(t, events, 1)
})
}
// --- HIGH-1: per-org rate limiter ---
func TestRateLimiter_AllowsBurstThenLimits(t *testing.T) {
l := newRateLimiter(0.0001, 3) // ~no refill within the test window
org := valuer.GenerateUUID()
assert.True(t, l.allow(org))
assert.True(t, l.allow(org))
assert.True(t, l.allow(org))
assert.False(t, l.allow(org), "burst exhausted → limited")
// A different org has its own bucket.
assert.True(t, l.allow(valuer.GenerateUUID()))
}
// --- MEDIUM-2: secret redaction (always) + PII scrub (default) ---
func TestSanitize_AlwaysRedactsSecrets(t *testing.T) {
cases := []string{
"key sk-abcdef0123456789ABCDEF leaked",
"aws AKIAIOSFODNN7EXAMPLE creds",
"Authorization: Bearer abcdef123456ghijkl",
"postgres://user:s3cr3tpw@db:5432/app",
"card 4111 1111 1111 1111 charged",
"hanzo hk-0123456789abcdef0123 token",
}
for _, c := range cases {
// Even with PII capture ON, secrets are still removed.
got := sanitize(c, true)
assert.Contains(t, got, redactedMark, "secret must be redacted: %q -> %q", c, got)
}
}
func TestScrubPII_MasksEmailAndIP_ByDefault(t *testing.T) {
scrubbed := sanitize("user a@b.com from 10.0.0.5 failed", false)
assert.NotContains(t, scrubbed, "a@b.com")
assert.NotContains(t, scrubbed, "10.0.0.5")
assert.Contains(t, scrubbed, emailMark)
assert.Contains(t, scrubbed, ipMark)
// With capture ON, PII is retained (but secrets still go).
kept := sanitize("user a@b.com from 10.0.0.5 failed", true)
assert.Contains(t, kept, "a@b.com")
assert.Contains(t, kept, "10.0.0.5")
}
// The normalizer scrubs by default (fail-secure) — the stored value and the sample
// carry no secret/PII.
func TestNormalize_ScrubsByDefault(t *testing.T) {
e := mustEvent(t, `{"event_id":"a","exception":{"values":[{"type":"AuthError","value":"token sk-DEADBEEFdeadbeef012345 for a@b.com"}]}}`)
occ := normalizeEvent(e) // default opts → scrub
assert.NotContains(t, occ.Value, "sk-DEADBEEFdeadbeef012345")
assert.NotContains(t, occ.Value, "a@b.com")
}
// --- MEDIUM-3: versioned, per-org-revocable DSN keys ---
func TestVerifyKey_VersionedAndRevocation(t *testing.T) {
secret := []byte("kms")
v1 := publicKeyForVersion(secret, "acme", 1)
v2 := publicKeyForVersion(secret, "acme", 2)
require.True(t, strings.HasPrefix(v1, "1:"))
require.True(t, strings.HasPrefix(v2, "2:"))
// Below the org's min-version is rejected; at/above verifies.
assert.True(t, verifyKey(secret, "acme", v1, 0), "v1 valid when nothing revoked")
assert.False(t, verifyKey(secret, "acme", v1, 2), "v1 revoked once min-version is 2")
assert.True(t, verifyKey(secret, "acme", v2, 2), "v2 still valid at min-version 2")
// A malformed version prefix fails closed.
assert.False(t, verifyKey(secret, "acme", "notanumber:"+v1, 0))
assert.False(t, verifyKey(secret, "acme", "0:"+strings.TrimPrefix(v1, "1:"), 0))
}
func TestSQLRevocations_RotateIsolatesOneOrg(t *testing.T) {
ctx := context.Background()
store := newTestStore(t)
r := NewSQLRevocations(store)
orgA := valuer.GenerateUUID()
orgB := valuer.GenerateUUID()
assert.Equal(t, 0, r.MinVersion(ctx, orgA), "default min-version is 0")
require.NoError(t, r.Rotate(ctx, orgA, 2))
assert.Equal(t, 2, r.MinVersion(ctx, orgA), "rotated org sees its new watermark")
assert.Equal(t, 0, r.MinVersion(ctx, orgB), "other orgs are untouched (isolated rotation)")
}
// --- LOW-1: optimistic concurrency on lifecycle update ---
func TestStore_OptimisticUpdateConflict(t *testing.T) {
ctx := context.Background()
mod, orgA, _ := newTestModule(t)
mustIngest(t, mod, ctx, orgA, occ("fp-oc", "E", "x", time.Now().UTC()))
list, _, err := mod.ListIssues(ctx, orgA, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
id := list[0].ID
// Two operators load the SAME version.
first, err := mod.GetIssue(ctx, orgA, id)
require.NoError(t, err)
second, err := mod.GetIssue(ctx, orgA, id)
require.NoError(t, err)
require.Equal(t, first.Issue.Version, second.Issue.Version)
staleVersion := second.Issue.Version
// First operator resolves — succeeds, bumping the row's version.
_, err = mod.UpdateIssue(ctx, orgA, id, &errortrackingtypes.UpdateIssue{Status: strp(string(errortrackingtypes.StatusResolved))})
require.NoError(t, err)
// The second operator's write, carrying the STALE version, must conflict.
stale := second.Issue
stale.Status = errortrackingtypes.StatusIgnored
stale.UpdatedAt = time.Now().UTC()
err = moduleStore(mod).UpdateIssue(ctx, stale, staleVersion)
require.Error(t, err, "a stale-version write must conflict, not clobber")
}
// --- retention/TTL ---
func TestStore_DeleteStale(t *testing.T) {
ctx := context.Background()
s := NewStore(newTestStore(t))
org := valuer.GenerateUUID()
old := &errortrackingtypes.Issue{Identifiable: types.Identifiable{ID: valuer.GenerateUUID()}, OrgID: org, Fingerprint: "old", Type: "E", Level: "error", Status: errortrackingtypes.StatusResolved, FirstSeen: time.Now().Add(-100 * 24 * time.Hour), LastSeen: time.Now().Add(-100 * 24 * time.Hour), Count: 1}
recent := &errortrackingtypes.Issue{Identifiable: types.Identifiable{ID: valuer.GenerateUUID()}, OrgID: org, Fingerprint: "new", Type: "E", Level: "error", Status: errortrackingtypes.StatusUnresolved, FirstSeen: time.Now(), LastSeen: time.Now(), Count: 1}
_, err := s.UpsertIssues(ctx, org, []*errortrackingtypes.Issue{old, recent}, 100)
require.NoError(t, err)
n, err := s.DeleteStale(ctx, time.Now().Add(-90*24*time.Hour))
require.NoError(t, err)
assert.Equal(t, int64(1), n, "only the stale issue is purged")
_, total, err := s.ListIssues(ctx, org, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
assert.Equal(t, 1, total)
}
func strp(s string) *string { return &s }
// moduleStore reaches the concrete module's store for the concurrency test.
func moduleStore(m interface{}) errortrackingtypes.Store {
return m.(*module).store
}
@@ -0,0 +1,98 @@
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
}
@@ -0,0 +1,55 @@
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())
}
@@ -0,0 +1,229 @@
package implerrortracking
import (
"context"
"encoding/json"
"time"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/types"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
const (
// maxIssuesPerOrg caps distinct fingerprints an org may accumulate — backpressure
// against a fingerprint-explosion DoS. New fingerprints past the cap are dropped
// at ingest; existing issues keep counting.
maxIssuesPerOrg = 10000
// retentionSweepInterval is how often the (optional) TTL sweeper runs.
retentionSweepInterval = 6 * time.Hour
)
type module struct {
store errortrackingtypes.Store
sink OccurrenceSink
retention time.Duration
}
// Option configures the module at construction.
type Option func(*module)
// WithRetention enables a background TTL sweep that purges issues whose last_seen
// predates the given age. Zero (the default) disables the sweeper — so tests that
// construct a module never spawn a goroutine.
func WithRetention(d time.Duration) Option {
return func(m *module) { m.retention = d }
}
// NewModule wires the issue store and the (default no-op) occurrence sink. When a
// positive retention is configured it starts the TTL sweeper.
func NewModule(store errortrackingtypes.Store, sink OccurrenceSink, opts ...Option) errortracking.Module {
if sink == nil {
sink = NoopSink{}
}
m := &module{store: store, sink: sink}
for _, o := range opts {
o(m)
}
if m.retention > 0 {
go m.retentionLoop(context.Background())
}
return m
}
// Ingest groups a whole request's occurrences into their issues. Occurrences are
// FIRST collapsed by fingerprint (a flood of identical events becomes one upsert
// with an incremented count) and then written in ONE transaction under the per-org
// ceiling — so a single request can never fan out into a transaction-per-event
// write storm. The occurrence sink is fail-soft (owns its own errors).
func (m *module) Ingest(ctx context.Context, orgID valuer.UUID, occs []*errortrackingtypes.Occurrence) (int, error) {
if orgID.IsZero() {
return 0, errors.Newf(errors.TypeInvalidInput, errortrackingtypes.ErrCodeErrorTrackingInvalidInput, "ingest has no org")
}
groups := aggregateByFingerprint(occs)
if len(groups) == 0 {
return 0, nil
}
now := nowUTC()
issues := make([]*errortrackingtypes.Issue, 0, len(groups))
for fp, g := range groups {
issues = append(issues, issueFromGroup(orgID, fp, g, now))
}
written, err := m.store.UpsertIssues(ctx, orgID, issues, maxIssuesPerOrg)
if err != nil {
return 0, err
}
// Reused occurrence store; bounded to one write per distinct fingerprint and
// fail-soft — the issues are already durable.
for _, g := range groups {
_ = m.sink.Write(ctx, orgID, g.sample)
}
return written, nil
}
func (m *module) ListIssues(ctx context.Context, orgID valuer.UUID, q *errortrackingtypes.IssuesQuery) ([]*errortrackingtypes.Issue, int, error) {
return m.store.ListIssues(ctx, orgID, q)
}
func (m *module) GetIssue(ctx context.Context, orgID, id valuer.UUID) (*errortrackingtypes.GettableIssue, error) {
issue, err := m.store.GetIssue(ctx, orgID, id)
if err != nil {
return nil, err
}
g := &errortrackingtypes.GettableIssue{Issue: issue}
if issue.SampleEvent != "" {
var occ errortrackingtypes.Occurrence
if err := json.Unmarshal([]byte(issue.SampleEvent), &occ); err == nil {
g.LatestEvent = &occ
}
}
return g, nil
}
// UpdateIssue applies a lifecycle transition scoped to the caller's org, with an
// optimistic-concurrency guard on the loaded version so a stale write conflicts
// instead of clobbering a concurrent operator's change.
func (m *module) UpdateIssue(ctx context.Context, orgID, id valuer.UUID, in *errortrackingtypes.UpdateIssue) (*errortrackingtypes.Issue, error) {
if err := in.Validate(); err != nil {
return nil, err
}
issue, err := m.store.GetIssue(ctx, orgID, id)
if err != nil {
return nil, err
}
expectedVersion := issue.Version
now := nowUTC()
if in.Status != nil {
status := errortrackingtypes.IssueStatus(*in.Status)
issue.Status = status
switch status {
case errortrackingtypes.StatusResolved:
issue.ResolvedAt = &now
issue.Regressed = false
case errortrackingtypes.StatusUnresolved:
issue.ResolvedAt = nil
issue.Regressed = false
case errortrackingtypes.StatusIgnored:
// muted: leave resolved_at untouched
}
}
if in.Assignee != nil {
issue.Assignee = *in.Assignee
}
issue.UpdatedAt = now
if err := m.store.UpdateIssue(ctx, issue, expectedVersion); err != nil {
return nil, err
}
issue.Version = expectedVersion + 1
return issue, nil
}
func (m *module) retentionLoop(ctx context.Context) {
t := time.NewTicker(retentionSweepInterval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
_, _ = m.store.DeleteStale(ctx, nowUTC().Add(-m.retention))
}
}
}
// occurrenceGroup is the per-fingerprint rollup of a batch.
type occurrenceGroup struct {
count int64
first time.Time
last time.Time
sample *errortrackingtypes.Occurrence
}
// aggregateByFingerprint collapses a batch to one group per fingerprint, tracking
// count and the first/last timestamps, keeping the latest occurrence as the sample.
func aggregateByFingerprint(occs []*errortrackingtypes.Occurrence) map[string]*occurrenceGroup {
groups := map[string]*occurrenceGroup{}
for _, occ := range occs {
if occ == nil || occ.Fingerprint == "" {
continue
}
g := groups[occ.Fingerprint]
if g == nil {
g = &occurrenceGroup{first: occ.Timestamp, last: occ.Timestamp, sample: occ}
groups[occ.Fingerprint] = g
}
g.count++
ts := occ.Timestamp
if ts.IsZero() {
continue
}
if g.first.IsZero() || ts.Before(g.first) {
g.first = ts
}
if ts.After(g.last) {
g.last = ts
g.sample = occ
}
}
return groups
}
func issueFromGroup(orgID valuer.UUID, fingerprint string, g *occurrenceGroup, now time.Time) *errortrackingtypes.Issue {
first := g.first
if first.IsZero() {
first = now
}
last := g.last
if last.IsZero() {
last = now
}
sample, _ := json.Marshal(g.sample)
return &errortrackingtypes.Issue{
Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
TimeAuditable: types.TimeAuditable{CreatedAt: now, UpdatedAt: now},
OrgID: orgID,
Fingerprint: fingerprint,
Type: firstNonEmpty(g.sample.Type, "Error"),
Value: g.sample.Value,
Culprit: g.sample.Culprit,
Level: firstNonEmpty(g.sample.Level, errortrackingtypes.DefaultLevel),
Platform: g.sample.Platform,
Status: errortrackingtypes.StatusUnresolved,
FirstSeen: first,
LastSeen: last,
Count: g.count,
Environment: g.sample.Environment,
Release: g.sample.Release,
ServiceName: g.sample.ServiceName,
SampleEvent: string(sample),
}
}
@@ -0,0 +1,295 @@
package implerrortracking
import (
"encoding/json"
"strconv"
"strings"
"time"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
)
const (
maxValueLen = 8192
maxCulpritLen = 512
maxFrames = 250
)
// ingestOpts carries the request-scoped ingest policy into the (otherwise pure)
// normalizer — currently just whether to retain end-user PII.
type ingestOpts struct {
capturePII bool
}
// normalizeEvent turns a decoded Sentry event into the canonical Occurrence and
// stamps its fingerprint. It is total: any missing/odd field degrades to a safe
// default rather than erroring, because an ingest endpoint must never 500 on a
// malformed client payload. Secrets are always redacted and PII scrubbed (unless
// capture is enabled) BEFORE the value enters the fingerprint. The default (no
// opts) is fail-secure: scrub.
func normalizeEvent(e *errortrackingtypes.SentryEvent, opts ...ingestOpts) *errortrackingtypes.Occurrence {
var o ingestOpts
if len(opts) > 0 {
o = opts[0]
}
occ := &errortrackingtypes.Occurrence{
EventID: e.EventID,
Level: firstNonEmpty(strings.ToLower(e.Level), errortrackingtypes.DefaultLevel),
Platform: e.Platform,
Timestamp: parseTimestamp(e.Timestamp),
Environment: e.Environment,
Release: e.Release,
ServerName: e.ServerName,
Transaction: e.Transaction,
Tags: parseTags(e.Tags),
}
if val := primaryException(e.Exception); val != nil {
occ.Type = strings.TrimSpace(val.Type)
occ.Value = truncate(val.Value, maxValueLen)
if val.Stacktrace != nil {
occ.Frames = convertFrames(val.Stacktrace.Frames)
}
}
// Message-only event (no exception): the message IS the grouping value.
if occ.Type == "" && occ.Value == "" {
if msg := parseMessage(e.Message); msg != "" {
occ.Type = "Message"
occ.Value = truncate(msg, maxValueLen)
}
}
occ.ServiceName = serviceName(e)
occ.TraceID, occ.SpanID = traceContext(e.Contexts)
occ.User = convertUser(e.User)
occ.Culprit = truncate(culprit(occ, e), maxCulpritLen)
// Redact secrets (always) + PII (unless captured) before the value is hashed, so
// grouping is stable and nothing sensitive reaches the fingerprint or storage.
sanitizeOccurrence(occ, o.capturePII)
occ.Fingerprint = computeFingerprint(occ, e.Fingerprint)
return occ
}
// primaryException returns the thrown exception — the last value with content —
// following the Sentry convention that chained causes precede the raised error.
func primaryException(ex *errortrackingtypes.SentryException) *errortrackingtypes.SentryExceptionValue {
if ex == nil || len(ex.Values) == 0 {
return nil
}
for i := len(ex.Values) - 1; i >= 0; i-- {
v := ex.Values[i]
if v.Type != "" || v.Value != "" || v.Stacktrace != nil {
return &ex.Values[i]
}
}
return &ex.Values[len(ex.Values)-1]
}
func convertFrames(in []errortrackingtypes.SentryFrame) []errortrackingtypes.Frame {
if len(in) > maxFrames {
in = in[len(in)-maxFrames:]
}
out := make([]errortrackingtypes.Frame, 0, len(in))
for _, f := range in {
out = append(out, errortrackingtypes.Frame{
Function: f.Function,
Module: f.Module,
Filename: f.Filename,
AbsPath: f.AbsPath,
Lineno: f.Lineno,
Colno: f.Colno,
InApp: f.InApp != nil && *f.InApp,
})
}
return out
}
func convertUser(u *errortrackingtypes.SentryUser) *errortrackingtypes.EventUser {
if u == nil {
return nil
}
if u.ID == "" && u.Email == "" && u.Username == "" && u.IPAddress == "" {
return nil
}
return &errortrackingtypes.EventUser{ID: u.ID, Email: u.Email, Username: u.Username, IP: u.IPAddress}
}
// culprit is the human-readable location shown on the issue: the transaction if
// set, else the crash frame rendered readably, else the logger.
func culprit(occ *errortrackingtypes.Occurrence, e *errortrackingtypes.SentryEvent) string {
if e.Transaction != "" {
return e.Transaction
}
if f := pickCrashFrame(occ.Frames); f != nil {
loc := firstNonEmpty(f.Module, baseName(f.Filename), baseName(f.AbsPath))
switch {
case f.Function != "" && loc != "":
return f.Function + " in " + loc
case f.Function != "":
return f.Function
default:
return loc
}
}
return e.Logger
}
// serviceName resolves the service the error belongs to: an explicit `server_name`
// tag / the `service_name` tag / the SDK-reported server, defaulting to the SDK name.
func serviceName(e *errortrackingtypes.SentryEvent) string {
tags := parseTags(e.Tags)
if v := tags["service_name"]; v != "" {
return v
}
if v := tags["server_name"]; v != "" {
return v
}
if e.ServerName != "" {
return e.ServerName
}
if e.SDK != nil {
return e.SDK.Name
}
return ""
}
// traceContext extracts distributed-trace linkage so an error can be pivoted to
// its trace in the same o11y plane.
func traceContext(contexts map[string]json.RawMessage) (traceID, spanID string) {
raw, ok := contexts["trace"]
if !ok {
return "", ""
}
var tc struct {
TraceID string `json:"trace_id"`
SpanID string `json:"span_id"`
}
if err := json.Unmarshal(raw, &tc); err != nil {
return "", ""
}
return tc.TraceID, tc.SpanID
}
// parseTimestamp accepts the two shapes the SDKs emit: a unix-seconds number
// (possibly fractional) or an ISO-8601 string. Unparseable → now.
func parseTimestamp(raw json.RawMessage) time.Time {
s := strings.TrimSpace(string(raw))
if s == "" || s == "null" {
return time.Now().UTC()
}
if s[0] == '"' {
var str string
if err := json.Unmarshal(raw, &str); err == nil {
for _, layout := range []string{time.RFC3339Nano, time.RFC3339, "2006-01-02T15:04:05.999999", "2006-01-02T15:04:05"} {
if t, err := time.Parse(layout, str); err == nil {
return t.UTC()
}
}
}
return time.Now().UTC()
}
if f, err := strconv.ParseFloat(s, 64); err == nil && f > 0 {
sec := int64(f)
nsec := int64((f - float64(sec)) * 1e9)
return time.Unix(sec, nsec).UTC()
}
return time.Now().UTC()
}
// parseMessage accepts the top-level message as a string or {message,formatted}.
func parseMessage(raw json.RawMessage) string {
s := strings.TrimSpace(string(raw))
if s == "" || s == "null" {
return ""
}
if s[0] == '"' {
var str string
_ = json.Unmarshal(raw, &str)
return str
}
var m errortrackingtypes.SentryMessage
if err := json.Unmarshal(raw, &m); err == nil {
return firstNonEmpty(m.Formatted, m.Message)
}
return ""
}
// parseTags accepts the two tag encodings: {k:v} or [[k,v],...]. Values are
// coerced to strings; non-scalar values are dropped.
func parseTags(raw json.RawMessage) map[string]string {
s := strings.TrimSpace(string(raw))
if s == "" || s == "null" {
return nil
}
out := map[string]string{}
switch s[0] {
case '{':
var m map[string]any
if err := json.Unmarshal(raw, &m); err != nil {
return nil
}
for k, v := range m {
if sv := scalarString(v); sv != "" {
out[k] = sv
}
}
case '[':
var pairs [][]any
if err := json.Unmarshal(raw, &pairs); err != nil {
return nil
}
for _, p := range pairs {
if len(p) == 2 {
if k := scalarString(p[0]); k != "" {
out[k] = scalarString(p[1])
}
}
}
}
if len(out) == 0 {
return nil
}
return out
}
func scalarString(v any) string {
switch x := v.(type) {
case string:
return x
case float64:
return strconv.FormatFloat(x, 'f', -1, 64)
case bool:
return strconv.FormatBool(x)
case json.Number:
return x.String()
}
return ""
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if v != "" {
return v
}
}
return ""
}
func baseName(path string) string {
path = strings.ReplaceAll(path, "\\", "/")
path = strings.TrimRight(path, "/")
if i := strings.LastIndex(path, "/"); i >= 0 {
return path[i+1:]
}
return path
}
func truncate(s string, max int) string {
if len(s) <= max {
return s
}
return s[:max]
}
@@ -0,0 +1,106 @@
package implerrortracking
import (
"encoding/json"
"testing"
"time"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func mustEvent(t *testing.T, j string) *errortrackingtypes.SentryEvent {
t.Helper()
var e errortrackingtypes.SentryEvent
require.NoError(t, json.Unmarshal([]byte(j), &e))
return &e
}
func TestNormalize_UnixTimestamp(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","timestamp":1700000000.5,"exception":{"values":[{"type":"E","value":"v"}]}}`))
assert.Equal(t, int64(1700000000), occ.Timestamp.Unix())
}
func TestNormalize_ISOTimestamp(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","timestamp":"2023-11-14T22:13:20Z","exception":{"values":[{"type":"E"}]}}`))
assert.Equal(t, 2023, occ.Timestamp.Year())
assert.Equal(t, time.November, occ.Timestamp.Month())
}
func TestNormalize_MissingTimestampDefaultsNow(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","exception":{"values":[{"type":"E"}]}}`))
assert.WithinDuration(t, time.Now().UTC(), occ.Timestamp, 5*time.Second)
}
func TestNormalize_MessageString(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","message":"plain failure"}`))
assert.Equal(t, "Message", occ.Type)
assert.Equal(t, "plain failure", occ.Value)
}
func TestNormalize_MessageObject(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","message":{"message":"raw %s","formatted":"raw boom"}}`))
assert.Equal(t, "raw boom", occ.Value, "formatted preferred over template")
}
func TestNormalize_TagsMap(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","message":"m","tags":{"env":"prod","code":500}}`))
assert.Equal(t, "prod", occ.Tags["env"])
assert.Equal(t, "500", occ.Tags["code"], "numeric tag coerced to string")
}
func TestNormalize_TagsArray(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","message":"m","tags":[["env","staging"],["region","sfo"]]}`))
assert.Equal(t, "staging", occ.Tags["env"])
assert.Equal(t, "sfo", occ.Tags["region"])
}
func TestNormalize_PrimaryExceptionIsLast(t *testing.T) {
// Chained: cause first, thrown last. Sentry treats the last value as primary.
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","exception":{"values":[
{"type":"IOError","value":"disk"},
{"type":"ServiceError","value":"upstream failed","stacktrace":{"frames":[{"function":"call","module":"svc","in_app":true}]}}
]}}`))
assert.Equal(t, "ServiceError", occ.Type)
assert.Equal(t, "upstream failed", occ.Value)
require.Len(t, occ.Frames, 1)
assert.Equal(t, "call", occ.Frames[0].Function)
}
func TestNormalize_CulpritFromTransaction(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","transaction":"GET /users","exception":{"values":[{"type":"E"}]}}`))
assert.Equal(t, "GET /users", occ.Culprit)
}
func TestNormalize_CulpritFromCrashFrame(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","exception":{"values":[{"type":"E","stacktrace":{"frames":[
{"function":"outer","module":"a","in_app":true},
{"function":"boom","filename":"/srv/app/w.py","in_app":true}
]}}]}}`))
assert.Equal(t, "boom in w.py", occ.Culprit)
}
func TestNormalize_LevelDefaultsError(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","exception":{"values":[{"type":"E"}]}}`))
assert.Equal(t, "error", occ.Level)
}
func TestNormalize_TraceContext(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","exception":{"values":[{"type":"E"}]},"contexts":{"trace":{"trace_id":"abc123","span_id":"def456"}}}`))
assert.Equal(t, "abc123", occ.TraceID)
assert.Equal(t, "def456", occ.SpanID)
}
// An empty/garbage event must not panic and must still yield a fingerprint.
func TestNormalize_EmptyEventSafe(t *testing.T) {
occ := normalizeEvent(&errortrackingtypes.SentryEvent{})
require.NotNil(t, occ)
assert.NotEmpty(t, occ.Fingerprint)
assert.Equal(t, "error", occ.Level)
}
func TestNormalize_StampsFingerprint(t *testing.T) {
occ := normalizeEvent(mustEvent(t, `{"event_id":"a","exception":{"values":[{"type":"E","value":"v"}]}}`))
assert.Len(t, occ.Fingerprint, 64)
}
@@ -0,0 +1,95 @@
package implerrortracking
import (
"regexp"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
)
// Error payloads routinely embed secrets (API keys, bearer/JWT tokens, DB
// connection strings, private keys, PANs) and end-user PII (emails, IPs). We treat
// storing those verbatim as a leak-at-rest. Two layers, mirroring the llmobs
// capture-messages precedent (default-secure):
//
// - Secret patterns are ALWAYS redacted — there is no mode in which we persist an
// sk-… key or a password in a DSN. Non-negotiable.
// - PII (email/IP) is scrubbed UNLESS the operator opts in via
// O11Y_ERRORTRACKING_CAPTURE_PII (default false → scrub). Fail-secure.
//
// Redaction runs before the value enters the fingerprint, so two errors that differ
// only by an embedded secret/email still group together.
const (
redactedMark = "[redacted]"
emailMark = "[email]"
ipMark = "[ip]"
)
var secretPatterns = []*regexp.Regexp{
regexp.MustCompile(`-----BEGIN[ A-Z]*PRIVATE KEY-----[\s\S]*?-----END[ A-Z]*PRIVATE KEY-----`),
regexp.MustCompile(`eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{6,}\.[A-Za-z0-9_-]{6,}`), // JWT
regexp.MustCompile(`(?i)\bbearer\s+[A-Za-z0-9._~+/-]{12,}=*`), // bearer token
regexp.MustCompile(`\b(?:sk|pk|rk)-[A-Za-z0-9]{2,}-?[A-Za-z0-9]{12,}`), // openai-style
regexp.MustCompile(`\b(?:sk|pk)_(?:live|test)_[A-Za-z0-9]{16,}`), // stripe
regexp.MustCompile(`\bhk-[A-Za-z0-9]{16,}`), // hanzo key
regexp.MustCompile(`\bAKIA[0-9A-Z]{16}\b`), // aws access key id
regexp.MustCompile(`\bASIA[0-9A-Z]{16}\b`), // aws sts key id
regexp.MustCompile(`\bAIza[0-9A-Za-z_-]{20,}`), // google api key
regexp.MustCompile(`\bgh[posru]_[A-Za-z0-9]{20,}`), // github token
regexp.MustCompile(`\bxox[baprs]-[A-Za-z0-9-]{10,}`), // slack token
regexp.MustCompile(`[a-zA-Z][a-zA-Z0-9+.-]*://[^\s:@/]+:[^\s@/]+@`), // creds in a URL/DSN
regexp.MustCompile(`\b(?:\d[ -]?){13,19}\b`), // PAN-like digit run
}
var (
reEmail = regexp.MustCompile(`[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}`)
reIPv4 = regexp.MustCompile(`\b(?:\d{1,3}\.){3}\d{1,3}\b`)
reIPv6 = regexp.MustCompile(`\b(?:[0-9A-Fa-f]{1,4}:){2,7}[0-9A-Fa-f]{1,4}\b`)
)
// redactSecrets removes known secret shapes. Always applied.
func redactSecrets(s string) string {
for _, re := range secretPatterns {
s = re.ReplaceAllString(s, redactedMark)
}
return s
}
// scrubPII masks emails and IPs. Applied unless PII capture is enabled.
func scrubPII(s string) string {
s = reEmail.ReplaceAllString(s, emailMark)
s = reIPv6.ReplaceAllString(s, ipMark)
s = reIPv4.ReplaceAllString(s, ipMark)
return s
}
// sanitize applies the redaction policy to a free-text field.
func sanitize(s string, capturePII bool) string {
if s == "" {
return s
}
s = redactSecrets(s)
if !capturePII {
s = scrubPII(s)
}
return s
}
// sanitizeOccurrence redacts the fields that carry attacker/user-controlled text.
// Frames (code locations) are left intact; value/tags/user are the leak surface.
func sanitizeOccurrence(occ *errortrackingtypes.Occurrence, capturePII bool) {
occ.Value = sanitize(occ.Value, capturePII)
for k, v := range occ.Tags {
occ.Tags[k] = sanitize(v, capturePII)
}
if occ.User != nil {
occ.User.Email = sanitize(occ.User.Email, capturePII)
occ.User.Username = sanitize(occ.User.Username, capturePII)
if !capturePII {
occ.User.IP = ""
}
if occ.User.ID == "" && occ.User.Email == "" && occ.User.Username == "" && occ.User.IP == "" {
occ.User = nil
}
}
}
@@ -0,0 +1,38 @@
package implerrortracking
import (
"context"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
// OccurrenceSink is the OPTIONAL bridge that would also persist each occurrence to
// the shared telemetry store (o11y_logs as an ERROR-severity record) for OTel-native
// drill-down. It is deliberately a seam, and the default is NoopSink:
//
// - Authoritative storage is o11y_issues — the grouped issue plus the latest
// occurrence sample. Error capture is fully viewable with the no-op sink; the
// sink is enrichment, not the source of truth.
// - When a real sink is wired, its write is fail-soft at the call site: a
// telemetry-store hiccup must never drop the durable issue upsert.
//
// NOTE (honest status): the ClickHouse logs sink is NOT implemented in this build —
// only NoopSink exists. A raw logs_v2 INSERT couples to resource-fingerprint /
// ts-bucket schema that must be byte-verified against a LIVE datastore, not
// reconstructed, so it is a deliberate fast-follow. Today occurrences are NOT
// written to the telemetry store; the count and the latest sample live on the
// issue row.
type OccurrenceSink interface {
Write(ctx context.Context, orgID valuer.UUID, occ *errortrackingtypes.Occurrence) error
}
// NoopSink discards occurrences. It is a complete, correct sink for the MVP: the
// authoritative issue (with its latest sample) is persisted independently.
type NoopSink struct{}
func NewNoopSink() OccurrenceSink { return NoopSink{} }
func (NoopSink) Write(context.Context, valuer.UUID, *errortrackingtypes.Occurrence) error {
return nil
}
@@ -0,0 +1,263 @@
package implerrortracking
import (
"context"
"time"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/sqlstore"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/uptrace/bun"
)
const (
defaultIssueLimit = 50
maxIssueLimit = 100
)
type store struct {
sqlstore sqlstore.SQLStore
}
// NewStore backs the o11y_issues lifecycle table. Every method is org-scoped.
func NewStore(sqlstore sqlstore.SQLStore) errortrackingtypes.Store {
return &store{sqlstore: sqlstore}
}
// UpsertIssues writes a whole envelope's grouped issues in ONE transaction — the
// batch is already collapsed to one row per fingerprint (Count = occurrences in the
// batch), so a request that reports N events causes at most (distinct fingerprints)
// upserts, not N transactions. New fingerprints are admitted only while the org is
// under `ceiling`; existing issues always bump. This bounds single-request write
// amplification and caps per-org issue growth (fingerprint-explosion backpressure).
//
// Portable SET exprs: `count = count + EXCLUDED.count` and `EXCLUDED.x` behave
// identically on SQLite and PostgreSQL; the boolean/enum writes use bound
// placeholders. A separate idempotent statement reopens a RESOLVED issue on
// recurrence (an IGNORED issue stays muted). Ingest never touches `version`.
func (s *store) UpsertIssues(ctx context.Context, orgID valuer.UUID, issues []*errortrackingtypes.Issue, ceiling int) (int, error) {
if len(issues) == 0 {
return 0, nil
}
tx, err := s.sqlstore.BunDB().BeginTx(ctx, nil)
if err != nil {
return 0, err
}
defer func() { _ = tx.Rollback() }()
current, err := tx.NewSelect().Model((*errortrackingtypes.Issue)(nil)).Where("org_id = ?", orgID).Count(ctx)
if err != nil {
return 0, err
}
// Which of the batch's fingerprints already exist for this org (they bump even at
// ceiling); only NEW fingerprints consume headroom.
fps := make([]string, 0, len(issues))
for _, iss := range issues {
fps = append(fps, iss.Fingerprint)
}
existing := map[string]bool{}
var rows []struct {
Fingerprint string `bun:"fingerprint"`
}
if err := tx.NewSelect().
Model((*errortrackingtypes.Issue)(nil)).
Column("fingerprint").
Where("org_id = ?", orgID).
Where("fingerprint IN (?)", bun.In(fps)).
Scan(ctx, &rows); err != nil {
return 0, err
}
for _, r := range rows {
existing[r.Fingerprint] = true
}
headroom := ceiling - current
written := 0
for _, issue := range issues {
if !existing[issue.Fingerprint] {
if headroom <= 0 {
continue // per-org ceiling reached: drop the NEW fingerprint (backpressure)
}
headroom--
existing[issue.Fingerprint] = true
}
if _, err := tx.NewInsert().
Model(issue).
On("CONFLICT (org_id, fingerprint) DO UPDATE").
Set("count = count + EXCLUDED.count").
Set("last_seen = EXCLUDED.last_seen").
Set("value = EXCLUDED.value").
Set("level = EXCLUDED.level").
Set("culprit = EXCLUDED.culprit").
Set("platform = EXCLUDED.platform").
Set("environment = EXCLUDED.environment").
Set("release = EXCLUDED.release").
Set("service_name = EXCLUDED.service_name").
Set("sample_event = EXCLUDED.sample_event").
Set("updated_at = EXCLUDED.updated_at").
Exec(ctx); err != nil {
return 0, err
}
if _, err := tx.NewUpdate().
Model((*errortrackingtypes.Issue)(nil)).
Set("status = ?", errortrackingtypes.StatusUnresolved).
Set("regressed = ?", true).
Set("updated_at = ?", issue.LastSeen).
Where("org_id = ?", issue.OrgID).
Where("fingerprint = ?", issue.Fingerprint).
Where("status = ?", errortrackingtypes.StatusResolved).
Exec(ctx); err != nil {
return 0, err
}
written++
}
if err := tx.Commit(); err != nil {
return 0, err
}
return written, nil
}
func (s *store) ListIssues(ctx context.Context, orgID valuer.UUID, q *errortrackingtypes.IssuesQuery) ([]*errortrackingtypes.Issue, int, error) {
issues := make([]*errortrackingtypes.Issue, 0)
// MANDATORY tenant boundary, first predicate. Every issue row belongs to exactly
// one org; there is no code path that lists issues without this filter.
query := s.sqlstore.
BunDBCtx(ctx).
NewSelect().
Model(&issues).
Where("org_id = ?", orgID)
if q.Status != "" {
query = query.Where("status = ?", q.Status)
}
if q.Level != "" {
query = query.Where("level = ?", q.Level)
}
if q.Environment != "" {
query = query.Where("environment = ?", q.Environment)
}
if q.ServiceName != "" {
query = query.Where("service_name = ?", q.ServiceName)
}
if q.Query != "" {
like := "%" + q.Query + "%"
query = query.Where("(type LIKE ? OR value LIKE ? OR culprit LIKE ?)", like, like, like)
}
count, err := query.
OrderExpr(sortColumn(q.Sort)).
Offset(clampOffset(q.Offset)).
Limit(clampLimit(q.Limit)).
ScanAndCount(ctx)
if err != nil {
return nil, 0, err
}
return issues, count, nil
}
func (s *store) GetIssue(ctx context.Context, orgID, id valuer.UUID) (*errortrackingtypes.Issue, error) {
issue := new(errortrackingtypes.Issue)
err := s.sqlstore.
BunDBCtx(ctx).
NewSelect().
Model(issue).
Where("org_id = ?", orgID).
Where("id = ?", id).
Scan(ctx)
if err != nil {
return nil, s.sqlstore.WrapNotFoundErrf(err, errortrackingtypes.ErrCodeErrorTrackingNotFound, "issue %s not found in the org", id)
}
return issue, nil
}
// UpdateIssue writes the mutable lifecycle columns with an optimistic-concurrency
// guard: the WHERE pins the loaded version, and the write bumps it. Zero rows means
// either the row vanished (not-found) or another operator wrote first (conflict) —
// distinguished by a cheap existence probe so the caller gets the right status.
func (s *store) UpdateIssue(ctx context.Context, issue *errortrackingtypes.Issue, expectedVersion int64) error {
res, err := s.sqlstore.
BunDBCtx(ctx).
NewUpdate().
Model((*errortrackingtypes.Issue)(nil)).
Set("status = ?", issue.Status).
Set("assignee = ?", issue.Assignee).
Set("resolved_at = ?", issue.ResolvedAt).
Set("regressed = ?", issue.Regressed).
Set("updated_at = ?", issue.UpdatedAt).
Set("version = version + 1").
Where("org_id = ?", issue.OrgID).
Where("id = ?", issue.ID).
Where("version = ?", expectedVersion).
Exec(ctx)
if err != nil {
return err
}
n, err := res.RowsAffected()
if err != nil {
return err
}
if n == 0 {
exists, _ := s.sqlstore.BunDBCtx(ctx).NewSelect().
Model((*errortrackingtypes.Issue)(nil)).
Where("org_id = ?", issue.OrgID).
Where("id = ?", issue.ID).
Exists(ctx)
if exists {
return errors.Newf(errors.TypeAlreadyExists, errortrackingtypes.ErrCodeErrorTrackingConflict, "issue %s was modified concurrently; reload and retry", issue.ID)
}
return errors.Newf(errors.TypeNotFound, errortrackingtypes.ErrCodeErrorTrackingNotFound, "issue %s not found in the org", issue.ID)
}
return nil
}
func (s *store) DeleteStale(ctx context.Context, cutoff time.Time) (int64, error) {
res, err := s.sqlstore.
BunDBCtx(ctx).
NewDelete().
Model((*errortrackingtypes.Issue)(nil)).
Where("last_seen < ?", cutoff).
Exec(ctx)
if err != nil {
return 0, err
}
return res.RowsAffected()
}
// sortColumn maps the API sort key to a safe ORDER BY expression (never user text).
func sortColumn(sort string) string {
switch sort {
case "firstSeen":
return "first_seen DESC"
case "count":
return "count DESC"
default:
return "last_seen DESC"
}
}
func clampLimit(limit int) int {
if limit <= 0 {
return defaultIssueLimit
}
if limit > maxIssueLimit {
return maxIssueLimit
}
return limit
}
func clampOffset(offset int) int {
if offset < 0 {
return 0
}
return offset
}
// nowUTC is the single time source for lifecycle writes (overridable in tests).
var nowUTC = func() time.Time { return time.Now().UTC() }
@@ -0,0 +1,199 @@
package implerrortracking
import (
"context"
"path/filepath"
"testing"
"time"
"github.com/hanzoai/o11y/pkg/factory/factorytest"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/sqlstore"
"github.com/hanzoai/o11y/pkg/sqlstore/sqlitesqlstore"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// newTestStore builds a real sqlite store with the o11y_issues table and its
// (org_id, fingerprint) unique index — the shape the migration ships, so ON CONFLICT
// upserts behave exactly as in production.
func newTestStore(t *testing.T) sqlstore.SQLStore {
t.Helper()
dbPath := filepath.Join(t.TempDir(), "test.db")
store, err := sqlitesqlstore.New(context.Background(), factorytest.NewSettings(), sqlstore.Config{
Provider: "sqlite",
Connection: sqlstore.ConnectionConfig{
MaxOpenConns: 1,
MaxConnLifetime: 0,
},
Sqlite: sqlstore.SqliteConfig{
Path: dbPath,
Mode: "wal",
BusyTimeout: 5 * time.Second,
TransactionMode: "deferred",
},
})
require.NoError(t, err)
_, err = store.BunDB().NewCreateTable().
Model((*errortrackingtypes.Issue)(nil)).
IfNotExists().
Exec(context.Background())
require.NoError(t, err)
_, err = store.BunDB().Exec(`CREATE UNIQUE INDEX IF NOT EXISTS uq_o11y_issues_org_fingerprint ON o11y_issues (org_id, fingerprint)`)
require.NoError(t, err)
_, err = store.BunDB().NewCreateTable().
Model((*ingestRevocation)(nil)).
IfNotExists().
Exec(context.Background())
require.NoError(t, err)
return store
}
func newTestModule(t *testing.T) (errortracking.Module, valuer.UUID, valuer.UUID) {
t.Helper()
m := NewModule(NewStore(newTestStore(t)), NewNoopSink())
return m, valuer.GenerateUUID(), valuer.GenerateUUID()
}
func occ(fp, typ, val string, ts time.Time) *errortrackingtypes.Occurrence {
return &errortrackingtypes.Occurrence{
Fingerprint: fp,
Type: typ,
Value: val,
Level: "error",
Timestamp: ts,
EventID: "evt-" + val,
}
}
// mustIngest sends a batch (one or more occurrences) through the module.
func mustIngest(t *testing.T, m errortracking.Module, ctx context.Context, org valuer.UUID, occs ...*errortrackingtypes.Occurrence) {
t.Helper()
_, err := m.Ingest(ctx, org, occs)
require.NoError(t, err)
}
// TestStore_TwoOrgIsolation is the load-bearing tenancy proof: two orgs report the
// SAME fingerprint; each org sees ONLY its own issue, and neither can read/mutate
// the other's issue by id.
func TestStore_TwoOrgIsolation(t *testing.T) {
ctx := context.Background()
mod, orgA, orgB := newTestModule(t)
now := time.Now().UTC()
mustIngest(t, mod, ctx, orgA, occ("fp-shared", "TypeError", "from-A", now))
mustIngest(t, mod, ctx, orgB, occ("fp-shared", "TypeError", "from-B", now))
aList, aTotal, err := mod.ListIssues(ctx, orgA, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
require.Equal(t, 1, aTotal)
require.Len(t, aList, 1)
assert.Equal(t, "from-A", aList[0].Value, "org A must see only its own occurrence value")
assert.Equal(t, orgA, aList[0].OrgID)
bList, bTotal, err := mod.ListIssues(ctx, orgB, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
require.Equal(t, 1, bTotal)
require.Len(t, bList, 1)
assert.Equal(t, "from-B", bList[0].Value, "org B must see only its own occurrence value")
assert.Equal(t, orgB, bList[0].OrgID)
_, err = mod.GetIssue(ctx, orgB, aList[0].ID)
require.Error(t, err, "org B must NOT read org A's issue by id")
reopen := "resolved"
_, err = mod.UpdateIssue(ctx, orgB, aList[0].ID, &errortrackingtypes.UpdateIssue{Status: &reopen})
require.Error(t, err, "org B must NOT update org A's issue")
got, err := mod.GetIssue(ctx, orgA, aList[0].ID)
require.NoError(t, err)
assert.Equal(t, errortrackingtypes.StatusUnresolved, got.Issue.Status)
}
// TestStore_UpsertGroupsByFingerprint proves the fingerprint bucket: repeated
// occurrences of the same (org, fingerprint) collapse into one issue with a running
// count and advancing last-seen, first-seen preserved.
func TestStore_UpsertGroupsByFingerprint(t *testing.T) {
ctx := context.Background()
mod, orgA, _ := newTestModule(t)
first := time.Now().UTC().Add(-time.Hour)
later := time.Now().UTC()
mustIngest(t, mod, ctx, orgA, occ("fp1", "ValueError", "boom", first))
mustIngest(t, mod, ctx, orgA, occ("fp1", "ValueError", "boom", later))
mustIngest(t, mod, ctx, orgA, occ("fp1", "ValueError", "boom", later))
list, total, err := mod.ListIssues(ctx, orgA, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
require.Equal(t, 1, total, "same fingerprint must not create new issues")
require.Len(t, list, 1)
assert.Equal(t, int64(3), list[0].Count, "count must track occurrences")
assert.WithinDuration(t, first, list[0].FirstSeen, time.Second, "first-seen preserved")
assert.WithinDuration(t, later, list[0].LastSeen, time.Second, "last-seen advanced")
mustIngest(t, mod, ctx, orgA, occ("fp2", "KeyError", "missing", later))
_, total2, err := mod.ListIssues(ctx, orgA, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
assert.Equal(t, 2, total2)
}
// TestStore_RegressionReopensResolved proves regression detection: a resolved issue
// that recurs flips back to unresolved and is flagged regressed; an ignored issue
// stays muted.
func TestStore_RegressionReopensResolved(t *testing.T) {
ctx := context.Background()
mod, orgA, _ := newTestModule(t)
now := time.Now().UTC()
mustIngest(t, mod, ctx, orgA, occ("fp-reg", "TypeError", "x", now))
list, _, err := mod.ListIssues(ctx, orgA, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
id := list[0].ID
resolved := string(errortrackingtypes.StatusResolved)
updated, err := mod.UpdateIssue(ctx, orgA, id, &errortrackingtypes.UpdateIssue{Status: &resolved})
require.NoError(t, err)
require.Equal(t, errortrackingtypes.StatusResolved, updated.Status)
require.NotNil(t, updated.ResolvedAt)
mustIngest(t, mod, ctx, orgA, occ("fp-reg", "TypeError", "x", now.Add(time.Minute)))
got, err := mod.GetIssue(ctx, orgA, id)
require.NoError(t, err)
assert.Equal(t, errortrackingtypes.StatusUnresolved, got.Issue.Status, "resolved issue must reopen on recurrence")
assert.True(t, got.Issue.Regressed, "reopened issue must be flagged a regression")
assert.Equal(t, int64(2), got.Issue.Count)
ignored := string(errortrackingtypes.StatusIgnored)
_, err = mod.UpdateIssue(ctx, orgA, id, &errortrackingtypes.UpdateIssue{Status: &ignored})
require.NoError(t, err)
mustIngest(t, mod, ctx, orgA, occ("fp-reg", "TypeError", "x", now.Add(2*time.Minute)))
got, err = mod.GetIssue(ctx, orgA, id)
require.NoError(t, err)
assert.Equal(t, errortrackingtypes.StatusIgnored, got.Issue.Status, "ignored issue must stay muted")
}
// TestModule_GetIssueParsesLatestEvent proves the detail view is fully served from
// SQL (the stored sample), independent of any occurrence-store read path.
func TestModule_GetIssueParsesLatestEvent(t *testing.T) {
ctx := context.Background()
mod, orgA, _ := newTestModule(t)
now := time.Now().UTC()
o := occ("fp-detail", "RuntimeError", "kaput", now)
o.Culprit = "handler in server.go"
mustIngest(t, mod, ctx, orgA, o)
list, _, err := mod.ListIssues(ctx, orgA, &errortrackingtypes.IssuesQuery{})
require.NoError(t, err)
detail, err := mod.GetIssue(ctx, orgA, list[0].ID)
require.NoError(t, err)
require.NotNil(t, detail.LatestEvent, "detail must carry the latest occurrence sample")
assert.Equal(t, "kaput", detail.LatestEvent.Value)
assert.Equal(t, "handler in server.go", detail.LatestEvent.Culprit)
}
+38
View File
@@ -1,6 +1,11 @@
package o11y
import (
"os"
"strconv"
"strings"
"time"
"github.com/hanzoai/o11y/pkg/alertmanager"
"github.com/hanzoai/o11y/pkg/alertmanager/o11yalertmanager"
"github.com/hanzoai/o11y/pkg/analytics"
@@ -18,6 +23,8 @@ import (
"github.com/hanzoai/o11y/pkg/modules/cloudintegration/implcloudintegration"
"github.com/hanzoai/o11y/pkg/modules/dashboard"
"github.com/hanzoai/o11y/pkg/modules/dashboard/impldashboard"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/modules/errortracking/implerrortracking"
"github.com/hanzoai/o11y/pkg/modules/fields"
"github.com/hanzoai/o11y/pkg/modules/fields/implfields"
"github.com/hanzoai/o11y/pkg/modules/inframonitoring"
@@ -87,6 +94,7 @@ type Handlers struct {
RulerHandler ruler.Handler
LLMPricingRuleHandler llmpricingrule.Handler
LLMObsHandler llmobs.Handler
ErrorTrackingHandler errortracking.Handler
StatsHandler statsreporter.Handler
}
@@ -136,6 +144,36 @@ func NewHandlers(
RulerHandler: o11yruler.NewHandler(rulerService),
LLMPricingRuleHandler: impllmpricingrule.NewHandler(modules.LLMPricingRule),
LLMObsHandler: impllmobs.NewHandler(modules.LLMObs),
ErrorTrackingHandler: implerrortracking.NewHandler(modules.ErrorTracking, errorTrackingIngestSecret(), errorTrackingCapturePII(), modules.ErrorTrackingRevocations),
StatsHandler: statsreporter.NewHandler(statsAggregator),
}
}
// errorTrackingIngestSecret is the platform secret used to verify Sentry DSN keys
// on the public error-ingest endpoints. It is sourced from KMS (synced to this env
// var via a KMSSecret CRD) — never committed, never plaintext at rest. When unset,
// the ingest endpoints fail closed (503) while the IAM-scoped read endpoints keep
// working.
func errorTrackingIngestSecret() []byte {
return []byte(os.Getenv("O11Y_ERRORTRACKING_INGEST_SECRET"))
}
// errorTrackingCapturePII reports whether the error-ingest path retains end-user
// PII (email/IP). Default false = scrub (fail-secure), mirroring the llmobs
// O11Y_GENAI_CAPTURE_MESSAGES precedent. Secrets are always redacted regardless.
func errorTrackingCapturePII() bool {
v := strings.ToLower(strings.TrimSpace(os.Getenv("O11Y_ERRORTRACKING_CAPTURE_PII")))
return v == "true" || v == "1" || v == "yes"
}
// errorTrackingRetention is the age past which resolved-or-stale issues are swept.
// Default 90 days; O11Y_ERRORTRACKING_RETENTION_DAYS overrides; 0 disables the sweep.
func errorTrackingRetention() time.Duration {
days := 90
if v := strings.TrimSpace(os.Getenv("O11Y_ERRORTRACKING_RETENTION_DAYS")); v != "" {
if n, err := strconv.Atoi(v); err == nil && n >= 0 {
days = n
}
}
return time.Duration(days) * 24 * time.Hour
}
+14 -2
View File
@@ -15,6 +15,8 @@ import (
"github.com/hanzoai/o11y/pkg/modules/authdomain/implauthdomain"
"github.com/hanzoai/o11y/pkg/modules/cloudintegration"
"github.com/hanzoai/o11y/pkg/modules/dashboard"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/modules/errortracking/implerrortracking"
"github.com/hanzoai/o11y/pkg/modules/inframonitoring"
"github.com/hanzoai/o11y/pkg/modules/inframonitoring/implinframonitoring"
"github.com/hanzoai/o11y/pkg/modules/llmobs"
@@ -97,7 +99,11 @@ type Modules struct {
SpanMapper spanmapper.Module
LLMPricingRule llmpricingrule.Module
LLMObs llmobs.Module
Tag tag.Module
ErrorTracking errortracking.Module
// ErrorTrackingRevocations backs per-org DSN-key rotation; the handler consults
// it on every ingest. Built here because it needs the sqlstore.
ErrorTrackingRevocations implerrortracking.RevocationStore
Tag tag.Module
}
func NewModules(
@@ -165,6 +171,12 @@ func NewModules(
SpanMapper: implspanmapper.NewModule(implspanmapper.NewStore(sqlstore), fl),
LLMPricingRule: impllmpricingrule.NewModule(impllmpricingrule.NewStore(sqlstore), fl),
LLMObs: impllmobs.NewModule(querier, impllmobs.NewStore(sqlstore)),
Tag: tagModule,
ErrorTracking: implerrortracking.NewModule(
implerrortracking.NewStore(sqlstore),
implerrortracking.NewNoopSink(),
implerrortracking.WithRetention(errorTrackingRetention()),
),
ErrorTrackingRevocations: implerrortracking.NewSQLRevocations(sqlstore),
Tag: tagModule,
}
}
+2
View File
@@ -20,6 +20,7 @@ import (
"github.com/hanzoai/o11y/pkg/modules/authdomain"
"github.com/hanzoai/o11y/pkg/modules/cloudintegration"
"github.com/hanzoai/o11y/pkg/modules/dashboard"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/modules/fields"
"github.com/hanzoai/o11y/pkg/modules/inframonitoring"
"github.com/hanzoai/o11y/pkg/modules/llmobs"
@@ -89,6 +90,7 @@ func NewOpenAPI(ctx context.Context, instrumentation instrumentation.Instrumenta
struct{ ruler.Handler }{},
struct{ statsreporter.Handler }{},
struct{ llmobs.Handler }{},
struct{ errortracking.Handler }{},
).New(ctx, instrumentation.ToProviderSettings(), apiserver.Config{})
if err != nil {
return nil, err
+2
View File
@@ -220,6 +220,7 @@ func NewSQLMigrationProviderFactories(
sqlmigration.NewAddMetricReductionRulesFactory(sqlstore, sqlschema),
sqlmigration.NewRemoveOrganizationTuplesFactory(sqlstore),
sqlmigration.NewAddLLMObsFactory(sqlstore, sqlschema),
sqlmigration.NewAddErrorTrackingFactory(sqlstore, sqlschema),
)
}
@@ -319,6 +320,7 @@ func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.Au
handlers.RulerHandler,
handlers.StatsHandler,
handlers.LLMObsHandler,
handlers.ErrorTrackingHandler,
),
)
}
+108
View File
@@ -0,0 +1,108 @@
package sqlmigration
import (
"context"
"github.com/hanzoai/o11y/pkg/factory"
"github.com/hanzoai/o11y/pkg/sqlschema"
"github.com/hanzoai/o11y/pkg/sqlstore"
"github.com/uptrace/bun"
"github.com/uptrace/bun/migrate"
)
// addErrorTracking creates the one net-new table backing error/crash tracking:
// o11y_issues (grouped-error lifecycle). Occurrences stay in the telemetry store
// (o11y_traces / o11y_logs); only non-derivable lifecycle state lives here. The
// unique index on (org_id, fingerprint) is the grouping key the ingest upsert
// conflicts on; (org_id, last_seen) serves the default list ordering.
type addErrorTracking struct {
sqlschema sqlschema.SQLSchema
sqlstore sqlstore.SQLStore
}
func NewAddErrorTrackingFactory(sqlstore sqlstore.SQLStore, sqlschema sqlschema.SQLSchema) factory.ProviderFactory[SQLMigration, Config] {
return factory.NewProviderFactory(factory.MustNewName("add_error_tracking"), func(_ context.Context, _ factory.ProviderSettings, _ Config) (SQLMigration, error) {
return &addErrorTracking{sqlschema: sqlschema, sqlstore: sqlstore}, nil
})
}
func (migration *addErrorTracking) Register(migrations *migrate.Migrations) error {
return migrations.Register(migration.Up, migration.Down)
}
func (migration *addErrorTracking) Up(ctx context.Context, db *bun.DB) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
orgFK := func(col string) *sqlschema.ForeignKeyConstraint {
return &sqlschema.ForeignKeyConstraint{
ReferencingColumnName: sqlschema.ColumnName(col),
ReferencedTableName: sqlschema.TableName("organizations"),
ReferencedColumnName: sqlschema.ColumnName("id"),
}
}
sqls := migration.sqlschema.Operator().CreateTable(&sqlschema.Table{
Name: "o11y_issues",
Columns: []*sqlschema.Column{
{Name: "id", DataType: sqlschema.DataTypeText, Nullable: false},
{Name: "created_at", DataType: sqlschema.DataTypeTimestamp, Nullable: false},
{Name: "updated_at", DataType: sqlschema.DataTypeTimestamp, Nullable: false},
{Name: "org_id", DataType: sqlschema.DataTypeText, Nullable: false},
{Name: "fingerprint", DataType: sqlschema.DataTypeText, Nullable: false},
{Name: "type", DataType: sqlschema.DataTypeText, Nullable: false},
{Name: "value", DataType: sqlschema.DataTypeText, Nullable: true},
{Name: "culprit", DataType: sqlschema.DataTypeText, Nullable: true},
{Name: "level", DataType: sqlschema.DataTypeText, Nullable: false, Default: "'error'"},
{Name: "platform", DataType: sqlschema.DataTypeText, Nullable: true},
{Name: "status", DataType: sqlschema.DataTypeText, Nullable: false, Default: "'unresolved'"},
{Name: "assignee", DataType: sqlschema.DataTypeText, Nullable: true},
{Name: "first_seen", DataType: sqlschema.DataTypeTimestamp, Nullable: false},
{Name: "last_seen", DataType: sqlschema.DataTypeTimestamp, Nullable: false},
{Name: "count", DataType: sqlschema.DataTypeBigInt, Nullable: false, Default: "0"},
{Name: "resolved_at", DataType: sqlschema.DataTypeTimestamp, Nullable: true},
{Name: "regressed", DataType: sqlschema.DataTypeBoolean, Nullable: false, Default: "false"},
{Name: "environment", DataType: sqlschema.DataTypeText, Nullable: true},
{Name: "release", DataType: sqlschema.DataTypeText, Nullable: true},
{Name: "service_name", DataType: sqlschema.DataTypeText, Nullable: true},
{Name: "version", DataType: sqlschema.DataTypeBigInt, Nullable: false, Default: "0"},
{Name: "sample_event", DataType: sqlschema.DataTypeText, Nullable: true},
},
PrimaryKeyConstraint: &sqlschema.PrimaryKeyConstraint{ColumnNames: []sqlschema.ColumnName{"id"}},
ForeignKeyConstraints: []*sqlschema.ForeignKeyConstraint{orgFK("org_id")},
})
// o11y_ingest_revocations: the per-org DSN-key rotation watermark. A DSN key is
// "<version>:<hmac>"; raising min_version for ONE org revokes only that org's
// below-min DSNs — isolated rotation without a global secret roll.
sqls = append(sqls, migration.sqlschema.Operator().CreateTable(&sqlschema.Table{
Name: "o11y_ingest_revocations",
Columns: []*sqlschema.Column{
{Name: "org_id", DataType: sqlschema.DataTypeText, Nullable: false},
{Name: "min_version", DataType: sqlschema.DataTypeBigInt, Nullable: false, Default: "0"},
{Name: "updated_at", DataType: sqlschema.DataTypeTimestamp, Nullable: false},
},
PrimaryKeyConstraint: &sqlschema.PrimaryKeyConstraint{ColumnNames: []sqlschema.ColumnName{"org_id"}},
})...)
// The grouping key (ingest upserts ON CONFLICT here) and the list-ordering index.
sqls = append(sqls,
[]byte(`CREATE UNIQUE INDEX IF NOT EXISTS uq_o11y_issues_org_fingerprint ON o11y_issues (org_id, fingerprint)`),
[]byte(`CREATE INDEX IF NOT EXISTS idx_o11y_issues_org_last_seen ON o11y_issues (org_id, last_seen)`),
)
for _, sql := range sqls {
if _, err := tx.ExecContext(ctx, string(sql)); err != nil {
return err
}
}
return tx.Commit()
}
func (migration *addErrorTracking) Down(context.Context, *bun.DB) error {
return nil
}
+136
View File
@@ -0,0 +1,136 @@
package errortrackingtypes
import (
"time"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/types"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/uptrace/bun"
)
var (
ErrCodeErrorTrackingInvalidInput = errors.MustNewCode("errortracking_invalid_input")
ErrCodeErrorTrackingNotFound = errors.MustNewCode("errortracking_not_found")
ErrCodeErrorTrackingUnauthorized = errors.MustNewCode("errortracking_unauthorized")
ErrCodeErrorTrackingDisabled = errors.MustNewCode("errortracking_disabled")
ErrCodeErrorTrackingConflict = errors.MustNewCode("errortracking_conflict")
)
// IssueStatus is the lifecycle state of an issue (Sentry-class).
type IssueStatus string
const (
StatusUnresolved IssueStatus = "unresolved"
StatusResolved IssueStatus = "resolved"
StatusIgnored IssueStatus = "ignored"
)
// Valid reports whether s is one of the three known lifecycle states.
func (s IssueStatus) Valid() bool {
switch s {
case StatusUnresolved, StatusResolved, StatusIgnored:
return true
}
return false
}
// Default severity level for an issue when the SDK sends none.
const DefaultLevel = "error"
// Issue is the grouped error — a fingerprint bucket. It is the ONE net-new table
// backing error tracking. Occurrences live in the telemetry store (o11y_traces /
// o11y_logs); only the lifecycle state that CANNOT be derived from telemetry —
// status, assignee, first/last-seen, running count, regression — lives here.
// Grouping is done at INGEST (the shim computes the fingerprint), so the Issues
// list is a plain org-scoped SELECT, never an unscoped scan over an org-less
// exception table.
//
// Tenancy: OrgID is the mandatory boundary. It is the o11y org UUID — identical
// to the claims.OrgID the read path derives from the gateway-asserted X-Org-Id,
// and to the UUID the ingest path derives from the DSN project via the SAME
// UUIDv5 mapping — so a row written by ingest is found by exactly one tenant's
// read. Every store query filters `org_id = ?`; there is no code path that reads
// issues across orgs.
type Issue struct {
bun.BaseModel `bun:"table:o11y_issues,alias:o11y_issues" json:"-"`
types.Identifiable
types.TimeAuditable
OrgID valuer.UUID `bun:"org_id,type:text,notnull" json:"-"`
Fingerprint string `bun:"fingerprint,type:text,notnull" json:"fingerprint"`
Type string `bun:"type,type:text,notnull" json:"type"`
Value string `bun:"value,type:text" json:"value"`
Culprit string `bun:"culprit,type:text" json:"culprit"`
Level string `bun:"level,type:text,notnull,default:'error'" json:"level"`
Platform string `bun:"platform,type:text" json:"platform,omitempty"`
Status IssueStatus `bun:"status,type:text,notnull,default:'unresolved'" json:"status"`
Assignee string `bun:"assignee,type:text" json:"assignee,omitempty"`
FirstSeen time.Time `bun:"first_seen,notnull" json:"firstSeen"`
LastSeen time.Time `bun:"last_seen,notnull" json:"lastSeen"`
Count int64 `bun:"count,notnull,default:0" json:"count"`
ResolvedAt *time.Time `bun:"resolved_at" json:"resolvedAt,omitempty"`
Regressed bool `bun:"regressed,notnull,default:false" json:"regressed"`
Environment string `bun:"environment,type:text" json:"environment,omitempty"`
Release string `bun:"release,type:text" json:"release,omitempty"`
ServiceName string `bun:"service_name,type:text" json:"serviceName,omitempty"`
// Version is the optimistic-concurrency guard for lifecycle updates: bumped only
// by UpdateIssue, never by ingest, so an operator's resolve/ignore cannot clobber
// a concurrent operator's write (last-writer-wins) — a stale version is a conflict.
Version int64 `bun:"version,type:bigint,notnull,default:0" json:"-"`
// SampleEvent is the latest normalized Occurrence, stored as JSON so the issue
// detail is fully viewable straight from SQL — no dependency on the (fast-follow)
// occurrence-in-logs read path. Never serialized on the list; parsed into
// GettableIssue.LatestEvent on detail.
SampleEvent string `bun:"sample_event,type:text" json:"-"`
}
// IssuesQuery is the filter for GET /v1/o11y/errortracking/issues. It carries NO
// org field on purpose — the tenant is passed as a separate, server-validated
// argument to the store (mirroring llmobs ScoresQuery), so no client query param
// can widen the scope.
type IssuesQuery struct {
Status string `query:"status" json:"status"`
Level string `query:"level" json:"level"`
Environment string `query:"environment" json:"environment"`
ServiceName string `query:"serviceName" json:"serviceName"`
Query string `query:"query" json:"query"`
Sort string `query:"sort" json:"sort"`
Offset int `query:"offset" json:"offset"`
Limit int `query:"limit" json:"limit"`
}
type GettableIssues struct {
Items []*Issue `json:"items" required:"true"`
Total int `json:"total" required:"true"`
Offset int `json:"offset" required:"true"`
Limit int `json:"limit" required:"true"`
}
// GettableIssue is the issue detail: the lifecycle row plus its latest occurrence
// (parsed from SampleEvent) so the drill-down renders without the occurrence store.
type GettableIssue struct {
Issue *Issue `json:"issue" required:"true"`
LatestEvent *Occurrence `json:"latestEvent,omitempty"`
}
// UpdateIssue is the PATCH body to change lifecycle state (resolve / ignore /
// reopen / assign). Nil fields are left unchanged.
type UpdateIssue struct {
Status *string `json:"status,omitempty"`
Assignee *string `json:"assignee,omitempty"`
}
// Validate enforces the minimal invariants of a lifecycle update.
func (u *UpdateIssue) Validate() error {
if u == nil {
return errors.Newf(errors.TypeInvalidInput, ErrCodeErrorTrackingInvalidInput, "update payload is null")
}
if u.Status != nil && !IssueStatus(*u.Status).Valid() {
return errors.Newf(errors.TypeInvalidInput, ErrCodeErrorTrackingInvalidInput, "invalid status %q (want unresolved|resolved|ignored)", *u.Status)
}
return nil
}
@@ -0,0 +1,50 @@
package errortrackingtypes
import "time"
// Occurrence is a single normalized error event (one exception instance). It is
// derived from a Sentry event (envelope / legacy store item) or, later, from an
// OTel exception span-event. It is the OTel-shaped occurrence the shim persists
// to o11y_logs (the reused occurrence store) and the "latest event" sample kept
// on the issue for the detail view. Purely a value — no store, no tags of its own.
type Occurrence struct {
EventID string `json:"eventId"`
Fingerprint string `json:"fingerprint"`
Type string `json:"type"`
Value string `json:"value"`
Culprit string `json:"culprit"`
Level string `json:"level"`
Platform string `json:"platform,omitempty"`
Timestamp time.Time `json:"timestamp"`
Environment string `json:"environment,omitempty"`
Release string `json:"release,omitempty"`
ServiceName string `json:"serviceName,omitempty"`
ServerName string `json:"serverName,omitempty"`
Transaction string `json:"transaction,omitempty"`
TraceID string `json:"traceId,omitempty"`
SpanID string `json:"spanId,omitempty"`
Frames []Frame `json:"frames,omitempty"`
Tags map[string]string `json:"tags,omitempty"`
User *EventUser `json:"user,omitempty"`
}
// Frame is a single stack frame, normalized from the Sentry frame shape. Frames
// are stored innermost-last (crash site last), matching the Sentry convention.
type Frame struct {
Function string `json:"function,omitempty"`
Module string `json:"module,omitempty"`
Filename string `json:"filename,omitempty"`
AbsPath string `json:"absPath,omitempty"`
Lineno int `json:"lineno,omitempty"`
Colno int `json:"colno,omitempty"`
InApp bool `json:"inApp"`
}
// EventUser is the reporting end-user context (used for "users affected"); PII is
// the caller's responsibility — we store only what the SDK sent.
type EventUser struct {
ID string `json:"id,omitempty"`
Email string `json:"email,omitempty"`
Username string `json:"username,omitempty"`
IP string `json:"ipAddress,omitempty"`
}
+91
View File
@@ -0,0 +1,91 @@
package errortrackingtypes
import "encoding/json"
// The types below are the subset of the Sentry SDK wire payload the shim consumes:
// the JSON of a legacy `/store/` body and of an `event`-type item inside an
// `/envelope/`. They are a from-scratch reimplementation of the PUBLIC, documented
// Sentry ingest protocol (develop.sentry.dev) — no upstream (FSL-licensed) code is
// used. Only the fields error-grouping needs are modeled; everything else is ignored.
// SentryEvent is one error event as sent by any Sentry SDK.
type SentryEvent struct {
EventID string `json:"event_id"`
Timestamp json.RawMessage `json:"timestamp"` // unix-seconds number OR ISO-8601 string
Platform string `json:"platform"`
Level string `json:"level"`
Logger string `json:"logger"`
ServerName string `json:"server_name"`
Environment string `json:"environment"`
Release string `json:"release"`
Transaction string `json:"transaction"`
Fingerprint []string `json:"fingerprint"`
Message json.RawMessage `json:"message"` // string OR {message,formatted,params}
Exception *SentryException `json:"exception"`
Tags json.RawMessage `json:"tags"` // {k:v} OR [[k,v],...]
User *SentryUser `json:"user"`
Contexts map[string]json.RawMessage `json:"contexts"`
SDK *SentrySDK `json:"sdk"`
}
// SentryException wraps one or more exception values (chained exceptions). The
// LAST value is the primary/thrown exception.
type SentryException struct {
Values []SentryExceptionValue `json:"values"`
}
type SentryExceptionValue struct {
Type string `json:"type"`
Value string `json:"value"`
Module string `json:"module"`
Stacktrace *SentryStacktrace `json:"stacktrace"`
}
// SentryStacktrace lists frames oldest-first; the crashing frame is last.
type SentryStacktrace struct {
Frames []SentryFrame `json:"frames"`
}
type SentryFrame struct {
Filename string `json:"filename"`
Function string `json:"function"`
Module string `json:"module"`
AbsPath string `json:"abs_path"`
Lineno int `json:"lineno"`
Colno int `json:"colno"`
InApp *bool `json:"in_app"`
}
type SentryUser struct {
ID string `json:"id"`
Email string `json:"email"`
Username string `json:"username"`
IPAddress string `json:"ip_address"`
}
type SentrySDK struct {
Name string `json:"name"`
Version string `json:"version"`
}
// SentryMessage is the object form of the top-level `message` field.
type SentryMessage struct {
Message string `json:"message"`
Formatted string `json:"formatted"`
}
// EnvelopeHeader is the first line of a Sentry envelope. Only DSN is load-bearing
// (a fallback source for the ingest key when the X-Sentry-Auth header is absent).
type EnvelopeHeader struct {
EventID string `json:"event_id"`
DSN string `json:"dsn"`
SentAt string `json:"sent_at"`
}
// EnvelopeItemHeader precedes each envelope item; Type selects the item and Length
// (when present) frames a binary/opaque payload exactly.
type EnvelopeItemHeader struct {
Type string `json:"type"`
Length *int `json:"length"`
ContentType string `json:"content_type"`
}
+31
View File
@@ -0,0 +1,31 @@
package errortrackingtypes
import (
"context"
"time"
"github.com/hanzoai/o11y/pkg/valuer"
)
// Store persists the one net-new table (o11y_issues). Every method is org-scoped:
// writes stamp org_id, reads filter it. There is deliberately no "list all issues"
// or cross-org accessor.
type Store interface {
// UpsertIssues groups a batch of occurrences (already collapsed to one Issue per
// fingerprint, with Count = occurrences-in-batch) in ONE transaction. New
// fingerprints are admitted only while the org is under `ceiling` (the per-org
// issue cap — backpressure against a fingerprint-explosion DoS); existing issues
// always bump count/last-seen and reopen-on-regression. Returns issues written.
UpsertIssues(ctx context.Context, orgID valuer.UUID, issues []*Issue, ceiling int) (int, error)
ListIssues(ctx context.Context, orgID valuer.UUID, q *IssuesQuery) ([]*Issue, int, error)
GetIssue(ctx context.Context, orgID, id valuer.UUID) (*Issue, error)
// UpdateIssue applies a lifecycle change to one issue scoped by (org_id, id) with
// OPTIMISTIC concurrency: it writes only when the row's version still equals
// expectedVersion, bumping it. A stale version is a conflict, not a silent clobber.
UpdateIssue(ctx context.Context, issue *Issue, expectedVersion int64) error
// DeleteStale purges issues whose last_seen predates cutoff (retention/TTL sweep).
DeleteStale(ctx context.Context, cutoff time.Time) (int64, error)
}