auditlog

package
v0.1.61 Latest Latest
Warning

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

 Go to latest Published: Jul 26, 2026 License: MIT Imports: 39 Imported by: 0

Documentation

Overview

Package auditlog provides audit logging for the AI gateway. It captures request/response metadata and stores it in configurable backends.

Index

Constants

View Source 
const (
	CacheTypeExact    = "exact"
	CacheTypeSemantic = "semantic"

	AuthMethodAPIKey    = "api_key"
	AuthMethodMasterKey = "master_key"
	AuthMethodNoKey     = "no_key"
)
View Source 
const (
	AttemptKindPrimary  = "primary"
	AttemptKindFailover = "failover"
	AttemptKindRetry    = "retry"
)
View Source 
const (
	LiveEventAuditStarted   = "audit.started"
	LiveEventAuditUpdated   = "audit.updated"
	LiveEventAuditStream    = "audit.stream"
	LiveEventAuditCompleted = "audit.completed"
	LiveEventAuditFailed    = "audit.failed"
	LiveEventAuditFlushed   = "audit.flushed"
	LiveEventAuditRemoved   = "audit.removed"
)
View Source 
const (
	// MaxBodyCapture is the maximum size of request/response bodies to capture (1MB).
	// Prevents memory exhaustion from large payloads.
	MaxBodyCapture = 1024 * 1024

	// MaxContentCapture is the maximum size of accumulated streaming content (1MB).
	// Used by the stream observer to limit reconstructed response body size.
	MaxContentCapture = 1024 * 1024

	// BatchFlushThreshold is the number of entries that triggers an immediate flush.
	// When the batch reaches this size, it's written to storage without waiting for the timer.
	BatchFlushThreshold = 100

	// APIKeyHashPrefixLength is the number of hex characters from SHA256 hash.
	// 16 hex chars = 64 bits of entropy for identification without exposure.
	APIKeyHashPrefixLength = 16
)

Buffer and capture limits for audit logging.

View Source 
const (
	// LogEntryKey is the context key for storing the log entry.
	LogEntryKey contextKey = "auditlog_entry"

	// LogEntryStreamingKey is the context key for marking a request as streaming.
	// When true, the middleware skips logging because the stream observer path
	// handles streaming audit logging.
	LogEntryStreamingKey contextKey = "auditlog_entry_streaming"

	// LogEntryLivePublisherKey stores an optional realtime dashboard publisher.
	LogEntryLivePublisherKey contextKey = "auditlog_live_publisher"
)
View Source 
const (
	StatsIntervalHour = "hour"
	StatsIntervalDay  = "day"
)

Request stats bucket granularities. The admin handler picks hourly buckets for short ranges and daily buckets for longer ones.

View Source 
const CleanupInterval = 1 * time.Hour

CleanupInterval is how often the cleanup goroutine runs to delete old log entries.

Variables

View Source 
var ErrPartialWrite = errors.New("partial write failure")

ErrPartialWrite indicates that a batch write only partially succeeded. Use errors.As to extract details about the failure.

Functions

func CaptureAttemptResponseBody 

func CaptureAttemptResponseBody(body []byte) any

CaptureAttemptResponseBody parses a raw upstream error body into a value suitable for audit storage: a JSON value when the body is valid JSON, a UTF-8 string otherwise, or nil when empty.

func CaptureInternalJSONExchange 

func CaptureInternalJSONExchange(
	entry *LogEntry,
	ctx context.Context,
	method,
	path string,
	requestBody,
	responseBody any,
	responseErr error,
	cfg Config,
)

CaptureInternalJSONExchange applies normal audit capture policy to an internal JSON request/response pair without requiring the caller to synthesize HTTP transport details in the server layer.

func CaptureLoggedBody 

func CaptureLoggedBody(bodyBytes []byte) any

CaptureLoggedBody converts raw body bytes into the representation audit entries store: parsed JSON when possible, otherwise a valid-UTF-8 string.

func EnrichEntry 

func EnrichEntry(c *echo.Context, model, provider string)

EnrichEntry retrieves the log entry from context for enrichment by handlers. This allows handlers to add model and provider information.

func EnrichEntryWithAttempts 

func EnrichEntryWithAttempts(c *echo.Context, attempts []AttemptSnapshot)

EnrichEntryWithAttempts attaches provider attempt summaries to the live audit entry. The attempt list belongs to the logical request, not to separate top-level audit rows.

func EnrichEntryWithAuthKeyID 

func EnrichEntryWithAuthKeyID(c *echo.Context, authKeyID string)

EnrichEntryWithAuthKeyID attaches the authenticated managed auth key id to the live audit entry.

func EnrichEntryWithAuthMethod 

func EnrichEntryWithAuthMethod(c *echo.Context, method string)

EnrichEntryWithAuthMethod records which authentication mechanism was used for the request.

func EnrichEntryWithCacheType 

func EnrichEntryWithCacheType(c *echo.Context, cacheType string)

EnrichEntryWithCacheType attaches cache-hit metadata to the live audit entry. The value is intentionally sourced directly from the cache middleware, not inferred from response headers after the fact.

func EnrichEntryWithCachedStreamResponse 

func EnrichEntryWithCachedStreamResponse(c *echo.Context, path string, body []byte)

EnrichEntryWithCachedStreamResponse reconstructs the OpenAI-compatible response body for a cached SSE replay when audit body capture is enabled.

func EnrichEntryWithCapturedResponseBody 

func EnrichEntryWithCapturedResponseBody(c *echo.Context, body any, truncated bool)

EnrichEntryWithCapturedResponseBody sets the response body and truncation flag from a handler-owned response capture — for responses the middleware's own writer deliberately skips, like MCP's SSE-framed POST replies.

func EnrichEntryWithError 

func EnrichEntryWithError(c *echo.Context, errorType, errorMessage string, errorCode ...string)

EnrichEntryWithError adds error information to the log entry.

func EnrichEntryWithFailover 

func EnrichEntryWithFailover(c *echo.Context, targetModel string)

EnrichEntryWithFailover records the configured failover selector used for the live request when translated execution redirected away from the primary selector.

func EnrichEntryWithRawRequestBody 

func EnrichEntryWithRawRequestBody(c *echo.Context, body []byte)

EnrichEntryWithRawRequestBody captures a raw request payload from a handler whose endpoint is not ingress-managed (no request snapshot to read from), applying the same size cap and JSON/UTF-8 coercion as snapshot-backed capture. Callers gate on the audit LogBodies setting; an already-populated request body is preserved.

func EnrichEntryWithRequestBody 

func EnrichEntryWithRequestBody(c *echo.Context, body any)

EnrichEntryWithRequestBody sets the audit request body from a handler that captures its own request payload (e.g. audio endpoints, which are not ingress-managed and so have no request snapshot to read from). A nil body or missing entry is a no-op; an already-populated request body is preserved.

func EnrichEntryWithRequestRevision 

func EnrichEntryWithRequestRevision(c *echo.Context, revision RequestRevisionSnapshot)

EnrichEntryWithRequestRevision appends one ingress request-rewrite revision to the live audit entry, assigning the next sequence number. A missing entry is a no-op.

func EnrichEntryWithRequestedModel 

func EnrichEntryWithRequestedModel(c *echo.Context, requestedModel string)

EnrichEntryWithRequestedModel attaches early requested-model metadata to the live audit entry before the final workflow policy has been resolved.

func EnrichEntryWithResolvedRoute 

func EnrichEntryWithResolvedRoute(c *echo.Context, resolvedModel, providerType, providerName string)

EnrichEntryWithResolvedRoute attaches the final executed route to the live audit entry after execution resolved to a concrete provider/model.

func EnrichEntryWithResponseBody 

func EnrichEntryWithResponseBody(c *echo.Context, body any)

EnrichEntryWithResponseBody sets the audit response body from a handler that captures its own response payload (e.g. audio output served as raw bytes). A nil body or missing entry is a no-op.

func EnrichEntryWithStream 

func EnrichEntryWithStream(c *echo.Context, stream bool)

EnrichEntryWithStream marks the log entry as a streaming request.

func EnrichEntryWithUserPath 

func EnrichEntryWithUserPath(c *echo.Context, userPath string)

EnrichEntryWithUserPath attaches the effective user path to the live audit entry.

func EnrichEntryWithWorkflow 

func EnrichEntryWithWorkflow(c *echo.Context, workflow *core.Workflow)

EnrichEntryWithWorkflow attaches workflow metadata to the live audit entry. This is preferred over requested-model-only enrichment once workflow resolution has completed for the request.

func EnrichLogEntryWithAttempts 

func EnrichLogEntryWithAttempts(entry *LogEntry, attempts []AttemptSnapshot)

EnrichLogEntryWithAttempts attaches provider attempt summaries directly to an existing audit log entry.

func EnrichLogEntryWithFailover 

func EnrichLogEntryWithFailover(entry *LogEntry, targetModel string)

EnrichLogEntryWithFailover attaches failover redirect metadata directly to an existing audit log entry.

func EnrichLogEntryWithRequestContext 

func EnrichLogEntryWithRequestContext(entry *LogEntry, ctx context.Context)

EnrichLogEntryWithRequestContext attaches auth and effective user-path metadata from context directly to an existing log entry.

func EnrichLogEntryWithResolvedRoute 

func EnrichLogEntryWithResolvedRoute(entry *LogEntry, resolvedModel, providerType, providerName string)

EnrichLogEntryWithResolvedRoute attaches the final executed route directly to an existing audit log entry.

func EnrichLogEntryWithWorkflow 

func EnrichLogEntryWithWorkflow(entry *LogEntry, workflow *core.Workflow)

EnrichLogEntryWithWorkflow attaches workflow metadata directly to an existing log entry. Internal translated executors can use this without depending on Echo middleware state.

func IsAudioContentType 

func IsAudioContentType(contentType string) bool

IsAudioContentType reports whether a Content-Type denotes an audio payload.

func IsEntryMarkedAsStreaming 

func IsEntryMarkedAsStreaming(c interface{ Get(string) any }) bool

IsEntryMarkedAsStreaming checks if the entry is marked as streaming.

func MarkEntryAsStreaming 

func MarkEntryAsStreaming(c interface{ Set(string, any) }, isStreaming bool)

MarkEntryAsStreaming marks the entry as a streaming request so the middleware knows not to log it (the stream observer path will handle logging).

func Middleware 

func Middleware(logger LoggerInterface) echo.MiddlewareFunc

Middleware creates an Echo middleware for audit logging. It captures request metadata at the start and response metadata at the end, then writes the log entry asynchronously.

func PopulateRequestData 

func PopulateRequestData(entry *LogEntry, req *http.Request, cfg Config)

PopulateRequestData copies the configured request capture fields into the log entry. Streaming handlers call this before creating the detached stream entry so request metadata is preserved even though the middleware finishes later.

func PopulateRequestHeaders 

func PopulateRequestHeaders(entry *LogEntry, headers http.Header)

PopulateRequestHeaders copies redacted request headers into the log entry.

func PopulateResponseData 

func PopulateResponseData(entry *LogEntry, headers http.Header, body []byte, bodyTruncated bool, cfg Config)

PopulateResponseData copies the configured response capture fields into the log entry from already-buffered response bytes.

func PopulateResponseHeaders 

func PopulateResponseHeaders(entry *LogEntry, headers http.Header)

PopulateResponseHeaders copies response headers into the log entry when header logging is enabled.

func RedactAttemptResponseHeaders 

func RedactAttemptResponseHeaders(header http.Header) map[string]string

RedactAttemptResponseHeaders flattens and redacts the upstream response headers of a failed attempt for audit storage.

func RedactHeaders 

func RedactHeaders(headers map[string]string) map[string]string

RedactHeaders redacts credential headers (core.IsCredentialHeader) from a header map. Values are replaced with "[REDACTED]" to prevent leaking secrets. The original map is not modified; a new map is returned.

Types

type AttemptSnapshot 

type AttemptSnapshot struct {
	Seq          int       `json:"seq" bson:"seq"`
	Kind         string    `json:"kind" bson:"kind"`
	ProviderType string    `json:"provider_type,omitempty" bson:"provider_type,omitempty"`
	ProviderName string    `json:"provider_name,omitempty" bson:"provider_name,omitempty"`
	Model        string    `json:"model,omitempty" bson:"model,omitempty"`
	StatusCode   int       `json:"status_code,omitempty" bson:"status_code,omitempty"`
	Success      bool      `json:"success" bson:"success"`
	ErrorType    string    `json:"error_type,omitempty" bson:"error_type,omitempty"`
	ErrorCode    string    `json:"error_code,omitempty" bson:"error_code,omitempty"`
	ErrorMessage string    `json:"error_message,omitempty" bson:"error_message,omitempty"`
	StartedAt    time.Time `json:"started_at" bson:"started_at,omitempty"`
	DurationNs   int64     `json:"duration_ns,omitempty" bson:"duration_ns,omitempty"`

	// ResponseBody and ResponseHeaders capture the raw upstream error response
	// of a failed attempt. ResponseBody is the parsed JSON (or a string when the
	// body is not JSON); ResponseHeaders is redacted. Both are populated only
	// when audit body/header logging is enabled.
	ResponseBody    any               `json:"response_body,omitempty" bson:"response_body,omitempty"`
	ResponseHeaders map[string]string `json:"response_headers,omitempty" bson:"response_headers,omitempty"`
}

AttemptSnapshot stores one external provider attempt made for a logical request. It intentionally stores structured errors, not raw upstream bodies.

func GateAttemptCapture 

func GateAttemptCapture(attempts []AttemptSnapshot, cfg Config) []AttemptSnapshot

GateAttemptCapture clears per-attempt response bodies and/or headers that the audit configuration did not opt into, leaving the structured error fields (type / code / status / message) untouched. It mutates and returns attempts.

type AudioBodyLog 

type AudioBodyLog struct {
	Audio       bool           `json:"__audio__" bson:"__audio__"`
	ContentType string         `json:"content_type,omitempty" bson:"content_type,omitempty"`
	Bytes       int            `json:"bytes" bson:"bytes"`
	Encoding    string         `json:"encoding,omitempty" bson:"encoding,omitempty"`
	Data        string         `json:"data,omitempty" bson:"data,omitempty"`
	Stored      bool           `json:"stored" bson:"stored"`
	TooLarge    bool           `json:"too_large,omitempty" bson:"too_large,omitempty"`
	Meta        map[string]any `json:"meta,omitempty" bson:"meta,omitempty"`
}

AudioBodyLog is the audit representation of an audio request/response body. The "__audio__" marker lets the dashboard detect audio payloads and render a player (when Data is present) or a labeled placeholder. When Data is set it holds the base64-encoded audio, suitable for a data: URL of ContentType.

func BuildAudioResponseBody 

func BuildAudioResponseBody(contentType string, data []byte, storeBytes bool) AudioBodyLog

BuildAudioResponseBody builds the audit value for a binary audio response. When storeBytes is true and the payload fits within audioBodyMaxBytes the audio is embedded as base64 for playback; otherwise only metadata is kept.

func BuildAudioUploadBody 

func BuildAudioUploadBody(contentType string, data []byte, storeBytes bool, meta map[string]any) AudioBodyLog

BuildAudioUploadBody builds the audit value for an uploaded audio request (e.g. a transcription input). It behaves like BuildAudioResponseBody but attaches request metadata (model, params) alongside the audio so the dashboard can show both a player and the parameters.

type Config 

type Config struct {
	// Enabled controls whether audit logging is active
	Enabled bool

	// LogBodies enables logging of full request/response bodies
	LogBodies bool

	// LogAudioBodies refines LogBodies for audio endpoints (base64 audio for
	// /v1/audio/speech, upload metadata for transcriptions). Requires LogBodies:
	// when LogBodies is off no audio body is captured; when LogBodies is on but
	// this is off, audio responses are recorded as a lightweight placeholder.
	LogAudioBodies bool

	// LogHeaders enables logging of request/response headers
	LogHeaders bool

	// BufferSize is the number of log entries to buffer before flushing
	BufferSize int

	// FlushInterval is how often to flush buffered logs
	FlushInterval time.Duration

	// RetentionDays is how long to keep logs (0 = forever)
	RetentionDays int

	// OnlyModelInteractions limits logging to AI model endpoints only
	// When true, only /v1/chat/completions, /v1/responses, /v1/embeddings, /v1/files, and /v1/batches are logged
	OnlyModelInteractions bool
}

Config holds audit logging configuration

type ConversationResult 

type ConversationResult struct {
	AnchorID string     `json:"anchor_id"`
	Entries  []LogEntry `json:"entries"`
}

ConversationResult holds a linear conversation thread centered around an anchor log.

type FailoverSnapshot 

type FailoverSnapshot struct {
	TargetModel string `json:"target_model,omitempty" bson:"target_model,omitempty"`
}

FailoverSnapshot stores the runtime failover selection used for one request. The target model is the configured failover selector, not the model echoed by the provider response body.

type LiveEventEmitter 

type LiveEventEmitter interface {
	PublishLiveEvent(eventType string, entry *LogEntry)
}

LiveEventEmitter is implemented by loggers that can publish audit lifecycle previews before the entry is persisted.

type LiveEventPublisher 

type LiveEventPublisher interface {
	PublishAuditEvent(eventType string, entry *LogEntry)
}

LiveEventPublisher receives compact audit lifecycle snapshots for realtime dashboard preview. Implementations must not block request handling.

type LiveSubscriberReporter 

type LiveSubscriberReporter interface {
	HasLiveSubscribers() bool
}

LiveSubscriberReporter is optionally implemented by live publishers (and emitters) that can report whether any dashboard subscriber is currently connected. Publishers that cannot tell are treated as always subscribed.

type LogData 

type LogData struct {
	// Identity
	UserAgent  string `json:"user_agent,omitempty" bson:"user_agent,omitempty"`
	APIKeyHash string `json:"api_key_hash,omitempty" bson:"api_key_hash,omitempty"`

	// Labels are request labels extracted from configured tagging headers.
	Labels []string `json:"labels,omitempty" bson:"labels,omitempty"`

	// WorkflowFeatures captures the request-time effective workflow features
	// after runtime caps were applied. This keeps audit views historically accurate
	// even if the active process config changes later.
	WorkflowFeatures *WorkflowFeaturesSnapshot `json:"workflow_features,omitempty" bson:"workflow_features,omitempty"`

	// Failover captures runtime redirect details when translated execution
	// moved from the primary selector to a configured failover target.
	Failover *FailoverSnapshot `json:"failover,omitempty" bson:"failover,omitempty"`

	// Attempts captures provider calls made for this logical request. SQL
	// stores split this into audit_log_attempts; Mongo stores it embedded.
	Attempts []AttemptSnapshot `json:"attempts,omitempty" bson:"attempts,omitempty"`

	// RequestRevisions captures the ingress request-rewrite chain: one entry
	// per registered rewriter that ran, in application order. Rewriters that
	// changed the body carry the rewritten body; those that left it alone are
	// recorded with NoChange so the audit trail still shows the step ran.
	// RequestBody always remains the original client request; the last
	// changed revision is what was forwarded downstream — when every rewriter
	// was a no-op there is no such revision and the original body is what
	// went upstream.
	RequestRevisions []RequestRevisionSnapshot `json:"request_revisions,omitempty" bson:"request_revisions,omitempty"`

	// Request parameters
	Temperature *float64 `json:"temperature,omitempty" bson:"temperature,omitempty"`
	MaxTokens   *int     `json:"max_tokens,omitempty" bson:"max_tokens,omitempty"`

	// Error details (message can be long, so kept in JSON)
	ErrorMessage string `json:"error_message,omitempty" bson:"error_message,omitempty"`
	ErrorCode    string `json:"error_code,omitempty" bson:"error_code,omitempty"`

	// Optional headers (when LOGGING_LOG_HEADERS=true)
	// Sensitive headers are auto-redacted
	RequestHeaders  map[string]string `json:"request_headers,omitempty" bson:"request_headers,omitempty"`
	ResponseHeaders map[string]string `json:"response_headers,omitempty" bson:"response_headers,omitempty"`

	// Optional bodies (when LOGGING_LOG_BODIES=true)
	// Stored as interface{} so MongoDB serializes as native BSON documents (queryable/readable)
	// instead of BSON Binary (base64 in Compass)
	RequestBody  any `json:"request_body,omitempty" bson:"request_body,omitempty"`
	ResponseBody any `json:"response_body,omitempty" bson:"response_body,omitempty"`

	// Body capture status flags (set when body exceeds 1MB limit)
	RequestBodyTooBigToHandle  bool `json:"request_body_too_big_to_handle,omitempty" bson:"request_body_too_big_to_handle,omitempty"`
	ResponseBodyTooBigToHandle bool `json:"response_body_too_big_to_handle,omitempty" bson:"response_body_too_big_to_handle,omitempty"`
}

LogData contains flexible request/response information. Fields that are commonly filtered are stored as columns in LogEntry. This struct contains the remaining flexible data.

type LogEntry 

type LogEntry struct {
	// ID is a unique identifier for this log entry (UUID)
	ID string `json:"id" bson:"_id"`

	// Timestamp is when the request started
	Timestamp time.Time `json:"timestamp" bson:"timestamp"`

	// DurationNs is the request duration in nanoseconds
	DurationNs int64 `json:"duration_ns" bson:"duration_ns"`

	// Core fields (indexed for queries)
	RequestedModel    string `json:"requested_model" bson:"requested_model,omitempty"`
	ResolvedModel     string `json:"resolved_model,omitempty" bson:"resolved_model,omitempty"`
	Provider          string `json:"provider" bson:"provider"` // canonical provider type used for routing and filters
	ProviderName      string `json:"provider_name,omitempty" bson:"provider_name,omitempty"`
	AliasUsed         bool   `json:"alias_used,omitempty" bson:"alias_used,omitempty"`
	WorkflowVersionID string `json:"workflow_version_id,omitempty" bson:"workflow_version_id,omitempty"`
	CacheType         string `json:"cache_type,omitempty" bson:"cache_type,omitempty"`
	StatusCode        int    `json:"status_code" bson:"status_code"`

	// Extracted fields for efficient filtering (indexed in relational DBs)
	RequestID  string `json:"request_id,omitempty" bson:"request_id,omitempty"`
	AuthKeyID  string `json:"auth_key_id,omitempty" bson:"auth_key_id,omitempty"`
	AuthMethod string `json:"auth_method,omitempty" bson:"auth_method,omitempty"`
	ClientIP   string `json:"client_ip,omitempty" bson:"client_ip,omitempty"`
	Method     string `json:"method,omitempty" bson:"method,omitempty"`
	Path       string `json:"path,omitempty" bson:"path,omitempty"`
	UserPath   string `json:"user_path,omitempty" bson:"user_path,omitempty"`
	Stream     bool   `json:"stream,omitempty" bson:"stream,omitempty"`
	ErrorType  string `json:"error_type,omitempty" bson:"error_type,omitempty"`

	// Data contains flexible request/response information as JSON
	Data *LogData `json:"data,omitempty" bson:"data,omitempty"`
}

LogEntry represents a single audit log entry. Core fields are indexed for efficient queries.

func CreateStreamEntry 

func CreateStreamEntry(baseEntry *LogEntry) *LogEntry

CreateStreamEntry creates a new log entry for a streaming request. This should be called before starting the stream.

func GetStreamEntryFromContext 

func GetStreamEntryFromContext(c interface{ Get(string) any }) *LogEntry

GetStreamEntryFromContext retrieves the log entry from Echo context for streaming. This allows handlers to get the entry for wrapping streams.

type LogListResult 

type LogListResult struct {
	Entries []LogEntry `json:"entries"`
	Total   int        `json:"total"`
	Limit   int        `json:"limit"`
	Offset  int        `json:"offset"`
}

LogListResult holds a paginated list of audit log entries.

type LogQueryParams 

type LogQueryParams struct {
	QueryParams
	RequestedModel string
	Provider       string // filter by provider name or provider type
	Method         string
	Path           string
	UserPath       string
	ErrorType      string
	Search         string
	StatusCode     *int
	Stream         *bool
	Limit          int
	Offset         int
}

LogQueryParams specifies query parameters for paginated audit log retrieval.

type LogStore 

type LogStore interface {
	// WriteBatch writes multiple log entries to storage.
	// This is called by the Logger when flushing buffered entries.
	WriteBatch(ctx context.Context, entries []*LogEntry) error

	// Flush forces any pending writes to complete.
	// Called during graceful shutdown.
	Flush(ctx context.Context) error

	// Close releases resources and flushes pending writes.
	Close() error
}

LogStore defines the interface for audit log storage backends. Implementations must be safe for concurrent use.

type Logger 

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

Logger provides async buffered logging with batch writes. It collects log entries in a channel and flushes them to storage either when the buffer is full or at regular intervals.

func NewLogger 

func NewLogger(store LogStore, cfg Config) *Logger

NewLogger creates a new async buffered Logger. The logger starts a background goroutine for flushing entries.

func (*Logger) Close 

func (l *Logger) Close() error

Close stops the logger and flushes remaining entries. This should be called during graceful shutdown. Close is idempotent - calling it multiple times is safe.

func (*Logger) Config 

func (l *Logger) Config() Config

Config returns the logger configuration

func (*Logger) HasLiveSubscribers 

func (l *Logger) HasLiveSubscribers() bool

HasLiveSubscribers reports whether the attached live publisher currently has connected dashboard subscribers. Used to skip building live body previews that nobody would receive; a publisher that cannot tell counts as subscribed.

func (*Logger) PublishLiveEvent 

func (l *Logger) PublishLiveEvent(eventType string, entry *LogEntry)

PublishLiveEvent publishes a compact lifecycle preview when live logs are enabled.

func (*Logger) SetLivePublisher 

func (l *Logger) SetLivePublisher(p LiveEventPublisher)

SetLivePublisher attaches the optional realtime dashboard publisher.

func (*Logger) Write 

func (l *Logger) Write(entry *LogEntry)

Write queues a log entry for async writing. This method is non-blocking. If the buffer is full or the logger is closed, the entry is dropped and a warning is logged.

type LoggerInterface 

type LoggerInterface interface {
	Write(entry *LogEntry)
	Config() Config
	Close() error
}

LoggerInterface defines the interface for loggers (both real and noop)

type MongoDBReader 

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

MongoDBReader implements Reader for MongoDB.

func NewMongoDBReader 

func NewMongoDBReader(database *mongo.Database) (*MongoDBReader, error)

NewMongoDBReader creates a new MongoDB audit log reader.

func (*MongoDBReader) GetConversation 

func (r *MongoDBReader) GetConversation(ctx context.Context, logID string, limit int) (*ConversationResult, error)

GetConversation returns a linear conversation thread around a seed log entry.

func (*MongoDBReader) GetLogByID 

func (r *MongoDBReader) GetLogByID(ctx context.Context, id string) (*LogEntry, error)

GetLogByID returns a single audit log entry by ID.

func (*MongoDBReader) GetLogs 

func (r *MongoDBReader) GetLogs(ctx context.Context, params LogQueryParams) (*LogListResult, error)

GetLogs returns a paginated list of audit log entries.

func (*MongoDBReader) GetRequestStats 

func (r *MongoDBReader) GetRequestStats(ctx context.Context, params RequestStatsParams) (*RequestStats, error)

GetRequestStats returns time-bucketed status-class counts and per-provider latency aggregates for the dashboard charts.

type MongoDBStore 

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

MongoDBStore implements LogStore for MongoDB.

func NewMongoDBStore 

func NewMongoDBStore(database *mongo.Database, retentionDays int) (*MongoDBStore, error)

NewMongoDBStore creates a new MongoDB audit log store. It creates the collection and indexes if they don't exist. MongoDB handles TTL-based cleanup automatically via TTL indexes.

func (*MongoDBStore) Close 

func (s *MongoDBStore) Close() error

Close is a no-op for MongoDB as the client is managed by the storage layer.

func (*MongoDBStore) Flush 

func (s *MongoDBStore) Flush(_ context.Context) error

Flush is a no-op for MongoDB as writes are synchronous.

func (*MongoDBStore) WriteBatch 

func (s *MongoDBStore) WriteBatch(ctx context.Context, entries []*LogEntry) error

WriteBatch writes multiple log entries to MongoDB using InsertMany.

type NoopLogger 

type NoopLogger struct{}

NoopLogger is a logger that does nothing (used when logging is disabled)

func (*NoopLogger) Close 

func (l *NoopLogger) Close() error

Close does nothing

func (*NoopLogger) Config 

func (l *NoopLogger) Config() Config

Config returns an empty config

func (*NoopLogger) Write 

func (l *NoopLogger) Write(_ *LogEntry)

Write does nothing

type PartialWriteError 

type PartialWriteError struct {
	TotalEntries int
	FailedCount  int
	Cause        mongo.BulkWriteException
}

PartialWriteError wraps a mongo.BulkWriteException with additional context about how many entries failed vs succeeded.

func (*PartialWriteError) Error 

func (e *PartialWriteError) Error() string

func (*PartialWriteError) Unwrap 

func (e *PartialWriteError) Unwrap() error

type ProviderLatencySeries 

type ProviderLatencySeries struct {
	Provider      string     `json:"provider"`
	Requests      []int64    `json:"requests"`
	AvgDurationMs []*float64 `json:"avg_duration_ms"`
}

ProviderLatencySeries is one provider's average request duration per bucket, aligned index-by-index with RequestStats.Buckets. Entries are nil for buckets where the provider served no successful uncached request, so charts can render gaps instead of misleading zeros. Durations are gateway-observed request durations of successful (2xx) requests, excluding local cache hits.

type QueryParams 

type QueryParams struct {
	StartDate time.Time // Inclusive start (day precision)
	EndDate   time.Time // Inclusive end (day precision)
}

QueryParams specifies the date range for audit log retrieval.

type Reader 

type Reader interface {
	// GetLogs returns a paginated list of audit log entries with optional filtering.
	GetLogs(ctx context.Context, params LogQueryParams) (*LogListResult, error)

	// GetLogByID returns a single audit log entry by ID.
	// Returns (nil, nil) when no entry exists for the given ID.
	GetLogByID(ctx context.Context, id string) (*LogEntry, error)

	// GetConversation returns a linear conversation thread around a seed log entry.
	// It follows Responses API linkage fields when available:
	// request_body.previous_response_id and response_body.id.
	GetConversation(ctx context.Context, logID string, limit int) (*ConversationResult, error)

	// GetRequestStats returns time-bucketed status-class counts and
	// per-provider latency aggregates for the dashboard charts.
	GetRequestStats(ctx context.Context, params RequestStatsParams) (*RequestStats, error)
}

Reader provides read access to audit log data for the admin API.

func NewReader 

func NewReader(store storage.Storage) (Reader, error)

NewReader creates an audit log Reader from a storage backend. Returns nil when store is nil.

type RequestRevisionSnapshot 

type RequestRevisionSnapshot struct {
	Seq         int    `json:"seq" bson:"seq"`
	Rewriter    string `json:"rewriter" bson:"rewriter"`
	BytesBefore int    `json:"bytes_before" bson:"bytes_before"`
	BytesAfter  int    `json:"bytes_after" bson:"bytes_after"`

	// TokensSaved is the rewriter-reported estimate of prompt tokens this
	// revision saved (e.g. token compression); zero when the rewriter does
	// not report savings.
	TokensSaved int `json:"tokens_saved,omitempty" bson:"tokens_saved,omitempty"`

	// Body is the request body after this revision (parsed JSON, or a string
	// when not valid JSON). Populated only when body logging is enabled and
	// the body is within the capture limit.
	Body any `json:"body,omitempty" bson:"body,omitempty"`

	// Detail is an optional rewriter-provided structured summary of what
	// changed (for example a compression block report).
	Detail any `json:"detail,omitempty" bson:"detail,omitempty"`

	// NoChange marks a rewriter that ran and left the body untouched. Such
	// revisions record the step for operators — BytesAfter equals BytesBefore,
	// Body is empty and TokensSaved is zero, though Detail may explain why
	// nothing changed — but are not part of the chain that produced the
	// forwarded request. Absent on entries written before no-change steps were
	// tracked, which is why the flag is positive: an old revision always
	// changed the body.
	NoChange bool `json:"no_change,omitempty" bson:"no_change,omitempty"`
}

RequestRevisionSnapshot records one ingress rewrite of the request body, so operators can trace how a request changed on its way to the provider.

type RequestStats 

type RequestStats struct {
	Interval        string                  `json:"interval"`
	Buckets         []RequestStatsBucket    `json:"buckets"`
	Summary         RequestStatsSummary     `json:"summary"`
	ProviderLatency []ProviderLatencySeries `json:"provider_latency"`
}

RequestStats is the time-bucketed request breakdown for the dashboard's status-code and provider-latency charts.

func EmptyRequestStats 

func EmptyRequestStats(interval string) *RequestStats

EmptyRequestStats returns a zero-value result for the disabled-reader fast path so the response shape matches an enabled reader's.

type RequestStatsBucket 

type RequestStatsBucket struct {
	Start       time.Time `json:"start"`
	Requests    int64     `json:"requests"`
	Status2xx   int64     `json:"status_2xx"`
	Status4xx   int64     `json:"status_4xx"`
	Status5xx   int64     `json:"status_5xx"`
	StatusOther int64     `json:"status_other"`
}

RequestStatsBucket counts requests by status class within one time bucket.

type RequestStatsParams 

type RequestStatsParams struct {
	QueryParams

	// Interval is the bucket granularity: StatsIntervalHour or StatsIntervalDay.
	Interval string

	// Location is the dashboard timezone. Daily buckets start at local
	// midnight in this location; hourly buckets are timezone-independent.
	Location *time.Location

	// Now bounds zero-filling: buckets are emitted from the range start up to
	// the earlier of the range end and Now, so a range ending today does not
	// trail empty future buckets.
	Now time.Time
}

RequestStatsParams selects the range and bucketing for GetRequestStats.

type RequestStatsSummary 

type RequestStatsSummary struct {
	Requests      int64    `json:"requests"`
	Status2xx     int64    `json:"status_2xx"`
	Status4xx     int64    `json:"status_4xx"`
	Status5xx     int64    `json:"status_5xx"`
	StatusOther   int64    `json:"status_other"`
	SuccessRate   *float64 `json:"success_rate,omitempty"`
	AvgDurationMs *float64 `json:"avg_duration_ms,omitempty"`
}

RequestStatsSummary aggregates the whole range. SuccessRate is the 2xx share of all requests; AvgDurationMs averages successful, uncached requests. Both are nil when the range has no qualifying requests.

type Result 

type Result struct {
	Logger LoggerInterface
}

Result holds the initialized audit logger. The caller is responsible for calling Close() to release resources.

func New 

func New(ctx context.Context, cfg *config.Config, store storage.Storage) (*Result, error)

New creates an audit logger on the shared storage connection. The caller must call Result.Close() during shutdown.

If logging is disabled in the config, returns a NoopLogger and never touches storage.

func (*Result) Close 

func (r *Result) Close() error

Close releases all resources held by the audit logger. Safe to call multiple times.

type SQLReader added in v0.1.60 

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

SQLReader implements Reader for SQL databases.

func NewSQLReader added in v0.1.60 

func NewSQLReader(db sqlx.DB) (*SQLReader, error)

NewSQLReader creates an audit log reader over a SQL database.

func (*SQLReader) GetConversation added in v0.1.60 

func (r *SQLReader) GetConversation(ctx context.Context, logID string, limit int) (*ConversationResult, error)

GetConversation returns a linear conversation thread around a seed log entry.

func (*SQLReader) GetLogByID added in v0.1.60 

func (r *SQLReader) GetLogByID(ctx context.Context, id string) (*LogEntry, error)

GetLogByID returns a single audit log entry by ID.

func (*SQLReader) GetLogs added in v0.1.60 

func (r *SQLReader) GetLogs(ctx context.Context, params LogQueryParams) (*LogListResult, error)

GetLogs returns a paginated list of audit log entries.

func (*SQLReader) GetRequestStats added in v0.1.60 

func (r *SQLReader) GetRequestStats(ctx context.Context, params RequestStatsParams) (*RequestStats, error)

GetRequestStats returns time-bucketed status-class counts and per-provider latency aggregates for the dashboard charts.

type SQLStore added in v0.1.60 

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

SQLStore implements LogStore for SQL databases.

func NewSQLStore added in v0.1.60 

func NewSQLStore(ctx context.Context, db sqlx.DB, retentionDays int) (*SQLStore, error)

NewSQLStore creates a SQL audit log store, creating its tables if needed and starting the retention sweep when one is configured.

func (*SQLStore) Close added in v0.1.60 

func (s *SQLStore) Close() error

Close stops the cleanup goroutine. The connection is managed by the storage layer. Safe to call multiple times.

func (*SQLStore) Flush added in v0.1.60 

func (s *SQLStore) Flush(_ context.Context) error

Flush is a no-op: writes are synchronous.

func (*SQLStore) WriteBatch added in v0.1.60 

func (s *SQLStore) WriteBatch(ctx context.Context, entries []*LogEntry) error

WriteBatch writes log entries, chunked to stay within the parameter limit.

type StreamLogObserver 

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

StreamLogObserver reconstructs stream metadata and optional response bodies from parsed SSE JSON payloads.

func NewStreamLogObserver 

func NewStreamLogObserver(logger LoggerInterface, entry *LogEntry, path string) *StreamLogObserver

func (*StreamLogObserver) OnJSONEvent 

func (o *StreamLogObserver) OnJSONEvent(event map[string]any)

func (*StreamLogObserver) OnStreamClose 

func (o *StreamLogObserver) OnStreamClose()

func (*StreamLogObserver) WantsJSONEvent 

func (o *StreamLogObserver) WantsJSONEvent([]byte) bool

WantsJSONEvent reports whether this observer consumes stream payloads at all. With body capture disabled it consumes none, letting the observed stream skip per-chunk JSON decoding on its behalf.

type WorkflowFeaturesSnapshot 

type WorkflowFeaturesSnapshot struct {
	Cache      bool `json:"cache" bson:"cache"`
	Audit      bool `json:"audit" bson:"audit"`
	Usage      bool `json:"usage" bson:"usage"`
	Budget     bool `json:"budget" bson:"budget"`
	Guardrails bool `json:"guardrails" bson:"guardrails"`
	Failover   bool `json:"failover" bson:"failover"`
}

WorkflowFeaturesSnapshot stores the effective workflow feature state that applied to one request. Fields intentionally do not use omitempty so "false" remains explicit once the snapshot exists.

Source Files

View all Source files

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.