cluster

type AdmitFunc ¶ added in v0.38.0

type AdmitFunc func(tenant signal.TenantID, b *metric.Batch) bool

AdmitFunc is the origin-side ingest valve: it reports whether a projected batch is admitted, and a false sheds the whole batch before it is framed. Nil admits everything.

It is called once per projected batch in projection order, so an implementation may cache the state it resolves per tenant across a tenant-contiguous run.

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
	// MemberTTL is the TTL of the etcd lease this node's membership registration hangs off:
	// how long the node may be unable to reach etcd before its peers evict it from the ring.
	// Lowering it detects a dead node sooner at the cost of evicting a live but stalled one
	// (a GC pause, a starved CPU, an etcd blip). An evicted node re-registers on its own, so
	// this sets how often that happens, not whether the cluster recovers. Zero ⇒
	// [etcd.DefaultTTL].
	MemberTTL time.Duration
	// 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

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(data []byte) ([]KeyInfo, error)

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

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 MetricFrames ¶ added in v0.38.0

type MetricFrames struct {
	// Shards maps a shard key to its WAL-encoded run of records.
	Shards map[signal.TenantID][]byte
	// Emitted is how many points projection produced, admitted or not.
	Emitted int
	// Shed is how many points the admit valve dropped before framing.
	Shed int
}

MetricFrames is what FrameMetrics produced: one WAL-encoded payload per shard key, ready to be routed to that shard's primary, plus what projection saw.

func FrameMetrics(md metric.Metrics, shards int, tenantOf TenantFunc, admit AdmitFunc) MetricFrames

type PrimaryWriteFunc ¶ added in v0.38.0

type PrimaryWriteFunc func(ctx context.Context, sig signal.Signal, shardKey string, walBytes []byte) (Reject, error)

PrimaryWriteFunc applies a write as the addressed shard's primary and returns what it rejected. walBytes is the WAL-encoded run of records for the shard, framed by EncodeWrite.

type RecordAdmitFunc ¶ added in v0.38.0

type RecordAdmitFunc func(tenant signal.TenantID, b *recordengine.Batch) bool

RecordAdmitFunc is the origin-side ingest valve for records: false sheds the whole stream batch before it is framed. Nil admits everything. Batches arrive in projection order, so an implementation may cache per-tenant state across a contiguous run.

type RecordFrames ¶ added in v0.38.0

type RecordFrames struct {
	// Shards maps a shard key to its WAL-encoded run of records.
	Shards map[signal.TenantID][]byte
	// Emitted is how many records projection produced, admitted or not.
	Emitted int
	// Shed is how many records the admit valve dropped before framing.
	Shed int
}

RecordFrames is what FrameRecords produced: one WAL-encoded payload per shard key.

func FrameRecords(project RecordProjector, shards int, tenantOf TenantFunc, admit RecordAdmitFunc) RecordFrames

type RecordProjector ¶ added in v0.38.0

type RecordProjector func(emit func(*recordengine.Batch)) int

RecordProjector projects a record signal's ingest batch, calling emit once per stream and returning the total record count. It is what log.Project, trace.Project and profile.Project present, so the three signals share one framing path.

type Reject ¶ added in v0.38.0

type Reject struct {
	OOO         int
	Cardinality int
	InFlight    int
}

Reject is the per-reason rejection breakdown a primary reports back to the write's origin, so ingest can attribute OTLP partial-success exactly like the single-node path. The rate valve is applied at the origin, so it is not carried here.

func SendPrimaryWrite(ctx context.Context, client *http.Client, addr string, payload []byte) (Reject, error)

func (r Reject) Total() int

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

func (a *RemoteAggregator) Aggregate(
	ctx context.Context, tenant string, start, end, step int64, eq []fetch.EqualMatcher,
) ([]engine.NamedAgg, error)

func (a *RemoteAggregator) AggregateWindow(
	ctx context.Context, tenant string, start, end int64, spec engine.WindowSpec, eq []fetch.EqualMatcher,
) ([]engine.NamedWindowAgg, error)

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

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

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

type SideFunc ¶ added in v0.2.0

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.

type TenantFunc ¶ added in v0.38.0

type TenantFunc func(signal.Resource, signal.Scope) signal.TenantID

TenantFunc derives a batch's tenant from its resource and scope, so one OTLP batch may fan out to many tenants. A nil func, or one returning "", routes everything to DefaultTenant.