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 DecodeBatches(data []byte) ([]*fetch.Batch, error)
• func DecodeFetchRequest(data []byte) (tenant string, start, end int64, eq []fetch.EqualMatcher, err error)
• func DecodeWrite(data []byte) (tenant string, walBytes []byte, err error)
• func EncodeBatches(batches []*fetch.Batch) []byte
• func EncodeFetchRequest(tenant string, start, end int64, eq []fetch.EqualMatcher) []byte
• func EncodeWrite(tenant string, walBytes []byte) []byte
• func ReadHandler(fetchFn FetchFunc) http.Handler
• type Config
• type FetchFunc
• type RemoteFetcher
  • func NewRemoteFetcher(addr string, client *http.Client) *RemoteFetcher
  • func (f *RemoteFetcher) Fetch(ctx context.Context, r fetch.Request) (fetch.Iterator, error)
• type RingSource
• type Writer
  • func NewWriter(rf int, r RingSource, addrOf func(nodeID string) string, ...) *Writer
  • func (w *Writer) Write(ctx context.Context, tenant string, payload []byte) error

Constants

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 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) (tenant string, start, end int64, eq []fetch.EqualMatcher, err error)

DecodeFetchRequest parses a request made by EncodeFetchRequest.

func DecodeWrite

func DecodeWrite(data []byte) (tenant string, walBytes []byte, err error)

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

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(tenant string, start, end int64, eq []fetch.EqualMatcher) []byte

EncodeFetchRequest frames a fetch request: tenant, window, and any serializable equality matchers to push down to the peer (other predicates are re-checked by the requester).

func EncodeWrite

func EncodeWrite(tenant string, walBytes []byte) []byte

EncodeWrite frames tenant ‖ walBytes into a replication payload.

func ReadHandler

func ReadHandler(fetchFn FetchFunc) http.Handler

ReadHandler returns the HTTP handler that serves fetches from the local store, reconstructing the pushed-down equality matchers. Mount it on the node's server at ReadPath.

Types

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
	// 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 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(addr string, client *http.Client) *RemoteFetcher

NewRemoteFetcher returns a fetcher that reads from the peer at addr. A nil client uses http.DefaultClient.

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 are not forwarded — the requester re-applies the full matcher set to the (possibly superset) result.

type RingSource

type RingSource interface {
	Ring() *ring.Ring
}

RingSource provides the current ring (e.g. an etcd-backed Membership).

type Writer

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

Writer is the cluster write path: it routes a tenant's write to that tenant's ring-owners and replicates it to a write quorum via the replica.Replicator. Sharding is by tenant — a whole tenant's stream is pinned to its owner set (locality); finer per-series sharding is a later refinement.

func NewWriter

func NewWriter(rf int, r RingSource, addrOf func(nodeID string) string, repl *replica.Replicator) *Writer

NewWriter constructs a cluster writer. addrOf maps a ring node ID to the network address the transport reaches it on (typically backed by the membership's Member.Addr).

func (*Writer) Write

func (w *Writer) Write(ctx context.Context, tenant string, payload []byte) error

Write replicates payload to the owners of tenant at write quorum. payload must be the EncodeWrite framing of this tenant's write (so a receiving node can apply it).