Documentation
¶
Overview ¶
Package upstream implements parent-proxy chaining with failover and circuit-breaker protection (ADR-0002 extraction; engine was upstream.go in package main).
Culvert can route traffic through one or more parent HTTP proxies with automatic failover. When all upstreams are down the proxy falls back to direct connections. The package owns the pool state machine (round-robin selection, health flags, per-proxy circuit breakers) and the health-check loop; package main keeps the singleton, the transport wiring (applyUpstreamProxy), and persistence via admin_settings.
Configuration (config.yaml):
upstream:
proxies:
- url: "http://parent1.corp.com:3128"
- url: "http://parent2.corp.com:3128"
health_interval: "30s"
circuit_breaker:
threshold: 5 # failures before opening circuit
timeout: "60s" # how long circuit stays open before half-open probe
Index ¶
- func FormatSummary(entries []Entry) string
- func RunHealthCheckLoop(ctx context.Context, pool *Pool, interval time.Duration)
- type CircuitBreaker
- func (cb *CircuitBreaker) Allow() bool
- func (cb *CircuitBreaker) Failures() int64
- func (cb *CircuitBreaker) OpenedAt() time.Time
- func (cb *CircuitBreaker) Params() (threshold int, timeout time.Duration)
- func (cb *CircuitBreaker) RecordFailure()
- func (cb *CircuitBreaker) RecordSuccess()
- func (cb *CircuitBreaker) State() string
- type Config
- type Entry
- type Pool
- func (p *Pool) CBParams() (threshold int, timeout time.Duration)
- func (p *Pool) Configure(entries []Entry, cbThreshold int, cbTimeout time.Duration)
- func (p *Pool) Enabled() bool
- func (p *Pool) Entries() []Entry
- func (p *Pool) HealthCheck()
- func (p *Pool) List() []Status
- func (p *Pool) Next() *Proxy
- func (p *Pool) ProxyFunc() func(*http.Request) (*url.URL, error)
- func (p *Pool) SetProxies(entries []Entry)
- type Proxy
- type Status
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func FormatSummary ¶
FormatSummary returns a log-friendly summary like "2 proxies (parent1:3128, parent2:3128)".
func RunHealthCheckLoop ¶
RunHealthCheckLoop runs pool.HealthCheck at the given interval until ctx is cancelled, stopping the underlying ticker on exit. Extracted so the shutdown invariant — "the loop must exit on context cancellation" — is unit-testable without spinning up the rest of initUpstreamPool.
Defensive contract: returns immediately for a nil pool or a non-positive interval. The production caller (initUpstreamPool) already validates these, but the standalone helper guards itself so future callers cannot panic (nil-deref) or wedge on a zero-interval ticker. P1.3 / S4.UpstreamHealth.
Types ¶
type CircuitBreaker ¶
type CircuitBreaker struct {
// contains filtered or unexported fields
}
CircuitBreaker tracks consecutive failures for an upstream proxy.
func (*CircuitBreaker) Allow ¶
func (cb *CircuitBreaker) Allow() bool
Allow returns true if the circuit permits a request.
func (*CircuitBreaker) Failures ¶ added in v1.0.48
func (cb *CircuitBreaker) Failures() int64
Failures returns the current consecutive-failure count. Exported for admin API / diagnostics surfacing; not used on the request path.
func (*CircuitBreaker) OpenedAt ¶ added in v1.0.48
func (cb *CircuitBreaker) OpenedAt() time.Time
OpenedAt returns when the circuit last tripped open (zero Time if it has never opened, or has since been reset by RecordSuccess). Exported for admin API / diagnostics surfacing; not used on the request path.
func (*CircuitBreaker) Params ¶
func (cb *CircuitBreaker) Params() (threshold int, timeout time.Duration)
Params returns the breaker's configured threshold and timeout. Exported for the main-side persistence-contract tests (SetProxies must inherit Configure's params) and for diagnostics; not used on the request path.
func (*CircuitBreaker) RecordFailure ¶
func (cb *CircuitBreaker) RecordFailure()
RecordFailure increments the failure count and opens the circuit if threshold is reached.
func (*CircuitBreaker) RecordSuccess ¶
func (cb *CircuitBreaker) RecordSuccess()
RecordSuccess resets the failure count and closes the circuit.
func (*CircuitBreaker) State ¶
func (cb *CircuitBreaker) State() string
State returns the current circuit state name.
type Config ¶
type Config struct {
Proxies []Entry `yaml:"proxies" json:"proxies"`
HealthInterval string `yaml:"health_interval" json:"healthInterval"` // Go duration
CircuitBreaker struct {
Threshold int `yaml:"threshold" json:"threshold"` // failures before open
Timeout string `yaml:"timeout" json:"timeout"` // Go duration
} `yaml:"circuit_breaker" json:"circuitBreaker"`
}
Config is the "upstream" section of config.yaml.
type Entry ¶
type Entry struct {
URL string `yaml:"url" json:"url"`
}
Entry is one parent proxy from config.yaml.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool manages a set of parent proxies with failover. The zero value is a usable empty pool.
func (*Pool) CBParams ¶
CBParams returns the circuit-breaker parameters remembered from the last Configure. Exported for the main-side snapshot/restore test helper and for diagnostics; not used on the request path.
func (*Pool) Configure ¶
Configure sets the list of upstream proxies and the circuit-breaker parameters (startup / YAML-reload / import path).
func (*Pool) Entries ¶
Entries returns a copy of the raw accepted entries. URLs may embed inline credentials — this is for persistence (admin_settings.json, mode 0600) only; use List() for anything user-facing.
func (*Pool) HealthCheck ¶
func (p *Pool) HealthCheck()
HealthCheck runs a connectivity check against each upstream proxy. Called periodically from a background goroutine.
func (*Pool) List ¶
List returns the current upstream proxy statuses for the UI/API. URLs are redacted (inline credentials stripped).
func (*Pool) Next ¶
Next returns the next healthy upstream proxy using round-robin selection. Returns nil if no healthy proxy is available (caller should fall back to direct).
func (*Pool) ProxyFunc ¶
ProxyFunc returns an http.Transport-compatible proxy selector. When upstreams are configured, it returns the next healthy proxy URL. Falls back to nil (direct connection) when no upstream is available.
func (*Pool) SetProxies ¶
SetProxies replaces the proxy list while keeping the circuit-breaker parameters from the last Configure (admin API / persisted-settings path).
type Proxy ¶
type Proxy struct {
URL *url.URL
Healthy atomic.Bool
CB *CircuitBreaker
}
Proxy represents one parent proxy in the chain.
type Status ¶
type Status struct {
URL string `json:"url"`
Healthy bool `json:"healthy"`
Circuit string `json:"circuit"`
// Failures is the current consecutive-failure count tracked by the
// circuit breaker (resets to 0 on RecordSuccess).
Failures int64 `json:"failures"`
// OpenedAtMs is when the circuit last tripped open, as Unix
// milliseconds; 0 when the circuit is closed.
OpenedAtMs int64 `json:"openedAtMs,omitempty"`
// RetryAfterMs is the remaining time (ms) until the breaker allows a
// half-open probe; 0 when the circuit is not open.
RetryAfterMs int64 `json:"retryAfterMs,omitempty"`
}
Status is returned by the admin API.
Source Files