upstream

README

upstream

A backend pool for proxies, tunnels and load balancers. It picks a backend for every connection or request, tracks backend health actively (probes) and passively (real traffic), drains and removes backends gracefully, and reports the state of every backend.

• Lock-free, allocation-free picks; membership and health changes publish a new immutable view atomically, so updates never disturb picks in flight.
• Balancers: round-robin, smooth weighted round-robin, weighted random, power-of-two-choices, least connections and consistent hashing with optional bounded loads, or your own.
• Priority tiers for active-backup setups, with per-backend caps on requests in flight and optional overflow to the next tier.
• Passive health: ejection on consecutive failures or a failure rate, with exponential, capped backoff and half-open trials.
• Active health: TCP connect checks by default, HTTP checks or any function on request; rise/fall thresholds, jitter, bounded concurrency, latency statistics and failure classification.
• Retries on backends not tried yet, bounded by attempts and a retry budget; a dialer and an http.RoundTripper over a pool.
• Runtime reconfiguration: balancer, passive health, retries, slow start and fallback change on a running pool without losing state.
• Slow start for backends that join or recover, explicit all-down behaviour, edge-triggered events.
• Standard library only. Each piece works on its own: balancers without the pool, the outlier detector for a single target, the health checker for any set of addresses.

go get github.com/djylb/upstream

Package	Contents
upstream	Pool, balancers, retries, OutlierDetector, Clock
health	Checker for active health checks, TCP check, Watch and Attach to drive pools
uphttp	HTTP health check, per-backend http.Transport cache, RoundTripper over a pool
upnet	Dialer that connects to a pool's backends

The root package does not import net or net/http; upnet does not import net/http.

Picking backends

pool := upstream.New[string](upstream.Config{
	Balancer:  upstream.PowerOfTwo(),
	Outlier:   upstream.OutlierConfig{ConsecutiveFailures: 5},
	Retry:     upstream.RetryConfig{Attempts: 3, Budget: 0.2},
	SlowStart: 30 * time.Second,
})
defer pool.Close()

err := pool.Sync([]upstream.Target[string]{
	{ID: "a", Value: "10.0.0.1:8080", Weight: 2},
	{ID: "b", Value: "10.0.0.2:8080"},
})

lease, err := pool.Pick() // or PickKey(clientIP) for affinity
if err != nil {
	return err // ErrNoBackends, ErrUnavailable, ErrSaturated or ErrClosed
}
conn, err := net.Dial("tcp", lease.Value())
lease.Report(upstream.OutcomeOf(err)) // the dial decides the outcome
if err != nil {
	lease.Release(upstream.Failure)
	return err
}
defer lease.Release(upstream.Neutral) // the slot follows the connection
defer conn.Close()

upnet.Dialer does all of this, with retries.

A Lease holds one in-flight slot of its backend until it is released, with Success, Failure or Neutral (for example when the client went away). Release every lease exactly once. Releasing the same lease variable twice is a no-op. When the outcome is known long before the lease ends, as for a connection that stays open, Report records it at once and Release later only returns the slot; a half-open backend's trial then ends with the report.

Lease.Observe records a latency sample, such as the time to the first byte, for the backend's moving average; upnet and uphttp observe dial and response-header times themselves.

Sync replaces the membership in one step; Put adds or replaces one backend. A backend keeps its counters and health across updates as long as its ID stays the same. Leases keep the version of the backend they were issued with. An update that changes neither a member's weight nor its value (compared with ==, or with Config.Equal) keeps the backend, emits no event and rebuilds nothing. An empty key carries no affinity: PickKey("") is Pick.

Retries

err := pool.Do(ctx, func(b *upstream.Backend[string]) error {
	conn, err = net.Dial("tcp", b.Value())
	return err
})

Do and DoKey pick a backend, run the function and release the lease with its outcome, even if the function panics. A Failure is retried on a backend not tried yet, up to RetryConfig.Attempts backends, while the context is not done; the last error is returned. WithOutcome(err, upstream.Neutral) marks an error that is not the backend's fault, which is neither counted against it nor retried.

With RetryConfig.Budget, retries within any Window number at most MinRetries plus Budget times the requests started, so retries cannot multiply the load when many requests fail at once.

Pool.Request(key) is the retry loop for leases that outlive a function: each Next leases a backend not returned before, or fails with ErrRetryLimit, ErrRetryBudget or ErrUnavailable, and Request.Do runs the loop with a function that may hand its lease on with Lease.Take:

var conn net.Conn
var kept upstream.Lease[string]
req := pool.Request(clientIP)
err := req.Do(ctx, func(l *upstream.Lease[string]) (retry bool, err error) {
	if conn, err = net.Dial("tcp", l.Value()); err != nil {
		return true, err // released as a Failure, then retried
	}
	kept = l.Take() // released when the connection closes
	return false, nil
})

Dialing and HTTP

upnet.Dialer dials a pool's backends with retries and returns a Conn that holds the backend's lease until it is closed, so least-connections balancing and drains follow real connections. The dial reports the outcome; with DeferReport, the caller reports it with Conn.Report after a handshake such as TLS, so handshake failures count against the backend. The Conn forwards CloseWrite, CloseRead, ReadFrom and WriteTo so that relays keep half-close and the zero-copy paths of io.Copy.

dialer := &upnet.Dialer[string]{Pool: pool}
conn, err := dialer.DialKey(ctx, clientIP)

uphttp.RoundTripper balances an http.Client or a reverse proxy over a pool: it points each request at a picked backend, takes an affinity key from the request, judges responses (5xx is a failure by default) and retries on another backend when the request can be sent again: its body can be replayed and its method is idempotent, or the connection was never established. The lease lasts until the response body is read to the end or closed; for an upgraded connection (101), until it is closed, since the client may still be writing. The outgoing request keeps the incoming Host; with https, the TLS server name and certificate check use the backend's address unless the transport's TLSClientConfig.ServerName says otherwise. For *url.URL values the URL's path and query are joined before the request's, as a reverse proxy to that URL does. Outcome judges a response for health and Retry decides whether to retry it, so that, say, a 503 is retried without counting against the backend. Two timeouts keep a backend that does not answer from using up the whole request's deadline; an attempt they cut before its headers may be retried. AttemptTimeout gives each attempt a real context deadline that lasts until its body ends (dialer, TLS and servers that honour deadlines see it) and fails with ErrAttemptTimeout; HeaderTimeout bounds only the wait for the response headers, sets no deadline and leaves the body unbounded, failing with ErrHeaderTimeout. For streaming responses use HeaderTimeout alone; for bounded responses use AttemptTimeout, or both to fail fast on headers within a longer overall bound.

A panic in a function given to Do, Request.Do, upnet.Dialer or uphttp.RoundTripper releases the backend's lease and goes on.

A backend's address is one concept for every package: upstream.AddressOf takes a string value as it is and calls Address() on values that implement upstream.Addresser, such as a struct with a zone or labels. health.Watch, health.Attach, upnet.Dialer and uphttp.RoundTripper use it unless given a function of their own, and fail with upstream.ErrNoAddress for pools whose values have no address.

client := &http.Client{Transport: &uphttp.RoundTripper[string]{
	Pool:      pool,
	Transport: func(b *upstream.Backend[string]) http.RoundTripper { return cache.Get(b.ID()) },
	Key:       func(r *http.Request) string { return sessionCookie(r) },
}}

Balancers

Balancer	Behaviour
RoundRobin()	Cycles in order, ignores weights.
WeightedRoundRobin()	Smooth weighted round-robin (the default): exact proportions, picks spread through the cycle.
Random()	Weighted random, constant time.
PowerOfTwo()	Two random candidates, the one with fewer requests in flight per weight.
LeastConn()	Fewest requests in flight per weight, random tie-break; scans all.
ConsistentHash(cfg)	Hash ring keyed by PickKey/PickHash; optional bounded loads.

Consistent hashing places backends by ID, so every process with the same IDs maps a key to the same backend, and adding or removing one of n backends moves only about 1/n of the keys. With LoadFactor set, a backend already carrying that multiple of its fair share is passed over for the next one on the ring.

A Balancer builds a Picker from Candidate values and can be used without a pool. Implement the two interfaces for custom strategies. A candidate carries the backend's in-flight count and moving average latency, both read live at pick time, and Meta, which a pool sets to the *Backend[T]: a zone-aware or least-latency balancer reads what it needs from there. A pool builds one picker per tier and calls Build for a tier only when its candidates change, never two at once, and coalesces changes that arrive during a build into one more. A balancer whose build is costly also implements Rebuilder to update the previous picker instead; ConsistentHash does, so a member that goes down or comes back on a large ring costs a linear merge, not a new sort.

Tiers

Target.Priority puts backends in tiers; lower values are preferred and the default, 0, puts every backend in one tier. A pick goes to the most preferred tier that has an eligible backend and is balanced within that tier by the pool's balancer. It moves to the next tier when no backend of the tier can take it: all are down, ejected, offline, draining or half-open without a free trial slot, or a retry has tried them all. Distinct priorities make an active-backup pool.

Target.MaxInFlight caps a backend's leases in flight; a backend at its cap refuses picks, in panic mode too, and the pick goes to the other backends of its tier. When every backend of a tier that the pick may use is full, the pick fails with ErrSaturated, so that a backup takes traffic only when the tier above is down, not when it is busy; Config.Overflow lets such picks go on to the next tier instead, and they fail with ErrSaturated only if they found a tier full and no tier could take them.

ErrUnavailable means no backend can take the pick for reasons of health, drains or offline marks; ErrSaturated means backends refused it for capacity and calls for backpressure: it clears as leases are released. errors.Is(ErrSaturated, ErrUnavailable) holds, so code that only asks whether a pick failed for lack of backends needs one check.

pool := upstream.New[string](upstream.Config{Overflow: true})
err := pool.Sync([]upstream.Target[string]{
	{ID: "a", Value: "10.0.0.1:8080", MaxInFlight: 100},
	{ID: "b", Value: "10.0.0.2:8080", MaxInFlight: 100},
	{ID: "backup", Value: "10.0.0.9:8080", Priority: 1},
})

Health and eligibility

A backend takes new picks when it is

• not draining or removed,
• not marked offline by the caller with MarkOffline,
• not reported down by active health checks (HealthUnknown and HealthUp both take traffic), and
• not ejected by passive health checking. A half-open backend takes up to HalfOpenRequests trial requests at a time.

Each of the last three is a veto of its own, and none overrides another: MarkOnline lifts only the caller's mark, and a backend that checks report down stays out. A backend's State names the first that applies, in the order draining, offline, unhealthy, ejected, half-open.

MarkOffline is for what the caller knows better than any probe, such as a backend reached through a session that dropped. Like ReportCheck, it takes the *Backend and ignores calls for a member that has since been removed or added again, so a late call from an old session cannot take a new one out. The mark follows the member's ID across updates, emits EventStateChanged with the given cause (or ErrOffline), and rebuilds the view like any other change. A backend that comes back online ramps up under slow start.

When too few backends are eligible, Config.Fallback decides:

• FallbackNone (default): Pick returns ErrUnavailable when no member can take the request.
• FallbackAll: panic mode. Pick chooses among all members that are neither draining nor offline, ignoring health, on the grounds that a signal marking everything down is more often wrong than right; tiers still apply. The pool panics when no member is eligible, when fewer than Config.PanicThreshold percent of the members that are neither draining nor offline are, and for a pick that every eligible member refuses, such as half-open members out of trial slots after a whole-pool ejection. Outcomes still count. Offline marks are not health signals: panic mode never picks an offline member.

An empty pool returns ErrNoBackends; a closed one ErrClosed.

Passive health

OutlierConfig ejects a backend after ConsecutiveFailures failures in a row, or when FailurePercent of the requests in a sliding Window failed (once MinRequests were seen). The ejection lasts BaseEjection, doubling with each ejection that is not followed by a recovery, up to MaxEjection. Then the backend is half-open: HalfOpenSuccesses successful trials restore it and reset the backoff; a failed trial ejects it again. Each ejection is one event, however many failures follow. MaxEjectionPercent caps the share of the pool that may be ejected or half-open at once, so a fault that all backends share cannot eject them all; refused ejections are counted in OutlierStatus.Refused.

OutlierDetector is the same logic for a single target, usable without a pool: Admit before a request, Record its outcome after.

Configuration files

ParseBalancer and ParseFallback take the names that the built-in balancers and Fallback print, such as least-conn and all. State, Health, OutlierState, Outcome, EventKind, Fallback and health.Reason encode and decode as their names in JSON and text. A Balancer is an interface: a configuration names it and ParseBalancer builds it.

Pool.SetConfig applies a reloaded configuration to a running pool, keeping its members, counters, ejections and leases: a new balancer builds a new view, new outlier settings apply from the next request, disabling passive health checking clears ejections, and retry settings apply from the next retry. Checker.SetConfig does the same for active checks.

Active health

watcher, err := health.Watch(ctx, pool, nil, // nil: TCP connect check of each value
	health.Config[string]{Options: health.Options{Interval: 5 * time.Second, Rise: 2, Fall: 3}})
if err != nil {
	return err
}
defer watcher.Close()

Watch follows the pool's membership and reports each result with Pool.ReportCheck, matching reports to members by generation so that a re-added member never takes an old verdict. Give a target function to choose the address, a Probe (such as uphttp.Probe) or per-member options. Thresholds apply in the checker: before its first check a target is HealthUnknown; the first check decides at once; after that Rise consecutive passes bring it up and Fall consecutive failures take it down. Transitions are reported once each. Closing the watcher before the pool sets the members' health back to HealthUnknown, so no member stays out of rotation on a verdict nothing will revise.

Attach connects a pool to one group of a shared checker instead, for example one pool per tunnel. The checker then holds targets by address (TargetKey): an address that several pools or members use is checked once and reported to each of them, members with one ID but different addresses stay apart, and a member that joins an address already checked takes its verdict at once. Closing a pool, or calling the function Attach returns, ends its group. Attach fails with health.ErrClosed for a closed checker and Watch with the context's error once it is done. An empty network and tcp are the same network, so pools that spell it differently share one check.

Draining

Drain(id) stops new picks while the backend stays a member; Remove(id) also takes it out of the pool. Both return a *Drain whose Done channel is closed, and whose Wait returns, when the backend's last lease is released. Resume ends a drain early; the Drain then reports Canceled and Wait returns ErrDrainCanceled. Members that Sync removes drain the same way.

Slow start

With Config.SlowStart, a backend that joins or becomes healthy again starts with a small share of traffic that grows linearly to its full weight over the given duration.

Events and status

Subscribe delivers EventAdded, EventUpdated, EventRemoved, EventStateChanged and EventBalancerPanic in order, starting with the current members. A Balancer that panics while the pool builds its view does not crash the program or reach the caller that changed the pool: the pool keeps routing with its last view, reports a *BalancerPanicError with the panic value and stack in an EventBalancerPanic, and builds again on its next change. Callbacks run without any lock held and may call the pool. Status and Backend.Status return in-flight counts, outcome counters (successes, failures and neutral outcomes, which with the in-flight count make up every pick), the latest check result, the outlier state and the routing state of every backend. Available counts the members that are eligible for traffic, for readiness checks; it is zero in panic mode, when FallbackAll still lets picks through.

Health checks on their own

The checker works on any set of targets. Each target has a key of your choosing and may belong to groups; a key used by several groups is checked once and stops when its last group lets go of it.

checker := health.New[string, int](ctx, health.Config[string]{
	Options: health.Options{Interval: 10 * time.Second, Timeout: 3 * time.Second},
	OnChange: func(r health.Report[string]) {
		report(r.Key, r.Health, r.Reason, r.Latency) // one call per transition
	},
})
defer checker.Close()

// Apply a full list for tunnel 7: adds new targets, keeps unchanged ones
// with their state, removes the rest. Put and AddToGroup add one target.
checker.SyncGroup(7, map[string]health.Target{
	"10.0.0.5:22":  {Address: "10.0.0.5:22"},
	"db.lan:5432":  {Address: "db.lan:5432", Options: health.Options{Fall: 5}},
	"10.0.0.9:53":  {Network: "udp", Address: "10.0.0.9:53"}, // not checked
})

checker.RemoveGroup(7) // tunnel deleted: its checks stop at once

• TCP check (the default for tcp, tcp4, tcp6): connects and closes at once, without writing or reading application data. Host names are resolved by every check; the result names the address that answered. The latency covers the connection to that address only, not name resolution, and is measured on the checker's Clock, which also runs check timeouts: the probe's context carries the check's deadline. The normal close leaves a short-lived TIME_WAIT entry on the checking host; TCPOptions.Reset closes with a reset instead, which avoids that but may be logged by the target. TCPOptions.Dial plugs in your own dialer or resolver.
• HTTP check: uphttp.Probe(uphttp.Check{...}) as a target's Probe, with method, path, Host header, TLS server name (by default the Host header's host), expected status range, body match and your own client for other TLS settings. It runs only for targets that name it. The default client connects directly, ignoring proxy settings in the environment.
• UDP has no handshake to observe, so there is no generic UDP check. UDP targets without a Probe are not checked and stay HealthUnknown; give a Probe if you know the protocol.
• Results carry the health before and after the check (Changed compares them), the latency and its statistics, the answering address and a Reason (refused, timeout, unreachable for unreachable or down hosts and networks and unusable local addresses, dns, reset, response, other) derived from error values and system error numbers. Status adds the last latency, a moving average and the minimum and maximum over a recent window.
• Options (interval, unhealthy interval, timeout, jitter, rise, fall, disabled) can be set globally and per target, merged field by field before defaults apply, so a target's own interval also sets its default timeout and jitter, and changed at any time with Put, AddToGroup, SyncGroup or SetConfig; the next check is rescheduled at once. CheckNow checks a target ahead of schedule, for example after real traffic to it failed. A target whose checks stop goes back to HealthUnknown.
• Removal cancels the target's check in progress through its context, discards the result, stops its timer and frees its slot. Once Remove, RemoveFromGroup, RemoveGroup or SyncGroup returns, no callback for the removed target starts (one already running may finish, as with time.Timer.Stop); OnRemove reports its key and generation last. Adding the key again, or changing its network or address, starts a new generation; OnRemove reports the end of the old one too, and every report carries the generation so that buffered reports of an old target can be told apart.
• Callbacks run one at a time, in order, on a goroutine the checker owns, without locks held. They may call any method, Close included. A blocked callback holds back later callbacks but not checks; results that change nothing are coalesced per target meanwhile.

Config.Concurrency bounds how many checks run at once; a hanging target holds one slot and never delays the others.

HTTP transports

uphttp.TransportCache keeps one *http.Transport per key (for example per backend ID), dropping transports that have been idle for IdleTTL and the least recently used beyond MaxEntries. Dropping a transport closes its idle connections; give transports an IdleConnTimeout so that connections still in use when it was dropped close once they go idle. Keyed by backend ID, remove a transport on EventRemoved and on EventUpdated, which a pool emits only when a member's value, weight, priority or cap really changed.

Concurrency and lifetimes

• All types are safe for concurrent use.
• No lock is held while calling a balancer, a clock, a probe or a callback, so callbacks may call back into the pool or the checker.
• No method waits for a probe or a callback, so none deadlocks when called re-entrantly, Close included.
• The pool owns its ejection and slow-start timers, which Close stops. Pool.Done reports that the pool was closed.
• A pool rebuilds its view on the goroutine that changed it, without locks held; a change that arrives during a build is left to the building goroutine, so the method returns before its new members are visible to picks. A goroutine builds at most twice per change, its own and one for the changes that arrived meanwhile, and hands anything later to a short-lived goroutine of the pool, so a request whose release ejects a backend is not held up by a stream of changes. A Balancer panic is recovered on every build path and reported as EventBalancerPanic; changes that arrived during that build are still built, and otherwise the next change retries. Removals, drains and health verdicts take effect for picks at once regardless.
• The checker owns its timers, one goroutine per running check and one goroutine for callbacks. Close (or the end of its context) stops them and returns at once; Done is closed, and Wait returns, once every one of them has ended. Waiting on Done from inside a probe or callback waits for itself.
• A watcher started with Watch belongs to its pool and closes with it; a pool attached with Attach leaves its checker when it closes.
• Time comes from an injectable Clock, so behaviour over time can be tested without waiting.

Limits

• Weights range from 1 to MaxWeight (2³¹−1); zero means 1.
• Smooth weighted round-robin precomputes its cycle when the cycle times the number of backends is at most 2²²; beyond that, which takes extreme weight ratios, it picks at random by weight, lock-free, so proportions hold on average.
• Bounded-load consistent hashing reads every backend's in-flight count on each pick; retries (not first picks) on rings of more than 256 backends allocate.
• A Request allocates from the fifth attempt of one request on; Pool.Do does not allocate, Request.Do allocates once.
• Slow start works through weights within a tier, so it has no effect with RoundRobin, on a backend alone in its tier, or between tiers: traffic moves back to a recovered tier at once.
• The failure-rate window counts up to 2³²−1 requests per window and backend; the retry budget, up to 2³²−1 requests per window and pool.
• The checker compares targets by key; two groups that give different targets for one key share the latest one.

License

Apache License 2.0

Documentation

Overview

Package upstream is a backend pool for proxies, tunnels and load balancers: it picks a backend for each connection or request, tracks backend health actively and passively, and reports the state of every backend.

The pieces compose but also stand alone:

• A Balancer builds a Picker from Candidate values and needs no pool: RoundRobin, WeightedRoundRobin, Random, PowerOfTwo, LeastConn and ConsistentHash (a hash ring with optional bounded loads), or any implementation of the interface. Candidates carry in-flight counts, a moving average latency and the pool's backend, so custom balancers can route by load, latency or the backend's value; a Rebuilder updates its Picker incrementally.
• An OutlierDetector is passive health checking for one backend: consecutive failures or a failure rate eject it, with exponential backoff, and half-open trials let it back in.
• Package health runs active health checks on any set of targets and drives pools with them; package uphttp adds an HTTP check, a per-backend transport cache and an http.RoundTripper over a pool; package upnet dials the backends of a pool. All three find a backend's address the same way, with AddressOf: a string value, or a value that implements Addresser.
• A Pool combines them: it holds the members, routes picks through the balancer to eligible backends, counts requests in flight through a Lease, feeds outcomes to an outlier detector per backend, takes active health reports with Pool.ReportCheck, drains and removes backends gracefully, and announces every change as an Event.

Picking

Pick, PickKey and PickHash return a Lease, which must be released exactly once with the request's Outcome; Lease.Report records the outcome early for leases that live long, such as connections, and Lease.Observe feeds the backend's latency. Pool.Do picks, runs a function and releases, retrying failures on other backends as RetryConfig allows, within a retry budget; a Request does the same for leases that outlive a function. Picks are lock-free and do not allocate: they read an immutable routing view that changes rebuild and publish atomically, so updates never disturb picks in flight. Updates that change no routing rebuild nothing, and rebuilds that overlap are coalesced. Pool.SetConfig changes the balancer, health, retry and fallback settings of a running pool.

Target.Priority puts members in tiers: picks go to the most preferred tier with a member that can take them, balanced within it, which makes active-backup pools. Target.MaxInFlight caps a member's leases, and Config.Overflow lets picks go on to the next tier when a tier is full.

Health

A backend is eligible when it is not draining, not marked offline with Pool.MarkOffline, not reported unhealthy by active checks and not ejected by passive checks; each of these vetoes picks on its own. A half-open backend takes up to OutlierConfig.HalfOpenRequests trial requests at a time. When no backend is eligible, Pick fails with ErrUnavailable, or, with FallbackAll, picks among all members that are neither draining nor offline; panic mode also covers too few eligible members (Config.PanicThreshold) and eligible members that all refuse a pick. OutlierConfig.MaxEjectionPercent keeps passive health checking from ejecting too much of a pool.

State changes are edge-triggered: a backend that keeps failing is reported down once.

Concurrency

All types are safe for concurrent use. No lock is held while calling a Balancer, a Clock, a probe or a subscriber, so callbacks may call back into the pool. The pool rebuilds its view on the goroutine that changed it, at most twice per change; a rebuild left over after that runs on a short-lived goroutine of the pool, which builds nothing once Close has begun. A Balancer that panics during a build does not reach any caller: the pool keeps its last view, emits EventBalancerPanic and builds again on its next change. The only timers the pool owns are those ending ejections and those stepping slow starts, which Close stops. No Close in this module waits; types that own goroutines have a Done channel for callers that need to wait for them.

Example

package main

import (
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	pool := upstream.New[string](upstream.Config{
		Balancer: upstream.PowerOfTwo(),
		Outlier:  upstream.OutlierConfig{ConsecutiveFailures: 5},
	})
	defer pool.Close()
	err := pool.Sync([]upstream.Target[string]{
		{ID: "a", Value: "10.0.0.1:8080", Weight: 2},
		{ID: "b", Value: "10.0.0.2:8080"},
	})
	if err != nil {
		fmt.Println(err)
		return
	}

	lease, err := pool.Pick()
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Println("in flight:", lease.Backend().InFlight())
	// Connect to lease.Value() here, then report how it went.
	lease.Release(upstream.Success)
	fmt.Println("in flight:", pool.Status()[0].InFlight+pool.Status()[1].InFlight)
}

Output:
in flight: 1
in flight: 0

Index

• Constants
• Variables
• func AddressOf[T any](v T) (addr string, ok bool)
• func Addressable[T any]() bool
• func WithOutcome(err error, o Outcome) error
• type Addresser
• type Backend
  • func (b *Backend[T]) ID() string
  • func (b *Backend[T]) InFlight() int64
  • func (b *Backend[T]) Latency() time.Duration
  • func (b *Backend[T]) MaxInFlight() int
  • func (b *Backend[T]) Priority() int
  • func (b *Backend[T]) State() State
  • func (b *Backend[T]) Status() Status
  • func (b *Backend[T]) Value() T
  • func (b *Backend[T]) Weight() int
• type Balancer
  • func ConsistentHash(cfg HashConfig) Balancer
  • func LeastConn() Balancer
  • func ParseBalancer(name string) (Balancer, error)
  • func PowerOfTwo() Balancer
  • func Random() Balancer
  • func RoundRobin() Balancer
  • func WeightedRoundRobin() Balancer
• type BalancerPanicError
  • func (e *BalancerPanicError) Error() string
  • func (e *BalancerPanicError) Unwrap() error
• type Candidate
• type CheckResult
• type Clock
  • func SystemClock() Clock
• type Config
• type Drain
  • func (d *Drain) Canceled() bool
  • func (d *Drain) Done() <-chan struct{}
  • func (d *Drain) Wait(ctx context.Context) error
• type Event
• type EventKind
  • func (k EventKind) MarshalText() ([]byte, error)
  • func (k EventKind) String() string
  • func (k *EventKind) UnmarshalText(b []byte) (err error)
• type Fallback
  • func ParseFallback(s string) (Fallback, error)
  • func (f Fallback) MarshalText() ([]byte, error)
  • func (f Fallback) String() string
  • func (f *Fallback) UnmarshalText(b []byte) (err error)
• type HashConfig
• type Health
  • func (h Health) MarshalText() ([]byte, error)
  • func (h Health) String() string
  • func (h *Health) UnmarshalText(b []byte) (err error)
• type Lease
  • func (l *Lease[T]) Backend() *Backend[T]
  • func (l *Lease[T]) Observe(d time.Duration)
  • func (l *Lease[T]) Release(o Outcome)
  • func (l *Lease[T]) Report(o Outcome)
  • func (l *Lease[T]) Take() Lease[T]
  • func (l *Lease[T]) Value() T
• type Outcome
  • func OutcomeOf(err error) Outcome
  • func (o Outcome) MarshalText() ([]byte, error)
  • func (o Outcome) String() string
  • func (o *Outcome) UnmarshalText(b []byte) (err error)
• type OutlierConfig
  • func (c OutlierConfig) Enabled() bool
• type OutlierDetector
  • func NewOutlierDetector(cfg OutlierConfig, clock Clock, onChange func()) *OutlierDetector
  • func (d *OutlierDetector) Admit() (Ticket, bool)
  • func (d *OutlierDetector) Close()
  • func (d *OutlierDetector) Record(t Ticket, o Outcome)
  • func (d *OutlierDetector) Reset()
  • func (d *OutlierDetector) SetConfig(cfg OutlierConfig)
  • func (d *OutlierDetector) State() OutlierState
  • func (d *OutlierDetector) Status() OutlierStatus
• type OutlierState
  • func (s OutlierState) MarshalText() ([]byte, error)
  • func (s OutlierState) String() string
  • func (s *OutlierState) UnmarshalText(b []byte) (err error)
• type OutlierStatus
• type Picker
• type Pool
  • func New[T any](cfg Config) *Pool[T]
  • func (p *Pool[T]) Available() int
  • func (p *Pool[T]) Backend(id string) (*Backend[T], bool)
  • func (p *Pool[T]) Backends() []*Backend[T]
  • func (p *Pool[T]) Clock() Clock
  • func (p *Pool[T]) Close()
  • func (p *Pool[T]) Do(ctx context.Context, fn func(*Backend[T]) error) error
  • func (p *Pool[T]) DoKey(ctx context.Context, key string, fn func(*Backend[T]) error) error
  • func (p *Pool[T]) Done() <-chan struct{}
  • func (p *Pool[T]) Drain(id string) (*Drain, bool)
  • func (p *Pool[T]) Len() int
  • func (p *Pool[T]) MarkOffline(b *Backend[T], cause error) bool
  • func (p *Pool[T]) MarkOnline(b *Backend[T]) bool
  • func (p *Pool[T]) Pick() (Lease[T], error)
  • func (p *Pool[T]) PickHash(hash uint64) (Lease[T], error)
  • func (p *Pool[T]) PickKey(key string) (Lease[T], error)
  • func (p *Pool[T]) Put(t Target[T]) error
  • func (p *Pool[T]) Remove(id string) (*Drain, bool)
  • func (p *Pool[T]) ReportCheck(b *Backend[T], h Health, r CheckResult) bool
  • func (p *Pool[T]) Request(key string) Request[T]
  • func (p *Pool[T]) ResetOutlier(id string) bool
  • func (p *Pool[T]) Resume(id string) bool
  • func (p *Pool[T]) SetConfig(cfg Config) error
  • func (p *Pool[T]) Status() []Status
  • func (p *Pool[T]) Subscribe(fn func(Event[T])) (cancel func())
  • func (p *Pool[T]) Sync(targets []Target[T]) error
• type Rebuilder
• type Request
  • func (r *Request[T]) Do(ctx context.Context, fn func(l *Lease[T]) (retry bool, err error)) error
  • func (r *Request[T]) Next() (Lease[T], error)
  • func (r *Request[T]) Tried() int
• type RetryConfig
• type State
  • func (s State) MarshalText() ([]byte, error)
  • func (s State) String() string
  • func (s *State) UnmarshalText(b []byte) (err error)
• type Status
• type Target
• type Ticket

Examples

• Package
• Candidate
• ConsistentHash
• Lease.Report
• OutlierDetector
• ParseBalancer
• Pool.Do
• Pool.MarkOffline
• Pool.PickKey
• Pool.Remove
• Pool.Subscribe
• Request
• Request.Do
• Target (Priority)

Constants

const MaxWeight = math.MaxInt32

MaxWeight is the largest Target weight. It keeps the products of loads and weights that balancers compare far from overflowing.

Variables

var (
	// ErrConsecutiveFailures means the backend failed
	// OutlierConfig.ConsecutiveFailures requests in a row.
	ErrConsecutiveFailures = errors.New("upstream: too many consecutive failures")

	// ErrFailureRate means the share of failed requests in the window
	// reached OutlierConfig.FailurePercent.
	ErrFailureRate = errors.New("upstream: failure rate too high")

	// ErrTrialFailed means a half-open trial request failed.
	ErrTrialFailed = errors.New("upstream: half-open trial failed")
)

Causes recorded when an OutlierDetector ejects its backend.

var (
	// ErrNoBackends means the pool has no members.
	ErrNoBackends = errors.New("upstream: no backends")

	// ErrUnavailable means no member can take the pick now: every one is
	// unhealthy, ejected, draining, offline, out of half-open trial slots
	// or, for a retry, already tried. ErrSaturated is an ErrUnavailable
	// whose cause is capacity.
	ErrUnavailable = errors.New("upstream: no backend available")

	// ErrSaturated means that members that could take the pick refused it
	// because they were at their Target.MaxInFlight: the pick reached a
	// full tier and Config.Overflow is off, or every tier it could go on to
	// was full or unavailable too. It clears as leases are released, so it
	// calls for backpressure rather than for an alarm.
	// errors.Is(ErrSaturated, ErrUnavailable) holds.
	ErrSaturated = fmt.Errorf("upstream: backends at capacity: %w", ErrUnavailable)

	// ErrClosed means the pool was closed.
	ErrClosed = errors.New("upstream: pool closed")

	// ErrInvalidTarget means a Target has an empty or duplicate ID or a
	// weight out of range.
	ErrInvalidTarget = errors.New("upstream: invalid target")
)

Errors returned by Pool.

var (
	// ErrRetryLimit means the request has tried RetryConfig.Attempts
	// backends.
	ErrRetryLimit = errors.New("upstream: retry limit reached")

	// ErrRetryBudget means the pool's retry budget is spent.
	ErrRetryBudget = errors.New("upstream: retry budget exhausted")
)

Errors returned by Request.Next when it declines a retry.

var ErrDrainCanceled = errors.New("upstream: drain canceled")

ErrDrainCanceled is returned by Drain.Wait when Resume ended the drain before the backend's leases were all released.

var ErrNoAddress = errors.New("upstream: backend value has no address")

ErrNoAddress means a backend value has no network address: it is neither a string nor an Addresser, and no function was given to derive one. Packages health, upnet and uphttp return it for such pools.

var ErrOffline = errors.New("upstream: backend offline")

ErrOffline is the Err of an EventStateChanged to StateOffline when Pool.MarkOffline was given no cause.

Functions

func AddressOf

func AddressOf[T any](v T) (addr string, ok bool)

AddressOf returns the address of a backend value: the value itself if its type is a string type, or the result of its Address method if it is an Addresser. ok is false for other values and for a nil Addresser.

func Addressable

func Addressable[T any]() bool

Addressable reports whether every value of type T has an address for AddressOf: T is a string type, or T implements Addresser. Packages that derive addresses check it once, up front, and fail with ErrNoAddress.

func WithOutcome

func WithOutcome(err error, o Outcome) error

WithOutcome returns an error that wraps err and that OutcomeOf maps to o, so that a function run by Pool.Do can say how its error reflects on the backend: for example Neutral for a request the client got wrong, which is neither counted against the backend nor retried. It returns nil if err is nil.

Types

type Addresser

type Addresser interface {
	// Address returns the backend's address, as for net.Dial.
	Address() string
}

An Addresser is a backend value that knows its network address, usually a host and port. Pools whose values carry more than an address, such as a struct with a zone or labels, implement it so that health checks, dialers and HTTP transports find the address without a function of their own.

type Backend

type Backend[T any] struct {
	// contains filtered or unexported fields
}

A Backend is one member of a Pool: a Target as the pool currently routes to it. Changing a target's weight or value in the pool creates a new Backend that shares the old one's counters and health state, so leases keep the version they were issued with; an update that changes neither keeps the Backend.

func (*Backend[T]) ID

func (b *Backend[T]) ID() string

ID returns the backend's ID.

func (*Backend[T]) InFlight

func (b *Backend[T]) InFlight() int64

InFlight returns the number of leases on the backend not yet released.

func (*Backend[T]) Latency

func (b *Backend[T]) Latency() time.Duration

Latency returns the moving average of the latencies reported with Lease.Observe, or zero before the first one.

func (*Backend[T]) MaxInFlight

func (b *Backend[T]) MaxInFlight() int

MaxInFlight returns the backend's cap on leases in flight, or zero if it has none.

func (*Backend[T]) Priority

func (b *Backend[T]) Priority() int

Priority returns the backend's tier: lower values are preferred.

func (*Backend[T]) State

func (b *Backend[T]) State() State

State returns the backend's routing state.

func (*Backend[T]) Status

func (b *Backend[T]) Status() Status

Status returns a snapshot of the backend.

func (*Backend[T]) Value

func (b *Backend[T]) Value() T

Value returns the value of the target the backend was created from.

func (*Backend[T]) Weight

func (b *Backend[T]) Weight() int

Weight returns the backend's weight, which is at least 1.

type Balancer

type Balancer interface {
	// Build returns a Picker for candidates, which is never empty. The
	// Picker may keep the slice; the caller does not modify it afterwards.
	Build(candidates []Candidate) Picker
}

A Balancer builds Pickers for sets of candidates. A Pool builds one Picker per tier (Target.Priority) and calls Build for a tier when its candidates change: when members join, leave or change weight, go down or come back, are ejected or restored, and at the steps of a slow start; a tier whose candidates did not change keeps its Picker. Calls never overlap for one pool and never happen while the pool holds a lock; picks use the previous Picker until Build returns. If Build or Rebuild panics, the pool recovers, keeps the previous Picker, reports an EventBalancerPanic and calls the Balancer again on its next change. A Balancer that can update a Picker more cheaply than it builds one also implements Rebuilder.

Balancers can be used without a Pool: build a Picker from candidates and call Pick.

func ConsistentHash

func ConsistentHash(cfg HashConfig) Balancer

ConsistentHash returns a Balancer that maps keys to candidates on a hash ring.

Each candidate owns ring points derived from its ID, in number proportional to its weight, and a key belongs to the first point at or after the key's hash. Adding or removing a candidate therefore moves only the keys of the ring segments it gains or loses, about 1/n of all keys, and every process that uses the same IDs maps a key to the same candidate. Retries walk the ring to the next distinct candidates, which are the candidates the key would move to if the earlier ones left.

Pick hashes the key it is given again, so any 64-bit key works. Picks do not allocate, except retries (attempts above 0) on rings of more than 256 candidates, which need a scratch bitset. The Balancer is a Rebuilder: when members come and go, a Pool updates the ring instead of building it again.

Example

package main

import (
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	ring := upstream.ConsistentHash(upstream.HashConfig{}).Build([]upstream.Candidate{
		{ID: "cache-1"}, {ID: "cache-2"}, {ID: "cache-3"},
	})
	key := uint64(12345)
	primary := ring.Pick(key, 0)
	backup := ring.Pick(key, 1) // where the key goes if the primary leaves
	fmt.Println(primary != backup)
}

Output:
true

func LeastConn

func LeastConn() Balancer

LeastConn returns a Balancer that picks the candidate with the fewest requests in flight relative to its weight, breaking ties at random. Each pick scans every candidate. A retry (attempt above 0) picks at random so that a candidate that cannot take the request is not chosen again.

func ParseBalancer

func ParseBalancer(name string) (Balancer, error)

ParseBalancer returns the built-in Balancer with the given name, as its String method writes it: "round-robin", "weighted-round-robin", "random", "power-of-two", "least-conn" or "consistent-hash" (with the default HashConfig). It suits configuration files; build a ConsistentHash with options in code.

Example

package main

import (
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	// Balancers and fallbacks by name, as a configuration file gives them.
	b, err := upstream.ParseBalancer("least-conn")
	if err != nil {
		fmt.Println(err)
		return
	}
	f, _ := upstream.ParseFallback("all")
	pool := upstream.New[string](upstream.Config{Balancer: b, Fallback: f})
	defer pool.Close()
	fmt.Println(b, f)
}

Output:
least-conn all

func PowerOfTwo

func PowerOfTwo() Balancer

PowerOfTwo returns a Balancer that samples two distinct candidates at random and picks the one with fewer requests in flight relative to its weight. It approaches least-connections balancing at constant cost and avoids the herding that strict least-connections shows when many clients see the same idle backend.

func Random

func Random() Balancer

Random returns a Balancer that picks candidates at random in proportion to their weights, using the alias method so each pick takes constant time.

func RoundRobin

func RoundRobin() Balancer

RoundRobin returns a Balancer that cycles through the candidates in order and ignores weights. Each Picker starts at a random position so that many pools do not hit the same backend in step.

func WeightedRoundRobin

func WeightedRoundRobin() Balancer

WeightedRoundRobin returns a Balancer that implements smooth weighted round-robin: over each cycle every candidate is picked in proportion to its weight, and its picks are spread through the cycle instead of bunched together. For weights 5, 1, 1 a cycle is a a b a c a a.

The cycle is precomputed, which makes Pick a single atomic add, when it is short enough: the sum of the weights divided by their greatest common divisor, times the number of candidates, at most 1<<22. Longer cycles, which take extreme weight ratios, fall back to weighted random picks: proportions then hold on average rather than exactly in every cycle, and picks stay lock-free.

type BalancerPanicError

type BalancerPanicError struct {
	// Value is the value passed to panic.
	Value any

	// Stack is the stack of the panicking goroutine, as runtime/debug.Stack
	// formats it.
	Stack []byte
}

A BalancerPanicError is the Err of an EventBalancerPanic: what a Balancer's Build or Rebuild panicked with, and where.

func (*BalancerPanicError) Error

func (e *BalancerPanicError) Error() string

func (*BalancerPanicError) Unwrap

func (e *BalancerPanicError) Unwrap() error

Unwrap returns Value if it is an error, such as a runtime.Error.

type Candidate

type Candidate struct {
	// InFlight counts the backend's requests in flight. Load-aware
	// balancers read it; nil means no load information, read as zero.
	InFlight *atomic.Int64

	// Latency is the backend's moving average latency in nanoseconds, as
	// reported with Lease.Observe; zero means no sample yet, and nil no
	// latency information. Latency-aware balancers read it at pick time.
	Latency *atomic.Int64

	// Meta is whatever the caller attaches to the candidate. A Pool sets
	// it to the *Backend[T], so that a balancer can route by the
	// backend's value, such as its zone or labels, or by its Status.
	Meta any

	// ID identifies the backend. Consistent hashing places backends by ID,
	// so equal IDs map keys to the same place in every process.
	ID string

	// Weight is the backend's relative capacity. Values below 1 mean 1.
	Weight int
}

A Candidate is a balancer's view of one backend.

Example

// Balancers work without a pool.
var loadA, loadB atomic.Int64
loadA.Store(5)
p := LeastConn().Build([]Candidate{
	{ID: "a", Weight: 1, InFlight: &loadA},
	{ID: "b", Weight: 1, InFlight: &loadB},
})
fmt.Println(p.Pick(0, 0))

Output:
1

type CheckResult

type CheckResult struct {
	// Time is when the check started.
	Time time.Time

	// Err is why the check failed, or nil.
	Err error

	// Addr is the address that answered, if the check reported it, such as
	// the IP address a host name resolved to.
	Addr string

	// Latency is the time the target took to answer.
	Latency time.Duration
}

CheckResult is the result of one active health check. Package health embeds it in its own results, so a pool keeps everything a check saw except the checker's thresholds.

type Clock

type Clock interface {
	// Now returns the current time.
	Now() time.Time

	// AfterFunc arranges for f to run in its own goroutine once d has
	// elapsed, and returns a function that cancels the call.
	AfterFunc(d time.Duration, f func()) (stop func() bool)

	// WithTimeout returns a copy of ctx that ends d from now on this Clock.
	WithTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc)
}

A Clock tells the time and schedules callbacks. The pool, the outlier detector and the health checker read time only through a Clock, so tests and simulations can substitute a fake one.

Implementations must be safe for concurrent use. AfterFunc must call f in its own goroutine, as time.AfterFunc does; f may start before AfterFunc returns, for example when d is not positive, and callers in this module allow for that. The returned stop function reports whether it prevented f from running, like time.Timer.Stop.

WithTimeout behaves like context.WithTimeout on the Clock's time: the returned context's Deadline reports Now()+d (or the parent's deadline if that is sooner), and once the Clock reaches it the context is canceled with Err and Cause both context.DeadlineExceeded. Every timeout in this module comes from it, so probes and transports see a real deadline.

func SystemClock

func SystemClock() Clock

SystemClock returns the Clock backed by package time.

type Config

type Config struct {
	// Balancer chooses backends. Nil means WeightedRoundRobin.
	Balancer Balancer

	// Clock is the time source. Nil means SystemClock.
	Clock Clock

	// Equal reports whether two target values mean the same backend, so
	// that Sync and Put keep a member whose weight and value did not
	// change: same Backend, no EventUpdated, no rebuild. Nil compares
	// values with ==, through reflection, when they are comparable, and
	// treats other values, such as slices, as always changed.
	Equal func(a, b any) bool

	// Outlier configures passive health checking, which is disabled by
	// default.
	Outlier OutlierConfig

	// Retry configures how Do, DoKey and Request retry on other
	// backends. There are no retries by default.
	Retry RetryConfig

	// SlowStart, if positive, ramps up the traffic of a backend that joins
	// the pool or becomes healthy again: its weight grows linearly from a
	// small fraction to its full value over SlowStart, so a cold backend is
	// not flooded. The pool rebuilds its view 16 times over the ramp. With
	// ConsistentHash the ramp moves keys to the backend gradually. Slow
	// start works through weights within a tier, so it has no effect with
	// RoundRobin, which ignores weights, on a backend alone in its tier, or
	// between tiers.
	SlowStart time.Duration

	// PanicThreshold, from 1 to 100, makes a pool with FallbackAll panic
	// while fewer than this percentage of its members that are not
	// draining are eligible, instead of only when none is: a few survivors
	// would otherwise take the whole load and fail in turn. Zero panics
	// only when no member is eligible.
	PanicThreshold int

	// Fallback says what to do when too few backends are eligible.
	Fallback Fallback

	// Overflow lets a pick move to the next tier when every member of its
	// tier that it may use is at its Target.MaxInFlight. Without Overflow
	// such a pick fails with ErrSaturated, so that a lower tier, such as
	// a costly backup, takes traffic only when the tier above is down, not
	// when it is busy. Members that are down, ejected, out of half-open
	// trial slots or already tried by the request always pass the pick on.
	Overflow bool
}

Config configures a Pool. The zero value is a pool with smooth weighted round-robin, no passive health checking, no slow start, no retries and FallbackNone. Pool.SetConfig changes everything but the Clock while the pool runs.

type Drain

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

A Drain follows a backend that takes no new picks until its leases in flight are released.

func (*Drain) Canceled

func (d *Drain) Canceled() bool

Canceled reports whether Resume ended the drain before the backend's leases were all released. It is meaningful once Done is closed, and false before.

func (*Drain) Done

func (d *Drain) Done() <-chan struct{}

Done returns a channel that is closed when the drain ends: when the backend's last lease is released, or when Resume cancels the drain.

func (*Drain) Wait

func (d *Drain) Wait(ctx context.Context) error

Wait waits until the drain ends. It returns nil once the backend's last lease is released, ErrDrainCanceled if Resume ended the drain first, and ctx's error if ctx is done first.

type Event

type Event[T any] struct {
	// Err is the cause of an EventStateChanged to StateUnhealthy (the last
	// check error) or to StateEjected or StateHalfOpen (the ejection
	// cause), the *BalancerPanicError of an EventBalancerPanic, and nil
	// otherwise.
	Err error

	// Backend is the member concerned; nil for EventBalancerPanic.
	Backend *Backend[T]

	// Kind says what happened.
	Kind EventKind

	// From and To are the backend's states before and after the event.
	From, To State
}

An Event reports a change in a Pool. State changes are edge-triggered: a backend that keeps failing produces one event when it goes down, not one per failure.

type EventKind

type EventKind uint8

EventKind says what an Event reports.

const (
	// EventAdded reports a new member.
	EventAdded EventKind = iota

	// EventUpdated reports a member replaced with a new weight or value;
	// Backend is the new version. Updates that change neither emit none.
	EventUpdated

	// EventRemoved reports a member that left the pool.
	EventRemoved

	// EventStateChanged reports a change of a member's State.
	EventStateChanged

	// EventBalancerPanic reports that the Balancer panicked while the pool
	// built its view. Err is the *BalancerPanicError; Backend is nil. The
	// pool keeps routing with the view it published before, and builds
	// again on its next change.
	EventBalancerPanic
)

func (EventKind) MarshalText

func (k EventKind) MarshalText() ([]byte, error)

MarshalText encodes the kind as its name.

func (EventKind) String

func (k EventKind) String() string

String returns the name of the kind.

func (*EventKind) UnmarshalText

func (k *EventKind) UnmarshalText(b []byte) (err error)

UnmarshalText decodes a kind from its name.

type Fallback

type Fallback uint8

Fallback says what a Pool does when too few backends are eligible.

const (
	// FallbackNone makes Pick fail with ErrUnavailable.
	FallbackNone Fallback = iota

	// FallbackAll enters panic mode: Pick chooses among all members that
	// are not draining, ignoring active and passive health. A health signal
	// that marks every backend down is more often wrong than right, and
	// degraded service beats none. The pool panics when no member is
	// eligible, when fewer than Config.PanicThreshold percent of the
	// members that are not draining are eligible, and, for a single pick,
	// when every eligible member refuses it, as half-open members out of
	// trial slots do. Leases picked in panic mode still report outcomes.
	FallbackAll
)

func ParseFallback

func ParseFallback(s string) (Fallback, error)

ParseFallback returns the Fallback named s: "none" or "all".

func (Fallback) MarshalText

func (f Fallback) MarshalText() ([]byte, error)

MarshalText encodes the fallback as its name.

func (Fallback) String

func (f Fallback) String() string

String returns the name of the fallback, as ParseFallback accepts it.

func (*Fallback) UnmarshalText

func (f *Fallback) UnmarshalText(b []byte) (err error)

UnmarshalText decodes a fallback from its name, as ParseFallback does.

type HashConfig

type HashConfig struct {
	// VirtualNodes is the average number of ring points per candidate. More
	// points spread keys more evenly at the cost of memory and build time.
	// The default is 160.
	VirtualNodes int

	// LoadFactor enables bounded loads when above zero: a candidate that
	// already carries LoadFactor times its weighted share of the requests
	// in flight is passed over for the next one on the ring. Values below
	// 1 mean 1. Bounded loads keep a hot key from overloading one backend
	// but cost a scan of the in-flight counts on every pick.
	LoadFactor float64
}

HashConfig configures ConsistentHash.

type Health

type Health uint8

Health is the verdict of active health checking on a backend.

const (
	// HealthUnknown means no check has decided yet, or checks are not
	// running for the backend. Unknown backends take traffic.
	HealthUnknown Health = iota

	// HealthUp means checks pass.
	HealthUp

	// HealthDown means checks fail; the backend takes no traffic.
	HealthDown
)

func (Health) MarshalText

func (h Health) MarshalText() ([]byte, error)

MarshalText encodes the verdict as its name.

func (Health) String

func (h Health) String() string

String returns the name of the verdict.

func (*Health) UnmarshalText

func (h *Health) UnmarshalText(b []byte) (err error)

UnmarshalText decodes a verdict from its name.

type Lease

type Lease[T any] struct {
	// contains filtered or unexported fields
}

A Lease is a backend picked for one request or connection. It holds one of the backend's in-flight slots until Release returns it with the request's outcome.

Every lease returned without error must be released exactly once, or the backend's in-flight count stays raised and a drain never completes; Pool.Do, Pool.DoKey and Request.Do do this for the caller. Release zeroes the lease, so releasing the same variable again does nothing. Copies of a lease share one slot: release only one of them. Making every copy idempotent would need state shared by the copies, one allocation per pick, so the pool instead detects what it can without cost: releasing a copy after another copy has been released panics if the backend has no slot left to return, like a negative sync.WaitGroup counter; if other leases are in flight, the extra release is not detected and frees one of their slots.

func (*Lease[T]) Backend

func (l *Lease[T]) Backend() *Backend[T]

Backend returns the leased backend, or nil after Release.

func (*Lease[T]) Observe

func (l *Lease[T]) Observe(d time.Duration)

Observe records a latency sample of the backend, such as the time to a response's first byte or to an established connection, for latency-aware balancers: it feeds the moving average that Candidate.Latency, Backend.Latency and Status.Latency report. Observe may be called any number of times before Release, and does nothing on a released or zero Lease.

func (*Lease[T]) Release

func (l *Lease[T]) Release(o Outcome)

Release returns the lease's slot and records the outcome for passive health checking, unless Report already recorded one. It does nothing on a released or zero Lease. If the outcome ejects or restores the backend, Release also rebuilds the pool's view and may deliver events to subscribers on the calling goroutine; the slot is returned first, so a panicking subscriber cannot leak it.

func (*Lease[T]) Report

func (l *Lease[T]) Report(o Outcome)

Report records the outcome for passive health checking now, while the lease keeps its in-flight slot. Use it when the outcome is known long before the lease ends, as for a connection that was established and then stays open: a half-open backend's trial ends with the report instead of with the connection. Only the first Report counts; Release then returns the slot and ignores its outcome. Report does nothing on a released or zero Lease.

Example

package main

import (
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	pool := upstream.New[string](upstream.Config{Balancer: upstream.LeastConn()})
	defer pool.Close()
	_ = pool.Sync([]upstream.Target[string]{{ID: "relay", Value: "10.0.0.1:9000"}})

	lease, _ := pool.Pick()
	// The connection is up: report the outcome now and keep the in-flight
	// slot until the connection closes.
	lease.Report(upstream.Success)
	s := pool.Status()[0]
	fmt.Println(s.InFlight, s.Successes)
	lease.Release(upstream.Neutral) // the reported outcome stands
	s = pool.Status()[0]
	fmt.Println(s.InFlight, s.Successes, s.Neutrals)
}

Output:
1 1
0 1 0

func (*Lease[T]) Take

func (l *Lease[T]) Take() Lease[T]

Take returns the lease and leaves l zero, handing the lease to a new owner, such as a connection that releases it when closed. Releasing l afterwards does nothing.

func (*Lease[T]) Value

func (l *Lease[T]) Value() T

Value returns the leased backend's value, or the zero value after Release.

type Outcome

type Outcome uint8

Outcome classifies a finished request or connection for passive health checking.

const (
	// Success means the backend served the request. It resets the
	// consecutive-failure count.
	Success Outcome = iota

	// Failure means the backend failed the request: it refused the
	// connection, timed out or answered with an error the caller blames on
	// the backend.
	Failure

	// Neutral means the result says nothing about the backend, for example
	// because the client went away. It counts towards neither successes nor
	// failures.
	Neutral
)

func OutcomeOf

func OutcomeOf(err error) Outcome

OutcomeOf maps an error to an Outcome: nil is Success, an error from WithOutcome has the outcome given there, an error wrapping context.Canceled is Neutral, and any other error is Failure. context.DeadlineExceeded is a Failure because a backend that does not answer in time is the usual cause.

func (Outcome) MarshalText

func (o Outcome) MarshalText() ([]byte, error)

MarshalText encodes the outcome as its name.

func (Outcome) String

func (o Outcome) String() string

String returns the name of the outcome.

func (*Outcome) UnmarshalText

func (o *Outcome) UnmarshalText(b []byte) (err error)

UnmarshalText decodes an outcome from its name.

type OutlierConfig

type OutlierConfig struct {
	// Window is the span over which FailurePercent is measured. The
	// default is 10s. The rate is a sliding estimate over the current and
	// the previous window.
	Window time.Duration

	// BaseEjection is the first ejection time. The default is 30s.
	BaseEjection time.Duration

	// MaxEjection caps the ejection time. The default is 5m; a value below
	// BaseEjection is raised to it.
	MaxEjection time.Duration

	// ConsecutiveFailures ejects the backend after this many failures in a
	// row. Zero disables the rule.
	ConsecutiveFailures int

	// FailurePercent ejects the backend when at least this percentage of
	// the requests in Window failed, once MinRequests requests have been
	// seen. Zero disables the rule; values above 100 mean 100.
	FailurePercent int

	// MinRequests is the request volume the failure-rate rule needs before
	// it judges the backend. The default is 20.
	MinRequests int

	// HalfOpenRequests is the number of trial requests a half-open backend
	// admits at the same time. The default is 1.
	HalfOpenRequests int

	// HalfOpenSuccesses is the number of successful trials that restore a
	// half-open backend. The default is 1.
	HalfOpenSuccesses int

	// MaxEjectionPercent, from 1 to 100, caps the share of a pool's members
	// that passive health checking may hold ejected or half-open at once,
	// so that a fault shared by every backend, such as a database they all
	// depend on, cannot eject the whole pool. One member may always be
	// ejected. A failure that would eject a member beyond the cap leaves it
	// in rotation and is counted in OutlierStatus.Refused. Zero means no
	// cap. It applies to the detectors of a Pool, not to an
	// OutlierDetector on its own.
	MaxEjectionPercent int
}

OutlierConfig configures passive health checking, which judges a backend by the outcomes of real traffic. The zero value disables it; set ConsecutiveFailures, FailurePercent or both to enable it. Zero durations and counts take the defaults documented on each field.

An ejected backend receives no traffic for an ejection time that starts at BaseEjection and doubles with every ejection that is not followed by a recovery, up to MaxEjection. When the time is up the backend becomes half-open: it admits at most HalfOpenRequests concurrent trial requests. HalfOpenSuccesses successful trials restore it and reset the ejection time; one failed trial ejects it again.

func (OutlierConfig) Enabled

func (c OutlierConfig) Enabled() bool

Enabled reports whether the configuration enables passive health checking.

type OutlierDetector

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

An OutlierDetector tracks the passive health of one backend. Callers ask it for admission before each request and report each request's outcome afterwards. It is safe for concurrent use. Admit and Record are lock-free apart from state transitions, which are rare.

The detector owns one timer, which ends an ejection. Close stops it.

Example

package main

import (
	"errors"
	"fmt"
	"net"

	"github.com/djylb/upstream"
)

func main() {
	// Passive health checking for a single connection target, without a
	// pool.
	d := upstream.NewOutlierDetector(upstream.OutlierConfig{ConsecutiveFailures: 2}, nil, nil)
	defer d.Close()
	dial := func() error {
		t, ok := d.Admit()
		if !ok {
			return errors.New("backend ejected")
		}
		_, err := net.Dial("tcp", "invalid address")
		d.Record(t, upstream.OutcomeOf(err))
		return err
	}
	for range 3 {
		_ = dial()
	}
	fmt.Println(d.State(), d.Status().Ejections)
}

Output:
ejected 1

func NewOutlierDetector

func NewOutlierDetector(cfg OutlierConfig, clock Clock, onChange func()) *OutlierDetector

NewOutlierDetector returns a detector for one backend that reads time from clock; nil means SystemClock. onChange, if not nil, is called after every state transition, without any lock held, from the goroutine that caused the transition: one calling Record, Reset or Close, or the detector's timer goroutine. Calls may overlap, so onChange should read State instead of assuming an order.

A detector whose configuration is not Enabled admits everything and never ejects.

func (*OutlierDetector) Admit

func (d *OutlierDetector) Admit() (Ticket, bool)

Admit reports whether the backend may take a new request now. A healthy backend always admits; an ejected one never does; a half-open one admits while fewer than HalfOpenRequests trials are in flight. When Admit returns true the caller must pass the Ticket to Record exactly once.

func (*OutlierDetector) Close

func (d *OutlierDetector) Close()

Close stops the detector's timer. The detector keeps its current state and still answers Admit and Record, but it no longer changes state. Close is idempotent.

func (*OutlierDetector) Record

func (d *OutlierDetector) Record(t Ticket, o Outcome)

Record reports the outcome of a request admitted with t.

func (*OutlierDetector) Reset

func (d *OutlierDetector) Reset()

Reset clears any ejection and the ejection backoff, as if the backend had just recovered. It is meant for operators overriding the detector.

func (*OutlierDetector) SetConfig

func (d *OutlierDetector) SetConfig(cfg OutlierConfig)

SetConfig replaces the detector's configuration. The current state and any ejection in progress stay; the new thresholds and times apply from the next request and the next ejection. A configuration that is not Enabled also clears any ejection, as Reset does.

func (*OutlierDetector) State

func (d *OutlierDetector) State() OutlierState

State returns the current passive health state.

func (*OutlierDetector) Status

func (d *OutlierDetector) Status() OutlierStatus

Status returns a snapshot of the detector.

type OutlierState

type OutlierState uint32

OutlierState is the passive health state of a backend.

const (
	// OutlierOK means the backend is not ejected.
	OutlierOK OutlierState = iota

	// OutlierEjected means the backend is ejected and admits no traffic
	// until its ejection time is up.
	OutlierEjected

	// OutlierHalfOpen means the ejection time is up and the backend admits
	// a limited number of trial requests.
	OutlierHalfOpen
)

func (OutlierState) MarshalText

func (s OutlierState) MarshalText() ([]byte, error)

MarshalText encodes the state as its name.

func (OutlierState) String

func (s OutlierState) String() string

String returns the name of the state.

func (*OutlierState) UnmarshalText

func (s *OutlierState) UnmarshalText(b []byte) (err error)

UnmarshalText decodes a state from its name.

type OutlierStatus

type OutlierStatus struct {
	// EjectedUntil is when the current ejection ends; it is zero unless
	// State is OutlierEjected.
	EjectedUntil time.Time

	// Cause is why the backend was last ejected; it is nil while State is
	// OutlierOK.
	Cause error

	// Ejections counts every ejection since the detector was created.
	Ejections uint64

	// Refused counts the failures that would have ejected the backend but
	// that OutlierConfig.MaxEjectionPercent refused.
	Refused uint64

	// ConsecutiveFailures is the current run of failures.
	ConsecutiveFailures uint32

	// State is the passive health state.
	State OutlierState
}

OutlierStatus is a snapshot of an OutlierDetector.

type Picker

type Picker interface {
	// Pick returns the index of the chosen candidate. key is the request's
	// hash for key-affine balancers, or a random number when the caller
	// has no key. attempt is 0 for the first choice; when the caller
	// cannot use the chosen backend it asks again with attempt 1, 2 and so
	// on, and a Picker that is deterministic for a key should then return
	// other candidates.
	Pick(key uint64, attempt int) int
}

A Picker chooses among the candidates it was built from. It must be safe for concurrent use and should not allocate.

type Pool

type Pool[T any] struct {
	// contains filtered or unexported fields
}

A Pool is a set of backends with a balancer, health state and accounting. Pick and Release are lock-free and do not allocate; updates to membership and health build a new immutable routing view and publish it atomically, so picks in progress are never disturbed.

A backend is eligible for picks when it is not draining, not reported down by active checks and not ejected by passive checks; a half-open backend is eligible for its trial requests. When too few backends are eligible, Config.Fallback decides.

Rebuilds are coalesced: an update that comes while another goroutine builds the view leaves the next build to that goroutine and returns. A method that changes the pool has published its change when it returns, unless such a build is in progress, which then publishes it. Removals, drains and health verdicts take effect for picks at once either way, because a pick checks the backend's state itself; only new members and changed weights wait for the view.

The pool starts no goroutines. It owns the passive health timers and the slow-start timer, which Close stops. Pick never locks; Release locks only when its outcome ejects or restores a backend, to rebuild the view.

Locking: mu guards membership and settings and is never held while calling a Balancer, a Clock or an event subscriber. Inside mu the pool may take qmu (the event queue) and an OutlierDetector's mutex to read its status; neither calls out while held. rampMu (the slow-start timer) is taken alone.

func New

func New[T any](cfg Config) *Pool[T]

New returns an empty pool.

func (*Pool[T]) Available

func (p *Pool[T]) Available() int

Available returns the number of members that are eligible for new picks, as the routing view sees them: not draining, not offline, not down and not ejected, in every tier. A half-open member counts although it takes only its trial requests, and a full one although it takes no pick until a lease is released. Available is zero while the pool is empty or closed and in panic mode, when Pick, under FallbackAll, still picks among members that are not eligible; a readiness check that should fail in panic mode tests Available() > 0.

func (*Pool[T]) Backend

func (p *Pool[T]) Backend(id string) (*Backend[T], bool)

Backend returns the member with the given ID.

func (*Pool[T]) Backends

func (p *Pool[T]) Backends() []*Backend[T]

Backends returns the members in order.

func (*Pool[T]) Clock

func (p *Pool[T]) Clock() Clock

Clock returns the pool's time source, for packages that measure the latencies they observe on its leases.

func (*Pool[T]) Close

func (p *Pool[T]) Close()

Close stops the passive health and slow-start timers and ends all subscriptions: no subscriber call starts once Close has begun, and one in progress may finish. Close does not wait: a timer that has already fired may still finish its own bookkeeping, but runs no Balancer, Clock or subscriber code, and a rebuild already in progress publishes nothing. Close may be called from anywhere, including a subscriber or a Balancer. Picks fail with ErrClosed once Close returns and updates are refused; leases already issued may still be released. Close is idempotent.

func (*Pool[T]) Do

func (p *Pool[T]) Do(ctx context.Context, fn func(*Backend[T]) error) error

Do picks a backend, calls fn with it and releases the lease with OutcomeOf of fn's error, also when fn panics, in which case the outcome is Failure and the panic continues. When the outcome is Failure and ctx is not done, Do retries on another backend as Config.Retry allows. It returns nil once fn succeeds, fn's last error if fn failed, and otherwise ctx's or the first pick's error.

Example

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	pool := upstream.New[string](upstream.Config{
		Retry: upstream.RetryConfig{Attempts: 3, Budget: 0.2},
	})
	defer pool.Close()
	_ = pool.Sync([]upstream.Target[string]{
		{ID: "primary", Value: "10.0.0.1:8080"},
		{ID: "backup", Value: "10.0.0.2:8080", Priority: 1},
	})

	// Do releases each lease with the outcome of the function's error and
	// retries a failure on a backend it has not tried.
	err := pool.Do(context.Background(), func(b *upstream.Backend[string]) error {
		if b.ID() == "primary" {
			return errors.New("connection refused")
		}
		fmt.Println("served by", b.ID())
		return nil
	})
	fmt.Println(err)
	fmt.Println("primary failures:", pool.Status()[0].Failures)
}

Output:
served by backup
<nil>
primary failures: 1

func (*Pool[T]) DoKey

func (p *Pool[T]) DoKey(ctx context.Context, key string, fn func(*Backend[T]) error) error

DoKey is like Do with backends picked for key, as with PickKey.

func (*Pool[T]) Done

func (p *Pool[T]) Done() <-chan struct{}

Done returns a channel that is closed when Close is called. The pool starts no goroutines but the short-lived one that finishes a rebuild a committer handed off, which builds nothing once Close has begun, so nothing of it starts user code after that, and helpers attached to the pool, such as a health checker started with health.Watch, stop with it.

func (*Pool[T]) Drain

func (p *Pool[T]) Drain(id string) (*Drain, bool)

Drain stops new picks of the backend with the given ID while it stays a member, for maintenance or before removal. The returned Drain is done when the backend has no leases in flight, or canceled when Resume ends the drain first. Draining a backend that is already draining returns the same Drain. Drain reports false if there is no such member.

func (*Pool[T]) Len

func (p *Pool[T]) Len() int

Len returns the number of members.

func (*Pool[T]) MarkOffline

func (p *Pool[T]) MarkOffline(b *Backend[T], cause error) bool

MarkOffline takes b out of rotation on the caller's word, for example because the session the backend is reached through dropped, until MarkOnline. It is one of three inputs that each veto picks: an offline backend, one that active checks report down and one that passive checks ejected take no new picks, and MarkOnline lifts only its own veto. While several apply, the backend's State is the first in the order StateDraining, StateOffline, StateUnhealthy, StateEjected, StateHalfOpen. Unlike the health verdicts, the mark also holds in panic mode: FallbackAll never picks an offline backend, which also does not count towards Config.PanicThreshold.

Taking b offline emits EventStateChanged with cause as its Err, or ErrOffline if cause is nil, and rebuilds the view like any other change; marking an offline backend again only replaces the cause, which the next event reports. The mark follows the member's ID across updates. A call for a backend that has been removed, or replaced by a new member with the same ID, is ignored; MarkOffline reports whether b was a current member of an open pool.

Example

package main

import (
	"errors"
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	pool := upstream.New[string](upstream.Config{})
	defer pool.Close()
	_ = pool.Sync([]upstream.Target[string]{{ID: "edge-1", Value: "edge-1"}})
	defer pool.Subscribe(func(ev upstream.Event[string]) {
		if ev.Kind == upstream.EventStateChanged {
			fmt.Println(ev.Backend.ID(), ev.From, "->", ev.To, ev.Err)
		}
	})()

	// The session the backend is reached through dropped, and came back.
	b, _ := pool.Backend("edge-1")
	pool.MarkOffline(b, errors.New("session closed"))
	_, err := pool.Pick()
	fmt.Println(err)
	pool.MarkOnline(b)
}

Output:
edge-1 healthy -> offline session closed
upstream: no backend available
edge-1 offline -> healthy <nil>

func (*Pool[T]) MarkOnline

func (p *Pool[T]) MarkOnline(b *Backend[T]) bool

MarkOnline lifts the mark of MarkOffline; a backend that comes back online ramps up under Config.SlowStart. Like MarkOffline, it ignores backends that are no longer members, and reports whether b was a current member of an open pool.

func (*Pool[T]) Pick

func (p *Pool[T]) Pick() (Lease[T], error)

Pick leases a backend chosen by the balancer without a key. Key-affine balancers pick at random.

func (*Pool[T]) PickHash

func (p *Pool[T]) PickHash(hash uint64) (Lease[T], error)

PickHash is like PickKey for callers that hash keys themselves.

func (*Pool[T]) PickKey

func (p *Pool[T]) PickKey(key string) (Lease[T], error)

PickKey leases a backend chosen for key, such as a client address or a session ID. With ConsistentHash the same key keeps mapping to the same backend while it is eligible; other balancers ignore the key. An empty key carries no affinity: PickKey("") is Pick.

Example

package main

import (
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	pool := upstream.New[string](upstream.Config{Balancer: upstream.ConsistentHash(upstream.HashConfig{})})
	defer pool.Close()
	_ = pool.Sync([]upstream.Target[string]{{ID: "a"}, {ID: "b"}, {ID: "c"}})

	first, _ := pool.PickKey("198.51.100.7")
	again, _ := pool.PickKey("198.51.100.7")
	fmt.Println("sticky:", first.Backend() == again.Backend())
	first.Release(upstream.Success)
	again.Release(upstream.Success)
}

Output:
sticky: true

func (*Pool[T]) Put

func (p *Pool[T]) Put(t Target[T]) error

Put adds a target, or replaces the member with the same ID, keeping its position, health state and counters. It emits EventAdded or EventUpdated; a Put that changes none of the weight, priority, cap and value of a member does nothing.

func (*Pool[T]) Remove

func (p *Pool[T]) Remove(id string) (*Drain, bool)

Remove removes the backend with the given ID gracefully: new picks avoid it at once, and leases already issued finish normally. The returned Drain is done when the last of them is released. Remove reports false if there is no such member.

Example

package main

import (
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	pool := upstream.New[string](upstream.Config{})
	defer pool.Close()
	_ = pool.Sync([]upstream.Target[string]{{ID: "old"}, {ID: "new"}})

	lease, _ := pool.PickKey("")
	for lease.Backend().ID() != "old" {
		lease.Release(upstream.Neutral)
		lease, _ = pool.PickKey("")
	}
	drained, _ := pool.Remove("old")
	select {
	case <-drained.Done():
	default:
		fmt.Println("waiting for the last lease")
	}
	lease.Release(upstream.Success)
	<-drained.Done()
	fmt.Println("drained")
}

Output:
waiting for the last lease
drained

func (*Pool[T]) ReportCheck

func (p *Pool[T]) ReportCheck(b *Backend[T], h Health, r CheckResult) bool

ReportCheck records the result of an active health check of b and sets its health. A health checker calls it after every check; the checker, not the pool, applies rise and fall thresholds. HealthDown takes the backend out of rotation; HealthUp and HealthUnknown keep it in, so a change between them only updates the health and rebuilds nothing. Reports for a backend that has been removed, or replaced by a new member with the same ID, are ignored; ReportCheck reports whether b was a current member of an open pool.

func (*Pool[T]) Request

func (p *Pool[T]) Request(key string) Request[T]

Request returns the picker for one request routed by key; an empty key carries no affinity, as with PickKey.

func (*Pool[T]) ResetOutlier

func (p *Pool[T]) ResetOutlier(id string) bool

ResetOutlier clears the passive health ejection of the backend with the given ID, if any. It reports false if there is no such member.

func (*Pool[T]) Resume

func (p *Pool[T]) Resume(id string) bool

Resume ends a drain started by Drain; the Drain reports Canceled unless it had already completed. Resume reports false if there is no such member.

func (*Pool[T]) SetConfig

func (p *Pool[T]) SetConfig(cfg Config) error

SetConfig replaces the pool's configuration, except Clock, keeping its members with their counters, health and leases. A new Balancer builds a new view from scratch; the same Balancer, such as the same pointer or an equal ConsistentHash value, goes on rebuilding its Pickers; new outlier settings apply to the next request of every member, and disabling passive health checking clears all ejections; new retry settings apply to the next Request.Next. It returns ErrClosed if the pool is closed.

func (*Pool[T]) Status

func (p *Pool[T]) Status() []Status

Status returns a snapshot of every member, in order.

func (*Pool[T]) Subscribe

func (p *Pool[T]) Subscribe(fn func(Event[T])) (cancel func())

Subscribe calls fn for every later event, starting with an EventAdded for each current member, and returns a function that ends the subscription.

Events are delivered one at a time in the order they happened, never while the pool holds a lock, on the goroutine that caused them or on one already delivering events. fn may call any Pool method; events that call causes are delivered after fn returns. fn should be quick, because the goroutine delivering events may be one releasing a lease. An event may be delivered slightly before picks observe the change it reports. After cancel returns, fn may still be running for an event already in progress but is not called again.

Example

package main

import (
	"errors"
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	pool := upstream.New[string](upstream.Config{})
	defer pool.Close()
	_ = pool.Sync([]upstream.Target[string]{{ID: "a"}})

	cancel := pool.Subscribe(func(ev upstream.Event[string]) {
		fmt.Println(ev.Kind, ev.Backend.ID(), ev.From, "->", ev.To, ev.Err)
	})
	defer cancel()
	a, _ := pool.Backend("a")
	refused := errors.New("refused")
	// Repeated failures report one transition.
	pool.ReportCheck(a, upstream.HealthDown, upstream.CheckResult{Err: refused})
	pool.ReportCheck(a, upstream.HealthDown, upstream.CheckResult{Err: refused})
	pool.ReportCheck(a, upstream.HealthUp, upstream.CheckResult{})
}

Output:
added a healthy -> healthy <nil>
state-changed a healthy -> unhealthy refused
state-changed a unhealthy -> healthy <nil>

func (*Pool[T]) Sync

func (p *Pool[T]) Sync(targets []Target[T]) error

Sync replaces the pool's membership with targets, in order. Members whose ID is in targets keep their health state and counters and take the new weight and value; new IDs join as healthy; members missing from targets are removed gracefully: they take no new picks while their leases finish. Sync emits EventRemoved, then EventAdded or EventUpdated for each target that is new or changed, in order. A Sync that changes nothing, order included, does nothing. On error nothing changes.

type Rebuilder

type Rebuilder interface {
	Balancer

	// Rebuild returns a Picker for candidates, which is never empty. prev
	// was returned by Build or Rebuild of this Balancer and may still be
	// in use by picks: Rebuild must not modify it.
	Rebuild(prev Picker, candidates []Candidate) Picker
}

A Rebuilder is a Balancer that derives a Picker from the previous one, for balancers whose Build is expensive, such as hash rings. A Pool calls Rebuild instead of Build when it has a Picker that the same Balancer built. The result must be what Build would return for candidates; Rebuild only saves work, such as by keeping the parts of prev whose candidates did not change.

type Request

type Request[T any] struct {
	// contains filtered or unexported fields
}

A Request picks distinct backends for one request: the first Next is a pick like PickKey, and every later Next is a retry that leases a backend this Request has not returned before, as the pool's RetryConfig allows. It reads the pool's current routing view and retry settings on every call, so a retry avoids backends that went down meanwhile.

Do runs the whole retry loop; Pool.Do is Do with a function of the backend. Use Next directly for loops that Do cannot express. A Request is for one goroutine. It does not allocate for up to four attempts.

Example

package main

import (
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	pool := upstream.New[string](upstream.Config{Retry: upstream.RetryConfig{Attempts: 2}})
	defer pool.Close()
	_ = pool.Sync([]upstream.Target[string]{{ID: "a"}, {ID: "b"}})

	// A Request leases distinct backends for one request: each Next
	// leases a backend not tried before, as the retry settings allow.
	req := pool.Request("client-key")
	first, _ := req.Next()
	first.Release(upstream.Failure)
	second, _ := req.Next()
	fmt.Println(first.Backend() == nil, second.Backend() != nil)
	second.Release(upstream.Success)
	_, err := req.Next()
	fmt.Println(err)
}

Output:
true true
upstream: retry limit reached

func (*Request[T]) Do

func (r *Request[T]) Do(ctx context.Context, fn func(l *Lease[T]) (retry bool, err error)) error

Do calls fn with a lease from Next, again with a new lease for as long as fn asks for a retry, ctx is not done and Next allows one. fn owns its lease: it may release it, or hand it on with Lease.Take, as for a connection that outlives fn. A lease fn leaves unreleased is released when fn returns, with OutcomeOf of fn's error, or with Failure if fn panics, in which case the panic goes on.

Do returns fn's error once fn asks for no retry. When a retry is declined, it returns fn's last error, or, if that was nil, ctx's error or the pick's; so does it when no lease could be picked at all. Do never returns nil unless fn's last call did.

Example

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	pool := upstream.New[string](upstream.Config{Retry: upstream.RetryConfig{Attempts: 2}})
	defer pool.Close()
	_ = pool.Sync([]upstream.Target[string]{{ID: "a"}, {ID: "b"}})

	// Do runs the retry loop for leases that outlive it, such as
	// connections: Take hands the lease of the attempt that worked on.
	var kept upstream.Lease[string]
	req := pool.Request("")
	err := req.Do(context.Background(), func(l *upstream.Lease[string]) (retry bool, err error) {
		if req.Tried() == 1 {
			return true, errors.New("refused") // released as a Failure, then retried
		}
		kept = l.Take()
		return false, nil
	})
	fmt.Println(err, req.Tried(), kept.Backend() != nil)
	kept.Release(upstream.Success)
}

Output:
<nil> 2 true

func (*Request[T]) Next

func (r *Request[T]) Next() (Lease[T], error)

Next leases the next backend. The first call fails like PickHash; a retry fails with ErrRetryLimit after RetryConfig.Attempts leases, with ErrRetryBudget when the budget is spent, with ErrUnavailable when every backend that takes traffic has been tried, and with ErrSaturated when those left are full. Release the previous lease, usually with Failure, before or after calling Next.

func (*Request[T]) Tried

func (r *Request[T]) Tried() int

Tried returns the number of leases Next has returned.

type RetryConfig

type RetryConfig struct {
	// Window is the span over which the budget counts requests and
	// retries. The default is 10s. The counts are a sliding estimate over
	// the current and the previous window.
	Window time.Duration

	// Budget, if positive, is the largest ratio of retries to requests,
	// such as 0.2 for one retry per five requests. Zero means no budget:
	// only Attempts bounds retries.
	Budget float64

	// Attempts is the most backends one request tries, the first one
	// included. Zero and one mean no retries.
	Attempts int

	// MinRetries is the number of retries per Window that the budget
	// allows whatever the request rate, so that a pool with little
	// traffic can still retry. Zero means 10; a negative value means none.
	MinRetries int
}

RetryConfig configures retries on other backends, by Pool.Do, Pool.DoKey and Request. The zero value allows no retries.

A retry budget bounds the extra load that retries put on the backends when many requests fail at once: within any Window, retries may number at most MinRetries plus Budget times the requests started.

type State

type State uint8

State is the routing state of a backend, combining its membership, the caller's availability input, its active health and its passive health. When several conditions hold, the state is the first that applies in this order: removed, draining, offline, unhealthy, ejected, half-open, healthy.

const (
	// StateHealthy means the backend takes new traffic.
	StateHealthy State = iota

	// StateHalfOpen means the backend's ejection is over and it takes a
	// limited number of trial requests.
	StateHalfOpen

	// StateEjected means passive health checking ejected the backend.
	StateEjected

	// StateUnhealthy means active health checks report the backend down.
	StateUnhealthy

	// StateDraining means the backend takes no new traffic but its leases
	// may still be in flight.
	StateDraining

	// StateRemoved means the backend is no longer a member of its pool.
	StateRemoved

	// StateOffline means the caller marked the backend unavailable with
	// Pool.MarkOffline, for example because the session it is reached
	// through dropped.
	StateOffline
)

func (State) MarshalText

func (s State) MarshalText() ([]byte, error)

MarshalText encodes the state as its name.

func (State) String

func (s State) String() string

String returns the name of the state.

func (*State) UnmarshalText

func (s *State) UnmarshalText(b []byte) (err error)

UnmarshalText decodes a state from its name.

type Status

type Status struct {
	// LastCheck is the latest active health check result; its Time is zero
	// before the first check.
	LastCheck CheckResult

	// Outlier is the passive health snapshot. Its State is always
	// OutlierOK when passive health checking is disabled.
	Outlier OutlierStatus

	// ID identifies the backend.
	ID string

	// Weight is the backend's weight.
	Weight int

	// Priority is the backend's tier: lower values are preferred.
	Priority int

	// MaxInFlight is the backend's cap on leases in flight; zero means
	// none.
	MaxInFlight int

	// InFlight is the number of leases not yet released.
	InFlight int64

	// Latency is the moving average of the latencies reported with
	// Lease.Observe, or zero before the first one.
	Latency time.Duration

	// Successes, Failures and Neutrals count the outcomes of leases, as
	// reported by Lease.Report or Lease.Release. Together with InFlight
	// they count every pick: each lease is in flight until released and
	// has one outcome.
	Successes uint64
	Failures  uint64
	Neutrals  uint64

	// State is the routing state.
	State State

	// Health is the latest active health verdict, whatever the State.
	Health Health

	// Draining reports whether the backend is draining or removed.
	Draining bool

	// Offline reports whether Pool.MarkOffline marked the backend
	// unavailable, whatever the State.
	Offline bool
}

Status is a snapshot of one backend.

type Target

type Target[T any] struct {
	// ID identifies the backend. It must be unique within the pool. Health
	// state and counters follow the ID across updates, so give a backend a
	// new ID when its value stops meaning the same server.
	ID string

	// Weight is the backend's relative capacity within its tier, from 0 to
	// MaxWeight. Zero means 1; to stop sending traffic to a backend, drain
	// it.
	Weight int

	// Priority is the backend's tier. Picks go to the tier with the lowest
	// Priority that has an eligible member, balanced among that tier's
	// members, and move to the next tier only when no member of the tier
	// can take the pick; see Config.Overflow for tiers that are full.
	// Members of different priorities make an active-backup setup. The
	// default, 0, puts every member in one tier.
	Priority int

	// MaxInFlight, if positive, caps the backend's leases in flight: a
	// backend at its cap refuses picks, in panic mode too, until a lease
	// is released. Zero means no cap. The in-flight count may read above
	// the cap for a moment while a pick that lost a race for the last slot
	// backs out; the leases never exceed it.
	MaxInFlight int

	// Value is what callers connect to, such as an address.
	Value T
}

A Target describes a backend to a Pool.

Example (Priority)

package main

import (
	"fmt"

	"github.com/djylb/upstream"
)

func main() {
	// Two primaries share the load; the backup takes traffic when both are
	// down, and, with Overflow, when both are at their MaxInFlight.
	pool := upstream.New[string](upstream.Config{Overflow: true})
	defer pool.Close()
	_ = pool.Sync([]upstream.Target[string]{
		{ID: "primary-1", Value: "10.0.0.1:8080", MaxInFlight: 1},
		{ID: "primary-2", Value: "10.0.0.2:8080", MaxInFlight: 1},
		{ID: "backup", Value: "10.0.0.9:8080", Priority: 1},
	})

	var leases []upstream.Lease[string]
	for range 3 {
		l, err := pool.Pick()
		if err != nil {
			fmt.Println(err)
			return
		}
		leases = append(leases, l)
	}
	fmt.Println("third pick:", leases[2].Backend().ID())
	for i := range leases {
		leases[i].Release(upstream.Success)
	}
}

Output:
third pick: backup

type Ticket

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

A Ticket is the admission returned by OutlierDetector.Admit. Pass it back to Record exactly once when the request ends.