Documentation
¶
Overview ¶
Package sqlite is obsdb's default backend: the observability schema in a single SQLite file on the CGO-free modernc.org/sqlite driver (thread/sqlite's choice). Open returns an obsdb.DB that also implements interface{ Hub() obsdb.Hub } — every Write publishes its frames to that in-process hub before returning, which is what setup A's live lane runs on (weft/otel's local sink writes, Studio's SSE handler subscribes, no network).
The handle shape is one writer connection plus a read pool: the local sink's synchronous processors write on the run's goroutine, Studio reads concurrently; WAL lets a reader in another process see the same data through its own Open.
Index ¶
- Variables
- func Open(path string, opts ...Option) (obsdb.DB, error)
- type DB
- func (d *DB) Breakpoints(ctx context.Context, runtimeID string) (_ []string, _ bool, err error)
- func (d *DB) Catalogs(ctx context.Context, runID string) (_ []obsdb.ToolsRecord, err error)
- func (d *DB) ChildRuns(ctx context.Context, parentIDs []string) (_ map[string][]obsdb.RunRow, err error)
- func (d *DB) Close() error
- func (d *DB) Commands(ctx context.Context, q obsdb.CommandQuery) (_ []obsdb.StoredCommand, err error)
- func (d *DB) Compactions(ctx context.Context, runID string) (_ []obsdb.Compaction, err error)
- func (d *DB) CountLogs(ctx context.Context, runID string, minSeverity int) (_ obsdb.LogCounts, err error)
- func (d *DB) DeleteRun(ctx context.Context, id string) (err error)
- func (d *DB) DeleteSession(ctx context.Context, id string) (err error)
- func (d *DB) Events(ctx context.Context, runID string, after int64, limit int) (_ obsdb.EventPage, err error)
- func (d *DB) Experiment(ctx context.Context, id string) (_ obsdb.Experiment, err error)
- func (d *DB) Experiments(ctx context.Context) (_ []obsdb.Experiment, err error)
- func (d *DB) Hub() obsdb.Hub
- func (d *DB) Manifest(ctx context.Context, hash string) (_ obsdb.StoredManifest, _ bool, err error)
- func (d *DB) OtherLogs(ctx context.Context, runID string, q obsdb.LogQuery) (_ obsdb.LogPage, err error)
- func (d *DB) Path() string
- func (d *DB) Prompt(ctx context.Context, runID, hash string) (_ obsdb.PromptRecord, err error)
- func (d *DB) PruneCommands(ctx context.Context, before time.Time, keep int) (n int, err error)
- func (d *DB) PutBreakpoints(ctx context.Context, runtimeID string, tools []string) (err error)
- func (d *DB) PutCommand(ctx context.Context, c obsdb.StoredCommand) (err error)
- func (d *DB) PutManifest(ctx context.Context, m obsdb.StoredManifest) (err error)
- func (d *DB) PutRoute(ctx context.Context, r obsdb.RunRoute) (err error)
- func (d *DB) PutRuntime(ctx context.Context, r obsdb.StoredRuntime) (err error)
- func (d *DB) Requests(ctx context.Context, runID string, q obsdb.RequestQuery) (_ []obsdb.RequestRecord, err error)
- func (d *DB) ResolvePublicID(ctx context.Context, publicID string) (_ string, err error)
- func (d *DB) Route(ctx context.Context, runID string) (_ obsdb.RunRoute, _ bool, err error)
- func (d *DB) Run(ctx context.Context, id string) (_ obsdb.RunDetail, err error)
- func (d *DB) RunSpans(ctx context.Context, runID string) (_ []obsdb.Span, err error)
- func (d *DB) RunTreeSpans(ctx context.Context, q obsdb.TreeQuery) (_ obsdb.RunTree, err error)
- func (d *DB) Runs(ctx context.Context, q obsdb.RunQuery) (_ obsdb.RunPage, err error)
- func (d *DB) Runtimes(ctx context.Context) (_ []obsdb.StoredRuntime, err error)
- func (d *DB) SaveExperiment(ctx context.Context, e obsdb.Experiment) (err error)
- func (d *DB) Search(ctx context.Context, q obsdb.SearchQuery) (_ obsdb.RunPage, err error)
- func (d *DB) Session(ctx context.Context, id string) (_ obsdb.SessionDetail, err error)
- func (d *DB) SessionEvents(ctx context.Context, q obsdb.SessionEventQuery) (_ obsdb.SessionEventPage, err error)
- func (d *DB) Sessions(ctx context.Context, q obsdb.SessionQuery) (_ obsdb.SessionPage, err error)
- func (d *DB) Sweep(ctx context.Context, before time.Time) (n int, err error)
- func (d *DB) Tools(ctx context.Context, runID, hash string) (obsdb.ToolsRecord, error)
- func (d *DB) Trace(ctx context.Context, traceID string) (_ []obsdb.Span, err error)
- func (d *DB) Traces(ctx context.Context, q obsdb.TraceQuery) (_ obsdb.TracePage, err error)
- func (d *DB) Transcript(ctx context.Context, runID string) ([]json.RawMessage, error)
- func (d *DB) TranscriptBatches(ctx context.Context, runID string) (_ []obsdb.TranscriptBatch, err error)
- func (d *DB) Vacuum(ctx context.Context) (err error)
- func (d *DB) Write(ctx context.Context, b obsdb.Batch) (err error)
- type Option
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ErrNewerSchema = errors.New("sqlite: database schema is newer than this obsdb")
ErrNewerSchema is returned by Open when the file's obsdb_migrations is ahead of this binary's highest migration: a database written by a newer weft fails loudly instead of silently ignoring tables it does not know.
Functions ¶
func Open ¶
Open opens (creating if needed) the database at path and brings its schema up to date. ":memory:" works — one in-process database. File connections carry journal_mode=WAL (a persistent property of the file, switched once), synchronous=NORMAL, busy_timeout=30000, foreign_keys=ON and _txlock=immediate (preventing deferred-to-writer upgrade deadlocks); the writer handle allows a single connection so every write is serialized, the read pool serves queries.
Types ¶
type DB ¶
type DB struct {
// contains filtered or unexported fields
}
DB is the SQLite backend. Create it with Open; the zero value is not usable.
func (*DB) Breakpoints ¶ added in v0.15.0
Breakpoints answers runtimeID's stored set.
func (*DB) ChildRuns ¶ added in v0.17.0
func (d *DB) ChildRuns(ctx context.Context, parentIDs []string) (_ map[string][]obsdb.RunRow, err error)
ChildRuns is obsdb.TreeReader's: the parents' children in one statement per childChunk ids, `parent_run_id IN (…)` on runs_started (0014: parent_run_id leads it), grouped in Go; each statement reads at most what is left of obsdb.ChildRunsCap and one more, the one more answering obsdb.ErrChildRunsCapped.
func (*DB) Commands ¶ added in v0.15.0
func (d *DB) Commands(ctx context.Context, q obsdb.CommandQuery) (_ []obsdb.StoredCommand, err error)
Commands answers the rows q selects, newest first.
func (*DB) Compactions ¶ added in v0.10.0
func (*DB) CountLogs ¶ added in v0.16.0
func (d *DB) CountLogs(ctx context.Context, runID string, minSeverity int) (_ obsdb.LogCounts, err error)
CountLogs counts a run's app log records at or above minSeverity (obsdb.CountOtherLogs), the candidates narrowed by severity in the query.
func (*DB) DeleteRun ¶ added in v0.14.0
DeleteRun removes a run and its subagent children (obsdb.DB).
func (*DB) DeleteSession ¶ added in v0.14.0
DeleteSession removes every run of a session, children included (obsdb.DB).
func (*DB) Experiment ¶
Experiment returns one experiment; ErrNotFound otherwise.
func (*DB) Experiments ¶
Experiments lists the saved experiments, newest update first.
func (*DB) Hub ¶
Hub returns the DB's in-process live hub: every Write publishes its frames to it before returning (D4 — this handle is what setup A's studio.DB(otel.LocalDB()) shares).
func (*DB) OtherLogs ¶ added in v0.10.0
func (d *DB) OtherLogs(ctx context.Context, runID string, q obsdb.LogQuery) (_ obsdb.LogPage, err error)
OtherLogs reads a run's app log records from other_logs, where Write puts every record with no weft.run.id (obsdb.ReadOtherLogs: the attribution through the run's spans, the order, the paging). The candidates are read by trace id (other_logs_trace) inside the run's time window, the first obsdb.MaxLogCandidates by time.
func (*DB) Path ¶ added in v0.11.0
Path returns the database file's absolute path, as Open resolved it against the working directory at open time; "" for ":memory:". Studio reads it for api/meta's db.path (through interface{ Path() string }, so another backend may answer too).
Example ¶
ExampleDB_Path: the file a database lives in, through the interface Studio's api/meta reads (db.path).
package main
import (
"fmt"
"os"
"path/filepath"
"github.com/weftgo/weft/obsdb/sqlite"
// The test writes a fake newer migration by hand.
_ "modernc.org/sqlite"
)
func main() {
dir, _ := os.MkdirTemp("", "obsdb")
defer func() { _ = os.RemoveAll(dir) }()
db, err := sqlite.Open(filepath.Join(dir, "weft.db"))
if err != nil {
panic(err)
}
defer func() { _ = db.Close() }()
p := db.(interface{ Path() string }).Path()
fmt.Println(filepath.Base(p), filepath.IsAbs(p))
}
Output: weft.db true
func (*DB) PruneCommands ¶ added in v0.15.0
PruneCommands applies the store's retention. The terminal commands go in passes of at most obsdb.SweepBatch rows, each its own writer transaction with ctx checked between them — the age rule's passes, then the keep rule's, which recounts in each — so a large backlog never holds the writer for the whole prune, as Sweep never does; the routes, breakpoints and runtimes clean-up is one transaction after them. On an error the passes before it stay deleted and are counted.
func (*DB) PutBreakpoints ¶ added in v0.15.0
PutBreakpoints replaces runtimeID's set whole.
func (*DB) PutCommand ¶ added in v0.15.0
PutCommand upserts one command row whole.
func (*DB) PutManifest ¶ added in v0.16.0
PutManifest stores m unless its hash is stored: the first write's row stays.
func (*DB) PutRuntime ¶ added in v0.15.0
PutRuntime upserts one runtime's registration whole.
func (*DB) Requests ¶ added in v0.10.0
func (d *DB) Requests(ctx context.Context, runID string, q obsdb.RequestQuery) (_ []obsdb.RequestRecord, err error)
func (*DB) ResolvePublicID ¶
func (*DB) RunTreeSpans ¶ added in v0.17.0
RunTreeSpans is obsdb.TreeReader's. A weft run's subtree is found from the run rows before a span is read (treeRowsSQL, one recursive statement; obsdb.TreeMembers places a row whose parent chain does not reach the run by the id rule), then the members' spans are read by run id, in any trace, capped (treeSpansSQL, obsdb.CapTree): the run's trace is never read whole. A derived run (no run id on its spans) is cut from its trace, as RunSpans reads it, the trace's first obsdb.TreeTraceCap spans (stockTreeWhere). Every statement is bounded by the tree's own size: the walk visits each member row once.
func (*DB) Runtimes ¶ added in v0.15.0
Runtimes answers every stored runtime, most recently seen first.
func (*DB) SaveExperiment ¶
SaveExperiment upserts one experiment by id. A new id stamps Created; an existing one keeps it and moves Updated.
func (*DB) SessionEvents ¶ added in v0.18.0
func (d *DB) SessionEvents(ctx context.Context, q obsdb.SessionEventQuery) (_ obsdb.SessionEventPage, err error)
SessionEvents pages a session's thread records and compaction markers from the session_events index (obsdb.SessionEventReader; migration 0022): one session (its public id's under q.PublicID) in (time, run, kind, body) order — the ClickHouse reader's order, body the tie-break two records of one time, run and kind need — session_events_time serving both the count and the page.
func (*DB) Sessions ¶
func (d *DB) Sessions(ctx context.Context, q obsdb.SessionQuery) (_ obsdb.SessionPage, err error)
Sessions is a GROUP BY session_id over top-level, non-playground runs (S3.4): turns, first/last seen, summed usage, and the newest turn's derived status. A PublicID scopes the runs before the GROUP BY (RunQuery.PublicID's WHERE): every aggregate, the order, Total and the cursor are that public id's runs alone.
func (*DB) Sweep ¶ added in v0.14.0
Sweep deletes the runs not running whose last_seen is before the cutoff, with their children that are not running (obsdb.DB). A run's subtree is swept from its topmost not-running ancestor: a candidate whose parent is a stored run is reached through it, or — when the parent is running, or was never stored — is its own root. Each pass is one writer transaction over at most obsdb.SweepBatch roots, so a write waits on one pass, never on the whole sweep; a pass that found fewer roots than the batch was the last. The passes done, the orphan spans older than the cutoff go too (sweepOrphanSpans).
func (*DB) Traces ¶ added in v0.14.0
Traces groups the spans table by trace id (obsdb.DB.Traces): the filters in HAVING over the group's aggregates — a trace's Start is MIN(start_ns), never one span's — and three keyed reads for the page's roots, services and run ids. With no window it is one scan of the primary key's order; a Since window narrows the groups first to the traces with a span at or after it (spans_start, migration 0019; a necessary condition of MIN(start_ns) >= since, so HAVING still decides — obsdb/clickhouse's traceInner narrows the same way). The scope filter reads weft.public_id off every span's attrs, the attribute /api/traces/{id}'s scope check reads.
func (*DB) Transcript ¶
func (*DB) TranscriptBatches ¶ added in v0.10.0
func (*DB) Vacuum ¶ added in v0.14.0
Vacuum folds the search index's segments and runs VACUUM (obsdb.DB): the whole file is rewritten, every table's contents unchanged — thread's too, when it shares the file. VACUUM needs the file to itself for its duration; a writer in another process waits on the busy timeout.
func (*DB) Write ¶
Write stores one batch in a single transaction: spans and the event/messages records first (INSERT OR IGNORE on the transport idempotency keys), then one runs upsert per run the batch touched.
The rules (S3.4, D3): deltas are counted, never inserted unless KeepDeltas; heartbeats are never inserted, they only move last_seen; counts increment only for rows actually inserted, so retries don't inflate them; any record or span — heartbeats and deltas included — sets last_seen to its max, started_ns to its min (a span's start, the earliest winning: run_start's time, or the invoke_agent span's just before it) and fills empty identity columns; run_finish sets the terminal fields; the invoke_agent span sets trace, finish, usage, steps, root_ended, weft_version (its attribute, else its resource's; the larger beside run_start's, ClickHouse's max) and, on error, failed and err; any weft span fills an empty manifest_hash. A run first seen through a later record (a reordered batch) gets a provisional started_ns, corrected when the earlier rows land. A stock GenAI span (no weft.run.id, obsdb.StockCandidate) re-derives its trace's derived runs (deriveStock, docs/record-contract.md §9).
After the commit, every frame the batch produced is published to the DB's hub before Write returns — an in-process subscriber sees a record within this call (setup A's live lane).
type Option ¶
type Option interface {
// contains filtered or unexported methods
}
Option configures Open.
func KeepDeltas ¶
func KeepDeltas() Option
KeepDeltas turns delta storage on for debugging (Q4, closed): deltas are counted but never stored by default, on their own counter, so storing them can never open a hole in the durable event sequence.
Source Files