Documentation
¶
Overview ¶
Package cluster implements the L0 distribution layer (DESIGN.md §3, §11, §14 M6–M7): rendezvous (HRW) hashing with spread-minimizing tokens, etcd-backed ring state and leases, RF=3 quorum replication, and rebalancing.
This is a scaffold stub: the ring, etcd integration, replication, and rebalancing are filled in at M6. Single-node users skip this package entirely ([Options.Cluster] == nil ⇒ no cluster layer).
Index ¶
- Constants
- Variables
- func AggregateHandler(fn AggregateFunc) http.Handler
- func AggregateWindowHandler(fn AggregateWindowFunc) http.Handler
- func DecodeAggregateRequest(data []byte) (tenant string, start, end, step int64, eq []fetch.EqualMatcher, err error)
- func DecodeAggregateWindowRequest(data []byte) (tenant string, start, end int64, spec engine.WindowSpec, ...)
- func DecodeAggregates(data []byte) ([]engine.NamedAgg, error)
- func DecodeBatches(data []byte) ([]*fetch.Batch, error)
- func DecodeFetchRequest(data []byte) (sig signal.Signal, tenant string, start, end int64, eq []fetch.EqualMatcher, ...)
- func DecodeLogBatches(data []byte) ([]*fetch.Batch, error)
- func DecodeSeriesList(data []byte) ([]signal.Series, error)
- func DecodeSideTables(data []byte) (map[string][]byte, error)
- func DecodeWindowAggregates(data []byte) ([]engine.NamedWindowAgg, error)
- func DecodeWrite(data []byte) (sig signal.Signal, tenant string, walBytes []byte, err error)
- func EncodeAggregateRequest(tenant string, start, end, step int64, eq []fetch.EqualMatcher) []byte
- func EncodeAggregateWindowRequest(tenant string, start, end int64, spec engine.WindowSpec, ...) []byte
- func EncodeAggregates(aggs []engine.NamedAgg) []byte
- func EncodeBatches(batches []*fetch.Batch) []byte
- func EncodeFetchRequest(sig signal.Signal, tenant string, start, end int64, eq []fetch.EqualMatcher) []byte
- func EncodeKeyList(keys []KeyInfo) []byte
- func EncodeLogBatches(batches []*fetch.Batch) []byte
- func EncodeSeriesList(series []signal.Series) []byte
- func EncodeSideTables(tables map[string][]byte) []byte
- func EncodeWindowAggregates(aggs []engine.NamedWindowAgg) []byte
- func EncodeWrite(sig signal.Signal, tenant string, walBytes []byte) []byte
- func FetchSeries(ctx context.Context, client *http.Client, addr string, sig signal.Signal, ...) ([]signal.Series, error)
- func FetchSide(ctx context.Context, client *http.Client, addr string, sig signal.Signal, ...) (map[string][]byte, error)
- func KeysHandler(fn KeysFunc) http.Handler
- func ReadHandler(metricFn, logFn, traceFn, profileFn FetchFunc) http.Handler
- func SeriesHandler(fn SeriesFunc) http.Handler
- func SideHandler(fn SideFunc) http.Handler
- type AggregateFunc
- type AggregateWindowFunc
- type Config
- type FetchFunc
- type KeyInfo
- type KeysFunc
- type RemoteAggregator
- type RemoteFetcher
- type SeriesFunc
- type SideFunc
Constants ¶
const ( SeriesPath = "/internal/series" SidePath = "/internal/side" KeysPath = "/internal/keys" )
SeriesPath, SidePath, and KeysPath are the HTTP paths of the series-listing, side-store, and attribute-key enumeration servers.
const AggregatePath = "/internal/aggregate"
AggregatePath is the HTTP path the cluster aggregate-pushdown server serves. A peer runs the step-bucketed aggregate over its local shard (using its stats sidecar where it applies) and returns one compact engine.NamedAgg per series — identity + buckets — instead of every sample, so a coordinator gathers and unions across shards without shipping raw points.
const AggregateWindowPath = "/internal/aggregate/window"
AggregateWindowPath is the HTTP path the overlapping-window aggregate server serves. It is a separate endpoint from AggregatePath rather than a widened request on it: the two carry different results (windows keyed by their evaluation timestamp, buckets by their start), and a peer too old to know about windows answers 404 — an error the coordinator fails over on — instead of silently returning disjoint buckets for an overlapping question.
const DefaultRF = 3
DefaultRF is the replication factor used when Config.RF is unset.
const DefaultRoot = "/oteldb"
DefaultRoot is the etcd key prefix used when Config.Root is empty.
const (
ReadPath = "/internal/fetch"
)
ReadPath is the HTTP path the cluster read (fetch fan-out) server serves.
Variables ¶
var ErrShardAbsent = errors.New("cluster: shard not held by peer")
ErrShardAbsent is a peer's answer that it does not hold the requested shard at all: it has no engine for the shard, whatever the ring says (a fresh owner after a rebalance, a node whose membership view lags the writer's, a spare that has not backfilled yet).
It is deliberately not an empty result. An empty success is indistinguishable from "this shard really has no data", so a caller that accepts it silently drops every row the shard holds elsewhere; the caller must instead fail over to another owner, and report empty only when every owner disclaims the shard.
Functions ¶
func AggregateHandler ¶ added in v0.8.0
func AggregateHandler(fn AggregateFunc) http.Handler
AggregateHandler returns the HTTP handler that serves an aggregate from the local store. Mount it at AggregatePath. the shared halves (body read, matcher rebuild) are already factored out.
func AggregateWindowHandler ¶ added in v0.36.0
func AggregateWindowHandler(fn AggregateWindowFunc) http.Handler
AggregateWindowHandler returns the HTTP handler that serves an overlapping-window aggregate from the local store. Mount it at AggregateWindowPath. the shared halves (body read, matcher rebuild) are already factored out.
func DecodeAggregateRequest ¶ added in v0.8.0
func DecodeAggregateRequest(data []byte) (tenant string, start, end, step int64, eq []fetch.EqualMatcher, err error)
DecodeAggregateRequest parses a request made by EncodeAggregateRequest.
func DecodeAggregateWindowRequest ¶ added in v0.36.0
func DecodeAggregateWindowRequest(data []byte) ( tenant string, start, end int64, spec engine.WindowSpec, eq []fetch.EqualMatcher, err error, )
DecodeAggregateWindowRequest parses a request made by EncodeAggregateWindowRequest.
func DecodeAggregates ¶ added in v0.8.0
DecodeAggregates parses an EncodeAggregates payload. It bounds-checks every length before slicing, so it never panics on a malformed or truncated response.
func DecodeBatches ¶
DecodeBatches parses EncodeBatches output, recomputing each batch's id from its identity.
func DecodeFetchRequest ¶
func DecodeFetchRequest(data []byte) (sig signal.Signal, tenant string, start, end int64, eq []fetch.EqualMatcher, err error)
DecodeFetchRequest parses a request made by EncodeFetchRequest.
func DecodeLogBatches ¶ added in v0.2.0
DecodeLogBatches parses EncodeLogBatches output, recomputing each batch's id from its identity.
func DecodeSeriesList ¶ added in v0.2.0
DecodeSeriesList parses EncodeSeriesList output.
func DecodeSideTables ¶ added in v0.2.0
DecodeSideTables parses EncodeSideTables output.
func DecodeWindowAggregates ¶ added in v0.36.0
func DecodeWindowAggregates(data []byte) ([]engine.NamedWindowAgg, error)
DecodeWindowAggregates parses an EncodeWindowAggregates payload. It bounds-checks every length before slicing, so it never panics on a malformed or truncated response.
func DecodeWrite ¶
DecodeWrite splits a payload made by EncodeWrite into the signal, tenant id, and WAL bytes.
func EncodeAggregateRequest ¶ added in v0.8.0
func EncodeAggregateRequest(tenant string, start, end, step int64, eq []fetch.EqualMatcher) []byte
EncodeAggregateRequest frames an aggregate request: tenant, window, step, and the serializable equality matchers to push to the peer (the coordinator re-checks the full set on the response).
func EncodeAggregateWindowRequest ¶ added in v0.36.0
func EncodeAggregateWindowRequest(tenant string, start, end int64, spec engine.WindowSpec, eq []fetch.EqualMatcher) []byte
EncodeAggregateWindowRequest frames a window-aggregate request: tenant, range, the evaluation grid (step, window width, anchor), and the serializable equality matchers to push to the peer (the coordinator re-checks the full set on the response).
func EncodeAggregates ¶ added in v0.8.0
EncodeAggregates serializes per-series aggregates: a count, then per series the identity (the reversible hash pre-image) and its step buckets.
func EncodeBatches ¶
EncodeBatches serializes fetch batches: each series' identity (reversible hash pre-image) followed by its (timestamp, value) samples. The id is recomputed from the identity on decode, so it is not sent.
func EncodeFetchRequest ¶
func EncodeFetchRequest(sig signal.Signal, tenant string, start, end int64, eq []fetch.EqualMatcher) []byte
EncodeFetchRequest frames a fetch request: the signal, tenant, window, and any serializable equality matchers to push down to the peer (other predicates are re-checked by the requester).
func EncodeKeyList ¶ added in v0.10.0
EncodeKeyList serializes a list of distinct attribute keys: a uvarint count, then per key a uvarint length, the key bytes, and a single scope byte.
func EncodeLogBatches ¶ added in v0.2.0
EncodeLogBatches serializes log fetch batches: each stream's identity, its record timestamps, and its named per-record columns (each tagged by physical kind). The id is recomputed from the identity on decode, so it is not sent.
func EncodeSeriesList ¶ added in v0.2.0
EncodeSeriesList serializes stream identities as length-prefixed reversible hash pre-images.
func EncodeSideTables ¶ added in v0.2.0
EncodeSideTables serializes a side-store table set (sorted by name for determinism).
func EncodeWindowAggregates ¶ added in v0.36.0
func EncodeWindowAggregates(aggs []engine.NamedWindowAgg) []byte
EncodeWindowAggregates serializes per-series window aggregates: a count, then per series the identity (the reversible hash pre-image) and its windows, each keyed by its evaluation timestamp.
func EncodeWrite ¶
EncodeWrite frames signal ‖ tenant ‖ walBytes into a replication payload.
func FetchSeries ¶ added in v0.2.0
func FetchSeries( ctx context.Context, client *http.Client, addr string, sig signal.Signal, tenant string, start, end int64, eq []fetch.EqualMatcher, ) ([]signal.Series, error)
FetchSeries lists a peer's stream identities for the signal+tenant+window, pushing down the serializable (equality) matchers; the caller re-applies any non-equality matchers.
func FetchSide ¶ added in v0.2.0
func FetchSide(ctx context.Context, client *http.Client, addr string, sig signal.Signal, tenant string) (map[string][]byte, error)
FetchSide returns a peer's side-store tables for the signal+tenant.
func KeysHandler ¶ added in v0.10.0
KeysHandler serves KeysPath: it enumerates the distinct record-attribute keys for the request's signal+tenant+window via fn (matchers are not used — keys are window-scoped, not matcher-scoped).
func ReadHandler ¶
ReadHandler returns the HTTP handler that serves fetches from the local store, reconstructing the pushed-down equality matchers and dispatching to the metric, log, trace, or profile fetch by the request's signal (encoding the result with the matching batch codec — samples for metrics, columns for the record signals). Mount it at ReadPath.
func SeriesHandler ¶ added in v0.2.0
func SeriesHandler(fn SeriesFunc) http.Handler
SeriesHandler serves SeriesPath: it reconstructs the pushed-down equality matchers and lists the matching stream identities via fn, dispatched to the right engine by the request's signal.
Types ¶
type AggregateFunc ¶ added in v0.8.0
type AggregateFunc func(ctx context.Context, tenant string, start, end, step int64, matchers []fetch.Matcher) ([]engine.NamedAgg, error)
AggregateFunc computes a node-local step-bucketed aggregate of a tenant's metric series matching the (pushed-down equality) matchers, returning each series' identity so a coordinator can re-check the full matcher set and union across shards. It is what AggregateHandler serves.
type AggregateWindowFunc ¶ added in v0.36.0
type AggregateWindowFunc func( ctx context.Context, tenant string, start, end int64, spec engine.WindowSpec, matchers []fetch.Matcher, ) ([]engine.NamedWindowAgg, error)
AggregateWindowFunc computes a node-local overlapping-window aggregate of a tenant's metric series matching the (pushed-down equality) matchers: one aggregate per step-aligned evaluation timestamp over the half-open window (t-window, t]. It is what AggregateWindowHandler serves.
type Config ¶
type Config struct {
// Etcd is the etcd endpoint list for membership coordination.
Etcd []string
// Self is this node's identity: ID (ring identity), Zone (failure domain), and Addr
// (host:port the node listens on for replication and reaches peers at).
Self etcd.Member
// RF is the replication factor (replicas per write). Zero ⇒ 3.
RF int
// ShardsPerTenant splits each tenant's metric series into this many independently-placed
// shards (series → shard = hash(seriesID) % N), so a single large tenant spreads its ingest,
// storage, and compaction across up to N nodes instead of being pinned to one owner set. Zero
// or one ⇒ a single shard (the tenant is the shard; on-disk layout and placement are identical
// to the unsharded path). Applies to metrics only; the record signals are a single shard.
ShardsPerTenant int
// Root is the etcd key prefix for this cluster's state. Empty ⇒ "/oteldb".
Root string
// PrivateBackend declares that this node's backend is private to it (a local disk, not a
// shared object store): peers cannot read the parts this node flushes. The cluster then
// replicates flushed parts node-to-node — replicas mirror their owner's backend objects
// over the parts endpoints (cluster/partsync) instead of loading them from a shared store,
// and an owner backfills from its peers before compacting. False (the default) keeps the
// shared-store model: flushed parts are exchanged through the backend, never over the
// cluster transport.
PrivateBackend bool
}
Config is the cluster configuration. It is optional: a nil [storage.Options].Cluster means single-node mode (the cluster layer is absent). When set, the storage facade joins the etcd-coordinated cluster, runs the replica server on Config.Self.Addr, and routes writes to their ring-owners at replication factor Config.RF.
type FetchFunc ¶
type FetchFunc func(ctx context.Context, tenant string, start, end int64, matchers []fetch.Matcher) ([]*fetch.Batch, error)
FetchFunc fetches a tenant's series within [start, end] from the local store, applying the pushed-down matchers. It is what ReadHandler serves.
type KeyInfo ¶ added in v0.10.0
KeyInfo is one distinct attribute key and the scope(s) it was observed in, as carried over the keys-enumeration RPC. Scope mirrors the record engine's KeyScope bitset (resource/scope/record).
func DecodeKeyList ¶ added in v0.10.0
DecodeKeyList parses EncodeKeyList output, bounds-checking every length so a malformed or truncated peer response is rejected rather than panicking.
type KeysFunc ¶ added in v0.10.0
type KeysFunc func(ctx context.Context, sig signal.Signal, tenant string, start, end int64) ([]KeyInfo, error)
KeysFunc returns the distinct record-attribute keys (with their scope bitset) present in a signal+tenant's records within the window.
type RemoteAggregator ¶ added in v0.8.0
type RemoteAggregator struct {
// contains filtered or unexported fields
}
RemoteAggregator runs an aggregate over a peer node's AggregateHandler.
func NewRemoteAggregator ¶ added in v0.8.0
func NewRemoteAggregator(addr string, client *http.Client) *RemoteAggregator
NewRemoteAggregator returns an aggregator over the peer at addr. A nil client uses http.DefaultClient.
func (*RemoteAggregator) Aggregate ¶ added in v0.8.0
func (a *RemoteAggregator) Aggregate( ctx context.Context, tenant string, start, end, step int64, eq []fetch.EqualMatcher, ) ([]engine.NamedAgg, error)
Aggregate pushes the tenant, window, step, and equality matchers to the peer and returns its per-series aggregates.
func (*RemoteAggregator) AggregateWindow ¶ added in v0.36.0
func (a *RemoteAggregator) AggregateWindow( ctx context.Context, tenant string, start, end int64, spec engine.WindowSpec, eq []fetch.EqualMatcher, ) ([]engine.NamedWindowAgg, error)
AggregateWindow pushes the tenant, range, evaluation grid and equality matchers to the peer and returns its per-series evaluation windows.
type RemoteFetcher ¶
type RemoteFetcher struct {
// contains filtered or unexported fields
}
RemoteFetcher is a fetch.Fetcher over a peer node's ReadHandler. It forwards only the request's tenant and window (matchers are re-applied by the caller), so it returns the peer's full window — a superset the fetch contract permits.
func NewRemoteFetcher ¶
NewRemoteFetcher returns a fetcher that reads the given signal from the peer at addr. A nil client uses http.DefaultClient. The zero signal value reads metrics.
type SeriesFunc ¶ added in v0.2.0
type SeriesFunc func( ctx context.Context, sig signal.Signal, tenant string, start, end int64, matchers []fetch.Matcher, ) ([]signal.Series, error)
SeriesFunc lists the local store's stream identities for a signal+tenant matching matchers within the window (a zero window disables the time filter). The signal selects the engine (logs / traces / profiles share one enumeration RPC, dispatched by the request's signal byte).
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package ec erasure-codes flushed-part backend objects for shared-nothing durability at sub-replica storage cost: an object split into Data shards plus Parity parity shards (systematic Reed-Solomon) survives the loss of any Parity shards while storing only (Data+Parity)/Data of the logical bytes — e.g.
|
Package ec erasure-codes flushed-part backend objects for shared-nothing durability at sub-replica storage cost: an object split into Data shards plus Parity parity shards (systematic Reed-Solomon) survives the loss of any Parity shards while storing only (Data+Parity)/Data of the logical bytes — e.g. |
|
Package etcd backs the L0 cluster ring with etcd: a node registers itself under a lease and watches the member set, so membership is live and self-healing — a crashed node's lease expires and it drops out of every other node's ring within the TTL, with no manual deregistration.
|
Package etcd backs the L0 cluster ring with etcd: a node registers itself under a lease and watches the member set, so membership is live and self-healing — a crashed node's lease expires and it drops out of every other node's ring within the TTL, with no manual deregistration. |
|
Package partsync replicates flushed, immutable parts between nodes whose backends are per-node private (shared-nothing cluster mode).
|
Package partsync replicates flushed, immutable parts between nodes whose backends are per-node private (shared-nothing cluster mode). |
|
Package rebalance computes the minimal ownership changes to apply a cluster membership change (DESIGN.md §11).
|
Package rebalance computes the minimal ownership changes to apply a cluster membership change (DESIGN.md §11). |
|
Package replica is the L0 write-replication layer: it fans an opaque write payload out to the ring-owners of a key and returns once a write **quorum** has durably applied it, so the unflushed head survives the loss of a minority of replicas (DESIGN.md §11; RF=3, quorum (RF/2)+1=2).
|
Package replica is the L0 write-replication layer: it fans an opaque write payload out to the ring-owners of a key and returns once a write **quorum** has durably applied it, so the unflushed head survives the loss of a minority of replicas (DESIGN.md §11; RF=3, quorum (RF/2)+1=2). |
|
Package ring implements rendezvous (highest-random-weight, HRW) hashing — the L0 sharding primitive (DESIGN.md §11).
|
Package ring implements rendezvous (highest-random-weight, HRW) hashing — the L0 sharding primitive (DESIGN.md §11). |