Documentation
¶
Overview ¶
Package router routes reads and writes by the cluster ring without joining it.
A storage node holds shards, so it obtains the ring as a side effect of registering itself as a member. The other tiers — a query front end, an ingester — need the same routing view while owning no data: they must resolve a shard's primary and its owners, but must never be placed as one. etcd.Watch gives them a ring without a lease, and this package turns it into the clients that ring is for: the primary write, and the read surface — Router.Fetcher, the enumeration (Router.Series, Router.Keys, Router.Side) and the metric aggregate pushdown (Router.Aggregate, Router.AggregateWindow). Each carries the RPC over the Router's own HTTP client under its own retry/hedge profile, so an off-ring reader fails over between a shard's owners exactly as a member node does.
A Router is safe for concurrent use and follows membership live: a node joining or failing is reflected in the next placement lookup, with no reconnect on the caller's part.
Index ¶
- type Config
- type Placement
- type Router
- func (r *Router) Aggregate(ctx context.Context, shardKey signal.TenantID, start, end, step int64, ...) ([]engine.NamedAgg, error)
- func (r *Router) AggregateWindow(ctx context.Context, shardKey signal.TenantID, start, end int64, ...) ([]engine.NamedWindowAgg, error)
- func (r *Router) Close(ctx context.Context) error
- func (r *Router) Fetcher(sig signal.Signal, shardKey signal.TenantID) fetch.Fetcher
- func (r *Router) Keys(ctx context.Context, sig signal.Signal, shardKey signal.TenantID, ...) ([]cluster.KeyInfo, error)
- func (r *Router) Members() []etcd.Member
- func (r *Router) Owners(shardKey signal.TenantID) []string
- func (r *Router) Primary(shardKey signal.TenantID) (string, bool)
- func (r *Router) PrimaryWrite(ctx context.Context, sig signal.Signal, shardKey signal.TenantID, ...) (cluster.Reject, error)
- func (r *Router) Series(ctx context.Context, sig signal.Signal, shardKey signal.TenantID, ...) ([]signal.Series, error)
- func (r *Router) ShardCount() int
- func (r *Router) ShardKey(tenant signal.TenantID, id signal.SeriesID) signal.TenantID
- func (r *Router) ShardKeys(tenant signal.TenantID) []signal.TenantID
- func (r *Router) Side(ctx context.Context, sig signal.Signal, shardKey signal.TenantID) (map[string][]byte, error)
- func (r *Router) WriteLogs(ctx context.Context, ld log.Logs, tenantOf cluster.TenantFunc) (Written, error)
- func (r *Router) WriteMetrics(ctx context.Context, md metric.Metrics, tenantOf cluster.TenantFunc) (Written, error)
- func (r *Router) WriteProfiles(ctx context.Context, pd *profile.Profiles, tenantOf cluster.TenantFunc) (Written, error)
- func (r *Router) WriteTraces(ctx context.Context, td trace.Traces, tenantOf cluster.TenantFunc) (Written, error)
- type Written
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// Etcd is the etcd endpoint list the cluster coordinates membership through. Required.
Etcd []string
// Root is the etcd key prefix for this cluster's state. Empty ⇒ [cluster.DefaultRoot]. It must
// match the storage nodes' [cluster.Config].Root.
Root string
// RF is the default replication factor. Zero ⇒ [cluster.DefaultRF]. It must match the nodes'
// [cluster.Config].RF, or reads and writes resolve a different owner set than the nodes do.
RF int
// ShardsPerTenant must match the nodes' [cluster.Config].ShardsPerTenant. A mismatch routes
// every write to the wrong shard, silently.
ShardsPerTenant int
// Placement, when set, resolves a shard's placement individually — for a cluster where tenants
// override RF or use erasure coding, which the cluster-wide defaults above cannot express.
// Nil ⇒ every shard uses RF with the replica placement.
Placement func(shardKey signal.TenantID) Placement
// Retry tunes the RPC reliability profile. Zero ⇒ [reliability.Default].
Retry reliability.RetryConfig
// HTTP is the client used for node-to-node RPCs. Nil ⇒ one built from Retry.
HTTP *http.Client
// Logger records membership changes. Nil ⇒ no logging.
Logger *zap.Logger
// DialTimeout bounds the initial etcd connection. Zero ⇒ 5s.
DialTimeout time.Duration
}
Config configures a Router.
type Placement ¶
type Placement struct {
// RF is the replication factor: how many owners the shard has. The ring clamps it to the
// membership size.
RF int
// Balanced selects the failure-domain-balanced placement erasure-coded shards use, instead of
// the zone-aware replica placement. It must match what the shard's tenant is configured with,
// or the router resolves a different owner set than the node does.
Balanced bool
}
Placement is how one shard is placed on the ring.
type Router ¶
type Router struct {
// contains filtered or unexported fields
}
Router resolves shard placement from a live ring view and carries the routed RPCs.
func Open ¶
Open connects to etcd and starts following the cluster's membership. The Router registers nothing: it never appears in another node's ring and is never placed as an owner.
func (*Router) Aggregate ¶ added in v0.39.0
func (r *Router) Aggregate( ctx context.Context, shardKey signal.TenantID, start, end, step int64, matchers []fetch.Matcher, ) ([]engine.NamedAgg, error)
Aggregate runs the step-bucketed metric aggregate on one shard's owner, hedged across its owners. The owner folds its own samples and ships one compact entry per series — identity + buckets, never raw points — so the pushdown the `*_over_time` paths rest on survives being run off the ring.
A step of zero asks for one whole-range bucket per series. Series that fail the full matcher set are dropped here: the owner applied only the serializable (equality) subset.
func (*Router) AggregateWindow ¶ added in v0.39.0
func (r *Router) AggregateWindow( ctx context.Context, shardKey signal.TenantID, start, end int64, spec engine.WindowSpec, matchers []fetch.Matcher, ) ([]engine.NamedWindowAgg, error)
AggregateWindow is the overlapping-window form of Router.Aggregate: the owner slides a range vector's evaluation windows itself and returns one entry per series per window.
It is a separate endpoint rather than a widened request on purpose: an owner that predates windows answers 404, which fails over, instead of silently returning disjoint step buckets for an overlapping question.
func (*Router) Fetcher ¶
Fetcher returns a fetch.Fetcher over one shard, racing its owners under the hedged read policy. Every owner holds a complete copy, so the first success is authoritative.
An owner answering cluster.ErrShardAbsent is a failover, not a result: it means the ring points at that node but the node holds no data for the shard (a fresh owner after a rebalance, a node whose membership view lags, one that has not backfilled). Accepting its empty answer would drop every row the shard holds elsewhere, so the read moves to the next owner and reports empty only when all of them disclaim it.
The owner applies only the serializable (fetch.Matcher.Spec) subset of a request's matchers, so its answer is a superset; the returned fetcher re-applies the full set (fetch.Filter) before yielding, exactly as the node's own fan-out does. A caller therefore gets what it asked for and not a superset it has to know to narrow itself.
func (*Router) Keys ¶ added in v0.39.0
func (r *Router) Keys( ctx context.Context, sig signal.Signal, shardKey signal.TenantID, start, end int64, ) ([]cluster.KeyInfo, error)
Keys lists one shard's distinct record-attribute keys within the window, with the scope(s) each was observed in, hedged across the shard's owners. Keys are window-scoped, not matcher-scoped, so nothing is pushed down beyond the window.
func (*Router) Owners ¶
Owners returns the addresses of a shard's ring owners, in placement order (the first is the primary). It is empty when the ring is empty.
func (*Router) Primary ¶
Primary returns the address of a shard's ring primary — the single authority that admits its writes — or false when the ring is empty or the primary has no known address.
func (*Router) PrimaryWrite ¶
func (r *Router) PrimaryWrite( ctx context.Context, sig signal.Signal, shardKey signal.TenantID, walBytes []byte, ) (cluster.Reject, error)
PrimaryWrite sends a shard's WAL-framed records to its ring primary and returns what the primary rejected. The primary is the shard's single authority, so routing every write for a shard to it is what makes the admission decision and the accepted set identical across all its replicas.
It retries only when the request provably never reached the server (retry.ConnFailure): a write is not idempotent at this layer, so one that may have been applied is never re-sent. A caller with its own idempotency key (a Kafka offset watermark, say) can retry more aggressively itself.
func (*Router) Series ¶ added in v0.39.0
func (r *Router) Series( ctx context.Context, sig signal.Signal, shardKey signal.TenantID, matchers []fetch.Matcher, start, end int64, ) ([]signal.Series, error)
Series lists one shard's stream identities matching matchers within the window, hedged across the shard's owners. It pushes down the serializable (equality) matchers and re-applies the full set to the owner's superset, so the result is exactly what matchers select.
The signal selects the peer's engine: logs, traces, profiles and metric series share one enumeration RPC, dispatched by the request's signal byte.
func (*Router) ShardCount ¶
ShardCount is how many shards each tenant is split into.
func (*Router) ShardKey ¶
ShardKey returns the shard key a series routes to. Both tiers derive it the same way, so an ingester's routing and a node's placement agree without coordinating.
func (*Router) ShardKeys ¶
ShardKeys returns every shard key of a tenant, in index order — the fan-out set for a read, which cannot know which shard holds a series before matching one.
func (*Router) Side ¶ added in v0.39.0
func (r *Router) Side( ctx context.Context, sig signal.Signal, shardKey signal.TenantID, ) (map[string][]byte, error)
Side returns one shard's side-store tables (the profile symbol store, for stack resolution), hedged across the shard's owners. Symbols ride the write path, so every owner's copy is complete.
func (*Router) WriteLogs ¶
func (r *Router) WriteLogs(ctx context.Context, ld log.Logs, tenantOf cluster.TenantFunc) (Written, error)
WriteLogs frames a logs batch by shard and routes each shard to its ring primary.
func (*Router) WriteMetrics ¶
func (r *Router) WriteMetrics(ctx context.Context, md metric.Metrics, tenantOf cluster.TenantFunc) (Written, error)
WriteMetrics frames md by shard and routes each shard to its ring primary, in parallel.
It is the write half of a storage node without the node: the same framing, the same shard keys, and the same per-shard primary authority — so an ingester's writes are indistinguishable from a node's, and land in the same places.
The origin-side ingest-rate valve is not applied here: it is per-tenant policy the router does not hold. The cardinality and in-flight-memory valves are head-enforced and still apply, so they come back in the returned breakdown.
type Written ¶
type Written struct {
// Accepted is how many points the primaries took.
Accepted int
// Rejected is the combined per-reason breakdown across every shard.
Rejected cluster.Reject
}
Written is the outcome of a Router.WriteMetrics: what the shards accepted, and what each primary refused and why.
Source Files