cluster

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
• func AggregateHandler(fn AggregateFunc) http.Handler
• func DecodeAggregateRequest(data []byte) (tenant string, start, end, step int64, eq []fetch.EqualMatcher, err error)
• 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 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 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 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 Config
• type FetchFunc
• type KeyInfo
  • func DecodeKeyList(data []byte) ([]KeyInfo, error)
  • func FetchKeys(ctx context.Context, client *http.Client, addr string, sig signal.Signal, ...) ([]KeyInfo, error)
• type KeysFunc
• type RemoteAggregator
  • func NewRemoteAggregator(addr string, client *http.Client) *RemoteAggregator
  • func (a *RemoteAggregator) Aggregate(ctx context.Context, tenant string, start, end, step int64, ...) ([]engine.NamedAgg, error)
• type RemoteFetcher
  • func NewRemoteFetcher(sig signal.Signal, addr string, client *http.Client) *RemoteFetcher
  • func (f *RemoteFetcher) Fetch(ctx context.Context, r fetch.Request) (fetch.Iterator, error)
• 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 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.

Functions

func AggregateHandler

func AggregateHandler(fn AggregateFunc) http.Handler

AggregateHandler returns the HTTP handler that serves an aggregate from the local store. Mount it at AggregatePath.

func DecodeAggregateRequest

func DecodeAggregateRequest(data []byte) (tenant string, start, end, step int64, eq []fetch.EqualMatcher, err error)

DecodeAggregateRequest parses a request made by EncodeAggregateRequest.

func DecodeAggregates

func DecodeAggregates(data []byte) ([]engine.NamedAgg, error)

DecodeAggregates parses an EncodeAggregates payload. It bounds-checks every length before slicing, so it never panics on a malformed or truncated response.

func DecodeBatches

func DecodeBatches(data []byte) ([]*fetch.Batch, error)

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

func DecodeLogBatches(data []byte) ([]*fetch.Batch, error)

DecodeLogBatches parses EncodeLogBatches output, recomputing each batch's id from its identity.

func DecodeSeriesList

func DecodeSeriesList(data []byte) ([]signal.Series, error)

DecodeSeriesList parses EncodeSeriesList output.

func DecodeSideTables

func DecodeSideTables(data []byte) (map[string][]byte, error)

DecodeSideTables parses EncodeSideTables output.

func DecodeWrite

func DecodeWrite(data []byte) (sig signal.Signal, tenant string, walBytes []byte, err error)

DecodeWrite splits a payload made by EncodeWrite into the signal, tenant id, and WAL bytes.

func EncodeAggregateRequest

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 EncodeAggregates

func EncodeAggregates(aggs []engine.NamedAgg) []byte

EncodeAggregates serializes per-series aggregates: a count, then per series the identity (the reversible hash pre-image) and its step buckets.

func EncodeBatches

func EncodeBatches(batches []*fetch.Batch) []byte

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

func EncodeKeyList(keys []KeyInfo) []byte

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

func EncodeLogBatches(batches []*fetch.Batch) []byte

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

func EncodeSeriesList(series []signal.Series) []byte

EncodeSeriesList serializes stream identities as length-prefixed reversible hash pre-images.

func EncodeSideTables

func EncodeSideTables(tables map[string][]byte) []byte

EncodeSideTables serializes a side-store table set (sorted by name for determinism).

func EncodeWrite

func EncodeWrite(sig signal.Signal, tenant string, walBytes []byte) []byte

EncodeWrite frames signal ‖ tenant ‖ walBytes into a replication payload.

func FetchSeries

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

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

func KeysHandler(fn KeysFunc) http.Handler

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

func ReadHandler(metricFn, logFn, traceFn, profileFn FetchFunc) http.Handler

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

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.

func SideHandler

func SideHandler(fn SideFunc) http.Handler

SideHandler serves SidePath: it returns the tenant's side-store tables via fn.

Types

type AggregateFunc

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 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
}

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

type KeyInfo struct {
	Key   []byte
	Scope uint8
}

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

func DecodeKeyList(data []byte) ([]KeyInfo, error)

DecodeKeyList parses EncodeKeyList output, bounds-checking every length so a malformed or truncated peer response is rejected rather than panicking.

func FetchKeys

func FetchKeys(
	ctx context.Context, client *http.Client, addr string, sig signal.Signal, tenant string, start, end int64,
) ([]KeyInfo, error)

FetchKeys returns a peer's distinct record-attribute keys for the signal+tenant+window.

type KeysFunc

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

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

RemoteAggregator runs an aggregate over a peer node's AggregateHandler.

func NewRemoteAggregator

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

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.

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

func NewRemoteFetcher(sig signal.Signal, addr string, client *http.Client) *RemoteFetcher

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.

func (*RemoteFetcher) Fetch

func (f *RemoteFetcher) Fetch(ctx context.Context, r fetch.Request) (fetch.Iterator, error)

Fetch forwards r's tenant, window, and serializable (equality) matchers to the peer and returns the decoded batches. Non-equality matchers (and columnar conditions) are not forwarded — the requester re-applies them to the (possibly superset) result.

type SeriesFunc

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).

type SideFunc

type SideFunc func(ctx context.Context, tenant string) (map[string][]byte, error)

SideFunc returns the local store's side-store tables (name → encoded payload) for a tenant.