Documentation
¶
Index ¶
- Constants
- Variables
- func DecodeTimestamp(key []byte, dims uint8) (time.Time, error)
- func DecodeValue(val []byte) (nums []float64, payload []byte, err error)
- func EncodeKey(tid TimelineID, dims uint8, dv []uint64, ts time.Time) ([]byte, error)
- func EncodePrefixKey(tid TimelineID, dv []uint64) []byte
- func EncodeValue(nums []float64, payload []byte) ([]byte, error)
- func KeySize(dims uint8) int
- type Adapter
- func (a *Adapter) Execute(ctx context.Context, store dxp.ParticipantStore, c dxp.Claim) (dxp.Result, error)
- func (a *Adapter) PostCommit(ctx context.Context, c dxp.Claim) error
- func (a *Adapter) Release(ctx context.Context, c dxp.Claim) error
- func (a *Adapter) Reserve(ctx context.Context, tenant string, op dxp.OpParams, txn, participantID string, ...) (dxp.Claim, error)
- func (a *Adapter) Validate(ctx context.Context, c dxp.Claim) error
- type AggregateQuery
- type AppendParams
- type Bucket
- type DefaultManager
- type Event
- type LatestQuery
- type Manager
- type PebbleConfig
- type PebbleDBProvider
- type PebbleStore
- func (s *PebbleStore) Aggregate(ctx context.Context, q AggregateQuery) ([]Bucket, error)
- func (s *PebbleStore) Append(ctx context.Context, e Event) error
- func (s *PebbleStore) AppendBatch(ctx context.Context, events []Event, maxBatch int) (int, error)
- func (s *PebbleStore) Close() error
- func (s *PebbleStore) DefaultRetentionDays() int
- func (s *PebbleStore) DefineRollup(sourceTID TimelineID, def RollupDef) (RollupID, error)
- func (s *PebbleStore) DefineSystemTimeline(id TimelineID, cfg TimelineConfig) error
- func (s *PebbleStore) DefineTimeline(id TimelineID, cfg TimelineConfig) error
- func (s *PebbleStore) Delete(ctx context.Context, e Event) error
- func (s *PebbleStore) DeleteKeys(ctx context.Context, keys [][]byte) error
- func (s *PebbleStore) DeleteRollup(sourceTID TimelineID, id RollupID) error
- func (s *PebbleStore) DeleteTimeline(ctx context.Context, id TimelineID) error
- func (s *PebbleStore) DeleteTimelineData(ctx context.Context, id TimelineID) error
- func (s *PebbleStore) GetRollup(sourceTID TimelineID, id RollupID) (RollupDef, error)
- func (s *PebbleStore) Latest(ctx context.Context, q LatestQuery) ([]Event, error)
- func (s *PebbleStore) ListRollups(sourceTID TimelineID) ([]RollupDef, error)
- func (s *PebbleStore) PebbleDB() *pebble.DB
- func (s *PebbleStore) Purge(ctx context.Context) error
- func (s *PebbleStore) PurgeTimelineRange(ctx context.Context, id TimelineID, from, to time.Time) error
- func (s *PebbleStore) QueryRange(ctx context.Context, q RangeQuery) ([]Event, error)
- func (s *PebbleStore) RangeAggregate(ctx context.Context, q RangeAllQuery) (*RangeAggregateResult, error)
- func (s *PebbleStore) RangeAvg(ctx context.Context, q RangeNumQuery) (float64, error)
- func (s *PebbleStore) RangeCount(ctx context.Context, q RangeNumQuery) (uint64, error)
- func (s *PebbleStore) RangeFullAggregate(ctx context.Context, q RangeFullQuery) (*RangeFullResult, error)
- func (s *PebbleStore) RangeMax(ctx context.Context, q RangeNumQuery) (float64, error)
- func (s *PebbleStore) RangeMedian(ctx context.Context, q RangeNumQuery) (float64, error)
- func (s *PebbleStore) RangeMin(ctx context.Context, q RangeNumQuery) (float64, error)
- func (s *PebbleStore) RangeQuantile(ctx context.Context, q RangeNumQuery, quantile float64) (float64, error)
- func (s *PebbleStore) RangeSum(ctx context.Context, q RangeNumQuery) (float64, error)
- func (s *PebbleStore) RollupParent(tid TimelineID) (RollupDef, bool)
- func (s *PebbleStore) RollupStatus(sourceTID TimelineID, id RollupID) (RollupStatusReport, error)
- func (s *PebbleStore) RollupTree() *RollupTreeNode
- func (s *PebbleStore) RunRollup(ctx context.Context, sourceTID TimelineID, id RollupID, from, to time.Time, ...) error
- func (s *PebbleStore) SetDefaultRetentionDays(days int) error
- func (s *PebbleStore) SetWriteConfig(id TimelineID, cfg TimelineWriteConfig) error
- func (s *PebbleStore) Stats(_ context.Context) (*StoreStats, error)
- func (s *PebbleStore) SysmaskWidth() SysmaskWidth
- func (s *PebbleStore) Timeline(id TimelineID) (TimelineConfig, bool)
- func (s *PebbleStore) TimelineStats(ctx context.Context, id TimelineID) (*TimelineStats, error)
- func (s *PebbleStore) Timelines() []TimelineID
- func (s *PebbleStore) UpdateTimeline(id TimelineID, cfg TimelineConfig) error
- func (s *PebbleStore) WriteConfig(id TimelineID) TimelineWriteConfig
- type RangeAggregateResult
- type RangeAllQuery
- type RangeFullQuery
- type RangeFullResult
- type RangeNumQuery
- type RangeQuery
- type RetentionWorker
- type RollupDef
- type RollupID
- type RollupStatusReport
- type RollupTreeNode
- type Store
- type StoreConfig
- type StoreFactory
- type StoreStats
- type SysmaskWidth
- type TimelineConfig
- type TimelineID
- type TimelineStats
- type TimelineWriteConfig
Constants ¶
const ( MinDims = 1 MaxDims = 5 )
MinDims and MaxDims bound the number of dimensions a timeline may declare.
const MaxRollupDepth = 4
MaxRollupDepth is the maximum allowed depth of the rollup tree (not counting timeline 0 as the root). Controlled at runtime by dynconfig key ts.rollup_max_depth; this is the compiled-in default.
Variables ¶
var ErrBucketLimitExceeded = fmt.Errorf("ts: aggregate bucket limit exceeded (%s)", xoluerr.ErrTSBucketLimit)
ErrBucketLimitExceeded is returned when an aggregate query would produce more buckets than the configured MaxBuckets limit.
var ErrDeleteNotSupported = fmt.Errorf("ts: delete not supported by this backend")
ErrDeleteNotSupported is returned by Store implementations that do not support event deletion. Callers must not assume the targeted event was removed when this error is received.
var ErrScanLimitExceeded = fmt.Errorf("ts: scan limit exceeded (XOLU-TS013)")
ErrScanLimitExceeded is returned when a query exceeds its MaxScanEvents budget.
Functions ¶
func DecodeTimestamp ¶
DecodeTimestamp extracts the timestamp from a key without decoding dimensions. Faster than DecodeKey when only the timestamp is needed (e.g. in Purge).
func DecodeValue ¶
DecodeValue decodes a value encoded by EncodeValue.
func EncodeKey ¶
EncodeKey encodes a Pebble key for the given timeline, dimension values, and timestamp. dims must be the timeline's declared dimension count (1–5); len(dv) must equal dims.
func EncodePrefixKey ¶
func EncodePrefixKey(tid TimelineID, dv []uint64) []byte
EncodePrefixKey encodes a key prefix for range scanning using a leading dimension slice (1 ≤ len(dv) ≤ dims). The timestamp is not included.
func EncodeValue ¶
EncodeValue encodes nums and payload into a compact binary value. len(nums) must be 0–7. NaN values are rejected.
Types ¶
type Adapter ¶ added in v0.26.0
type Adapter struct {
// contains filtered or unexported fields
}
Adapter is ts's dxp.Participant. One Adapter per PebbleStore; safe for concurrent use.
func NewAdapter ¶ added in v0.26.0
func NewAdapter(store *PebbleStore, cache *dxp.MemCache) *Adapter
NewAdapter returns an Adapter ready to register with a dxp coordinator under the primitive key "ts".
func (*Adapter) Execute ¶ added in v0.26.0
func (a *Adapter) Execute(ctx context.Context, store dxp.ParticipantStore, c dxp.Claim) (dxp.Result, error)
Execute writes the event directly into the coordinator-supplied batch — EncodeKey/EncodeValue + Set, the same pair Append's own non-coalesced path uses on s.db, just targeting store.Batch instead. Deliberately bypasses Append/the coalescer entirely (see this file's own doc for why that's safe here).
func (*Adapter) PostCommit ¶ added in v0.26.0
PostCommit is a no-op — ts's own event write IS the primary store, not a guard-plane write with a separate derived/advisory index fed afterward (unlike cal's H1/H3 split). Implemented so Adapter satisfies dxp.Participant without a second interface-change ripple later.
func (*Adapter) Release ¶ added in v0.26.0
Release drops txn's stashed params, if any. Idempotent and unconditional, matching every other adapter's own error taxonomy (dxp-reservation-cache.md §16): releasing an unknown or already-cleared txn is a no-op, never an error.
func (*Adapter) Reserve ¶ added in v0.26.0
func (a *Adapter) Reserve(ctx context.Context, tenant string, op dxp.OpParams, txn, participantID string, deadline int64, w dxp.Weight) (dxp.Claim, error)
Reserve validates the event (validateEvent, pure) and checks the timeline is defined (the one real admission check — see this file's own doc). On consent it Holds a trivial claim (Amount: 1 — a slot-shaped resource per Claim's own doc, not a conserved one like bal's) and stashes the params for Execute.
func (*Adapter) Validate ¶ added in v0.26.0
Validate re-checks the timeline still exists — the only fact Reserve established that could have changed (a delete between Reserve and Validate). No balance/occupancy-shaped invariant exists here to re-verify, unlike bal/cal — append is independent of every other event on the timeline, live or not.
type AggregateQuery ¶
type AggregateQuery struct {
Timeline TimelineID
Dims []uint64
From time.Time
To time.Time
NumField uint8 // index into Nums (0-based, max 6)
Function string // "avg", "min", "max", "sum", "count"
Interval time.Duration // 0 = scalar result; > 0 = time-bucketed
MaxScanEvents int // 0 = no scan limit
MaxBuckets int // 0 = no bucket limit; > 0 aborts when exceeded (XOLU-TS019)
}
AggregateQuery computes an aggregate over a numeric field for all events matching the dimension prefix and time range.
type AppendParams ¶ added in v0.26.0
type AppendParams struct {
Timeline TimelineID `json:"timeline"`
Dims []uint64 `json:"dims,omitempty"`
TimeUnixNs int64 `json:"time_unix_ns"`
Nums []float64 `json:"nums,omitempty"`
Payload []byte `json:"payload,omitempty"`
}
AppendParams is ts's dxp.OpParams — one event to append.
func (AppendParams) Primitive ¶ added in v0.26.0
func (AppendParams) Primitive() string
Primitive satisfies dxp.OpParams.
type DefaultManager ¶
type DefaultManager struct {
// contains filtered or unexported fields
}
DefaultManager manages per-tenant Store lifecycle. It is backend-agnostic — the StoreFactory controls which engine is used.
func NewManager ¶
func NewManager(baseDir string, factory StoreFactory, cfg StoreConfig) (*DefaultManager, error)
NewManager creates a timeseries manager rooted at the data root (baseDir). Each tenant's timeseries store lives at <baseDir>/tXXXX/ts (tenant-first, derived by pkg/storelayout). NewManager scans the data root for existing tenant directories that already contain a ts/ subdirectory and registers them as provisioned (lazy-open on first request). Tenant names for pre-existing stores are not known at scan time; they are recorded on the first StoreFor call that provides a name via Provision.
func (*DefaultManager) Close ¶
func (m *DefaultManager) Close() error
Close shuts down all open stores.
func (*DefaultManager) IsProvisioned ¶
func (m *DefaultManager) IsProvisioned(tenantID tenant.TenantID) bool
IsProvisioned reports whether a tenant has timeseries storage.
type Event ¶
type Event struct {
Timeline TimelineID
Dims []uint64 // len must equal timeline's Dims
Time time.Time
Nums []float64 // optional, up to 7; nil means no numeric fields
Payload []byte // optional, caller-defined opaque bytes
}
Event is a single timeseries record written to or read from a timeline.
type LatestQuery ¶
type LatestQuery struct {
Timeline TimelineID
Dims []uint64
N int // default 10, max 10000
From time.Time // optional lower bound (zero = unbounded)
To time.Time // optional upper bound (zero = unbounded)
}
LatestQuery retrieves the N most recent events matching a dimension prefix.
Dims may be a leading prefix of the timeline's declared dimension count; all events matching that prefix are considered, across all remaining dimension values. This is intentional and useful for "latest across all sub-dimensions" queries.
From and To are optional time bounds. When non-zero, only events within [From, To] are returned. This is applied as a Go-side filter, consistent with the partial-prefix time filter in QueryRange.
type Manager ¶
type Manager interface {
// Provision creates a timeseries store for a tenant.
// tenantName is used to scope dynconfig lookups.
Provision(ctx context.Context, tenantID tenant.TenantID, tenantName string) error
// StoreFor returns the Store for a tenant, or an error if not provisioned.
StoreFor(tenantID tenant.TenantID) (Store, error)
// IsProvisioned reports whether a tenant has timeseries storage.
IsProvisioned(tenantID tenant.TenantID) bool
// Close shuts down all stores.
Close() error
}
Manager manages per-tenant Store lifecycle.
type PebbleConfig ¶
type PebbleConfig struct {
MemtableSize int // bytes; default 67108864 (64 MB)
BlockSize int // bytes; default 32768 (32 KB)
Compression string // "snappy", "zstd", or "none"; default "zstd"
L0CompactionThreshold int // L0 files before compaction; default 4
MaxOpenFiles int // per-store file descriptor limit; default 500
// Write coalescer tuning. Zero values fall back to package-level defaults
// (10ms flush interval, 2000 max events). Only relevant when the coalescer
// is enabled via dynconfig key ts.writecoal for the tenant or globally.
CoalFlushIntervalMs int // flush window in milliseconds; default 10
CoalMaxEvents int // early-flush threshold in events; default 2000
}
PebbleConfig holds LSM-tree tuning parameters specific to the Pebble storage engine. It is consumed only by NewPebbleStore / NewPebbleStoreFactory and has no meaning to other backends.
Zero values are safe: NewPebbleStore applies sensible defaults for any field that is ≤ 0 or empty.
type PebbleDBProvider ¶ added in v0.26.0
PebbleDBProvider is the Pebble-side analog of storage.WriterDBProvider — gives access to the underlying raw *pebble.DB. Used exactly once: by the dxp coordinator's phased execution path (pkg/server/v2_dxp_dispatch.go), which needs to construct one independent *pebble.Batch per Pebble-backed participant, the same way it constructs one independent *sql.Tx per SQL-backed one.
type PebbleStore ¶
type PebbleStore struct {
// contains filtered or unexported fields
}
PebbleStore implements Store backed by a single Pebble instance per tenant.
func (*PebbleStore) Aggregate ¶
func (s *PebbleStore) Aggregate(ctx context.Context, q AggregateQuery) ([]Bucket, error)
func (*PebbleStore) AppendBatch ¶
func (*PebbleStore) Close ¶
func (s *PebbleStore) Close() error
func (*PebbleStore) DefaultRetentionDays ¶
func (s *PebbleStore) DefaultRetentionDays() int
DefaultRetentionDays returns the store-level default retention in days.
func (*PebbleStore) DefineRollup ¶
func (s *PebbleStore) DefineRollup(sourceTID TimelineID, def RollupDef) (RollupID, error)
func (*PebbleStore) DefineSystemTimeline ¶ added in v0.16.3
func (s *PebbleStore) DefineSystemTimeline(id TimelineID, cfg TimelineConfig) error
defineSystemTimeline is the system-internal timeline definition path (@S §8). It draws from the system region and is the only way to mint a system id. It is deliberately unexported and not wired to the tenant HTTP surface — platform components call it directly. It refuses ids that are NOT system-scope (the symmetric guard), so the two paths partition the id space between them with no overlap.
func (*PebbleStore) DefineTimeline ¶
func (s *PebbleStore) DefineTimeline(id TimelineID, cfg TimelineConfig) error
DefineTimeline is the user-facing timeline definition path (@S §8). It refuses any id in the system region: under the store's sysmask width, an id whose selector bits are non-zero is system-scope and cannot be minted by a user request. With the default width 0 no id is system-scope, so this guard is a no-op until an operator opts in.
System timelines are created via DefineSystemTimeline, which is not reachable through the tenant HTTP surface.
func (*PebbleStore) Delete ¶
func (s *PebbleStore) Delete(ctx context.Context, e Event) error
Delete removes the event identified by e from the store. The key is encoded from (Timeline, Dims, Time); if no event exists at that key the delete is a no-op (Pebble tombstones are written regardless, which is correct — the key is then absent from all subsequent reads). The timeline event counter is decremented on success; the counter is approximate by design so a small discrepancy from a missed write is acceptable.
func (*PebbleStore) DeleteKeys ¶
func (s *PebbleStore) DeleteKeys(ctx context.Context, keys [][]byte) error
DeleteKeys removes events by their pre-encoded Pebble keys. Keys must have been produced by EncodeKey; passing arbitrary byte slices is undefined behaviour. All deletes are issued in a single Pebble batch committed with Sync durability, so either all succeed or none do.
Event counters are NOT adjusted — they are approximate by design and the caller is expected to use Delete when an exact decrement matters. This method exists as a fast rollback path for /commit, where the keys are already in hand and counter precision is not required.
func (*PebbleStore) DeleteRollup ¶
func (s *PebbleStore) DeleteRollup(sourceTID TimelineID, id RollupID) error
func (*PebbleStore) DeleteTimeline ¶
func (s *PebbleStore) DeleteTimeline(ctx context.Context, id TimelineID) error
DeleteTimeline removes a timeline's definition, its event data, and its rollups. It is the inverse of DefineTimeline and is distinct from DeleteTimelineData (which clears events but keeps the definition).
Cascade follows the same RollupCascadeDelete policy that DeleteRollup uses:
- cascade on (default): the timeline's rollups are removed first, then its data, then its definition.
- cascade off: if the timeline still has rollups, the call returns an error and changes nothing; the caller must delete the rollups first.
Concurrency: before tearing anything down, the timeline is marked deleting in the registry, after which get reports it as not-found. Concurrent readers and writers therefore fail fast with a clean not-found instead of racing the data teardown and observing a defined-but-empty timeline. If the cascade-off check rejects the delete — or any teardown step fails — the marker is cleared and the timeline remains fully usable (the operation is all-or-nothing as seen by callers). The internal sequence still runs rollups → data → definition so no step leaves a rollup pointing at a source whose data or definition is gone. Timeline 0 is the structural root and cannot be deleted.
func (*PebbleStore) DeleteTimelineData ¶
func (s *PebbleStore) DeleteTimelineData(ctx context.Context, id TimelineID) error
func (*PebbleStore) GetRollup ¶
func (s *PebbleStore) GetRollup(sourceTID TimelineID, id RollupID) (RollupDef, error)
func (*PebbleStore) Latest ¶
func (s *PebbleStore) Latest(ctx context.Context, q LatestQuery) ([]Event, error)
func (*PebbleStore) ListRollups ¶
func (s *PebbleStore) ListRollups(sourceTID TimelineID) ([]RollupDef, error)
func (*PebbleStore) PebbleDB ¶ added in v0.26.0
func (s *PebbleStore) PebbleDB() *pebble.DB
PebbleDB returns the store's underlying *pebble.DB.
func (*PebbleStore) Purge ¶
func (s *PebbleStore) Purge(ctx context.Context) error
Purge deletes events older than the applicable retention window for each timeline. Timelines with effective RetentionDays == 0 are skipped (no expiry).
func (*PebbleStore) PurgeTimelineRange ¶
func (s *PebbleStore) PurgeTimelineRange(ctx context.Context, id TimelineID, from, to time.Time) error
func (*PebbleStore) QueryRange ¶
func (s *PebbleStore) QueryRange(ctx context.Context, q RangeQuery) ([]Event, error)
func (*PebbleStore) RangeAggregate ¶
func (s *PebbleStore) RangeAggregate(ctx context.Context, q RangeAllQuery) (*RangeAggregateResult, error)
func (*PebbleStore) RangeAvg ¶
func (s *PebbleStore) RangeAvg(ctx context.Context, q RangeNumQuery) (float64, error)
RangeAvg returns the average of num field q.NumField over the query range. Returns 0 with no error when no events carry the field. Syntax sugar over RangeAggregate; performs one full scan pass.
func (*PebbleStore) RangeCount ¶
func (s *PebbleStore) RangeCount(ctx context.Context, q RangeNumQuery) (uint64, error)
RangeCount returns the count of events in the query range that carry num field q.NumField. Syntax sugar over RangeAggregate; performs one full scan pass.
func (*PebbleStore) RangeFullAggregate ¶
func (s *PebbleStore) RangeFullAggregate(ctx context.Context, q RangeFullQuery) (*RangeFullResult, error)
RangeFullAggregate computes exact sum/avg/min/max/count for all seven numeric fields AND approximate quantiles for the requested fields in a single Pebble scan pass.
Digests are allocated at the start of the scan (one per requested field, ~16 KB each at compression=100) and discarded after quantile extraction. They are never stored in the result; RangeAggregateResult remains a plain value type.
If q.Quantiles is empty the method is equivalent to RangeAggregate and no digests are allocated.
func (*PebbleStore) RangeMax ¶
func (s *PebbleStore) RangeMax(ctx context.Context, q RangeNumQuery) (float64, error)
RangeMax returns the maximum of num field q.NumField over the query range. Returns 0 with no error when no events carry the field. Syntax sugar over RangeAggregate; performs one full scan pass.
func (*PebbleStore) RangeMedian ¶
func (s *PebbleStore) RangeMedian(ctx context.Context, q RangeNumQuery) (float64, error)
RangeMedian returns the approximate P50 for q.NumField over the query range. Syntax sugar over RangeQuantile(ctx, q, 0.5). NOT OPTIMISED: performs a separate scan pass from RangeAggregate. See RangeQuantile for the suggested future optimisation.
func (*PebbleStore) RangeMin ¶
func (s *PebbleStore) RangeMin(ctx context.Context, q RangeNumQuery) (float64, error)
RangeMin returns the minimum of num field q.NumField over the query range. Returns 0 with no error when no events carry the field. Syntax sugar over RangeAggregate; performs one full scan pass.
func (*PebbleStore) RangeQuantile ¶
func (s *PebbleStore) RangeQuantile(ctx context.Context, q RangeNumQuery, quantile float64) (float64, error)
RangeQuantile returns an approximate quantile estimate for q.NumField over the query range using a t-digest (compression=100, ~16 KB per call).
NOT OPTIMISED: this method performs its own full Pebble scan pass, separate from RangeAggregate. A caller needing both quantile and sum/avg/min/max for the same range must issue two queries and pay for two scans.
If a single-pass combined result is ever needed, introduce a separate RangeFullQuery / RangeFullResult pair rather than embedding *tdigest.TDigest into RangeAggregateResult. Keeping the types separate preserves RangeAggregateResult as a plain value type (no heap pointers, trivially copyable and serialisable) and avoids surfacing the estimator implementation through the Store contract.
func (*PebbleStore) RangeSum ¶
func (s *PebbleStore) RangeSum(ctx context.Context, q RangeNumQuery) (float64, error)
RangeSum returns the sum of num field q.NumField over the query range. Syntax sugar over RangeAggregate; performs one full scan pass.
func (*PebbleStore) RollupParent ¶
func (s *PebbleStore) RollupParent(tid TimelineID) (RollupDef, bool)
func (*PebbleStore) RollupStatus ¶
func (s *PebbleStore) RollupStatus(sourceTID TimelineID, id RollupID) (RollupStatusReport, error)
func (*PebbleStore) RollupTree ¶
func (s *PebbleStore) RollupTree() *RollupTreeNode
func (*PebbleStore) RunRollup ¶
func (s *PebbleStore) RunRollup(ctx context.Context, sourceTID TimelineID, id RollupID, from, to time.Time, cascade bool) error
func (*PebbleStore) SetDefaultRetentionDays ¶
func (s *PebbleStore) SetDefaultRetentionDays(days int) error
SetDefaultRetentionDays updates the store-level default retention and persists it.
func (*PebbleStore) SetWriteConfig ¶
func (s *PebbleStore) SetWriteConfig(id TimelineID, cfg TimelineWriteConfig) error
SetWriteConfig updates the write performance configuration for a timeline. The timeline must already be defined. The new config is persisted to disk.
func (*PebbleStore) Stats ¶
func (s *PebbleStore) Stats(_ context.Context) (*StoreStats, error)
func (*PebbleStore) SysmaskWidth ¶ added in v0.16.3
func (s *PebbleStore) SysmaskWidth() SysmaskWidth
SysmaskWidth returns the store's immutable system/user partition width (@S), loaded once from meta.json at open.
func (*PebbleStore) Timeline ¶
func (s *PebbleStore) Timeline(id TimelineID) (TimelineConfig, bool)
func (*PebbleStore) TimelineStats ¶
func (s *PebbleStore) TimelineStats(ctx context.Context, id TimelineID) (*TimelineStats, error)
func (*PebbleStore) Timelines ¶
func (s *PebbleStore) Timelines() []TimelineID
func (*PebbleStore) UpdateTimeline ¶
func (s *PebbleStore) UpdateTimeline(id TimelineID, cfg TimelineConfig) error
func (*PebbleStore) WriteConfig ¶
func (s *PebbleStore) WriteConfig(id TimelineID) TimelineWriteConfig
WriteConfig returns the write performance configuration for the given timeline. Returns the zero value (NoSync=false) if no config has been set.
type RangeAggregateResult ¶
type RangeAggregateResult struct {
Count uint64
Sums [7]float64
Avgs [7]float64 // populated after scan: Sums[i]/Count; NaN if Count==0
Mins [7]float64
Maxs [7]float64
Fields [7]bool // true if field i appeared in at least one event
}
RangeAggregateResult holds per-field statistics from a single scan pass. Fields[i] indicates whether num field i was present in at least one event; entries for absent fields carry zero values.
type RangeAllQuery ¶
type RangeAllQuery struct {
Timeline TimelineID
Dims []uint64
From time.Time
To time.Time
MaxScanEvents int // 0 = no limit
}
RangeAllQuery is the query shape for RangeAggregate, which computes statistics over all populated numeric fields in a single scan pass. No NumField — the result covers every field present in the matched events.
type RangeFullQuery ¶
type RangeFullQuery struct {
RangeAllQuery
Quantiles []float64 // quantile values to estimate, e.g. [0.5, 0.9, 0.99]
QuantileFields []uint8 // num fields to estimate quantiles for (0–6); nil = all fields
}
RangeFullQuery is the query shape for RangeFullAggregate, which computes sum, avg, min, max, count (via RangeAggregateResult) AND approximate quantiles for selected numeric fields — all in a single Pebble scan pass.
Quantiles lists the desired quantile values, e.g. [0.5, 0.9, 0.99]. Each value must be in [0, 1]; RangeFullAggregate returns an error otherwise.
QuantileFields lists which numeric fields (0–6) should have quantiles computed. If nil, quantiles are computed for all seven fields, allocating ~16 KB of t-digest state per field. Callers should be explicit about which fields they need to avoid unnecessary allocation.
If Quantiles is empty, RangeFullAggregate behaves identically to RangeAggregate (no digests allocated).
type RangeFullResult ¶
type RangeFullResult struct {
Aggregate RangeAggregateResult
Quantiles [7][]float64
}
RangeFullResult holds the combined output of RangeFullAggregate. Aggregate contains the exact statistics (same as RangeAggregate). Quantiles[i][j] is the estimate for field i at Quantiles[j] from RangeFullQuery. A nil inner slice means field i was not requested or carried no events. The outer array is always length 7 (one slot per field).
type RangeNumQuery ¶
type RangeNumQuery struct {
Timeline TimelineID
Dims []uint64
From time.Time
To time.Time
NumField uint8 // 0–6
MaxScanEvents int // 0 = no limit
}
Bucket holds one time bucket of an aggregation result. RangeNumQuery is the query shape for single-field scalar range functions: RangeSum, RangeAvg, RangeMin, RangeMax, RangeCount. NumField is validated (0–6) and must correspond to a populated field.
type RangeQuery ¶
type RangeQuery struct {
Timeline TimelineID
Dims []uint64
From time.Time
To time.Time
Limit int // default 1000, max 10000
Order string // "asc" (default) or "desc"
MaxScanEvents int // 0 = use store/server default; aborts scan if exceeded
}
RangeQuery retrieves events from a timeline over a time range. Dims is a leading prefix: 1 ≤ len(Dims) ≤ timeline.Dims.
type RetentionWorker ¶
type RetentionWorker struct {
// contains filtered or unexported fields
}
RetentionWorker runs periodic retention sweeps against all provisioned tenant stores managed by a Manager.
func NewRetentionWorker ¶
func NewRetentionWorker(m *DefaultManager, interval time.Duration) *RetentionWorker
NewRetentionWorker creates a RetentionWorker that calls Purge on every open store at the given interval.
func (*RetentionWorker) Start ¶
func (w *RetentionWorker) Start()
Start launches the retention goroutine.
func (*RetentionWorker) Stop ¶
func (w *RetentionWorker) Stop()
Stop signals the worker to stop and waits for it to exit.
type RollupDef ¶
type RollupDef struct {
ID RollupID `json:"id"`
SourceTID TimelineID `json:"source_tid"`
DestTID TimelineID `json:"dest_tid"`
BucketDuration time.Duration `json:"bucket_duration"`
// LateWindow is how long after a bucket closes to wait before computing
// the rollup, to absorb late-arriving events. Default 0.
LateWindow time.Duration `json:"late_window,omitempty"`
// Running records whether the worker was active when last persisted.
// Workers are not started automatically at DefineRollup time; they are
// started explicitly via RunRollup (which starts the worker and cascades
// to all descendants). On store reopen, only definitions with Running=true
// restart their workers.
Running bool `json:"running"`
CreatedAt time.Time `json:"created_at"`
}
RollupDef describes one rollup definition: read from SourceTID every BucketDuration, aggregate, and write into DestTID.
type RollupID ¶
type RollupID string
RollupID uniquely identifies a rollup definition within a source timeline.
type RollupStatusReport ¶
type RollupStatusReport struct {
ID RollupID `json:"id"`
SourceTID TimelineID `json:"source_tid"`
DestTID TimelineID `json:"dest_tid"`
LastRunAt time.Time `json:"last_run_at,omitempty"`
LastBucketEnd time.Time `json:"last_bucket_end,omitempty"`
EventsWritten int64 `json:"events_written"`
LastError string `json:"last_error,omitempty"`
Running bool `json:"running"`
}
RollupStatusReport carries the last-known operational state of a rollup worker.
type RollupTreeNode ¶
type RollupTreeNode struct {
TID TimelineID `json:"tid"`
Def *RollupDef `json:"def,omitempty"` // nil for the root (tid=0) and raw timelines
Children []*RollupTreeNode `json:"children,omitempty"`
}
RollupTreeNode is one node in the tenant rollup tree as returned by RollupTree.
type Store ¶
type Store interface {
// Timeline management
DefineTimeline(id TimelineID, cfg TimelineConfig) error
// DefineSystemTimeline is the system-internal define path (@S §8):
// it mints a system-region id and is not exposed on the tenant HTTP
// surface. Refuses non-system ids (the symmetric guard).
DefineSystemTimeline(id TimelineID, cfg TimelineConfig) error
UpdateTimeline(id TimelineID, cfg TimelineConfig) error // name + RetentionDays only
Timeline(id TimelineID) (TimelineConfig, bool)
Timelines() []TimelineID
// SysmaskWidth returns the store's immutable system/user partition
// width (@S). 0 means no system reservation.
SysmaskWidth() SysmaskWidth
// Write
Append(ctx context.Context, e Event) error
AppendBatch(ctx context.Context, events []Event, maxBatch int) (int, error)
// Delete removes the event identified by e from the store by computing its
// encoded key from (Timeline, Dims, Time) and issuing a hard delete.
// The event counter for the timeline is decremented on success.
//
// Implementors: returning nil from a backend that does not actually delete
// the event is a silent correctness bug. The /commit endpoint calls Delete
// and DeleteKeys as a rollback mechanism; a no-op implementation means
// orphaned timeseries events will silently survive a SQLite failure and
// become permanently inconsistent with entity state. Always return
// ErrDeleteNotSupported if your backend cannot honour the deletion.
Delete(ctx context.Context, e Event) error
// DeleteKeys removes events by their pre-encoded keys. Keys must be produced
// by EncodeKey; passing arbitrary byte slices produces undefined behaviour.
// This is the preferred path when the caller already holds encoded keys
// (e.g. during /commit rollback) and wants to avoid re-encoding overhead.
// Because raw keys carry no Go-level timeline identity, implementations that
// successfully delete via this method do NOT adjust event counters — the
// counter is already documented as approximate (see storeMeta). Callers that
// require an exact counter decrement should use Delete instead.
//
// Implementors: the same obligation as Delete applies. A silent no-op here
// is not safe — it will cause /commit to silently succeed the rollback path
// while leaving orphaned Pebble entries in place. Return ErrDeleteNotSupported
// if your backend cannot delete by raw key; the /commit handler will log
// XOLU-CM016 and alert operators rather than silently corrupting the store.
DeleteKeys(ctx context.Context, keys [][]byte) error
// Read
QueryRange(ctx context.Context, q RangeQuery) ([]Event, error)
Latest(ctx context.Context, q LatestQuery) ([]Event, error)
// Aggregate — bucketed or scalar, single numeric field
Aggregate(ctx context.Context, q AggregateQuery) ([]Bucket, error)
// Single-field scalar range functions. Each performs one scan pass
// over [From, To] for the given NumField. Kept alongside RangeAggregate
// to allow direct performance comparison via benchmarks.
RangeSum(ctx context.Context, q RangeNumQuery) (float64, error)
RangeAvg(ctx context.Context, q RangeNumQuery) (float64, error)
RangeMin(ctx context.Context, q RangeNumQuery) (float64, error)
RangeMax(ctx context.Context, q RangeNumQuery) (float64, error)
RangeCount(ctx context.Context, q RangeNumQuery) (uint64, error)
// RangeAggregate computes Count, Sum, Avg, Min, Max for all seven
// numeric fields simultaneously in a single scan pass.
RangeAggregate(ctx context.Context, q RangeAllQuery) (*RangeAggregateResult, error)
// RangeQuantile returns an approximate quantile estimate for a single
// numeric field over [From, To] using a t-digest (compression=100).
//
// q must be in [0, 1]. Returns (0, nil) when no events carry NumField.
//
// Performance note: RangeQuantile performs its own full scan pass and
// cannot be combined with RangeAggregate in a single pass. A caller
// needing both sum/avg/min/max AND a quantile estimate for the same
// range must issue two separate queries and pay for two scans.
//
// Future optimisation: if a single-pass combined result becomes
// necessary, introduce a separate RangeFullQuery / RangeFullResult pair
// rather than embedding *tdigest.TDigest into RangeAggregateResult.
// Keeping the types separate preserves RangeAggregateResult as a plain
// value type (no heap pointers, trivially copyable and serialisable) and
// avoids surfacing the quantile estimator implementation as part of the
// Store contract.
RangeQuantile(ctx context.Context, q RangeNumQuery, quantile float64) (float64, error)
// RangeMedian returns the approximate P50 for a single numeric field
// over [From, To]. Syntax sugar over RangeQuantile(ctx, q, 0.5).
// Carries the same two-scan limitation; see RangeQuantile.
RangeMedian(ctx context.Context, q RangeNumQuery) (float64, error)
// RangeFullAggregate computes exact sum/avg/min/max/count for all seven
// numeric fields AND approximate quantiles for selected fields in a single
// Pebble scan pass.
//
// This is the efficient alternative to calling RangeAggregate and
// RangeQuantile separately when both are needed. RangeAggregateResult
// is kept as a plain value type; digests are allocated during the scan
// and discarded after quantile extraction, never stored in the result.
//
// If RangeFullQuery.Quantiles is empty the call is equivalent to
// RangeAggregate with no additional cost.
RangeFullAggregate(ctx context.Context, q RangeFullQuery) (*RangeFullResult, error)
// Retention
Purge(ctx context.Context) error
// Retention configuration
DefaultRetentionDays() int
SetDefaultRetentionDays(days int) error
// WriteConfig returns the write performance configuration for a timeline.
// Returns the zero value (both flags false) if the timeline has no explicit
// config set or if id is not defined.
WriteConfig(id TimelineID) TimelineWriteConfig
// SetWriteConfig updates the write performance configuration for a timeline.
// The timeline must already exist. The config is persisted to disk so it
// survives store restarts.
SetWriteConfig(id TimelineID, cfg TimelineWriteConfig) error
// Diagnostics
Stats(ctx context.Context) (*StoreStats, error)
TimelineStats(ctx context.Context, id TimelineID) (*TimelineStats, error)
// Rollup management
//
// DefineRollup creates or updates a rollup definition on sourceTID.
// Timeline 0 is rejected (XOLU-TS022). The destination must not already
// be the target of another definition (XOLU-TS026). Adding a definition
// that would create a cycle (XOLU-TS023) or exceed the depth limit
// (XOLU-TS024) is rejected. Returns the assigned RollupID.
DefineRollup(sourceTID TimelineID, def RollupDef) (RollupID, error)
// GetRollup returns a specific rollup definition by its ID on sourceTID.
GetRollup(sourceTID TimelineID, id RollupID) (RollupDef, error)
// ListRollups returns all rollup definitions where sourceTID is the source.
ListRollups(sourceTID TimelineID) ([]RollupDef, error)
// DeleteRollup removes a rollup definition and stops its worker.
// Data already written to the destination timeline is not affected.
DeleteRollup(sourceTID TimelineID, id RollupID) error
// RollupParent returns the rollup definition for which sourceTID is the
// destination — i.e. the definition that feeds into this timeline.
// Returns (RollupDef{}, false) if this timeline has no parent rollup.
RollupParent(sourceTID TimelineID) (RollupDef, bool)
// RollupTree returns the full rollup tree for this store, rooted at
// timeline 0. Each node carries its definition and its children.
RollupTree() *RollupTreeNode
// RunRollup executes a rollup definition immediately for the given time
// range, writing the results into the destination timeline. If from and
// to are both zero, runs for the most recently closed bucket.
// If cascade is true, after completing the specified bucket RunRollup
// also runs all descendant definitions for the corresponding time windows,
// walking down the tree in source→destination order. The worker goroutine
// is started for this definition and all cascaded descendants if not
// already running.
RunRollup(ctx context.Context, sourceTID TimelineID, id RollupID, from, to time.Time, cascade bool) error
// RollupStatus returns the operational status of a rollup definition.
RollupStatus(sourceTID TimelineID, id RollupID) (RollupStatusReport, error)
// Data deletion
//
// DeleteTimelineData removes all events from a timeline's Pebble key
// range. The timeline definition is preserved. Timeline 0 is rejected.
DeleteTimelineData(ctx context.Context, id TimelineID) error
// DeleteTimeline removes a timeline's definition together with its event
// data and its rollups. It is the inverse of DefineTimeline and is distinct
// from DeleteTimelineData (which keeps the definition). Rollup cascade
// follows RollupCascadeDelete: when off and the timeline still has rollups,
// the call returns an error and changes nothing. Timeline 0 is rejected.
DeleteTimeline(ctx context.Context, id TimelineID) error
// PurgeTimelineRange removes events in [from, to] from a timeline.
// Timeline 0 is rejected.
PurgeTimelineRange(ctx context.Context, id TimelineID, from, to time.Time) error
// Lifecycle
Close() error
}
Store is the interface for a single tenant's timeseries backend. Implementations must be safe for concurrent use.
func NewPebbleStore ¶
func NewPebbleStore(dir string, cfg StoreConfig, pcfg PebbleConfig, tenantName string, dc *dynconfig.DynConfig) (Store, error)
NewPebbleStore opens or creates a Pebble timeseries store in dir. cfg carries the backend-agnostic settings (retention); pcfg carries the Pebble-specific tuning parameters. Zero values in pcfg are safe — sensible defaults are applied for each unset field.
type StoreConfig ¶
type StoreConfig struct {
DefaultRetentionDays int // store-level fallback; 0 = no expiry
// SysmaskWidth is the immutable system/user partition width (@S),
// applied ONLY at store creation. On reopen the persisted value in
// meta.json wins and this field is ignored — the width can never
// change for the life of a store. 0 (the default) means no system
// reservation.
SysmaskWidth SysmaskWidth
// RollupCascadeDelete controls whether DeleteRollup automatically removes
// all descendant definitions. When true (default), deleting a parent
// removes its entire subtree. When false, deleting a definition that has
// descendants returns an error; the caller must delete bottom-up.
RollupCascadeDelete bool
}
StoreConfig holds configuration that is meaningful to any timeseries store backend. It is passed through the StoreFactory contract and must not contain engine-specific knobs.
type StoreFactory ¶
type StoreFactory func(dir string, cfg StoreConfig, tenantName string) (Store, error)
StoreFactory creates a Store for a given data directory and tenant name. The tenantName is used to scope dynconfig lookups to the tenant's namespace.
func NewPebbleStoreFactory ¶
func NewPebbleStoreFactory(pcfg PebbleConfig, dc *dynconfig.DynConfig) StoreFactory
NewPebbleStoreFactory returns a StoreFactory backed by Pebble. The supplied PebbleConfig is captured in the closure; callers only need to thread the backend-agnostic StoreConfig through the factory contract.
A zero-value PebbleConfig is valid — NewPebbleStore applies sensible defaults for every unset field.
type StoreStats ¶
StoreStats holds aggregate diagnostics for the entire tenant store.
type SysmaskWidth ¶ added in v0.16.3
type SysmaskWidth uint8
SysmaskWidth is the immutable count of top bits of a 32-bit TimelineID that form the system/user region selector. See docs/proposals/system-bookkeeping-scope.md (@S).
It is a *width*, not a mask value and not a prefix: the number of high bits reserved as the selector. An id is system-scope iff those selector bits are non-zero.
- width 0 → no bits reserved; the entire space is user-space (the default: a store that never opts in behaves as if the feature is off).
- width 32 → every bit is a selector; the only all-zero value is id 0 (already reserved), so the store is effectively system-only.
- 0 < width < 32 → a split: low ids (zero selector) are user, high ids (non-zero selector) are system.
The width is frozen at store initialisation and never changes for the life of the store. Changing a store's effective width is done offline by transvasing into a new store (@S §6), never by mutation.
const MaxSysmaskWidth SysmaskWidth = 32
MaxSysmaskWidth is the widest legal sysmask (the full 32-bit id).
func (SysmaskWidth) IsSystem ¶ added in v0.16.3
func (w SysmaskWidth) IsSystem(id TimelineID) bool
IsSystem reports whether id falls in the system region under this width. One shift and one compare, with both extremes handled:
- width 0: no selector bits, so nothing is system. Special-cased because `id >> 32` is not well-defined for a 32-bit value.
- width 32: every bit selects, so system iff id != 0.
func (SysmaskWidth) IsUser ¶ added in v0.16.3
func (w SysmaskWidth) IsUser(id TimelineID) bool
IsUser is the complement of IsSystem.
func (SysmaskWidth) MaxUserID ¶ added in v0.16.3
func (w SysmaskWidth) MaxUserID() TimelineID
MaxUserID returns the highest id that is still user-space under this width — i.e. the id whose selector bits are all zero and whose remaining bits are all one. For width 0 this is MaxTimelineID (the whole space); for width 32 it is 0 (no user ids exist beyond the reserved id 0). Used by the user allocator to detect user-space exhaustion without crossing the partition.
func (SysmaskWidth) String ¶ added in v0.16.3
func (w SysmaskWidth) String() string
String renders the width for display, e.g. in `iolu db status`.
func (SysmaskWidth) Valid ¶ added in v0.16.3
func (w SysmaskWidth) Valid() bool
Valid reports whether the width is in the legal range 0–32.
type TimelineConfig ¶
type TimelineConfig struct {
Name string // optional, human-readable label
Dims uint8 // 1–5, immutable after FirstWriteAt is set
RetentionDays int // 0 = use store-level default
CreatedAt time.Time
FirstWriteAt time.Time // zero until first event written; Dims locks here
}
TimelineConfig describes a timeline's declaration. Dims is immutable after the first write; Name and RetentionDays may be updated freely.
type TimelineID ¶
type TimelineID uint32
TimelineID is a uint32 identifier for a timeline within a tenant store. ID 0x00000000 is reserved; valid IDs are 0x00000001–0xFFFFFFFF. Widened from uint16 in the wave-1 per-primitive ID pass (@P wave 1) so per-tenant timeline counts scale past 65k — the ceiling that mid-market workloads reach long before a machine reaches its tenant count. On-disk Pebble key format widens from 2 to 4 bytes for the TimelineID prefix; done before production data exists so no migration is owed.
const MaxTimelineID TimelineID = 0xFFFFFFFF
MaxTimelineID is the highest valid timeline ID.
func TimelineIDFromJSON ¶ added in v0.16.3
func TimelineIDFromJSON(v int64) (TimelineID, error)
TimelineIDFromJSON converts a JSON-decoded numeric timeline id (carried as int64, the only width that holds the full uint32 range on every platform including 32-bit) into a TimelineID, validating the range.
This is the single sanctioned crossing from the JSON/transport boundary into the typed id space. Callers must not hand-narrow a JSON number to TimelineID directly: a plain `int` cannot hold ids above 2^31 on 32-bit targets (silent truncation), and `int(MaxTimelineID)` overflows the constant on those targets (compile error). Routing every boundary through this function makes both failure modes impossible.
The id must be in [1, MaxTimelineID]: 0 is reserved (registry rejects it too), and anything above the uint32 ceiling is out of range.
type TimelineStats ¶
type TimelineStats struct {
TotalEvents int64
TotalEventsApproximate bool // always true; counter is eventually consistent
OldestEvent time.Time
NewestEvent time.Time
}
TimelineStats holds diagnostics for a single timeline. TotalEvents is derived from an in-memory counter that is persisted periodically to meta.json. After a crash without a clean Close, the counter may be stale; TotalEventsApproximate is always true for the current PebbleStore implementation.
type TimelineWriteConfig ¶
type TimelineWriteConfig struct {
// NoSync — when true, AppendBatch commits this timeline's events with
// pebble.NoSync instead of pebble.Sync. The OS page cache is not flushed
// to durable storage before the call returns. Data loss is possible if
// the process crashes before the OS writes the WAL to disk.
NoSync bool `json:"nosync"`
}
TimelineWriteConfig holds the per-timeline performance configuration. The only remaining per-timeline flag is NoSync; write coalescing is controlled process-wide (or per-tenant) via dynconfig keys:
ts.writecoal bool — enable the write coalescer (default false) ts.coal_flush_interval_ms int — flush window in ms (default 10) ts.coal_max_events int — early-flush threshold (default 2000)
These keys are read from the tenant's dynconfig namespace first ("tenant.{name}"), falling back to "global" if absent.