audit

package
v1.101.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 11, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

Documentation

Overview

Package audit provides audit logging for the platform.

Index

Constants

View Source 
const (
	// DefaultAsyncQueueCapacity is the bounded-queue depth used when no
	// capacity option is supplied. Sized so a brief store stall (a slow
	// transaction, a failover) buffers rather than drops, while a
	// sustained outage sheds load instead of growing without bound.
	DefaultAsyncQueueCapacity = 1024

	// DefaultAsyncWriteTimeout bounds a single store write. The store must
	// honor context cancellation (the PostgreSQL store's ExecContext
	// does); a write that exceeds this deadline is abandoned and the
	// writer proceeds to the next event.
	DefaultAsyncWriteTimeout = 5 * time.Second
)

Variables

View Source 
var ValidBreakdownDimensions = map[BreakdownDimension]bool{
	BreakdownByToolName:    true,
	BreakdownByUserID:      true,
	BreakdownByPersona:     true,
	BreakdownByToolkitKind: true,
	BreakdownByConnection:  true,
}

ValidBreakdownDimensions is the set of allowed group-by values.

View Source 
var ValidResolutions = map[Resolution]bool{
	ResolutionMinute: true,
	ResolutionHour:   true,
	ResolutionDay:    true,
}

ValidResolutions is the set of allowed resolution values.

View Source 
var ValidSortColumns = map[string]bool{
	"timestamp":          true,
	"user_id":            true,
	"tool_name":          true,
	"toolkit_kind":       true,
	"source":             true,
	"connection":         true,
	"duration_ms":        true,
	"success":            true,
	"enrichment_applied": true,
	"enrichment_mode":    true,
}

ValidSortColumns lists columns that can be used for ORDER BY.

Functions

func SanitizeParameters 

func SanitizeParameters(params map[string]any) map[string]any

SanitizeParameters removes sensitive parameters from the event.

Types

type AsyncOption added in v1.101.0 

type AsyncOption func(*AsyncWriter)

AsyncOption configures an AsyncWriter.

func WithMetrics added in v1.101.0 

func WithMetrics(m *observability.Metrics) AsyncOption

WithMetrics attaches the observability recorder used to count dropped events. A nil recorder is accepted; observability.Metrics is itself nil-safe, so the writer works unchanged in deployments without metrics wired.

func WithQueueCapacity added in v1.101.0 

func WithQueueCapacity(n int) AsyncOption

WithQueueCapacity sets the bounded queue depth. Non-positive values are ignored so the default capacity stands.

func WithWriteTimeout added in v1.101.0 

func WithWriteTimeout(d time.Duration) AsyncOption

WithWriteTimeout sets the per-write timeout. Non-positive values are ignored so the default timeout stands.

type AsyncWriter added in v1.101.0 

type AsyncWriter struct {
	// contains filtered or unexported fields
}

AsyncWriter decouples audit emission from storage latency. Events are queued on a bounded channel and written by a single goroutine with a per-write timeout. When the queue is full the event is dropped and counted; audit delivery is best-effort by design.

The single-goroutine design is the bound: no matter how many concurrent tool calls enqueue or how long the store stalls, exactly one write goroutine exists, so a stalled database can never accumulate goroutines. Compare the old per-call detached goroutine, which grew without limit under a stalled store (issue #884).

func NewAsyncWriter added in v1.101.0 

func NewAsyncWriter(logger Logger, opts ...AsyncOption) *AsyncWriter

NewAsyncWriter wraps logger in a bounded async writer and starts its single drain goroutine. Callers must call Close to stop intake and drain the queue.

func (*AsyncWriter) Close added in v1.101.0 

func (w *AsyncWriter) Close(ctx context.Context) error

Close stops accepting new events and drains the queue, returning when the queue is empty or ctx expires (whichever comes first). When ctx expires with events still queued, baseCtx is canceled so the drain goroutine abandons the backlog immediately instead of writing into a store the caller is about to close; the abandoned events are counted as lost. Close is idempotent and safe to call concurrently with Log.

func (*AsyncWriter) Dropped added in v1.101.0 

func (w *AsyncWriter) Dropped() int64

Dropped returns the cumulative number of lost events: queue-full drops plus writes that failed or were abandoned at the per-write timeout.

func (*AsyncWriter) Log added in v1.101.0 

func (w *AsyncWriter) Log(_ context.Context, e Event) error

Log enqueues e without blocking. When the queue is full (or the writer is closing) the event is dropped, counted, and logged; the caller is never blocked or failed. The returned error is always nil — audit delivery is best-effort — but the signature satisfies the Logger contract so an AsyncWriter can wrap any Logger transparently.

func (*AsyncWriter) QueueDepth added in v1.101.0 

func (w *AsyncWriter) QueueDepth() int

QueueDepth returns the number of events currently queued. Exposed for tests and diagnostics; it is a point-in-time snapshot with no ordering guarantee.

type BreakdownDimension added in v0.17.1 

type BreakdownDimension string

BreakdownDimension defines valid group-by dimensions. 

const (
	// BreakdownByToolName groups by tool name.
	BreakdownByToolName BreakdownDimension = "tool_name"

	// BreakdownByUserID groups by user ID.
	BreakdownByUserID BreakdownDimension = "user_id"

	// BreakdownByPersona groups by persona.
	BreakdownByPersona BreakdownDimension = "persona"

	// BreakdownByToolkitKind groups by toolkit kind.
	BreakdownByToolkitKind BreakdownDimension = "toolkit_kind"

	// BreakdownByConnection groups by connection.
	BreakdownByConnection BreakdownDimension = "connection"
)

type BreakdownEntry added in v0.17.1 

type BreakdownEntry struct {
	Dimension     string  `json:"dimension" example:"trino_query"`
	Count         int     `json:"count" example:"65"`
	SuccessRate   float64 `json:"success_rate" example:"0.95"`
	AvgDurationMS float64 `json:"avg_duration_ms" example:"320.0"`
}

BreakdownEntry holds aggregated stats for a single dimension value.

type BreakdownFilter added in v0.17.1 

type BreakdownFilter struct {
	GroupBy   BreakdownDimension
	Limit     int
	StartTime *time.Time
	EndTime   *time.Time
	UserID    string
	// ToolName scopes the aggregation to a specific tool. Set when a
	// caller wants per-tool stats regardless of breakdown ranking — on
	// platforms with more than Limit distinct tools active in the
	// window, low-frequency tools would otherwise fall off the top-N
	// breakdown and appear as "no calls recorded" even when they have
	// activity (#343 bug 2).
	ToolName string
	// EventKind scopes the aggregation to a specific event category.
	// Lets the portal MCP view exclude apigateway noise without
	// modifying the breakdown call sites.
	EventKind string
}

BreakdownFilter controls breakdown query parameters.

type Config 

type Config struct {
	Enabled       bool
	LogToolCalls  bool
	RetentionDays int
}

Config configures audit logging.

type DiscoveryStats added in v0.25.0 

type DiscoveryStats struct {
	TotalSessions         int              `json:"total_sessions" example:"100"`
	DiscoverySessions     int              `json:"discovery_sessions" example:"75"`
	QuerySessions         int              `json:"query_sessions" example:"80"`
	DiscoveryBeforeQuery  int              `json:"discovery_before_query" example:"60"`
	DiscoveryRate         float64          `json:"discovery_rate" example:"0.75"`
	QueryWithoutDiscovery int              `json:"query_without_discovery" example:"20"`
	TopDiscoveryTools     []BreakdownEntry `json:"top_discovery_tools"`
}

DiscoveryStats holds discovery-before-query pattern statistics.

type EnrichmentStats added in v0.25.0 

type EnrichmentStats struct {
	TotalCalls       int     `json:"total_calls" example:"1500"`
	EnrichedCalls    int     `json:"enriched_calls" example:"1200"`
	EnrichmentRate   float64 `json:"enrichment_rate" example:"0.80"`
	FullCount        int     `json:"full_count" example:"800"`
	SummaryCount     int     `json:"summary_count" example:"300"`
	ReferenceCount   int     `json:"reference_count" example:"100"`
	NoneCount        int     `json:"none_count" example:"0"`
	TotalTokensFull  int64   `json:"total_tokens_full" example:"450000"`
	TotalTokensDedup int64   `json:"total_tokens_dedup" example:"120000"`
	TokensSaved      int64   `json:"tokens_saved" example:"330000"`
	AvgTokensFull    float64 `json:"avg_tokens_full" example:"375.0"`
	AvgTokensDedup   float64 `json:"avg_tokens_dedup" example:"100.0"`
	UniqueSessions   int     `json:"unique_sessions" example:"45"`
}

EnrichmentStats holds aggregate enrichment statistics.

type Event 

type Event struct {
	ID                    string         `json:"id" example:"evt_a1b2c3d4e5f6"`
	Timestamp             time.Time      `json:"timestamp" example:"2026-04-15T10:41:18Z"`
	DurationMS            int64          `json:"duration_ms" example:"143"`
	RequestID             string         `json:"request_id" example:"req_x9y8z7"`
	SessionID             string         `json:"session_id" example:"sess_abc123"`
	UserID                string         `json:"user_id" example:"550e8400-e29b-41d4-a716-446655440000"`
	UserEmail             string         `json:"user_email,omitempty" example:"marcus.johnson@example.com"`
	Persona               string         `json:"persona,omitempty" example:"data-engineer"`
	ToolName              string         `json:"tool_name" example:"datahub_get_schema"`
	ToolkitKind           string         `json:"toolkit_kind,omitempty" example:"datahub"`
	ToolkitName           string         `json:"toolkit_name,omitempty" example:"acme-catalog"`
	Connection            string         `json:"connection,omitempty" example:"acme-catalog"`
	Parameters            map[string]any `json:"parameters,omitempty"`
	Success               bool           `json:"success" example:"true"`
	ErrorMessage          string         `json:"error_message,omitempty"`
	ResponseChars         int            `json:"response_chars" example:"2450"`
	RequestChars          int            `json:"request_chars" example:"120"`
	ContentBlocks         int            `json:"content_blocks" example:"2"`
	Transport             string         `json:"transport" example:"http"`
	Source                string         `json:"source" example:"mcp"`
	EnrichmentApplied     bool           `json:"enrichment_applied" example:"true"`
	EnrichmentTokensFull  int            `json:"enrichment_tokens_full" example:"850"`
	EnrichmentTokensDedup int            `json:"enrichment_tokens_dedup" example:"350"`
	EnrichmentMode        string         `json:"enrichment_mode,omitempty" example:"summary"`
	// EnrichmentMatchKind records how the semantic enrichment matched
	// the target table or column: "urn" when the URN-equality lookup
	// resolved exactly, "semantic" when an exact lookup missed and the
	// platform fell back to similarity search (suggested match, not
	// asserted), or empty when no enrichment ran. Operators use this
	// to measure the false-positive rate of similarity-based
	// suggestions (issue #444).
	EnrichmentMatchKind string `json:"enrichment_match_kind,omitempty" example:"urn"`
	Authorized          bool   `json:"authorized" example:"true"`
	// EventKind is the high-level category of the event. Set to
	// "mcp_tool_call" for tools routed through the MCP toolkits (trino,
	// datahub, s3, mcp gateway) and "apigateway_invoke" for upstream
	// HTTP API calls via the apigateway toolkit. Lets the portal split
	// MCP activity from gateway noise without coupling to tool-name
	// patterns. See EventType constants in event.go.
	EventKind EventType `json:"event_kind,omitempty" example:"mcp_tool_call"`
}

Event represents an auditable event.

func NewEvent 

func NewEvent(toolName string) *Event

NewEvent creates a new audit event.

func (*Event) WithAuthorized added in v0.15.0 

func (e *Event) WithAuthorized(authorized bool) *Event

WithAuthorized records the authorization decision.

func (*Event) WithConnection 

func (e *Event) WithConnection(connection string) *Event

WithConnection adds connection information to the event.

func (*Event) WithEnrichment added in v0.15.0 

func (e *Event) WithEnrichment(applied bool) *Event

WithEnrichment records whether semantic enrichment was applied.

func (*Event) WithEnrichmentMatchKind added in v1.67.0 

func (e *Event) WithEnrichmentMatchKind(kind string) *Event

WithEnrichmentMatchKind records how enrichment matched its target: "urn" for exact URN equality, "semantic" for similarity-fallback, empty when no enrichment ran. See Event.EnrichmentMatchKind.

func (*Event) WithEnrichmentMode added in v0.25.0 

func (e *Event) WithEnrichmentMode(mode string) *Event

WithEnrichmentMode records the enrichment mode used for this event.

func (*Event) WithEnrichmentTokens added in v0.24.0 

func (e *Event) WithEnrichmentTokens(full, dedup int) *Event

WithEnrichmentTokens records estimated token counts for enrichment.

func (*Event) WithEventKind added in v1.69.0 

func (e *Event) WithEventKind(kind EventType) *Event

WithEventKind records the high-level category of the event (e.g. mcp_tool_call, apigateway_invoke). See Event.EventKind.

func (*Event) WithParameters 

func (e *Event) WithParameters(params map[string]any) *Event

WithParameters adds parameters to the event.

func (*Event) WithPersona 

func (e *Event) WithPersona(persona string) *Event

WithPersona adds persona information to the event.

func (*Event) WithRequestID 

func (e *Event) WithRequestID(requestID string) *Event

WithRequestID adds a request ID to the event.

func (*Event) WithRequestSize added in v0.15.0 

func (e *Event) WithRequestSize(chars int) *Event

WithRequestSize adds request size metrics to the event.

func (*Event) WithResponseSize added in v0.14.0 

func (e *Event) WithResponseSize(chars, contentBlocks int) *Event

WithResponseSize adds response size metrics to the event.

func (*Event) WithResult 

func (e *Event) WithResult(success bool, errorMsg string, durationMS int64) *Event

WithResult adds result information to the event.

func (*Event) WithSessionID added in v0.15.0 

func (e *Event) WithSessionID(sessionID string) *Event

WithSessionID adds session identification to the event.

func (*Event) WithToolkit 

func (e *Event) WithToolkit(kind, name string) *Event

WithToolkit adds toolkit information to the event.

func (*Event) WithTransport added in v0.15.0 

func (e *Event) WithTransport(transport, source string) *Event

WithTransport adds transport and source metadata to the event.

func (*Event) WithUser 

func (e *Event) WithUser(userID, email string) *Event

WithUser adds user information to the event.

type EventType 

type EventType string

EventType categorizes audit events. 

const (
	// EventTypeToolCall is a tool invocation event.
	EventTypeToolCall EventType = "tool_call"

	// EventTypeAuth is an authentication event.
	EventTypeAuth EventType = "auth"

	// EventTypeAdmin is an administrative event.
	EventTypeAdmin EventType = "admin"

	// EventTypeMCPToolCall categorizes an MCP tool invocation routed
	// through a non-apigateway toolkit (trino, datahub, s3, mcp gateway).
	// Stamped into Event.EventKind at write time so the portal Activity
	// view can exclude apigateway HTTP noise by default.
	EventTypeMCPToolCall EventType = "mcp_tool_call"

	// EventTypeAPIGatewayInvoke categorizes an HTTP API invocation
	// through the apigateway toolkit. Stamped into Event.EventKind at
	// write time so the portal Activity view can split these out from
	// the MCP-only view.
	EventTypeAPIGatewayInvoke EventType = "apigateway_invoke"
)

func EventKindForToolkit added in v1.69.0 

func EventKindForToolkit(toolkitKind string) EventType

EventKindForToolkit maps a toolkit kind to its high-level event category. The apigateway toolkit ("api") produces HTTP invocations; every other toolkit kind produces MCP tool calls. Used at audit write time so the portal can split gateway noise from MCP activity.

type Logger 

type Logger interface {
	// Log records an audit event.
	Log(ctx context.Context, event Event) error

	// Query retrieves audit events matching the filter.
	Query(ctx context.Context, filter QueryFilter) ([]Event, error)

	// Close releases resources.
	Close() error
}

Logger defines the interface for audit logging.

type MetricsFilter added in v0.36.0 

type MetricsFilter struct {
	StartTime *time.Time
	EndTime   *time.Time
	UserID    string
	EventKind string
}

MetricsFilter provides common filtering for aggregate metric queries.

type Overview added in v0.17.1 

type Overview struct {
	TotalCalls     int     `json:"total_calls" example:"196"`
	SuccessRate    float64 `json:"success_rate" example:"0.949"`
	AvgDurationMS  float64 `json:"avg_duration_ms" example:"522"`
	UniqueUsers    int     `json:"unique_users" example:"12"`
	UniqueTools    int     `json:"unique_tools" example:"12"`
	EnrichmentRate float64 `json:"enrichment_rate" example:"0.85"`
	ErrorCount     int     `json:"error_count" example:"10"`
}

Overview holds aggregate statistics for the audit log.

type PerformanceStats added in v0.17.1 

type PerformanceStats struct {
	P50MS            float64 `json:"p50_ms" example:"320"`
	P95MS            float64 `json:"p95_ms" example:"1450"`
	P99MS            float64 `json:"p99_ms" example:"2400"`
	AvgMS            float64 `json:"avg_ms" example:"522"`
	MaxMS            float64 `json:"max_ms" example:"5200"`
	AvgResponseChars float64 `json:"avg_response_chars" example:"1850"`
	AvgRequestChars  float64 `json:"avg_request_chars" example:"120"`
}

PerformanceStats holds latency percentile statistics.

type QueryFilter 

type QueryFilter struct {
	ID          string
	StartTime   *time.Time
	EndTime     *time.Time
	UserID      string
	SessionID   string
	ToolName    string
	ToolkitKind string
	Source      string
	EventKind   string
	Search      string
	Success     *bool
	SortBy      string
	SortOrder   SortOrder
	Limit       int
	Offset      int
}

QueryFilter defines criteria for querying audit events.

type Resolution added in v0.17.1 

type Resolution string

Resolution defines the time bucketing granularity for timeseries queries. 

const (
	// ResolutionMinute buckets by minute.
	ResolutionMinute Resolution = "minute"

	// ResolutionHour buckets by hour.
	ResolutionHour Resolution = "hour"

	// ResolutionDay buckets by day.
	ResolutionDay Resolution = "day"
)

type SortOrder added in v0.17.1 

type SortOrder string

SortOrder defines sort direction. 

const (
	// SortAsc sorts ascending.
	SortAsc SortOrder = "asc"

	// SortDesc sorts descending.
	SortDesc SortOrder = "desc"
)

type SyncOption added in v1.101.0 

type SyncOption func(*SyncWriter)

SyncOption configures a SyncWriter.

func WithSyncMetrics added in v1.101.0 

func WithSyncMetrics(m *observability.Metrics) SyncOption

WithSyncMetrics attaches the observability recorder used to count lost events. A nil recorder is accepted; observability.Metrics is itself nil-safe.

func WithSyncWriteTimeout added in v1.101.0 

func WithSyncWriteTimeout(d time.Duration) SyncOption

WithSyncWriteTimeout sets the per-write timeout. Non-positive values are ignored so the default timeout stands.

type SyncWriter added in v1.101.0 

type SyncWriter struct {
	// contains filtered or unexported fields
}

SyncWriter writes each event to the underlying Logger synchronously, on the caller's goroutine, bounded by a per-write timeout. It is the durable counterpart to AsyncWriter: it has no queue, so it sheds nothing on backpressure — a slow store slows the caller (backpressure) rather than dropping events. Compliance deployments choose sync delivery to trade tool-call latency for zero queue-overflow drops (issue #898).

A write that fails or exceeds the per-write timeout is a genuinely lost event: it is logged and counted (via the same audit_events_dropped_total metric AsyncWriter uses, so the loss is visible regardless of delivery mode). Log still returns nil in that case — audit must never fail a tool call.

Like AsyncWriter, each write runs against baseCtx (not the caller's request context, which the audit middleware deliberately passes as context.Background so a client disconnect cannot abandon an audit write mid-request). Close cancels baseCtx, so an in-flight write against a stalled store is abandoned at shutdown instead of holding a request goroutine — and a connection from the shared pool — for the full per-write timeout while the platform tears down (the guarantee #884 gave the async path, kept here for the sync path).

func NewSyncWriter added in v1.101.0 

func NewSyncWriter(logger Logger, opts ...SyncOption) *SyncWriter

NewSyncWriter wraps logger in a synchronous, timeout-bounded writer. Unlike NewAsyncWriter it starts no goroutine: each Log call does the store write inline. Call Close on shutdown to cancel any in-flight write.

func (*SyncWriter) Close added in v1.101.0 

func (w *SyncWriter) Close(_ context.Context) error

Close cancels baseCtx so any in-flight write against a stalled store is abandoned immediately rather than blocking shutdown for the per-write timeout. It has no queue to drain, so ctx is unused and Close returns nil at once; implementing this Close(ctx) signature lets the audit store adapter treat the sync and async writers uniformly (pkg/middleware auditFlusher). Idempotent.

func (*SyncWriter) Log added in v1.101.0 

func (w *SyncWriter) Log(_ context.Context, e Event) error

Log writes e to the underlying logger on the caller's goroutine, bounding the attempt with the per-write timeout on baseCtx. The passed context is ignored (matching AsyncWriter): the audit middleware hands this writer context.Background so a request ending does not abandon the write; shutdown cancellation flows through Close, not the request context. A failed or abandoned write is logged and counted, but Log always returns nil so a store outage never fails the tool call.

func (*SyncWriter) Lost added in v1.101.0 

func (w *SyncWriter) Lost() int64

Lost returns the cumulative number of events lost to a failed or abandoned synchronous write.

type TimeseriesBucket added in v0.17.1 

type TimeseriesBucket struct {
	Bucket        time.Time `json:"bucket" example:"2026-04-15T14:30:00Z"`
	Count         int       `json:"count" example:"12"`
	SuccessCount  int       `json:"success_count" example:"11"`
	ErrorCount    int       `json:"error_count" example:"1"`
	AvgDurationMS float64   `json:"avg_duration_ms" example:"245.5"`
}

TimeseriesBucket holds counts for a single time bucket.

func ZeroFill added in v0.36.1 

func ZeroFill(buckets []TimeseriesBucket, start, end time.Time, resolution Resolution) []TimeseriesBucket

ZeroFill expands a sparse set of timeseries buckets into a complete series covering [start, end] at the given resolution. Missing buckets are filled with zero values.

type TimeseriesFilter added in v0.17.1 

type TimeseriesFilter struct {
	Resolution Resolution
	StartTime  *time.Time
	EndTime    *time.Time
	UserID     string
	EventKind  string
}

TimeseriesFilter controls timeseries query parameters.

Source Files

View all Source files

Directories

Path Synopsis
Package postgres provides PostgreSQL storage for audit logs.
 Package postgres provides PostgreSQL storage for audit logs.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.