obs

package
v0.0.0-...-effd846 Latest Latest
Warning

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

 Go to latest Published: Jun 29, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package obs is the observability surface from spec 18: the metrics catalogue, the structured slow-query log, the flat-oracle recall sampler, and the health and readiness report. It is engine-agnostic. The metric updaters take plain values, the recall sampler takes search functions, and the health report takes a snapshot, so the package is testable without the storage engine and the server composes it by feeding it those inputs.

Metrics export as Prometheus text exposition (spec 18 section 1.5: pull-based, no out-of-band agent). The registry here is self-contained and adds no client dependency; a deployment that already runs the Prometheus client wraps the registry in a prometheus.Collector, which is the one adapter that lives outside this package.

Index

Constants

View Source 
const (
	MQueryDuration          = "vec_query_duration_seconds"
	MQueryTotal             = "vec_query_total"
	MQueryCandidatesVisited = "vec_query_candidates_visited"
	MQueryRerankCount       = "vec_query_rerank_count"
	MQueryK                 = "vec_query_k"
	MQueryEfEffective       = "vec_query_ef_effective"
	MQueryNprobeEffective   = "vec_query_nprobe_effective"
	MQueryFilterSelectivity = "vec_query_filter_selectivity"
	MPlanCacheHits          = "vec_query_plan_cache_hits_total"
	MPlanCacheMisses        = "vec_query_plan_cache_misses_total"
	MHybridRRFFused         = "vec_query_hybrid_rrf_fused_total"

	MRecallEstimate        = "vec_recall_estimate"
	MRecallShadowQueries   = "vec_recall_shadow_queries_total"
	MRecallRerankAgreement = "vec_recall_rerank_agreement"
	MRecallEstimateAge     = "vec_recall_estimate_age_seconds"
	MRecallAlarmTotal      = "vec_recall_alarm_total"

	MIndexBuildDuration   = "vec_index_build_duration_seconds"
	MIndexBuildProgress   = "vec_index_build_progress"
	MIndexBuildActive     = "vec_index_build_active"
	MIndexInsertDuration  = "vec_index_insert_duration_seconds"
	MIndexDeleteTombstone = "vec_index_delete_tombstone_total"

	MWALSizeBytes       = "vec_wal_size_bytes"
	MCheckpointDuration = "vec_checkpoint_duration_seconds"
	MCheckpointTotal    = "vec_checkpoint_total"
	MSegmentCount       = "vec_segment_count"
	MFragmentationRatio = "vec_fragmentation_ratio"
	MFileSizeBytes      = "vec_file_size_bytes"
	MPageCacheSizeBytes = "vec_page_cache_size_bytes"
	MPageCacheHits      = "vec_page_cache_hits_total"
	MPageCacheMisses    = "vec_page_cache_misses_total"

	MWriterQueueDepth   = "vec_writer_queue_depth"
	MWriterWaitDuration = "vec_writer_wait_duration_seconds"
	MUpsertTotal        = "vec_upsert_total"
	MDeleteTotal        = "vec_delete_total"

	MGCPauseDuration = "vec_gc_pause_duration_seconds"
	MHeapAllocBytes  = "vec_heap_alloc_bytes"
	MGoroutines      = "vec_goroutines"
)

Metric names from the spec 18 section 2 catalogue. They are exported so the server, the tests, and any external alerting rule reference the same strings.

View Source 
const (
	StatusOK       = "ok"
	StatusDegraded = "degraded"
	StatusNotReady = "not_ready"
)

Health status values (spec 18 §7.1). ok passes traffic; degraded passes traffic but fires an alert; not_ready blocks traffic.

Variables

View Source 
var GCBuckets = []float64{0.00005, 0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1}

GCBuckets covers GC stop-the-world pauses in seconds (spec 18 §2.8). The range runs from 50us to 100ms, which spans a healthy young-gen pause up to a pause long enough to show in the latency tail.

View Source 
var LatencyBuckets = []float64{0.0001, 0.0005, 0.001, 0.002, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 1.0}

LatencyBuckets is the upper-bound set for query latency histograms in seconds (spec 18 section 2.2). The buckets match the spec table exactly.

Functions

func HashVector 

func HashVector(raw []byte) string

HashVector returns the truncated SHA-256 of a query vector's float32 bytes (spec 18 §3.4). The log records this, never the vector itself, so an embedding does not leak into the log. The caller passes the raw little-endian float32 bytes.

func RerankAgreement 

func RerankAgreement(preTopK, finalTopK []uint64, k int) float64

RerankAgreement is the cheaper recall proxy of spec 18 §5.4: among the final top-k, the fraction that was already in the pre-rerank top-k. It is computed from data the query pipeline already holds, so it costs nothing extra. preTopK is the ann ranking before reranking; finalTopK is the ranking after.

Types

type CollectionHealth 

type CollectionHealth struct {
	Status             string  `json:"status"`
	IndexLoaded        bool    `json:"index_loaded"`
	WALReplayed        bool    `json:"wal_replayed"`
	RecallEstimate     float64 `json:"recall_estimate"`
	RecallEstimateAgeS float64 `json:"recall_estimate_age_s"`
	// Reason names the degraded or not_ready condition, empty when ok. It is not
	// in the spec's JSON example but the endpoint includes it so an operator sees
	// why without cross-referencing the metrics.
	Reason string `json:"reason,omitempty"`
}

CollectionHealth is the per-collection health detail (spec 18 §7.1).

type CollectionSnapshot 

type CollectionSnapshot struct {
	IndexLoaded        bool
	WALReplayed        bool
	RecallEstimate     float64
	RecallEstimateAgeS float64
	WALSizeBytes       int64
	Fragmentation      float64
}

CollectionSnapshot is the raw state of one collection that the health evaluator grades against the thresholds. The server fills it from the engine; the obs package does not reach into storage.

type Counter 

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

Counter is a monotone counter updated without a lock on the hot path (spec 18 section 2.10).

func (*Counter) Add 

func (c *Counter) Add(delta int64)

Add increments the counter by delta.

func (*Counter) Inc 

func (c *Counter) Inc()

Inc increments the counter by one.

func (*Counter) Value 

func (c *Counter) Value() int64

Value returns the current count.

type Gauge 

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

Gauge is a value that goes up and down (spec 18 section 2.5). It stores a float64 in an atomic uint64 bit pattern so reads and writes are lock-free.

func (*Gauge) Set 

func (g *Gauge) Set(v float64)

Set stores v.

func (*Gauge) Value 

func (g *Gauge) Value() float64

Value returns the current value.

type HealthReport 

type HealthReport struct {
	Status      string                      `json:"status"`
	Collections map[string]CollectionHealth `json:"collections"`
	UptimeS     int64                       `json:"uptime_s"`
	Version     string                      `json:"version"`
}

HealthReport is the body of the /health and /ready endpoints (spec 18 §7.1). The top-level status is the worst status across collections: not_ready if any collection is not ready, else degraded if any is degraded, else ok.

func BuildHealthReport 

func BuildHealthReport(t HealthThresholds, snaps map[string]CollectionSnapshot, version string, uptimeS int64) HealthReport

BuildHealthReport grades every collection snapshot and rolls the per-collection statuses up to a single top-level status (spec 18 §7.1). version and uptimeS are stamped by the caller, which holds the clock.

func (HealthReport) Ready 

func (r HealthReport) Ready() bool

Ready reports whether the report permits traffic (spec 18 §7.1). not_ready blocks; ok and degraded pass.

type HealthThresholds 

type HealthThresholds struct {
	// RecallWarnAgeS marks a collection degraded when its recall estimate is older
	// than this many seconds (default 3600, spec §17.5 health_recall_warn_age_s).
	RecallWarnAgeS float64
	// RecallFloor marks a collection degraded when its recall estimate is below
	// this floor. Zero disables the floor check.
	RecallFloor float64
	// WALSizeWarnBytes marks a collection degraded when its WAL exceeds this size
	// (default 128MB, spec §17.5 health_wal_size_warn_bytes).
	WALSizeWarnBytes int64
	// FragmentationWarn marks a collection degraded above this ratio (default 0.50,
	// spec §17.5 health_fragmentation_warn).
	FragmentationWarn float64
}

HealthThresholds are the limits that turn an ok collection degraded (spec 18 §7.1, §17.5). Zero on a field disables that check.

func DefaultHealthThresholds 

func DefaultHealthThresholds() HealthThresholds

DefaultHealthThresholds returns the spec §17.5 defaults.

func (HealthThresholds) Grade 

func (t HealthThresholds) Grade(s CollectionSnapshot) CollectionHealth

Grade turns one snapshot into a CollectionHealth (spec 18 §7.1). A collection that has not loaded its index or replayed its WAL is not_ready and blocks traffic. A loaded, replayed collection is degraded when any warn threshold is crossed, else ok. The first crossed condition is named in Reason.

type Histogram 

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

Histogram is a cumulative bucketed distribution (spec 18 section 2.2). Bucket counts, the running sum, and the total count are atomic, so Observe takes no lock on the query hot path.

func NewHistogram 

func NewHistogram(bounds []float64) *Histogram

NewHistogram builds a histogram over the given upper bounds, which must be sorted ascending. The implicit final +Inf bucket is added automatically.

func (*Histogram) Count 

func (h *Histogram) Count() int64

Count returns the number of observations.

func (*Histogram) Observe 

func (h *Histogram) Observe(v float64)

Observe records one sample.

func (*Histogram) Sum 

func (h *Histogram) Sum() float64

Sum returns the sum of all observations.

type Metrics 

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

Metrics is the engine-facing recorder. It wraps a Registry and offers typed methods for the hot paths so callers do not repeat metric names and label keys at every call site (spec 18 section 2.10). The same Metrics is shared across the whole DB; the library returns it from db.Metrics and the server mounts the same instance on /metrics, so there is one set of counters and no double count.

func NewMetrics 

func NewMetrics() *Metrics

NewMetrics builds a Metrics over a fresh registry.

func (*Metrics) PlanCacheHit 

func (m *Metrics) PlanCacheHit(collection string)

PlanCacheHit and PlanCacheMiss record the plan cache outcome (spec 18 §2.2).

func (*Metrics) PlanCacheMiss 

func (m *Metrics) PlanCacheMiss(collection string)

func (*Metrics) RecallAlarm 

func (m *Metrics) RecallAlarm(collection string)

RecallAlarm increments the recall regression counter (spec 18 §5.3).

func (*Metrics) RecordDelete 

func (m *Metrics) RecordDelete(collection, status string)

func (*Metrics) RecordQuery 

func (m *Metrics) RecordQuery(o QueryObservation)

RecordQuery folds one query's observation into the metrics. It is the single call the executor makes when a query finishes.

func (*Metrics) RecordRecall 

func (m *Metrics) RecordRecall(collection, index string, k int, estimate, ageSeconds float64)

RecordRecall records a recall estimate and its freshness (spec 18 §2.3).

func (*Metrics) RecordUpsert 

func (m *Metrics) RecordUpsert(collection, status string)

RecordUpsert and RecordDelete record write outcomes (spec 18 §2.7).

func (*Metrics) Registry 

func (m *Metrics) Registry() *Registry

Registry returns the underlying registry, for exposition and for the prometheus.Collector adapter a deployment may wrap around it.

func (*Metrics) SetFileSize 

func (m *Metrics) SetFileSize(collection string, bytes int64)

func (*Metrics) SetFragmentation 

func (m *Metrics) SetFragmentation(collection string, ratio float64)

func (*Metrics) SetWALSize 

func (m *Metrics) SetWALSize(collection string, bytes int64)

SetWALSize, SetFileSize, SetFragmentation record storage gauges (spec 18 §2.5).

type QueryObservation 

type QueryObservation struct {
	Collection        string
	Index             string
	FilterPresent     bool
	Status            string // ok / error / timeout
	DurationSeconds   float64
	CandidatesVisited int
	RerankCount       int
	K                 int
	EfEffective       int
	NprobeEffective   int
	FilterSelectivity float64 // -1 if unknown
}

QueryObservation is the set of values one completed query contributes to the metrics (spec 18 section 2.2). Zero-valued optional fields are skipped.

type RecallOptions 

type RecallOptions struct {
	// Collection and Index name the series the estimate is recorded under.
	Collection string
	Index      string
	// SampleSize is how many probe points to draw per run (default 100).
	SampleSize int
	// K is the neighbor count to measure recall at (default 10).
	K int
	// AlarmThreshold fires the recall-regression alarm when the estimate drops
	// below it (spec 18 §5.3). Zero disables the alarm.
	AlarmThreshold float64
}

RecallOptions configures one sampler (spec 18 §5.2 PRAGMAs).

type RecallResult 

type RecallResult struct {
	// Estimate is the mean recall@k across the sampled points, in [0,1].
	Estimate float64
	// Samples is the number of probe points that contributed (draws that failed a
	// search are skipped and do not count).
	Samples int
	// Alarmed reports whether the estimate fell below the alarm threshold.
	Alarmed bool
}

RecallResult is the outcome of one sampling run.

type RecallSampler 

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

RecallSampler estimates recall in production without ground-truth labels, by shadow-sampling against the exact flat search (spec 18 §5.2). It picks stored points at random, runs the ANN query and the exact query for each, and reports the average intersection over k. The two searches are injected as functions so the sampler does not depend on the index or the executor.

func NewRecallSampler 

func NewRecallSampler(opts RecallOptions, m *Metrics) *RecallSampler

NewRecallSampler builds a sampler that records into m. Zero SampleSize and K fall back to the spec defaults.

func (*RecallSampler) Run 

func (s *RecallSampler) Run(src SampleSource, ann, flat SearchFunc, ageSeconds float64) (RecallResult, error)

Run executes one sampling pass (spec 18 §5.2). It draws probe points from src, runs ann and flat for each, and averages the per-point recall. ageSeconds is the freshness stamp recorded with the estimate; the caller passes the seconds elapsed since the previous run (the sampler holds no clock, per the no-Date.now constraint of the workflow harness). A run that draws nothing returns a zero estimate without recording, so an empty collection does not look like a recall collapse.

type Registry 

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

Registry holds every metric family and renders Prometheus text exposition (spec 18 section 1.5, 2.10). It is safe for concurrent use.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry builds an empty registry.

func (*Registry) Counter 

func (r *Registry) Counter(name, help string, labelPairs ...string) *Counter

Counter returns the counter for a metric name and label pairs, creating it on first use (spec 18 section 2.2). Help is recorded the first time the name is seen.

func (*Registry) Gauge 

func (r *Registry) Gauge(name, help string, labelPairs ...string) *Gauge

Gauge returns the gauge for a metric name and label pairs (spec 18 section 2.5).

func (*Registry) Histogram 

func (r *Registry) Histogram(name, help string, bounds []float64, labelPairs ...string) *Histogram

Histogram returns the histogram for a metric name and label pairs (spec 18 section 2.2). The bounds are used only when the series is first created.

func (*Registry) Text 

func (r *Registry) Text() string

Text returns the exposition as a string.

func (*Registry) WriteText 

func (r *Registry) WriteText(w *strings.Builder)

WriteText renders the whole registry in Prometheus text exposition format (spec 18 section 1.5). Families come out in registration order and series within a family come out in sorted label order, so the output is deterministic.

type RuntimeCollector 

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

RuntimeCollector samples the Go runtime and folds the result into the runtime metrics of spec 18 §2.8: GC pause duration, heap allocated bytes, and live goroutine count. It is poll-based (spec §12.2): the server calls Collect on a ticker, default every 10 seconds. The collector keeps the last PauseTotalNs so each Collect records only the pauses since the previous call.

func NewRuntimeCollector 

func NewRuntimeCollector(m *Metrics) *RuntimeCollector

NewRuntimeCollector builds a collector that records into m.

func (*RuntimeCollector) Collect 

func (c *RuntimeCollector) Collect()

Collect reads runtime.MemStats once and updates the runtime metrics (spec 18 §12.2). The first call seeds the GC baseline and records the gauges but emits no pause samples, because there is no previous reading to diff against. Later calls observe each GC pause that completed since the previous call, read from the PauseNs circular buffer.

type SampleSource 

type SampleSource func(n int) (ids []uint64, vectors [][]float32, err error)

SampleSource yields the probe points for one run: the ids drawn at random and their vectors (spec 18 §5.2 step 1). The sampler does not own point storage, so the caller supplies the draw.

type SearchFunc 

type SearchFunc func(query []float32, k int) ([]uint64, error)

SearchFunc returns the ids of the top results for a query vector. The ANN variant returns the index's approximate neighbors; the flat variant returns the exact neighbors from a brute-force scan (spec 18 §5.2 steps 2 and 3).

type SlowQueryLogger 

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

SlowQueryLogger decides whether a query is logged and emits the record (spec 18 §3). It assigns each query a monotone id and samples with a lock-free PRNG so the decision adds no contention to the query hot path.

func NewSlowQueryLogger 

func NewSlowQueryLogger(opts SlowQueryOptions, seed uint64) *SlowQueryLogger

NewSlowQueryLogger builds a logger. seed seeds the sampling PRNG; the caller passes a value from crypto/rand at startup (spec 18 §3.3). A zero seed is replaced with a fixed nonzero constant so the generator never sticks at zero.

func (*SlowQueryLogger) Log 

func (l *SlowQueryLogger) Log(ctx context.Context, rec SlowQueryRecord)

Log emits a record if the query qualifies (spec 18 §3). The duration field on the record decides; ShouldLog is applied here so the caller has one entry point. Redaction is applied to the filter representation before writing.

func (*SlowQueryLogger) NextQueryID 

func (l *SlowQueryLogger) NextQueryID() uint64

NextQueryID returns the id for the next query. The executor stamps it into the record it builds, so the id is the same whether or not the query is logged.

func (*SlowQueryLogger) ShouldLog 

func (l *SlowQueryLogger) ShouldLog(d time.Duration) bool

ShouldLog reports whether a query of the given duration is logged (spec 18 §3.3). A query qualifies if it crosses the threshold or wins the sample.

type SlowQueryOptions 

type SlowQueryOptions struct {
	// Threshold logs any query slower than this. Zero logs every query, which is
	// only sane in combination with SampleRate well below 1.
	Threshold time.Duration
	// SampleRate logs this fraction of all queries regardless of latency, in
	// [0,1]. The threshold and the sample rate compose with OR.
	SampleRate float64
	// Redact replaces the filter text with "<redacted>" (spec 18 §3.4).
	Redact bool
	// Sink receives the records. A nil sink disables logging.
	Sink *slog.Logger
}

SlowQueryOptions configures the slow-query log (spec 18 §3.3, §3.5).

type SlowQueryRecord 

type SlowQueryRecord struct {
	Time              time.Time  `json:"time"`
	Collection        string     `json:"collection"`
	IndexType         string     `json:"index_type"`
	QueryID           uint64     `json:"query_id"`
	DurationMs        float64    `json:"duration_ms"`
	K                 int        `json:"k"`
	EfSearch          int        `json:"ef_search,omitempty"`
	Nprobe            int        `json:"nprobe,omitempty"`
	CandidatesVisited int        `json:"candidates_visited"`
	RerankCount       int        `json:"rerank_count"`
	FilterPresent     bool       `json:"filter_present"`
	FilterSelectiv    float64    `json:"filter_selectivity"`
	PlanShape         string     `json:"plan_shape"`
	RecallEstimate    float64    `json:"recall_estimate"`
	VectorHash        string     `json:"vector_hash"`
	FilterRepr        string     `json:"filter_repr"`
	ErrorMsg          string     `json:"error,omitempty"`
	StageDurations    StageTimes `json:"stage_durations_ms"`
}

SlowQueryRecord is one entry in the structured slow-query log (spec 18 §3.2). The field tags are normative so the JSON shape is stable across a Go caller and a downstream parser.

type StageTimes 

type StageTimes struct {
	Parse     float64 `json:"parse"`
	Plan      float64 `json:"plan"`
	IndexScan float64 `json:"index_scan"`
	Rerank    float64 `json:"rerank"`
	Filter    float64 `json:"filter"`
	Assemble  float64 `json:"assemble"`
}

StageTimes is the per-stage latency breakdown in milliseconds (spec 18 §3.2).

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.