Repository navigation
Replies: 2 comments 2 replies
ReviewA breaker in bedrock is defensible, but the proposal doesn't make the case yet. It leads with the weaker reason (dependency freshness policies) and gives the stronger one (bedrock's ethos) a single sentence. Before it turns into stories, it needs concrete reasons, a home, a scope and some design decisions. 1. Reword the freshness argument or drop it
2. Say exactly how existing breakers break the ethosThese come from reading
The proposal should also say why a wrapper isn't enough: a wrapper could require every setting, but it can't remove the It shouldn't claim "stdlib-only" as an advantage. gobreaker and cep21/circuit already depend only on testify, for their tests, and neither keeps global state. 3. Why bedrock rather than humus
It should be a nested module like 4. Draw the scope boundary explicitly
5. Design questions to settle before stories
|
2 replies
Proposed API: answers to the seven design questionsThis answers the design questions in the review above and proposes a package API for them. How it was produced:
How it was checked: a separate implementation was built from scratch against the proposal and needed no change to the exported API. It imports only
Both the implementation and the humus sketch are prototypes. Nothing is in the repository yet. The API at a glanceModule type State int // Closed, Open, HalfOpen; the zero value is invalid
type Outcome int // Success, Failure, Ignored; the zero value is invalid
const MaxCount = 1 << 16
type Config struct { Window, Probes int; OpenFor, ProbeTimeout time.Duration } // all required
type Counts struct { Successes, Failures, ConsecutiveSuccesses, ConsecutiveFailures, TimedOut int }
type Policy struct {
Trip func(Counts) bool // asked while closed
Probe func(Counts) State // asked while half-open: Closed, Open, or HalfOpen to keep probing
}
type Transition struct { From, To State; Generation uint64; Counts Counts }
var ErrRejected error
type RejectedError struct { State State; RetryAfter time.Duration } // Unwrap returns ErrRejected
func New(cfg Config, p Policy,
classify func(context.Context, error) Outcome,
observe func(context.Context, Transition)) (*Breaker, error)
func (b *Breaker) Do(ctx context.Context, fn func(context.Context) error) error
func (b *Breaker) Allow(ctx context.Context) (*Ticket, error)
func (b *Breaker) State() State // a pure read of the current state: no transition, no caller code
func (t *Ticket) Done(err error) bool // false: the result arrived after a state change and was discarded
func (t *Ticket) Release() // settles without a result; calls no caller codeThe full API with doc comments, the examples and the humus sketch are in the collapsed sections at the end. Answers1. Counts and trip policy
2. Classifying results
3. API shape
4. Panics without
5. Validation
6. Observer
7. Clock
What the prototype's tests coverEvery transition, run under synctest, and also:
Concerns
Decisions for the maintainerEach one has a recommendation:
Adoption costA third module:
Full API with doc comments// Package circuit models a circuit breaker.
//
// # Background
//
// A circuit breaker stops a process from calling a dependency that keeps
// failing, and lets a few calls through now and then to learn when it has
// recovered, as described in Michael Nygard's Release It! (2007) and in
// https://learn.microsoft.com/en-us/azure/architecture/patterns/circuit-breaker.
// A [Breaker] stands in front of one dependency, such as one host or one
// gRPC service, and is in one of three states:
//
// - [Closed]: every call is admitted, and its result is counted. After
// each success or failure, [Policy.Trip] decides whether to open;
// - [Open]: every call is rejected with a [*RejectedError], until
// [Config.OpenFor] has passed. The first call after that makes the
// Breaker half-open;
// - [HalfOpen]: at most [Config.Probes] calls, the probes, are in flight
// at once, and the rest are rejected. After each probe's success or
// failure, and after each probe that times out, [Policy.Probe] decides
// whether to close, to reopen or to keep probing.
//
// circuit models only that state machine. It does no I/O, starts no
// goroutine and sets no timer. Wrapping an http.RoundTripper or a gRPC
// connection, exporting metrics and reading [Config] from the environment
// belong to the framework built on it.
//
// # Calls
//
// [Breaker.Do] runs a call whose result is known when a function returns.
// [Breaker.Allow] admits a call whose result is known only later, such as a
// response whose body the caller is still reading, and returns a [*Ticket].
// The Ticket is settled exactly once: with [Ticket.Done] and the result, or
// with [Ticket.Release] when there is no result to report. Settling it
// twice panics.
//
// A rejected call gets a [*RejectedError], which matches [ErrRejected] and
// says how much longer the Breaker stays open.
//
// Each call belongs either at startup or on the request path:
//
// | Call | When |
// |--------------------------------------------|-------------------------------------------------|
// | Reading Config, building the Policy | Startup, inside App.Run |
// | New | Startup, once per dependency |
// | Do, or Allow and then Done or Release | Per outbound call |
// | the classifier | Per result, with the Breaker unlocked |
// | Policy functions | Per success, failure or timeout, Breaker locked |
// | observe | Per change of state, with the Breaker unlocked |
// | State | Whenever, such as in a metrics callback |
//
// One Breaker is shared by every goroutine that calls its dependency, so
// the classifier, the Policy functions and observe must be safe for
// concurrent use.
//
// # Classifying results
//
// circuit does not decide what a failure is. Every result goes to the
// classify function passed to [New], together with the context of the
// call, and classify returns a [Success], a [Failure] or [Ignored]. An
// Ignored result is evidence of nothing: it is not counted, it neither
// extends nor ends a run of results, and it frees the probe slot it held.
// circuit has no default classifier and treats no error specially, a nil
// error and the context errors included.
//
// The context lets classify tell the caller giving up from the dependency
// failing. ctx.Err() tells three cases apart:
//
// - context.Canceled: the caller gave up, which says nothing about the
// dependency;
// - context.DeadlineExceeded: the caller's deadline ran out while it
// waited for the dependency. http.Client.Timeout and a gRPC per-call
// deadline both end a call this way, because both are deadlines on the
// context the request is made with, which is the context the Breaker
// sees. A dependency that hangs ends every call this way, so a
// classifier that ignores this case never opens the Breaker. The
// deadline may also be one inherited from an inbound request;
// - nil, with an err that wraps context.DeadlineExceeded: a deadline set
// inside the call ran out.
//
// Which of these count against the dependency is the caller's decision.
// This classifier counts every deadline and ignores only the caller giving
// up:
//
// func classify(ctx context.Context, err error) circuit.Outcome {
// switch {
// case err == nil:
// return circuit.Success
// case errors.Is(ctx.Err(), context.Canceled):
// return circuit.Ignored // the caller gave up
// default:
// return circuit.Failure // including every deadline
// }
// }
//
// A result that is not an error, such as an HTTP status code, reaches
// classify as an error value that the caller builds, such as a
// StatusError{Code: 503} passed to Ticket.Done.
//
// # Counts and policy
//
// circuit decides which counts a Breaker keeps, and the caller decides what
// they mean. [Counts] holds the successes and failures among the latest
// [Config.Window] results, the length of the current run of successes or of
// failures, which the window does not bound, and the probes that timed out.
// Every change of state starts the Counts again from zero.
//
// A [Policy] decides from the Counts alone, so it is a pure function of a
// value, and can be tested without a Breaker:
//
// circuit.Policy{
// // Open after 5 failures in a row, or when half of the last 20 or
// // more results failed.
// Trip: func(c circuit.Counts) bool {
// n := c.Successes + c.Failures
// return c.ConsecutiveFailures >= 5 || (n >= 20 && 2*c.Failures >= n)
// },
// // Reopen on any failed probe, close after 3 successful ones, and
// // keep probing otherwise, including after a probe times out.
// Probe: func(c circuit.Counts) circuit.State {
// switch {
// case c.Failures > 0:
// return circuit.Open
// case c.ConsecutiveSuccesses >= 3:
// return circuit.Closed
// default:
// return circuit.HalfOpen
// }
// },
// }
//
// The window counts results, not time. At low traffic the latest Window
// results can reach back a long way, so a policy that must react only to
// recent results relies on the consecutive counts. circuit ships no
// Policy.
//
// Policy functions run with the Breaker locked, so they must be fast and
// must not call the Breaker.
//
// # Late results
//
// A result belongs to the state its call was admitted in. When it arrives
// after the Breaker has changed state, for example a slow call admitted
// while closed that finishes after the Breaker opened, it is discarded:
// [Ticket.Done] reports false, and the Counts of the newer state are
// untouched. A late result never closes, reopens or trips a Breaker in a
// state it was not admitted in.
//
// # Panics and lost results
//
// circuit never recovers a panic. A panic is a bug in the caller, not a sign
// that the dependency is unhealthy, so when the function passed to Do
// panics or calls runtime.Goexit, Do frees the slot the call held, counts
// nothing, and lets the panic continue. So does a classifier that panics or
// returns an invalid [Outcome]. The slot is freed in every case, so a probe
// that panics, and is recovered higher up as net/http's server does, does
// not keep the Breaker from probing again. Code that uses Allow gets the
// same guarantee by deferring [Ticket.Release] behind a flag, which is safe
// while a panic unwinds because Release calls no function of the caller's.
//
// No such guard exists for a Ticket that is never settled, nor for a call
// that never returns. A probe therefore holds its slot for at most
// [Config.ProbeTimeout]. A probe still in flight by then stops holding it,
// so that another probe can be admitted, and is counted in
// [Counts.TimedOut], and Policy.Probe decides what that means: a Policy
// that reopens when a probe times out treats a hung dependency as failing,
// and one that keeps probing waits for results. The timeout does not
// cancel the call, and its result still counts if it arrives before the
// Breaker changes state, so a dependency slower than ProbeTimeout can
// still close the Breaker. While closed, a Ticket that is never settled
// costs nothing.
//
// # Observing transitions
//
// The observe function passed to New receives every [Transition], one at a
// time and in order: each one's Generation is one more than the last one's,
// and its From is the last one's To. observe is never called with the
// Breaker locked, and never concurrently with itself, so it may call the
// Breaker's methods, and an observer that sets a gauge needs no lock of its
// own. observe receives the context of the call that made the change,
// which may already be done.
//
// A Transition is delivered before the call that made it returns, unless
// another goroutine is already delivering, in which case that goroutine
// delivers it. If observe panics, the panic continues, and the Transitions
// still queued are delivered by the next call to Do, Allow or Ticket.Done,
// which are the calls that make Transitions. A call to Do or Allow delivers
// what is queued before it is admitted, so observe never runs while that
// call holds a probe slot.
//
// Results and rejections are not observed, because the caller already sees
// them: classify receives every result, and Do and Allow return every
// rejection. A framework counts them by wrapping the classifier and by
// checking for ErrRejected. circuit ships no observe function.
//
// # Time
//
// circuit reads time.Now and nothing else: no clock is injected, and no
// timer or goroutine runs in the background. Every period is measured with
// the monotonic clock. A change that time makes due, becoming half-open or
// a probe timing out, is made by the next call to Do, Allow or Ticket.Done,
// as of the moment it fell due, so the Breaker behaves the same whichever
// call makes it. [Breaker.State] reports the state as time has left it
// without making the change, so a metrics gauge that reads it shows
// half-open as soon as the open period has passed, with no traffic.
//
// A test drives a Breaker through its states inside a testing/synctest
// bubble, advancing time with time.Sleep.
//
// # Out of scope
//
// circuit does not:
//
// - retry, back off, time out, rate limit or bound the concurrency of
// calls;
// - share a Breaker's state between processes: each process has its own
// Breakers;
// - keep a registry of named Breakers: the caller holds each Breaker it
// builds;
// - ship transports, policies, classifiers or observers.
package circuit
import (
"context"
"errors"
"time"
)
// State is the state of a [Breaker]. The zero State is not a state, so a
// [Policy.Probe] that returns one by mistake panics instead of choosing a
// state silently.
type State int
const (
// Closed admits every call and counts its result.
Closed State = iota + 1
// Open rejects every call until Config.OpenFor has passed.
Open
// HalfOpen admits at most Config.Probes calls at once, the probes, and
// counts their results.
HalfOpen
)
// String returns "closed", "open" or "half-open", or "State(n)" for a
// value that is not a State.
func (s State) String() string
// Outcome is how the result of one call counts. The zero Outcome is not an
// outcome, so a classifier that returns one by mistake panics instead of
// counting the result silently.
type Outcome int
const (
// Success counts toward the dependency being healthy.
Success Outcome = iota + 1
// Failure counts toward the dependency being unhealthy.
Failure
// Ignored is evidence of nothing. It is not counted, it neither extends
// nor ends a run of results, no Policy function is asked about it, and
// it frees the probe slot it held.
Ignored
)
// String returns "success", "failure" or "ignored", or "Outcome(n)" for a
// value that is not an Outcome.
func (o Outcome) String() string
// MaxCount is the largest Config.Window and Config.Probes that New accepts.
// Both are allocated when the Breaker is built, so the bound turns a typo in
// configuration into an error from New instead of a crash at startup.
const MaxCount = 1 << 16
// Config holds the settings of a Breaker that come from configuration. Every
// field is required: New rejects a zero or out-of-range value rather than
// replacing it with a default.
type Config struct {
// Window is how many of the latest successes and failures in the
// current state Counts.Successes and Counts.Failures cover, from 1 to
// MaxCount.
Window int
// Probes is how many calls a half-open Breaker has in flight at once,
// from 1 to MaxCount.
Probes int
// OpenFor is how long an open Breaker rejects every call before it
// becomes half-open. It must be positive.
OpenFor time.Duration
// ProbeTimeout is how long a probe holds its slot. A probe that has
// not been settled by then stops holding it, so that another probe can
// be admitted, and is counted in Counts.TimedOut. The call itself is
// not cancelled, and its result still counts if it arrives before the
// Breaker changes state. It must be positive.
ProbeTimeout time.Duration
}
// Counts are what a Policy decides from. They cover the current state of a
// Breaker only: every change of state starts them again from zero, and a
// result that arrives after the state its call was admitted in has ended is
// not counted.
type Counts struct {
// Successes and Failures count the outcomes among the latest
// Config.Window successes and failures, so their sum never exceeds
// Config.Window.
Successes int
Failures int
// ConsecutiveSuccesses and ConsecutiveFailures are the length of the
// current run of successes or of failures. They are not bounded by the
// window, so a run longer than Config.Window still trips a Policy that
// waits for it, and they stop at math.MaxInt rather than wrap. At most
// one of them is non-zero. An Ignored result neither extends nor ends a
// run.
ConsecutiveSuccesses int
ConsecutiveFailures int
// TimedOut counts the probes that held their slot for
// Config.ProbeTimeout without being settled. It is zero while closed.
// A probe that timed out still counts as a success or a failure if its
// result arrives before the Breaker changes state.
TimedOut int
}
// Policy decides when a Breaker changes state. Both functions are required.
// circuit ships no Policy: how many failures are too many is the caller's
// decision.
//
// They are called with the Breaker's lock held, with a copy of the Counts,
// on the goroutine of the call that was just counted, or, after a probe
// times out, of the call to Do, Allow or Ticket.Done that found it timed
// out, which may be a call of another goroutine's. They must be fast,
// must decide from the Counts alone, must be safe for concurrent use, and
// must not call the Breaker. A Policy function that panics leaves the
// Breaker consistent: what was counted stays counted, and the state does
// not change.
type Policy struct {
// Trip is called after every success or failure counted while closed.
// It reports whether the Breaker opens.
Trip func(Counts) bool
// Probe is called while half-open after every success or failure of a
// probe, and after every probe that times out. It returns Closed to
// close the Breaker, Open to reopen it, or HalfOpen to keep probing.
// Any other State panics.
Probe func(Counts) State
}
// Transition is a change of state of a Breaker, as observe receives it.
type Transition struct {
From, To State
// Generation numbers the Transitions of a Breaker from 1. Each one's
// is one more than the one delivered before it.
Generation uint64
// Counts are the final Counts of the state the Breaker left, including
// the result or timeout that made the Policy move it. They are zero
// for a move from Open to HalfOpen, which time makes.
Counts Counts
}
// ErrRejected is matched, with errors.Is, by every error that Breaker.Do and
// Breaker.Allow return for a call they do not admit.
var ErrRejected = errors.New("circuit: call rejected")
// RejectedError is the error of a call the Breaker does not admit. It
// matches ErrRejected.
type RejectedError struct {
// State is Open, or HalfOpen when every probe slot is taken.
State State
// RetryAfter is how much longer the Breaker stays open, measured when
// the call was rejected. It is zero when State is HalfOpen, since a slot
// is freed as soon as a probe is settled.
RetryAfter time.Duration
}
// Error implements the error interface. It renders as
// "circuit: call rejected: open for another 4.2s" or
// "circuit: call rejected: half-open, and every probe slot is taken".
func (e *RejectedError) Error() string
// Unwrap returns ErrRejected.
func (e *RejectedError) Unwrap() error
// Breaker is a circuit breaker for one dependency. Build one with New at
// startup and share it between every goroutine that calls that dependency.
// Its fields are unexported, so New is the only way to build one, and every
// method of a Breaker not built by New panics.
type Breaker struct {
// unexported
}
// New returns a closed Breaker with the settings cfg, that classifies every
// result with classify, changes state as p decides, and reports every
// Transition to observe.
//
// classify and observe must be safe for concurrent use. classify is called
// on the goroutine of the call it classifies, with the Breaker unlocked.
// observe is called with the Breaker unlocked, one Transition at a time and
// in order, so it may call the Breaker's methods.
//
// New panics if p.Trip, p.Probe, classify or observe is nil, which is a
// programming error. It returns an error naming every invalid field of cfg,
// because cfg comes from configuration and App.Run should exit non-zero at
// startup instead.
func New(cfg Config, p Policy, classify func(context.Context, error) Outcome, observe func(context.Context, Transition)) (*Breaker, error)
// Do calls fn with ctx if b admits the call, classifies the error fn
// returns together with ctx, and returns that error unchanged. If b does not
// admit the call, Do returns a [*RejectedError] without calling fn.
//
// If fn panics or calls runtime.Goexit, Do settles the call as with
// Ticket.Release before the panic continues: its slot is freed, it is not
// counted, and no function of the caller's runs on the way out. A
// classifier that panics or returns an invalid Outcome is treated the same
// way. Do never recovers a panic.
//
// Do panics if ctx or fn is nil, or if b was not built by New.
func (b *Breaker) Do(ctx context.Context, fn func(context.Context) error) error
// Allow admits a call whose result is known only later, such as a response
// whose body the caller is still reading, and returns the [*Ticket] that
// settles it. If b does not admit the call, Allow returns a nil Ticket and a
// [*RejectedError].
//
// ctx is the context of the call. The classifier receives it when the
// Ticket is settled with Ticket.Done, and observe receives it for any
// Transition that Allow makes.
//
// Allow panics if ctx is nil, or if b was not built by New.
func (b *Breaker) Allow(ctx context.Context) (*Ticket, error)
// State returns the state of b as time has left it. Once b has been open for
// Config.OpenFor, State returns HalfOpen, although b makes that Transition,
// and calls observe, only when the next call to Do, Allow or Ticket.Done
// arrives. State does not ask the Policy about a probe that has timed out:
// that call does, and may find b open again. State changes nothing and calls
// no function of the caller's, so a metrics callback may call it.
//
// State panics if b was not built by New.
func (b *Breaker) State() State
// Ticket settles one call admitted by Breaker.Allow. Settle it exactly once:
// with Done when the result is known, or with Release when there is no
// result to report. A Ticket may be settled from any goroutine, and
// settling it twice panics, so code that may settle it from several places,
// such as both Read and Close of a response body, guards it with a
// sync.Once.
//
// A Ticket that is never settled costs nothing while the Breaker is closed.
// While it is half-open, the probe holds its slot for at most
// Config.ProbeTimeout.
type Ticket struct {
// unexported
}
// Done classifies err with the context passed to Allow, and counts it. It
// reports whether the result reached the state the call was admitted in:
// false means the Breaker had changed state since, and the result was
// discarded.
//
// A classifier that panics or returns an invalid Outcome frees t's slot,
// as Release does, before the panic continues. Done panics if t is already
// settled, or if t was not returned by Breaker.Allow.
func (t *Ticket) Done(err error) bool
// Release settles t without a result, as if the classifier had returned
// Ignored, and without calling it. It is for a call that produced no
// result, such as one whose code panicked before the result was known.
// Every result goes through Done instead, so that one classifier decides
// how it counts.
//
// Release calls no function of the caller's: not the classifier, the
// Policy or observe. It is therefore safe to defer as a guard that frees
// the slot if the code between Allow and settling panics, as Do does:
//
// handedOff := false
// defer func() {
// if !handedOff {
// t.Release()
// }
// }()
// resp, err := next.RoundTrip(req)
// handedOff = true
//
// Release panics if t is already settled, or if t was not returned by
// Breaker.Allow.
func (t *Ticket) Release()Examples: Do, and Allow with a streamed body// classify counts every error against the dependency, every deadline
// included, except one the caller caused by giving up.
func classify(ctx context.Context, err error) circuit.Outcome {
switch {
case err == nil:
return circuit.Success
case errors.Is(ctx.Err(), context.Canceled):
return circuit.Ignored
default:
return circuit.Failure
}
}
// consecutive opens after n failures in a row, reopens on any failed probe,
// and closes after n successful probes in a row.
func consecutive(n int) circuit.Policy {
return circuit.Policy{
Trip: func(c circuit.Counts) bool { return c.ConsecutiveFailures >= n },
Probe: func(c circuit.Counts) circuit.State {
switch {
case c.Failures > 0:
return circuit.Open
case c.ConsecutiveSuccesses >= n:
return circuit.Closed
default:
return circuit.HalfOpen
}
},
}
}
func printTransition(_ context.Context, t circuit.Transition) {
fmt.Printf("observe: #%d %v -> %v after %d failed, %d succeeded\n",
t.Generation, t.From, t.To, t.Counts.Failures, t.Counts.Successes)
}
// cfg would come from configuration. Its open period is short so that the
// examples run quickly.
var cfg = circuit.Config{Window: 20, Probes: 1, OpenFor: 10 * time.Millisecond, ProbeTimeout: time.Second}
// Two failures in a row open the Breaker, and it rejects calls without
// making them until its open period has passed. Then it is half-open, and
// two successful probes close it.
func Example() {
ctx := context.Background()
// At startup.
b, err := circuit.New(cfg, consecutive(2), classify, printTransition)
if err != nil {
fmt.Println(err)
return
}
// Per outbound call.
report := func(err error) {
if rej, ok := errors.AsType[*circuit.RejectedError](err); ok {
fmt.Println("do: rejected while", rej.State)
return
}
fmt.Println("do:", err)
}
unavailable := errors.New("payments: unavailable")
for _, result := range []error{unavailable, unavailable, nil} {
report(b.Do(ctx, func(context.Context) error { return result }))
}
time.Sleep(cfg.OpenFor)
fmt.Println("state:", b.State())
for range 2 {
report(b.Do(ctx, func(context.Context) error { return nil }))
}
fmt.Println("state:", b.State())
// Output:
// do: payments: unavailable
// observe: #1 closed -> open after 2 failed, 0 succeeded
// do: payments: unavailable
// do: rejected while open
// state: half-open
// observe: #2 open -> half-open after 0 failed, 0 succeeded
// do: <nil>
// observe: #3 half-open -> closed after 0 failed, 2 succeeded
// do: <nil>
// state: closed
}
// body settles its Ticket once, whichever comes first: the end of the body,
// an error while reading it, or Close.
type body struct {
io.ReadCloser
once sync.Once
ticket *circuit.Ticket
}
func (b *body) settle(err error) { b.once.Do(func() { b.ticket.Done(err) }) }
func (b *body) Read(p []byte) (int, error) {
n, err := b.ReadCloser.Read(p)
switch {
case errors.Is(err, io.EOF):
b.settle(nil)
case err != nil:
b.settle(err) // the body failed partway through
}
return n, err
}
func (b *body) Close() error {
b.settle(nil) // the caller stopped reading, having got what it needed
return b.ReadCloser.Close()
}
type errReader struct{ err error }
func (r errReader) Read([]byte) (int, error) { return 0, r.err }
// A response whose body fails while the caller reads it counts as a
// failure. Allow admits the call when it starts, and the body settles the
// Ticket when the result is known.
func ExampleBreaker_Allow() {
b, err := circuit.New(cfg, consecutive(1), classify, printTransition)
if err != nil {
fmt.Println(err)
return
}
ticket, err := b.Allow(context.Background())
if err != nil {
fmt.Println(err)
return
}
resp := &body{
ReadCloser: io.NopCloser(io.MultiReader(
strings.NewReader("first chunk, "),
errReader{errors.New("connection reset")},
)),
ticket: ticket,
}
got, err := io.ReadAll(resp)
fmt.Printf("read %q: %v\n", got, err)
fmt.Println("close:", resp.Close())
fmt.Println("state:", b.State())
// Output:
// observe: #1 closed -> open after 1 failed, 0 succeeded
// read "first chunk, ": connection reset
// close: <nil>
// state: open
}humus sketch: config, a policy builder, classifiers, RoundTripper, gRPC interceptors, OTel// Package humuscircuit sketches the wiring z5labs/humus builds on
// github.com/z5labs/bedrock/circuit, using only its exported API:
// configuration, policies, classifiers, an http.RoundTripper, gRPC client
// interceptors and OpenTelemetry instruments. In humus these would be split
// by dependency (circuithttp, circuitgrpc, circuitotel); they share one file
// here so the sketch compiles as one package. It builds against grpc
// v1.84.0, otel v1.47.0 and bedrock/config. humus_test.go (in-memory, synctest,
// otel sdk/metric ManualReader) pins: a body read error after a 200 opens the
// breaker; http.Client.Timeout opens it; status-only callers (204, close
// without reading) close a half-open breaker; a panicking Next frees the probe;
// a 101 body stays an io.ReadWriteCloser; a rejected unary call is
// codes.Unavailable and still matches circuit.ErrRejected; CloseAndRecv settles
// a client-streaming call; a missed stream deadline counts and a cancelled
// stream is ignored; the state gauge reads half-open with no traffic.
package humuscircuit
import (
"context"
"errors"
"fmt"
"io"
"net/http"
"strconv"
"sync"
"time"
"github.com/z5labs/bedrock"
"github.com/z5labs/bedrock/circuit"
"github.com/z5labs/bedrock/config"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
// ---------------------------------------------------------------------------
// Configuration. circuit.Config is plain data, so config.Struct builds it
// directly and reports every missing or malformed variable at once;
// circuit.New then reports every value out of range. Any default is humus's
// to write out here with .Default(...); bedrock has none.
// Settings is everything a humus breaker reads from its environment.
type Settings struct {
Circuit circuit.Config
MinResults int // results the window needs before FailureRatio applies
FailureRatio float64 // share of failures in the window that opens the breaker
CloseAfter int // consecutive successful probes that close it
}
// ReadSettings reads the Settings of the breaker whose variables start with
// prefix, such as "PAYMENTS_BREAKER_".
func ReadSettings(env bedrock.Environment, prefix string) config.Value[Settings] {
return config.Struct(func(r *config.Resolver) Settings {
v := func(name string) config.Value[string] { return config.Env(env, prefix+name) }
return Settings{
Circuit: circuit.Config{
Window: r.Get(v("WINDOW").Parse(strconv.Atoi)),
Probes: r.Get(v("PROBES").Parse(strconv.Atoi)),
OpenFor: r.Get(v("OPEN_FOR").Parse(time.ParseDuration)),
ProbeTimeout: r.Get(v("PROBE_TIMEOUT").Parse(time.ParseDuration)),
},
MinResults: r.Get(v("MIN_RESULTS").Parse(strconv.Atoi)),
FailureRatio: r.Get(v("FAILURE_RATIO").Parse(config.ParseFloat64)),
CloseAfter: r.Get(v("CLOSE_AFTER").Parse(strconv.Atoi)),
}
})
}
// ---------------------------------------------------------------------------
// Policies: humus's choices, not bedrock's. A Policy is opaque to circuit,
// so a builder checks what circuit cannot, such as a threshold the window
// can never reach.
// FailureRate opens once the window holds at least minResults results and
// at least ratio of them failed, reopens on any failed probe, closes after
// closeAfter successful probes in a row, and keeps probing when a probe
// times out, so a probe that is slow or lost never reopens the breaker.
// Pair it with a client timeout shorter than ProbeTimeout: the timeout turns
// a dependency that hangs into failed probes, which reopen the breaker,
// and leaves ProbeTimeout to free the slots of Tickets that were lost.
func FailureRate(cfg circuit.Config, minResults int, ratio float64, closeAfter int) (circuit.Policy, error) {
var errs []error
if minResults < 1 || minResults > cfg.Window {
errs = append(errs, fmt.Errorf("circuit policy: minimum results must be between 1 and the window, %d, got %d", cfg.Window, minResults))
}
if ratio <= 0 || ratio > 1 {
errs = append(errs, fmt.Errorf("circuit policy: failure ratio must be in (0, 1], got %v", ratio))
}
if closeAfter < 1 {
errs = append(errs, fmt.Errorf("circuit policy: close after must be at least 1, got %d", closeAfter))
}
if err := errors.Join(errs...); err != nil {
return circuit.Policy{}, err
}
return circuit.Policy{
Trip: func(c circuit.Counts) bool {
n := c.Successes + c.Failures
return n >= minResults && float64(c.Failures) >= ratio*float64(n)
},
Probe: func(c circuit.Counts) circuit.State {
switch {
case c.Failures > 0:
return circuit.Open
case c.ConsecutiveSuccesses >= closeAfter:
return circuit.Closed
default:
return circuit.HalfOpen
}
},
}, nil
}
// ---------------------------------------------------------------------------
// Classifiers. Both ignore only the caller giving up. http.Client.Timeout
// and a gRPC per-call deadline are deadlines on the context the breaker
// sees, so a dependency that hangs ends with ctx.Err() ==
// context.DeadlineExceeded, which counts as a failure.
// StatusError is the result Transport reports for a response with a status
// of 400 or more, so that the classifier decides which statuses count.
// RoundTrip still returns the response to its caller.
type StatusError struct{ Code int }
func (e *StatusError) Error() string { return "http: status " + strconv.Itoa(e.Code) }
// ClassifyHTTP counts transport errors, body read errors, 5xx and 429 as
// failures, and every other status as a success: the server answered.
func ClassifyHTTP(ctx context.Context, err error) circuit.Outcome {
switch {
case err == nil:
return circuit.Success
case errors.Is(ctx.Err(), context.Canceled):
return circuit.Ignored // the caller gave up
}
if se, ok := errors.AsType[*StatusError](err); ok && se.Code < 500 && se.Code != http.StatusTooManyRequests {
return circuit.Success
}
return circuit.Failure
}
// ClassifyGRPC counts the codes that mean the server is unhealthy or
// overloaded as failures, and every other answer, such as InvalidArgument
// or NotFound, as a success.
func ClassifyGRPC(ctx context.Context, err error) circuit.Outcome {
switch {
case err == nil:
return circuit.Success
case errors.Is(ctx.Err(), context.Canceled):
return circuit.Ignored
}
switch status.Code(err) {
case codes.Unavailable, codes.DeadlineExceeded, codes.ResourceExhausted,
codes.Internal, codes.Unknown, codes.Aborted, codes.DataLoss:
return circuit.Failure
}
return circuit.Success
}
// ---------------------------------------------------------------------------
// net/http. The result of a call is known only once the caller has read the
// body, so Transport uses Allow, and the body settles the Ticket.
// Transport is an http.RoundTripper that sends requests through Breaker.
type Transport struct {
Breaker *circuit.Breaker
Next http.RoundTripper
}
// RoundTrip implements http.RoundTripper. A rejected request fails with a
// *circuit.RejectedError, which http.Client wraps in a *url.Error; both
// match circuit.ErrRejected.
func (t Transport) RoundTrip(req *http.Request) (*http.Response, error) {
ticket, err := t.Breaker.Allow(req.Context())
if err != nil {
return nil, err
}
handedOff := false
defer func() {
if !handedOff {
ticket.Release() // Next panicked or called runtime.Goexit
}
}()
resp, err := t.Next.RoundTrip(req)
handedOff = true
switch {
case err != nil:
ticket.Done(err)
return nil, err
case resp.StatusCode >= 400:
// Settled now: a caller that only checks the status may never read
// or close the body.
ticket.Done(&StatusError{Code: resp.StatusCode})
return resp, nil
case resp.StatusCode == http.StatusSwitchingProtocols:
ticket.Done(nil) // the body is an io.ReadWriteCloser now; leave it be
return resp, nil
}
resp.Body = &body{ReadCloser: resp.Body, ticket: ticket}
return resp, nil
}
// body settles its Ticket exactly once, whichever comes first of the end of
// the body, an error while reading it, and Close. The Ticket panics on a
// second settle, so the sync.Once is required.
type body struct {
io.ReadCloser
ticket *circuit.Ticket
once sync.Once
}
func (b *body) settle(err error) { b.once.Do(func() { b.ticket.Done(err) }) }
func (b *body) Read(p []byte) (int, error) {
n, err := b.ReadCloser.Read(p)
switch {
case errors.Is(err, io.EOF):
b.settle(nil)
case err != nil:
b.settle(err) // a reset partway through; Ignored if the caller cancelled
}
return n, err
}
func (b *body) Close() error {
err := b.ReadCloser.Close()
// Closed before the end: the healthy status line is all there is to go
// on. json.Decoder, a HEAD or a 204 often never reaches EOF.
b.settle(nil)
return err
}
// ---------------------------------------------------------------------------
// gRPC.
// rejected is a rejection as a gRPC status, codes.Unavailable, that still
// matches circuit.ErrRejected and *circuit.RejectedError.
type rejected struct{ *circuit.RejectedError }
func (r rejected) GRPCStatus() *status.Status { return status.New(codes.Unavailable, r.Error()) }
func (r rejected) Unwrap() error { return r.RejectedError }
func asStatus(err error) error {
if rej, ok := errors.AsType[*circuit.RejectedError](err); ok {
return rejected{rej}
}
return err
}
// UnaryClientInterceptor sends every unary call through b. The result is
// known when invoker returns, so it uses Do, which also frees the slot if
// invoker panics.
func UnaryClientInterceptor(b *circuit.Breaker) grpc.UnaryClientInterceptor {
return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
return asStatus(b.Do(ctx, func(ctx context.Context) error {
return invoker(ctx, method, req, reply, cc, opts...)
}))
}
}
// StreamClientInterceptor sends every stream through b with Allow. A stream
// is settled by the first of: a message received (the server answered,
// which also settles a client-streaming call's single response and a
// long-lived watch), the end of the stream, an error from RecvMsg, SendMsg
// or Header, or the end of its context. Each goes through the classifier,
// so a stream whose deadline passes counts as a failure, as a unary call's
// does, and one its caller cancels is ignored.
func StreamClientInterceptor(b *circuit.Breaker) grpc.StreamClientInterceptor {
return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
ticket, err := b.Allow(ctx)
if err != nil {
return nil, asStatus(err)
}
handedOff := false
defer func() {
if !handedOff {
ticket.Release() // streamer panicked or called runtime.Goexit
}
}()
cs, err := streamer(ctx, desc, cc, method, opts...)
handedOff = true
if err != nil {
ticket.Done(err)
return nil, err
}
s := &clientStream{ClientStream: cs, ticket: ticket}
// If ctx is already done, the function runs at once, on another
// goroutine, so it must not read s.stop.
s.stop = context.AfterFunc(ctx, func() { s.done(status.FromContextError(context.Cause(ctx)).Err()) })
return s, nil
}
}
type clientStream struct {
grpc.ClientStream
ticket *circuit.Ticket
once sync.Once
stop func() bool // stops settling s when its context ends
}
// settle settles the stream from one of its methods, which the caller can
// call only once s.stop is set.
func (s *clientStream) settle(err error) {
s.stop()
s.done(err)
}
func (s *clientStream) done(err error) { s.once.Do(func() { s.ticket.Done(err) }) }
func (s *clientStream) RecvMsg(m any) error {
err := s.ClientStream.RecvMsg(m)
if errors.Is(err, io.EOF) {
s.settle(nil)
} else {
s.settle(err) // nil: a message arrived
}
return err
}
func (s *clientStream) SendMsg(m any) error {
err := s.ClientStream.SendMsg(m)
if err != nil && !errors.Is(err, io.EOF) {
s.settle(err) // io.EOF means the status is waiting in RecvMsg
}
return err
}
func (s *clientStream) Header() (metadata.MD, error) {
md, err := s.ClientStream.Header()
if err != nil {
s.settle(err)
}
return md, err
}
// ---------------------------------------------------------------------------
// OpenTelemetry. Transitions come from observe, the state from Breaker.State,
// outcomes from wrapping the classifier, and rejections from where Do and
// Allow return them, so bedrock needs no per-call event.
// Observer returns an observe function for circuit.New that counts
// transitions. observe receives them one at a time and in order.
func Observer(meter metric.Meter, name string) (func(context.Context, circuit.Transition), error) {
transitions, err := meter.Int64Counter("circuit.transitions",
metric.WithDescription("Changes of state of a circuit breaker."))
if err != nil {
return nil, err
}
breaker := attribute.String("circuit.name", name)
return func(ctx context.Context, t circuit.Transition) {
// ctx may already be done; a metric does not care, so drop its
// cancellation rather than its values.
transitions.Add(context.WithoutCancel(ctx), 1, metric.WithAttributes(breaker,
attribute.String("circuit.from", t.From.String()),
attribute.String("circuit.to", t.To.String()),
attribute.Bool("circuit.probe_timed_out", t.Counts.TimedOut > 0),
))
}, nil
}
// CountOutcomes wraps classify so that every classified result is counted
// by outcome.
func CountOutcomes(meter metric.Meter, name string, classify func(context.Context, error) circuit.Outcome) (func(context.Context, error) circuit.Outcome, error) {
outcomes, err := meter.Int64Counter("circuit.outcomes",
metric.WithDescription("Results of calls through a circuit breaker, by outcome."))
if err != nil {
return nil, err
}
breaker := attribute.String("circuit.name", name)
return func(ctx context.Context, err error) circuit.Outcome {
o := classify(ctx, err)
outcomes.Add(context.WithoutCancel(ctx), 1, metric.WithAttributes(breaker, attribute.String("circuit.outcome", o.String())))
return o
}, nil
}
// RegisterState reports the state of b on every collection. Breaker.State
// is a read that reports half-open as soon as the open period has passed,
// with no traffic, and calls none of the breaker's functions, so it is safe
// in a collection callback.
func RegisterState(meter metric.Meter, name string, b *circuit.Breaker) (metric.Registration, error) {
state, err := meter.Int64ObservableGauge("circuit.state",
metric.WithDescription("1 for the state the circuit breaker is in, 0 for the others."))
if err != nil {
return nil, err
}
breaker := attribute.String("circuit.name", name)
return meter.RegisterCallback(func(_ context.Context, o metric.Observer) error {
current := b.State()
for _, s := range []circuit.State{circuit.Closed, circuit.Open, circuit.HalfOpen} {
v := int64(0)
if s == current {
v = 1
}
o.ObserveInt64(state, v, metric.WithAttributes(breaker, attribute.String("circuit.state", s.String())))
}
return nil
}, state)
}
// NewBreaker builds the breaker named name from its environment, with the
// OTel instruments of meter. It is called once per dependency in App.Run.
func NewBreaker(ctx context.Context, env bedrock.Environment, meter metric.Meter, name, prefix string, classify func(context.Context, error) circuit.Outcome) (*circuit.Breaker, error) {
s, err := ReadSettings(env, prefix).Read(ctx)
if err != nil {
return nil, err
}
policy, err := FailureRate(s.Circuit, s.MinResults, s.FailureRatio, s.CloseAfter)
if err != nil {
return nil, err
}
observe, err := Observer(meter, name)
if err != nil {
return nil, err
}
classify, err = CountOutcomes(meter, name, classify)
if err != nil {
return nil, err
}
b, err := circuit.New(s.Circuit, policy, classify, observe)
if err != nil {
return nil, err
}
if _, err := RegisterState(meter, name, b); err != nil {
return nil, err
}
return b, nil
}
// Wiring at startup, inside App.Run:
//
// b, err := humuscircuit.NewBreaker(ctx, p.Environment, meter, "payments", "PAYMENTS_BREAKER_", humuscircuit.ClassifyHTTP)
// client := &http.Client{Timeout: 5 * time.Second, Transport: humuscircuit.Transport{Breaker: b, Next: http.DefaultTransport}} |
Revision:
|
|
Closing as resolved. The design is settled, and it's filed as three stories. Each is blocked by the one before it:
Settled decisions:
The answers to the seven design questions are in the API proposal and the revision. The stories embed the source they need. |
0 replies
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
Many circuit breakers in the ecosystem have not received updates recently which is expected since a circuit breaker is a static architecture and once mature shouldn't require many updates. Unfortunately, many companies require dependencies to have an "recent" update so to help alleviate this somewhat, bedrock could provide a circuitbreaker package which would be "updated" every time a new bedrock release is made. Another reason for the request is to have a circuitbreaker implementation which better follows the ethos of bedrock
All reactions