From 779869fbe8081f84c3f044a60defd0d5e94252c7 Mon Sep 17 00:00:00 2001 From: Frank Schroeder Date: Fri, 29 Dec 2017 13:44:33 +0100 Subject: [PATCH 1/3] Vendoring in github.com/rubyist/circuitbreaker plus dependencies --- NOTICES.txt | 19 + vendor/github.com/cenk/backoff/LICENSE | 20 + vendor/github.com/cenk/backoff/README.md | 30 ++ vendor/github.com/cenk/backoff/backoff.go | 66 +++ vendor/github.com/cenk/backoff/context.go | 60 +++ vendor/github.com/cenk/backoff/exponential.go | 158 +++++++ vendor/github.com/cenk/backoff/retry.go | 78 +++ vendor/github.com/cenk/backoff/ticker.go | 84 ++++ vendor/github.com/cenk/backoff/tries.go | 35 ++ vendor/github.com/facebookgo/clock/LICENSE | 21 + vendor/github.com/facebookgo/clock/README.md | 104 ++++ vendor/github.com/facebookgo/clock/clock.go | 363 ++++++++++++++ .../rubyist/circuitbreaker/CHANGELOG.md | 230 +++++++++ .../github.com/rubyist/circuitbreaker/LICENSE | 22 + .../rubyist/circuitbreaker/README.md | 120 +++++ .../rubyist/circuitbreaker/circuitbreaker.go | 446 ++++++++++++++++++ .../rubyist/circuitbreaker/client.go | 170 +++++++ .../rubyist/circuitbreaker/panel.go | 144 ++++++ .../rubyist/circuitbreaker/window.go | 174 +++++++ vendor/golang.org/x/net/context/context.go | 56 +++ vendor/golang.org/x/net/context/go17.go | 72 +++ vendor/golang.org/x/net/context/go19.go | 20 + vendor/golang.org/x/net/context/pre_go17.go | 300 ++++++++++++ vendor/golang.org/x/net/context/pre_go19.go | 109 +++++ vendor/vendor.json | 4 + 25 files changed, 2905 insertions(+) create mode 100644 vendor/github.com/cenk/backoff/LICENSE create mode 100644 vendor/github.com/cenk/backoff/README.md create mode 100644 vendor/github.com/cenk/backoff/backoff.go create mode 100644 vendor/github.com/cenk/backoff/context.go create mode 100644 vendor/github.com/cenk/backoff/exponential.go create mode 100644 vendor/github.com/cenk/backoff/retry.go create mode 100644 vendor/github.com/cenk/backoff/ticker.go create mode 100644 vendor/github.com/cenk/backoff/tries.go create mode 100644 vendor/github.com/facebookgo/clock/LICENSE create mode 100644 vendor/github.com/facebookgo/clock/README.md create mode 100644 vendor/github.com/facebookgo/clock/clock.go create mode 100644 vendor/github.com/rubyist/circuitbreaker/CHANGELOG.md create mode 100644 vendor/github.com/rubyist/circuitbreaker/LICENSE create mode 100644 vendor/github.com/rubyist/circuitbreaker/README.md create mode 100644 vendor/github.com/rubyist/circuitbreaker/circuitbreaker.go create mode 100644 vendor/github.com/rubyist/circuitbreaker/client.go create mode 100644 vendor/github.com/rubyist/circuitbreaker/panel.go create mode 100644 vendor/github.com/rubyist/circuitbreaker/window.go create mode 100644 vendor/golang.org/x/net/context/context.go create mode 100644 vendor/golang.org/x/net/context/go17.go create mode 100644 vendor/golang.org/x/net/context/go19.go create mode 100644 vendor/golang.org/x/net/context/pre_go17.go create mode 100644 vendor/golang.org/x/net/context/pre_go19.go diff --git a/NOTICES.txt b/NOTICES.txt index d1ba4a54b4..41c4e20b38 100644 --- a/NOTICES.txt +++ b/NOTICES.txt @@ -153,7 +153,26 @@ https://github.com/rogpeppe/fastuuid.git License: BSD 3-clause (https://github.com/google/uuid/LICENSE) Copyright © 2014, Roger Peppe All rights reserved. + golang.org/x/sync/singleflight https://golang.org/x/sync/singleflight License: BSD 3-clause (https://golang.org/x/sync/LICENSE) Copyright (c) 2009 The Go Authors. All rights reserved. + + +github.com/rubyist/circuitbreaker +https://github.com/rubyist/circuitbreaker +License: MIT (https://github.com/rubyist/circuitbreaker/blob/master/LICENSE) +Copyright (c) 2014 Scott Barron + + +github.com/facebookgo/clock +https://github.com/facebookgo/clock +License: MIT (https://github.com/facebookgo/clock/blob/master/LICENSE) +Copyright (c) 2014 Ben Johnson + + +github.com/cenk/backoff +https://github.com/cenkalti/backoff +License: MIT (https://github.com/cenkalti/backoff/blob/master/LICENSE) +Copyright (c) 2014 Cenk Altı diff --git a/vendor/github.com/cenk/backoff/LICENSE b/vendor/github.com/cenk/backoff/LICENSE new file mode 100644 index 0000000000..89b8179965 --- /dev/null +++ b/vendor/github.com/cenk/backoff/LICENSE @@ -0,0 +1,20 @@ +The MIT License (MIT) + +Copyright (c) 2014 Cenk Altı + +Permission is hereby granted, free of charge, to any person obtaining a copy of +this software and associated documentation files (the "Software"), to deal in +the Software without restriction, including without limitation the rights to +use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software is furnished to do so, +subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS +FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER +IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/vendor/github.com/cenk/backoff/README.md b/vendor/github.com/cenk/backoff/README.md new file mode 100644 index 0000000000..13b347fb95 --- /dev/null +++ b/vendor/github.com/cenk/backoff/README.md @@ -0,0 +1,30 @@ +# Exponential Backoff [![GoDoc][godoc image]][godoc] [![Build Status][travis image]][travis] [![Coverage Status][coveralls image]][coveralls] + +This is a Go port of the exponential backoff algorithm from [Google's HTTP Client Library for Java][google-http-java-client]. + +[Exponential backoff][exponential backoff wiki] +is an algorithm that uses feedback to multiplicatively decrease the rate of some process, +in order to gradually find an acceptable rate. +The retries exponentially increase and stop increasing when a certain threshold is met. + +## Usage + +See https://godoc.org/github.com/cenkalti/backoff#pkg-examples + +## Contributing + +* I would like to keep this library as small as possible. +* Please don't send a PR without opening an issue and discussing it first. +* If proposed change is not a common use case, I will probably not accept it. + +[godoc]: https://godoc.org/github.com/cenkalti/backoff +[godoc image]: https://godoc.org/github.com/cenkalti/backoff?status.png +[travis]: https://travis-ci.org/cenkalti/backoff +[travis image]: https://travis-ci.org/cenkalti/backoff.png?branch=master +[coveralls]: https://coveralls.io/github/cenkalti/backoff?branch=master +[coveralls image]: https://coveralls.io/repos/github/cenkalti/backoff/badge.svg?branch=master + +[google-http-java-client]: https://github.com/google/google-http-java-client +[exponential backoff wiki]: http://en.wikipedia.org/wiki/Exponential_backoff + +[advanced example]: https://godoc.org/github.com/cenkalti/backoff#example_ diff --git a/vendor/github.com/cenk/backoff/backoff.go b/vendor/github.com/cenk/backoff/backoff.go new file mode 100644 index 0000000000..3676ee405d --- /dev/null +++ b/vendor/github.com/cenk/backoff/backoff.go @@ -0,0 +1,66 @@ +// Package backoff implements backoff algorithms for retrying operations. +// +// Use Retry function for retrying operations that may fail. +// If Retry does not meet your needs, +// copy/paste the function into your project and modify as you wish. +// +// There is also Ticker type similar to time.Ticker. +// You can use it if you need to work with channels. +// +// See Examples section below for usage examples. +package backoff + +import "time" + +// BackOff is a backoff policy for retrying an operation. +type BackOff interface { + // NextBackOff returns the duration to wait before retrying the operation, + // or backoff. Stop to indicate that no more retries should be made. + // + // Example usage: + // + // duration := backoff.NextBackOff(); + // if (duration == backoff.Stop) { + // // Do not retry operation. + // } else { + // // Sleep for duration and retry operation. + // } + // + NextBackOff() time.Duration + + // Reset to initial state. + Reset() +} + +// Stop indicates that no more retries should be made for use in NextBackOff(). +const Stop time.Duration = -1 + +// ZeroBackOff is a fixed backoff policy whose backoff time is always zero, +// meaning that the operation is retried immediately without waiting, indefinitely. +type ZeroBackOff struct{} + +func (b *ZeroBackOff) Reset() {} + +func (b *ZeroBackOff) NextBackOff() time.Duration { return 0 } + +// StopBackOff is a fixed backoff policy that always returns backoff.Stop for +// NextBackOff(), meaning that the operation should never be retried. +type StopBackOff struct{} + +func (b *StopBackOff) Reset() {} + +func (b *StopBackOff) NextBackOff() time.Duration { return Stop } + +// ConstantBackOff is a backoff policy that always returns the same backoff delay. +// This is in contrast to an exponential backoff policy, +// which returns a delay that grows longer as you call NextBackOff() over and over again. +type ConstantBackOff struct { + Interval time.Duration +} + +func (b *ConstantBackOff) Reset() {} +func (b *ConstantBackOff) NextBackOff() time.Duration { return b.Interval } + +func NewConstantBackOff(d time.Duration) *ConstantBackOff { + return &ConstantBackOff{Interval: d} +} diff --git a/vendor/github.com/cenk/backoff/context.go b/vendor/github.com/cenk/backoff/context.go new file mode 100644 index 0000000000..5d15709254 --- /dev/null +++ b/vendor/github.com/cenk/backoff/context.go @@ -0,0 +1,60 @@ +package backoff + +import ( + "time" + + "golang.org/x/net/context" +) + +// BackOffContext is a backoff policy that stops retrying after the context +// is canceled. +type BackOffContext interface { + BackOff + Context() context.Context +} + +type backOffContext struct { + BackOff + ctx context.Context +} + +// WithContext returns a BackOffContext with context ctx +// +// ctx must not be nil +func WithContext(b BackOff, ctx context.Context) BackOffContext { + if ctx == nil { + panic("nil context") + } + + if b, ok := b.(*backOffContext); ok { + return &backOffContext{ + BackOff: b.BackOff, + ctx: ctx, + } + } + + return &backOffContext{ + BackOff: b, + ctx: ctx, + } +} + +func ensureContext(b BackOff) BackOffContext { + if cb, ok := b.(BackOffContext); ok { + return cb + } + return WithContext(b, context.Background()) +} + +func (b *backOffContext) Context() context.Context { + return b.ctx +} + +func (b *backOffContext) NextBackOff() time.Duration { + select { + case <-b.Context().Done(): + return Stop + default: + return b.BackOff.NextBackOff() + } +} diff --git a/vendor/github.com/cenk/backoff/exponential.go b/vendor/github.com/cenk/backoff/exponential.go new file mode 100644 index 0000000000..d9de15a177 --- /dev/null +++ b/vendor/github.com/cenk/backoff/exponential.go @@ -0,0 +1,158 @@ +package backoff + +import ( + "math/rand" + "time" +) + +/* +ExponentialBackOff is a backoff implementation that increases the backoff +period for each retry attempt using a randomization function that grows exponentially. + +NextBackOff() is calculated using the following formula: + + randomized interval = + RetryInterval * (random value in range [1 - RandomizationFactor, 1 + RandomizationFactor]) + +In other words NextBackOff() will range between the randomization factor +percentage below and above the retry interval. + +For example, given the following parameters: + + RetryInterval = 2 + RandomizationFactor = 0.5 + Multiplier = 2 + +the actual backoff period used in the next retry attempt will range between 1 and 3 seconds, +multiplied by the exponential, that is, between 2 and 6 seconds. + +Note: MaxInterval caps the RetryInterval and not the randomized interval. + +If the time elapsed since an ExponentialBackOff instance is created goes past the +MaxElapsedTime, then the method NextBackOff() starts returning backoff.Stop. + +The elapsed time can be reset by calling Reset(). + +Example: Given the following default arguments, for 10 tries the sequence will be, +and assuming we go over the MaxElapsedTime on the 10th try: + + Request # RetryInterval (seconds) Randomized Interval (seconds) + + 1 0.5 [0.25, 0.75] + 2 0.75 [0.375, 1.125] + 3 1.125 [0.562, 1.687] + 4 1.687 [0.8435, 2.53] + 5 2.53 [1.265, 3.795] + 6 3.795 [1.897, 5.692] + 7 5.692 [2.846, 8.538] + 8 8.538 [4.269, 12.807] + 9 12.807 [6.403, 19.210] + 10 19.210 backoff.Stop + +Note: Implementation is not thread-safe. +*/ +type ExponentialBackOff struct { + InitialInterval time.Duration + RandomizationFactor float64 + Multiplier float64 + MaxInterval time.Duration + // After MaxElapsedTime the ExponentialBackOff stops. + // It never stops if MaxElapsedTime == 0. + MaxElapsedTime time.Duration + Clock Clock + + currentInterval time.Duration + startTime time.Time + random *rand.Rand +} + +// Clock is an interface that returns current time for BackOff. +type Clock interface { + Now() time.Time +} + +// Default values for ExponentialBackOff. +const ( + DefaultInitialInterval = 500 * time.Millisecond + DefaultRandomizationFactor = 0.5 + DefaultMultiplier = 1.5 + DefaultMaxInterval = 60 * time.Second + DefaultMaxElapsedTime = 15 * time.Minute +) + +// NewExponentialBackOff creates an instance of ExponentialBackOff using default values. +func NewExponentialBackOff() *ExponentialBackOff { + b := &ExponentialBackOff{ + InitialInterval: DefaultInitialInterval, + RandomizationFactor: DefaultRandomizationFactor, + Multiplier: DefaultMultiplier, + MaxInterval: DefaultMaxInterval, + MaxElapsedTime: DefaultMaxElapsedTime, + Clock: SystemClock, + random: rand.New(rand.NewSource(time.Now().UnixNano())), + } + b.Reset() + return b +} + +type systemClock struct{} + +func (t systemClock) Now() time.Time { + return time.Now() +} + +// SystemClock implements Clock interface that uses time.Now(). +var SystemClock = systemClock{} + +// Reset the interval back to the initial retry interval and restarts the timer. +func (b *ExponentialBackOff) Reset() { + b.currentInterval = b.InitialInterval + b.startTime = b.Clock.Now() +} + +// NextBackOff calculates the next backoff interval using the formula: +// Randomized interval = RetryInterval +/- (RandomizationFactor * RetryInterval) +func (b *ExponentialBackOff) NextBackOff() time.Duration { + // Make sure we have not gone over the maximum elapsed time. + if b.MaxElapsedTime != 0 && b.GetElapsedTime() > b.MaxElapsedTime { + return Stop + } + defer b.incrementCurrentInterval() + if b.random == nil { + b.random = rand.New(rand.NewSource(time.Now().UnixNano())) + } + return getRandomValueFromInterval(b.RandomizationFactor, b.random.Float64(), b.currentInterval) +} + +// GetElapsedTime returns the elapsed time since an ExponentialBackOff instance +// is created and is reset when Reset() is called. +// +// The elapsed time is computed using time.Now().UnixNano(). It is +// safe to call even while the backoff policy is used by a running +// ticker. +func (b *ExponentialBackOff) GetElapsedTime() time.Duration { + return b.Clock.Now().Sub(b.startTime) +} + +// Increments the current interval by multiplying it with the multiplier. +func (b *ExponentialBackOff) incrementCurrentInterval() { + // Check for overflow, if overflow is detected set the current interval to the max interval. + if float64(b.currentInterval) >= float64(b.MaxInterval)/b.Multiplier { + b.currentInterval = b.MaxInterval + } else { + b.currentInterval = time.Duration(float64(b.currentInterval) * b.Multiplier) + } +} + +// Returns a random value from the following interval: +// [randomizationFactor * currentInterval, randomizationFactor * currentInterval]. +func getRandomValueFromInterval(randomizationFactor, random float64, currentInterval time.Duration) time.Duration { + var delta = randomizationFactor * float64(currentInterval) + var minInterval = float64(currentInterval) - delta + var maxInterval = float64(currentInterval) + delta + + // Get a random value from the range [minInterval, maxInterval]. + // The formula used below has a +1 because if the minInterval is 1 and the maxInterval is 3 then + // we want a 33% chance for selecting either 1, 2 or 3. + return time.Duration(minInterval + (random * (maxInterval - minInterval + 1))) +} diff --git a/vendor/github.com/cenk/backoff/retry.go b/vendor/github.com/cenk/backoff/retry.go new file mode 100644 index 0000000000..5dbd825b5c --- /dev/null +++ b/vendor/github.com/cenk/backoff/retry.go @@ -0,0 +1,78 @@ +package backoff + +import "time" + +// An Operation is executing by Retry() or RetryNotify(). +// The operation will be retried using a backoff policy if it returns an error. +type Operation func() error + +// Notify is a notify-on-error function. It receives an operation error and +// backoff delay if the operation failed (with an error). +// +// NOTE that if the backoff policy stated to stop retrying, +// the notify function isn't called. +type Notify func(error, time.Duration) + +// Retry the operation o until it does not return error or BackOff stops. +// o is guaranteed to be run at least once. +// It is the caller's responsibility to reset b after Retry returns. +// +// If o returns a *PermanentError, the operation is not retried, and the +// wrapped error is returned. +// +// Retry sleeps the goroutine for the duration returned by BackOff after a +// failed operation returns. +func Retry(o Operation, b BackOff) error { return RetryNotify(o, b, nil) } + +// RetryNotify calls notify function with the error and wait duration +// for each failed attempt before sleep. +func RetryNotify(operation Operation, b BackOff, notify Notify) error { + var err error + var next time.Duration + + cb := ensureContext(b) + + b.Reset() + for { + if err = operation(); err == nil { + return nil + } + + if permanent, ok := err.(*PermanentError); ok { + return permanent.Err + } + + if next = b.NextBackOff(); next == Stop { + return err + } + + if notify != nil { + notify(err, next) + } + + t := time.NewTimer(next) + + select { + case <-cb.Context().Done(): + t.Stop() + return err + case <-t.C: + } + } +} + +// PermanentError signals that the operation should not be retried. +type PermanentError struct { + Err error +} + +func (e *PermanentError) Error() string { + return e.Err.Error() +} + +// Permanent wraps the given err in a *PermanentError. +func Permanent(err error) *PermanentError { + return &PermanentError{ + Err: err, + } +} diff --git a/vendor/github.com/cenk/backoff/ticker.go b/vendor/github.com/cenk/backoff/ticker.go new file mode 100644 index 0000000000..e742512fd3 --- /dev/null +++ b/vendor/github.com/cenk/backoff/ticker.go @@ -0,0 +1,84 @@ +package backoff + +import ( + "runtime" + "sync" + "time" +) + +// Ticker holds a channel that delivers `ticks' of a clock at times reported by a BackOff. +// +// Ticks will continue to arrive when the previous operation is still running, +// so operations that take a while to fail could run in quick succession. +type Ticker struct { + C <-chan time.Time + c chan time.Time + b BackOffContext + stop chan struct{} + stopOnce sync.Once +} + +// NewTicker returns a new Ticker containing a channel that will send +// the time at times specified by the BackOff argument. Ticker is +// guaranteed to tick at least once. The channel is closed when Stop +// method is called or BackOff stops. It is not safe to manipulate the +// provided backoff policy (notably calling NextBackOff or Reset) +// while the ticker is running. +func NewTicker(b BackOff) *Ticker { + c := make(chan time.Time) + t := &Ticker{ + C: c, + c: c, + b: ensureContext(b), + stop: make(chan struct{}), + } + t.b.Reset() + go t.run() + runtime.SetFinalizer(t, (*Ticker).Stop) + return t +} + +// Stop turns off a ticker. After Stop, no more ticks will be sent. +func (t *Ticker) Stop() { + t.stopOnce.Do(func() { close(t.stop) }) +} + +func (t *Ticker) run() { + c := t.c + defer close(c) + + // Ticker is guaranteed to tick at least once. + afterC := t.send(time.Now()) + + for { + if afterC == nil { + return + } + + select { + case tick := <-afterC: + afterC = t.send(tick) + case <-t.stop: + t.c = nil // Prevent future ticks from being sent to the channel. + return + case <-t.b.Context().Done(): + return + } + } +} + +func (t *Ticker) send(tick time.Time) <-chan time.Time { + select { + case t.c <- tick: + case <-t.stop: + return nil + } + + next := t.b.NextBackOff() + if next == Stop { + t.Stop() + return nil + } + + return time.After(next) +} diff --git a/vendor/github.com/cenk/backoff/tries.go b/vendor/github.com/cenk/backoff/tries.go new file mode 100644 index 0000000000..cfeefd9b76 --- /dev/null +++ b/vendor/github.com/cenk/backoff/tries.go @@ -0,0 +1,35 @@ +package backoff + +import "time" + +/* +WithMaxRetries creates a wrapper around another BackOff, which will +return Stop if NextBackOff() has been called too many times since +the last time Reset() was called + +Note: Implementation is not thread-safe. +*/ +func WithMaxRetries(b BackOff, max uint64) BackOff { + return &backOffTries{delegate: b, maxTries: max} +} + +type backOffTries struct { + delegate BackOff + maxTries uint64 + numTries uint64 +} + +func (b *backOffTries) NextBackOff() time.Duration { + if b.maxTries > 0 { + if b.maxTries <= b.numTries { + return Stop + } + b.numTries++ + } + return b.delegate.NextBackOff() +} + +func (b *backOffTries) Reset() { + b.numTries = 0 + b.delegate.Reset() +} diff --git a/vendor/github.com/facebookgo/clock/LICENSE b/vendor/github.com/facebookgo/clock/LICENSE new file mode 100644 index 0000000000..ce212cb1ce --- /dev/null +++ b/vendor/github.com/facebookgo/clock/LICENSE @@ -0,0 +1,21 @@ +The MIT License (MIT) + +Copyright (c) 2014 Ben Johnson + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/vendor/github.com/facebookgo/clock/README.md b/vendor/github.com/facebookgo/clock/README.md new file mode 100644 index 0000000000..5d4f4fe72e --- /dev/null +++ b/vendor/github.com/facebookgo/clock/README.md @@ -0,0 +1,104 @@ +clock [![Build Status](https://drone.io/github.com/benbjohnson/clock/status.png)](https://drone.io/github.com/benbjohnson/clock/latest) [![Coverage Status](https://coveralls.io/repos/benbjohnson/clock/badge.png?branch=master)](https://coveralls.io/r/benbjohnson/clock?branch=master) [![GoDoc](https://godoc.org/github.com/benbjohnson/clock?status.png)](https://godoc.org/github.com/benbjohnson/clock) ![Project status](http://img.shields.io/status/experimental.png?color=red) +===== + +Clock is a small library for mocking time in Go. It provides an interface +around the standard library's [`time`][time] package so that the application +can use the realtime clock while tests can use the mock clock. + +[time]: http://golang.org/pkg/time/ + + +## Usage + +### Realtime Clock + +Your application can maintain a `Clock` variable that will allow realtime and +mock clocks to be interchangable. For example, if you had an `Application` type: + +```go +import "github.com/benbjohnson/clock" + +type Application struct { + Clock clock.Clock +} +``` + +You could initialize it to use the realtime clock like this: + +```go +var app Application +app.Clock = clock.New() +... +``` + +Then all timers and time-related functionality should be performed from the +`Clock` variable. + + +### Mocking time + +In your tests, you will want to use a `Mock` clock: + +```go +import ( + "testing" + + "github.com/benbjohnson/clock" +) + +func TestApplication_DoSomething(t *testing.T) { + mock := clock.NewMock() + app := Application{Clock: mock} + ... +} +``` + +Now that you've initialized your application to use the mock clock, you can +adjust the time programmatically. The mock clock always starts from the Unix +epoch (midnight, Jan 1, 1970 UTC). + + +### Controlling time + +The mock clock provides the same functions that the standard library's `time` +package provides. For example, to find the current time, you use the `Now()` +function: + +```go +mock := clock.NewMock() + +// Find the current time. +mock.Now().UTC() // 1970-01-01 00:00:00 +0000 UTC + +// Move the clock forward. +mock.Add(2 * time.Hour) + +// Check the time again. It's 2 hours later! +mock.Now().UTC() // 1970-01-01 02:00:00 +0000 UTC +``` + +Timers and Tickers are also controlled by this same mock clock. They will only +execute when the clock is moved forward: + +``` +mock := clock.NewMock() +count := 0 + +// Kick off a timer to increment every 1 mock second. +go func() { + ticker := clock.Ticker(1 * time.Second) + for { + <-ticker.C + count++ + } +}() +runtime.Gosched() + +// Move the clock forward 10 second. +mock.Add(10 * time.Second) + +// This prints 10. +fmt.Println(count) +``` + + diff --git a/vendor/github.com/facebookgo/clock/clock.go b/vendor/github.com/facebookgo/clock/clock.go new file mode 100644 index 0000000000..bca1a7ba8b --- /dev/null +++ b/vendor/github.com/facebookgo/clock/clock.go @@ -0,0 +1,363 @@ +package clock + +import ( + "runtime" + "sort" + "sync" + "time" +) + +// Clock represents an interface to the functions in the standard library time +// package. Two implementations are available in the clock package. The first +// is a real-time clock which simply wraps the time package's functions. The +// second is a mock clock which will only make forward progress when +// programmatically adjusted. +type Clock interface { + After(d time.Duration) <-chan time.Time + AfterFunc(d time.Duration, f func()) *Timer + Now() time.Time + Sleep(d time.Duration) + Tick(d time.Duration) <-chan time.Time + Ticker(d time.Duration) *Ticker + Timer(d time.Duration) *Timer +} + +// New returns an instance of a real-time clock. +func New() Clock { + return &clock{} +} + +// clock implements a real-time clock by simply wrapping the time package functions. +type clock struct{} + +func (c *clock) After(d time.Duration) <-chan time.Time { return time.After(d) } + +func (c *clock) AfterFunc(d time.Duration, f func()) *Timer { + return &Timer{timer: time.AfterFunc(d, f)} +} + +func (c *clock) Now() time.Time { return time.Now() } + +func (c *clock) Sleep(d time.Duration) { time.Sleep(d) } + +func (c *clock) Tick(d time.Duration) <-chan time.Time { return time.Tick(d) } + +func (c *clock) Ticker(d time.Duration) *Ticker { + t := time.NewTicker(d) + return &Ticker{C: t.C, ticker: t} +} + +func (c *clock) Timer(d time.Duration) *Timer { + t := time.NewTimer(d) + return &Timer{C: t.C, timer: t} +} + +// Mock represents a mock clock that only moves forward programmically. +// It can be preferable to a real-time clock when testing time-based functionality. +type Mock struct { + mu sync.Mutex + now time.Time // current time + timers clockTimers // tickers & timers + + calls Calls + waiting []waiting + callsMutex sync.Mutex +} + +// NewMock returns an instance of a mock clock. +// The current time of the mock clock on initialization is the Unix epoch. +func NewMock() *Mock { + return &Mock{now: time.Unix(0, 0)} +} + +// Add moves the current time of the mock clock forward by the duration. +// This should only be called from a single goroutine at a time. +func (m *Mock) Add(d time.Duration) { + // Calculate the final current time. + t := m.now.Add(d) + + // Continue to execute timers until there are no more before the new time. + for { + if !m.runNextTimer(t) { + break + } + } + + // Ensure that we end with the new time. + m.mu.Lock() + m.now = t + m.mu.Unlock() + + // Give a small buffer to make sure the other goroutines get handled. + gosched() +} + +// runNextTimer executes the next timer in chronological order and moves the +// current time to the timer's next tick time. The next time is not executed if +// it's next time if after the max time. Returns true if a timer is executed. +func (m *Mock) runNextTimer(max time.Time) bool { + m.mu.Lock() + + // Sort timers by time. + sort.Sort(m.timers) + + // If we have no more timers then exit. + if len(m.timers) == 0 { + m.mu.Unlock() + return false + } + + // Retrieve next timer. Exit if next tick is after new time. + t := m.timers[0] + if t.Next().After(max) { + m.mu.Unlock() + return false + } + + // Move "now" forward and unlock clock. + m.now = t.Next() + m.mu.Unlock() + + // Execute timer. + t.Tick(m.now) + return true +} + +// After waits for the duration to elapse and then sends the current time on the returned channel. +func (m *Mock) After(d time.Duration) <-chan time.Time { + defer m.inc(&m.calls.After) + return m.Timer(d).C +} + +// AfterFunc waits for the duration to elapse and then executes a function. +// A Timer is returned that can be stopped. +func (m *Mock) AfterFunc(d time.Duration, f func()) *Timer { + defer m.inc(&m.calls.AfterFunc) + t := m.Timer(d) + t.C = nil + t.fn = f + return t +} + +// Now returns the current wall time on the mock clock. +func (m *Mock) Now() time.Time { + defer m.inc(&m.calls.Now) + m.mu.Lock() + defer m.mu.Unlock() + return m.now +} + +// Sleep pauses the goroutine for the given duration on the mock clock. +// The clock must be moved forward in a separate goroutine. +func (m *Mock) Sleep(d time.Duration) { + defer m.inc(&m.calls.Sleep) + <-m.After(d) +} + +// Tick is a convenience function for Ticker(). +// It will return a ticker channel that cannot be stopped. +func (m *Mock) Tick(d time.Duration) <-chan time.Time { + defer m.inc(&m.calls.Tick) + return m.Ticker(d).C +} + +// Ticker creates a new instance of Ticker. +func (m *Mock) Ticker(d time.Duration) *Ticker { + defer m.inc(&m.calls.Ticker) + m.mu.Lock() + defer m.mu.Unlock() + ch := make(chan time.Time) + t := &Ticker{ + C: ch, + c: ch, + mock: m, + d: d, + next: m.now.Add(d), + } + m.timers = append(m.timers, (*internalTicker)(t)) + return t +} + +// Timer creates a new instance of Timer. +func (m *Mock) Timer(d time.Duration) *Timer { + defer m.inc(&m.calls.Timer) + m.mu.Lock() + defer m.mu.Unlock() + ch := make(chan time.Time) + t := &Timer{ + C: ch, + c: ch, + mock: m, + next: m.now.Add(d), + } + m.timers = append(m.timers, (*internalTimer)(t)) + return t +} + +func (m *Mock) removeClockTimer(t clockTimer) { + m.mu.Lock() + defer m.mu.Unlock() + for i, timer := range m.timers { + if timer == t { + copy(m.timers[i:], m.timers[i+1:]) + m.timers[len(m.timers)-1] = nil + m.timers = m.timers[:len(m.timers)-1] + break + } + } + sort.Sort(m.timers) +} + +func (m *Mock) inc(addr *uint32) { + m.callsMutex.Lock() + defer m.callsMutex.Unlock() + *addr++ + var newWaiting []waiting + for _, w := range m.waiting { + if m.calls.atLeast(w.expected) { + close(w.done) + continue + } + newWaiting = append(newWaiting, w) + } + m.waiting = newWaiting +} + +// Wait waits for at least the relevant calls before returning. The expected +// Calls are always over the lifetime of the Mock. Values in the Calls struct +// are used as the minimum number of calls, this allows you to wait for only +// the calls you care about. +func (m *Mock) Wait(s Calls) { + m.callsMutex.Lock() + if m.calls.atLeast(s) { + m.callsMutex.Unlock() + return + } + done := make(chan struct{}) + m.waiting = append(m.waiting, waiting{expected: s, done: done}) + m.callsMutex.Unlock() + <-done +} + +// clockTimer represents an object with an associated start time. +type clockTimer interface { + Next() time.Time + Tick(time.Time) +} + +// clockTimers represents a list of sortable timers. +type clockTimers []clockTimer + +func (a clockTimers) Len() int { return len(a) } +func (a clockTimers) Swap(i, j int) { a[i], a[j] = a[j], a[i] } +func (a clockTimers) Less(i, j int) bool { return a[i].Next().Before(a[j].Next()) } + +// Timer represents a single event. +// The current time will be sent on C, unless the timer was created by AfterFunc. +type Timer struct { + C <-chan time.Time + c chan time.Time + timer *time.Timer // realtime impl, if set + next time.Time // next tick time + mock *Mock // mock clock, if set + fn func() // AfterFunc function, if set +} + +// Stop turns off the ticker. +func (t *Timer) Stop() { + if t.timer != nil { + t.timer.Stop() + } else { + t.mock.removeClockTimer((*internalTimer)(t)) + } +} + +type internalTimer Timer + +func (t *internalTimer) Next() time.Time { return t.next } +func (t *internalTimer) Tick(now time.Time) { + if t.fn != nil { + t.fn() + } else { + t.c <- now + } + t.mock.removeClockTimer((*internalTimer)(t)) + gosched() +} + +// Ticker holds a channel that receives "ticks" at regular intervals. +type Ticker struct { + C <-chan time.Time + c chan time.Time + ticker *time.Ticker // realtime impl, if set + next time.Time // next tick time + mock *Mock // mock clock, if set + d time.Duration // time between ticks +} + +// Stop turns off the ticker. +func (t *Ticker) Stop() { + if t.ticker != nil { + t.ticker.Stop() + } else { + t.mock.removeClockTimer((*internalTicker)(t)) + } +} + +type internalTicker Ticker + +func (t *internalTicker) Next() time.Time { return t.next } +func (t *internalTicker) Tick(now time.Time) { + select { + case t.c <- now: + case <-time.After(1 * time.Millisecond): + } + t.next = now.Add(t.d) + gosched() +} + +// Sleep momentarily so that other goroutines can process. +func gosched() { runtime.Gosched() } + +// Calls keeps track of the count of calls for each of the methods on the Clock +// interface. +type Calls struct { + After uint32 + AfterFunc uint32 + Now uint32 + Sleep uint32 + Tick uint32 + Ticker uint32 + Timer uint32 +} + +// atLeast returns true if at least the number of calls in o have been made. +func (c Calls) atLeast(o Calls) bool { + if c.After < o.After { + return false + } + if c.AfterFunc < o.AfterFunc { + return false + } + if c.Now < o.Now { + return false + } + if c.Sleep < o.Sleep { + return false + } + if c.Tick < o.Tick { + return false + } + if c.Ticker < o.Ticker { + return false + } + if c.Timer < o.Timer { + return false + } + return true +} + +type waiting struct { + expected Calls + done chan struct{} +} diff --git a/vendor/github.com/rubyist/circuitbreaker/CHANGELOG.md b/vendor/github.com/rubyist/circuitbreaker/CHANGELOG.md new file mode 100644 index 0000000000..6dbb18990a --- /dev/null +++ b/vendor/github.com/rubyist/circuitbreaker/CHANGELOG.md @@ -0,0 +1,230 @@ +# Changelog +All notable changes to this project will be documented in this file. + +## 2.2.0 - 2016-08-09 + +### Added +- Externally provided event listener channel (@spencerkimball) + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- Reduce allocations around last failure time storage +- Use the Clock for window code as well +- Remove test data race +- Fix race condition in `state()` (@tamird) + +## 2.1.7 - 2016-07-27 + +### Added +- Nothing + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- Set `Backoff.MaxElapsedTime` to 0 as default [@matope] +- Use a lock when modifying `nextBackoff` +- Fix goroutine leak when using timeouts [@isaldana] +- Fix window buckets that should be empty [@isaldana] +- Update backoff package, which has been renamed + +## 2.1.6 - 2016-02-02 + +### Added +- Nothing + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- client.Do() was not returning the error when it timed out [@ryanmurf] + +## 2.1.5 - 2015-11-19 + +### Added +- Nothing + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- Respect backoff.Stop [@bc-vincent-zhao] + +## 2.1.4 - 2015-09-01 + +### Added +- Nothing + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- HTTP client was using a new panel object instead of the one it added the breaker to [@ryanmurf] + +## 2.1.3 - 2015-08-05 + +### Added +- Configurable bucket time and number [@thraxil] +- Use mock clock for test [@andreas] + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- Bug in statsd bucket name documentation / example [@thraxil] + +## 2.1.2 - 2015-04-03 + +### Added +- Nothing + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- Simplify Call() for rate breaker, fixing a reset bug + +## 2.1.1 - 2014-10-29 + +### Added +- Nothing + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- Ensure the half opens counter resets when the breaker resets, or auto-resetting may not occur + +## 2.1.0 - 2014-10-16 + +### Added +- Failure, Sucess counts and Error Rate is now calculated over a sliding window +- Number of buckets in the window and the time the window spans are tuneable + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- A race condition in Call() + +## 2.0.2 - 2014-10-13 + +### Added +- ResetCounters + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- Nothing + +## 2.0.1 - 2014-10-13 + +### Added +- Nothing + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- Error rate should return 0.0 if there have been no samples + +## 2.0.0 - 2014-10-13 + +### Added +- All circuit breakers are now a Breaker with trip semantics handled by a TripFunc +- NewConsecutiveBreaker +- NewRateBreaker +- ConsecFailures +- ErrorRate +- Success +- Successes +- Retry logic now uses cenkalti/backoff, exponential backoff by default + +### Deprecated +- Nothing + +### Removed +- TrippableBreaker, ThresholdBreaker, FrequencyBreaker, TimeoutBreaker; all handled by Breaker now +- NewFrequencyBreaker, replaced by NewConsecutiveBreaker +- NewTimeoutBreaker, time out semantics are now handled by Call() +- NoOp(), use a Breaker with no TripFunc instead + +### Fixed +- Nothing + +## 1.1.2 - 2014-08-20 + +### Added +- Nothing + +### Deprecated +- Nothing + +### Fixed +- For a FrequencyBreaker, Failures() should return the count since the duration start, even after resetting. + +## 1.1.1 - 2014-08-20 + +### Added +- Nothing + +### Deprecated +- Nothing + +### Fixed +- Only send the reset event if the breaker was in a tripped state + +## 1.1.0 - 2014-08-16 + +### Added +- Re-export a Panels Circuits map. It's handy and if you mess it up, it's on you. + +### Deprecated +- Nothing + +### Removed +- Nothing + +### Fixed +- Nothing + +## 1.0.0 - 2014-08-16 + +### Added +- This will be the public API for version 1.0.0. This project will follow semver rules. diff --git a/vendor/github.com/rubyist/circuitbreaker/LICENSE b/vendor/github.com/rubyist/circuitbreaker/LICENSE new file mode 100644 index 0000000000..5a188a06b0 --- /dev/null +++ b/vendor/github.com/rubyist/circuitbreaker/LICENSE @@ -0,0 +1,22 @@ +(The MIT License) + +Copyright (c) 2014 Scott Barron + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +'Software'), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be +included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, +TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/vendor/github.com/rubyist/circuitbreaker/README.md b/vendor/github.com/rubyist/circuitbreaker/README.md new file mode 100644 index 0000000000..1e08423b58 --- /dev/null +++ b/vendor/github.com/rubyist/circuitbreaker/README.md @@ -0,0 +1,120 @@ +# circuitbreaker + +Circuitbreaker provides an easy way to use the Circuit Breaker pattern in a +Go program. + +Circuit breakers are typically used when your program makes remote calls. +Remote calls can often hang for a while before they time out. If your +application makes a lot of these requests, many resources can be tied +up waiting for these time outs to occur. A circuit breaker wraps these +remote calls and will trip after a defined amount of failures or time outs +occur. When a circuit breaker is tripped any future calls will avoid making +the remote call and return an error to the caller. In the meantime, the +circuit breaker will periodically allow some calls to be tried again and +will close the circuit if those are successful. + +You can read more about this pattern and how it's used at: +- [Martin Fowler's bliki](http://martinfowler.com/bliki/CircuitBreaker.html) +- [The Netflix Tech Blog](http://techblog.netflix.com/2012/02/fault-tolerance-in-high-volume.html) +- [Release It!](http://pragprog.com/book/mnee/release-it) + +[![GoDoc](https://godoc.org/github.com/rubyist/circuitbreaker?status.svg)](https://godoc.org/github.com/rubyist/circuitbreaker) + +## Installation + +``` + go get github.com/rubyist/circuitbreaker +``` + +## Examples + +Here is a quick example of what circuitbreaker provides + +```go +// Creates a circuit breaker that will trip if the function fails 10 times +cb := circuit.NewThresholdBreaker(10) + +events := cb.Subscribe() +go func() { + for { + e := <-events + // Monitor breaker events like BreakerTripped, BreakerReset, BreakerFail, BreakerReady + } +}() + +cb.Call(func() error { + // This is where you'll do some remote call + // If it fails, return an error +}, 0) +``` + +Circuitbreaker can also wrap a time out around the remote call. + +```go +// Creates a circuit breaker that will trip after 10 failures +// using a time out of 5 seconds +cb := circuit.NewThresholdBreaker(10) + +cb.Call(func() error { + // This is where you'll do some remote call + // If it fails, return an error +}, time.Second * 5) // This will time out after 5 seconds, which counts as a failure + +// Proceed as above + +``` + +Circuitbreaker can also trip based on the number of consecutive failures. + +```go +// Creates a circuit breaker that will trip if 10 consecutive failures occur +cb := circuit.NewConsecutiveBreaker(10) + +// Proceed as above +``` + +Circuitbreaker can trip based on the error rate. + +```go +// Creates a circuit breaker based on the error rate +cb := circuit.NewRateBreaker(0.95, 100) // trip when error rate hits 95%, with at least 100 samples + +// Proceed as above +``` + +If it doesn't make sense to wrap logic in Call(), breakers can be handled manually. + +```go +cb := circuit.NewThresholdBreaker(10) + +for { + if cb.Ready() { + // Breaker is not tripped, proceed + err := doSomething() + if err != nil { + cb.Fail() // This will trip the breaker once it's failed 10 times + continue + } + cb.Success() + } else { + // Breaker is in a tripped state. + } +} +``` + +Circuitbreaker also provides a wrapper around `http.Client` that will wrap a +time out around any request. + +```go +// Passing in nil will create a regular http.Client. +// You can also build your own http.Client and pass it in +client := circuit.NewHTTPClient(time.Second * 5, 10, nil) + +resp, err := client.Get("http://example.com/resource.json") +``` + +See the godoc for more examples. + +## Bugs, Issues, Feedback + +Right here on GitHub: [https://github.com/rubyist/circuitbreaker](https://github.com/rubyist/circuitbreaker) diff --git a/vendor/github.com/rubyist/circuitbreaker/circuitbreaker.go b/vendor/github.com/rubyist/circuitbreaker/circuitbreaker.go new file mode 100644 index 0000000000..620cd94a64 --- /dev/null +++ b/vendor/github.com/rubyist/circuitbreaker/circuitbreaker.go @@ -0,0 +1,446 @@ +// Package circuit implements the Circuit Breaker pattern. It will wrap +// a function call (typically one which uses remote services) and monitors for +// failures and/or time outs. When a threshold of failures or time outs has been +// reached, future calls to the function will not run. During this state, the +// breaker will periodically allow the function to run and, if it is successful, +// will start running the function again. +// +// Circuit includes three types of circuit breakers: +// +// A Threshold Breaker will trip when the failure count reaches a given threshold. +// It does not matter how long it takes to reach the threshold and the failures do +// not need to be consecutive. +// +// A Consecutive Breaker will trip when the consecutive failure count reaches a given +// threshold. It does not matter how long it takes to reach the threshold, but the +// failures do need to be consecutive. +// +// +// When wrapping blocks of code with a Breaker's Call() function, a time out can be +// specified. If the time out is reached, the breaker's Fail() function will be called. +// +// +// Other types of circuit breakers can be easily built by creating a Breaker and +// adding a custom TripFunc. A TripFunc is called when a Breaker Fail()s and receives +// the breaker as an argument. It then returns true or false to indicate whether the +// breaker should trip. +// +// The package also provides a wrapper around an http.Client that wraps all of +// the http.Client functions with a Breaker. +// +package circuit + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "time" + + "github.com/cenk/backoff" + "github.com/facebookgo/clock" +) + +// BreakerEvent indicates the type of event received over an event channel +type BreakerEvent int + +const ( + // BreakerTripped is sent when a breaker trips + BreakerTripped BreakerEvent = iota + + // BreakerReset is sent when a breaker resets + BreakerReset BreakerEvent = iota + + // BreakerFail is sent when Fail() is called + BreakerFail BreakerEvent = iota + + // BreakerReady is sent when the breaker enters the half open state and is ready to retry + BreakerReady BreakerEvent = iota +) + +// ListenerEvent includes a reference to the circuit breaker and the event. +type ListenerEvent struct { + CB *Breaker + Event BreakerEvent +} + +type state int + +const ( + open state = iota + halfopen state = iota + closed state = iota +) + +var ( + defaultInitialBackOffInterval = 500 * time.Millisecond + defaultBackoffMaxElapsedTime = 0 * time.Second +) + +// Error codes returned by Call +var ( + ErrBreakerOpen = errors.New("breaker open") + ErrBreakerTimeout = errors.New("breaker time out") +) + +// TripFunc is a function called by a Breaker's Fail() function and determines whether +// the breaker should trip. It will receive the Breaker as an argument and returns a +// boolean. By default, a Breaker has no TripFunc. +type TripFunc func(*Breaker) bool + +// Breaker is the base of a circuit breaker. It maintains failure and success counters +// as well as the event subscribers. +type Breaker struct { + // BackOff is the backoff policy that is used when determining if the breaker should + // attempt to retry. A breaker created with NewBreaker will use an exponential backoff + // policy by default. + BackOff backoff.BackOff + + // ShouldTrip is a TripFunc that determines whether a Fail() call should trip the breaker. + // A breaker created with NewBreaker will not have a ShouldTrip by default, and thus will + // never automatically trip. + ShouldTrip TripFunc + + // Clock is used for controlling time in tests. + Clock clock.Clock + + _ [4]byte // pad to fix golang issue #599 + consecFailures int64 + lastFailure int64 // stored as nanoseconds since the Unix epoch + halfOpens int64 + counts *window + nextBackOff time.Duration + tripped int32 + broken int32 + eventReceivers []chan BreakerEvent + listeners []chan ListenerEvent + backoffLock sync.Mutex +} + +// Options holds breaker configuration options. +type Options struct { + BackOff backoff.BackOff + Clock clock.Clock + ShouldTrip TripFunc + WindowTime time.Duration + WindowBuckets int +} + +// NewBreakerWithOptions creates a base breaker with a specified backoff, clock and TripFunc +func NewBreakerWithOptions(options *Options) *Breaker { + if options == nil { + options = &Options{} + } + + if options.Clock == nil { + options.Clock = clock.New() + } + + if options.BackOff == nil { + b := backoff.NewExponentialBackOff() + b.InitialInterval = defaultInitialBackOffInterval + b.MaxElapsedTime = defaultBackoffMaxElapsedTime + b.Clock = options.Clock + b.Reset() + options.BackOff = b + } + + if options.WindowTime == 0 { + options.WindowTime = DefaultWindowTime + } + + if options.WindowBuckets == 0 { + options.WindowBuckets = DefaultWindowBuckets + } + + return &Breaker{ + BackOff: options.BackOff, + Clock: options.Clock, + ShouldTrip: options.ShouldTrip, + nextBackOff: options.BackOff.NextBackOff(), + counts: newWindow(options.WindowTime, options.WindowBuckets), + } +} + +// NewBreaker creates a base breaker with an exponential backoff and no TripFunc +func NewBreaker() *Breaker { + return NewBreakerWithOptions(nil) +} + +// NewThresholdBreaker creates a Breaker with a ThresholdTripFunc. +func NewThresholdBreaker(threshold int64) *Breaker { + return NewBreakerWithOptions(&Options{ + ShouldTrip: ThresholdTripFunc(threshold), + }) +} + +// NewConsecutiveBreaker creates a Breaker with a ConsecutiveTripFunc. +func NewConsecutiveBreaker(threshold int64) *Breaker { + return NewBreakerWithOptions(&Options{ + ShouldTrip: ConsecutiveTripFunc(threshold), + }) +} + +// NewRateBreaker creates a Breaker with a RateTripFunc. +func NewRateBreaker(rate float64, minSamples int64) *Breaker { + return NewBreakerWithOptions(&Options{ + ShouldTrip: RateTripFunc(rate, minSamples), + }) +} + +// Subscribe returns a channel of BreakerEvents. Whenever the breaker changes state, +// the state will be sent over the channel. See BreakerEvent for the types of events. +func (cb *Breaker) Subscribe() <-chan BreakerEvent { + eventReader := make(chan BreakerEvent) + output := make(chan BreakerEvent, 100) + + go func() { + for v := range eventReader { + select { + case output <- v: + default: + <-output + output <- v + } + } + }() + cb.eventReceivers = append(cb.eventReceivers, eventReader) + return output +} + +// AddListener adds a channel of ListenerEvents on behalf of a listener. +// The listener channel must be buffered. +func (cb *Breaker) AddListener(listener chan ListenerEvent) { + cb.listeners = append(cb.listeners, listener) +} + +// RemoveListener removes a channel previously added via AddListener. +// Once removed, the channel will no longer receive ListenerEvents. +// Returns true if the listener was found and removed. +func (cb *Breaker) RemoveListener(listener chan ListenerEvent) bool { + for i, receiver := range cb.listeners { + if listener == receiver { + cb.listeners = append(cb.listeners[:i], cb.listeners[i+1:]...) + return true + } + } + return false +} + +// Trip will trip the circuit breaker. After Trip() is called, Tripped() will +// return true. +func (cb *Breaker) Trip() { + atomic.StoreInt32(&cb.tripped, 1) + now := cb.Clock.Now() + atomic.StoreInt64(&cb.lastFailure, now.UnixNano()) + cb.sendEvent(BreakerTripped) +} + +// Reset will reset the circuit breaker. After Reset() is called, Tripped() will +// return false. +func (cb *Breaker) Reset() { + atomic.StoreInt32(&cb.broken, 0) + atomic.StoreInt32(&cb.tripped, 0) + atomic.StoreInt64(&cb.halfOpens, 0) + cb.ResetCounters() + cb.sendEvent(BreakerReset) +} + +// ResetCounters will reset only the failures, consecFailures, and success counters +func (cb *Breaker) ResetCounters() { + atomic.StoreInt64(&cb.consecFailures, 0) + cb.counts.Reset() +} + +// Tripped returns true if the circuit breaker is tripped, false if it is reset. +func (cb *Breaker) Tripped() bool { + return atomic.LoadInt32(&cb.tripped) == 1 +} + +// Break trips the circuit breaker and prevents it from auto resetting. Use this when +// manual control over the circuit breaker state is needed. +func (cb *Breaker) Break() { + atomic.StoreInt32(&cb.broken, 1) + cb.Trip() +} + +// Failures returns the number of failures for this circuit breaker. +func (cb *Breaker) Failures() int64 { + return cb.counts.Failures() +} + +// ConsecFailures returns the number of consecutive failures that have occured. +func (cb *Breaker) ConsecFailures() int64 { + return atomic.LoadInt64(&cb.consecFailures) +} + +// Successes returns the number of successes for this circuit breaker. +func (cb *Breaker) Successes() int64 { + return cb.counts.Successes() +} + +// Fail is used to indicate a failure condition the Breaker should record. It will +// increment the failure counters and store the time of the last failure. If the +// breaker has a TripFunc it will be called, tripping the breaker if necessary. +func (cb *Breaker) Fail() { + cb.counts.Fail() + atomic.AddInt64(&cb.consecFailures, 1) + now := cb.Clock.Now() + atomic.StoreInt64(&cb.lastFailure, now.UnixNano()) + cb.sendEvent(BreakerFail) + if cb.ShouldTrip != nil && cb.ShouldTrip(cb) { + cb.Trip() + } +} + +// Success is used to indicate a success condition the Breaker should record. If +// the success was triggered by a retry attempt, the breaker will be Reset(). +func (cb *Breaker) Success() { + cb.backoffLock.Lock() + cb.BackOff.Reset() + cb.nextBackOff = cb.BackOff.NextBackOff() + cb.backoffLock.Unlock() + + state := cb.state() + if state == halfopen { + cb.Reset() + } + atomic.StoreInt64(&cb.consecFailures, 0) + cb.counts.Success() +} + +// ErrorRate returns the current error rate of the Breaker, expressed as a floating +// point number (e.g. 0.9 for 90%), since the last time the breaker was Reset. +func (cb *Breaker) ErrorRate() float64 { + return cb.counts.ErrorRate() +} + +// Ready will return true if the circuit breaker is ready to call the function. +// It will be ready if the breaker is in a reset state, or if it is time to retry +// the call for auto resetting. +func (cb *Breaker) Ready() bool { + state := cb.state() + if state == halfopen { + atomic.StoreInt64(&cb.halfOpens, 0) + cb.sendEvent(BreakerReady) + } + return state == closed || state == halfopen +} + +// Call wraps a function the Breaker will protect. A failure is recorded +// whenever the function returns an error. If the called function takes longer +// than timeout to run, a failure will be recorded. +func (cb *Breaker) Call(circuit func() error, timeout time.Duration) error { + return cb.CallContext(context.Background(), circuit, timeout) +} + +// CallContext is same as Call but if the ctx is canceled after the circuit returned an error, +// the error will not be marked as a failure because the call was canceled intentionally. +func (cb *Breaker) CallContext(ctx context.Context, circuit func() error, timeout time.Duration) error { + var err error + + if !cb.Ready() { + return ErrBreakerOpen + } + + if timeout == 0 { + err = circuit() + } else { + c := make(chan error, 1) + go func() { + c <- circuit() + close(c) + }() + + select { + case e := <-c: + err = e + case <-cb.Clock.After(timeout): + err = ErrBreakerTimeout + } + } + + if err != nil { + if ctx.Err() != context.Canceled { + cb.Fail() + } + return err + } + + cb.Success() + return nil +} + +// state returns the state of the TrippableBreaker. The states available are: +// closed - the circuit is in a reset state and is operational +// open - the circuit is in a tripped state +// halfopen - the circuit is in a tripped state but the reset timeout has passed +func (cb *Breaker) state() state { + tripped := cb.Tripped() + if tripped { + if atomic.LoadInt32(&cb.broken) == 1 { + return open + } + + last := atomic.LoadInt64(&cb.lastFailure) + since := cb.Clock.Now().Sub(time.Unix(0, last)) + + cb.backoffLock.Lock() + defer cb.backoffLock.Unlock() + + if cb.nextBackOff != backoff.Stop && since > cb.nextBackOff { + if atomic.CompareAndSwapInt64(&cb.halfOpens, 0, 1) { + cb.nextBackOff = cb.BackOff.NextBackOff() + return halfopen + } + return open + } + return open + } + return closed +} + +func (cb *Breaker) sendEvent(event BreakerEvent) { + for _, receiver := range cb.eventReceivers { + receiver <- event + } + for _, listener := range cb.listeners { + le := ListenerEvent{CB: cb, Event: event} + select { + case listener <- le: + default: + <-listener + listener <- le + } + } +} + +// ThresholdTripFunc returns a TripFunc with that trips whenever +// the failure count meets the threshold. +func ThresholdTripFunc(threshold int64) TripFunc { + return func(cb *Breaker) bool { + return cb.Failures() == threshold + } +} + +// ConsecutiveTripFunc returns a TripFunc that trips whenever +// the consecutive failure count meets the threshold. +func ConsecutiveTripFunc(threshold int64) TripFunc { + return func(cb *Breaker) bool { + return cb.ConsecFailures() == threshold + } +} + +// RateTripFunc returns a TripFunc that trips whenever the +// error rate hits the threshold. The error rate is calculated as such: +// f = number of failures +// s = number of successes +// e = f / (f + s) +// The error rate is calculated over a sliding window of 10 seconds (by default) +// This TripFunc will not trip until there have been at least minSamples events. +func RateTripFunc(rate float64, minSamples int64) TripFunc { + return func(cb *Breaker) bool { + samples := cb.Failures() + cb.Successes() + return samples >= minSamples && cb.ErrorRate() >= rate + } +} diff --git a/vendor/github.com/rubyist/circuitbreaker/client.go b/vendor/github.com/rubyist/circuitbreaker/client.go new file mode 100644 index 0000000000..e91c52b26c --- /dev/null +++ b/vendor/github.com/rubyist/circuitbreaker/client.go @@ -0,0 +1,170 @@ +package circuit + +import ( + "io" + "net/http" + "net/url" + "time" +) + +// HTTPClient is a wrapper around http.Client that provides circuit breaker capabilities. +// +// By default, the client will use its defaultBreaker. A BreakerLookup function may be +// provided to allow different breakers to be used based on the circumstance. See the +// implementation of NewHostBasedHTTPClient for an example of this. +type HTTPClient struct { + Client *http.Client + BreakerTripped func() + BreakerReset func() + BreakerLookup func(*HTTPClient, interface{}) *Breaker + Panel *Panel + timeout time.Duration +} + +var defaultBreakerName = "_default" + +// NewHTTPClient provides a circuit breaker wrapper around http.Client. +// It wraps all of the regular http.Client functions. Specifying 0 for timeout will +// give a breaker that does not check for time outs. +func NewHTTPClient(timeout time.Duration, threshold int64, client *http.Client) *HTTPClient { + breaker := NewThresholdBreaker(threshold) + return NewHTTPClientWithBreaker(breaker, timeout, client) +} + +// NewHostBasedHTTPClient provides a circuit breaker wrapper around http.Client. This +// client will use one circuit breaker per host parsed from the request URL. This allows +// you to use a single HTTPClient for multiple hosts with one host's breaker not affecting +// the other hosts. +func NewHostBasedHTTPClient(timeout time.Duration, threshold int64, client *http.Client) *HTTPClient { + brclient := NewHTTPClient(timeout, threshold, client) + + brclient.BreakerLookup = func(c *HTTPClient, val interface{}) *Breaker { + rawURL := val.(string) + parsedURL, err := url.Parse(rawURL) + if err != nil { + breaker, _ := c.Panel.Get(defaultBreakerName) + return breaker + } + host := parsedURL.Host + + cb, ok := c.Panel.Get(host) + if !ok { + cb = NewThresholdBreaker(threshold) + c.Panel.Add(host, cb) + } + + return cb + } + + return brclient +} + +// NewHTTPClientWithBreaker provides a circuit breaker wrapper around http.Client. +// It wraps all of the regular http.Client functions using the provided Breaker. +func NewHTTPClientWithBreaker(breaker *Breaker, timeout time.Duration, client *http.Client) *HTTPClient { + if client == nil { + client = &http.Client{} + } + + panel := NewPanel() + panel.Add(defaultBreakerName, breaker) + + brclient := &HTTPClient{Client: client, Panel: panel, timeout: timeout} + brclient.BreakerLookup = func(c *HTTPClient, val interface{}) *Breaker { + cb, _ := c.Panel.Get(defaultBreakerName) + return cb + } + + events := breaker.Subscribe() + go func() { + event := <-events + switch event { + case BreakerTripped: + brclient.runBreakerTripped() + case BreakerReset: + brclient.runBreakerReset() + } + }() + + return brclient +} + +// Do wraps http.Client Do() +func (c *HTTPClient) Do(req *http.Request) (*http.Response, error) { + var resp *http.Response + var err error + breaker := c.breakerLookup(req.URL.String()) + err = breaker.Call(func() error { + resp, err = c.Client.Do(req) + return err + }, c.timeout) + return resp, err +} + +// Get wraps http.Client Get() +func (c *HTTPClient) Get(url string) (*http.Response, error) { + var resp *http.Response + breaker := c.breakerLookup(url) + err := breaker.Call(func() error { + aresp, err := c.Client.Get(url) + resp = aresp + return err + }, c.timeout) + return resp, err +} + +// Head wraps http.Client Head() +func (c *HTTPClient) Head(url string) (*http.Response, error) { + var resp *http.Response + breaker := c.breakerLookup(url) + err := breaker.Call(func() error { + aresp, err := c.Client.Head(url) + resp = aresp + return err + }, c.timeout) + return resp, err +} + +// Post wraps http.Client Post() +func (c *HTTPClient) Post(url string, bodyType string, body io.Reader) (*http.Response, error) { + var resp *http.Response + breaker := c.breakerLookup(url) + err := breaker.Call(func() error { + aresp, err := c.Client.Post(url, bodyType, body) + resp = aresp + return err + }, c.timeout) + return resp, err +} + +// PostForm wraps http.Client PostForm() +func (c *HTTPClient) PostForm(url string, data url.Values) (*http.Response, error) { + var resp *http.Response + breaker := c.breakerLookup(url) + err := breaker.Call(func() error { + aresp, err := c.Client.PostForm(url, data) + resp = aresp + return err + }, c.timeout) + return resp, err +} + +func (c *HTTPClient) breakerLookup(val interface{}) *Breaker { + if c.BreakerLookup != nil { + return c.BreakerLookup(c, val) + } + cb, _ := c.Panel.Get(defaultBreakerName) + return cb +} + +func (c *HTTPClient) runBreakerTripped() { + if c.BreakerTripped != nil { + c.BreakerTripped() + } +} + +func (c *HTTPClient) runBreakerReset() { + if c.BreakerReset != nil { + c.BreakerReset() + } +} diff --git a/vendor/github.com/rubyist/circuitbreaker/panel.go b/vendor/github.com/rubyist/circuitbreaker/panel.go new file mode 100644 index 0000000000..cd4e7b6c7e --- /dev/null +++ b/vendor/github.com/rubyist/circuitbreaker/panel.go @@ -0,0 +1,144 @@ +package circuit + +import ( + "fmt" + "sync" + "time" +) + +var defaultStatsPrefixf = "circuit.%s" + +// Statter interface provides a way to gather statistics from breakers +type Statter interface { + Counter(sampleRate float32, bucket string, n ...int) + Timing(sampleRate float32, bucket string, d ...time.Duration) + Gauge(sampleRate float32, bucket string, value ...string) +} + +// PanelEvent wraps a BreakerEvent and provides the string name of the breaker +type PanelEvent struct { + Name string + Event BreakerEvent +} + +// Panel tracks a group of circuit breakers by name. +type Panel struct { + Statter Statter + StatsPrefixf string + + Circuits map[string]*Breaker + + lastTripTimes map[string]time.Time + tripTimesLock sync.RWMutex + panelLock sync.RWMutex + eventReceivers []chan PanelEvent +} + +// NewPanel creates a new Panel +func NewPanel() *Panel { + return &Panel{ + Circuits: make(map[string]*Breaker), + Statter: &noopStatter{}, + StatsPrefixf: defaultStatsPrefixf, + lastTripTimes: make(map[string]time.Time)} +} + +// Add sets the name as a reference to the given circuit breaker. +func (p *Panel) Add(name string, cb *Breaker) { + p.panelLock.Lock() + p.Circuits[name] = cb + p.panelLock.Unlock() + + events := cb.Subscribe() + + go func() { + for event := range events { + for _, receiver := range p.eventReceivers { + receiver <- PanelEvent{name, event} + } + switch event { + case BreakerTripped: + p.breakerTripped(name) + case BreakerReset: + p.breakerReset(name) + case BreakerFail: + p.breakerFail(name) + case BreakerReady: + p.breakerReady(name) + } + } + }() +} + +// Get retrieves a circuit breaker by name. If no circuit breaker exists, it +// returns the NoOp one and sets ok to false. +func (p *Panel) Get(name string) (*Breaker, bool) { + p.panelLock.RLock() + cb, ok := p.Circuits[name] + p.panelLock.RUnlock() + + if ok { + return cb, ok + } + + return NewBreaker(), ok +} + +// Subscribe returns a channel of PanelEvents. Whenever a breaker changes state, +// the PanelEvent will be sent over the channel. See BreakerEvent for the types of events. +func (p *Panel) Subscribe() <-chan PanelEvent { + eventReader := make(chan PanelEvent) + output := make(chan PanelEvent, 100) + + go func() { + for v := range eventReader { + select { + case output <- v: + default: + <-output + output <- v + } + } + }() + p.eventReceivers = append(p.eventReceivers, eventReader) + return output +} + +func (p *Panel) breakerTripped(name string) { + p.Statter.Counter(1.0, fmt.Sprintf(p.StatsPrefixf, name)+".tripped", 1) + p.tripTimesLock.Lock() + p.lastTripTimes[name] = time.Now() + p.tripTimesLock.Unlock() +} + +func (p *Panel) breakerReset(name string) { + bucket := fmt.Sprintf(p.StatsPrefixf, name) + + p.Statter.Counter(1.0, bucket+".reset", 1) + + p.tripTimesLock.RLock() + lastTrip := p.lastTripTimes[name] + p.tripTimesLock.RUnlock() + + if !lastTrip.IsZero() { + p.Statter.Timing(1.0, bucket+".trip-time", time.Since(lastTrip)) + p.tripTimesLock.Lock() + p.lastTripTimes[name] = time.Time{} + p.tripTimesLock.Unlock() + } +} + +func (p *Panel) breakerFail(name string) { + p.Statter.Counter(1.0, fmt.Sprintf(p.StatsPrefixf, name)+".fail", 1) +} + +func (p *Panel) breakerReady(name string) { + p.Statter.Counter(1.0, fmt.Sprintf(p.StatsPrefixf, name)+".ready", 1) +} + +type noopStatter struct { +} + +func (*noopStatter) Counter(sampleRate float32, bucket string, n ...int) {} +func (*noopStatter) Timing(sampleRate float32, bucket string, d ...time.Duration) {} +func (*noopStatter) Gauge(sampleRate float32, bucket string, value ...string) {} diff --git a/vendor/github.com/rubyist/circuitbreaker/window.go b/vendor/github.com/rubyist/circuitbreaker/window.go new file mode 100644 index 0000000000..ab83187f6c --- /dev/null +++ b/vendor/github.com/rubyist/circuitbreaker/window.go @@ -0,0 +1,174 @@ +package circuit + +import ( + "container/ring" + "sync" + "time" + + "github.com/facebookgo/clock" +) + +var ( + // DefaultWindowTime is the default time the window covers, 10 seconds. + DefaultWindowTime = time.Millisecond * 10000 + + // DefaultWindowBuckets is the default number of buckets the window holds, 10. + DefaultWindowBuckets = 10 +) + +// bucket holds counts of failures and successes +type bucket struct { + failure int64 + success int64 +} + +// Reset resets the counts to 0 +func (b *bucket) Reset() { + b.failure = 0 + b.success = 0 +} + +// Fail increments the failure count +func (b *bucket) Fail() { + b.failure++ +} + +// Sucecss increments the success count +func (b *bucket) Success() { + b.success++ +} + +// window maintains a ring of buckets and increments the failure and success +// counts of the current bucket. Once a specified time has elapsed, it will +// advance to the next bucket, reseting its counts. This allows the keeping of +// rolling statistics on the counts. +type window struct { + buckets *ring.Ring + bucketTime time.Duration + bucketLock sync.RWMutex + lastAccess time.Time + clock clock.Clock +} + +// newWindow creates a new window. windowTime is the time covering the entire +// window. windowBuckets is the number of buckets the window is divided into. +// An example: a 10 second window with 10 buckets will have 10 buckets covering +// 1 second each. +func newWindow(windowTime time.Duration, windowBuckets int) *window { + buckets := ring.New(windowBuckets) + for i := 0; i < buckets.Len(); i++ { + buckets.Value = &bucket{} + buckets = buckets.Next() + } + + clock := clock.New() + + bucketTime := time.Duration(windowTime.Nanoseconds() / int64(windowBuckets)) + return &window{ + buckets: buckets, + bucketTime: bucketTime, + clock: clock, + lastAccess: clock.Now(), + } +} + +// Fail records a failure in the current bucket. +func (w *window) Fail() { + w.bucketLock.Lock() + b := w.getLatestBucket() + b.Fail() + w.bucketLock.Unlock() +} + +// Success records a success in the current bucket. +func (w *window) Success() { + w.bucketLock.Lock() + b := w.getLatestBucket() + b.Success() + w.bucketLock.Unlock() +} + +// Failures returns the total number of failures recorded in all buckets. +func (w *window) Failures() int64 { + w.bucketLock.RLock() + + var failures int64 + w.buckets.Do(func(x interface{}) { + b := x.(*bucket) + failures += b.failure + }) + + w.bucketLock.RUnlock() + return failures +} + +// Successes returns the total number of successes recorded in all buckets. +func (w *window) Successes() int64 { + w.bucketLock.RLock() + + var successes int64 + w.buckets.Do(func(x interface{}) { + b := x.(*bucket) + successes += b.success + }) + w.bucketLock.RUnlock() + return successes +} + +// ErrorRate returns the error rate calculated over all buckets, expressed as +// a floating point number (e.g. 0.9 for 90%) +func (w *window) ErrorRate() float64 { + var total int64 + var failures int64 + + w.bucketLock.RLock() + w.buckets.Do(func(x interface{}) { + b := x.(*bucket) + total += b.failure + b.success + failures += b.failure + }) + w.bucketLock.RUnlock() + + if total == 0 { + return 0.0 + } + + return float64(failures) / float64(total) +} + +// Reset resets the count of all buckets. +func (w *window) Reset() { + w.bucketLock.Lock() + + w.buckets.Do(func(x interface{}) { + x.(*bucket).Reset() + }) + w.bucketLock.Unlock() +} + +// getLatestBucket returns the current bucket. If the bucket time has elapsed +// it will move to the next bucket, resetting its counts and updating the last +// access time before returning it. getLatestBucket assumes that the caller has +// locked the bucketLock +func (w *window) getLatestBucket() *bucket { + var b *bucket + b = w.buckets.Value.(*bucket) + elapsed := w.clock.Now().Sub(w.lastAccess) + + if elapsed > w.bucketTime { + // Reset the buckets between now and number of buckets ago. If + // that is more that the existing buckets, reset all. + for i := 0; i < w.buckets.Len(); i++ { + w.buckets = w.buckets.Next() + b = w.buckets.Value.(*bucket) + b.Reset() + elapsed = time.Duration(int64(elapsed) - int64(w.bucketTime)) + if elapsed < w.bucketTime { + // Done resetting buckets. + break + } + } + w.lastAccess = w.clock.Now() + } + return b +} diff --git a/vendor/golang.org/x/net/context/context.go b/vendor/golang.org/x/net/context/context.go new file mode 100644 index 0000000000..a3c021d3f8 --- /dev/null +++ b/vendor/golang.org/x/net/context/context.go @@ -0,0 +1,56 @@ +// Copyright 2014 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// Package context defines the Context type, which carries deadlines, +// cancelation signals, and other request-scoped values across API boundaries +// and between processes. +// As of Go 1.7 this package is available in the standard library under the +// name context. https://golang.org/pkg/context. +// +// Incoming requests to a server should create a Context, and outgoing calls to +// servers should accept a Context. The chain of function calls between must +// propagate the Context, optionally replacing it with a modified copy created +// using WithDeadline, WithTimeout, WithCancel, or WithValue. +// +// Programs that use Contexts should follow these rules to keep interfaces +// consistent across packages and enable static analysis tools to check context +// propagation: +// +// Do not store Contexts inside a struct type; instead, pass a Context +// explicitly to each function that needs it. The Context should be the first +// parameter, typically named ctx: +// +// func DoSomething(ctx context.Context, arg Arg) error { +// // ... use ctx ... +// } +// +// Do not pass a nil Context, even if a function permits it. Pass context.TODO +// if you are unsure about which Context to use. +// +// Use context Values only for request-scoped data that transits processes and +// APIs, not for passing optional parameters to functions. +// +// The same Context may be passed to functions running in different goroutines; +// Contexts are safe for simultaneous use by multiple goroutines. +// +// See http://blog.golang.org/context for example code for a server that uses +// Contexts. +package context // import "golang.org/x/net/context" + +// Background returns a non-nil, empty Context. It is never canceled, has no +// values, and has no deadline. It is typically used by the main function, +// initialization, and tests, and as the top-level Context for incoming +// requests. +func Background() Context { + return background +} + +// TODO returns a non-nil, empty Context. Code should use context.TODO when +// it's unclear which Context to use or it is not yet available (because the +// surrounding function has not yet been extended to accept a Context +// parameter). TODO is recognized by static analysis tools that determine +// whether Contexts are propagated correctly in a program. +func TODO() Context { + return todo +} diff --git a/vendor/golang.org/x/net/context/go17.go b/vendor/golang.org/x/net/context/go17.go new file mode 100644 index 0000000000..d20f52b7de --- /dev/null +++ b/vendor/golang.org/x/net/context/go17.go @@ -0,0 +1,72 @@ +// Copyright 2016 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// +build go1.7 + +package context + +import ( + "context" // standard library's context, as of Go 1.7 + "time" +) + +var ( + todo = context.TODO() + background = context.Background() +) + +// Canceled is the error returned by Context.Err when the context is canceled. +var Canceled = context.Canceled + +// DeadlineExceeded is the error returned by Context.Err when the context's +// deadline passes. +var DeadlineExceeded = context.DeadlineExceeded + +// WithCancel returns a copy of parent with a new Done channel. The returned +// context's Done channel is closed when the returned cancel function is called +// or when the parent context's Done channel is closed, whichever happens first. +// +// Canceling this context releases resources associated with it, so code should +// call cancel as soon as the operations running in this Context complete. +func WithCancel(parent Context) (ctx Context, cancel CancelFunc) { + ctx, f := context.WithCancel(parent) + return ctx, CancelFunc(f) +} + +// WithDeadline returns a copy of the parent context with the deadline adjusted +// to be no later than d. If the parent's deadline is already earlier than d, +// WithDeadline(parent, d) is semantically equivalent to parent. The returned +// context's Done channel is closed when the deadline expires, when the returned +// cancel function is called, or when the parent context's Done channel is +// closed, whichever happens first. +// +// Canceling this context releases resources associated with it, so code should +// call cancel as soon as the operations running in this Context complete. +func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc) { + ctx, f := context.WithDeadline(parent, deadline) + return ctx, CancelFunc(f) +} + +// WithTimeout returns WithDeadline(parent, time.Now().Add(timeout)). +// +// Canceling this context releases resources associated with it, so code should +// call cancel as soon as the operations running in this Context complete: +// +// func slowOperationWithTimeout(ctx context.Context) (Result, error) { +// ctx, cancel := context.WithTimeout(ctx, 100*time.Millisecond) +// defer cancel() // releases resources if slowOperation completes before timeout elapses +// return slowOperation(ctx) +// } +func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) { + return WithDeadline(parent, time.Now().Add(timeout)) +} + +// WithValue returns a copy of parent in which the value associated with key is +// val. +// +// Use context Values only for request-scoped data that transits processes and +// APIs, not for passing optional parameters to functions. +func WithValue(parent Context, key interface{}, val interface{}) Context { + return context.WithValue(parent, key, val) +} diff --git a/vendor/golang.org/x/net/context/go19.go b/vendor/golang.org/x/net/context/go19.go new file mode 100644 index 0000000000..d88bd1db12 --- /dev/null +++ b/vendor/golang.org/x/net/context/go19.go @@ -0,0 +1,20 @@ +// Copyright 2017 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// +build go1.9 + +package context + +import "context" // standard library's context, as of Go 1.7 + +// A Context carries a deadline, a cancelation signal, and other values across +// API boundaries. +// +// Context's methods may be called by multiple goroutines simultaneously. +type Context = context.Context + +// A CancelFunc tells an operation to abandon its work. +// A CancelFunc does not wait for the work to stop. +// After the first call, subsequent calls to a CancelFunc do nothing. +type CancelFunc = context.CancelFunc diff --git a/vendor/golang.org/x/net/context/pre_go17.go b/vendor/golang.org/x/net/context/pre_go17.go new file mode 100644 index 0000000000..0f35592df5 --- /dev/null +++ b/vendor/golang.org/x/net/context/pre_go17.go @@ -0,0 +1,300 @@ +// Copyright 2014 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// +build !go1.7 + +package context + +import ( + "errors" + "fmt" + "sync" + "time" +) + +// An emptyCtx is never canceled, has no values, and has no deadline. It is not +// struct{}, since vars of this type must have distinct addresses. +type emptyCtx int + +func (*emptyCtx) Deadline() (deadline time.Time, ok bool) { + return +} + +func (*emptyCtx) Done() <-chan struct{} { + return nil +} + +func (*emptyCtx) Err() error { + return nil +} + +func (*emptyCtx) Value(key interface{}) interface{} { + return nil +} + +func (e *emptyCtx) String() string { + switch e { + case background: + return "context.Background" + case todo: + return "context.TODO" + } + return "unknown empty Context" +} + +var ( + background = new(emptyCtx) + todo = new(emptyCtx) +) + +// Canceled is the error returned by Context.Err when the context is canceled. +var Canceled = errors.New("context canceled") + +// DeadlineExceeded is the error returned by Context.Err when the context's +// deadline passes. +var DeadlineExceeded = errors.New("context deadline exceeded") + +// WithCancel returns a copy of parent with a new Done channel. The returned +// context's Done channel is closed when the returned cancel function is called +// or when the parent context's Done channel is closed, whichever happens first. +// +// Canceling this context releases resources associated with it, so code should +// call cancel as soon as the operations running in this Context complete. +func WithCancel(parent Context) (ctx Context, cancel CancelFunc) { + c := newCancelCtx(parent) + propagateCancel(parent, c) + return c, func() { c.cancel(true, Canceled) } +} + +// newCancelCtx returns an initialized cancelCtx. +func newCancelCtx(parent Context) *cancelCtx { + return &cancelCtx{ + Context: parent, + done: make(chan struct{}), + } +} + +// propagateCancel arranges for child to be canceled when parent is. +func propagateCancel(parent Context, child canceler) { + if parent.Done() == nil { + return // parent is never canceled + } + if p, ok := parentCancelCtx(parent); ok { + p.mu.Lock() + if p.err != nil { + // parent has already been canceled + child.cancel(false, p.err) + } else { + if p.children == nil { + p.children = make(map[canceler]bool) + } + p.children[child] = true + } + p.mu.Unlock() + } else { + go func() { + select { + case <-parent.Done(): + child.cancel(false, parent.Err()) + case <-child.Done(): + } + }() + } +} + +// parentCancelCtx follows a chain of parent references until it finds a +// *cancelCtx. This function understands how each of the concrete types in this +// package represents its parent. +func parentCancelCtx(parent Context) (*cancelCtx, bool) { + for { + switch c := parent.(type) { + case *cancelCtx: + return c, true + case *timerCtx: + return c.cancelCtx, true + case *valueCtx: + parent = c.Context + default: + return nil, false + } + } +} + +// removeChild removes a context from its parent. +func removeChild(parent Context, child canceler) { + p, ok := parentCancelCtx(parent) + if !ok { + return + } + p.mu.Lock() + if p.children != nil { + delete(p.children, child) + } + p.mu.Unlock() +} + +// A canceler is a context type that can be canceled directly. The +// implementations are *cancelCtx and *timerCtx. +type canceler interface { + cancel(removeFromParent bool, err error) + Done() <-chan struct{} +} + +// A cancelCtx can be canceled. When canceled, it also cancels any children +// that implement canceler. +type cancelCtx struct { + Context + + done chan struct{} // closed by the first cancel call. + + mu sync.Mutex + children map[canceler]bool // set to nil by the first cancel call + err error // set to non-nil by the first cancel call +} + +func (c *cancelCtx) Done() <-chan struct{} { + return c.done +} + +func (c *cancelCtx) Err() error { + c.mu.Lock() + defer c.mu.Unlock() + return c.err +} + +func (c *cancelCtx) String() string { + return fmt.Sprintf("%v.WithCancel", c.Context) +} + +// cancel closes c.done, cancels each of c's children, and, if +// removeFromParent is true, removes c from its parent's children. +func (c *cancelCtx) cancel(removeFromParent bool, err error) { + if err == nil { + panic("context: internal error: missing cancel error") + } + c.mu.Lock() + if c.err != nil { + c.mu.Unlock() + return // already canceled + } + c.err = err + close(c.done) + for child := range c.children { + // NOTE: acquiring the child's lock while holding parent's lock. + child.cancel(false, err) + } + c.children = nil + c.mu.Unlock() + + if removeFromParent { + removeChild(c.Context, c) + } +} + +// WithDeadline returns a copy of the parent context with the deadline adjusted +// to be no later than d. If the parent's deadline is already earlier than d, +// WithDeadline(parent, d) is semantically equivalent to parent. The returned +// context's Done channel is closed when the deadline expires, when the returned +// cancel function is called, or when the parent context's Done channel is +// closed, whichever happens first. +// +// Canceling this context releases resources associated with it, so code should +// call cancel as soon as the operations running in this Context complete. +func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc) { + if cur, ok := parent.Deadline(); ok && cur.Before(deadline) { + // The current deadline is already sooner than the new one. + return WithCancel(parent) + } + c := &timerCtx{ + cancelCtx: newCancelCtx(parent), + deadline: deadline, + } + propagateCancel(parent, c) + d := deadline.Sub(time.Now()) + if d <= 0 { + c.cancel(true, DeadlineExceeded) // deadline has already passed + return c, func() { c.cancel(true, Canceled) } + } + c.mu.Lock() + defer c.mu.Unlock() + if c.err == nil { + c.timer = time.AfterFunc(d, func() { + c.cancel(true, DeadlineExceeded) + }) + } + return c, func() { c.cancel(true, Canceled) } +} + +// A timerCtx carries a timer and a deadline. It embeds a cancelCtx to +// implement Done and Err. It implements cancel by stopping its timer then +// delegating to cancelCtx.cancel. +type timerCtx struct { + *cancelCtx + timer *time.Timer // Under cancelCtx.mu. + + deadline time.Time +} + +func (c *timerCtx) Deadline() (deadline time.Time, ok bool) { + return c.deadline, true +} + +func (c *timerCtx) String() string { + return fmt.Sprintf("%v.WithDeadline(%s [%s])", c.cancelCtx.Context, c.deadline, c.deadline.Sub(time.Now())) +} + +func (c *timerCtx) cancel(removeFromParent bool, err error) { + c.cancelCtx.cancel(false, err) + if removeFromParent { + // Remove this timerCtx from its parent cancelCtx's children. + removeChild(c.cancelCtx.Context, c) + } + c.mu.Lock() + if c.timer != nil { + c.timer.Stop() + c.timer = nil + } + c.mu.Unlock() +} + +// WithTimeout returns WithDeadline(parent, time.Now().Add(timeout)). +// +// Canceling this context releases resources associated with it, so code should +// call cancel as soon as the operations running in this Context complete: +// +// func slowOperationWithTimeout(ctx context.Context) (Result, error) { +// ctx, cancel := context.WithTimeout(ctx, 100*time.Millisecond) +// defer cancel() // releases resources if slowOperation completes before timeout elapses +// return slowOperation(ctx) +// } +func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) { + return WithDeadline(parent, time.Now().Add(timeout)) +} + +// WithValue returns a copy of parent in which the value associated with key is +// val. +// +// Use context Values only for request-scoped data that transits processes and +// APIs, not for passing optional parameters to functions. +func WithValue(parent Context, key interface{}, val interface{}) Context { + return &valueCtx{parent, key, val} +} + +// A valueCtx carries a key-value pair. It implements Value for that key and +// delegates all other calls to the embedded Context. +type valueCtx struct { + Context + key, val interface{} +} + +func (c *valueCtx) String() string { + return fmt.Sprintf("%v.WithValue(%#v, %#v)", c.Context, c.key, c.val) +} + +func (c *valueCtx) Value(key interface{}) interface{} { + if c.key == key { + return c.val + } + return c.Context.Value(key) +} diff --git a/vendor/golang.org/x/net/context/pre_go19.go b/vendor/golang.org/x/net/context/pre_go19.go new file mode 100644 index 0000000000..b105f80be4 --- /dev/null +++ b/vendor/golang.org/x/net/context/pre_go19.go @@ -0,0 +1,109 @@ +// Copyright 2014 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// +build !go1.9 + +package context + +import "time" + +// A Context carries a deadline, a cancelation signal, and other values across +// API boundaries. +// +// Context's methods may be called by multiple goroutines simultaneously. +type Context interface { + // Deadline returns the time when work done on behalf of this context + // should be canceled. Deadline returns ok==false when no deadline is + // set. Successive calls to Deadline return the same results. + Deadline() (deadline time.Time, ok bool) + + // Done returns a channel that's closed when work done on behalf of this + // context should be canceled. Done may return nil if this context can + // never be canceled. Successive calls to Done return the same value. + // + // WithCancel arranges for Done to be closed when cancel is called; + // WithDeadline arranges for Done to be closed when the deadline + // expires; WithTimeout arranges for Done to be closed when the timeout + // elapses. + // + // Done is provided for use in select statements: + // + // // Stream generates values with DoSomething and sends them to out + // // until DoSomething returns an error or ctx.Done is closed. + // func Stream(ctx context.Context, out chan<- Value) error { + // for { + // v, err := DoSomething(ctx) + // if err != nil { + // return err + // } + // select { + // case <-ctx.Done(): + // return ctx.Err() + // case out <- v: + // } + // } + // } + // + // See http://blog.golang.org/pipelines for more examples of how to use + // a Done channel for cancelation. + Done() <-chan struct{} + + // Err returns a non-nil error value after Done is closed. Err returns + // Canceled if the context was canceled or DeadlineExceeded if the + // context's deadline passed. No other values for Err are defined. + // After Done is closed, successive calls to Err return the same value. + Err() error + + // Value returns the value associated with this context for key, or nil + // if no value is associated with key. Successive calls to Value with + // the same key returns the same result. + // + // Use context values only for request-scoped data that transits + // processes and API boundaries, not for passing optional parameters to + // functions. + // + // A key identifies a specific value in a Context. Functions that wish + // to store values in Context typically allocate a key in a global + // variable then use that key as the argument to context.WithValue and + // Context.Value. A key can be any type that supports equality; + // packages should define keys as an unexported type to avoid + // collisions. + // + // Packages that define a Context key should provide type-safe accessors + // for the values stores using that key: + // + // // Package user defines a User type that's stored in Contexts. + // package user + // + // import "golang.org/x/net/context" + // + // // User is the type of value stored in the Contexts. + // type User struct {...} + // + // // key is an unexported type for keys defined in this package. + // // This prevents collisions with keys defined in other packages. + // type key int + // + // // userKey is the key for user.User values in Contexts. It is + // // unexported; clients use user.NewContext and user.FromContext + // // instead of using this key directly. + // var userKey key = 0 + // + // // NewContext returns a new Context that carries value u. + // func NewContext(ctx context.Context, u *User) context.Context { + // return context.WithValue(ctx, userKey, u) + // } + // + // // FromContext returns the User value stored in ctx, if any. + // func FromContext(ctx context.Context) (*User, bool) { + // u, ok := ctx.Value(userKey).(*User) + // return u, ok + // } + Value(key interface{}) interface{} +} + +// A CancelFunc tells an operation to abandon its work. +// A CancelFunc does not wait for the work to stop. +// After the first call, subsequent calls to a CancelFunc do nothing. +type CancelFunc func() diff --git a/vendor/vendor.json b/vendor/vendor.json index f31daa9250..648ee6489c 100644 --- a/vendor/vendor.json +++ b/vendor/vendor.json @@ -3,11 +3,13 @@ "ignore": "test", "package": [ {"path":"github.com/armon/go-proxyproto","checksumSHA1":"eAall4ACaMG40mzSJ5Oc95GiF1A=","revision":"609d6338d3a76ec26ac3fe7045a164d9a58436e7","revisionTime":"2015-02-06T18:58:55-08:00"}, + {"path":"github.com/cenk/backoff","checksumSHA1":"MsfvQZcBtSZeatCqbSZRHVsb9yU=","revision":"2ea60e5f094469f9e65adb9cd103795b73ae743e","revisionTime":"2017-12-24T16:42:12Z"}, {"path":"github.com/circonus-labs/circonus-gometrics","checksumSHA1":"ZAdLZ0e/hin4AXxdS9F8y0yi/bg=","revision":"f8c68ed96a065c10344dcaf802f608781fc0a981","revisionTime":"2016-08-30T16:47:25Z"}, {"path":"github.com/circonus-labs/circonus-gometrics/api","checksumSHA1":"2hV0W0wM75u+OJ4kGv0i7B95cmY=","revision":"f8c68ed96a065c10344dcaf802f608781fc0a981","revisionTime":"2016-08-30T16:47:25Z"}, {"path":"github.com/circonus-labs/circonus-gometrics/checkmgr","checksumSHA1":"r3XLF0s89Y7UJgS9YP3NbdyafNY=","revision":"f8c68ed96a065c10344dcaf802f608781fc0a981","revisionTime":"2016-08-30T16:47:25Z"}, {"path":"github.com/circonus-labs/circonusllhist","checksumSHA1":"C4Z7+l5GOpOCW5DcvNYzheGvQRE=","revision":"d724266ae5270ae8b87a5d2e8081f04e307c3c18","revisionTime":"2016-05-26T04:38:13Z"}, {"path":"github.com/cyberdelia/go-metrics-graphite","checksumSHA1":"8/Q1JbAHUmL4sDURLq6yron4K/I=","revision":"b8345b7f01d571b05366d5791a034d872f1bb36f","revisionTime":"2015-08-25T20:22:00-07:00"}, + {"path":"github.com/facebookgo/clock","checksumSHA1":"imR2wF388/0fBU6RRWx8RvTi8Q8=","revision":"600d898af40aa09a7a93ecb9265d87b0504b6f03","revisionTime":"2015-04-10T01:09:13Z"}, {"path":"github.com/fatih/structs","checksumSHA1":"zKLQNNEpgfrJOwVrDDyBMYEIx/w=","revision":"5ada2f449b108d87dbd8c1e60c32cdd065c27886","revisionTime":"2016-06-01T09:31:17Z"}, {"path":"github.com/hashicorp/consul/api","checksumSHA1":"GsJ84gKbQno8KbojhVTgSVWNues=","revision":"402636ff2db998edef392ac6d59210d2170b3ebf","revisionTime":"2017-04-05T04:22:14Z","version":"v0.8.0","versionExact":"v0.8.0"}, {"path":"github.com/hashicorp/errwrap","checksumSHA1":"cdOCt0Yb+hdErz8NAQqayxPmRsY=","revision":"7554cd9344cec97297fa6649b055a8c98c2a1e55","revisionTime":"2014-10-27T22:47:10-07:00"}, @@ -34,8 +36,10 @@ {"path":"github.com/pkg/profile","checksumSHA1":"C3yiSMdTQxSY3xqKJzMV9T+KnIc=","revision":"5b67d428864e92711fcbd2f8629456121a56d91f","revisionTime":"2017-05-09T09:25:25Z"}, {"path":"github.com/rcrowley/go-metrics","checksumSHA1":"ODTWX4h8f+DW3oWZFT0yTmfHzdg=","revision":"3e5e593311103d49927c8d2b0fd93ccdfe4a525c","revisionTime":"2015-07-19T09:56:14-07:00"}, {"path":"github.com/rogpeppe/fastuuid","checksumSHA1":"ehRkDJisGCCSYdNgyvs1gSywSPE=","revision":"6724a57986aff9bff1a1770e9347036def7c89f6","revisionTime":"2015-01-06T09:31:45Z"}, + {"path":"github.com/rubyist/circuitbreaker","checksumSHA1":"q6JII/ziSLMqwajFGHj6w8V+RbA=","revision":"2074adba5ddc7d5f7559448a9c3066573521c5bf","revisionTime":"2017-03-30T19:36:47Z"}, {"path":"github.com/ryanuber/go-glob","checksumSHA1":"EYkx4sXDLSEd1xUtGoXRsfd5cpw=","revision":"572520ed46dbddaed19ea3d9541bdd0494163693","revisionTime":"2016-02-26T08:37:05Z"}, {"path":"github.com/sergi/go-diff/diffmatchpatch","checksumSHA1":"iWCtyR1TkJ22Bi/ygzfKDvOQdQY=","revision":"24e2351369ec4949b2ed0dc5c477afdd4c4034e8","revisionTime":"2017-01-18T13:12:30Z"}, + {"path":"golang.org/x/net/context","checksumSHA1":"GtamqiJoL7PGHsN454AoffBFMa8=","revision":"a8b9294777976932365dabb6640cf1468d95c70f","revisionTime":"2017-11-29T19:21:16Z"}, {"path":"golang.org/x/net/http2","checksumSHA1":"N1akwAdrHVfPPrsFOhG2ouP21VA=","revision":"f2499483f923065a842d38eb4c7f1927e6fc6e6d","revisionTime":"2017-01-14T04:22:49Z"}, {"path":"golang.org/x/net/http2/hpack","checksumSHA1":"HzuGD7AwgC0p1az1WAQnEFnEk98=","revision":"f2499483f923065a842d38eb4c7f1927e6fc6e6d","revisionTime":"2017-01-14T04:22:49Z"}, {"path":"golang.org/x/net/idna","checksumSHA1":"GIGmSrYACByf5JDIP9ByBZksY80=","revision":"f2499483f923065a842d38eb4c7f1927e6fc6e6d","revisionTime":"2017-01-14T04:22:49Z"}, From c2a2605c9e24b4f9170bee121c0dca02b5891b72 Mon Sep 17 00:00:00 2001 From: Frank Schroeder Date: Fri, 29 Dec 2017 14:23:13 +0100 Subject: [PATCH 2/3] route: add support to delete all routes for a destination route del * * http://host:port will delete all routes for that destination --- route/table.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/route/table.go b/route/table.go index 6a23021d46..0b0aa51d62 100644 --- a/route/table.go +++ b/route/table.go @@ -207,6 +207,15 @@ func (t Table) delRoute(d *RouteDef) error { } } + case d.Service == "*" && d.Src == "*": + for _, routes := range t { + for _, r := range routes { + r.filter(func(tg *Target) bool { + return tg.URL.String() == d.Dst + }) + } + } + case d.Src == "" && d.Dst == "": for _, routes := range t { for _, r := range routes { From 5bef5d3084478826084e8a3a4cc26caeb6c2bab5 Mon Sep 17 00:00:00 2001 From: Frank Schroeder Date: Fri, 29 Dec 2017 14:27:04 +0100 Subject: [PATCH 3/3] WIP: add circuit breaker to detect network failures This patch adds a circuit breaker monitor which issues route table fragments to delete routes for which the circuit breaker is open. The current implementation supports only the HTTP path. TCP and WS will follow. The integration test isn't testing the behavior automatically yet but can be used to eyeball the behavior. This needs to be improved. Also missing is the code to GC the circuit breakers that aren't tripped over time but that should be easy to implement in this code. Fixes #393 --- cb/monitor.go | 120 +++++++++++++++++++++++++++++++++ main.go | 17 +++-- proxy/http_integration_test.go | 76 +++++++++++++++++++++ proxy/http_proxy.go | 38 +++++++++++ 4 files changed, 246 insertions(+), 5 deletions(-) create mode 100644 cb/monitor.go diff --git a/cb/monitor.go b/cb/monitor.go new file mode 100644 index 0000000000..7ee4226f52 --- /dev/null +++ b/cb/monitor.go @@ -0,0 +1,120 @@ +package cb + +import ( + "log" + "strings" + "time" + + circuit "github.com/rubyist/circuitbreaker" +) + +// Monitor implements a circuit breaker monitor which manages +// multiple circuit breakers and generates routing table updates +// from the state changes. +// +// Monitor generates a new routing table fragment on a state change +// and on a regular basis. The default is to trigger the breaker +// after three consecutive failures. +type Monitor struct { + UpdateInterval time.Duration + ConsecFailures int + routes chan string + fail chan string + success chan string + done chan struct{} +} + +func NewMonitor() *Monitor { + return &Monitor{ + UpdateInterval: 15 * time.Second, + ConsecFailures: 3, + routes: make(chan string, 1), + fail: make(chan string, 100), + success: make(chan string, 100), + done: make(chan struct{}), + } +} + +func (m *Monitor) Stop() { + close(m.done) +} + +func (m *Monitor) Start() { + cbs := make(map[string]*circuit.Breaker) + + getcb := func(addr string) *circuit.Breaker { + cb := cbs[addr] + if cb == nil { + cb = circuit.NewConsecutiveBreaker(int64(m.ConsecFailures)) + cbs[addr] = cb + } + return cb + } + + ticker := time.NewTicker(m.UpdateInterval) + for { + select { + case <-ticker.C: + ready := 0 + for addr, cb := range cbs { + if cb.Tripped() && cb.Ready() { + ready++ + log.Printf("[INFO] breaker: retrying routes for %s", addr) + } + } + if ready > 0 { + m.routes <- m.update(cbs) + } + + case addr := <-m.fail: + cb := getcb(addr) + wasready := cb.Ready() + cb.Fail() + if wasready && cb.Tripped() { + log.Printf("[WARN] breaker: breaker for %s tripped", addr) + m.routes <- m.update(cbs) + } + + case addr := <-m.success: + cb := getcb(addr) + wasready := cb.Ready() + cb.Success() + if !wasready && !cb.Tripped() { + log.Printf("[INFO] breaker: breaker for %s recovered", addr) + m.routes <- m.update(cbs) + } + + case <-m.done: + return + } + } +} + +func (m *Monitor) SuccessHost(addr string) { + select { + case m.success <- addr: + default: + } +} + +func (m *Monitor) FailHost(addr string) { + select { + case m.fail <- addr: + default: + } +} + +func (m *Monitor) Routes() <-chan string { + return m.routes +} + +func (m *Monitor) update(cbs map[string]*circuit.Breaker) string { + var s []string + for addr, cb := range cbs { + if cb.Tripped() || !cb.Ready() { + s = append(s, "route del * * http://"+addr) + } + } + routes := strings.Join(s, "\n") + return routes +} diff --git a/main.go b/main.go index 03b5a93cce..3fd7256015 100644 --- a/main.go +++ b/main.go @@ -18,6 +18,7 @@ import ( "time" "github.com/fabiolb/fabio/admin" + "github.com/fabiolb/fabio/cb" "github.com/fabiolb/fabio/cert" "github.com/fabiolb/fabio/config" "github.com/fabiolb/fabio/exit" @@ -97,12 +98,16 @@ func main() { log.Printf("[INFO] Profile path %q", cfg.ProfilePath) } + cbmon := cb.NewMonitor() + go cbmon.Start() + exit.Listen(func(s os.Signal) { atomic.StoreInt32(&shuttingDown, 1) proxy.Shutdown(cfg.Proxy.ShutdownWait) if prof != nil { prof.Stop() } + cbmon.Stop() if registry.Default == nil { return } @@ -119,12 +124,12 @@ func main() { go watchNoRouteHTML(cfg) first := make(chan bool) - go watchBackend(cfg, first) + go watchBackend(cfg, first, cbmon) log.Print("[INFO] Waiting for first routing table") <-first // create proxies after metrics since they use the metrics registry. - startServers(cfg) + startServers(cfg, cbmon) exit.Wait() log.Print("[INFO] Down") } @@ -241,7 +246,7 @@ func startAdmin(cfg *config.Config) { }() } -func startServers(cfg *config.Config) { +func startServers(cfg *config.Config, cbmon *cb.Monitor) { for _, l := range cfg.Listen { l := l // capture loop var for go routines below tlscfg, err := makeTLSConfig(l) @@ -371,11 +376,12 @@ func initBackend(cfg *config.Config) { } } -func watchBackend(cfg *config.Config, first chan bool) { +func watchBackend(cfg *config.Config, first chan bool, cbmon *cb.Monitor) { var ( last string svccfg string mancfg string + cbcfg string once sync.Once ) @@ -387,11 +393,12 @@ func watchBackend(cfg *config.Config, first chan bool) { select { case svccfg = <-svc: case mancfg = <-man: + case cbcfg = <-cbmon.Routes(): } // manual config overrides service config // order matters - next := svccfg + "\n" + mancfg + next := svccfg + "\n" + mancfg + "\n" + cbcfg if next == last { continue } diff --git a/proxy/http_integration_test.go b/proxy/http_integration_test.go index 30b173b70f..4a63b90084 100644 --- a/proxy/http_integration_test.go +++ b/proxy/http_integration_test.go @@ -7,6 +7,7 @@ import ( "crypto/x509" "fmt" "io/ioutil" + "log" "net" "net/http" "net/http/httptest" @@ -15,9 +16,11 @@ import ( "regexp" "sort" "strings" + "sync/atomic" "testing" "time" + "github.com/fabiolb/fabio/cb" "github.com/fabiolb/fabio/config" "github.com/fabiolb/fabio/logger" "github.com/fabiolb/fabio/noroute" @@ -118,6 +121,7 @@ func TestProxyStripsPath(t *testing.T) { w.WriteHeader(404) } })) + defer server.Close() proxy := httptest.NewServer(&HTTPProxy{ Transport: http.DefaultTransport, @@ -137,6 +141,78 @@ func TestProxyStripsPath(t *testing.T) { } } +func TestProxyTripsBreaker(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, "OK") + })) + defer server.Close() + + // create a transport with a short dial timeout to speedup tests + // otherwise, we have to wait 30 sec for every request to time out + tr := &http.Transport{ + Dial: (&net.Dialer{Timeout: 100 * time.Millisecond}).Dial, + } + + // create a routing table with one good and one bad target + routes := "route add mock / http://127.0.0.99:12345\n" + routes += "route add mock / " + server.URL + + // create a circuit breaker monitor which will generate routing + // table updates + cbmon := cb.NewMonitor() + cbmon.UpdateInterval = time.Second // check every second for recovered CBs + go cbmon.Start() + defer cbmon.Stop() + + // create sync value to contain the routing table + var syncTbl atomic.Value + tbl, err := route.NewTable(routes) + if err != nil { + t.Fatal(err) + } + syncTbl.Store(tbl) + + go func() { + for cbroutes := range cbmon.Routes() { + src := routes + "\n" + cbroutes + tbl, err := route.NewTable(src) + if err != nil { + t.Fatal(err) + } + syncTbl.Store(tbl) + log.Println("new routing table:\n" + tbl.String()) + } + }() + + proxy := httptest.NewServer(&HTTPProxy{ + Transport: tr, + Lookup: func(r *http.Request) *route.Target { + tbl := syncTbl.Load().(route.Table) + return tbl.Lookup(r, "", route.Picker["rr"], route.Matcher["prefix"]) + }, + CBMon: cbmon, + }) + defer proxy.Close() + + call := func(wantStatus int, wantBody string) { + t.Helper() + resp, body := mustGet(proxy.URL + "/") + t.Logf("GET %s: status: %d body: %q", proxy.URL, resp.StatusCode, string(body)) + // if got, want := resp.StatusCode, wantStatus; got != want { + // t.Fatalf("got status %d want %d", got, want) + // } + // if got, want := string(body), wantBody; got != want { + // t.Fatalf("got body %q want %q", got, want) + // } + } + + for i := 0; i < 100; i++ { + call(502, "") + time.Sleep(100 * time.Millisecond) + // call(200, "OK") + } +} + func TestProxyHost(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, r.Host) diff --git a/proxy/http_proxy.go b/proxy/http_proxy.go index f9aa849626..ce7f9e9786 100644 --- a/proxy/http_proxy.go +++ b/proxy/http_proxy.go @@ -11,6 +11,7 @@ import ( "strings" "time" + "github.com/fabiolb/fabio/cb" "github.com/fabiolb/fabio/config" "github.com/fabiolb/fabio/logger" "github.com/fabiolb/fabio/metrics" @@ -55,6 +56,10 @@ type HTTPProxy struct { // UUID returns a unique id in uuid format. // If UUID is nil, uuid.NewUUID() is used. UUID func() string + + // CBMon is a circuit breaker monitor which manages failure + // and success events and generates routing table updates from them. + CBMon *cb.Monitor } func (p *HTTPProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -167,11 +172,24 @@ func (p *HTTPProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { timeNow = time.Now } + // wrap ResponseWriter to capture the status code + w = &statusRW{w, http.StatusOK} + start := timeNow() h.ServeHTTP(w, r) end := timeNow() dur := end.Sub(start) + if p.CBMon != nil { + addr := targetURL.Host + switch w.(*statusRW).code { + case 502: // BadGateway is returned on i/o timeout + p.CBMon.FailHost(addr) + default: + p.CBMon.SuccessHost(addr) + } + } + if p.Requests != nil { p.Requests.Update(dur) } @@ -213,3 +231,23 @@ func key(code int) string { b = strconv.AppendInt(b, int64(code), 10) return string(b) } + +// statusRW wraps a http.ResponseWriter +// to capture the status code set by WriteHeader +type statusRW struct { + w http.ResponseWriter + code int +} + +func (w *statusRW) Header() http.Header { + return w.w.Header() +} + +func (w *statusRW) Write(p []byte) (int, error) { + return w.w.Write(p) +} + +func (w *statusRW) WriteHeader(code int) { + w.code = code + w.w.WriteHeader(code) +}