merge: Hanzo Sentry /v1/sentry backend on the o11y+datastore substrate (Red-GO — cross-tenant trace isolation verified)

This commit is contained in:
Hanzo Dev
2026-07-11 12:35:26 -07:00
25 changed files with 3213 additions and 34 deletions
+12
View File
@@ -28,6 +28,7 @@ import (
"github.com/hanzoai/o11y/pkg/modules/promote"
"github.com/hanzoai/o11y/pkg/modules/rawdataexport"
"github.com/hanzoai/o11y/pkg/modules/rulestatehistory"
"github.com/hanzoai/o11y/pkg/modules/sentry"
"github.com/hanzoai/o11y/pkg/modules/serviceaccount"
"github.com/hanzoai/o11y/pkg/modules/session"
"github.com/hanzoai/o11y/pkg/modules/spanmapper"
@@ -77,6 +78,7 @@ type provider struct {
llmPricingRuleHandler llmpricingrule.Handler
llmObsHandler llmobs.Handler
errorTrackingHandler errortracking.Handler
sentryHandler sentry.Handler
statsHandler statsreporter.Handler
}
@@ -114,6 +116,7 @@ func NewFactory(
statsHandler statsreporter.Handler,
llmObsHandler llmobs.Handler,
errorTrackingHandler errortracking.Handler,
sentryHandler sentry.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(
@@ -153,6 +156,7 @@ func NewFactory(
statsHandler,
llmObsHandler,
errorTrackingHandler,
sentryHandler,
)
})
}
@@ -194,6 +198,7 @@ func newProvider(
statsHandler statsreporter.Handler,
llmObsHandler llmobs.Handler,
errorTrackingHandler errortracking.Handler,
sentryHandler sentry.Handler,
) (apiserver.APIServer, error) {
settings := factory.NewScopedProviderSettings(providerSettings, "github.com/hanzoai/o11y/pkg/apiserver/o11yapiserver")
router := mux.NewRouter().UseEncodedPath()
@@ -233,6 +238,7 @@ func newProvider(
llmPricingRuleHandler: llmPricingRuleHandler,
llmObsHandler: llmObsHandler,
errorTrackingHandler: errorTrackingHandler,
sentryHandler: sentryHandler,
statsHandler: statsHandler,
}
@@ -362,6 +368,12 @@ func (provider *provider) AddToRouter(router *mux.Router) error {
return err
}
// Hanzo Sentry product face — clean /v1/sentry/* routes on THIS router (no
// /v1/o11y→/api rewrite). Registered here so its literal paths precede any wildcard.
if err := provider.addSentryRoutes(router); err != nil {
return err
}
if err := provider.addTraceDetailRoutes(router); err != nil {
return err
}
+160
View File
@@ -0,0 +1,160 @@
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"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
)
// uuidPattern constrains the ingest {project} path var to a UUID, so the wildcard
// ingest routes can NEVER shadow a static /v1/sentry resource word (projects, issues,
// discover, events, logs, traces, stats). Combined with static-before-wildcard
// registration order below and the reserved-slug check at project creation, a project
// segment and a resource route are structurally unambiguous.
const uuidPattern = "[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}"
// addSentryRoutes serves Hanzo Sentry — the Sentry-parity product face — under the
// CLEAN /v1/sentry contract (no /api/ segment anywhere). Two families:
//
// - INGEST (public, DSN-authenticated in-handler): POST /v1/sentry/{project}/envelope/
// and /store/. OpenAccess (no IAM): a Sentry SDK presents a DSN key, not a Hanzo
// session; the handler verifies that key against the project's rotation watermark.
// The {project} var is UUID-constrained and these are registered LAST.
// - READ (Hanzo IAM authz, org-scoped): projects, issues, discover, events, logs,
// traces, stats — every one scoped to the caller's org from the validated claims.
//
// These paths are literal /v1/sentry/… on the SAME router the o11y read plane uses, so
// no /v1/o11y→/api rewrite applies (see createPublicServer's /v1/sentry passthrough).
func (provider *provider) addSentryRoutes(router *mux.Router) error {
h := provider.sentryHandler
// STATIC resource routes — registered BEFORE the ingest wildcard.
staticRoutes := []struct {
method string
path string
fn http.HandlerFunc
def handler.OpenAPIDef
}{
{http.MethodGet, "/v1/sentry/projects", provider.authzMiddleware.ViewAccess(h.ListProjects), handler.OpenAPIDef{
ID: "SentryListProjects", Tags: []string{"sentry"}, Summary: "List Sentry projects",
Response: new(sentrytypes.GettableProjects), ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK, SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodPost, "/v1/sentry/projects", provider.authzMiddleware.EditAccess(h.CreateProject), handler.OpenAPIDef{
ID: "SentryCreateProject", Tags: []string{"sentry"}, Summary: "Create a Sentry project",
Request: new(sentrytypes.PostableProject), RequestContentType: "application/json",
Response: new(sentrytypes.GettableProject), ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK, ErrorStatusCodes: []int{http.StatusBadRequest},
SecuritySchemes: newSecuritySchemes(types.RoleEditor),
}},
{http.MethodGet, "/v1/sentry/projects/{id}", provider.authzMiddleware.ViewAccess(h.GetProject), handler.OpenAPIDef{
ID: "SentryGetProject", Tags: []string{"sentry"}, Summary: "Get a Sentry project",
Response: new(sentrytypes.GettableProject), ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK, ErrorStatusCodes: []int{http.StatusNotFound},
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodPost, "/v1/sentry/projects/{id}/keys/rotate", provider.authzMiddleware.EditAccess(h.RotateProjectKey), handler.OpenAPIDef{
ID: "SentryRotateProjectKey", Tags: []string{"sentry"}, Summary: "Rotate a project's DSN key",
Response: new(sentrytypes.GettableProject), ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK, ErrorStatusCodes: []int{http.StatusNotFound},
SecuritySchemes: newSecuritySchemes(types.RoleEditor),
}},
{http.MethodGet, "/v1/sentry/issues", provider.authzMiddleware.ViewAccess(h.ListIssues), handler.OpenAPIDef{
ID: "SentryListIssues", Tags: []string{"sentry"}, Summary: "List error issues",
RequestQuery: new(errortrackingtypes.IssuesQuery), Response: new(errortrackingtypes.GettableIssues),
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodGet, "/v1/sentry/issues/{id}", provider.authzMiddleware.ViewAccess(h.GetIssue), handler.OpenAPIDef{
ID: "SentryGetIssue", Tags: []string{"sentry"}, Summary: "Get an error issue",
Response: new(errortrackingtypes.GettableIssue), ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK, ErrorStatusCodes: []int{http.StatusNotFound},
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodPut, "/v1/sentry/issues/{id}", provider.authzMiddleware.EditAccess(h.UpdateIssue), handler.OpenAPIDef{
ID: "SentryUpdateIssue", Tags: []string{"sentry"}, Summary: "Update an issue's lifecycle",
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),
}},
{http.MethodGet, "/v1/sentry/issues/{id}/events", provider.authzMiddleware.ViewAccess(h.IssueEvents), handler.OpenAPIDef{
ID: "SentryIssueEvents", Tags: []string{"sentry"}, Summary: "List an issue's occurrences",
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodPost, "/v1/sentry/discover", provider.authzMiddleware.ViewAccess(h.Discover), handler.OpenAPIDef{
ID: "SentryDiscover", Tags: []string{"sentry"}, Summary: "Query the events plane",
Request: new(sentrytypes.DiscoverRequest), RequestContentType: "application/json",
Response: new(sentrytypes.DiscoverResult), ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK, ErrorStatusCodes: []int{http.StatusBadRequest},
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodGet, "/v1/sentry/events/{id}", provider.authzMiddleware.ViewAccess(h.GetEvent), handler.OpenAPIDef{
ID: "SentryGetEvent", Tags: []string{"sentry"}, Summary: "Get one error event",
Response: new(sentrytypes.Event), ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK, ErrorStatusCodes: []int{http.StatusNotFound},
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodGet, "/v1/sentry/logs", provider.authzMiddleware.ViewAccess(h.ListLogs), handler.OpenAPIDef{
ID: "SentryListLogs", Tags: []string{"sentry"}, Summary: "List error-event logs",
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest}, SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodGet, "/v1/sentry/traces", provider.authzMiddleware.ViewAccess(h.ListTraces), handler.OpenAPIDef{
ID: "SentryListTraces", Tags: []string{"sentry"}, Summary: "List error-correlated traces",
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest}, SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodGet, "/v1/sentry/traces/{id}", provider.authzMiddleware.ViewAccess(h.GetTrace), handler.OpenAPIDef{
ID: "SentryGetTrace", Tags: []string{"sentry"}, Summary: "Get a trace waterfall",
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest, http.StatusNotFound}, SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
{http.MethodGet, "/v1/sentry/stats", provider.authzMiddleware.ViewAccess(h.Stats), handler.OpenAPIDef{
ID: "SentryStats", Tags: []string{"sentry"}, Summary: "Event-rate timeseries",
ResponseContentType: "application/json", SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest}, SecuritySchemes: newSecuritySchemes(types.RoleViewer),
}},
}
for _, rt := range staticRoutes {
if err := router.Handle(rt.path, handler.New(rt.fn, rt.def)).Methods(rt.method).GetError(); err != nil {
return err
}
}
// INGEST wildcard routes — registered LAST, UUID-constrained project segment.
ingestRoutes := []struct {
path string
fn http.HandlerFunc
def handler.OpenAPIDef
}{
{"/v1/sentry/{project:" + uuidPattern + "}/envelope/", provider.authzMiddleware.OpenAccess(h.EnvelopeIngest), handler.OpenAPIDef{
ID: "SentryIngestEnvelope", Tags: []string{"sentry"}, 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{},
}},
{"/v1/sentry/{project:" + uuidPattern + "}/store/", provider.authzMiddleware.OpenAccess(h.StoreIngest), handler.OpenAPIDef{
ID: "SentryIngestStore", Tags: []string{"sentry"}, 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{},
}},
}
for _, rt := range ingestRoutes {
if err := router.Handle(rt.path, handler.New(rt.fn, rt.def)).Methods(http.MethodPost).GetError(); err != nil {
return err
}
}
return nil
}
@@ -0,0 +1,81 @@
package implerrortracking
import (
"net/http"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
// This file is the DELIBERATE, exported reuse surface of the Sentry-wire ingest
// engine. The Hanzo Sentry product face (pkg/modules/sentry) COMPOSES these
// primitives verbatim — ONE ingest engine, two product faces: errortracking's
// org-keyed DSN path and sentry's project-keyed DSN path. Every symbol here is a
// thin, behavior-preserving alias over the unexported original errortracking's own
// handler already uses, so there is exactly one implementation of decode, parse,
// normalize/scrub, DSN-key derivation and rate limiting. No logic is duplicated and
// the reviewed ingest path stays byte-unchanged.
// DecodeBody decompresses a raw ingest body per Content-Encoding, bounded against a
// decompression bomb (the identical MaxDecodedBytes cap the errortracking path uses).
func DecodeBody(body []byte, encoding string) ([]byte, error) { return decodeBody(body, encoding) }
// ParseEnvelope extracts events from a modern Sentry envelope, capped at
// MaxEventsPerEnvelope so one request cannot fan out unbounded.
func ParseEnvelope(decoded []byte) ([]*errortrackingtypes.SentryEvent, error) {
return parseEnvelope(decoded)
}
// ParseStoreBody extracts the single event of a legacy /store/ body.
func ParseStoreBody(decoded []byte) ([]*errortrackingtypes.SentryEvent, error) {
return parseStoreBody(decoded)
}
// NormalizeEvent turns a decoded Sentry event into the canonical, fingerprinted
// Occurrence. Total (never errors on a malformed client payload) and fail-secure:
// secrets are always redacted and end-user PII is scrubbed unless capturePII is set.
func NormalizeEvent(e *errortrackingtypes.SentryEvent, capturePII bool) *errortrackingtypes.Occurrence {
return normalizeEvent(e, ingestOpts{capturePII: capturePII})
}
// PublicKeyForVersion derives the versioned DSN public key for a DSN path segment.
// The Sentry face passes the PROJECT UUID as the segment (errortracking passes the
// ORG); the HMAC-over-platform-secret derivation is identical.
func PublicKeyForVersion(secret []byte, segment string, version int) string {
return publicKeyForVersion(secret, segment, version)
}
// VerifyKey constant-time verifies a presented "<v>:<hmac>" DSN key for a segment,
// rejecting versions below minVersion (the rotation watermark). Fail-closed on an
// empty secret/key or a malformed/too-low version.
func VerifyKey(secret []byte, segment, presented string, minVersion int) bool {
return verifyKey(secret, segment, presented, minVersion)
}
// SentryKeyFromRequest extracts the presented DSN public key from the Sentry auth
// surface (X-Sentry-Auth header, then ?sentry_key). The envelope-body DSN is never
// trusted as a credential channel.
func SentryKeyFromRequest(r *http.Request) string { return sentryKeyFromRequest(r) }
// RateLimiter is the per-key token-bucket ingest limiter. The Sentry face keys it on
// the project UUID; errortracking keys it on the org UUID.
type RateLimiter = rateLimiter
// NewRateLimiter builds a token-bucket limiter (steady-state events/sec, burst) with
// the SAME policy the errortracking ingest path applies.
func NewRateLimiter(rate, burst float64) *RateLimiter { return newRateLimiter(rate, burst) }
// Allow reports whether the key (org or project UUID) is within its rate budget and
// consumes a token when it is. Buckets are created lazily and only for a key that
// already passed DSN verification.
func (l *rateLimiter) Allow(key valuer.UUID) bool { return l.allow(key) }
// Shared ingest-policy constants, so the Sentry face applies identical budgets and
// payload/event bounds to its own ingest pipeline.
const (
IngestRatePerSec = ingestRatePerSec
IngestBurst = ingestBurst
MaxDecodedBytes = maxDecodedBytes
MaxEventsPerEnvelope = maxEventsPerEnvelope
MaxCompressedBody = maxCompressedBody
)
@@ -150,6 +150,12 @@ func (s *store) ListIssues(ctx context.Context, orgID valuer.UUID, q *errortrack
like := "%" + q.Query + "%"
query = query.Where("(type LIKE ? OR value LIKE ? OR culprit LIKE ?)", like, like, like)
}
// Server-only narrowing to a set of fingerprints (the Sentry project projection).
// Never client-settable (IssuesQuery.Fingerprints has no query tag), still under
// the mandatory org_id filter above.
if len(q.Fingerprints) > 0 {
query = query.Where("fingerprint IN (?)", bun.In(q.Fingerprints))
}
count, err := query.
OrderExpr(sortColumn(q.Sort)).
+21
View File
@@ -0,0 +1,21 @@
package implsentry
import (
"github.com/hanzoai/o11y/pkg/modules/errortracking/implerrortracking"
"github.com/hanzoai/o11y/pkg/valuer"
)
// mintDSN builds the DSN an operator hands to an app to report into a project. The
// key is the reused errortracking derivation keyed on the PROJECT id + version; the
// endpoint is the CLEAN Hanzo route — no /api/ segment:
//
// https://<version>:<hmac>@<host>/v1/sentry/<projectID>
//
// A Sentry SDK given this DSN derives its ingest URL as
// https://<host>/v1/sentry/<projectID>/envelope/ — exactly the route addSentryRoutes
// registers. (Official-SDK builds that hardcode an extra /api/ segment are an
// edge-rewrite compat concern for the gateway, never baked in here.)
func mintDSN(secret []byte, host string, projectID valuer.UUID, version int) string {
key := implerrortracking.PublicKeyForVersion(secret, projectID.String(), version)
return "https://" + key + "@" + host + "/v1/sentry/" + projectID.String()
}
+58
View File
@@ -0,0 +1,58 @@
package implsentry
import (
"strings"
"testing"
"github.com/hanzoai/o11y/pkg/modules/errortracking/implerrortracking"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// extractKey pulls the "<v>:<hmac>" credential out of a minted DSN.
func extractKey(t *testing.T, dsn string) string {
t.Helper()
at := strings.Index(dsn, "@")
require.Greater(t, at, len("https://"))
return dsn[len("https://"):at]
}
func TestMintDSN_ShapeAndVerify(t *testing.T) {
secret := []byte("platform-ingest-secret")
proj := valuer.GenerateUUID()
dsn := mintDSN(secret, "api.hanzo.ai", proj, 1)
// CLEAN path — no /api/ segment.
assert.True(t, strings.HasPrefix(dsn, "https://"))
assert.Contains(t, dsn, "@api.hanzo.ai/v1/sentry/"+proj.String())
assert.NotContains(t, dsn, "/api/")
// The embedded key verifies for THIS project at v1.
key := extractKey(t, dsn)
assert.True(t, implerrortracking.VerifyKey(secret, proj.String(), key, 1))
}
func TestMintDSN_ProjectDomainSeparation(t *testing.T) {
secret := []byte("platform-ingest-secret")
a, b := valuer.GenerateUUID(), valuer.GenerateUUID()
keyA := extractKey(t, mintDSN(secret, "h", a, 1))
// A's key must NOT verify for B — the project id is part of the HMAC domain.
assert.False(t, implerrortracking.VerifyKey(secret, b.String(), keyA, 1))
assert.True(t, implerrortracking.VerifyKey(secret, a.String(), keyA, 1))
}
func TestMintDSN_RotationWatermark(t *testing.T) {
secret := []byte("s")
proj := valuer.GenerateUUID()
v1 := extractKey(t, mintDSN(secret, "h", proj, 1))
v2 := extractKey(t, mintDSN(secret, "h", proj, 2))
// After rotation to v2, a v1 key no longer verifies (minVersion=2); v2 does.
assert.False(t, implerrortracking.VerifyKey(secret, proj.String(), v1, 2))
assert.True(t, implerrortracking.VerifyKey(secret, proj.String(), v2, 2))
// A wrong secret never verifies.
assert.False(t, implerrortracking.VerifyKey([]byte("other"), proj.String(), v2, 1))
}
+310
View File
@@ -0,0 +1,310 @@
package implsentry
import (
"fmt"
"strings"
"time"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
)
// This file is the PURE query layer of the events plane: every function is a
// bytes-and-args builder with no connection, so the security-critical invariants —
// (1) org_id AND project_id are always the leading, BOUND predicates, (2) the time
// window is always bound, (3) every column/aggregation/interval a client can name is
// resolved through a fixed ALLOWLIST and never interpolated — are exhaustively
// unit-testable in isolation. The IO layer (eventstore.go) only executes what these
// return.
// eventColumns is the allowlist mapping an API field name to its physical column.
// A field NOT in this map is rejected — a client field name never reaches the SQL as
// an identifier. All values are constants defined here, never request data.
var eventColumns = map[string]colKind{
"timestamp": {"timestamp", kindTime},
"level": {"level", kindString},
"type": {"type", kindString},
"value": {"value", kindString},
"message": {"message", kindString},
"culprit": {"culprit", kindString},
"fingerprint": {"fingerprint", kindString},
"platform": {"platform", kindString},
"environment": {"environment", kindString},
"release": {"release", kindString},
"service_name": {"service_name", kindString},
"transaction": {"transaction", kindString},
"trace_id": {"trace_id", kindString},
"span_id": {"span_id", kindString},
"server_name": {"server_name", kindString},
"user_id": {"user_id", kindString},
"user_email": {"user_email", kindString},
}
// eventAggs is the aggregation allowlist. Each value is a FIXED expression with no
// client input, so an aggregation can never carry injected SQL. Keys are the API
// names; the alias is the key itself.
var eventAggs = map[string]colKind{
"count": {"count()", kindUint},
"users": {"count(DISTINCT user_id)", kindUint},
"traces": {"count(DISTINCT trace_id)", kindUint},
"fingerprints": {"count(DISTINCT fingerprint)", kindUint},
"first_seen": {"min(timestamp)", kindTime},
"last_seen": {"max(timestamp)", kindTime},
}
// periods maps a relative window token to its duration. Unknown => defaultPeriod.
var periods = map[string]time.Duration{
"1h": time.Hour,
"6h": 6 * time.Hour,
"24h": 24 * time.Hour,
"7d": 7 * 24 * time.Hour,
"14d": 14 * 24 * time.Hour,
"30d": 30 * 24 * time.Hour,
"90d": 90 * 24 * time.Hour,
}
const defaultPeriod = 24 * time.Hour
type kind int
const (
kindString kind = iota
kindTime
kindUint
)
type colKind struct {
expr string // physical column or fixed aggregation expression
kind kind
}
// discoverCol names an output column of a Discover result and the kind used to
// allocate its scan target.
type discoverCol struct {
Name string
Kind kind
}
const (
maxDiscoverGroupBy = 6
maxDiscoverLimit = 1000
defaultDiscoverLim = 100
maxReadLimit = 1000
defaultReadLimit = 100
)
// resolveWindow turns a relative period token into an absolute [from, now] window.
// now is injected so the window is deterministic in tests.
func resolveWindow(period string, now time.Time) sentrytypes.Window {
d, ok := periods[strings.TrimSpace(period)]
if !ok {
d = defaultPeriod
}
return sentrytypes.Window{From: now.Add(-d), To: now}
}
// scope is the mandatory (org, project) + window prefix shared by every read: two
// bound tenant predicates FIRST, then the bound time bounds. It returns the WHERE
// fragment and its args in order, so no read can omit the tenant boundary.
func scope(orgID, projectID string, w sentrytypes.Window) (string, []any) {
return "org_id = ? AND project_id = ? AND timestamp >= ? AND timestamp <= ?",
[]any{orgID, projectID, w.From, w.To}
}
// buildDiscover assembles the columnar aggregation. Returns the SQL, its bound args,
// and the output column descriptors (for typed scanning). Every groupBy/orderBy field
// resolves through eventColumns and every aggregation through eventAggs; anything else
// is an invalid-input error, never interpolated.
func buildDiscover(db, table, orgID, projectID string, req *sentrytypes.DiscoverRequest, w sentrytypes.Window) (string, []any, []discoverCol, error) {
if len(req.GroupBy) > maxDiscoverGroupBy {
return "", nil, nil, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "too many groupBy fields (max %d)", maxDiscoverGroupBy)
}
var selects []string
var cols []discoverCol
byAlias := map[string]bool{}
for _, g := range req.GroupBy {
ck, ok := eventColumns[g]
if !ok {
return "", nil, nil, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "unknown groupBy field %q", g)
}
selects = append(selects, ck.expr+" AS "+g)
cols = append(cols, discoverCol{Name: g, Kind: ck.kind})
byAlias[g] = true
}
aggs := req.Aggregations
if len(aggs) == 0 {
aggs = []string{"count"}
}
for _, a := range aggs {
ck, ok := eventAggs[a]
if !ok {
return "", nil, nil, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "unknown aggregation %q", a)
}
selects = append(selects, ck.expr+" AS "+a)
cols = append(cols, discoverCol{Name: a, Kind: ck.kind})
byAlias[a] = true
}
where, args := scope(orgID, projectID, w)
filterSQL, filterArgs, err := buildFilters(req.Filters)
if err != nil {
return "", nil, nil, err
}
where += filterSQL
args = append(args, filterArgs...)
var sb strings.Builder
fmt.Fprintf(&sb, "SELECT %s FROM %s.%s WHERE %s", strings.Join(selects, ", "), db, table, where)
if len(req.GroupBy) > 0 {
sb.WriteString(" GROUP BY " + strings.Join(req.GroupBy, ", "))
}
if req.OrderBy != "" {
if !byAlias[req.OrderBy] {
return "", nil, nil, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "orderBy %q must be a selected field or aggregation", req.OrderBy)
}
dir := "DESC"
if strings.EqualFold(req.OrderDir, "asc") {
dir = "ASC"
}
sb.WriteString(" ORDER BY " + req.OrderBy + " " + dir)
}
sb.WriteString(" LIMIT ?")
args = append(args, clampLimit(req.Limit, defaultDiscoverLim, maxDiscoverLimit))
return sb.String(), args, cols, nil
}
// buildFilters turns the equality/like predicates into bound clauses. The field is
// resolved through the allowlist; the value is ALWAYS a bound parameter. Op is a
// fixed set. Returns a leading " AND ..." fragment (possibly empty) and its args.
func buildFilters(filters []sentrytypes.DiscoverFilter) (string, []any, error) {
var sb strings.Builder
var args []any
for _, f := range filters {
ck, ok := eventColumns[f.Field]
if !ok {
return "", nil, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "unknown filter field %q", f.Field)
}
switch strings.ToLower(f.Op) {
case "", "eq":
sb.WriteString(" AND " + ck.expr + " = ?")
args = append(args, f.Value)
case "neq":
sb.WriteString(" AND " + ck.expr + " != ?")
args = append(args, f.Value)
case "like":
sb.WriteString(" AND " + ck.expr + " LIKE ?")
args = append(args, "%"+f.Value+"%")
default:
return "", nil, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "unknown filter op %q (want eq|neq|like)", f.Op)
}
}
return sb.String(), args, nil
}
// selectColumns is the fixed projection used by row reads (event detail / issue
// occurrences / logs). Order MUST match scanEvent in eventstore.go.
const selectColumns = "org_id, project_id, event_id, timestamp, received_at, level, type, value, message, culprit, fingerprint, platform, environment, release, service_name, transaction, trace_id, span_id, server_name, user_id, user_email, user_ip, tags, sample"
func buildGetEvent(db, table, orgID, projectID, eventID string) (string, []any) {
// Tenant boundary FIRST (org, then project — a project is the DSN-bearing
// isolation unit), then event id; a foreign org/project/event returns zero rows.
return fmt.Sprintf("SELECT %s FROM %s.%s WHERE org_id = ? AND project_id = ? AND event_id = ? ORDER BY timestamp DESC LIMIT 1", selectColumns, db, table),
[]any{orgID, projectID, eventID}
}
func buildListForFingerprint(db, table, orgID, projectID, fingerprint string, limit int) (string, []any) {
return fmt.Sprintf("SELECT %s FROM %s.%s WHERE org_id = ? AND project_id = ? AND fingerprint = ? ORDER BY timestamp DESC LIMIT ?", selectColumns, db, table),
[]any{orgID, projectID, fingerprint, clampLimit(limit, defaultReadLimit, maxReadLimit)}
}
// buildListForTrace returns the (org, project)-scoped error events that reference a
// trace id — the tenant-safe "errors in this trace" detail. The trace id is
// attacker-influenced (it comes from the ingested event body), so it is ANDed AFTER
// the bound org+project predicates: a caller only ever sees their OWN project's events
// for a trace, never another tenant's spans (the o11y_traces span plane has no general
// org column and is intentionally NOT read here — see the report).
func buildListForTrace(db, table, orgID, projectID, traceID string, limit int) (string, []any) {
return fmt.Sprintf("SELECT %s FROM %s.%s WHERE org_id = ? AND project_id = ? AND trace_id = ? ORDER BY timestamp ASC LIMIT ?", selectColumns, db, table),
[]any{orgID, projectID, traceID, clampLimit(limit, defaultReadLimit, maxReadLimit)}
}
func buildListLogs(db, table, orgID, projectID, query string, w sentrytypes.Window, limit int) (string, []any) {
where, args := scope(orgID, projectID, w)
if strings.TrimSpace(query) != "" {
where += " AND (message LIKE ? OR value LIKE ?)"
like := "%" + query + "%"
args = append(args, like, like)
}
args = append(args, clampLimit(limit, defaultReadLimit, maxReadLimit))
return fmt.Sprintf("SELECT %s FROM %s.%s WHERE %s ORDER BY timestamp DESC LIMIT ?", selectColumns, db, table, where), args
}
func buildDistinctFingerprints(db, table, orgID, projectID string, w sentrytypes.Window) (string, []any) {
where, args := scope(orgID, projectID, w)
return fmt.Sprintf("SELECT DISTINCT fingerprint FROM %s.%s WHERE %s AND fingerprint != ''", db, table, where), args
}
func buildListTraces(db, table, orgID, projectID string, w sentrytypes.Window, limit int) (string, []any) {
where, args := scope(orgID, projectID, w)
args = append(args, clampLimit(limit, defaultReadLimit, maxReadLimit))
return fmt.Sprintf(
"SELECT trace_id, count() AS c, min(timestamp) AS f, max(timestamp) AS l, argMax(sample, timestamp) AS s FROM %s.%s WHERE %s AND trace_id != '' GROUP BY trace_id ORDER BY l DESC LIMIT ?",
db, table, where), args
}
// statsFields is the allowlist selecting the counted subset for the stats timeseries.
// Each value is a fixed WHERE fragment with no client input.
var statsFields = map[string]string{
"": "",
"events": "",
"errors": " AND level IN ('error','fatal')",
"warnings": " AND level = 'warning'",
}
// buildStats assembles the bucketed event-count timeseries. The bucket width is
// derived from the window span over a FIXED ladder (never client input), so the
// INTERVAL literal is a value this code controls.
func buildStats(db, table, orgID, projectID, field string, w sentrytypes.Window) (string, []any, error) {
extra, ok := statsFields[strings.ToLower(strings.TrimSpace(field))]
if !ok {
return "", nil, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "unknown stats field %q", field)
}
bucket := bucketSeconds(w.To.Sub(w.From))
where, args := scope(orgID, projectID, w)
// bucket is a controlled int (bucketSeconds ladder), safe to format as the INTERVAL.
sql := fmt.Sprintf(
"SELECT toStartOfInterval(timestamp, INTERVAL %d SECOND) AS bucket, count() AS c FROM %s.%s WHERE %s%s GROUP BY bucket ORDER BY bucket ASC",
bucket, db, table, where, extra)
return sql, args, nil
}
// bucketSeconds picks a bucket width from a fixed ladder so a timeseries has a
// reasonable number of points regardless of the window.
func bucketSeconds(span time.Duration) int {
switch {
case span <= 2*time.Hour:
return 60 // 1m
case span <= 24*time.Hour:
return 3600 // 1h
case span <= 7*24*time.Hour:
return 6 * 3600 // 6h
default:
return 24 * 3600 // 1d
}
}
func clampLimit(limit, def, max int) int {
if limit <= 0 {
return def
}
if limit > max {
return max
}
return limit
}
@@ -0,0 +1,165 @@
package implsentry
import (
"strings"
"testing"
"time"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func testWindow() sentrytypes.Window {
return sentrytypes.Window{From: time.Unix(1000, 0), To: time.Unix(2000, 0)}
}
// TestScopeIsTenantFirst pins the single most important invariant: every read's WHERE
// prefix binds org_id AND project_id FIRST, then the window — so no read can omit the
// tenant boundary, and the tenant values are always bound params (never interpolated).
func TestScopeIsTenantFirst(t *testing.T) {
where, args := scope("org-A", "proj-1", testWindow())
assert.Equal(t, "org_id = ? AND project_id = ? AND timestamp >= ? AND timestamp <= ?", where)
require.Len(t, args, 4)
assert.Equal(t, "org-A", args[0])
assert.Equal(t, "proj-1", args[1])
}
func TestBuildDiscover_ScopedAndBound(t *testing.T) {
req := &sentrytypes.DiscoverRequest{
GroupBy: []string{"level"},
Aggregations: []string{"count", "users"},
Filters: []sentrytypes.DiscoverFilter{{Field: "environment", Op: "eq", Value: "prod"}},
OrderBy: "count",
OrderDir: "desc",
}
sql, args, cols, err := buildDiscover("o11y_sentry", "o11y_sentry_events", "org-A", "proj-1", req, testWindow())
require.NoError(t, err)
// org + project are the FIRST two bound args, always.
assert.Equal(t, "org-A", args[0])
assert.Equal(t, "proj-1", args[1])
assert.Contains(t, sql, "org_id = ? AND project_id = ?")
// The filter value is bound, never inlined.
assert.Contains(t, sql, "environment = ?")
assert.Contains(t, args, "prod")
// Fixed aggregation expressions, aliased by key.
assert.Contains(t, sql, "count() AS count")
assert.Contains(t, sql, "count(DISTINCT user_id) AS users")
assert.Contains(t, sql, "GROUP BY level")
assert.Contains(t, sql, "ORDER BY count DESC")
assert.Equal(t, []discoverCol{{"level", kindString}, {"count", kindUint}, {"users", kindUint}}, cols)
}
// TestBuildDiscover_RejectsInjection proves no client string reaches the SQL as an
// identifier: any field/aggregation/orderBy outside the allowlist is an error, never
// interpolated.
func TestBuildDiscover_RejectsInjection(t *testing.T) {
injection := "value) FROM o11y_sentry.o11y_sentry_events; DROP TABLE o11y_sentry_events --"
cases := []struct {
name string
req *sentrytypes.DiscoverRequest
}{
{"groupBy", &sentrytypes.DiscoverRequest{GroupBy: []string{injection}}},
{"aggregation", &sentrytypes.DiscoverRequest{Aggregations: []string{injection}}},
{"filter field", &sentrytypes.DiscoverRequest{Filters: []sentrytypes.DiscoverFilter{{Field: injection, Value: "x"}}}},
{"filter op", &sentrytypes.DiscoverRequest{Filters: []sentrytypes.DiscoverFilter{{Field: "level", Op: "; DROP", Value: "x"}}}},
{"orderBy", &sentrytypes.DiscoverRequest{GroupBy: []string{"level"}, OrderBy: injection}},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
sql, args, _, err := buildDiscover("db", "t", "org", "proj", c.req, testWindow())
require.Error(t, err, "malicious %s must be rejected, not built", c.name)
// The injection never becomes SQL: the builder returns an error and NO
// statement/args. (The error message may name the rejected field for the
// operator — that string is never executed.)
assert.Empty(t, sql)
assert.Empty(t, args)
})
}
}
// TestBuildDiscover_MaliciousFilterValueIsBound confirms a value that LOOKS like SQL is
// carried as a bound arg, not spliced into the statement.
func TestBuildDiscover_MaliciousFilterValueIsBound(t *testing.T) {
evil := "'; DROP TABLE o11y_sentry_events; --"
req := &sentrytypes.DiscoverRequest{
GroupBy: []string{"level"},
Filters: []sentrytypes.DiscoverFilter{{Field: "release", Op: "eq", Value: evil}},
}
sql, args, _, err := buildDiscover("db", "t", "org", "proj", req, testWindow())
require.NoError(t, err)
assert.NotContains(t, sql, "DROP TABLE", "value must not appear in the SQL text")
assert.Contains(t, args, evil, "value must be a bound arg")
}
func TestBuildStats_ScopedAndFieldAllowlist(t *testing.T) {
sql, args, err := buildStats("db", "t", "org-A", "proj-1", "errors", testWindow())
require.NoError(t, err)
assert.Equal(t, "org-A", args[0])
assert.Equal(t, "proj-1", args[1])
assert.Contains(t, sql, "org_id = ? AND project_id = ?")
assert.Contains(t, sql, "level IN ('error','fatal')")
_, _, err = buildStats("db", "t", "org", "proj", "sneaky') OR 1=1 --", testWindow())
require.Error(t, err, "unknown stats field must be rejected")
}
func TestRowReads_AreOrgAndProjectScoped(t *testing.T) {
// Event detail is org+project bound (a project is an isolation unit).
get, gArgs := buildGetEvent("db", "t", "org-A", "proj-1", "evt-1")
assert.Contains(t, get, "org_id = ? AND project_id = ? AND event_id = ?")
assert.Equal(t, "org-A", gArgs[0])
assert.Equal(t, "proj-1", gArgs[1])
// Issue occurrences are org+project bound.
fp, fArgs := buildListForFingerprint("db", "t", "org-A", "proj-1", "fp-1", 10)
assert.Contains(t, fp, "org_id = ? AND project_id = ? AND fingerprint = ?")
assert.Equal(t, "org-A", fArgs[0])
assert.Equal(t, "proj-1", fArgs[1])
// Trace detail reads the EVENTS plane (org+project+trace bound) — never o11y_traces.
ft, ftArgs := buildListForTrace("db", "t", "org-A", "proj-1", "trace-xyz", 10)
assert.Contains(t, ft, "org_id = ? AND project_id = ? AND trace_id = ?")
assert.Equal(t, "org-A", ftArgs[0])
assert.Equal(t, "proj-1", ftArgs[1])
assert.Equal(t, "trace-xyz", ftArgs[2])
logs, lArgs := buildListLogs("db", "t", "org-A", "proj-1", "boom", testWindow(), 10)
assert.Contains(t, logs, "org_id = ? AND project_id = ?")
assert.Equal(t, "org-A", lArgs[0])
assert.Equal(t, "proj-1", lArgs[1])
assert.Contains(t, logs, "message LIKE ? OR value LIKE ?")
tr, tArgs := buildListTraces("db", "t", "org-A", "proj-1", testWindow(), 10)
assert.Contains(t, tr, "org_id = ? AND project_id = ?")
assert.Equal(t, "org-A", tArgs[0])
df, dArgs := buildDistinctFingerprints("db", "t", "org-A", "proj-1", testWindow())
assert.Contains(t, df, "org_id = ? AND project_id = ?")
assert.Equal(t, "org-A", dArgs[0])
}
func TestResolveWindow(t *testing.T) {
now := time.Unix(1_000_000, 0)
w := resolveWindow("7d", now)
assert.Equal(t, now, w.To)
assert.Equal(t, now.Add(-7*24*time.Hour), w.From)
// Unknown period falls back to the default (24h), never an unbounded scan.
def := resolveWindow("garbage", now)
assert.Equal(t, now.Add(-defaultPeriod), def.From)
}
func TestClampLimit(t *testing.T) {
assert.Equal(t, 100, clampLimit(0, 100, 1000))
assert.Equal(t, 1000, clampLimit(5000, 100, 1000))
assert.Equal(t, 42, clampLimit(42, 100, 1000))
}
func TestBuildDiscover_TooManyGroupBy(t *testing.T) {
req := &sentrytypes.DiscoverRequest{GroupBy: strings.Split("a,b,c,d,e,f,g", ",")}
_, _, _, err := buildDiscover("db", "t", "org", "proj", req, testWindow())
require.Error(t, err)
}
+367
View File
@@ -0,0 +1,367 @@
package implsentry
import (
"context"
"fmt"
"sync"
"time"
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
"github.com/hanzoai/o11y/pkg/telemetrystore"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
// Default database + table for the columnar events plane. The o11y convention is one
// database per telemetry plane (o11y_traces / o11y_logs / o11y_metrics); the Sentry
// events plane is o11y_sentry, mirroring it.
const (
defaultEventsDB = "o11y_sentry"
defaultEventsTable = "o11y_sentry_events"
)
// insertSQL is the fixed-order INSERT the batch sink appends to. Column order is
// identical to selectColumns / scanEvent so a written row reads back field-for-field.
const insertSQL = "INSERT INTO %s.%s (org_id, project_id, event_id, timestamp, received_at, level, type, value, message, culprit, fingerprint, platform, environment, release, service_name, transaction, trace_id, span_id, server_name, user_id, user_email, user_ip, tags, sample)"
// createSchemaDDL is the events-plane schema.
//
// HONEST STATUS — NOT LIVE-VERIFIED. This DDL was designed against the o11y datastore
// conventions and is exercised by the clickhouse-go-mock round-trip test, but it has
// NOT been byte-verified against a live datastore in this build (no datastore was
// reachable — localhost:9000 closed, no O11Y_DATASTORE_DSN). Two things a live run
// must confirm before this is called done:
// 1. the datastore accepts this exact DDL (types, INDEX/TTL syntax);
// 2. on a MULTI-SHARD datastore this plain MergeTree must become the o11y
// local + Distributed(ON CLUSTER) split (the distributed_* convention) so a read
// on any node sees all shards. It is correct as-is only for a single-shard /
// replicated topology, which is why the engine + database are configurable
// (WithDatabase/WithEngine) — the follow-on is config, not a rewrite.
const createSchemaDDL = `CREATE TABLE IF NOT EXISTS %s.%s (
org_id String,
project_id String,
event_id String,
timestamp DateTime64(9),
received_at DateTime64(3),
level LowCardinality(String),
type LowCardinality(String),
value String,
message String,
culprit String,
fingerprint String,
platform LowCardinality(String),
environment LowCardinality(String),
release String,
service_name LowCardinality(String),
transaction String,
trace_id String,
span_id String,
server_name String,
user_id String,
user_email String,
user_ip String,
tags Map(String, String),
sample String,
INDEX idx_fingerprint fingerprint TYPE bloom_filter GRANULARITY 4,
INDEX idx_trace_id trace_id TYPE bloom_filter GRANULARITY 4
) ENGINE = %s
PARTITION BY toDate(timestamp)
ORDER BY (org_id, project_id, timestamp)
TTL toDateTime(timestamp) + INTERVAL %d DAY
SETTINGS index_granularity = 8192`
const defaultEngine = "MergeTree"
const defaultRetentionDays = 30
// eventStore is the datastore-backed EventStore. It ensures its schema lazily (once,
// retried until it succeeds) so it works identically in the standalone runtime and the
// cloud embed with no boot wiring, and touches the datastore only when actually used.
type eventStore struct {
store telemetrystore.TelemetryStore
db string
table string
engine string
retentionDays int
now func() time.Time
ensureMu sync.Mutex
ensureDone bool
}
// Option configures the event store.
type Option func(*eventStore)
func WithDatabase(db string) Option { return func(s *eventStore) { s.db = db } }
func WithTable(t string) Option { return func(s *eventStore) { s.table = t } }
func WithEngine(e string) Option { return func(s *eventStore) { s.engine = e } }
func WithRetentionDays(d int) Option {
return func(s *eventStore) {
if d > 0 {
s.retentionDays = d
}
}
}
// NewEventStore builds the events plane over the shared datastore connection.
func NewEventStore(store telemetrystore.TelemetryStore, opts ...Option) sentrytypes.EventStore {
s := &eventStore{
store: store,
db: defaultEventsDB,
table: defaultEventsTable,
engine: defaultEngine,
retentionDays: defaultRetentionDays,
now: func() time.Time { return time.Now().UTC() },
}
for _, o := range opts {
o(s)
}
return s
}
// ensureSchema creates the database + table idempotently. It caches success so the
// DDL runs once; a failure is returned (and retried next call) so a transient datastore
// outage does not permanently disable the plane.
func (s *eventStore) ensureSchema(ctx context.Context) error {
s.ensureMu.Lock()
defer s.ensureMu.Unlock()
if s.ensureDone {
return nil
}
conn := s.store.ClickhouseDB()
if err := conn.Exec(ctx, fmt.Sprintf("CREATE DATABASE IF NOT EXISTS %s", s.db)); err != nil {
return err
}
if err := conn.Exec(ctx, fmt.Sprintf(createSchemaDDL, s.db, s.table, s.engine, s.retentionDays)); err != nil {
return err
}
s.ensureDone = true
return nil
}
func (s *eventStore) Insert(ctx context.Context, orgID, projectID valuer.UUID, events []*sentrytypes.Event) error {
if len(events) == 0 {
return nil
}
if err := s.ensureSchema(ctx); err != nil {
return err
}
batch, err := s.store.ClickhouseDB().PrepareBatch(ctx,
fmt.Sprintf(insertSQL, s.db, s.table), driver.WithReleaseConnection())
if err != nil {
return err
}
defer func() { _ = batch.Abort() }()
received := s.now()
org, proj := orgID.String(), projectID.String()
for _, e := range events {
if err := batch.Append(
org, proj, e.EventID, e.Timestamp, received,
e.Level, e.Type, e.Value, e.Message, e.Culprit, e.Fingerprint,
e.Platform, e.Environment, e.Release, e.ServiceName, e.Transaction,
e.TraceID, e.SpanID, e.ServerName, e.UserID, e.UserEmail, e.UserIP,
mapOrEmpty(e.Tags), e.Sample,
); err != nil {
return err
}
}
return batch.Send()
}
func (s *eventStore) Discover(ctx context.Context, orgID, projectID valuer.UUID, req *sentrytypes.DiscoverRequest, w sentrytypes.Window) (*sentrytypes.DiscoverResult, error) {
if err := s.ensureSchema(ctx); err != nil {
return nil, err
}
sql, args, cols, err := buildDiscover(s.db, s.table, orgID.String(), projectID.String(), req, w)
if err != nil {
return nil, err
}
rows, err := s.store.ClickhouseDB().Query(ctx, sql, args...)
if err != nil {
return nil, err
}
defer rows.Close()
out := &sentrytypes.DiscoverResult{Columns: make([]string, len(cols))}
for i, c := range cols {
out.Columns[i] = c.Name
}
for rows.Next() {
dest, boxes := scanTargets(cols)
if err := rows.Scan(dest...); err != nil {
return nil, err
}
out.Rows = append(out.Rows, boxes())
}
return out, rows.Err()
}
func (s *eventStore) GetEvent(ctx context.Context, orgID, projectID valuer.UUID, eventID string) (*sentrytypes.Event, error) {
if err := s.ensureSchema(ctx); err != nil {
return nil, err
}
sql, args := buildGetEvent(s.db, s.table, orgID.String(), projectID.String(), eventID)
events, err := s.queryEvents(ctx, sql, args)
if err != nil || len(events) == 0 {
return nil, err
}
return events[0], nil
}
func (s *eventStore) ListForFingerprint(ctx context.Context, orgID, projectID valuer.UUID, fingerprint string, limit int) ([]*sentrytypes.Event, error) {
if err := s.ensureSchema(ctx); err != nil {
return nil, err
}
sql, args := buildListForFingerprint(s.db, s.table, orgID.String(), projectID.String(), fingerprint, limit)
return s.queryEvents(ctx, sql, args)
}
func (s *eventStore) ListForTrace(ctx context.Context, orgID, projectID valuer.UUID, traceID string, limit int) ([]*sentrytypes.Event, error) {
if traceID == "" {
return nil, nil
}
if err := s.ensureSchema(ctx); err != nil {
return nil, err
}
sql, args := buildListForTrace(s.db, s.table, orgID.String(), projectID.String(), traceID, limit)
return s.queryEvents(ctx, sql, args)
}
func (s *eventStore) ListLogs(ctx context.Context, orgID, projectID valuer.UUID, query string, w sentrytypes.Window, limit int) ([]*sentrytypes.Event, error) {
if err := s.ensureSchema(ctx); err != nil {
return nil, err
}
sql, args := buildListLogs(s.db, s.table, orgID.String(), projectID.String(), query, w, limit)
return s.queryEvents(ctx, sql, args)
}
func (s *eventStore) DistinctFingerprints(ctx context.Context, orgID, projectID valuer.UUID, w sentrytypes.Window) ([]string, error) {
if err := s.ensureSchema(ctx); err != nil {
return nil, err
}
sql, args := buildDistinctFingerprints(s.db, s.table, orgID.String(), projectID.String(), w)
rows, err := s.store.ClickhouseDB().Query(ctx, sql, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var fps []string
for rows.Next() {
var fp string
if err := rows.Scan(&fp); err != nil {
return nil, err
}
fps = append(fps, fp)
}
return fps, rows.Err()
}
func (s *eventStore) ListTraces(ctx context.Context, orgID, projectID valuer.UUID, w sentrytypes.Window, limit int) ([]*sentrytypes.TraceSummary, error) {
if err := s.ensureSchema(ctx); err != nil {
return nil, err
}
sql, args := buildListTraces(s.db, s.table, orgID.String(), projectID.String(), w, limit)
rows, err := s.store.ClickhouseDB().Query(ctx, sql, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []*sentrytypes.TraceSummary
for rows.Next() {
t := new(sentrytypes.TraceSummary)
if err := rows.Scan(&t.TraceID, &t.Count, &t.FirstSeen, &t.LastSeen, &t.Sample); err != nil {
return nil, err
}
out = append(out, t)
}
return out, rows.Err()
}
func (s *eventStore) Stats(ctx context.Context, orgID, projectID valuer.UUID, field string, w sentrytypes.Window) ([]sentrytypes.StatsPoint, error) {
if err := s.ensureSchema(ctx); err != nil {
return nil, err
}
sql, args, err := buildStats(s.db, s.table, orgID.String(), projectID.String(), field, w)
if err != nil {
return nil, err
}
rows, err := s.store.ClickhouseDB().Query(ctx, sql, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []sentrytypes.StatsPoint
for rows.Next() {
var p sentrytypes.StatsPoint
if err := rows.Scan(&p.Time, &p.Value); err != nil {
return nil, err
}
out = append(out, p)
}
return out, rows.Err()
}
// queryEvents runs a fixed-projection (selectColumns) query and scans rows into Events.
func (s *eventStore) queryEvents(ctx context.Context, sql string, args []any) ([]*sentrytypes.Event, error) {
rows, err := s.store.ClickhouseDB().Query(ctx, sql, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []*sentrytypes.Event
for rows.Next() {
e := new(sentrytypes.Event)
if e.Tags == nil {
e.Tags = map[string]string{}
}
if err := rows.Scan(
&e.OrgID, &e.ProjectID, &e.EventID, &e.Timestamp, &e.ReceivedAt,
&e.Level, &e.Type, &e.Value, &e.Message, &e.Culprit, &e.Fingerprint,
&e.Platform, &e.Environment, &e.Release, &e.ServiceName, &e.Transaction,
&e.TraceID, &e.SpanID, &e.ServerName, &e.UserID, &e.UserEmail, &e.UserIP,
&e.Tags, &e.Sample,
); err != nil {
return nil, err
}
out = append(out, e)
}
return out, rows.Err()
}
// scanTargets allocates a typed scan destination per Discover output column and
// returns the destinations plus a boxing closure that reads their values into []any
// (the untyped result row).
func scanTargets(cols []discoverCol) ([]any, func() []any) {
dest := make([]any, len(cols))
for i, c := range cols {
switch c.Kind {
case kindTime:
dest[i] = new(time.Time)
case kindUint:
dest[i] = new(uint64)
default:
dest[i] = new(string)
}
}
return dest, func() []any {
row := make([]any, len(dest))
for i, d := range dest {
switch v := d.(type) {
case *time.Time:
row[i] = *v
case *uint64:
row[i] = *v
case *string:
row[i] = *v
}
}
return row
}
}
func mapOrEmpty(m map[string]string) map[string]string {
if m == nil {
return map[string]string{}
}
return m
}
@@ -0,0 +1,70 @@
package implsentry
import (
"context"
"strings"
"testing"
cmock "github.com/hanzoai/clickhouse-go-mock"
"github.com/hanzoai/o11y/pkg/telemetrystore"
"github.com/hanzoai/o11y/pkg/telemetrystore/telemetrystoretest"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// anyMatcher lets the mock match on expectation type + args, not exact SQL text —
// the SQL text itself is asserted by the pure builder tests (eventsql_test.go).
type anyMatcher struct{}
func (anyMatcher) Match(string, string) error { return nil }
// TestEventStore_EnsureSchemaAndQuery exercises the REAL datastore IO path against the
// clickhouse-go-mock: ensureSchema runs the CREATE DATABASE + CREATE TABLE DDL exactly
// once, and a read binds + decodes rows. This verifies the wiring (query executes,
// rows scan) — it does NOT verify a live datastore accepts the DDL (see the honest
// caveat on createSchemaDDL; no live datastore was reachable in this build).
func TestEventStore_EnsureSchemaAndQuery(t *testing.T) {
provider := telemetrystoretest.New(telemetrystore.Config{}, anyMatcher{})
mock := provider.Mock()
mock.MatchExpectationsInOrder(false)
// ensureSchema → CREATE DATABASE + CREATE TABLE (idempotent, once).
mock.ExpectExec("CREATE DATABASE")
mock.ExpectExec("CREATE TABLE")
// DistinctFingerprints → one String column. The 4 bound args are the (org,
// project, from, to) tenant+window scope every read carries.
mock.ExpectQuery("SELECT DISTINCT fingerprint").
WithArgs(nil, nil, nil, nil). // nil = match-any (cmock.matchArg); 4 = the tenant+window scope
WillReturnRows(cmock.NewRows(
[]cmock.ColumnType{{Name: "fingerprint", Type: "String"}},
[][]any{{"fp-1"}, {"fp-2"}},
))
store := NewEventStore(telemetrystore.TelemetryStore(provider))
fps, err := store.DistinctFingerprints(context.Background(), valuer.GenerateUUID(), valuer.GenerateUUID(), testWindow())
require.NoError(t, err)
assert.Equal(t, []string{"fp-1", "fp-2"}, fps)
require.NoError(t, mock.ExpectationsWereMet())
}
// TestInsertTemplateColumnCount pins the insert-sink invariant: the INSERT column list
// and the batch.Append argument list are the SAME length (24) and in the SAME order as
// the read projection — so a written row reads back field-for-field. A drift here is
// exactly the class of bug the honest sink note warned about.
func TestInsertTemplateColumnCount(t *testing.T) {
// Column list between the first "(" and the first ")".
open := strings.Index(insertSQL, "(")
closeP := strings.Index(insertSQL, ")")
require.Greater(t, closeP, open)
insertCols := strings.Count(insertSQL[open:closeP], ",") + 1
selectCols := strings.Count(selectColumns, ",") + 1
assert.Equal(t, 24, insertCols, "insert must write all 24 columns")
assert.Equal(t, 24, selectCols, "read projection must match the 24 written columns")
}
var _ sentrytypes.EventStore = (*eventStore)(nil)
+480
View File
@@ -0,0 +1,480 @@
package implsentry
import (
"context"
"io"
"net/http"
"strconv"
"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/implerrortracking"
"github.com/hanzoai/o11y/pkg/modules/sentry"
"github.com/hanzoai/o11y/pkg/types/authtypes"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
const (
viewTimeout = 30 * time.Second
writeTimeout = 15 * time.Second
ingestTimeout = 15 * time.Second
)
// eventParser turns a decoded ingest body into events; the two wire formats differ
// only here (reused engine parsers).
type eventParser func([]byte) ([]*errortrackingtypes.SentryEvent, error)
type handler struct {
module sentry.Module
ingestEnabled bool
capturePII bool
}
// NewHandler builds the /v1/sentry HTTP surface. ingestEnabled reflects whether the
// KMS ingest secret is configured (empty => ingest fails closed 503, reads still
// work); capturePII retains end-user PII on ingest when true (default false = scrub).
func NewHandler(module sentry.Module, ingestEnabled, capturePII bool) sentry.Handler {
return &handler{module: module, ingestEnabled: ingestEnabled, capturePII: capturePII}
}
// --- ingest (public, DSN-authenticated) ---
func (h *handler) EnvelopeIngest(rw http.ResponseWriter, r *http.Request) {
h.ingest(rw, r, implerrortracking.ParseEnvelope)
}
func (h *handler) StoreIngest(rw http.ResponseWriter, r *http.Request) {
h.ingest(rw, r, implerrortracking.ParseStoreBody)
}
// ingest is the shared pipeline: enabled-check → parse the project id → resolve org +
// verify the DSN key against the project watermark (fail-closed) → per-project rate
// limit → bounded read+decode → parse (event-count capped) → normalize (scrub) →
// persist to the events plane AND the issue lifecycle. Every failure fails closed and
// leaks no internal detail to the untrusted client. Reuses the errortracking engine
// verbatim for decode/parse/normalize/key-verify/rate-limit.
func (h *handler) ingest(rw http.ResponseWriter, r *http.Request, parse eventParser) {
ctx, cancel := context.WithTimeout(r.Context(), ingestTimeout)
defer cancel()
if !h.ingestEnabled {
http.Error(rw, "sentry ingest is not configured", http.StatusServiceUnavailable)
return
}
projectID, err := valuer.NewUUID(mux.Vars(r)["project"])
if err != nil {
http.Error(rw, "invalid project", http.StatusBadRequest)
return
}
orgID, ok := h.module.ResolveIngest(ctx, projectID, implerrortracking.SentryKeyFromRequest(r))
if !ok {
// 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.module.RateAllow(projectID) {
rw.Header().Set("Retry-After", "1")
http.Error(rw, "rate limited", http.StatusTooManyRequests)
return
}
r.Body = http.MaxBytesReader(rw, r.Body, implerrortracking.MaxCompressedBody)
raw, err := io.ReadAll(r.Body)
if err != nil {
http.Error(rw, "payload too large", http.StatusRequestEntityTooLarge)
return
}
decoded, err := implerrortracking.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
}
occs := make([]*errortrackingtypes.Occurrence, 0, len(events))
lastID := ""
for _, ev := range events {
occ := implerrortracking.NormalizeEvent(ev, h.capturePII)
if occ.Fingerprint == "" {
continue
}
occs = append(occs, occ)
lastID = occ.EventID
}
if err := h.module.Ingest(ctx, orgID, projectID, occs); err != nil {
http.Error(rw, "ingest failed", http.StatusInternalServerError)
return
}
render.Success(rw, http.StatusOK, map[string]string{"id": lastID})
}
// --- projects ---
func (h *handler) ListProjects(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
}
out, err := h.module.ListProjects(ctx, orgID)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, out)
}
func (h *handler) CreateProject(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
}
req := new(sentrytypes.PostableProject)
if err := binding.JSON.BindBody(r.Body, req); err != nil {
render.Error(rw, err)
return
}
p, err := h.module.CreateProject(ctx, orgID, req)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, p)
}
func (h *handler) GetProject(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
}
p, err := h.module.GetProject(ctx, orgID, id)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, p)
}
func (h *handler) RotateProjectKey(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
}
p, err := h.module.RotateProjectKey(ctx, orgID, id)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, p)
}
// --- issues ---
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
}
var projectID *valuer.UUID
if raw := r.URL.Query().Get("project"); raw != "" {
id, err := valuer.NewUUID(raw)
if err != nil {
render.Error(rw, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "project is not a valid uuid"))
return
}
projectID = &id
}
out, err := h.module.ListIssues(ctx, orgID, projectID, &q, resolveWindow(r.URL.Query().Get("period"), time.Now().UTC()))
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, out)
}
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)
}
func (h *handler) IssueEvents(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
}
projectID, err := projectFromQuery(r)
if err != nil {
render.Error(rw, err)
return
}
events, err := h.module.IssueEvents(ctx, orgID, id, projectID, queryLimit(r))
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, map[string]any{"items": events})
}
// --- discover / events / logs / traces / stats ---
func (h *handler) Discover(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
}
req := new(sentrytypes.DiscoverRequest)
if err := binding.JSON.BindBody(r.Body, req); err != nil {
render.Error(rw, err)
return
}
out, err := h.module.Discover(ctx, orgID, req)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, out)
}
func (h *handler) GetEvent(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
}
projectID, err := projectFromQuery(r)
if err != nil {
render.Error(rw, err)
return
}
event, err := h.module.GetEvent(ctx, orgID, projectID, mux.Vars(r)["id"])
if err != nil {
render.Error(rw, err)
return
}
if event == nil {
render.Error(rw, errors.Newf(errors.TypeNotFound, sentrytypes.ErrCodeSentryNotFound, "event not found"))
return
}
render.Success(rw, http.StatusOK, event)
}
func (h *handler) ListLogs(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
}
projectID, err := projectFromQuery(r)
if err != nil {
render.Error(rw, err)
return
}
events, err := h.module.ListLogs(ctx, orgID, projectID, r.URL.Query().Get("query"), r.URL.Query().Get("period"), queryLimit(r))
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, map[string]any{"items": events})
}
func (h *handler) ListTraces(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
}
projectID, err := projectFromQuery(r)
if err != nil {
render.Error(rw, err)
return
}
traces, err := h.module.ListTraces(ctx, orgID, projectID, r.URL.Query().Get("period"), queryLimit(r))
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, map[string]any{"items": traces})
}
func (h *handler) GetTrace(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
}
projectID, err := projectFromQuery(r)
if err != nil {
render.Error(rw, err)
return
}
detail, err := h.module.TraceDetail(ctx, orgID, projectID, mux.Vars(r)["id"])
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, detail)
}
func (h *handler) Stats(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
}
projectID, err := projectFromQuery(r)
if err != nil {
render.Error(rw, err)
return
}
points, err := h.module.Stats(ctx, orgID, projectID, r.URL.Query().Get("field"), r.URL.Query().Get("period"))
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, map[string]any{"items": points})
}
// --- shared helpers ---
// orgFromContext resolves the caller's org UUID from the gateway-asserted claims. A
// malformed/absent org fails closed as unauthenticated rather than panicking.
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, sentrytypes.ErrCodeSentryUnauthorized, "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.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "id is not a valid uuid")
}
return id, nil
}
// projectFromQuery reads and validates the mandatory ?project= param for the
// project-scoped reads (logs/traces/stats). The value must be a UUID; ownership is
// enforced downstream by the org-scoped project lookup in the module.
func projectFromQuery(r *http.Request) (valuer.UUID, error) {
id, err := valuer.NewUUID(r.URL.Query().Get("project"))
if err != nil {
return valuer.UUID{}, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "a valid project query param is required")
}
return id, nil
}
func queryLimit(r *http.Request) int {
n, _ := strconv.Atoi(r.URL.Query().Get("limit"))
return n
}
+369
View File
@@ -0,0 +1,369 @@
package implsentry
import (
"context"
"encoding/json"
"strings"
"time"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/modules/errortracking/implerrortracking"
"github.com/hanzoai/o11y/pkg/modules/sentry"
"github.com/hanzoai/o11y/pkg/types"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
// nowUTC is the ONE package time source for lifecycle writes (overridable in tests).
var nowUTC = func() time.Time { return time.Now().UTC() }
// Config wires the sentry module's non-store dependencies.
type Config struct {
// IngestSecret is the KMS-sourced platform DSN-signing secret (the SAME secret the
// errortracking ingest path verifies against). Empty => ingest fails closed.
IngestSecret []byte
// Host is the DSN endpoint origin (e.g. "api.hanzo.ai"); the minted DSN points at
// https://<key>@<host>/v1/sentry/<project>.
Host string
// CapturePII retains end-user PII (email/ip) when true; default false = scrub.
CapturePII bool
}
type module struct {
projects sentrytypes.ProjectStore
events sentrytypes.EventStore
issues errortracking.Module // reused grouped-issue lifecycle (o11y_issues)
limiter *implerrortracking.RateLimiter
cfg Config
}
// NewModule composes the sentry product face over the reused engine, the projects
// store, the columnar events plane and the reused issue lifecycle. Trace/event/issue
// reads are all org+project scoped over the events plane (the o11y_traces span plane
// is intentionally NOT read — it has no general org column, so it cannot be
// tenant-scoped for this multi-tenant product; see the report).
func NewModule(projects sentrytypes.ProjectStore, events sentrytypes.EventStore, issues errortracking.Module, cfg Config) sentry.Module {
return &module{
projects: projects,
events: events,
issues: issues,
limiter: implerrortracking.NewRateLimiter(implerrortracking.IngestRatePerSec, implerrortracking.IngestBurst),
cfg: cfg,
}
}
// --- ingest ---
// Ingest persists a request's occurrences for (org, project): the columnar events
// plane (queryable, high-volume) AND the grouped-issue lifecycle (reused verbatim).
// The events write is fail-soft — a datastore hiccup must not drop the durable issue
// upsert — so its error is logged-and-swallowed here (the issue path is the source of
// truth for the Issues UI).
func (m *module) Ingest(ctx context.Context, orgID, projectID valuer.UUID, occs []*errortrackingtypes.Occurrence) error {
if orgID.IsZero() || projectID.IsZero() {
return errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "ingest has no org/project")
}
events := make([]*sentrytypes.Event, 0, len(occs))
for _, occ := range occs {
if occ == nil || occ.Fingerprint == "" {
continue
}
events = append(events, eventFromOccurrence(orgID, projectID, occ))
}
// Events plane (fail-soft) FIRST, then the durable issue lifecycle (authoritative).
_ = m.events.Insert(ctx, orgID, projectID, events)
if _, err := m.issues.Ingest(ctx, orgID, occs); err != nil {
return err
}
return nil
}
// ResolveIngest maps a DSN project id to its owning org, verifying the presented key
// against the project's rotation watermark. Fail-closed at every step: unknown or
// disabled project, or a key below the watermark, returns ok=false.
func (m *module) ResolveIngest(ctx context.Context, projectID valuer.UUID, presentedKey string) (valuer.UUID, bool) {
if len(m.cfg.IngestSecret) == 0 {
return valuer.UUID{}, false
}
orgID, keyVersion, status, found, err := m.projects.Resolve(ctx, projectID)
if err != nil || !found || status != sentrytypes.ProjectActive {
return valuer.UUID{}, false
}
if !implerrortracking.VerifyKey(m.cfg.IngestSecret, projectID.String(), presentedKey, keyVersion) {
return valuer.UUID{}, false
}
return orgID, true
}
func (m *module) RateAllow(projectID valuer.UUID) bool { return m.limiter.Allow(projectID) }
// --- projects ---
func (m *module) CreateProject(ctx context.Context, orgID valuer.UUID, in *sentrytypes.PostableProject) (*sentrytypes.GettableProject, error) {
name := strings.TrimSpace(in.Name)
if name == "" {
return nil, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "project name is required")
}
slug := slugify(in.Slug)
if slug == "" {
slug = slugify(name)
}
if reservedSlugs[slug] {
return nil, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "project slug %q is reserved", slug)
}
now := nowUTC()
p := &sentrytypes.Project{
Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
TimeAuditable: types.TimeAuditable{CreatedAt: now, UpdatedAt: now},
OrgID: orgID,
Name: name,
Slug: slug,
Platform: strings.TrimSpace(in.Platform),
Status: sentrytypes.ProjectActive,
KeyVersion: 1,
}
if err := m.projects.Create(ctx, p); err != nil {
return nil, err
}
return m.gettable(p), nil
}
func (m *module) ListProjects(ctx context.Context, orgID valuer.UUID) (*sentrytypes.GettableProjects, error) {
ps, err := m.projects.List(ctx, orgID)
if err != nil {
return nil, err
}
items := make([]*sentrytypes.GettableProject, 0, len(ps))
for _, p := range ps {
items = append(items, m.gettable(p))
}
return &sentrytypes.GettableProjects{Items: items, Total: len(items)}, nil
}
func (m *module) GetProject(ctx context.Context, orgID, id valuer.UUID) (*sentrytypes.GettableProject, error) {
p, err := m.projects.Get(ctx, orgID, id)
if err != nil {
return nil, err
}
return m.gettable(p), nil
}
func (m *module) RotateProjectKey(ctx context.Context, orgID, id valuer.UUID) (*sentrytypes.GettableProject, error) {
version, err := m.projects.Rotate(ctx, orgID, id)
if err != nil {
return nil, err
}
p, err := m.projects.Get(ctx, orgID, id)
if err != nil {
return nil, err
}
p.KeyVersion = version
return m.gettable(p), nil
}
// gettable derives the project's current DSN (never stored) and returns the API view.
func (m *module) gettable(p *sentrytypes.Project) *sentrytypes.GettableProject {
dsn := ""
if len(m.cfg.IngestSecret) > 0 {
dsn = mintDSN(m.cfg.IngestSecret, m.cfg.Host, p.ID, p.KeyVersion)
}
return &sentrytypes.GettableProject{Project: p, DSN: dsn}
}
// --- issues (reused errortracking lifecycle, org-scoped) ---
// ListIssues returns the org's grouped issues, optionally narrowed to a project. The
// project narrowing is the events-plane projection: an issue (fingerprint) belongs to
// a project iff it has captured events there in the window. The fingerprint set is
// server-derived and passed as a server-only filter, so no client can widen scope.
func (m *module) ListIssues(ctx context.Context, orgID valuer.UUID, projectID *valuer.UUID, q *errortrackingtypes.IssuesQuery, w sentrytypes.Window) (*errortrackingtypes.GettableIssues, error) {
if projectID != nil {
// Validate the project belongs to the caller's org (foreign id => not found).
if _, err := m.projects.Get(ctx, orgID, *projectID); err != nil {
return nil, err
}
fps, err := m.events.DistinctFingerprints(ctx, orgID, *projectID, w)
if err != nil {
return nil, err
}
if len(fps) == 0 {
// A project with no captured errors has no issues — do not run an unfiltered
// (whole-org) list.
return &errortrackingtypes.GettableIssues{Items: []*errortrackingtypes.Issue{}, Total: 0, Offset: 0, Limit: q.Limit}, nil
}
q.Fingerprints = fps
}
items, total, err := m.issues.ListIssues(ctx, orgID, q)
if err != nil {
return nil, err
}
return &errortrackingtypes.GettableIssues{Items: items, Total: total, Offset: q.Offset, Limit: q.Limit}, nil
}
func (m *module) GetIssue(ctx context.Context, orgID, id valuer.UUID) (*errortrackingtypes.GettableIssue, error) {
return m.issues.GetIssue(ctx, orgID, id)
}
func (m *module) UpdateIssue(ctx context.Context, orgID, id valuer.UUID, in *errortrackingtypes.UpdateIssue) (*errortrackingtypes.Issue, error) {
return m.issues.UpdateIssue(ctx, orgID, id, in)
}
// IssueEvents returns an issue's recent occurrences from the events plane, scoped to
// (org, project): the org-scoped GetIssue resolves the fingerprint, then the events
// read binds BOTH org and project — a project is an isolation unit, so occurrences are
// never read across projects.
func (m *module) IssueEvents(ctx context.Context, orgID, id, projectID valuer.UUID, limit int) ([]*sentrytypes.Event, error) {
if _, err := m.projects.Get(ctx, orgID, projectID); err != nil {
return nil, err
}
issue, err := m.issues.GetIssue(ctx, orgID, id)
if err != nil {
return nil, err
}
return m.events.ListForFingerprint(ctx, orgID, projectID, issue.Issue.Fingerprint, limit)
}
// --- discover / events / logs / traces / stats (events plane) ---
func (m *module) Discover(ctx context.Context, orgID valuer.UUID, req *sentrytypes.DiscoverRequest) (*sentrytypes.DiscoverResult, error) {
projectID, err := m.requireProject(ctx, orgID, req.Project)
if err != nil {
return nil, err
}
return m.events.Discover(ctx, orgID, projectID, req, resolveWindow(req.Period, nowUTC()))
}
func (m *module) GetEvent(ctx context.Context, orgID, projectID valuer.UUID, eventID string) (*sentrytypes.Event, error) {
if _, err := m.projects.Get(ctx, orgID, projectID); err != nil {
return nil, err
}
return m.events.GetEvent(ctx, orgID, projectID, eventID)
}
func (m *module) ListLogs(ctx context.Context, orgID, projectID valuer.UUID, query, period string, limit int) ([]*sentrytypes.Event, error) {
if _, err := m.projects.Get(ctx, orgID, projectID); err != nil {
return nil, err
}
return m.events.ListLogs(ctx, orgID, projectID, query, resolveWindow(period, nowUTC()), limit)
}
func (m *module) ListTraces(ctx context.Context, orgID, projectID valuer.UUID, period string, limit int) ([]*sentrytypes.TraceSummary, error) {
if _, err := m.projects.Get(ctx, orgID, projectID); err != nil {
return nil, err
}
return m.events.ListTraces(ctx, orgID, projectID, resolveWindow(period, nowUTC()), limit)
}
// TraceDetail returns the (org, project)-scoped error events referencing a trace —
// the tenant-safe "errors in this trace" view. The load-bearing scope is on the READ
// itself: the events query binds org AND project, so a caller only ever sees their own
// project's events for a trace, never another tenant's. The o11y_traces span waterfall
// is intentionally NOT read — that plane has no general org column (only gen_ai spans
// carry gen_ai.hanzo.org_id), so it cannot be tenant-scoped for a multi-tenant product;
// the full span waterfall is a follow-on gated on a general org column in o11y_traces.
func (m *module) TraceDetail(ctx context.Context, orgID, projectID valuer.UUID, traceID string) (any, error) {
if _, err := m.projects.Get(ctx, orgID, projectID); err != nil {
return nil, err
}
events, err := m.events.ListForTrace(ctx, orgID, projectID, traceID, 0)
if err != nil {
return nil, err
}
return map[string]any{"traceId": traceID, "events": events}, nil
}
func (m *module) Stats(ctx context.Context, orgID, projectID valuer.UUID, field, period string) ([]sentrytypes.StatsPoint, error) {
if _, err := m.projects.Get(ctx, orgID, projectID); err != nil {
return nil, err
}
return m.events.Stats(ctx, orgID, projectID, field, resolveWindow(period, nowUTC()))
}
// requireProject parses + org-validates a project param, returning a clear error when
// it is missing or foreign (the tenant boundary for every project-scoped read).
func (m *module) requireProject(ctx context.Context, orgID valuer.UUID, raw string) (valuer.UUID, error) {
id, err := valuer.NewUUID(strings.TrimSpace(raw))
if err != nil {
return valuer.UUID{}, errors.Newf(errors.TypeInvalidInput, sentrytypes.ErrCodeSentryInvalidInput, "a valid project is required")
}
if _, err := m.projects.Get(ctx, orgID, id); err != nil {
return valuer.UUID{}, err
}
return id, nil
}
// eventFromOccurrence maps a normalized Occurrence to a columnar events-plane row for
// (org, project). The full occurrence is retained as the sample JSON for event detail.
func eventFromOccurrence(orgID, projectID valuer.UUID, occ *errortrackingtypes.Occurrence) *sentrytypes.Event {
sample, _ := json.Marshal(occ)
e := &sentrytypes.Event{
OrgID: orgID.String(),
ProjectID: projectID.String(),
EventID: occ.EventID,
Timestamp: occ.Timestamp,
Level: occ.Level,
Type: occ.Type,
Value: occ.Value,
Message: firstNonEmpty(occ.Value, occ.Type),
Culprit: occ.Culprit,
Fingerprint: occ.Fingerprint,
Platform: occ.Platform,
Environment: occ.Environment,
Release: occ.Release,
ServiceName: occ.ServiceName,
Transaction: occ.Transaction,
TraceID: occ.TraceID,
SpanID: occ.SpanID,
ServerName: occ.ServerName,
Tags: occ.Tags,
Sample: string(sample),
}
if occ.Timestamp.IsZero() {
e.Timestamp = nowUTC()
}
if occ.User != nil {
e.UserID = occ.User.ID
e.UserEmail = occ.User.Email
e.UserIP = occ.User.IP
}
return e
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if v != "" {
return v
}
}
return ""
}
// reservedSlugs are the static /v1/sentry resource words a project slug may not take,
// so a slug can never be confused with a route (belt-and-suspenders alongside the
// UUID-constrained ingest route and static-before-wildcard registration).
var reservedSlugs = map[string]bool{
"projects": true, "issues": true, "discover": true, "events": true,
"logs": true, "traces": true, "stats": true, "envelope": true, "store": true,
}
// slugify lowercases and reduces a name to a URL-safe slug (a-z0-9 and single dashes).
func slugify(s string) string {
s = strings.ToLower(strings.TrimSpace(s))
var b strings.Builder
lastDash := false
for _, r := range s {
switch {
case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
b.WriteRune(r)
lastDash = false
case r == ' ' || r == '-' || r == '_' || r == '.':
if !lastDash && b.Len() > 0 {
b.WriteByte('-')
lastDash = true
}
}
}
return strings.Trim(b.String(), "-")
}
@@ -0,0 +1,380 @@
package implsentry
import (
"context"
"testing"
"time"
"github.com/hanzoai/o11y/pkg/modules/errortracking"
"github.com/hanzoai/o11y/pkg/modules/errortracking/implerrortracking"
"github.com/hanzoai/o11y/pkg/modules/sentry"
"github.com/hanzoai/o11y/pkg/sqlstore"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// fakeEvents is an in-memory EventStore: it records every write keyed by (org,
// project) so a test can assert BOTH that ingest reached the events plane AND that a
// read is only ever asked for the caller's own tenant.
type fakeEvents struct {
inserts map[[2]string][]*sentrytypes.Event
lastOrg string
lastProj string
discovers int
}
func newFakeEvents() *fakeEvents {
return &fakeEvents{inserts: map[[2]string][]*sentrytypes.Event{}}
}
func (f *fakeEvents) key(o, p valuer.UUID) [2]string { return [2]string{o.String(), p.String()} }
func (f *fakeEvents) Insert(_ context.Context, o, p valuer.UUID, e []*sentrytypes.Event) error {
f.inserts[f.key(o, p)] = append(f.inserts[f.key(o, p)], e...)
return nil
}
func (f *fakeEvents) Discover(_ context.Context, o, p valuer.UUID, _ *sentrytypes.DiscoverRequest, _ sentrytypes.Window) (*sentrytypes.DiscoverResult, error) {
f.lastOrg, f.lastProj, f.discovers = o.String(), p.String(), f.discovers+1
return &sentrytypes.DiscoverResult{}, nil
}
func (f *fakeEvents) GetEvent(_ context.Context, o, p valuer.UUID, id string) (*sentrytypes.Event, error) {
// Tenant boundary: only THIS (org, project)'s events — never another org's or
// another project's within the org.
for _, e := range f.inserts[f.key(o, p)] {
if e.EventID == id {
return e, nil
}
}
return nil, nil
}
func (f *fakeEvents) ListForFingerprint(_ context.Context, o, p valuer.UUID, fp string, _ int) ([]*sentrytypes.Event, error) {
var out []*sentrytypes.Event
for _, e := range f.inserts[f.key(o, p)] {
if e.Fingerprint == fp {
out = append(out, e)
}
}
return out, nil
}
func (f *fakeEvents) ListForTrace(_ context.Context, o, p valuer.UUID, traceID string, _ int) ([]*sentrytypes.Event, error) {
f.lastOrg, f.lastProj = o.String(), p.String()
var out []*sentrytypes.Event
for _, e := range f.inserts[f.key(o, p)] {
if e.TraceID == traceID {
out = append(out, e)
}
}
return out, nil
}
func (f *fakeEvents) DistinctFingerprints(_ context.Context, o, p valuer.UUID, _ sentrytypes.Window) ([]string, error) {
seen := map[string]bool{}
var out []string
for _, e := range f.inserts[f.key(o, p)] {
if !seen[e.Fingerprint] {
seen[e.Fingerprint] = true
out = append(out, e.Fingerprint)
}
}
return out, nil
}
func (f *fakeEvents) ListLogs(_ context.Context, o, p valuer.UUID, _ string, _ sentrytypes.Window, _ int) ([]*sentrytypes.Event, error) {
f.lastOrg, f.lastProj = o.String(), p.String()
return f.inserts[f.key(o, p)], nil
}
func (f *fakeEvents) ListTraces(_ context.Context, o, p valuer.UUID, _ sentrytypes.Window, _ int) ([]*sentrytypes.TraceSummary, error) {
f.lastOrg, f.lastProj = o.String(), p.String()
return nil, nil
}
func (f *fakeEvents) Stats(_ context.Context, o, p valuer.UUID, _ string, _ sentrytypes.Window) ([]sentrytypes.StatsPoint, error) {
f.lastOrg, f.lastProj = o.String(), p.String()
return nil, nil
}
const testSecret = "platform-ingest-secret"
type harness struct {
mod sentry.Module
events *fakeEvents
projects sentrytypes.ProjectStore
}
func newModuleHarness(t *testing.T) *harness {
t.Helper()
store := newModuleSQLStore(t)
projects := NewProjectStore(store)
issues := errortracking.Module(implerrortracking.NewModule(implerrortracking.NewStore(store), implerrortracking.NewNoopSink()))
events := newFakeEvents()
mod := NewModule(projects, events, issues, Config{IngestSecret: []byte(testSecret), Host: "api.hanzo.ai"})
return &harness{mod: mod, events: events, projects: projects}
}
// newModuleSQLStore is a sqlite store with BOTH the projects table and the
// errortracking o11y_issues lifecycle table (+ its unique index).
func newModuleSQLStore(t *testing.T) sqlstore.SQLStore {
t.Helper()
store := newTestSQLStore(t) // creates o11y_sentry_projects (+ unique index)
_, 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)
return store
}
func mustProject(t *testing.T, h *harness, org valuer.UUID, name string) *sentrytypes.GettableProject {
t.Helper()
p, err := h.mod.CreateProject(context.Background(), org, &sentrytypes.PostableProject{Name: name})
require.NoError(t, err)
return p
}
func occ(fp, eventID string) *errortrackingtypes.Occurrence {
return occTrace(fp, eventID, "")
}
func occTrace(fp, eventID, traceID string) *errortrackingtypes.Occurrence {
return &errortrackingtypes.Occurrence{
EventID: eventID, Fingerprint: fp, Type: "Error", Value: "boom",
Level: "error", Timestamp: time.Now().UTC(), TraceID: traceID,
}
}
// TestIngest_WritesEventsAndIssues proves the dual write: one Ingest lands the
// occurrence on BOTH the columnar events plane and the grouped-issue lifecycle.
func TestIngest_WritesEventsAndIssues(t *testing.T) {
ctx := context.Background()
h := newModuleHarness(t)
org := valuer.GenerateUUID()
proj := mustProject(t, h, org, "web")
pid := proj.Project.ID
require.NoError(t, h.mod.Ingest(ctx, org, pid, []*errortrackingtypes.Occurrence{occ("fp-1", "e1"), occ("fp-1", "e2")}))
// Events plane got both occurrences under (org, project).
assert.Len(t, h.events.inserts[[2]string{org.String(), pid.String()}], 2)
// Issue lifecycle grouped them into one issue for the org.
issues, err := h.mod.ListIssues(ctx, org, nil, &errortrackingtypes.IssuesQuery{}, testWindow())
require.NoError(t, err)
require.Len(t, issues.Items, 1)
assert.Equal(t, "fp-1", issues.Items[0].Fingerprint)
assert.Equal(t, int64(2), issues.Items[0].Count)
}
// TestResolveIngest_FailsClosed is the DSN gate: unknown project, disabled project,
// wrong key and a below-watermark (rotated) key all return ok=false; only a correct,
// current key resolves — to the OWNING org.
func TestResolveIngest_FailsClosed(t *testing.T) {
ctx := context.Background()
h := newModuleHarness(t)
org := valuer.GenerateUUID()
proj := mustProject(t, h, org, "web")
pid := proj.Project.ID
validKey := implerrortracking.PublicKeyForVersion([]byte(testSecret), pid.String(), 1)
// Valid key + active project -> resolves to the owning org.
gotOrg, ok := h.mod.ResolveIngest(ctx, pid, validKey)
require.True(t, ok)
assert.Equal(t, org, gotOrg)
// Unknown project -> closed.
_, ok = h.mod.ResolveIngest(ctx, valuer.GenerateUUID(), validKey)
assert.False(t, ok)
// Wrong key -> closed.
_, ok = h.mod.ResolveIngest(ctx, pid, "1:deadbeef")
assert.False(t, ok)
// Empty key -> closed.
_, ok = h.mod.ResolveIngest(ctx, pid, "")
assert.False(t, ok)
// After rotation, the old v1 key is below the watermark -> closed; the new one works.
_, err := h.projects.Rotate(ctx, org, pid)
require.NoError(t, err)
_, ok = h.mod.ResolveIngest(ctx, pid, validKey)
assert.False(t, ok, "a below-watermark (pre-rotation) key must stop resolving")
v2 := implerrortracking.PublicKeyForVersion([]byte(testSecret), pid.String(), 2)
_, ok = h.mod.ResolveIngest(ctx, pid, v2)
assert.True(t, ok)
}
func TestResolveIngest_DisabledProjectFailsClosed(t *testing.T) {
ctx := context.Background()
h := newModuleHarness(t)
org := valuer.GenerateUUID()
proj := mustProject(t, h, org, "web")
pid := proj.Project.ID
// Disable the project directly in the store.
disableProject(t, h, org, pid)
key := implerrortracking.PublicKeyForVersion([]byte(testSecret), pid.String(), 1)
_, ok := h.mod.ResolveIngest(ctx, pid, key)
assert.False(t, ok, "a disabled project must not resolve even with a valid key")
}
// TestReads_ForeignProjectDenied is the mandatory read isolation: a project id that
// belongs to another org is rejected (never silently scoped to the caller), so no
// cross-tenant read is possible via a client-supplied project.
func TestReads_ForeignProjectDenied(t *testing.T) {
ctx := context.Background()
h := newModuleHarness(t)
orgA, orgB := valuer.GenerateUUID(), valuer.GenerateUUID()
projA := mustProject(t, h, orgA, "a").Project.ID
_ = mustProject(t, h, orgB, "b")
// org B asks to Discover org A's project -> denied (project not found in B's org).
_, err := h.mod.Discover(ctx, orgB, &sentrytypes.DiscoverRequest{Project: projA.String()})
require.Error(t, err)
// Same for logs / traces / stats / trace-detail — every project-scoped read.
_, err = h.mod.ListLogs(ctx, orgB, projA, "", "24h", 10)
require.Error(t, err)
_, err = h.mod.ListTraces(ctx, orgB, projA, "24h", 10)
require.Error(t, err)
_, err = h.mod.Stats(ctx, orgB, projA, "events", "24h")
require.Error(t, err)
_, err = h.mod.TraceDetail(ctx, orgB, projA, "trace-1")
require.Error(t, err)
// The fake events store was NEVER asked for org A's data on B's behalf.
assert.NotEqual(t, orgA.String(), h.events.lastOrg)
}
// TestListIssues_ProjectFilterViaEventsPlane proves the org-grouped issue list is
// correctly projected to a single project through the events-plane fingerprints, and
// that a project with no captured errors yields zero issues (never the whole org).
func TestListIssues_ProjectFilterViaEventsPlane(t *testing.T) {
ctx := context.Background()
h := newModuleHarness(t)
org := valuer.GenerateUUID()
web := mustProject(t, h, org, "web").Project.ID
api := mustProject(t, h, org, "api").Project.ID
require.NoError(t, h.mod.Ingest(ctx, org, web, []*errortrackingtypes.Occurrence{occ("fp-web", "e1")}))
require.NoError(t, h.mod.Ingest(ctx, org, api, []*errortrackingtypes.Occurrence{occ("fp-api", "e2")}))
// Whole-org list sees BOTH issues.
all, err := h.mod.ListIssues(ctx, org, nil, &errortrackingtypes.IssuesQuery{}, testWindow())
require.NoError(t, err)
assert.Len(t, all.Items, 2)
// Project-scoped list sees only that project's issue.
webOnly, err := h.mod.ListIssues(ctx, org, &web, &errortrackingtypes.IssuesQuery{}, testWindow())
require.NoError(t, err)
require.Len(t, webOnly.Items, 1)
assert.Equal(t, "fp-web", webOnly.Items[0].Fingerprint)
// A foreign project on the issue list is denied.
otherOrg := valuer.GenerateUUID()
foreign := mustProject(t, h, otherOrg, "x").Project.ID
_, err = h.mod.ListIssues(ctx, org, &foreign, &errortrackingtypes.IssuesQuery{}, testWindow())
require.Error(t, err)
}
// TestTraceDetail_CrossTenantTraceIsolation is the exact scenario Red flagged: org B
// injects an event carrying org A's trace_id, then reads that trace. Because the
// load-bearing scope is on the events READ (org AND project bound), TraceDetail returns
// ONLY org B's own project events for the trace — ZERO of org A's data — and the
// o11y_traces span plane (which has no org column) is never read.
func TestTraceDetail_CrossTenantTraceIsolation(t *testing.T) {
ctx := context.Background()
h := newModuleHarness(t)
orgA, orgB := valuer.GenerateUUID(), valuer.GenerateUUID()
pA := mustProject(t, h, orgA, "a").Project.ID
pB := mustProject(t, h, orgB, "b").Project.ID
const victimTrace = "VICTIM-TRACE-DEADBEEF"
// org A's real event on the victim trace.
require.NoError(t, h.mod.Ingest(ctx, orgA, pA, []*errortrackingtypes.Occurrence{occTrace("fp-a", "a-secret", victimTrace)}))
// org B forges an event CLAIMING the same trace id in ITS OWN project.
require.NoError(t, h.mod.Ingest(ctx, orgB, pB, []*errortrackingtypes.Occurrence{occTrace("fp-b", "b-own", victimTrace)}))
// org B reads the trace in its own project: sees ONLY its own event, never org A's.
detail, err := h.mod.TraceDetail(ctx, orgB, pB, victimTrace)
require.NoError(t, err)
events := detail.(map[string]any)["events"].([]*sentrytypes.Event)
require.Len(t, events, 1)
assert.Equal(t, "b-own", events[0].EventID)
for _, e := range events {
assert.NotEqual(t, "a-secret", e.EventID, "org A's event must NEVER surface for org B")
assert.Equal(t, orgB.String(), e.OrgID, "only org B rows may be returned")
}
// org B cannot even target org A's project (foreign project → denied).
_, err = h.mod.TraceDetail(ctx, orgB, pA, victimTrace)
require.Error(t, err)
}
// TestGetEvent_ProjectScoped: a within-tenant cross-PROJECT read is denied — a project
// is the isolation unit, so an event in project X is not readable via project Y.
func TestGetEvent_ProjectScoped(t *testing.T) {
ctx := context.Background()
h := newModuleHarness(t)
org := valuer.GenerateUUID()
web := mustProject(t, h, org, "web").Project.ID
api := mustProject(t, h, org, "api").Project.ID
require.NoError(t, h.mod.Ingest(ctx, org, web, []*errortrackingtypes.Occurrence{occ("fp", "evt-web")}))
// Correct project → found.
got, err := h.mod.GetEvent(ctx, org, web, "evt-web")
require.NoError(t, err)
require.NotNil(t, got)
assert.Equal(t, "evt-web", got.EventID)
// Wrong project (same org) → not found (not a leak).
got, err = h.mod.GetEvent(ctx, org, api, "evt-web")
require.NoError(t, err)
assert.Nil(t, got, "event from another project must not be readable via a different project")
// Foreign project (another org) → denied.
other := valuer.GenerateUUID()
foreign := mustProject(t, h, other, "x").Project.ID
_, err = h.mod.GetEvent(ctx, org, foreign, "evt-web")
require.Error(t, err)
}
// TestIssueEvents_ProjectScoped: issue occurrences are read only for the named project.
func TestIssueEvents_ProjectScoped(t *testing.T) {
ctx := context.Background()
h := newModuleHarness(t)
org := valuer.GenerateUUID()
web := mustProject(t, h, org, "web").Project.ID
api := mustProject(t, h, org, "api").Project.ID
require.NoError(t, h.mod.Ingest(ctx, org, web, []*errortrackingtypes.Occurrence{occ("fp-shared", "e-web")}))
require.NoError(t, h.mod.Ingest(ctx, org, api, []*errortrackingtypes.Occurrence{occ("fp-shared", "e-api")}))
// One org-scoped issue exists for fp-shared; find it.
issues, err := h.mod.ListIssues(ctx, org, nil, &errortrackingtypes.IssuesQuery{}, testWindow())
require.NoError(t, err)
require.Len(t, issues.Items, 1)
issueID := issues.Items[0].ID
// Occurrences scoped to web → only the web event.
webEvents, err := h.mod.IssueEvents(ctx, org, issueID, web, 0)
require.NoError(t, err)
require.Len(t, webEvents, 1)
assert.Equal(t, "e-web", webEvents[0].EventID)
// Foreign project → denied.
other := valuer.GenerateUUID()
foreign := mustProject(t, h, other, "x").Project.ID
_, err = h.mod.IssueEvents(ctx, org, issueID, foreign, 0)
require.Error(t, err)
}
// disableProject flips a project's status to disabled via a direct store write.
func disableProject(t *testing.T, h *harness, org, id valuer.UUID) {
t.Helper()
ps := h.projects.(*projectStore)
_, err := ps.sqlstore.BunDB().NewUpdate().
Model((*sentrytypes.Project)(nil)).
Set("status = ?", sentrytypes.ProjectDisabled).
Where("org_id = ?", org).Where("id = ?", id).
Exec(context.Background())
require.NoError(t, err)
}
@@ -0,0 +1,106 @@
package implsentry
import (
"context"
"database/sql"
stderrors "errors"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/sqlstore"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
// projectStore backs o11y_sentry_projects. Every method is org-scoped except Resolve
// (the DSN-authenticated ingest lookup, keyed by the unguessable project id).
type projectStore struct {
sqlstore sqlstore.SQLStore
}
// NewProjectStore wires the relational projects store.
func NewProjectStore(sqlstore sqlstore.SQLStore) sentrytypes.ProjectStore {
return &projectStore{sqlstore: sqlstore}
}
func (s *projectStore) Create(ctx context.Context, p *sentrytypes.Project) error {
_, err := s.sqlstore.BunDBCtx(ctx).NewInsert().Model(p).Exec(ctx)
if err != nil {
return s.sqlstore.WrapAlreadyExistsErrf(err, sentrytypes.ErrCodeSentryConflict, "a project named %q already exists in the org", p.Slug)
}
return nil
}
func (s *projectStore) List(ctx context.Context, orgID valuer.UUID) ([]*sentrytypes.Project, error) {
projects := make([]*sentrytypes.Project, 0)
// MANDATORY tenant boundary, first predicate — there is no unscoped project list.
err := s.sqlstore.BunDBCtx(ctx).
NewSelect().
Model(&projects).
Where("org_id = ?", orgID).
OrderExpr("created_at DESC").
Scan(ctx)
if err != nil {
return nil, err
}
return projects, nil
}
func (s *projectStore) Get(ctx context.Context, orgID, id valuer.UUID) (*sentrytypes.Project, error) {
p := new(sentrytypes.Project)
err := s.sqlstore.BunDBCtx(ctx).
NewSelect().
Model(p).
Where("org_id = ?", orgID).
Where("id = ?", id).
Scan(ctx)
if err != nil {
return nil, s.sqlstore.WrapNotFoundErrf(err, sentrytypes.ErrCodeSentryNotFound, "project %s not found in the org", id)
}
return p, nil
}
// Rotate bumps the project's key watermark (invalidating below-version DSNs). The
// bump is org-scoped and the loaded row's version drives the returned new version;
// zero rows affected means the project does not belong to the caller's org.
func (s *projectStore) Rotate(ctx context.Context, orgID, id valuer.UUID) (int, error) {
p, err := s.Get(ctx, orgID, id)
if err != nil {
return 0, err
}
res, err := s.sqlstore.BunDBCtx(ctx).
NewUpdate().
Model((*sentrytypes.Project)(nil)).
Set("key_version = key_version + 1").
Set("updated_at = ?", nowUTC()).
Where("org_id = ?", orgID).
Where("id = ?", id).
Exec(ctx)
if err != nil {
return 0, err
}
if n, _ := res.RowsAffected(); n == 0 {
return 0, errors.Newf(errors.TypeNotFound, sentrytypes.ErrCodeSentryNotFound, "project %s not found in the org", id)
}
return p.KeyVersion + 1, nil
}
// Resolve is the ingest-time lookup: it maps a project id to its owning org, current
// key version and status WITHOUT an org filter (ingest carries no IAM principal — the
// DSN key is the credential, verified by the caller). Fail-closed: an unknown project
// returns found=false, so a forged/unknown DSN never resolves to a tenant.
func (s *projectStore) Resolve(ctx context.Context, id valuer.UUID) (valuer.UUID, int, sentrytypes.ProjectStatus, bool, error) {
p := new(sentrytypes.Project)
err := s.sqlstore.BunDB().
NewSelect().
Model(p).
Column("org_id", "key_version", "status").
Where("id = ?", id).
Scan(ctx)
if err != nil {
if stderrors.Is(err, sql.ErrNoRows) {
return valuer.UUID{}, 0, "", false, nil // fail-closed: unknown project never resolves
}
return valuer.UUID{}, 0, "", false, err
}
return p.OrgID, p.KeyVersion, p.Status, true, nil
}
@@ -0,0 +1,132 @@
package implsentry
import (
"context"
"path/filepath"
"testing"
"time"
"github.com/hanzoai/o11y/pkg/factory/factorytest"
"github.com/hanzoai/o11y/pkg/sqlstore"
"github.com/hanzoai/o11y/pkg/sqlstore/sqlitesqlstore"
"github.com/hanzoai/o11y/pkg/types"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
"github.com/hanzoai/o11y/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// newTestSQLStore builds a real sqlite store with the o11y_sentry_projects table and
// its (org_id, slug) unique index — the shape the migration ships.
func newTestSQLStore(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((*sentrytypes.Project)(nil)).IfNotExists().Exec(context.Background())
require.NoError(t, err)
_, err = store.BunDB().Exec(`CREATE UNIQUE INDEX IF NOT EXISTS uq_o11y_sentry_projects_org_slug ON o11y_sentry_projects (org_id, slug)`)
require.NoError(t, err)
return store
}
func newProject(orgID valuer.UUID, name, slug string) *sentrytypes.Project {
now := time.Now().UTC()
return &sentrytypes.Project{
Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
TimeAuditable: types.TimeAuditable{CreatedAt: now, UpdatedAt: now},
OrgID: orgID,
Name: name,
Slug: slug,
Status: sentrytypes.ProjectActive,
KeyVersion: 1,
}
}
func TestProjectStore_CRUDAndRotate(t *testing.T) {
ctx := context.Background()
store := NewProjectStore(newTestSQLStore(t))
org := valuer.GenerateUUID()
p := newProject(org, "Web", "web")
require.NoError(t, store.Create(ctx, p))
got, err := store.Get(ctx, org, p.ID)
require.NoError(t, err)
assert.Equal(t, "web", got.Slug)
assert.Equal(t, 1, got.KeyVersion)
list, err := store.List(ctx, org)
require.NoError(t, err)
require.Len(t, list, 1)
// Rotate bumps the key watermark.
v, err := store.Rotate(ctx, org, p.ID)
require.NoError(t, err)
assert.Equal(t, 2, v)
got, err = store.Get(ctx, org, p.ID)
require.NoError(t, err)
assert.Equal(t, 2, got.KeyVersion)
// Resolve (ingest path) returns the owning org + watermark.
rOrg, ver, status, found, err := store.Resolve(ctx, p.ID)
require.NoError(t, err)
assert.True(t, found)
assert.Equal(t, org, rOrg)
assert.Equal(t, 2, ver)
assert.Equal(t, sentrytypes.ProjectActive, status)
}
// TestProjectStore_TenantIsolation is the mandatory two-org isolation test: org B can
// neither read nor rotate org A's project, and each org lists only its own.
func TestProjectStore_TenantIsolation(t *testing.T) {
ctx := context.Background()
store := NewProjectStore(newTestSQLStore(t))
orgA, orgB := valuer.GenerateUUID(), valuer.GenerateUUID()
pa := newProject(orgA, "A app", "a-app")
pb := newProject(orgB, "B app", "b-app")
require.NoError(t, store.Create(ctx, pa))
require.NoError(t, store.Create(ctx, pb))
// org B cannot GET org A's project — foreign id is not found in B's scope.
_, err := store.Get(ctx, orgB, pa.ID)
require.Error(t, err)
// org B cannot ROTATE org A's project.
_, err = store.Rotate(ctx, orgB, pa.ID)
require.Error(t, err)
// ...and org A's key is untouched by B's attempt.
stillA, err := store.Get(ctx, orgA, pa.ID)
require.NoError(t, err)
assert.Equal(t, 1, stillA.KeyVersion)
// Each org lists only its own.
la, _ := store.List(ctx, orgA)
lb, _ := store.List(ctx, orgB)
require.Len(t, la, 1)
require.Len(t, lb, 1)
assert.Equal(t, pa.ID, la[0].ID)
assert.Equal(t, pb.ID, lb[0].ID)
}
func TestProjectStore_ResolveUnknownFailsClosed(t *testing.T) {
ctx := context.Background()
store := NewProjectStore(newTestSQLStore(t))
_, _, _, found, err := store.Resolve(ctx, valuer.GenerateUUID())
require.NoError(t, err)
assert.False(t, found, "an unknown project must resolve to found=false, never a tenant")
}
+87
View File
@@ -0,0 +1,87 @@
// Package sentry is Hanzo Sentry — the Sentry-parity error/log/trace product face
// under /v1/sentry. It is a COMPOSITION, not a refork: the ingest engine (envelope
// parse, fingerprint, DSN verify, scrub, rate-limit) is the reused errortracking
// engine; identity is Hanzo IAM; storage is the ONE datastore (columnar events) plus
// o11y_issues (grouped-issue lifecycle, reused verbatim). This package owns only the
// product surface: projects, the events plane, and the read/query shapes that give
// Discover / logs / traces / stats their Sentry semantics.
package sentry
import (
"context"
"net/http"
"github.com/hanzoai/o11y/pkg/types/errortrackingtypes"
"github.com/hanzoai/o11y/pkg/types/sentrytypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
// Module is the org-scoped business surface. Ingest is the only method that takes a
// project (resolved from the DSN, no principal); every other method is scoped to the
// caller's org and validates any project against it.
type Module interface {
// Ingest persists a request's occurrences for (org, project): the columnar events
// plane (Insert) AND the grouped-issue lifecycle (reused errortracking upsert).
Ingest(ctx context.Context, orgID, projectID valuer.UUID, occs []*errortrackingtypes.Occurrence) error
// Projects — org-scoped CRUD + DSN rotation. Create/Get/List/Rotate all stamp or
// filter org_id; the DSN is derived, never stored.
CreateProject(ctx context.Context, orgID valuer.UUID, in *sentrytypes.PostableProject) (*sentrytypes.GettableProject, error)
ListProjects(ctx context.Context, orgID valuer.UUID) (*sentrytypes.GettableProjects, error)
GetProject(ctx context.Context, orgID, id valuer.UUID) (*sentrytypes.GettableProject, error)
RotateProjectKey(ctx context.Context, orgID, id valuer.UUID) (*sentrytypes.GettableProject, error)
// ResolveIngest maps a DSN project id to its owning org, verifying the presented
// DSN key against the project's rotation watermark. Fail-closed: an unknown,
// disabled or below-watermark project/key returns ok=false.
ResolveIngest(ctx context.Context, projectID valuer.UUID, presentedKey string) (orgID valuer.UUID, ok bool)
// RateAllow reports whether the project is within its ingest rate budget.
RateAllow(projectID valuer.UUID) bool
// Issues — reused errortracking lifecycle, org-scoped, optionally narrowed to a
// project via the events-plane fingerprint projection.
ListIssues(ctx context.Context, orgID valuer.UUID, projectID *valuer.UUID, q *errortrackingtypes.IssuesQuery, w sentrytypes.Window) (*errortrackingtypes.GettableIssues, 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)
// IssueEvents lists an issue's occurrences scoped to (org, project) — a project is
// an isolation unit, so the caller declares which project's occurrences to read.
IssueEvents(ctx context.Context, orgID, id, projectID valuer.UUID, limit int) ([]*sentrytypes.Event, error)
// Discover / event detail / logs / traces / stats — all over the events plane.
Discover(ctx context.Context, orgID valuer.UUID, req *sentrytypes.DiscoverRequest) (*sentrytypes.DiscoverResult, error)
GetEvent(ctx context.Context, orgID, projectID valuer.UUID, eventID string) (*sentrytypes.Event, error)
ListLogs(ctx context.Context, orgID valuer.UUID, projectID valuer.UUID, query, period string, limit int) ([]*sentrytypes.Event, error)
ListTraces(ctx context.Context, orgID valuer.UUID, projectID valuer.UUID, period string, limit int) ([]*sentrytypes.TraceSummary, error)
TraceDetail(ctx context.Context, orgID, projectID valuer.UUID, traceID string) (any, error)
Stats(ctx context.Context, orgID, projectID valuer.UUID, field, period string) ([]sentrytypes.StatsPoint, error)
}
// Handler is the HTTP surface under /v1/sentry. The two ingest endpoints are PUBLIC
// (DSN-authenticated in-handler); everything else is behind Hanzo IAM authz and
// org-scoped from the validated claims.
type Handler interface {
// Ingest (public, DSN-auth): POST /v1/sentry/{project}/envelope|store/.
EnvelopeIngest(http.ResponseWriter, *http.Request)
StoreIngest(http.ResponseWriter, *http.Request)
// Projects.
ListProjects(http.ResponseWriter, *http.Request)
CreateProject(http.ResponseWriter, *http.Request)
GetProject(http.ResponseWriter, *http.Request)
RotateProjectKey(http.ResponseWriter, *http.Request)
// Issues.
ListIssues(http.ResponseWriter, *http.Request)
GetIssue(http.ResponseWriter, *http.Request)
UpdateIssue(http.ResponseWriter, *http.Request)
IssueEvents(http.ResponseWriter, *http.Request)
// Discover / events / logs / traces / stats.
Discover(http.ResponseWriter, *http.Request)
GetEvent(http.ResponseWriter, *http.Request)
ListLogs(http.ResponseWriter, *http.Request)
ListTraces(http.ResponseWriter, *http.Request)
GetTrace(http.ResponseWriter, *http.Request)
Stats(http.ResponseWriter, *http.Request)
}
+14
View File
@@ -45,6 +45,8 @@ import (
"github.com/hanzoai/o11y/pkg/modules/rulestatehistory/implrulestatehistory"
"github.com/hanzoai/o11y/pkg/modules/savedview"
"github.com/hanzoai/o11y/pkg/modules/savedview/implsavedview"
"github.com/hanzoai/o11y/pkg/modules/sentry"
"github.com/hanzoai/o11y/pkg/modules/sentry/implsentry"
"github.com/hanzoai/o11y/pkg/modules/serviceaccount"
"github.com/hanzoai/o11y/pkg/modules/serviceaccount/implserviceaccount"
"github.com/hanzoai/o11y/pkg/modules/services"
@@ -95,6 +97,7 @@ type Handlers struct {
LLMPricingRuleHandler llmpricingrule.Handler
LLMObsHandler llmobs.Handler
ErrorTrackingHandler errortracking.Handler
SentryHandler sentry.Handler
StatsHandler statsreporter.Handler
}
@@ -145,6 +148,7 @@ func NewHandlers(
LLMPricingRuleHandler: impllmpricingrule.NewHandler(modules.LLMPricingRule),
LLMObsHandler: impllmobs.NewHandler(modules.LLMObs),
ErrorTrackingHandler: implerrortracking.NewHandler(modules.ErrorTracking, errorTrackingIngestSecret(), errorTrackingCapturePII(), modules.ErrorTrackingRevocations),
SentryHandler: implsentry.NewHandler(modules.Sentry, len(errorTrackingIngestSecret()) > 0, errorTrackingCapturePII()),
StatsHandler: statsreporter.NewHandler(statsAggregator),
}
}
@@ -166,6 +170,16 @@ func errorTrackingCapturePII() bool {
return v == "true" || v == "1" || v == "yes"
}
// sentryIngestHost is the DSN endpoint origin the Sentry product mints into project
// DSNs (https://<key>@<host>/v1/sentry/<project>). Defaults to the public API host;
// O11Y_SENTRY_INGEST_HOST overrides per deployment/brand.
func sentryIngestHost() string {
if v := strings.TrimSpace(os.Getenv("O11Y_SENTRY_INGEST_HOST")); v != "" {
return v
}
return "api.hanzo.ai"
}
// 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 {
+58 -34
View File
@@ -43,6 +43,8 @@ import (
"github.com/hanzoai/o11y/pkg/modules/rulestatehistory/implrulestatehistory"
"github.com/hanzoai/o11y/pkg/modules/savedview"
"github.com/hanzoai/o11y/pkg/modules/savedview/implsavedview"
"github.com/hanzoai/o11y/pkg/modules/sentry"
"github.com/hanzoai/o11y/pkg/modules/sentry/implsentry"
"github.com/hanzoai/o11y/pkg/modules/serviceaccount"
"github.com/hanzoai/o11y/pkg/modules/services"
"github.com/hanzoai/o11y/pkg/modules/services/implservices"
@@ -103,7 +105,12 @@ type Modules struct {
// 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
// Sentry is the /v1/sentry product face: it COMPOSES the reused errortracking
// engine + issue lifecycle, the columnar events plane (telemetryStore) and the
// reused tracedetail read. Built here because it needs BOTH the sqlstore (projects)
// and the telemetryStore (events plane).
Sentry sentry.Module
Tag tag.Module
}
func NewModules(
@@ -142,41 +149,58 @@ func NewModules(
ruleStore := sqlrulestore.NewRuleStore(sqlstore, queryParser, providerSettings)
authDomainModule := implauthdomain.NewModule(implauthdomain.NewStore(sqlstore), authNs, authz)
// Error tracking (o11y_issues lifecycle) and trace detail (o11y_traces waterfall)
// are pulled into locals so the Sentry product face can COMPOSE them rather than
// reconstruct them — one issue lifecycle, one trace read, two product faces.
errorTrackingModule := implerrortracking.NewModule(
implerrortracking.NewStore(sqlstore),
implerrortracking.NewNoopSink(),
implerrortracking.WithRetention(errorTrackingRetention()),
)
traceDetailModule := impltracedetail.NewModule(impltracedetail.NewTraceStore(telemetryStore), providerSettings, config.TraceDetail)
sentryModule := implsentry.NewModule(
implsentry.NewProjectStore(sqlstore),
implsentry.NewEventStore(telemetryStore),
errorTrackingModule,
implsentry.Config{
IngestSecret: errorTrackingIngestSecret(),
Host: sentryIngestHost(),
CapturePII: errorTrackingCapturePII(),
},
)
return Modules{
OrgGetter: orgGetter,
OrgSetter: orgSetter,
Preference: implpreference.NewModule(implpreference.NewStore(sqlstore), preferencetypes.NewAvailablePreference()),
SavedView: implsavedview.NewModule(sqlstore),
Apdex: implapdex.NewModule(sqlstore),
Dashboard: dashboard,
UserSetter: userSetter,
UserGetter: userGetter,
RetentionGetter: retentionGetter,
QuickFilter: quickfilter,
TraceFunnel: impltracefunnel.NewModule(impltracefunnel.NewStore(sqlstore)),
RawDataExport: implrawdataexport.NewModule(querier),
AuthDomain: authDomainModule,
Session: implsession.NewModule(providerSettings, authNs, userSetter, userGetter, authDomainModule, tokenizer, orgGetter, authz),
SpanPercentile: implspanpercentile.NewModule(querier, providerSettings),
Services: implservices.NewModule(querier, telemetryStore),
MetricsExplorer: implmetricsexplorer.NewModule(telemetryStore, telemetryMetadataStore, cache, ruleStore, dashboard, fl, providerSettings, config.MetricsExplorer),
MetricReductionRule: metricReductionRule,
InfraMonitoring: implinframonitoring.NewModule(telemetryStore, telemetryMetadataStore, querier, fl, providerSettings, config.InfraMonitoring),
Promote: implpromote.NewModule(telemetryMetadataStore, telemetryStore),
ServiceAccount: serviceAccount,
LogsPipeline: impllogspipeline.NewModule(sqlstore),
RuleStateHistory: implrulestatehistory.NewModule(implrulestatehistory.NewStore(telemetryStore, telemetryMetadataStore, providerSettings.Logger)),
CloudIntegration: cloudIntegrationModule,
TraceDetail: impltracedetail.NewModule(impltracedetail.NewTraceStore(telemetryStore), providerSettings, config.TraceDetail),
SpanMapper: implspanmapper.NewModule(implspanmapper.NewStore(sqlstore), fl),
LLMPricingRule: impllmpricingrule.NewModule(impllmpricingrule.NewStore(sqlstore), fl),
LLMObs: impllmobs.NewModule(querier, impllmobs.NewStore(sqlstore)),
ErrorTracking: implerrortracking.NewModule(
implerrortracking.NewStore(sqlstore),
implerrortracking.NewNoopSink(),
implerrortracking.WithRetention(errorTrackingRetention()),
),
OrgGetter: orgGetter,
OrgSetter: orgSetter,
Preference: implpreference.NewModule(implpreference.NewStore(sqlstore), preferencetypes.NewAvailablePreference()),
SavedView: implsavedview.NewModule(sqlstore),
Apdex: implapdex.NewModule(sqlstore),
Dashboard: dashboard,
UserSetter: userSetter,
UserGetter: userGetter,
RetentionGetter: retentionGetter,
QuickFilter: quickfilter,
TraceFunnel: impltracefunnel.NewModule(impltracefunnel.NewStore(sqlstore)),
RawDataExport: implrawdataexport.NewModule(querier),
AuthDomain: authDomainModule,
Session: implsession.NewModule(providerSettings, authNs, userSetter, userGetter, authDomainModule, tokenizer, orgGetter, authz),
SpanPercentile: implspanpercentile.NewModule(querier, providerSettings),
Services: implservices.NewModule(querier, telemetryStore),
MetricsExplorer: implmetricsexplorer.NewModule(telemetryStore, telemetryMetadataStore, cache, ruleStore, dashboard, fl, providerSettings, config.MetricsExplorer),
MetricReductionRule: metricReductionRule,
InfraMonitoring: implinframonitoring.NewModule(telemetryStore, telemetryMetadataStore, querier, fl, providerSettings, config.InfraMonitoring),
Promote: implpromote.NewModule(telemetryMetadataStore, telemetryStore),
ServiceAccount: serviceAccount,
LogsPipeline: impllogspipeline.NewModule(sqlstore),
RuleStateHistory: implrulestatehistory.NewModule(implrulestatehistory.NewStore(telemetryStore, telemetryMetadataStore, providerSettings.Logger)),
CloudIntegration: cloudIntegrationModule,
TraceDetail: traceDetailModule,
SpanMapper: implspanmapper.NewModule(implspanmapper.NewStore(sqlstore), fl),
LLMPricingRule: impllmpricingrule.NewModule(impllmpricingrule.NewStore(sqlstore), fl),
LLMObs: impllmobs.NewModule(querier, impllmobs.NewStore(sqlstore)),
ErrorTracking: errorTrackingModule,
ErrorTrackingRevocations: implerrortracking.NewSQLRevocations(sqlstore),
Sentry: sentryModule,
Tag: tagModule,
}
}
+2
View File
@@ -32,6 +32,7 @@ import (
"github.com/hanzoai/o11y/pkg/modules/promote"
"github.com/hanzoai/o11y/pkg/modules/rawdataexport"
"github.com/hanzoai/o11y/pkg/modules/rulestatehistory"
"github.com/hanzoai/o11y/pkg/modules/sentry"
"github.com/hanzoai/o11y/pkg/modules/serviceaccount"
"github.com/hanzoai/o11y/pkg/modules/session"
"github.com/hanzoai/o11y/pkg/modules/spanmapper"
@@ -91,6 +92,7 @@ func NewOpenAPI(ctx context.Context, instrumentation instrumentation.Instrumenta
struct{ statsreporter.Handler }{},
struct{ llmobs.Handler }{},
struct{ errortracking.Handler }{},
struct{ sentry.Handler }{},
).New(ctx, instrumentation.ToProviderSettings(), apiserver.Config{})
if err != nil {
return nil, err
+2
View File
@@ -221,6 +221,7 @@ func NewSQLMigrationProviderFactories(
sqlmigration.NewRemoveOrganizationTuplesFactory(sqlstore),
sqlmigration.NewAddLLMObsFactory(sqlstore, sqlschema),
sqlmigration.NewAddErrorTrackingFactory(sqlstore, sqlschema),
sqlmigration.NewAddSentryProjectsFactory(sqlstore, sqlschema),
)
}
@@ -321,6 +322,7 @@ func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.Au
handlers.StatsHandler,
handlers.LLMObsHandler,
handlers.ErrorTrackingHandler,
handlers.SentryHandler,
),
)
}
+10
View File
@@ -231,6 +231,16 @@ func (s *Server) createPublicServer(api *APIHandler, web web.Web) (*http.Server,
return
}
// Hanzo Sentry is served under the CLEAN /v1/sentry contract (no /api/,
// no /v1/o11y rewrite): its routes are registered on r at their literal
// /v1/sentry/… path. StripPrefix(routePrefix=/v1/o11y) would 404 them, so
// pass them straight to the router — the same escape hatch the /api/ paths
// use, for the same reason.
if strings.HasPrefix(req.URL.Path, "/v1/sentry/") {
r.ServeHTTP(w, req)
return
}
prefixed.ServeHTTP(w, req)
})
}
@@ -0,0 +1,81 @@
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"
)
// addSentryProjects creates o11y_sentry_projects — the DSN-bearing unit of the Hanzo
// Sentry product face. It is a thin relational row: the DSN is NEVER stored (it is
// derived from the platform ingest secret + id + key_version), so rotation is a
// single-row bump. Raw events live on the columnar datastore plane and grouped-issue
// lifecycle stays in o11y_issues — this table only names projects and carries the DSN
// key watermark. The unique index on (org_id, slug) is the per-org name key; the index
// on (org_id) serves the org-scoped project list.
type addSentryProjects struct {
sqlschema sqlschema.SQLSchema
sqlstore sqlstore.SQLStore
}
func NewAddSentryProjectsFactory(sqlstore sqlstore.SQLStore, sqlschema sqlschema.SQLSchema) factory.ProviderFactory[SQLMigration, Config] {
return factory.NewProviderFactory(factory.MustNewName("add_sentry_projects"), func(_ context.Context, _ factory.ProviderSettings, _ Config) (SQLMigration, error) {
return &addSentryProjects{sqlschema: sqlschema, sqlstore: sqlstore}, nil
})
}
func (migration *addSentryProjects) Register(migrations *migrate.Migrations) error {
return migrations.Register(migration.Up, migration.Down)
}
func (migration *addSentryProjects) Up(ctx context.Context, db *bun.DB) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
orgFK := &sqlschema.ForeignKeyConstraint{
ReferencingColumnName: sqlschema.ColumnName("org_id"),
ReferencedTableName: sqlschema.TableName("organizations"),
ReferencedColumnName: sqlschema.ColumnName("id"),
}
sqls := migration.sqlschema.Operator().CreateTable(&sqlschema.Table{
Name: "o11y_sentry_projects",
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: "name", DataType: sqlschema.DataTypeText, Nullable: false},
{Name: "slug", DataType: sqlschema.DataTypeText, Nullable: false},
{Name: "platform", DataType: sqlschema.DataTypeText, Nullable: true},
{Name: "status", DataType: sqlschema.DataTypeText, Nullable: false, Default: "'active'"},
{Name: "key_version", DataType: sqlschema.DataTypeBigInt, Nullable: false, Default: "1"},
},
PrimaryKeyConstraint: &sqlschema.PrimaryKeyConstraint{ColumnNames: []sqlschema.ColumnName{"id"}},
ForeignKeyConstraints: []*sqlschema.ForeignKeyConstraint{orgFK},
})
sqls = append(sqls,
[]byte(`CREATE UNIQUE INDEX IF NOT EXISTS uq_o11y_sentry_projects_org_slug ON o11y_sentry_projects (org_id, slug)`),
[]byte(`CREATE INDEX IF NOT EXISTS idx_o11y_sentry_projects_org ON o11y_sentry_projects (org_id)`),
)
for _, sql := range sqls {
if _, err := tx.ExecContext(ctx, string(sql)); err != nil {
return err
}
}
return tx.Commit()
}
func (migration *addSentryProjects) Down(context.Context, *bun.DB) error {
return nil
}
+7
View File
@@ -101,6 +101,13 @@ type IssuesQuery struct {
Sort string `query:"sort" json:"sort"`
Offset int `query:"offset" json:"offset"`
Limit int `query:"limit" json:"limit"`
// Fingerprints is a SERVER-ONLY narrowing set (note the absence of a `query` tag:
// no URL param can populate it). The Hanzo Sentry face sets it to a project's
// fingerprints — derived from the org+project-scoped events plane — so an org's
// grouped issue list can be projected to one project without widening scope. Empty
// leaves the list org-scoped, exactly as before.
Fingerprints []string `json:"-"`
}
type GettableIssues struct {
+166
View File
@@ -0,0 +1,166 @@
// Package sentrytypes holds the value types and store seams for Hanzo Sentry — the
// Sentry-parity error/log/trace product face served under /v1/sentry. It COMPOSES
// the shared observability substrate rather than reforking it:
//
// - Projects are the DSN-bearing unit under an IAM org (relational lifecycle).
// - Raw error EVENTS are columnar rows on the ONE datastore (high-volume, queried
// by Discover / events / stats / logs / traces).
// - Grouped ISSUE lifecycle stays in o11y_issues (errortracking, reused verbatim).
//
// Every read is org-scoped from the validated IAM principal; the client never names
// its own tenant. A project param is always validated against the caller's org
// before it scopes a query, so a foreign project id returns zero rows, not a leak.
package sentrytypes
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 (
ErrCodeSentryInvalidInput = errors.MustNewCode("sentry_invalid_input")
ErrCodeSentryNotFound = errors.MustNewCode("sentry_not_found")
ErrCodeSentryUnauthorized = errors.MustNewCode("sentry_unauthorized")
ErrCodeSentryDisabled = errors.MustNewCode("sentry_disabled")
ErrCodeSentryConflict = errors.MustNewCode("sentry_conflict")
)
// ProjectStatus is a project's lifecycle state. A revoked/archived project fails
// ingest closed (its DSN stops verifying) without deleting its history.
type ProjectStatus string
const (
ProjectActive ProjectStatus = "active"
ProjectDisabled ProjectStatus = "disabled"
)
// Project is a DSN-bearing unit under an IAM org. It is a thin relational row: the
// DSN itself is NOT stored — it is derived on demand from the platform ingest secret
// (KMS) + the project id + KeyVersion, so rotating a project is a single-row bump
// with no secret at rest. Tenancy: OrgID is the mandatory boundary; every store
// query filters org_id.
type Project struct {
bun.BaseModel `bun:"table:o11y_sentry_projects,alias:o11y_sentry_projects" json:"-"`
types.Identifiable
types.TimeAuditable
OrgID valuer.UUID `bun:"org_id,type:text,notnull" json:"-"`
Name string `bun:"name,type:text,notnull" json:"name"`
Slug string `bun:"slug,type:text,notnull" json:"slug"`
Platform string `bun:"platform,type:text" json:"platform,omitempty"`
Status ProjectStatus `bun:"status,type:text,notnull,default:'active'" json:"status"`
// KeyVersion is the per-project DSN rotation watermark. A DSN key is
// "<version>:<hmac>"; a key whose version is below KeyVersion no longer verifies.
// Rotation bumps this by one — isolated to THIS project, no global secret roll and
// no shared revocation table.
KeyVersion int `bun:"key_version,type:bigint,notnull,default:1" json:"-"`
}
// GettableProject is the API view of a project including its freshly-derived DSN.
type GettableProject struct {
*Project
DSN string `json:"dsn"`
}
// PostableProject creates a project. Only Name (and optional Slug/Platform) are
// client-supplied; org/id/dsn/key are server-assigned.
type PostableProject struct {
Name string `json:"name"`
Slug string `json:"slug,omitempty"`
Platform string `json:"platform,omitempty"`
}
type GettableProjects struct {
Items []*GettableProject `json:"items" required:"true"`
Total int `json:"total" required:"true"`
}
// Event is one columnar error occurrence on the datastore events plane. It carries
// exactly the fields Discover / events / stats / logs / traces need, org+project
// scoped and timestamp-bucketed. It is the realized "raw error events" sink the
// errortracking OccurrenceSink note deferred (that seam was org-only; the events
// plane needs the project dimension, so it lives here in the product face).
type Event struct {
OrgID string `json:"orgId"`
ProjectID string `json:"projectId"`
EventID string `json:"eventId"`
Timestamp time.Time `json:"timestamp"`
ReceivedAt time.Time `json:"receivedAt"`
Level string `json:"level"`
Type string `json:"type"`
Value string `json:"value"`
Message string `json:"message"`
Culprit string `json:"culprit"`
Fingerprint string `json:"fingerprint"`
Platform string `json:"platform,omitempty"`
Environment string `json:"environment,omitempty"`
Release string `json:"release,omitempty"`
ServiceName string `json:"serviceName,omitempty"`
Transaction string `json:"transaction,omitempty"`
TraceID string `json:"traceId,omitempty"`
SpanID string `json:"spanId,omitempty"`
ServerName string `json:"serverName,omitempty"`
UserID string `json:"userId,omitempty"`
UserEmail string `json:"userEmail,omitempty"`
UserIP string `json:"userIp,omitempty"`
Tags map[string]string `json:"tags,omitempty"`
Sample string `json:"-"` // full normalized Occurrence JSON, for event detail
}
// Window is a resolved absolute time range [From, To]. Every columnar read is bounded
// by a window so a query can never scan the whole retention.
type Window struct {
From time.Time
To time.Time
}
// DiscoverFilter is one equality/like predicate. Field is resolved against the
// column allowlist (or a validated tags[key]) — never interpolated; Value is always
// a bound parameter.
type DiscoverFilter struct {
Field string `json:"field"`
Op string `json:"op"` // eq | neq | like
Value string `json:"value"`
}
// DiscoverRequest is a columnar aggregation over the events plane. Project is
// mandatory and validated against the caller's org before it scopes the scan.
type DiscoverRequest struct {
Project string `json:"project"`
Filters []DiscoverFilter `json:"filters,omitempty"`
Aggregations []string `json:"aggregations,omitempty"` // allowlist keys; empty => count
GroupBy []string `json:"groupBy,omitempty"` // allowlist column keys
Period string `json:"period,omitempty"` // relative window, e.g. 1h|24h|7d|14d|30d
OrderBy string `json:"orderBy,omitempty"` // a groupBy key or an aggregation key
OrderDir string `json:"orderDir,omitempty"` // asc | desc
Limit int `json:"limit,omitempty"`
}
// DiscoverResult is a tabular result: named columns and value rows, in column order.
type DiscoverResult struct {
Columns []string `json:"columns"`
Rows [][]any `json:"rows"`
}
// StatsPoint is one bucket of an event-rate timeseries.
type StatsPoint struct {
Time time.Time `json:"time"`
Value uint64 `json:"value"`
}
// TraceSummary is an error-correlated trace: the trace id plus the count and span of
// captured error events that referenced it, for the project.
type TraceSummary struct {
TraceID string `json:"traceId"`
Count uint64 `json:"count"`
FirstSeen time.Time `json:"firstSeen"`
LastSeen time.Time `json:"lastSeen"`
Sample string `json:"sample,omitempty"`
}
+69
View File
@@ -0,0 +1,69 @@
package sentrytypes
import (
"context"
"github.com/hanzoai/o11y/pkg/valuer"
)
// ProjectStore persists o11y_sentry_projects. Every method is org-scoped EXCEPT
// Resolve, the ingest-time lookup that maps an unguessable project id to its owning
// org + key watermark (ingest carries no IAM principal — it authenticates the DSN).
type ProjectStore interface {
Create(ctx context.Context, p *Project) error
List(ctx context.Context, orgID valuer.UUID) ([]*Project, error)
Get(ctx context.Context, orgID, id valuer.UUID) (*Project, error)
// Rotate bumps the project's KeyVersion (invalidating below-watermark DSNs) and
// returns the new version, org-scoped and idempotent-safe.
Rotate(ctx context.Context, orgID, id valuer.UUID) (int, error)
// Resolve maps a project id to its owning org, current key version and status —
// the ONLY non-org-scoped read, used by the public DSN-authenticated ingest path.
// Fail-closed: an unknown project returns found=false.
Resolve(ctx context.Context, id valuer.UUID) (orgID valuer.UUID, keyVersion int, status ProjectStatus, found bool, err error)
}
// EventStore is the columnar events plane on the ONE datastore. Insert is the finished
// ingest sink; the reads back Discover / event detail / issue occurrences / logs /
// traces / stats. Every read takes the org (mandatory tenant boundary) and a project
// as separate, server-validated arguments — no query shape carries a client-named
// tenant.
type EventStore interface {
// Insert writes a batch of occurrences for one (org, project). Fail-soft is the
// caller's contract: the durable issue upsert must not depend on this write.
Insert(ctx context.Context, orgID, projectID valuer.UUID, events []*Event) error
// Discover runs a bounded, allowlist-checked aggregation scoped to (org, project).
Discover(ctx context.Context, orgID, projectID valuer.UUID, req *DiscoverRequest, w Window) (*DiscoverResult, error)
// GetEvent returns one event by id within (org, project) — a project is an
// isolation unit, so a cross-project id in the same tenant returns not-found.
// Not found => (nil, nil).
GetEvent(ctx context.Context, orgID, projectID valuer.UUID, eventID string) (*Event, error)
// ListForFingerprint returns the latest occurrences of an issue (by fingerprint)
// within (org, project), newest first.
ListForFingerprint(ctx context.Context, orgID, projectID valuer.UUID, fingerprint string, limit int) ([]*Event, error)
// ListForTrace returns the (org, project)-scoped error events referencing a trace
// id — the tenant-safe "errors in this trace" detail (the o11y_traces span plane is
// NOT read: it has no general org column and cannot be tenant-scoped).
ListForTrace(ctx context.Context, orgID, projectID valuer.UUID, traceID string, limit int) ([]*Event, error)
// DistinctFingerprints returns the set of issue fingerprints seen for (org,
// project) inside the window — the project→issue projection used to scope the
// org-grouped issue list to a project.
DistinctFingerprints(ctx context.Context, orgID, projectID valuer.UUID, w Window) ([]string, error)
// ListLogs returns error-event log lines for (org, project), newest first,
// optionally narrowed by a free-text query over message/value.
ListLogs(ctx context.Context, orgID, projectID valuer.UUID, query string, w Window, limit int) ([]*Event, error)
// ListTraces returns error-correlated traces for (org, project) in the window.
ListTraces(ctx context.Context, orgID, projectID valuer.UUID, w Window, limit int) ([]*TraceSummary, error)
// Stats returns an event-count timeseries bucketed across the window for (org,
// project). field selects the counted subset (allowlist).
Stats(ctx context.Context, orgID, projectID valuer.UUID, field string, w Window) ([]StatsPoint, error)
}