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
- type AppendLimits
- type AppendResult
- type Config
- type DownsampleTier
- type Engine
- func (e *Engine) Append(s signal.Series, ts int64, value float64) (bool, error)
- func (e *Engine) AppendBatch(ids []signal.SeriesID, ts []int64, values, sf []float64, ...) (AppendResult, error)
- func (e *Engine) ApplyPrimary(data []byte) (accepted []byte, rejected int, err error)
- func (e *Engine) ApplyReplicated(data []byte) error
- func (e *Engine) Close(ctx context.Context) error
- func (e *Engine) CloseWAL() error
- func (e *Engine) Fetch(ctx context.Context, r fetch.Request) (fetch.Iterator, error)
- func (e *Engine) Flush(ctx context.Context) error
- func (e *Engine) HeadBytes() int64
- func (e *Engine) HeadSampleCount() int
- func (e *Engine) LoadParts(ctx context.Context) error
- func (e *Engine) Merge(ctx context.Context, retainFrom int64) error
- func (e *Engine) MergeWith(ctx context.Context, opts MergeOptions) error
- func (e *Engine) PartCount() int
- func (e *Engine) RefreshReplica(ctx context.Context) error
- func (e *Engine) Replay(dir string) error
- func (e *Engine) Reset(ctx context.Context) error
- func (e *Engine) SeriesCount() int
- func (e *Engine) SyncWAL() error
- type MergeOptions
- type RecompressSpec
Constants ¶
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
// 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
}
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 Config ¶
type Config struct {
// OOOWindow rejects samples older than newest-OOOWindow (nanoseconds). 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
}
Config configures an Engine.
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 (*Engine) Append ¶
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
ApplyPrimary applies a write as the shard's **primary**: it appends each sample through the out-of-order-checked path (the single OOO decision for the shard) and re-frames the *accepted* samples into a WAL payload to replicate to the secondary owners. It returns that accepted payload and the number of samples rejected as out-of-order. Because only the primary OOO-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 ¶
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) Close ¶
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
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) Fetch ¶
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 returns one batch per series with its samples in the window, merged across the head buffer and every part by timestamp.
func (*Engine) Flush ¶
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
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 ¶
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) LoadParts ¶
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 and advances the part sequence past the highest existing part. A head-only engine (no backend) is a no-op.
func (*Engine) Merge ¶
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) 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) RefreshReplica ¶
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) Reset ¶
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. Flushed part objects are deleted from the backend so none are orphaned. 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) SeriesCount ¶
SeriesCount returns the number of distinct series in the head.
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
}
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 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 default codec-only framing. 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.
Source Files