Help us shape the future of Go! Take the 2026 Go Developer Survey

upstream

package module
v0.0.0-...-13dc683 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 25, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

README

upstream

Go Reference

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, failover (active-backup) and consistent hashing with optional bounded loads, or your own.
  • 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.
  • 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 or ErrClosed
}
conn, err := net.Dial("tcp", lease.Value())
lease.Release(upstream.OutcomeOf(err))

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.

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

Attempts is the same for leases that outlive a function: each Next leases a backend not returned before, or fails with ErrRetryLimit, ErrRetryBudget or ErrUnavailable.

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.

A panic in a function given to Do, upnet.Dialer or uphttp.RoundTripper releases the backend's lease and goes on. Where a pool's values cannot serve as addresses and no function says how to use them, health.Watch and health.Attach return health.ErrNoTarget, upnet returns ErrNoAddress and uphttp returns ErrNoRewrite.

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.
Failover() First eligible backend in membership order (active-backup).
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.

Health and eligibility

A backend takes new picks when it is

  • not draining or removed,
  • 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.

When no backend is eligible, Config.Fallback decides:

  • FallbackNone (default): Pick returns ErrUnavailable.
  • FallbackAll: panic mode. Pick chooses among all members that are not draining, ignoring health, on the grounds that a signal marking everything down is more often wrong than right. Outcomes still count.

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.

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.

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.

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 and EventStateChanged in order, starting with the current members. 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. 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, expected status range, body match and your own client for TLS. 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 starts a new generation, 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.

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 starts no goroutines; it owns its ejection and slow-start timers, which Close stops. Pool.Done reports that the pool was closed.
  • 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 on rings of more than 256 backends allocate.
  • Attempts allocates from the fifth attempt of one request on.
  • 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, Failover and ConsistentHash (a hash ring with optional bounded loads), or any implementation of the interface.
  • 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.
  • 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. Pool.Do picks, runs a function and releases, retrying failures on other backends as RetryConfig allows, within a retry budget; Attempts does the same for leases that outlive a function. Picks are lock-free and do not allocate: they read an immutable routing view that every change rebuilds and publishes atomically, so updates never disturb picks in flight.

Health

A backend is eligible when it is not draining, not reported unhealthy by active checks and not ejected by passive checks. 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 not draining.

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 starts no goroutines; the only timers it 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

Examples

Constants

View Source 
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

View Source 
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.

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

	// ErrUnavailable means no member can take traffic now: every one is
	// unhealthy, ejected, draining or out of half-open trial slots.
	ErrUnavailable = errors.New("upstream: no backend available")

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

View Source 
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 Attempts.Next when it declines a retry.

View Source 
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.

Functions

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 Attempts 

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

Attempts 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 Attempts has not returned before, as the pool's RetryConfig allows. It reads the pool's current routing view on every call, so a retry avoids backends that went down meanwhile.

Pool.Do is Attempts with a function; use Attempts directly when a lease outlives a function call, such as for a connection or a response body. An Attempts 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"}})

	// Attempts is Do for leases that outlive a function, such as
	// connections: each Next leases a backend not tried before.
	attempts := pool.Attempts("client-key")
	first, _ := attempts.Next()
	first.Release(upstream.Failure)
	second, _ := attempts.Next()
	fmt.Println(first.Backend() == nil, second.Backend() != nil)
	second.Release(upstream.Success)
	_, err := attempts.Next()
	fmt.Println(err)
}

Output:
true true
upstream: retry limit reached

func (*Attempts[T]) Count 

func (a *Attempts[T]) Count() int

Count returns the number of leases Next has returned.

func (*Attempts[T]) Next 

func (a *Attempts[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, and with ErrUnavailable when every backend that takes traffic has been tried. Release the previous lease, usually with Failure, before or after calling Next.

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.

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]) 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 calls Build whenever the set of backends that may receive traffic changes, never while holding a lock, and keeps using the previous Picker until Build returns.

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 and bounded-load walks on rings of more than 256 candidates, which need a scratch bitset.

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 Failover 

func Failover() Balancer

Failover returns a Balancer for active-backup setups: it always picks the first candidate, in membership order, that can take traffic. A Pool leaves unhealthy and drained backends out of the candidates, so traffic moves to the next backend in order and returns when the first recovers.

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", "failover" 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 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

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

	// Latency is how long the check took.
	Latency time.Duration
}

CheckResult is the result of one active health check.

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

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.

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

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

	// Retry configures how Do, DoKey and Attempts 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.
	SlowStart time.Duration

	// Fallback says what to do when no backend is eligible.
	Fallback Fallback
}

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.

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), and nil otherwise.
	Err error

	// Backend is the member concerned.
	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.
	EventUpdated

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

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

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 no backend is eligible. 

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

	// FallbackAll makes Pick choose among all members that are not
	// draining, ignoring active and passive health. This is panic mode: a
	// health signal that marks every backend down is more often wrong than
	// right, and degraded service beats none. 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 and Pool.DoKey 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]) 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]) 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
}

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

	// 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 no backend is eligible, Config.Fallback decides.

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 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. pubMu (view publication) and rampMu (the slow-start timer) are taken alone.

func New 

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

New returns an empty pool.

func (*Pool[T]) Attempts 

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

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

func (*Pool[T]) Available 

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

Available returns the number of members that are eligible for new picks now, as the routing view sees them. It is zero while the pool is empty or closed, and while no member is eligible: then Pick fails with ErrUnavailable, or, under FallbackAll, still succeeds in panic mode, picking 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]) 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. Close may be called from anywhere, including a subscriber or a Balancer. Later picks fail with ErrClosed 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{
		Balancer: upstream.Failover(),
		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"},
	})

	// 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, so nothing of it runs 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]) 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.

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)

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. Reports for a backend that has been removed, or replaced by a new member with the same ID, are ignored.

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]) 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 in order. On error nothing changes.

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 Attempts. 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, its active health and its passive health. When several conditions hold, the state is the first that applies in this order: removed, draining, 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
)

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

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

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

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, from 0 to MaxWeight. Zero
	// means 1; to stop sending traffic to a backend, drain it.
	Weight int

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

A Target describes a backend to a Pool.

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.

Source Files

View all Source files

Directories

Path Synopsis
Package health runs active health checks: it probes targets on jittered intervals, applies rise and fall thresholds, measures latency and classifies failures, and reports results and transitions.
 Package health runs active health checks: it probes targets on jittered intervals, applies rise and fall thresholds, measures latency and classifies failures, and reports results and transitions.
internal
fakeclock
Package fakeclock provides a manually advanced clock for tests.
 Package fakeclock provides a manually advanced clock for tests.
Package uphttp holds the HTTP-specific parts of upstream: an HTTP health check, a cache of per-backend transports and an http.RoundTripper that balances requests over a pool.
 Package uphttp holds the HTTP-specific parts of upstream: an HTTP health check, a cache of per-backend transports and an http.RoundTripper that balances requests over a pool.
Package upnet connects to the backends of an upstream.Pool: a Dialer picks a backend, dials it, retries on another backend when the dial fails, and holds the backend's lease for as long as the connection is open, so that in-flight counts, least-connections balancing and drains follow real connections.
 Package upnet connects to the backends of an upstream.Pool: a Dialer picks a backend, dials it, retries on another backend when the dial fails, and holds the backend's lease for as long as the connection is open, so that in-flight counts, least-connections balancing and drains follow real connections.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.