engine

package
v0.40.1 Latest Latest
Warning

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

 Go to latest Published: Aug 22, 2026 License: Apache-2.0 Imports: 39 Imported by: 0

Documentation

Overview

Package engine is the L3 single-node storage engine: an in-memory head that absorbs writes (indexing labels and buffering samples) with a bounded out-of-order window, optionally backed by a write-ahead log for crash recovery. It implements the fetch contract over the head. Flush to immutable parts and background merge (compaction, retention, downsampling) are added in later slices.

Index

Constants

View Source 
const DefaultDecodeBudgetForceAfter = 30 * time.Second

DefaultDecodeBudgetForceAfter is how long a query waits for decode budget before it is admitted anyway. See DecodeBudget.

View Source 
const DefaultMetricBlockRows = 1024

DefaultMetricBlockRows is the metric part block size used when Config.MetricBlockRows is 0. It is finer than the historical 8192-row granule so a sparse selector (a small fraction of a part's series, scattered by SeriesID hash) touches a smaller fraction of blocks.

View Source 
const SampleBytes = 16

SampleBytes is the in-flight memory charged per buffered sample: an int64 timestamp plus a float64 value. It is the unit for the head's byte accounting and for callers sizing a batch against a rate budget. It deliberately ignores per-series index overhead (a small, amortized constant), so it measures the dominant, sample-proportional cost of the unflushed head.

Variables

This section is empty.

Functions

This section is empty.

Types

type AppendLimits added in v0.4.0 

type AppendLimits struct {
	// MaxSeries caps the number of distinct series buffered in the head. A sample that would
	// register a new series once the head already holds MaxSeries is rejected (cardinality
	// backpressure); samples for already-known series are unaffected. 0 ⇒ unlimited.
	MaxSeries int64
	// MaxSeriesSoft, when 0 < MaxSeriesSoft <= MaxSeries together with a non-nil Overflow, is the
	// soft cardinality budget: a *new* series arriving once the head holds at least MaxSeriesSoft is
	// routed (via Overflow) to an overflow series instead of being registered, until MaxSeries is
	// reached (then it is rejected). 0 ⇒ no soft budget (a hard reject at MaxSeries).
	MaxSeriesSoft int64
	// Overflow, when non-nil, builds the overflow series identity for a new series that crosses the
	// soft budget (the caller supplies it so the head stays signal-agnostic — e.g. metrics map to
	// {__name__, __overflow__}). The overflow series itself is exempt from the cap. nil ⇒ no
	// overflow routing.
	Overflow func(s signal.Series) signal.Series
	// MaxInFlightBytes caps the head's buffered sample bytes ([SampleBytes] each). A sample
	// arriving while the head is at or over the cap is rejected (memory backpressure) until a
	// flush drains the head. 0 ⇒ unlimited.
	MaxInFlightBytes int64
}

AppendLimits are the per-call admission limits the head enforces while buffering a batch. The zero value imposes no limit. They are passed per Engine.AppendBatch call (not stored on the engine) so a consumer's hot-reloaded tenant policy takes effect on the next write. The engine stays policy-agnostic — it sees only these numbers, never a tenant or a [tenant.Policy].

type AppendResult added in v0.4.0 

type AppendResult struct {
	Accepted            int
	RejectedOOO         int // older than the out-of-order window
	RejectedCardinality int // would exceed AppendLimits.MaxSeries (a new series)
	RejectedBytes       int // head at or over AppendLimits.MaxInFlightBytes
	// Overflowed counts samples for new series past the soft budget that were routed to an overflow
	// series instead of being rejected. They are also counted in Accepted (the data is retained).
	Overflowed int
}

AppendResult reports the disposition of an Engine.AppendBatch run: how many samples were accepted and how many were rejected, by reason, so the caller can attribute an OTLP partial-success precisely.

func (AppendResult) Rejected added in v0.4.0 

func (r AppendResult) Rejected() int

Rejected returns the total number of rejected samples across all reasons.

type BucketAgg added in v0.6.0 

type BucketAgg struct {
	SeriesAgg

	Start int64
}

BucketAgg is one step-aligned bucket's aggregate for a series: the bucket's start timestamp and the count/sum/min/max of the samples that fall in it.

type CardinalityStat added in v0.12.0 

type CardinalityStat struct {
	TotalSeries        int64
	DistinctLabelNames int
	SymbolCount        int
	Top                []LabelCard // sorted by Series desc, then Name; truncated to the requested top-N
}

CardinalityStat summarizes the engine's label cardinality (the head's index spans head ∪ flushed series). TotalSeries and SymbolCount are exact; Top is the highest-cardinality label names.

type ColumnStat added in v0.12.0 

type ColumnStat struct {
	Name     string
	Kind     string // physical type: int64 / float64 / bytes / int128
	Codec    string // value codec
	Compress string // block-compression algorithm
	Level    int    // block-compression level (0 ⇒ algorithm default or uncompressed)
}

ColumnStat is one part column's physical description (from the manifest).

type Config 

type Config struct {
	// OOOWindow is a per-series lateness bound: a sample older than OOOWindow (nanoseconds) behind
	// the series' own newest admitted sample is rejected. 0 disables.
	OOOWindow int64
	// WAL, when non-nil, durably logs series and samples for crash recovery. nil is the
	// ephemeral in-memory engine.
	WAL *wal.SegmentWriter
	// Backend stores flushed parts. Required for [Engine.Flush]; nil is a head-only engine.
	Backend backend.Backend
	// Prefix is the backend key prefix under which this engine's parts are written
	// (typically "{tenant}/metrics").
	Prefix string
	// Term reports this writer's current ownership term for Prefix — which tenure of the shard
	// this engine is writing as. It is stamped into every bucket index written, so a reader can
	// order two indexes of the same prefix even when neither the part names nor FlushedEpoch
	// moved; see [github.com/oteldb/storage/backend/bucketindex.Generation]. nil is a writer with
	// no cluster, whose generation is then a plain local counter.
	Term func() uint64
	// WriterID is this writer's stable identity — the cluster node id — under which its WAL flush
	// watermark is kept in the bucket index. It matters because that index is *shared*: over a
	// shared object store every replica of a shard commits one index object under one prefix, and
	// the watermark is a per-node count of that node's own flushes, so one scalar in a shared
	// object is meaningless to whichever node did not write it (see
	// [github.com/oteldb/storage/backend/bucketindex.WriterEpoch]).
	//
	// Empty is the anonymous writer: a single-writer engine, which keeps the sole pre-v4 slot.
	// Leaving it empty where two engines do share a prefix makes them share one slot, which is
	// the defect the slots exist to prevent.
	WriterID string
	// Obs is the observability handle (spans + metrics). nil ⇒ a no-op handle, so an engine
	// constructed without one logs/spans/counts nothing.
	Obs *obs.Obs
	// DecodeCacheBytes enables a cross-fetch cache of decoded part columns, sized to this many
	// bytes (LRU). It skips the column re-decode that the backend read cache cannot, and applies to
	// every backend (a decode is CPU even when the read is RAM-fast). Zero disables it.
	DecodeCacheBytes int64
	// MaxPartBytes caps a *flushed* part's (approximate, uncompressed) size: a flush splits its
	// output so no single part exceeds it. 0 ⇒ unlimited. A merged part is sized by
	// MergeCeilingBytes instead.
	MaxPartBytes int64
	// MergeCeilingBytes is the upper bound on a merged part's size *on disk*; the effective cap is
	// the least of it, this merge's share of the backend's free space, and — over a backend that
	// takes objects whole — what the merge may hold in memory (see mergecap.go). 0 ⇒
	// defaultMergeCeilingBytes; negative ⇒ unlimited (never seal).
	MergeCeilingBytes int64
	// MergeMemoryBytes is how much memory all concurrent merges together may hold. Over a backend
	// that takes objects whole a merged part is buffered encoded in RAM until it is sealed, so this —
	// not free space — is what stops a part from outgrowing the process on a node whose disk dwarfs
	// its memory limit. Over one implementing backend.ObjectCreator the part streams out as it is
	// encoded, so this bounds the per-series state a merge still holds and the disk sizes the part.
	// 0 ⇒ a share of GOMEMLIMIT, or defaultMergeMemoryBytes when the process declares no limit;
	// negative ⇒ unbounded (only the ceiling and free space then apply).
	MergeMemoryBytes int64
	// MergeConcurrency reports how many merges may run concurrently against this backend, dividing
	// the free space so they cannot collectively exhaust the disk. nil or ≤ 1 ⇒ no division.
	//
	// A callback because the answer moves: fan-out is bounded by the node's engine count as much as
	// by its worker limit, and engines appear lazily. Fixing it at engine creation would divide a
	// single-tenant node's disk by its core count, undoing most of the widening.
	MergeConcurrency func() int
	// AggregateStats writes a per-series aggregate sidecar (count/sum/min/max) alongside each part,
	// so [Engine.AggregateRange] answers a range-covering aggregate from it without decoding the
	// value column. It costs a little storage per series; off by default. AggregateRange works
	// without it (via decoding), just without the fast path.
	AggregateStats bool
	// RecentWindow enables an in-memory recent tier (nanoseconds): the most recent flush window is
	// mirrored in RAM across flushes so a query whose [Start, End] falls inside it is answered from
	// the tier ∪ the head without decoding any file part — first-touch recent-range queries skip the
	// decode path entirely (the decode cache only helps repeats). It trades a bounded uncompressed
	// window of resident memory for that latency. 0 disables it (the default); the head is then just
	// the unflushed tail, drained on every flush as before.
	RecentWindow int64
	// DecodeMemoryBytes caps the total in-flight decoded column bytes across concurrent queries: a
	// query reserves its estimated decode footprint before reading parts and releases it when done,
	// blocking when the budget is exhausted. It bounds the query-concurrency RSS cliff (N heavy
	// queries each materializing whole columns) by serializing decode through the cap rather than
	// letting concurrency multiply resident decoded bytes. 0 ⇒ unlimited (no admission control). A
	// query larger than the whole budget runs alone (it cannot be bounded below its own footprint).
	DecodeMemoryBytes int64
	// DecodeBudget, when non-nil, is a pre-built decode-memory budget this engine reserves from
	// instead of building its own from DecodeMemoryBytes. Share one [DecodeBudget] across engines
	// (one engine per tenant) so the cap bounds the process-wide in-flight decoded bytes rather
	// than multiplying per tenant. Takes precedence over DecodeMemoryBytes.
	DecodeBudget *DecodeBudget
	// MinFreeBytes is the headroom the engine leaves unused on a backend that reports its free
	// space: a flush is refused, and the ingest path starts rejecting, once the medium holds less
	// than the pending part plus this. It leaves a merge room for its output — a merge must write
	// before it can retire the inputs it frees, so a disk at 100% cannot compact its way out. 0 ⇒
	// [diskguard.DefaultReserveBytes]; negative ⇒ the byte axis is not checked.
	MinFreeBytes int64
	// MinFreeInodes is the same headroom on the object-count axis, for a backend that reports free
	// inodes. It is checked separately because a part is many small objects: an inode table can
	// exhaust with the disk half empty, and byte accounting cannot see it. 0 ⇒
	// [diskguard.DefaultReserveInodes]; negative ⇒ the inode axis is not checked.
	MinFreeInodes int64
	// MetricBlockRows sets the row block size for metric part columns (ts/value/sf): the columns are
	// split into independently decodable blocks of this many rows, so a query can decode only the
	// blocks its matched series' row ranges touch (sub-part seek) instead of the whole column, and
	// the block boundaries drive the part's marks granules. 0 ⇒ [DefaultMetricBlockRows]. A finer
	// size skips more on sparse selectors at a small per-block header cost.
	MetricBlockRows int
}

Config configures an Engine.

type DecodeBudget added in v0.25.0 

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

DecodeBudget caps the total in-flight decoded bytes across concurrent queries, so query concurrency cannot drive RSS past a bound. Each query estimates its decode footprint (the column buffers it will materialize across the parts it touches) and acquires that many bytes before decoding, releasing them when the fetch ends; an acquire blocks until enough is free. Under load this trades latency for a memory ceiling — N heavy concurrent queries serialize through the budget instead of each allocating GBs at once (the concurrency cliff).

A query whose own estimate exceeds the whole budget is admitted alone (it cannot be bounded below its own footprint), so an unsatisfiable request never deadlocks. The budget is acquired once per query (the whole estimate up front), not incrementally per part, so two queries cannot each hold a partial reservation while waiting on the other.

The ceiling is soft in two more ways, both liveness guards for a caller that holds several reads open at once (which the library cannot detect and must not hang on):

  • the wait is cancellable — a done context aborts it with an error and no reservation, so a query deadline or a client disconnect always recovers the goroutine;
  • the wait is bounded — a waiter that sits for DefaultDecodeBudgetForceAfter without the budget draining at all is admitted regardless, counted and logged. Trading an RSS overshoot for liveness is the same trade the admit-alone rule already makes.

One budget may be shared by multiple engines (via Config.DecodeBudget) so the cap bounds the process-wide decode footprint rather than a per-engine (per-tenant) one.

func NewDecodeBudget added in v0.25.0 

func NewDecodeBudget(maxBytes int64) *DecodeBudget

NewDecodeBudget returns a budget capping in-flight decoded bytes at maxBytes. maxBytes ≤ 0 disables it (every acquire/release is a no-op).

type DecodeCacheStats added in v0.6.0 

type DecodeCacheStats struct {
	Hits, Misses int64
	Bytes        int64
	Items        int // cached blocks
}

DecodeCacheStats is a snapshot of the decoded-block cache's effectiveness.

type DownsampleTier added in v0.4.0 

type DownsampleTier struct {
	Before   int64 // samples with ts < Before are subject to this tier
	Interval int64 // bucket width, nanoseconds
	Agg      signal.Aggregation
}

DownsampleTier is the absolute (wall-clock-free) form of a tenant downsampling tier: every sample older than Before is rolled up into one representative per Interval-wide bucket, the bucket's samples combined by Agg. Buckets are aligned to absolute multiples of Interval, so a time range's rollup does not depend on when the merge runs. A tier with Interval ≤ 0 is ignored. The caller ([storage.Storage]) builds these from [tenant.DownsampleTier] and the current time.

type Engine 

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

Engine is a single tenant's storage engine. Safe for concurrent use.

func New 

func New(cfg Config) *Engine

New returns an engine with an empty head.

func (*Engine) AggregateRange added in v0.6.0 

func (e *Engine) AggregateRange(ctx context.Context, r fetch.Request) (map[signal.SeriesID]SeriesAgg, error)

AggregateRange returns a per-series aggregate (count, sum, min, max — enough for avg) over [r.Start, r.End] for the series matching r.Matchers. It is the aggregate-pushdown read path: for parts the range fully covers, it folds each part's precomputed stats sidecar instead of decoding the value column — so an aggregate over many points returns one number per series for almost no I/O — and falls back to decoding + merging when that would not be exact (a part only partially in range, parts overlapping in time so timestamps could be duplicated, or a sampled/ sidecar-less part). The fast path is taken in the common compacted, time-disjoint case.

Series with no sample in the window are omitted from the result.

func (*Engine) AggregateStep added in v0.6.0 

func (e *Engine) AggregateStep(ctx context.Context, r fetch.Request, step int64) (map[signal.SeriesID][]BucketAgg, error)

AggregateStep returns, per series, the aggregate of each non-empty step-aligned bucket in [r.Start, r.End] — the range-vector shape an embedder's `*_over_time` needs. Buckets align to the absolute grid (multiples of step), so a range's bucketing is independent of when it runs; the returned buckets are sorted ascending by Start. step ≤ 0 collapses to a single whole-range bucket (equivalent to Engine.AggregateRange).

It reuses the stats sidecar where it still applies: when the plan is pushdown-safe (parts fully covered and time-disjoint) and a part falls entirely within one bucket, that bucket folds the part's sidecar without decoding. Parts that straddle buckets, or an unsafe plan, decode (and an unsafe plan merges first, to dedup by timestamp).

func (*Engine) AggregateStepNamed added in v0.8.0 

func (e *Engine) AggregateStepNamed(ctx context.Context, r fetch.Request, step int64) ([]NamedAgg, error)

AggregateStepNamed is Engine.AggregateStep returning each series' identity alongside its buckets, for the cluster aggregate RPC (a peer pushes the aggregate down and ships identities so the coordinator can re-filter and union).

func (*Engine) AggregateWindow added in v0.36.0 

func (e *Engine) AggregateWindow(
	ctx context.Context, r fetch.Request, spec WindowSpec,
) (map[signal.SeriesID][]WindowAgg, error)

AggregateWindow returns, per series, the aggregate of every step-aligned evaluation window (t-window, t] that holds a sample — the *overlapping* range-vector shape, `<fn>_over_time(m[W])` evaluated at a step finer than W. Engine.AggregateStep is the disjoint special case; this is the 12x-overlap one (a 1h range at a 5m step).

The cost is proportional to the data in the request window, not to window/step times it: samples fold once into disjoint step-wide fine buckets (reusing the sidecar pushdown of Engine.AggregateStep — a part wholly inside one fine bucket never decodes), and a sliding accumulator then walks those buckets once per series, adding the bucket that enters each window and subtracting the one that leaves — and, since an extremum cannot be subtracted back out, tracking min/max on monotonic deques. Advancing a step is O(1) regardless of how wide the window is.

Windows align to the absolute grid (End is a multiple of step) and are returned sorted ascending by End; empty windows and series with no sample are omitted. The window is half-open on the left: a sample exactly at End-window is excluded, one exactly at End included.

Only windows ending inside [r.Start, r.End] are returned, and each covers only the fetched data — so a caller wanting complete windows from its first evaluation point must fetch a window's worth of lead-in before it, as a PromQL engine does.

window ≤ 0 means window = step (disjoint windows).

func (*Engine) AggregateWindowNamed added in v0.36.0 

func (e *Engine) AggregateWindowNamed(
	ctx context.Context, r fetch.Request, spec WindowSpec,
) ([]NamedWindowAgg, error)

AggregateWindowNamed is Engine.AggregateWindow returning each series' identity alongside its windows, for a caller that must re-check matchers or render labels (the cluster aggregate RPC).

func (*Engine) Append 

func (e *Engine) Append(s signal.Series, ts int64, value float64) (bool, error)

Append ingests one sample for series s, logging to the WAL when durable. It returns whether the sample was accepted (false ⇒ rejected as out-of-order beyond the window).

func (*Engine) AppendBatch 

func (e *Engine) AppendBatch(
	ids []signal.SeriesID, ts []int64, values, sf []float64, materialize func(i int) signal.Series, limits AppendLimits,
) (AppendResult, error)

AppendBatch ingests a run of samples whose content ids are already computed (by the projection layer, on a reused buffer). ids[i], ts[i], values[i] describe sample i; materialize(i) returns sample i's full identity and is called only when its series is new (first sight), so a repeat series costs just a map probe and a buffer append, with no per-point signal.Series construction or hashing. The whole run is appended under a single lock. limits caps cardinality and in-flight memory (0 fields ⇒ unlimited). It returns an AppendResult breaking accepted/rejected down by reason, so the caller can report an exact OTLP partial-success. sf carries each sample's lossy-sampling weight (nil ⇒ every weight is 1); it is non-nil only when the caller's admission layer sampled the batch. Safe for concurrent use.

func (*Engine) ApplyPrimary added in v0.2.0 

func (e *Engine) ApplyPrimary(data []byte, limits AppendLimits) (accepted []byte, res AppendResult, err error)

ApplyPrimary applies a write as the shard's **primary**: it runs each sample through the admission-checked append path (the single OOO decision for the shard, plus the cardinality and in-flight-memory valves from limits) and re-frames the *accepted* samples into a WAL payload to replicate to the secondary owners. It returns that accepted payload and an AppendResult breaking the disposition down by reason, so the clustered ingest path can attribute OTLP partial-success exactly like the single-node path. Because only the primary admission-checks and it dictates the accepted set, every replica converges on the same data regardless of concurrent writers. Safe for concurrent use.

func (*Engine) ApplyReplicated 

func (e *Engine) ApplyReplicated(data []byte) error

ApplyReplicated applies a replicated write from the shard's primary to this secondary's head: it registers each series and appends its samples **verbatim** (no OOO re-check — the primary already decided the accepted set, the same way WAL Engine.Replay trusts the log), so all replicas hold identical data. A replica holds the unflushed head this way; after a flush the shared object store reconciles them. Safe for concurrent use.

func (*Engine) Cardinality added in v0.12.0 

func (e *Engine) Cardinality(topN int) CardinalityStat

Cardinality summarizes the engine's label cardinality from the head's inverted index (which spans every series ever seen, flushed or not). topN bounds the returned Top slice (≤0 returns all label names). It takes a read lock and does no backend I/O.

func (*Engine) Close 

func (e *Engine) Close(ctx context.Context) error

Close flushes any buffered samples to a part and closes the WAL. It does not stop a background loop — the owner ([storage.Storage]) does that before calling Close.

func (*Engine) CloseWAL added in v0.4.0 

func (e *Engine) CloseWAL() error

CloseWAL closes the engine's open WAL segment file handle without flushing the head or checkpointing — modeling a process crash, where the OS reclaims open descriptors but the on-disk WAL segments survive for replay. The head is left as-is (and lost, as a crash would lose it). A crash-recovery test uses this to release the file handle so the WAL directory can be removed even on platforms that refuse to delete a file held open by a live process (Windows). No-op without a WAL.

func (*Engine) Count added in v0.17.0 

func (e *Engine) Count(ctx context.Context, r fetch.Request) (int, error)

Count returns the number of series matching r.Matchers that have at least one sample in [r.Start, r.End]. It is the count-pushdown read path for PromQL `count(<selector>)`: it resolves the matched series and checks each for an in-window sample without materializing samples (no result batches, no value copies) or labels (no projection) — so a query that needs only a cardinality pays none of the per-series decode-to-result cost a full Fetch incurs.

Correctness mirrors Engine.Fetch: parts are decoded once (shared, pooled) and a series' in-window existence is found by binary search over its sorted timestamp run; head, recent, and mid-flush windows are scanned in memory. A series counts if any source holds an in-window sample. Series with no sample in the window are omitted.

func (*Engine) CountBy added in v0.25.0 

func (e *Engine) CountBy(ctx context.Context, r fetch.Request, label []byte) (map[string]int, error)

CountBy is the grouped variant of Engine.Count: it returns, for each distinct value of the given label among the matched series, how many of them have at least one sample in [r.Start, r.End]. It backs the PromQL `count by (label)(<selector>)` pushdown — and, via the result's length, `count(count by (label)(...))`, i.e. distinct-label-values — with the same zero-decode economics as Count: group membership comes from the snapshotted series identities (no label projection into results) and in-window existence from the part index / timestamp-only edge decode.

Grouping key: the label's canonical text (map key), read from the series identity over the same key space the postings index (and matchers) see — point attributes first, then the otel.scope.name/version synthetics, scope attributes, and resource attributes. Matched series without the label group under "" (indistinguishable from an explicit empty-text value, matching PromQL's absent-label grouping). Groups whose every series is empty in the window are omitted.

func (*Engine) DecodeCacheStats added in v0.6.0 

func (e *Engine) DecodeCacheStats() (DecodeCacheStats, bool)

DecodeCacheStats returns the engine's decoded-block cache statistics and whether a cache is configured (Config.DecodeCacheBytes > 0). For operator visibility into cache effectiveness.

func (*Engine) Fetch 

func (e *Engine) Fetch(ctx context.Context, r fetch.Request) (fetch.Iterator, error)

Fetch implements fetch.Fetcher over the head ∪ flushed parts: it resolves the request's matchers to series (the index spans every series ever seen, flushed or not) and yields one batch per series with its samples in the window, merged across the head buffer and every part by timestamp.

The iterator is **streaming**: a series' samples are gathered on the fetch.Iterator.Next that yields them, so a consumer that folds and releases each batch stays O(1) in matched series instead of O(matched series x samples). The acquired parts, the decode-memory reservation and the profiling/observability accounting therefore live for the whole iteration and are settled by fetch.Iterator.Close — a caller **must** Close (directly or via fetch.Drain), otherwise the parts stay pinned and the decode budget stays reserved.

func (*Engine) Flush 

func (e *Engine) Flush(ctx context.Context) error

Flush writes the head's buffered samples to a new immutable part and clears the buffers (the series index is retained). It is a no-op if the head holds no samples. Requires a Config.Backend.

func (*Engine) HeadBytes added in v0.4.0 

func (e *Engine) HeadBytes() int64

HeadBytes returns the head's current buffered sample bytes — the in-flight memory measure a consumer compares against a per-tenant cap (see AppendLimits.MaxInFlightBytes) and the basis for a size-triggered flush.

func (*Engine) HeadSampleCount 

func (e *Engine) HeadSampleCount() int

HeadSampleCount returns the number of samples currently buffered in the head (across all series) — for introspection and tests (e.g. to observe replica head trimming).

func (*Engine) IdentityBytes added in v0.37.0 

func (e *Engine) IdentityBytes() int64

IdentityBytes returns the resident bytes of the engine's identity state — the symbol table, the series index, the postings lists and the per-series out-of-order watermarks. It is reported separately from Engine.HeadBytes because a flush does not drain it: identities outlive their samples and are cleared only by Engine.Reset, so this number tracks the engine's all-time series count, not its buffered data.

func (*Engine) LabelNames added in v0.37.0 

func (e *Engine) LabelNames(ctx context.Context, r fetch.Request) ([]string, error)

LabelNames returns, sorted, the distinct attribute names carried by the series matching r.Matchers that hold samples in [r.Start, r.End]. LabelValues is its per-name twin, returning the distinct canonical-text values of one name.

Both answer from the **inverted index**, not from series: with no matchers the walk is over the postings' (name → values) map, so an unmatched label query — the shape a dashboard's template variable issues — costs O(distinct values), not O(series). That is the difference between reading a few thousand symbols and materializing a million identities to project and deduplicate them. With matchers it narrows to the matched ids and reads only the requested name off each identity (never a whole label set).

Liveness is what keeps the index-driven answer honest: the head's series index is **all-time** (it outlives flushes, and retention prunes samples and parts, never identities), so a value is emitted only once some series carrying it is found live — an in-window in-memory sample, or membership in a part overlapping the window. Like Engine.Series the part test is overlap-granular, and the probe stops at the first live series, so a live value costs one probe rather than a scan of its postings list.

func (*Engine) LabelValues added in v0.37.0 

func (e *Engine) LabelValues(ctx context.Context, r fetch.Request, name []byte) ([]string, error)

LabelValues returns, sorted, the distinct canonical-text values of name across the series matching r.Matchers that hold samples in [r.Start, r.End]. See Engine.LabelNames for the shared contract.

func (*Engine) LoadParts 

func (e *Engine) LoadParts(ctx context.Context) error

LoadParts reconstructs the engine's durable state from the object store: the part set from the bucket index, and the series identity index (postings + labels) from the persisted identity object. It is how a fresh engine over an existing prefix serves reads with no in-memory state carried over from the writer (the stateless read path); typically called once after New during recovery. WAL Engine.Replay is complementary — it restores the unflushed head samples — but is not required to query flushed data.

It replaces any current parts. A head-only engine (no backend) is a no-op. It assumes this node owns the prefix — it sweeps the part objects the index does not name (see [Engine.sweepOrphansLocked]).

func (*Engine) Merge 

func (e *Engine) Merge(ctx context.Context, retainFrom int64) error

Merge compacts every flushed part into a single new part, dropping samples older than retainFrom (retention; retainFrom ≤ 0 disables it). It is a no-op when there is nothing to gain — fewer than two parts and no retention cutoff. Source parts are deleted from the backend after the new part is durably written.

Retention is expressed as an absolute timestamp (unix nanoseconds), so the engine stays free of wall-clock dependencies; the caller derives it from the tenant policy. For downsampling, use Engine.MergeWith.

func (*Engine) MergeBacklog added in v0.12.0 

func (e *Engine) MergeBacklog() int

MergeBacklog returns the parts a merge may still take — the flushed parts less the sealed ones, which no merge will reconsider. It is MergeShape.Backlog; use Engine.MergeShape for the rest of the selector's inputs.

func (*Engine) MergeRunning added in v0.12.0 

func (e *Engine) MergeRunning() bool

MergeRunning reports whether a merge/compaction is currently executing on this engine (an in-memory liveness flag for introspection).

func (*Engine) MergeShape added in v0.37.0 

func (e *Engine) MergeShape() MergeShape

MergeShape returns the selector's view of the engine's parts. It takes a brief read lock, does no backend I/O and decodes nothing, so it is safe to poll at dashboard cadence.

func (*Engine) MergeWith added in v0.4.0 

func (e *Engine) MergeWith(ctx context.Context, opts MergeOptions) error

MergeWith compacts every flushed part into a single new part, applying retention and downsampling per opts. It is the one background-merge entry point; compaction, retention, and downsampling are the same pass over the immutable parts (no separate subsystem).

func (*Engine) PartCount 

func (e *Engine) PartCount() int

PartCount returns the number of flushed parts (testing/introspection).

func (*Engine) Parts added in v0.12.0 

func (e *Engine) Parts() []PartStat

Parts returns an in-memory snapshot of the parts the engine can serve — its own, plus the ones it adopted from a rival writer's index — under a read lock, with no backend I/O and no decode, so it is safe to poll. It is the servable set rather than this writer's own, because that is what a read answers from and what the completeness accounting has to measure. For byte sizes, codecs, and chunk counts, use Engine.PartsDetailed.

func (*Engine) PartsDetailed added in v0.12.0 

func (e *Engine) PartsDetailed(ctx context.Context) ([]PartDetailStat, error)

PartsDetailed augments Engine.Parts with each part's on-backend byte size, column/codec layout, and chunk (granule) count. It reads from the backend (object sizes), so unlike Parts it is not hot-path-free — call it for a drill-down view, not a high-frequency poll. Each part is ref-held for the duration so a concurrent merge cannot reclaim its objects mid-read.

func (*Engine) PruneIdentities added in v0.37.0 

func (e *Engine) PruneIdentities(ctx context.Context) (int, error)

PruneIdentities drops the identities no live data names any more — those the retention side of a merge left behind — rebuilding the resident index around the survivors and shrinking the durable identity object to match. It returns the number of identities removed (0 when nothing could have died, when too few have, or when the engine holds no parts).

Every node may call it, owner or replica: identity is scoped to the part that holds it, so the live set is derived from *this node's* parts and means exactly "what this node can still serve". A part a replica has not yet synced brings its identities with it when it arrives, so pruning ahead of a sync loses nothing. It takes the same flush/merge exclusion as those paths, so no publish can add an identity underneath it.

func (*Engine) PruneIdentitiesWith added in v0.37.0 

func (e *Engine) PruneIdentitiesWith(ctx context.Context, opts PruneOptions) (int, error)

PruneIdentitiesWith is Engine.PruneIdentities with explicit options.

func (*Engine) RefreshReplica 

func (e *Engine) RefreshReplica(ctx context.Context) error

RefreshReplica brings a replica node's view up to date with the shared object store: it reconstructs the flushed parts from the bucket index and trims its head to the still-unflushed window — samples a primary has already flushed (covered by a part) are dropped, bounding replica memory. With no shared store (this node cannot see the parts), it is a safe no-op: nothing loads, so nothing is trimmed.

func (*Engine) Replay 

func (e *Engine) Replay(dir string) error

Replay rebuilds the head from the WAL segments in dir (durable restart). It skips segments at or below the flush watermark recovered by Engine.LoadParts (call LoadParts first), so records already in a flushed part are not re-applied — exactly-once recovery.

func (*Engine) Reset 

func (e *Engine) Reset(ctx context.Context) error

Reset discards all of the engine's data — the in-memory head (samples + series index) and every flushed part — returning it to the empty state of a freshly New'd engine, without reallocating the engine itself. It waits for an in-flight flush or merge to finish first, so that operation cannot publish its part into the reset engine. Flushed part objects are deleted from the backend so none are orphaned — except those a concurrent fetch is still reading, which are retired and deleted by the deferred reclaim once the reader drains. It is destructive (it wipes this engine's parts under Config.Prefix) and is meant for the ephemeral in-memory engine in tests and benchmarks, letting a long-lived engine be reused across runs. Safe for concurrent use.

func (*Engine) Series added in v0.37.0 

func (e *Engine) Series(ctx context.Context, r fetch.Request) ([]signal.Series, error)

Series returns the identities of the series matching r.Matchers that hold samples in [r.Start, r.End], without reading a single value or timestamp. It is the metrics twin of recordengine's stream enumeration and backs the label endpoints (/api/v1/labels, /api/v1/label/<name>/values, /series): those need identities only, so paying a full fetch — decoding and copying every sample of every matching series just to read b.Series — costs (cardinality x window) for a list of strings.

The read is **series-only**: matched ids come from the head's postings index (which outlives a flush), and each in-window part contributes the matched ids its series index holds. No sample column is touched, so the cost is proportional to matched cardinality alone — not to the window's depth. Consequently the time filter is **part-overlap granular**, exactly as [recordengine.Engine.Series] and Prometheus' own block-granular label endpoints are: a returned series is guaranteed to match the matchers and to live in a part overlapping the window (or to have an in-memory sample inside it, which is checked exactly), but a series whose samples sit just outside the window in an overlapping part may still be listed. Use Engine.Count where the "has >= 1 sample in the window" test must be exact — it pays a timestamp decode at the window edges for it.

The returned identities alias engine-owned, immutable interned memory; copy them to retain past an engine Reset.

func (*Engine) SeriesCount 

func (e *Engine) SeriesCount() int

SeriesCount returns the number of distinct series in the head.

func (*Engine) Stats added in v0.10.0 

func (e *Engine) Stats() Stats

Stats returns an in-memory snapshot of the engine's state under a single read lock. It does no backend I/O and decodes nothing, so it is safe to poll at dashboard cadence without touching the hot path. Part byte sizes are not included (they would require backend stat calls).

func (*Engine) SyncWAL added in v0.3.0 

func (e *Engine) SyncWAL() error

SyncWAL fsyncs the engine's WAL, if any (the background WALSyncInterval path). No-op without a WAL.

func (*Engine) WALState added in v0.12.0 

func (e *Engine) WALState() (segments int, bytes int64, epoch uint64, ok bool)

WALState returns the current WAL segment count, the open segment's byte size, and the active flush epoch. ok is false when the engine has no WAL (the ephemeral in-memory engine). It takes a read lock, excluding concurrent appends (which hold the write lock).

type LabelCard added in v0.12.0 

type LabelCard struct {
	Name           string
	Series         int64
	DistinctValues int
}

LabelCard is one label name's cardinality: how many series carry it and how many distinct values it takes across them.

type MergeOptions added in v0.4.0 

type MergeOptions struct {
	// RetainFrom drops samples with a timestamp < RetainFrom before the merged part is
	// written (retention). ≤ 0 disables it. It is an absolute unix-nanosecond cutoff so the
	// engine stays free of wall-clock dependencies; the caller derives it from tenant policy.
	RetainFrom int64
	// Downsample, when non-empty, rolls up old samples at merge time (coarsening resolution
	// with age). It reuses the one merge engine — no separate subsystem. The cutoffs are
	// absolute (the caller resolves now − After into Before), keeping the merge deterministic.
	Downsample []DownsampleTier
	// Recompress, when non-nil, rewrites a fully-cold merged part (every sample older than its
	// Before cutoff) with a higher-ratio compression profile — the fourth merge mode after
	// compaction, retention, and downsampling, still one pass over the parts. nil keeps the
	// default (codec-only) compression.
	Recompress *RecompressSpec
	// Precision, when non-empty, re-encodes a cold part's value column lossily at merge — fewer
	// significant mantissa bits for older data (age-tiered) — the fifth merge mode, still one
	// pass. The cutoffs are absolute (the caller resolves now − After into Before). Empty keeps
	// every part lossless.
	Precision []PrecisionTier
	// Force takes the best run of unsealed parts even when it does not earn its rewrite, instead of
	// selecting nothing — the operator escape from a fixed point where every run scores below
	// [minMergeMultiplier] and the engine would sit on its part count until the idle waiver fires.
	// It bypasses the selection heuristic only: the seal threshold and the run's cumulative-bytes
	// cap still bound what one merge reads, writes, and holds.
	Force bool
}

MergeOptions parameterizes a merge. The zero value is a plain compaction (no retention, no downsampling) — the same effect as the historical Merge(ctx, 0).

type MergeShape added in v0.37.0 

type MergeShape struct {
	// Parts is the flushed parts; Sealed those too large to be a useful merge input, which no merge
	// will reconsider; Backlog the rest — the parts a merge may still take.
	Parts   int
	Sealed  int
	Backlog int
	// Candidates is how many parts the next size-driven merge would select right now. 0 with a
	// non-zero Backlog is the stuck state: parts remain mergeable but no run of them qualifies.
	Candidates int
	// CapBytes is the seal threshold in effect, in bytes on disk. It is derived per merge from free
	// space and the merge memory allowance, so it is reported as of the last merge — 0 before the
	// first one, and 0 when sealing is disabled.
	CapBytes int64
	// BestMultiplier is the best output-to-largest-input ratio any eligible run reaches;
	// MinMultiplier the ratio a run must reach to be selected on its own merits.
	BestMultiplier float64
	MinMultiplier  float64
	// IdleRounds is the consecutive merges that selected nothing; after WaiveAfter of them the
	// selector takes its best run regardless of the ratio.
	IdleRounds int
	WaiveAfter int
}

MergeShape is the merge selector's view of the flushed parts: the inputs to the decision the background merge makes each cycle. Without them an engine sitting on a part count it will never reduce is indistinguishable from an idle healthy one — the two differ only in whether the parts are sealed and whether any run of the rest is worth rewriting.

type NamedAgg added in v0.8.0 

type NamedAgg struct {
	Series  signal.Series
	Buckets []BucketAgg
}

NamedAgg pairs a series' identity with its step buckets — the cluster-facing aggregate result, carrying the labels a coordinator re-checks the full matcher set against before unioning shards.

type NamedWindowAgg added in v0.36.0 

type NamedWindowAgg struct {
	Series  signal.Series
	Windows []WindowAgg
}

NamedWindowAgg pairs a series' identity with its evaluation windows — the labeled form of Engine.AggregateWindow, for a caller that renders the result as a PromQL range vector.

type PartDetailStat added in v0.12.0 

type PartDetailStat struct {
	PartStat

	Bytes   int64        // sum of the part's backend object sizes
	Chunks  int          // sparse-index granules: ceil(RowCount / GranuleSize)
	Columns []ColumnStat // per-column physical layout
}

PartDetailStat augments PartStat with fields that need a backend read: the on-backend byte size (summed over the part's objects) and the column/codec layout and chunk count from the manifest (the manifest is already cached on the open part, so only Bytes incurs additional I/O).

type PartStat added in v0.12.0 

type PartStat struct {
	ID      string // the part's backend key prefix
	MinTime int64  // inclusive unix-ns bounds of the part's samples
	MaxTime int64
	Series  int   // distinct series in the part (len of its row-range index)
	Rows    int64 // total samples (sum of the per-series row spans)
	// SizeBytes is the part's recorded on-disk size — what the merge cap compares against, and so
	// what explains why a part is or is not sealed. It needs no I/O and counts the column and marks
	// objects only; [PartDetailStat.Bytes] measures every object at the cost of a backend read.
	SizeBytes int64
}

PartStat is one flushed part's in-memory shape (no backend I/O, no decode): identity, time bounds, and the series/row counts derivable from the part's in-memory row-range index.

type PrecisionTier added in v0.6.0 

type PrecisionTier struct {
	Before int64 // a part whose maxTime < Before is subject to this tier
	Bits   uint8 // significant mantissa bits to retain (1..63); 0 or ≥64 ⇒ lossless, ignored
}

PrecisionTier is the absolute (wall-clock-free) form of a tenant float-precision policy: a part whose newest sample is older than Before is re-encoded, at merge, retaining only Bits significant mantissa bits in its value column (scaled-decimal, lossy). Fewer bits ⇒ denser, less accurate. It mirrors RecompressSpec (a per-part, age-gated cold rewrite) but as a list of coarsening tiers, so only old data trades accuracy for size. The caller ([storage.Storage]) builds these from [tenant.PrecisionTier] and the current time.

type PruneOptions added in v0.37.0 

type PruneOptions struct {
	// Force runs the prune even when the background thresholds would skip it — an engine that has
	// merged nothing away since the last pass, or one whose dead set is too small to pay for the
	// rebuild. It is what an operator-triggered sweep wants: "prune now", with the count it removed
	// as the answer, rather than a silent no-op whose reason is a threshold.
	Force bool
}

PruneOptions tunes an identity prune.

type RecompressSpec added in v0.4.0 

type RecompressSpec struct {
	Before    int64 // a part whose maxTime < Before is cold and is rewritten with the profile below
	Algorithm compress.Algorithm
	Level     compress.Level
}

RecompressSpec is the absolute (wall-clock-free) form of a tenant recompression policy: a merged part whose newest sample is older than Before (it is fully cold) is written with Algorithm at Level instead of the ladder level its size would select. The level is decode-irrelevant — the reader reconstructs the decompressor from the per-column algorithm recorded in the manifest — so this is a pure ratio/CPU trade-off with no format change. The caller ([storage.Storage]) builds it from [tenant.Recompress] and the current time.

type SeriesAgg added in v0.6.0 

type SeriesAgg struct {
	Count    int64
	Sum      float64
	Min, Max float64
}

SeriesAgg is a per-series aggregate over a value window — enough to answer count, sum, min, max, and avg (Sum/Count) without the raw samples. It is the unit of the aggregate-pushdown fast path: a part precomputes one per series at write time (the stats sidecar) so a query whose range fully covers the part folds these instead of decoding its value column.

type Stats added in v0.10.0 

type Stats struct {
	Series      int64 // distinct series ever seen (index span: head ∪ flushed)
	HeadSamples int64 // samples currently buffered in the head (unflushed)
	HeadBytes   int64 // head's buffered sample bytes (the in-flight memory measure)
	// IdentityBytes is the resident identity state (symbols + series index + postings + OOO
	// watermarks) — memory a flush does not drain, and which no other counter here reports.
	IdentityBytes int64
	Parts         int   // flushed immutable parts
	MinTime       int64 // oldest flushed sample time (unix ns); 0 when no parts
	MaxTime       int64 // newest sample time across parts and the head (unix ns); 0 when empty
	// OutOfSpace is set while the engine refuses writes because its backend is out of bytes or
	// inodes. Reads still answer from what is on disk; it clears when a flush finds room again.
	OutOfSpace bool
}

Stats is an in-memory snapshot of an engine's state for introspection (no backend I/O, no decode).

type WindowAgg added in v0.36.0 

type WindowAgg struct {
	SeriesAgg

	End int64
}

WindowAgg is one evaluation step's range-vector aggregate: the count/sum/min/max of the samples in the half-open window (End-window, End], keyed by the evaluation timestamp End. Unlike BucketAgg, consecutive windows overlap whenever the window is wider than the step, so a sample contributes to window/step of them.

type WindowSpec added in v0.36.0 

type WindowSpec struct {
	// Step is the distance between evaluation timestamps. Must be > 0.
	Step int64

	// Window is the width of each evaluation window, (t-Window, t]. A value ≤ 0 means Step: one
	// aggregate per step over disjoint windows, no overlap.
	Window int64

	// Anchor is any timestamp on the evaluation grid — windows end at Anchor + k*Step. The zero
	// value anchors on the absolute grid (multiples of Step). A PromQL range query's grid is
	// anchored at the query's start, which is only a multiple of the step by coincidence, so an
	// embedder must pass it: the answer is otherwise computed at timestamps nobody asked about.
	Anchor int64
}

WindowSpec is the evaluation grid of an overlapping range-vector aggregate.

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.